Responses context compaction

export API_BASE_URL="https://gateway.example.com/v1"
export API_KEY="YOUR_API_KEY"
export API_MODEL="YOUR_MODEL_ID"
.venv/bin/python examples/compaction.py

import os
from openai import OpenAI, APIConnectionError, APIStatusError

history = [
    {"role": "user", "content": "Plan a two-day visit to Kyoto without driving."},
    {"role": "assistant", "content": "Day one: eastern temples. Day two: Arashiyama by train."},
]
instructions = "Answer briefly in plain text. Preserve the user's travel constraints."
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
            api_key=os.environ["API_KEY"], timeout=60.0, max_retries=0) as client:
    try:
        compacted = client.post("/responses/compact", cast_to=dict, body={
            "model": os.environ["API_MODEL"], "input": history,
            "instructions": instructions,
        })
        if (not isinstance(compacted, dict) or compacted.get("error")
                or compacted.get("object") != "response.compaction"):
            raise ValueError("Invalid compaction response.")
        window = compacted.get("output")
        if not isinstance(window, list) or not window:
            raise ValueError("Empty compacted window.")
        found_compaction = False
        for item in window:
            if not isinstance(item, dict) or not isinstance(item.get("type"), str) or not item["type"]:
                raise ValueError("Invalid context item.")
            if item["type"] == "compaction":
                encrypted = item.get("encrypted_content")
                if not isinstance(encrypted, str) or not encrypted.strip():
                    raise ValueError("Missing encrypted content.")
                found_compaction = True
        if not found_compaction:
            raise ValueError("Missing compaction item.")
        # Keep every returned item, in order, including unknown metadata.
        next_input = [*window, {"role": "user", "content": "Add a rainy-day alternative."}]
        result = client.responses.create(
            model=os.environ["API_MODEL"], input=next_input,
            instructions=instructions, store=False,
        )
        if result.status != "completed" or not result.output_text.strip():
            raise ValueError("No complete text reply.")
        for item in result.output:
            if item.type == "reasoning":
                continue
            if (item.type != "message" or item.status != "completed"
                    or any(part.type != "output_text" for part in item.content)):
                raise ValueError("Unexpected or incomplete output item.")
        print(result.output_text)
    except APIStatusError as error:
        raise SystemExit(f"Request failed (HTTP {error.status_code}). No automatic retry.")
    except APIConnectionError:
        raise SystemExit("Connection failed. Original history must be retained.")
    except (ValueError, TypeError, AttributeError):
        raise SystemExit("Invalid compaction or continuation. Original history must be retained.")