diff --git a/python/PACKAGE_STATUS.md b/python/PACKAGE_STATUS.md index b8701ea2d8c..4ce55d57ba0 100644 --- a/python/PACKAGE_STATUS.md +++ b/python/PACKAGE_STATUS.md @@ -143,6 +143,14 @@ listed below. - `agent-framework-core`: `FunctionInvocationContext.add_tools` and `FunctionInvocationContext.remove_tools` from `agent_framework/_middleware.py` +#### `SERVER_COMPACTION` + +- `agent-framework-core`: `Content.from_compaction` and the `compaction` content type from + `agent_framework/_types.py` +- `agent-framework-openai`: `ContextManagementOptions` and the `context_management` option on + `OpenAIChatOptions` (also used by `FoundryChatClient`), including replay of compaction items for + `store=False` requests + #### `SESSION_STORE` - `agent-framework-core`: `SessionStore` and `FileSessionStore` from diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 13361bbeaa9..4f82d6ca860 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -74,6 +74,9 @@ agent_framework/ constructors are experimental under `COMPUTER_USE`; the rest of `Content` retains its existing stage. Computer results can omit screenshots in core; OpenAI-based connectors require them when converting to Responses items. Completed call/result pairs remain in the transcript but are not user-input requests. +- **Server-side compaction** - `Content.from_compaction` and the `compaction` content type are experimental under + `SERVER_COMPACTION`. The content carries a provider's opaque compacted conversation state in `protected_data`; + how it composes with `CompactionStrategy` / `CompactionProvider` is not settled yet. - **`ChatOptions`** - TypedDict for chat request options ### Tools (`_tools.py`) diff --git a/python/packages/core/agent_framework/_feature_stage.py b/python/packages/core/agent_framework/_feature_stage.py index 24309e0b144..4578f4a1618 100644 --- a/python/packages/core/agent_framework/_feature_stage.py +++ b/python/packages/core/agent_framework/_feature_stage.py @@ -63,6 +63,7 @@ class ExperimentalFeature(str, Enum): MCP_LONG_RUNNING_TASKS = "MCP_LONG_RUNNING_TASKS" MCP_SKILLS = "MCP_SKILLS" PROGRESSIVE_TOOLS = "PROGRESSIVE_TOOLS" + SERVER_COMPACTION = "SERVER_COMPACTION" SESSION_STORE = "SESSION_STORE" TO_PROMPT_AGENT = "TO_PROMPT_AGENT" VECTOR_STORES = "VECTOR_STORES" diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index f98743794a4..b01573d732d 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -387,6 +387,7 @@ def _restore_compaction_annotation_in_additional_properties( "function_approval_request", "function_approval_response", "oauth_consent_request", + "compaction", ] @@ -1505,6 +1506,43 @@ def from_oauth_consent_request( raw_representation=raw_representation, ) + @classmethod + @experimental(feature_id=ExperimentalFeature.SERVER_COMPACTION) + def from_compaction( + cls: type[ContentT], + *, + protected_data: str, + id: str | None = None, + annotations: Sequence[Annotation] | None = None, + additional_properties: MutableMapping[str, Any] | None = None, + raw_representation: Any = None, + ) -> ContentT: + """Create compaction content. + + A compaction item is an opaque, provider-encrypted summary of earlier conversation + state, produced by service-side context compaction (for example the OpenAI Responses + API ``context_management`` option). It carries that state into later turns in fewer + tokens, and providers that support it replay it as input on the next request. + + Keyword Args: + protected_data: The provider's encrypted compaction payload. + id: The provider's identifier for the compaction item. + annotations: Optional annotations. + additional_properties: Optional additional properties. + raw_representation: Optional raw representation from the provider. + + Returns: + A new Content instance with type ``compaction``. + """ + return cls( + "compaction", + id=id, + protected_data=protected_data, + annotations=annotations, + additional_properties=additional_properties, + raw_representation=raw_representation, + ) + def to_function_approval_response( self, approved: bool, diff --git a/python/packages/core/tests/core/test_feature_stage.py b/python/packages/core/tests/core/test_feature_stage.py index 797fc859e8d..2284276328e 100644 --- a/python/packages/core/tests/core/test_feature_stage.py +++ b/python/packages/core/tests/core/test_feature_stage.py @@ -82,6 +82,20 @@ def test_computer_use_content_is_experimental_without_staging_all_content() -> N assert "[COMPUTER_USE]" in str(caught[0].message) +def test_server_compaction_content_is_experimental() -> None: + constructor = Content.from_compaction + assert getattr(constructor, "__feature_stage__", None) == "experimental" + assert getattr(constructor, "__feature_id__", None) == PublicExperimentalFeature.SERVER_COMPACTION.value + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", ExperimentalWarning) + content = Content.from_compaction(id="cmp_1", protected_data="state") + + assert content.type == "compaction" + assert len(caught) == 1 + assert "[SERVER_COMPACTION]" in str(caught[0].message) + + def test_experimental_decorator_accepts_feature_enum() -> None: with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 05d474687f4..5e7fcff95b5 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -5982,6 +5982,18 @@ def test_oauth_consent_request_serialization_roundtrip(): assert d["user_input_request"] is True +def test_compaction_content_creation_and_serialization_roundtrip(): + """Test Content.from_compaction keeps the encrypted payload and id through serialization.""" + content = Content.from_compaction(id="cmp_1", protected_data="encrypted-state") + + assert content.type == "compaction" + assert content.id == "cmp_1" + assert content.protected_data == "encrypted-state" + + restored = Content.from_dict(content.to_dict()) + assert (restored.type, restored.id, restored.protected_data) == ("compaction", "cmp_1", "encrypted-state") + + # endregion diff --git a/python/packages/foundry/tests/foundry/test_foundry_chat_client.py b/python/packages/foundry/tests/foundry/test_foundry_chat_client.py index a4660da378c..080d1f3aeb1 100644 --- a/python/packages/foundry/tests/foundry/test_foundry_chat_client.py +++ b/python/packages/foundry/tests/foundry/test_foundry_chat_client.py @@ -2224,3 +2224,27 @@ async def test_shared_chat_client_keeps_concurrent_agent_runs_isolated( assert first_session.service_session_id == "response-first" assert second_session.service_session_id == "response-second" assert transport.max_active_requests == 2 + + +async def test_foundry_chat_client_replays_compaction_item_when_store_false() -> None: + """FoundryChatClient forwards context_management and replays the latest compaction item statelessly.""" + mock_project_client = MagicMock() + mock_project_client.get_openai_client.return_value = _make_mock_openai_client() + client = FoundryChatClient(project_client=mock_project_client, model=_TEST_FOUNDRY_MODEL) + context_management = [{"type": "compaction", "compact_threshold": 200000}] + + _, run_options, _ = await client._prepare_request( + [ + Message(role="user", contents=["Earlier question"]), + Message( + role="assistant", + contents=[Content.from_compaction(id="cmp_1", protected_data="state"), "Earlier answer"], + ), + Message(role="user", contents=["Next question"]), + ], + {"store": False, "context_management": context_management}, + ) + + assert run_options["context_management"] == context_management + assert run_options["input"][0] == {"type": "compaction", "encrypted_content": "state", "id": "cmp_1"} + assert "Earlier question" not in str(run_options["input"]) diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index 150160aac62..60932976f51 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -43,7 +43,10 @@ GROUP_ID_KEY, CompactionStrategy, TokenizerProtocol, + included_messages, + set_excluded, ) +from agent_framework._feature_stage import ExperimentalFeature, experimental from agent_framework._middleware import ChatAndFunctionMiddlewareTypes, ChatMiddlewareLayer from agent_framework._settings import SecretString from agent_framework._telemetry import USER_AGENT_KEY, mark_feature_used @@ -204,6 +207,17 @@ class ReasoningOptions(TypedDict, total=False): """How to summarize reasoning in the response.""" +@experimental(feature_id=ExperimentalFeature.SERVER_COMPACTION) +class ContextManagementOptions(TypedDict, total=False): + """Service-side context management entry for the Responses API.""" + + type: Literal["compaction"] + """The context management strategy. Only ``compaction`` is supported.""" + + compact_threshold: int + """Token count at which the service compacts the context.""" + + class StreamOptions(TypedDict, total=False): """Options for streaming responses.""" @@ -258,6 +272,14 @@ class OpenAIChatOptions(ChatOptions[ResponseFormatT], Generic[ResponseFormatT], Sending this option requires openai 2.45.0 or later. See: https://developers.openai.com/api/docs/guides/prompt-caching#prompt-cache-breakpoints""" + context_management: list[ContextManagementOptions] + """Service-side context management, such as automatic compaction. + When the context crosses ``compact_threshold`` the service emits an encrypted + compaction item, surfaced as ``compaction`` content. With ``store=False`` the + latest compaction item is replayed on later requests and the history before it + (other than system/developer messages) is no longer sent. + See: https://learn.microsoft.com/azure/foundry/openai/how-to/responses#server-side-compaction""" + reasoning: ReasoningOptions """Configuration for reasoning models (gpt-5, o-series). See: https://platform.openai.com/docs/guides/reasoning""" @@ -1749,6 +1771,7 @@ def _prepare_messages_for_openai( """ reasoning_items: dict[str, dict[str, Any]] = {} if not request_uses_service_side_storage: + chat_messages = self._exclude_messages_before_latest_compaction(chat_messages) reasoning_items = self._prepare_reasoning_items_for_openai(chat_messages) self._validate_reasoning_groups_for_stateless_replay( chat_messages, @@ -1771,6 +1794,34 @@ def _prepare_messages_for_openai( # items (drop unmatched). See `_AF_MCP_PENDING_OUTPUT_KEY`. return self._coalesce_pending_mcp_results(flat) + @staticmethod + def _exclude_messages_before_latest_compaction(chat_messages: Sequence[Message]) -> Sequence[Message]: + """Mark the history that a service-side compaction item already summarizes as excluded. + + The latest compaction item with a replayable payload carries the context needed to + continue, so for stateless replay the messages before the one that contains it are marked + excluded rather than removed, matching how the built-in compaction strategies work. System + and developer messages are kept because they carry instructions rather than conversation + state. Whole messages are marked or kept so tool calls stay paired with their results. + Returns the messages that remain included. + """ + latest_index = next( + ( + index + for index in range(len(chat_messages) - 1, -1, -1) + if any( + content.type == "compaction" and content.protected_data for content in chat_messages[index].contents + ) + ), + None, + ) + if latest_index is None or latest_index == 0: + return chat_messages + for message in chat_messages[:latest_index]: + if message.role not in ("system", "developer"): + set_excluded(message, excluded=True, reason="superseded by service-side compaction") + return included_messages(list(chat_messages)) + def _validate_reasoning_groups_for_stateless_replay( self, chat_messages: Sequence[Message], @@ -1905,6 +1956,20 @@ def _prepare_message_for_openai( all_messages.append(reasoning_item) serialized_reasoning_ids.add(content.id) continue + case "compaction": + # With service-side storage the service already holds the compaction item. + if request_uses_service_side_storage or not content.protected_data: + continue + if "content" in args or "tool_calls" in args: + all_messages.append(args) + args = {"type": "message", "role": message.role} + compaction_item: dict[str, Any] = { + "type": "compaction", + "encrypted_content": content.protected_data, + } + if content.id: + compaction_item["id"] = content.id + all_messages.append(compaction_item) case "function_result": new_args: dict[str, Any] = {} new_args.update( @@ -3264,6 +3329,23 @@ def _get_finish_reason_from_openai_response(self, response: Any) -> FinishReason return FinishReason("tool_calls") return FinishReason("stop") + @staticmethod + def _parse_compaction_content(item: Any) -> Content | None: + """Convert a Responses API compaction item into compaction content.""" + encrypted_content = getattr(item, "encrypted_content", None) + if not isinstance(encrypted_content, str) or not encrypted_content: + logger.debug("Ignoring compaction item without encrypted content: %s", item) + return None + additional_properties: dict[str, Any] = {} + if created_by := getattr(item, "created_by", None): + additional_properties["created_by"] = created_by + return Content.from_compaction( + id=getattr(item, "id", None), + protected_data=encrypted_content, + additional_properties=additional_properties or None, + raw_representation=item, + ) + def _parse_response_from_openai( self, response: OpenAIResponse | ParsedResponse[BaseModel], @@ -3539,6 +3621,9 @@ def _parse_response_from_openai( contents.extend(self._image_generation_item_to_contents(item)) case "shell_call" | "local_shell_call" | "shell_call_output": contents.extend(self._shell_item_to_contents(item, local_shell_tool_name)) + case "compaction": # ResponseCompactionItem + if compaction_content := self._parse_compaction_content(item): + contents.append(compaction_content) case _: logger.debug("Unparsed output of type: %s: %s", item.type, item) self._mark_completed_computer_calls(contents) @@ -4123,6 +4208,9 @@ def parse_annotations( raw_representation=done_item, ) ) + elif getattr(done_item, "type", None) == "compaction": + if compaction_content := self._parse_compaction_content(done_item): + contents.append(compaction_content) elif done_item.type == "message": for content_index, part in enumerate(done_item.content): if part.type == "output_text": diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index bd576e61321..9509cbaced5 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -10974,3 +10974,219 @@ async def test_prepare_options_prompt_cache_options_guarded_on_old_openai(monkey # endregion + + +# region Server-side compaction + + +def _compaction_mock_response(response_id: str, output: list[Any]) -> MagicMock: + mock_response = MagicMock() + mock_response.output_parsed = None + mock_response.metadata = {} + mock_response.usage = None + mock_response.id = response_id + mock_response.model = "test-model" + mock_response.created_at = 1000000000 + mock_response.status = "completed" + mock_response.finish_reason = "stop" + mock_response.incomplete = None + mock_response.conversation = None + mock_response.output = output + return mock_response + + +def _compaction_text_item(text: str) -> MagicMock: + text_content = MagicMock() + text_content.type = "output_text" + text_content.text = text + text_content.annotations = [] + text_content.logprobs = None + text_item = MagicMock() + text_item.type = "message" + text_item.content = [text_content] + return text_item + + +def test_parse_response_preserves_compaction_item() -> None: + from openai.types.responses import ResponseCompactionItem + + client = OpenAIChatClient(model="test-model", api_key="test-key") + compaction_item = ResponseCompactionItem(id="cmp_1", type="compaction", encrypted_content="encrypted-state") + mock_response = _compaction_mock_response("resp-1", [compaction_item, _compaction_text_item("Done.")]) + + response = client._parse_response_from_openai(mock_response, options={}) # type: ignore[arg-type] + + compaction = response.messages[0].contents[0] + assert compaction.type == "compaction" + assert compaction.id == "cmp_1" + assert compaction.protected_data == "encrypted-state" + assert compaction.raw_representation is compaction_item + assert response.text == "Done." + + +def test_parse_response_ignores_compaction_item_without_encrypted_content() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + compaction_item = MagicMock() + compaction_item.type = "compaction" + compaction_item.id = "cmp_empty" + compaction_item.encrypted_content = None + mock_response = _compaction_mock_response("resp-1", [compaction_item, _compaction_text_item("Done.")]) + + response = client._parse_response_from_openai(mock_response, options={}) # type: ignore[arg-type] + + assert [content.type for content in response.messages[0].contents] == ["text"] + + +def test_streamed_compaction_item_is_preserved() -> None: + from openai.types.responses import ResponseCompactionItem + + client = OpenAIChatClient(model="test-model", api_key="test-key") + done_event = MagicMock() + done_event.type = "response.output_item.done" + done_event.item = ResponseCompactionItem( + id="cmp_stream", type="compaction", encrypted_content="encrypted-state", created_by="system" + ) + + update = client._parse_chunk_from_openai(done_event, {}, {}) + response = ChatResponse.from_updates([update]) + + compaction = response.messages[0].contents[0] + assert (compaction.type, compaction.id, compaction.protected_data) == ( + "compaction", + "cmp_stream", + "encrypted-state", + ) + assert compaction.additional_properties["created_by"] == "system" + + +async def test_stateless_request_replays_latest_compaction_and_drops_earlier_history() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + messages = [ + Message(role="system", contents=["You are helpful."]), + Message(role="user", contents=["First question"]), + Message( + role="assistant", + contents=[Content.from_compaction(id="cmp_old", protected_data="old-state"), "First answer"], + ), + Message(role="user", contents=["Second question"]), + Message( + role="assistant", + contents=[Content.from_compaction(id="cmp_new", protected_data="new-state"), "Second answer"], + ), + Message(role="user", contents=["Third question"]), + ] + + _, run_options, _ = await client._prepare_request(messages, {"store": False}) + + request_input = run_options["input"] + assert request_input[0]["role"] == "system" + assert request_input[1] == {"type": "compaction", "encrypted_content": "new-state", "id": "cmp_new"} + assert [item.get("role") for item in request_input[2:]] == ["assistant", "user"] + serialized = json.dumps(request_input) + assert "First question" not in serialized + assert "Second question" not in serialized + assert "old-state" not in serialized + # Earlier history is marked excluded rather than removed from the caller's messages. + assert [m.additional_properties.get("_excluded", False) for m in messages] == [ + False, + True, + True, + True, + False, + False, + ] + + +async def test_stateless_request_ignores_compaction_without_replayable_payload() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + messages = [ + Message(role="user", contents=["First question"]), + Message(role="assistant", contents=[Content.from_compaction(id="cmp_empty", protected_data=""), "Answer"]), + Message(role="user", contents=["Second question"]), + ] + + _, run_options, _ = await client._prepare_request(messages, {"store": False}) + + serialized = json.dumps(run_options["input"]) + assert "First question" in serialized + assert not any(m.additional_properties.get("_excluded", False) for m in messages) + + +async def test_request_with_service_side_storage_does_not_replay_or_prune_compaction() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + messages = [ + Message(role="user", contents=["First question"]), + Message(role="assistant", contents=[Content.from_compaction(id="cmp_1", protected_data="state"), "Answer"]), + Message(role="user", contents=["Second question"]), + ] + + _, run_options, _ = await client._prepare_request(messages, {"conversation_id": "resp_previous"}) + + assert all(item.get("type") != "compaction" for item in run_options["input"]) + assert "First question" in json.dumps(run_options["input"]) + + +async def test_compaction_content_round_trips_through_serialized_history() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + assistant = Message( + role="assistant", contents=[Content.from_compaction(id="cmp_1", protected_data="state"), "Answer"] + ) + restored = Message.from_dict(assistant.to_dict()) + + _, run_options, _ = await client._prepare_request( + [Message(role="user", contents=["Old"]), restored, Message(role="user", contents=["New"])], + {"store": False}, + ) + + assert run_options["input"][0] == {"type": "compaction", "encrypted_content": "state", "id": "cmp_1"} + assert "Old" not in json.dumps(run_options["input"]) + + +async def test_context_management_option_is_forwarded() -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + context_management = [{"type": "compaction", "compact_threshold": 200000}] + + _, run_options, _ = await client._prepare_request( + [Message(role="user", contents=["Hello"])], + {"store": False, "context_management": context_management}, + ) + + assert run_options["context_management"] == context_management + + +async def test_agent_with_history_replays_compaction_on_next_turn_when_store_false() -> None: + from openai.types.responses import ResponseCompactionItem + + client = OpenAIChatClient(model="test-model", api_key="test-key") + first = _compaction_mock_response( + "resp-1", + [ + ResponseCompactionItem(id="cmp_1", type="compaction", encrypted_content="compacted-state"), + _compaction_text_item("First answer"), + ], + ) + second = _compaction_mock_response("resp-2", [_compaction_text_item("Second answer")]) + default_options: OpenAIChatOptions = { + "store": False, + "context_management": [{"type": "compaction", "compact_threshold": 1000}], + } + agent = Agent( + client=client, + instructions="Be brief.", + context_providers=[InMemoryHistoryProvider()], + default_options=default_options, + ) + session = agent.create_session() + + with patch.object(client.client.responses, "create", side_effect=[_as_raw(first), _as_raw(second)]) as create: + await agent.run("First question", session=session) + await agent.run("Second question", session=session) + + second_input = create.call_args_list[1].kwargs["input"] + assert {"type": "compaction", "encrypted_content": "compacted-state", "id": "cmp_1"} in second_input + assert "First question" not in json.dumps(second_input) + assert "Second question" in json.dumps(second_input) + assert create.call_args_list[1].kwargs["context_management"] == [{"type": "compaction", "compact_threshold": 1000}] + + +# endregion