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.")