feat: #2333 support input-based responses compaction with store-aware auto mode (#2334)

seratch · web-flow · commit 42f2c401c60f · 2026-01-20T09:35:00.000+09:00
diff --git a/examples/memory/compaction_session_example.py b/examples/memory/compaction_session_example.py
@@ -45,9 +45,9 @@ async def main():
         result = await Runner.run(agent, prompt, session=session)
         print(f"Assistant: {result.final_output}\n")
 
-    # Show final session state
+    # Show session state after automatic compaction (if triggered)
     items = await session.get_items()
-    print("=== Final Session State ===")
+    print("=== Session State (Auto Compaction) ===")
     print(f"Total items: {len(items)}")
     for item in items:
         # Some inputs are stored as easy messages (only `role` and `content`).
@@ -59,6 +59,27 @@ async def main():
             print(f"  - message ({role})")
         else:
             print(f"  - {item_type}")
+    print()
+
+    # Manual compaction after inspecting the auto-compacted state.
+    print("=== Manual Compaction ===")
+    await session.run_compaction({"force": True})
+    print("Done")
+    print()
+
+    # Show final session state after manual compaction
+    items = await session.get_items()
+    print("=== Session State (Manual Compaction) ===")
+    print(f"Total items: {len(items)}")
+    for item in items:
+        item_type = item.get("type") or ("message" if "role" in item else "unknown")
+        if item_type == "compaction":
+            print("  - compaction (encrypted content)")
+        elif item_type == "message":
+            role = item.get("role", "unknown")
+            print(f"  - message ({role})")
+        else:
+            print(f"  - {item_type}")
 
 
 if __name__ == "__main__":
diff --git a/examples/memory/compaction_session_stateless_example.py b/examples/memory/compaction_session_stateless_example.py
@@ -0,0 +1,85 @@
+"""
+Example demonstrating stateless compaction with store=False.
+
+In auto mode, OpenAIResponsesCompactionSession uses input-based compaction when
+responses are not stored on the server.
+"""
+
+import asyncio
+
+from agents import Agent, ModelSettings, OpenAIResponsesCompactionSession, Runner, SQLiteSession
+
+
+async def main():
+    # Create an underlying session for storage
+    underlying = SQLiteSession(":memory:")
+
+    # Wrap with compaction session in auto mode. When store=False, this will
+    # compact using the locally stored input items.
+    session = OpenAIResponsesCompactionSession(
+        session_id="demo-session",
+        underlying_session=underlying,
+        model="gpt-4.1",
+        compaction_mode="auto",
+        should_trigger_compaction=lambda ctx: len(ctx["compaction_candidate_items"]) >= 3,
+    )
+
+    agent = Agent(
+        name="Assistant",
+        instructions="Reply concisely. Keep answers to 1-2 sentences.",
+        model_settings=ModelSettings(store=False),
+    )
+
+    print("=== Stateless Compaction Session Example ===\n")
+
+    prompts = [
+        "What is the tallest mountain in the world?",
+        "How tall is it in feet?",
+        "When was it first climbed?",
+        "Who was on that expedition?",
+    ]
+
+    for i, prompt in enumerate(prompts, 1):
+        print(f"Turn {i}:")
+        print(f"User: {prompt}")
+        result = await Runner.run(agent, prompt, session=session)
+        print(f"Assistant: {result.final_output}\n")
+
+    # Show session state after automatic compaction (if triggered)
+    items = await session.get_items()
+    print("=== Session State (Auto Compaction) ===")
+    print(f"Total items: {len(items)}")
+    for item in items:
+        item_type = item.get("type") or ("message" if "role" in item else "unknown")
+        if item_type == "compaction":
+            print("  - compaction (encrypted content)")
+        elif item_type == "message":
+            role = item.get("role", "unknown")
+            print(f"  - message ({role})")
+        else:
+            print(f"  - {item_type}")
+    print()
+
+    # Manual compaction in stateless mode.
+    print("=== Manual Compaction ===")
+    await session.run_compaction({"force": True})
+    print("Done")
+    print()
+
+    # Show final session state
+    items = await session.get_items()
+    print("=== Final Session State ===")
+    print(f"Total items: {len(items)}")
+    for item in items:
+        item_type = item.get("type") or ("message" if "role" in item else "unknown")
+        if item_type == "compaction":
+            print("  - compaction (encrypted content)")
+        elif item_type == "message":
+            role = item.get("role", "unknown")
+            print(f"  - message ({role})")
+        else:
+            print(f"  - {item_type}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Literal
 
 from openai import AsyncOpenAI
 
@@ -21,6 +21,8 @@
 
 DEFAULT_COMPACTION_THRESHOLD = 10
 
+OpenAIResponsesCompactionMode = Literal["previous_response_id", "input", "auto"]
+
 
 def select_compaction_candidate_items(
     items: list[TResponseInputItem],
@@ -85,6 +87,7 @@ def __init__(
         *,
         client: AsyncOpenAI | None = None,
         model: str = "gpt-4.1",
+        compaction_mode: OpenAIResponsesCompactionMode = "auto",
         should_trigger_compaction: Callable[[dict[str, Any]], bool] | None = None,
     ):
         """Initialize the compaction session.
@@ -97,6 +100,9 @@ def __init__(
                 get_default_openai_client() or new AsyncOpenAI().
             model: Model to use for responses.compact. Defaults to "gpt-4.1". Must be an
                 OpenAI model name (gpt-*, o*, or ft:gpt-*).
+            compaction_mode: Controls how the compaction request provides conversation
+                history. "auto" (default) uses input when the last response was not
+                stored or no response_id is available.
             should_trigger_compaction: Custom decision hook. Defaults to triggering when
                 10+ compaction candidates exist.
         """
@@ -113,6 +119,7 @@ def __init__(
         self.underlying_session = underlying_session
         self._client = client
         self.model = model
+        self.compaction_mode = compaction_mode
         self.should_trigger_compaction = (
             should_trigger_compaction or default_should_trigger_compaction
         )
@@ -122,21 +129,54 @@ def __init__(
         self._session_items: list[TResponseInputItem] | None = None
         self._response_id: str | None = None
         self._deferred_response_id: str | None = None
+        self._last_unstored_response_id: str | None = None
 
     @property
     def client(self) -> AsyncOpenAI:
         if self._client is None:
             self._client = get_default_openai_client() or AsyncOpenAI()
         return self._client
 
+    def _resolve_compaction_mode_for_response(
+        self,
+        *,
+        response_id: str | None,
+        store: bool | None,
+        requested_mode: OpenAIResponsesCompactionMode | None,
+    ) -> _ResolvedCompactionMode:
+        mode = requested_mode or self.compaction_mode
+        if (
+            mode == "auto"
+            and store is None
+            and response_id is not None
+            and response_id == self._last_unstored_response_id
+        ):
+            return "input"
+        return _resolve_compaction_mode(mode, response_id=response_id, store=store)
+
     async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None:
         """Run compaction using responses.compact API."""
         if args and args.get("response_id"):
             self._response_id = args["response_id"]
+        requested_mode = args.get("compaction_mode") if args else None
+        if args and "store" in args:
+            store = args["store"]
+            if store is False and self._response_id:
+                self._last_unstored_response_id = self._response_id
+            elif store is True and self._response_id == self._last_unstored_response_id:
+                self._last_unstored_response_id = None
+        else:
+            store = None
+        resolved_mode = self._resolve_compaction_mode_for_response(
+            response_id=self._response_id,
+            store=store,
+            requested_mode=requested_mode,
+        )
 
-        if not self._response_id:
+        if resolved_mode == "previous_response_id" and not self._response_id:
             raise ValueError(
-                "OpenAIResponsesCompactionSession.run_compaction requires a response_id"
+                "OpenAIResponsesCompactionSession.run_compaction requires a response_id "
+                "when using previous_response_id compaction."
             )
 
         compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
@@ -145,23 +185,32 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
         should_compact = force or self.should_trigger_compaction(
             {
                 "response_id": self._response_id,
+                "compaction_mode": resolved_mode,
                 "compaction_candidate_items": compaction_candidate_items,
                 "session_items": session_items,
             }
         )
 
         if not should_compact:
-            logger.debug(f"skip: decision hook declined compaction for {self._response_id}")
+            logger.debug(
+                f"skip: decision hook declined compaction for {self._response_id} "
+                f"(mode={resolved_mode})"
+            )
             return
 
         self._deferred_response_id = None
-        logger.debug(f"compact: start for {self._response_id} using {self.model}")
-
-        compacted = await self.client.responses.compact(
-            previous_response_id=self._response_id,
-            model=self.model,
+        logger.debug(
+            f"compact: start for {self._response_id} using {self.model} (mode={resolved_mode})"
         )
 
+        compact_kwargs: dict[str, Any] = {"model": self.model}
+        if resolved_mode == "previous_response_id":
+            compact_kwargs["previous_response_id"] = self._response_id
+        else:
+            compact_kwargs["input"] = session_items
+
+        compacted = await self.client.responses.compact(**compact_kwargs)
+
         await self.underlying_session.clear_session()
         output_items: list[TResponseInputItem] = []
         if compacted.output:
@@ -183,19 +232,26 @@ async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None
 
         logger.debug(
             f"compact: done for {self._response_id} "
-            f"(output={len(output_items)}, candidates={len(self._compaction_candidate_items)})"
+            f"(mode={resolved_mode}, output={len(output_items)}, "
+            f"candidates={len(self._compaction_candidate_items)})"
         )
 
     async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
         return await self.underlying_session.get_items(limit)
 
-    async def _defer_compaction(self, response_id: str) -> None:
+    async def _defer_compaction(self, response_id: str, store: bool | None = None) -> None:
         if self._deferred_response_id is not None:
             return
         compaction_candidate_items, session_items = await self._ensure_compaction_candidates()
+        resolved_mode = self._resolve_compaction_mode_for_response(
+            response_id=response_id,
+            store=store,
+            requested_mode=None,
+        )
         should_compact = self.should_trigger_compaction(
             {
                 "response_id": response_id,
+                "compaction_mode": resolved_mode,
                 "compaction_candidate_items": compaction_candidate_items,
                 "session_items": session_items,
             }
@@ -247,3 +303,21 @@ async def _ensure_compaction_candidates(
             f"candidates: initialized (history={len(history)}, candidates={len(candidates)})"
         )
         return (candidates[:], history[:])
+
+
+_ResolvedCompactionMode = Literal["previous_response_id", "input"]
+
+
+def _resolve_compaction_mode(
+    requested_mode: OpenAIResponsesCompactionMode,
+    *,
+    response_id: str | None,
+    store: bool | None,
+) -> _ResolvedCompactionMode:
+    if requested_mode != "auto":
+        return requested_mode
+    if store is False:
+        return "input"
+    if not response_id:
+        return "input"
+    return "previous_response_id"
diff --git a/src/agents/memory/session.py b/src/agents/memory/session.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Literal, Protocol, runtime_checkable
 
 from typing_extensions import TypedDict, TypeGuard
 
@@ -107,6 +107,20 @@ class OpenAIResponsesCompactionArgs(TypedDict, total=False):
     response_id: str
     """The ID of the last response to use for compaction."""
 
+    compaction_mode: Literal["previous_response_id", "input", "auto"]
+    """How to provide history for compaction.
+
+    - "auto": Use input when the last response was not stored or no response ID is available.
+    - "previous_response_id": Use server-managed response history.
+    - "input": Send locally stored session items as input.
+    """
+
+    store: bool
+    """Whether the last model response was stored on the server.
+
+    When set to False, compaction should avoid "previous_response_id" unless explicitly requested.
+    """
+
     force: bool
     """Whether to force compaction even if the threshold is not met."""
 
diff --git a/src/agents/run.py b/src/agents/run.py
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
