diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 895764eb93d..11fe82b46c9 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -165,6 +165,22 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID Shared serialization normalizes stored data; codecs and connector overrides remain trusted Python code - **`SupportsVectorUpsert`** / **`SupportsVectorSearch`** - Structural protocols for vector store capabilities +### Compaction (`_compaction.py`) + +- **`AgentTriggeredCompactionStrategy`** - Wraps one existing compaction strategy and invokes it only after its + explicitly registered trigger tool completes in the latest message group. Trigger consumption is stored on the + completed trigger call occurrence rather than on the strategy instance so shared agents remain safe across + concurrent runs and composed trigger tools do not suppress each other. The trigger group + is temporarily excluded while the inner strategy runs, so it does not consume retention counts, then restored by + default; set `exclude_trigger=True` to remove a trigger-only group from the next model call. Mixed groups that + contain unrelated tool calls are always restored atomically. The agent-triggered compaction APIs are experimental + under the shared `AGENT_TRIGGERED_COMPACTION` feature ID and require client-managed history (`store=False` with no + service-managed conversation ID). +- **`UnloadSkillsCompactionStrategy`** - Intended as an inner agent-triggered strategy. It supplies an + `unload_skill` trigger-tool factory that `AgentTriggeredCompactionStrategy` adopts automatically, then removes + complete `load_skill` and `read_skill_resource` groups for one named skill, or all skills when no name is + supplied. Mixed tool groups and `run_skill_script` results remain intact. + ### Middleware (`_middleware.py`) - **`AgentMiddleware`** - Intercepts agent `run()` calls diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index 5a0b6800ab5..3895cc5a51f 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -69,6 +69,7 @@ "SUMMARIZED_BY_SUMMARY_ID_KEY", "SUMMARY_OF_GROUP_IDS_KEY", "SUMMARY_OF_MESSAGE_IDS_KEY", + "AgentTriggeredCompactionStrategy", "CharacterEstimatorTokenizer", "CompactionProvider", "CompactionStrategy", @@ -80,6 +81,7 @@ "TokenizerProtocol", "ToolResultCompactionStrategy", "TruncationStrategy", + "UnloadSkillsCompactionStrategy", "annotate_message_groups", "apply_compaction", "included_messages", @@ -452,6 +454,7 @@ "AgentResponseUpdate", "AgentRunInputs", "AgentSession", + "AgentTriggeredCompactionStrategy", "AggregatingSkillsSource", "Annotation", "BackgroundAgentsProvider", @@ -645,6 +648,7 @@ "ToolTypes", "TruncationStrategy", "TypeCompatibilityError", + "UnloadSkillsCompactionStrategy", "UpdateT", "UsageDetails", "UserInputRequiredException", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index c5d5b0a250e..3eac3908a17 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -31,6 +31,7 @@ from ._compaction import ( SUMMARIZED_BY_SUMMARY_ID_KEY, SUMMARY_OF_GROUP_IDS_KEY, SUMMARY_OF_MESSAGE_IDS_KEY, + AgentTriggeredCompactionStrategy, CharacterEstimatorTokenizer, CompactionProvider, CompactionStrategy, @@ -42,6 +43,7 @@ from ._compaction import ( TokenizerProtocol, ToolResultCompactionStrategy, TruncationStrategy, + UnloadSkillsCompactionStrategy, annotate_message_groups, apply_compaction, included_messages, @@ -407,6 +409,7 @@ __all__ = [ "AgentResponseUpdate", "AgentRunInputs", "AgentSession", + "AgentTriggeredCompactionStrategy", "AggregatingSkillsSource", "Annotation", "BackgroundAgentsProvider", @@ -600,6 +603,7 @@ __all__ = [ "ToolTypes", "TruncationStrategy", "TypeCompatibilityError", + "UnloadSkillsCompactionStrategy", "UpdateT", "UsageDetails", "UserInputRequiredException", diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index 564d78aa660..1eed178ac18 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -38,6 +38,7 @@ ResponseStream, validate_chat_options, ) +from .exceptions import ChatClientInvalidRequestException if sys.version_info >= (3, 13): from typing import TypeVar # pragma: no cover @@ -288,6 +289,8 @@ async def _stream(): When ``True``, the agent skips auto-injecting ``InMemoryHistoryProvider`` unless the user explicitly sets ``store=False``. """ + SERVICE_CONTINUATION_OPTION_NAMES: ClassVar[tuple[str, ...]] = ("conversation_id",) + """Option names that activate provider-managed conversation continuation.""" # OTEL_PROVIDER_NAME is used for OTel setup, should be overridden in subclasses def __init__( @@ -373,8 +376,37 @@ async def _prepare_messages_for_model_call( *, compaction_strategy: CompactionStrategy | None = None, tokenizer: TokenizerProtocol | None = None, + options: Mapping[str, Any] | None = None, ) -> list[Message]: prepared_messages = list(messages) + if compaction_strategy is not None: + from ._compaction import _contains_agent_triggered_compaction_strategy # pyright: ignore[reportPrivateUsage] + from ._sessions import is_local_history_conversation_id + + if _contains_agent_triggered_compaction_strategy(compaction_strategy): + store = options.get("store") if options is not None else None + service_stores_history = ( + getattr(self, "STORES_BY_DEFAULT", False) if store is None else store is not False + ) + has_service_conversation = False + if options is not None: + for option_name in self.SERVICE_CONTINUATION_OPTION_NAMES: + value = options.get(option_name) + if value is None or value == "": + continue + if ( + option_name == "conversation_id" + and isinstance(value, str) + and is_local_history_conversation_id(value) + ): + continue + has_service_conversation = True + break + if service_stores_history or has_service_conversation: + raise ChatClientInvalidRequestException( + "Agent-triggered compaction requires client-managed history. " + "Set store=False and do not use a service-managed continuation option." + ) if compaction_strategy is None: if tokenizer is None: return prepared_messages @@ -531,6 +563,7 @@ async def _get_stream() -> ResponseStream[ChatResponseUpdate, ChatResponse[Any]] prepared_messages = await self._prepare_messages_for_model_call( messages, **compaction_overrides, + options=options, ) stream_response = self._inner_get_response( messages=prepared_messages, @@ -551,6 +584,7 @@ async def _get_response() -> ChatResponse[Any]: prepared_messages = await self._prepare_messages_for_model_call( messages, **compaction_overrides, + options=options, ) return await self._inner_get_response( messages=prepared_messages, diff --git a/python/packages/core/agent_framework/_compaction.py b/python/packages/core/agent_framework/_compaction.py index 6f4078eca35..64e37a4fbb2 100644 --- a/python/packages/core/agent_framework/_compaction.py +++ b/python/packages/core/agent_framework/_compaction.py @@ -16,12 +16,14 @@ runtime_checkable, ) +from ._feature_stage import ExperimentalFeature, experimental from ._sessions import ContextProvider from ._telemetry import FeatureIndex, mark_feature_used -from ._types import ChatResponse, Content, Message +from ._types import APPROVAL_REJECTION_RESULT_KEY, ChatResponse, Content, Message if TYPE_CHECKING: from ._clients import SupportsChatGetResponse + from ._tools import ApprovalMode, FunctionTool GroupKind: TypeAlias = Literal["system", "user", "assistant_text", "tool_call"] GROUP_ANNOTATION_KEY = "_group" @@ -41,6 +43,9 @@ SUMMARY_OF_MESSAGE_IDS_KEY = "_summary_of_message_ids" SUMMARY_OF_GROUP_IDS_KEY = "_summary_of_group_ids" SUMMARIZED_BY_SUMMARY_ID_KEY = "_summarized_by_summary_id" +_AGENT_COMPACTION_TRIGGER_CONSUMED_KEY = "_agent_compaction_trigger_consumed" +_SKILL_UNLOAD_SELECTION_KEY = "_skill_unload_selection" +_SKILL_CONTEXT_TOOL_NAMES: Final[frozenset[str]] = frozenset({"load_skill", "read_skill_resource"}) logger = logging.getLogger("agent_framework") @@ -52,6 +57,7 @@ "code_interpreter_tool_call", "shell_tool_call", "image_generation_tool_call", + "search_tool_call", } @@ -1180,6 +1186,511 @@ def _count_included_tokens(messages: list[Message]) -> int: return included_token_count(messages) +def _latest_group(messages: list[Message], *, include_excluded: bool) -> list[Message]: + grouped = _group_messages_by_id(messages) + for message in reversed(messages): + if not include_excluded and message.additional_properties.get(EXCLUDED_KEY, False): + continue + group_id = _group_id(message) + if group_id is not None: + return grouped.get(group_id, []) + return [] + + +def _latest_included_group(messages: list[Message]) -> list[Message]: + return _latest_group(messages, include_excluded=False) + + +def _function_call_result_pairs( + messages: Sequence[Message], + *, + include_excluded: bool = False, +) -> list[tuple[Content, Content]]: + candidate_messages = ( + list(messages) + if include_excluded + else [message for message in messages if not message.additional_properties.get(EXCLUDED_KEY, False)] + ) + open_calls_by_call_id: dict[str, list[Content]] = {} + open_calls_by_occurrence_id: dict[str, list[Content]] = {} + closed_call_identities: set[int] = set() + pairs: list[tuple[Content, Content]] = [] + + for message in candidate_messages: + for content in message.contents: + if content.type == "function_call" and content.call_id: + open_calls_by_call_id.setdefault(content.call_id, []).append(content) + if content.id: + open_calls_by_occurrence_id.setdefault(content.id, []).append(content) + continue + if content.type != "function_result" or not content.call_id: + continue + + matched_call: Content | None = None + if content.id is not None: + matched_call = next( + ( + call + for call in open_calls_by_occurrence_id.get(content.id, ()) + if id(call) not in closed_call_identities and call.call_id == content.call_id + ), + None, + ) + else: + matched_call = next( + ( + call + for call in open_calls_by_call_id.get(content.call_id, ()) + if id(call) not in closed_call_identities + ), + None, + ) + if matched_call is None: + continue + + closed_call_identities.add(id(matched_call)) + pairs.append((matched_call, content)) + + return pairs + + +def _completed_function_call_pairs( + messages: Sequence[Message], + *, + tool_name: str, + include_consumed: bool = False, + include_excluded: bool = False, +) -> list[tuple[Content, Content]]: + return [ + (call, result) + for call, result in _function_call_result_pairs(messages, include_excluded=include_excluded) + if ( + call.name == tool_name + and result.exception is None + and result.additional_properties.get(APPROVAL_REJECTION_RESULT_KEY) is not True + and (include_consumed or call.additional_properties.get(_AGENT_COMPACTION_TRIGGER_CONSUMED_KEY) is not True) + ) + ] + + +def _completed_function_calls( + messages: Sequence[Message], + *, + tool_name: str, + include_consumed: bool = False, + include_excluded: bool = False, +) -> list[Content]: + return [ + call + for call, _ in _completed_function_call_pairs( + messages, + tool_name=tool_name, + include_consumed=include_consumed, + include_excluded=include_excluded, + ) + ] + + +def _contains_agent_triggered_compaction_strategy( + strategy: object | None, + *, + _seen: set[int] | None = None, +) -> bool: + if strategy is None: + return False + seen: set[int] = _seen if _seen is not None else set() + strategy_id = id(strategy) + if strategy_id in seen: + return False + seen.add(strategy_id) + + if isinstance(strategy, AgentTriggeredCompactionStrategy): + return True + + nested_strategy: object | None = getattr(strategy, "strategy", None) + if ( + nested_strategy is not None + and nested_strategy is not strategy + and _contains_agent_triggered_compaction_strategy(nested_strategy, _seen=seen) + ): + return True + + nested_strategies: object = getattr(strategy, "strategies", None) + if isinstance(nested_strategies, Sequence) and not isinstance(nested_strategies, (str, bytes, bytearray)): + return any( + _contains_agent_triggered_compaction_strategy(nested, _seen=seen) + for nested in cast("Sequence[object]", nested_strategies) + ) + return False + + +@experimental(feature_id=ExperimentalFeature.AGENT_TRIGGERED_COMPACTION) +class AgentTriggeredCompactionStrategy: + """Run an existing compaction strategy after an agent invokes a trigger tool. + + The trigger tool must be registered explicitly on the agent alongside this + strategy. A completed trigger call in the latest included message group is + consumed once, then the wrapped strategy runs before the next model call. + Wrapped strategies may provide ``create_trigger_tool()`` to replace the + default ``compact_context`` tool with one specialized for their inputs. + + The consumed marker is stored on the completed trigger call occurrence + rather than on this strategy instance, so one strategy can be shared safely + by concurrent runs and composed trigger tools do not suppress each other. + + This strategy requires client-managed history (for example ``store=False``). + Chat clients reject it when service-side storage or a service-managed + conversation ID is active because local compaction cannot remove remote + conversation context. + """ + + DEFAULT_TOOL_NAME: Final[str] = "compact_context" + + def __init__( + self, + strategy: CompactionStrategy, + *, + tool_name: str = DEFAULT_TOOL_NAME, + approval_mode: ApprovalMode = "never_require", + exclude_trigger: bool = False, + ) -> None: + """Create an agent-triggered wrapper around one compaction strategy. + + Args: + strategy: Existing strategy to invoke after a completed trigger call. + + Keyword Args: + tool_name: Name for the default trigger tool when the wrapped + strategy does not provide ``create_trigger_tool()``. + approval_mode: Approval mode for the generated default trigger tool. + exclude_trigger: Whether to exclude the completed trigger call/result + group before the next model call. Defaults to False so the model + can observe that compaction completed. A trigger group containing + unrelated tool calls is always retained atomically. + """ + if not callable(strategy): + raise TypeError("strategy must be callable.") + if not tool_name.strip(): + raise ValueError("tool_name must not be empty.") + + trigger_tool_factory = getattr(strategy, "create_trigger_tool", None) + tool = ( + trigger_tool_factory() + if callable(trigger_tool_factory) + else self._create_default_tool(tool_name, approval_mode) + ) + from ._tools import FunctionTool + + if not isinstance(tool, FunctionTool): + raise TypeError("create_trigger_tool() must return a FunctionTool.") + if not tool.name.strip(): + raise ValueError("trigger tool name must not be empty.") + + self.strategy = strategy + self.tool = tool + self.exclude_trigger = exclude_trigger + + @property + def tokenizer(self) -> TokenizerProtocol | None: + """Return the wrapped strategy's tokenizer, when it exposes one.""" + tokenizer = getattr(self.strategy, "tokenizer", None) + return tokenizer if isinstance(tokenizer, TokenizerProtocol) else None + + @staticmethod + def _create_default_tool(tool_name: str, approval_mode: ApprovalMode) -> FunctionTool: + from ._tools import FunctionTool + + def request_compaction(reason: str | None = None) -> str: + """Request configured context compaction before the next model call.""" + message = "Compaction requested. The configured strategy will run before the next model call." + normalized_reason = reason.strip() if isinstance(reason, str) else "" + return f"{message} Reason: {normalized_reason}" if normalized_reason else message + + return FunctionTool( + name=tool_name, + description=( + "Request that the configured context compaction strategy run before the next model call. " + "Call this tool by itself, after any other tool calls whose results should be eligible for compaction; " + "do not call it in parallel with other tools." + ), + func=request_compaction, + approval_mode=approval_mode, + input_model={ + "type": "object", + "properties": { + "reason": { + "type": "string", + "description": "Why the agent is requesting compaction before the next model call.", + }, + }, + "additionalProperties": False, + }, + ) + + async def __call__(self, messages: list[Message]) -> bool: + """Run the wrapped strategy once for the latest completed trigger call.""" + annotate_message_groups(messages) + grouped = _group_messages_by_id(messages) + trigger_group = _latest_included_group(messages) + trigger_calls = _completed_function_calls( + trigger_group, + tool_name=self.tool.name, + ) + if not trigger_calls: + return False + + groups_to_hide: list[list[Message]] = [trigger_group] + trigger_group_ids = {id(message) for message in trigger_group} + for group in grouped.values(): + if any(id(message) in trigger_group_ids for message in group): + continue + if not any( + content.type == "function_call" + and content.name == self.tool.name + and content.additional_properties.get(_AGENT_COMPACTION_TRIGGER_CONSUMED_KEY) is True + for message in group + if not message.additional_properties.get(EXCLUDED_KEY, False) + for content in message.contents + ): + continue + if self._is_trigger_only_group(group): + groups_to_hide.append(group) + + hidden_messages = [message for group in groups_to_hide for message in group] + trigger_state = [ + ( + message, + EXCLUDED_KEY in message.additional_properties, + message.additional_properties.get(EXCLUDED_KEY), + EXCLUDE_REASON_KEY in message.additional_properties, + message.additional_properties.get(EXCLUDE_REASON_KEY), + ) + for message in hidden_messages + ] + for message in hidden_messages: + set_excluded(message, excluded=True, reason="agent_compaction_trigger") + + try: + changed = await self.strategy(messages) + except BaseException: + self._restore_trigger_group(trigger_state) + raise + + for trigger_call in trigger_calls: + trigger_call.additional_properties[_AGENT_COMPACTION_TRIGGER_CONSUMED_KEY] = True + + has_unrelated_tool_calls = not self._is_trigger_only_group(trigger_group, include_excluded=True) + trigger_message_ids = {id(message) for message in trigger_group} + has_other_included_context = any( + id(message) not in trigger_message_ids and not message.additional_properties.get(EXCLUDED_KEY, False) + for message in messages + ) + trigger_changed = self.exclude_trigger and not has_unrelated_tool_calls and has_other_included_context + prior_trigger_state = [state for state in trigger_state if id(state[0]) not in trigger_message_ids] + self._restore_trigger_group(prior_trigger_state) + if not trigger_changed: + current_trigger_state = [state for state in trigger_state if id(state[0]) in trigger_message_ids] + self._restore_trigger_group(current_trigger_state) + return changed or trigger_changed + + def _is_trigger_only_group( + self, + group: Sequence[Message], + *, + include_excluded: bool = False, + ) -> bool: + tool_calls = [ + content + for message in group + if include_excluded or not message.additional_properties.get(EXCLUDED_KEY, False) + for content in message.contents + if content.type in _TOOL_CALL_CONTENT_TYPES + ] + return bool(tool_calls) and all( + content.type == "function_call" and content.name == self.tool.name for content in tool_calls + ) + + @staticmethod + def _restore_trigger_group( + trigger_state: Sequence[tuple[Message, bool, Any, bool, Any]], + ) -> None: + for message, had_excluded, excluded, had_reason, reason in trigger_state: + if had_excluded: + message.additional_properties[EXCLUDED_KEY] = excluded + else: + message.additional_properties.pop(EXCLUDED_KEY, None) + if had_reason: + message.additional_properties[EXCLUDE_REASON_KEY] = reason + else: + message.additional_properties.pop(EXCLUDE_REASON_KEY, None) + + +@experimental(feature_id=ExperimentalFeature.AGENT_TRIGGERED_COMPACTION) +class UnloadSkillsCompactionStrategy: + """Exclude loaded skill bodies and resources selected by an unload tool. + + This strategy targets complete tool-call groups containing only + ``load_skill`` and ``read_skill_resource`` calls. The latest completed + ``unload_skill`` request selects one skill by name, or all skills when its + ``skill_name`` argument is omitted or blank. + + Use this as the wrapped strategy of :class:`AgentTriggeredCompactionStrategy`. + The wrapper obtains the specialized ``unload_skill`` tool through + :meth:`create_trigger_tool` and exposes that single tool to the caller. + """ + + DEFAULT_TOOL_NAME: Final[str] = "unload_skill" + + def __init__( + self, + *, + tool_name: str = DEFAULT_TOOL_NAME, + approval_mode: ApprovalMode = "never_require", + ) -> None: + """Create a skill-unloading strategy and its trigger tool. + + Keyword Args: + tool_name: Name of the tool that requests skill unloading. + approval_mode: Approval mode for the generated unload tool. + """ + if not tool_name.strip(): + raise ValueError("tool_name must not be empty.") + self.tool_name: str = tool_name + self.approval_mode: ApprovalMode = approval_mode + + def create_trigger_tool(self) -> FunctionTool: + """Create the ``unload_skill`` trigger tool used by the wrapper.""" + from ._tools import FunctionTool + + def request_skill_unload(skill_name: str | Sequence[str] | None = None) -> list[Content]: + """Unload the full instructions and resources for selected skills.""" + if skill_name is None: + requested_names: Sequence[str] = [] + elif isinstance(skill_name, str): + requested_names = [skill_name] + elif isinstance(skill_name, Sequence) and not isinstance(skill_name, (bytes, bytearray)): + if any(not isinstance(name, str) for name in skill_name): + raise TypeError("skill_name must contain only strings.") + requested_names = skill_name + else: + raise TypeError("skill_name must be a string, a sequence of strings, or None.") + normalized_names = [name.strip() for name in requested_names if name.strip()] + if normalized_names: + formatted_names = ", ".join(f"'{name}'" for name in normalized_names) + message = f"Skill(s) {formatted_names} will be unloaded before the next model call." + else: + message = "All loaded skills will be unloaded before the next model call." + return [ + Content.from_text( + message, + additional_properties={_SKILL_UNLOAD_SELECTION_KEY: normalized_names}, + ) + ] + + return FunctionTool( + name=self.tool_name, + description=( + "Unloads the full instructions and resources for one or more loaded skills. " + "Omit skill_name to unload all loaded skills. Call this tool by itself after the relevant skill work " + "is complete; do not call it in parallel with other tools." + ), + func=request_skill_unload, + approval_mode=self.approval_mode, + input_model={ + "type": "object", + "properties": { + "skill_name": { + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}}, + ], + "description": ( + "The name of a skill to unload, a list of skill names to unload, " + "or omit to unload all loaded skills." + ), + }, + }, + "additionalProperties": False, + }, + ) + + def _requested_skill_names(self, messages: list[Message]) -> set[str] | None: + requests = _completed_function_call_pairs( + _latest_group(messages, include_excluded=True), + tool_name=self.tool_name, + include_consumed=True, + include_excluded=True, + ) + if not requests: + return set() + + requested_names: set[str] = set() + for _, result in requests: + selections = [ + item.additional_properties.get(_SKILL_UNLOAD_SELECTION_KEY) + for item in (result.items or []) + if item.type == "text" + ] + if len(selections) != 1 or not isinstance(selections[0], list): + return set() + skill_names = cast("list[Any]", selections[0]) + if any(not isinstance(skill_name, str) for skill_name in skill_names): + return set() + normalized_names = {skill_name.strip().casefold() for skill_name in skill_names if skill_name.strip()} + if not normalized_names: + return None + requested_names.update(normalized_names) + return requested_names + + @staticmethod + def _skill_names_for_removable_group(group: Sequence[Message]) -> set[str] | None: + included_messages = [message for message in group if not message.additional_properties.get(EXCLUDED_KEY, False)] + tool_calls = [ + content + for message in included_messages + for content in message.contents + if content.type in _TOOL_CALL_CONTENT_TYPES + ] + if not tool_calls or any( + call.type != "function_call" or call.name not in _SKILL_CONTEXT_TOOL_NAMES for call in tool_calls + ): + return None + calls = tool_calls + + completed_call_identities = {id(call) for call, _ in _function_call_result_pairs(included_messages)} + if any(id(call) not in completed_call_identities for call in calls): + return None + + skill_names: set[str] = set() + for call in calls: + arguments = call.parse_arguments() or {} + skill_name = arguments.get("skill_name") + if not isinstance(skill_name, str) or not skill_name.strip(): + return None + skill_names.add(skill_name.strip().casefold()) + return skill_names + + async def __call__(self, messages: list[Message]) -> bool: + """Exclude matching loaded skill context groups.""" + ordered_group_ids = annotate_message_groups(messages) + requested_skill_names = self._requested_skill_names(messages) + if requested_skill_names == set(): + return False + + grouped = _group_messages_by_id(messages) + changed = False + for group_id in ordered_group_ids: + group = grouped.get(group_id, []) + skill_names = self._skill_names_for_removable_group(group) + if skill_names is None: + continue + if requested_skill_names is not None and not skill_names.issubset(requested_skill_names): + continue + for message in group: + changed = set_excluded(message, excluded=True, reason="skill_unload") or changed + return changed + + class TruncationStrategy: """Oldest-first compaction using a single metric threshold. @@ -1429,7 +1940,11 @@ async def __call__(self, messages: list[Message]) -> bool: for content in msg.contents: if content.type == "function_call" and content.call_id and content.name: call_id_to_name[content.call_id] = content.name - elif content.type == "mcp_server_tool_call" and content.call_id and content.tool_name: + elif ( + content.type in ("mcp_server_tool_call", "search_tool_call") + and content.call_id + and content.tool_name + ): call_id_to_name[content.call_id] = content.tool_name # Collect tool results with the tool name for context. @@ -1448,6 +1963,11 @@ async def __call__(self, messages: list[Message]) -> bool: tool_name = call_id_to_name.get(content.call_id or "", "") label = f"{tool_name}: {result_text}" if tool_name else result_text tool_results.append(label.strip()) + elif content.type == "search_tool_result": + result_text = _tool_result_text(content.result if content.result is not None else content.items) + tool_name = call_id_to_name.get(content.call_id or "", "") + label = f"{tool_name}: {result_text}" if tool_name else result_text + tool_results.append(label.strip()) summary_text = self._summary_text(tool_results) summary_id = f"tool_summary_{group_id}" @@ -2369,6 +2889,7 @@ async def __call__(self, messages: list[Message]) -> bool: "SUMMARIZED_BY_SUMMARY_ID_KEY", "SUMMARY_OF_GROUP_IDS_KEY", "SUMMARY_OF_MESSAGE_IDS_KEY", + "AgentTriggeredCompactionStrategy", "CharacterEstimatorTokenizer", "CompactionProvider", "CompactionStrategy", @@ -2381,6 +2902,7 @@ async def __call__(self, messages: list[Message]) -> bool: "TokenizerProtocol", "ToolResultCompactionStrategy", "TruncationStrategy", + "UnloadSkillsCompactionStrategy", "annotate_message_groups", "annotate_token_counts", "append_compaction_message", diff --git a/python/packages/core/agent_framework/_feature_stage.py b/python/packages/core/agent_framework/_feature_stage.py index 9421c12cc27..b4988db3c19 100644 --- a/python/packages/core/agent_framework/_feature_stage.py +++ b/python/packages/core/agent_framework/_feature_stage.py @@ -50,6 +50,7 @@ class ExperimentalFeature(str, Enum): on enum membership or attribute presence over time. """ + AGENT_TRIGGERED_COMPACTION = "AGENT_TRIGGERED_COMPACTION" AGENT_HOOKS = "AGENT_HOOKS" COMPUTER_USE = "COMPUTER_USE" DECLARATIVE_AGENTS = "DECLARATIVE_AGENTS" diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 08f00fefc11..53b64bce9d3 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -416,9 +416,7 @@ def _parse_inputs( # pyright: ignore[reportUnusedFunction] if inputs is None: return [] - from ._types import ( - Content, - ) + from ._types import Content parsed_inputs: list[Content] = [] if not isinstance(inputs, list): @@ -4449,9 +4447,7 @@ def _replace_approval_contents_with_results( Returns: The terminal contents produced while resolving the approval responses, in response order. """ - from ._types import ( - Content, - ) + from ._types import APPROVAL_REJECTION_RESULT_KEY, Content result_groups_by_call_id: dict[str, deque[list[Content]]] = {} for result_group in approved_function_result_groups: @@ -4578,7 +4574,10 @@ def find_approval_occurrence(approval_id: str) -> _ApprovalCallOccurrence | None call_id=call_id, id=occurrence.function_call.id if occurrence is not None else content.function_call.id, result="Error: Tool call invocation was rejected by user.", - additional_properties=content.function_call.additional_properties, + additional_properties={ + **content.function_call.additional_properties, + APPROVAL_REJECTION_RESULT_KEY: True, + }, ) ] if not replacements: diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 629593b8839..0909e4fd173 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -459,6 +459,7 @@ def _restore_compaction_annotation_in_additional_properties( "code_interpreter_tool_call", "code_interpreter_tool_result", }) +APPROVAL_REJECTION_RESULT_KEY = "_approval_rejection_result" _CONTENT_TYPES: Final[frozenset[str]] = frozenset(get_args(ContentType)) diff --git a/python/packages/core/tests/core/test_agents.py b/python/packages/core/tests/core/test_agents.py index 1481ce8a77b..a3703420a80 100644 --- a/python/packages/core/tests/core/test_agents.py +++ b/python/packages/core/tests/core/test_agents.py @@ -21,6 +21,7 @@ AgentResponse, AgentResponseUpdate, AgentSession, + AgentTriggeredCompactionStrategy, ChatContext, ChatOptions, ChatResponse, @@ -28,6 +29,7 @@ Content, ContextProvider, Embedding, + FunctionMiddleware, FunctionTool, GeneratedEmbeddings, HistoryProvider, @@ -45,6 +47,7 @@ ToolApprovalMiddleware, ToolResultCompactionStrategy, TruncationStrategy, + UnloadSkillsCompactionStrategy, VectorStoreHistoryProvider, chat_middleware, enqueue_messages, @@ -3446,6 +3449,155 @@ async def capturing_inner( assert captured_token_counts == [[9, 9]] +@pytest.mark.parametrize("stream", [False, True]) +async def test_agent_triggered_compaction_applies_before_next_model_call( + chat_client_base: MockBaseChatClient, + stream: bool, +) -> None: + agent_compaction = AgentTriggeredCompactionStrategy( + TruncationStrategy(max_n=2, compact_to=2, preserve_first_user_group=True), + ) + function_call = Content.from_function_call( + call_id="compact-1", + name=agent_compaction.tool.name, + arguments={"reason": "planning complete"}, + ) + if stream: + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(contents=[function_call], role="assistant", finish_reason="tool_calls")], + [ChatResponseUpdate(contents=[Content.from_text("done")], role="assistant", finish_reason="stop")], + ] + else: + chat_client_base.run_responses = [ + ChatResponse(messages=Message(role="assistant", contents=[function_call])), + ChatResponse(messages=Message(role="assistant", contents=["done"])), + ] + + captured_roles: list[list[str]] = [] + captured_texts: list[list[str | None]] = [] + original_inner = chat_client_base._inner_get_response + + def capture_inner( + *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any + ) -> Any: + captured_roles.append([message.role for message in messages]) + captured_texts.append([message.text for message in messages]) + return original_inner(messages=messages, stream=stream, options=options, **kwargs) + + agent = Agent( + client=chat_client_base, + tools=[agent_compaction.tool], + compaction_strategy=agent_compaction, + ) + + with patch.object(chat_client_base, "_inner_get_response", side_effect=capture_inner): + run_messages = [ + Message(role="user", contents=["original task"]), + Message(role="assistant", contents=["obsolete planning detail"]), + Message(role="user", contents=["plan"]), + ] + if stream: + result_text = "".join([update.text or "" async for update in agent.run(run_messages, stream=True)]) + else: + result_text = (await agent.run(run_messages)).text + + assert result_text == "done" + assert captured_roles == [ + ["user", "assistant", "user"], + ["user", "user", "assistant", "tool"], + ] + assert "obsolete planning detail" in captured_texts[0] + assert "obsolete planning detail" not in captured_texts[1] + + +async def test_unload_skills_uses_middleware_rewritten_selection( + chat_client_base: MockBaseChatClient, +) -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + + class RewriteUnloadSelectionMiddleware(FunctionMiddleware): + async def process(self, context: FunctionInvocationContext, call_next: Any) -> None: + if context.function.name == strategy.tool.name: + context.arguments = {"skill_name": "excel"} + await call_next() + + chat_client_base.run_responses = [ + ChatResponse( + messages=Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="unload-1", + name=strategy.tool.name, + arguments={"skill_name": "pdf"}, + ) + ], + ) + ), + ChatResponse(messages=Message(role="assistant", contents=["done"])), + ] + + captured_calls: list[list[tuple[str | None, Any]]] = [] + original_inner = chat_client_base._inner_get_response + + def capture_inner( + *, messages: MutableSequence[Message], stream: bool, options: dict[str, Any], **kwargs: Any + ) -> Any: + captured_calls.append([ + (content.name, content.parse_arguments()) + for message in messages + for content in message.contents + if content.type == "function_call" + ]) + return original_inner(messages=messages, stream=stream, options=options, **kwargs) + + agent = Agent( + client=chat_client_base, + tools=[strategy.tool], + compaction_strategy=strategy, + middleware=[RewriteUnloadSelectionMiddleware()], + ) + run_messages = [ + Message(role="user", contents=["work"]), + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="load-pdf", + name="load_skill", + arguments={"skill_name": "pdf"}, + ) + ], + ), + Message( + role="tool", + contents=[Content.from_function_result(call_id="load-pdf", result="pdf instructions")], + ), + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="load-excel", + name="load_skill", + arguments={"skill_name": "excel"}, + ) + ], + ), + Message( + role="tool", + contents=[Content.from_function_result(call_id="load-excel", result="excel instructions")], + ), + Message(role="user", contents=["unload the selected skill"]), + ] + + with patch.object(chat_client_base, "_inner_get_response", side_effect=capture_inner): + result = await agent.run(run_messages) + + assert result.text == "done" + assert ("load_skill", {"skill_name": "pdf"}) in captured_calls[1] + assert ("load_skill", {"skill_name": "excel"}) not in captured_calls[1] + + async def test_chat_agent_uses_client_compaction_defaults_when_agent_unset(chat_client_base: Any) -> None: captured_roles: list[list[str]] = [] original_inner = chat_client_base._inner_get_response diff --git a/python/packages/core/tests/core/test_clients.py b/python/packages/core/tests/core/test_clients.py index b122b9c5ebf..5ee1af25799 100644 --- a/python/packages/core/tests/core/test_clients.py +++ b/python/packages/core/tests/core/test_clients.py @@ -14,8 +14,10 @@ GROUP_TOKEN_COUNT_KEY, SUMMARIZED_BY_SUMMARY_ID_KEY, SUMMARY_OF_MESSAGE_IDS_KEY, + AgentTriggeredCompactionStrategy, BaseChatClient, ChatMiddleware, + ChatOptions, ChatResponse, ChatResponseUpdate, Content, @@ -30,6 +32,8 @@ included_messages, tool, ) +from agent_framework._sessions import LOCAL_HISTORY_CONVERSATION_ID +from agent_framework.exceptions import ChatClientInvalidRequestException class _NoOpChatMiddleware(ChatMiddleware): @@ -247,6 +251,103 @@ async def _capture( assert captured_roles == [["assistant"]] +@pytest.mark.parametrize("stream", [False, True]) +async def test_agent_triggered_compaction_rejects_default_service_storage( + chat_client_base: SupportsChatGetResponse, + stream: bool, +) -> None: + chat_client_base.function_invocation_configuration["enabled"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + strategy = AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1)) + + with pytest.raises(ChatClientInvalidRequestException, match="store=False"): + if stream: + response_stream = chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + stream=True, + compaction_strategy=strategy, + ) + async for _ in response_stream: + pass + else: + response = chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + stream=False, + compaction_strategy=strategy, + ) + _ = await response + + +async def test_agent_triggered_compaction_rejects_explicit_store_true( + chat_client_base: SupportsChatGetResponse, +) -> None: + chat_client_base.function_invocation_configuration["enabled"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + strategy = AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1)) + + with pytest.raises(ChatClientInvalidRequestException, match="store=False"): + await chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + options=ChatOptions(store=True), + compaction_strategy=strategy, + ) + + +async def test_nested_agent_triggered_compaction_rejects_service_storage( + chat_client_base: SupportsChatGetResponse, +) -> None: + class StrategyWrapper: + def __init__(self, strategy: Any) -> None: + self.strategy = strategy + + async def __call__(self, messages: list[Message]) -> bool: + return await self.strategy(messages) + + chat_client_base.function_invocation_configuration["enabled"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + strategy = StrategyWrapper(AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1))) + + with pytest.raises(ChatClientInvalidRequestException, match="store=False"): + await chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + compaction_strategy=strategy, + ) + + +async def test_agent_triggered_compaction_rejects_service_conversation_id( + chat_client_base: SupportsChatGetResponse, +) -> None: + chat_client_base.function_invocation_configuration["enabled"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + strategy = AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1)) + + with pytest.raises(ChatClientInvalidRequestException, match="service-managed continuation"): + await chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + options=ChatOptions(store=False, conversation_id="service-conversation"), + compaction_strategy=strategy, + ) + + +@pytest.mark.parametrize("conversation_id", [None, LOCAL_HISTORY_CONVERSATION_ID]) +async def test_agent_triggered_compaction_allows_client_managed_history( + chat_client_base: SupportsChatGetResponse, + conversation_id: str | None, +) -> None: + chat_client_base.function_invocation_configuration["enabled"] = False # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + chat_client_base.STORES_BY_DEFAULT = True # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + strategy = AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1)) + options = ChatOptions(store=False) + if conversation_id is not None: + options["conversation_id"] = conversation_id + + response = await chat_client_base.get_response( + [Message(role="user", contents=["Hello"])], + options=options, + compaction_strategy=strategy, + ) + + assert response.text == "test response - Hello" + + async def test_base_client_per_call_tokenizer_override_annotates_messages( chat_client_base: SupportsChatGetResponse, ) -> None: diff --git a/python/packages/core/tests/core/test_compaction.py b/python/packages/core/tests/core/test_compaction.py index 7c5ed25de54..4e55f07fb34 100644 --- a/python/packages/core/tests/core/test_compaction.py +++ b/python/packages/core/tests/core/test_compaction.py @@ -19,6 +19,7 @@ SUMMARIZED_BY_SUMMARY_ID_KEY, SUMMARY_OF_GROUP_IDS_KEY, SUMMARY_OF_MESSAGE_IDS_KEY, + AgentTriggeredCompactionStrategy, CharacterEstimatorTokenizer, ChatResponse, CompactionProvider, @@ -31,18 +32,23 @@ TokenBudgetComposedStrategy, ToolResultCompactionStrategy, TruncationStrategy, + UnloadSkillsCompactionStrategy, annotate_message_groups, apply_compaction, included_messages, included_token_count, ) from agent_framework._compaction import ( + _SKILL_UNLOAD_SELECTION_KEY, + _completed_function_calls, _format_summary_message, _select_summary_input_groups, _serialize_message, append_compaction_message, extend_compaction_messages, ) +from agent_framework._feature_stage import _WARNED_FEATURES, ExperimentalFeature, ExperimentalWarning +from agent_framework._types import APPROVAL_REJECTION_RESULT_KEY def _assistant_function_call(call_id: str) -> Message: @@ -52,6 +58,17 @@ def _assistant_function_call(call_id: str) -> Message: ) +def _assistant_named_function_call( + call_id: str, + name: str, + arguments: dict[str, Any] | None = None, +) -> Message: + return Message( + role="assistant", + contents=[Content.from_function_call(call_id=call_id, name=name, arguments=arguments or {})], + ) + + def _assistant_function_call_with_result(call_id: str, result: str) -> Message: return Message( role="assistant", @@ -112,6 +129,23 @@ def _tool_result(call_id: str, result: str, message_id: str | None = None) -> Me ) +def _skill_unload_result(call_id: str, skill_names: list[str]) -> Message: + return Message( + role="tool", + contents=[ + Content.from_function_result( + call_id=call_id, + result=[ + Content.from_text( + "Skills will be unloaded before the next model call.", + additional_properties={_SKILL_UNLOAD_SELECTION_KEY: skill_names}, + ) + ], + ) + ], + ) + + def _group_id(message: Message) -> str | None: annotation = message.additional_properties.get(GROUP_ANNOTATION_KEY) if not isinstance(annotation, dict): @@ -848,6 +882,612 @@ def test_truncation_strategy_validates_token_targets() -> None: raise AssertionError("Expected ValueError when compact_to is greater than max_n.") +class _RecordingCompactionStrategy: + def __init__(self, *, changed: bool = True) -> None: + self.call_count = 0 + self.changed = changed + + async def __call__(self, messages: list[Message]) -> bool: + self.call_count += 1 + return self.changed + + +def test_agent_triggered_compaction_apis_are_experimental() -> None: + feature_id = ExperimentalFeature.AGENT_TRIGGERED_COMPACTION.value + for api in (AgentTriggeredCompactionStrategy, UnloadSkillsCompactionStrategy): + assert getattr(api, "__feature_stage__", None) == "experimental" + assert getattr(api, "__feature_id__", None) == feature_id + assert ".. warning:: Experimental" in (api.__doc__ or "") + + _WARNED_FEATURES.discard((ExperimentalWarning, feature_id)) + with pytest.warns(ExperimentalWarning, match=feature_id): + AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + + +async def test_agent_triggered_compaction_runs_only_after_completed_trigger() -> None: + inner = _RecordingCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(inner) + messages = [ + Message(role="user", contents=["plan"]), + _assistant_named_function_call("compact-1", strategy.tool.name, {"reason": "planning complete"}), + ] + + assert await strategy(messages) is False + assert inner.call_count == 0 + + messages.append(_tool_result("compact-1", "Compaction requested.")) + + assert await strategy(messages) is True + assert inner.call_count == 1 + assert messages[1].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is not True + assert await strategy(messages) is False + assert inner.call_count == 1 + + +async def test_agent_triggered_compaction_consumes_request_when_inner_strategy_is_noop() -> None: + inner = _RecordingCompactionStrategy(changed=False) + strategy = AgentTriggeredCompactionStrategy(inner) + messages = [ + _assistant_named_function_call("compact-1", strategy.tool.name), + _tool_result("compact-1", "Compaction requested."), + ] + + assert await strategy(messages) is False + assert await strategy(messages) is False + assert inner.call_count == 1 + + +async def test_agent_triggered_compaction_can_exclude_trigger_group() -> None: + inner = _RecordingCompactionStrategy(changed=False) + strategy = AgentTriggeredCompactionStrategy(inner, exclude_trigger=True) + messages = [ + Message(role="user", contents=["keep working"]), + _assistant_named_function_call("compact-1", strategy.tool.name), + _tool_result("compact-1", "Compaction requested."), + ] + + assert await strategy(messages) is True + assert messages[0].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[1].additional_properties.get(EXCLUDED_KEY) is True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is True + + +async def test_agent_trigger_does_not_count_toward_tool_group_retention() -> None: + strategy = AgentTriggeredCompactionStrategy(SelectiveToolCallCompactionStrategy(keep_last_tool_call_groups=1)) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("work-1", "work_tool"), + _tool_result("work-1", "work result"), + _assistant_named_function_call("compact-1", strategy.tool.name), + _tool_result("compact-1", "Compaction requested."), + ] + + assert await strategy(messages) is False + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages) + + +@pytest.mark.parametrize("trigger_position", [0, 1, 2]) +@pytest.mark.parametrize("exclude_trigger", [False, True]) +async def test_agent_trigger_in_mixed_tool_batch_is_position_independent_and_retained( + trigger_position: int, + exclude_trigger: bool, +) -> None: + strategy = AgentTriggeredCompactionStrategy( + SelectiveToolCallCompactionStrategy(keep_last_tool_call_groups=0), + exclude_trigger=exclude_trigger, + ) + batch_calls = [ + Content.from_function_call(call_id="work-a", name="work_a", arguments={}), + Content.from_function_call(call_id="work-b", name="work_b", arguments={}), + ] + batch_calls.insert( + trigger_position, + Content.from_function_call(call_id="compact-1", name=strategy.tool.name, arguments={}), + ) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("old-work", "old_work"), + _tool_result("old-work", "old result"), + Message(role="assistant", contents=batch_calls), + Message( + role="tool", + contents=[ + Content.from_function_result(call_id=call.call_id or "", result=f"{call.name} result") + for call in batch_calls + ], + ), + ] + + changed = await strategy(messages) + + assert changed is True + assert messages[1].additional_properties.get(EXCLUDED_KEY) is True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is True + assert messages[3].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[4].additional_properties.get(EXCLUDED_KEY) is not True + + +async def test_agent_trigger_in_mixed_search_batch_is_retained() -> None: + strategy = AgentTriggeredCompactionStrategy( + SelectiveToolCallCompactionStrategy(keep_last_tool_call_groups=0), + exclude_trigger=True, + ) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("old-work", "old_work"), + _tool_result("old-work", "old result"), + Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="compact-1", name=strategy.tool.name, arguments={}), + Content.from_search_tool_call( + call_id="search-1", + tool_name="web_search", + arguments={"query": "latest"}, + ), + ], + ), + Message( + role="tool", + contents=[ + Content.from_function_result(call_id="compact-1", result="Compaction requested."), + Content.from_search_tool_result( + call_id="search-1", + tool_name="web_search", + result="search result", + ), + ], + ), + ] + + changed = await strategy(messages) + + assert changed is True + assert messages[1].additional_properties.get(EXCLUDED_KEY) is True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is True + assert messages[3].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[4].additional_properties.get(EXCLUDED_KEY) is not True + + +async def test_successive_triggers_do_not_consume_tool_group_retention() -> None: + strategy = AgentTriggeredCompactionStrategy(SelectiveToolCallCompactionStrategy(keep_last_tool_call_groups=2)) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("work-1", "work_tool"), + _tool_result("work-1", "work result 1"), + _assistant_named_function_call("work-2", "work_tool"), + _tool_result("work-2", "work result 2"), + _assistant_named_function_call("compact-1", strategy.tool.name), + _tool_result("compact-1", "Compaction requested."), + ] + + assert await strategy(messages) is False + + messages.extend([ + _assistant_named_function_call("work-3", "work_tool"), + _tool_result("work-3", "work result 3"), + _assistant_named_function_call("compact-2", strategy.tool.name), + _tool_result("compact-2", "Compaction requested."), + ]) + + assert await strategy(messages) is True + assert messages[1].additional_properties.get(EXCLUDED_KEY) is True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is True + assert messages[3].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[4].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[5].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[6].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[7].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[8].additional_properties.get(EXCLUDED_KEY) is not True + + +async def test_agent_triggered_compaction_restores_request_when_inner_strategy_fails() -> None: + class _FailingStrategy: + async def __call__(self, messages: list[Message]) -> bool: + raise RuntimeError("compaction failed") + + strategy = AgentTriggeredCompactionStrategy(_FailingStrategy()) + trigger_message = _assistant_named_function_call("compact-1", strategy.tool.name) + messages = [trigger_message, _tool_result("compact-1", "Compaction requested.")] + + with pytest.raises(RuntimeError, match="compaction failed"): + await strategy(messages) + + assert "_agent_compaction_trigger_consumed" not in trigger_message.contents[0].additional_properties + + +@pytest.mark.parametrize("reverse_order", [False, True]) +async def test_composed_trigger_tools_each_execute_once(reverse_order: bool) -> None: + first_inner = _RecordingCompactionStrategy(changed=False) + second_inner = _RecordingCompactionStrategy(changed=False) + first = AgentTriggeredCompactionStrategy(first_inner, tool_name="compact_first") + second = AgentTriggeredCompactionStrategy(second_inner, tool_name="compact_second") + messages = [ + Message(role="user", contents=["work"]), + Message( + role="assistant", + contents=[ + Content.from_function_call(call_id="first", name=first.tool.name, arguments={}), + Content.from_function_call(call_id="second", name=second.tool.name, arguments={}), + ], + ), + Message( + role="tool", + contents=[ + Content.from_function_result(call_id="first", result="First compaction requested."), + Content.from_function_result(call_id="second", result="Second compaction requested."), + ], + ), + ] + strategies = [second, first] if reverse_order else [first, second] + + assert await strategies[0](messages) is False + assert await strategies[1](messages) is False + assert first_inner.call_count == 1 + assert second_inner.call_count == 1 + assert await first(messages) is False + assert await second(messages) is False + assert first_inner.call_count == 1 + assert second_inner.call_count == 1 + + +async def test_agent_triggered_compaction_ignores_trigger_outside_latest_group() -> None: + inner = _RecordingCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(inner) + messages = [ + _assistant_named_function_call("compact-1", strategy.tool.name), + _tool_result("compact-1", "Compaction requested."), + Message(role="assistant", contents=["continued"]), + ] + + assert await strategy(messages) is False + assert inner.call_count == 0 + + +async def test_agent_triggered_compaction_uses_terminal_result_for_nonadjacent_group_recency() -> None: + inner = _RecordingCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(inner) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("compact-1", strategy.tool.name), + Message(role="assistant", contents=["intervening context"]), + _tool_result("compact-1", "Compaction requested."), + ] + + assert await strategy(messages) is True + assert inner.call_count == 1 + + +async def test_agent_triggered_compaction_ignores_rejected_approval_result() -> None: + inner = _RecordingCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(inner) + rejection_result = Content.from_function_result( + call_id="compact-1", + result="Error: Tool call invocation was rejected by user.", + additional_properties={APPROVAL_REJECTION_RESULT_KEY: True}, + ) + messages = [ + _assistant_named_function_call("compact-1", strategy.tool.name), + Message(role="tool", contents=[rejection_result]), + ] + + assert await strategy(messages) is False + assert inner.call_count == 0 + + +def test_completed_function_calls_correlates_reused_call_id_by_occurrence() -> None: + rejected_call = Content.from_function_call( + id="occ-rejected", + call_id="shared", + name="compact_context", + arguments={"reason": "rejected"}, + ) + approved_call = Content.from_function_call( + id="occ-approved", + call_id="shared", + name="compact_context", + arguments={"reason": "approved"}, + ) + unresolved_call = Content.from_function_call( + id="occ-unresolved", + call_id="shared", + name="compact_context", + arguments={"reason": "unresolved"}, + ) + messages = [ + Message(role="assistant", contents=[rejected_call, approved_call, unresolved_call]), + Message( + role="tool", + contents=[ + Content.from_function_result( + id="occ-rejected", + call_id="shared", + result="Error: Tool call invocation was rejected by user.", + additional_properties={APPROVAL_REJECTION_RESULT_KEY: True}, + ), + Content.from_function_result( + id="occ-approved", + call_id="shared", + result="Compaction requested.", + ), + ], + ), + ] + + completed = _completed_function_calls(messages, tool_name="compact_context") + + assert completed == [approved_call] + + +def test_completed_function_calls_uses_fifo_for_idless_reused_call_id() -> None: + first_call = Content.from_function_call(call_id="shared", name="compact_context", arguments={}) + second_call = Content.from_function_call(call_id="shared", name="compact_context", arguments={}) + messages = [ + Message(role="assistant", contents=[first_call, second_call]), + Message( + role="tool", + contents=[Content.from_function_result(call_id="shared", result="Compaction requested.")], + ), + ] + + completed = _completed_function_calls(messages, tool_name="compact_context") + + assert completed == [first_call] + + +def test_agent_triggered_compaction_tool_describes_reason_parameter() -> None: + strategy = AgentTriggeredCompactionStrategy(_RecordingCompactionStrategy()) + + reason_schema = strategy.tool.parameters()["properties"]["reason"] + + assert reason_schema["description"] == "Why the agent is requesting compaction before the next model call." + assert "Call this tool by itself" in strategy.tool.description + assert "do not call it in parallel" in strategy.tool.description + + +async def test_unload_skills_compaction_removes_named_skill_only() -> None: + skill_strategy = UnloadSkillsCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(skill_strategy) + messages = [ + Message(role="user", contents=["work"]), + _assistant_named_function_call("load-pdf", "load_skill", {"skill_name": "pdf"}), + _tool_result("load-pdf", "pdf instructions"), + _assistant_named_function_call("load-excel", "load_skill", {"skill_name": "excel"}), + _tool_result("load-excel", "excel instructions"), + _assistant_named_function_call("unload-pdf", strategy.tool.name, {"skill_name": "pdf"}), + _skill_unload_result("unload-pdf", ["pdf"]), + ] + + changed = await strategy(messages) + + assert changed is True + assert messages[1].additional_properties.get(EXCLUDED_KEY) is True + assert messages[2].additional_properties.get(EXCLUDED_KEY) is True + assert messages[3].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[4].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[5].additional_properties.get(EXCLUDED_KEY) is not True + assert messages[6].additional_properties.get(EXCLUDED_KEY) is not True + + +async def test_unload_skills_compaction_removes_multiple_named_skills() -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + messages = [ + _assistant_named_function_call("load-pdf", "load_skill", {"skill_name": "pdf"}), + _tool_result("load-pdf", "pdf instructions"), + _assistant_named_function_call("load-excel", "load_skill", {"skill_name": "excel"}), + _tool_result("load-excel", "excel instructions"), + _assistant_named_function_call("load-word", "load_skill", {"skill_name": "word"}), + _tool_result("load-word", "word instructions"), + _assistant_named_function_call( + "unload-selected", + strategy.tool.name, + {"skill_name": ["pdf", "excel"]}, + ), + _skill_unload_result("unload-selected", ["pdf", "excel"]), + ] + + changed = await strategy(messages) + + assert changed is True + assert all(message.additional_properties.get(EXCLUDED_KEY) is True for message in messages[:4]) + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages[4:6]) + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages[6:]) + + +def test_unload_skills_tool_describes_skill_name_parameter() -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + + skill_name_schema = strategy.tool.parameters()["properties"]["skill_name"] + + assert skill_name_schema["description"] == ( + "The name of a skill to unload, a list of skill names to unload, or omit to unload all loaded skills." + ) + assert "Call this tool by itself" in strategy.tool.description + assert "do not call it in parallel" in strategy.tool.description + + +@pytest.mark.parametrize( + "skill_name", + [123, [123], ["pdf", 123], b"pdf"], +) +async def test_unload_skills_tool_rejects_malformed_skill_names(skill_name: Any) -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + + with pytest.raises(TypeError, match="skill_name"): + await strategy.tool.invoke(arguments={"skill_name": skill_name}) + + +async def test_unload_skills_compaction_empty_name_removes_all_skill_context() -> None: + skill_strategy = UnloadSkillsCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(skill_strategy) + messages = [ + _assistant_named_function_call("load-pdf", "load_skill", {"skill_name": "pdf"}), + _tool_result("load-pdf", "pdf instructions"), + _assistant_named_function_call( + "resource-excel", + "read_skill_resource", + {"skill_name": "excel", "resource_name": "reference"}, + ), + _tool_result("resource-excel", "excel reference"), + _assistant_named_function_call("unload-all", strategy.tool.name, {"skill_name": " "}), + _skill_unload_result("unload-all", []), + ] + + changed = await strategy(messages) + + assert changed is True + assert all(message.additional_properties.get(EXCLUDED_KEY) is True for message in messages[:4]) + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages[4:]) + + +async def test_unload_skills_compaction_preserves_script_results_and_mixed_groups() -> None: + skill_strategy = UnloadSkillsCompactionStrategy() + strategy = AgentTriggeredCompactionStrategy(skill_strategy) + mixed_call = Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="load-pdf", + name="load_skill", + arguments={"skill_name": "pdf"}, + ), + Content.from_function_call( + call_id="search", + name="search", + arguments={"query": "pdf"}, + ), + ], + ) + messages = [ + mixed_call, + Message( + role="tool", + contents=[ + Content.from_function_result(call_id="load-pdf", result="pdf instructions"), + Content.from_function_result(call_id="search", result="search result"), + ], + ), + _assistant_named_function_call( + "run-pdf", + "run_skill_script", + {"skill_name": "pdf", "script_name": "convert"}, + ), + _tool_result("run-pdf", "converted.pdf"), + _assistant_named_function_call("unload-pdf", strategy.tool.name, {"skill_name": "pdf"}), + _skill_unload_result("unload-pdf", ["pdf"]), + ] + + changed = await strategy(messages) + + assert changed is False + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages) + + +async def test_unload_skills_compaction_preserves_group_with_hosted_tool_call() -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + messages = [ + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="load-pdf", + name="load_skill", + arguments={"skill_name": "pdf"}, + ), + Content.from_shell_tool_call(call_id="shell-1", commands=["echo hi"]), + ], + ), + _tool_result("load-pdf", "pdf instructions"), + _assistant_named_function_call("unload-pdf", strategy.tool.name, {"skill_name": "pdf"}), + _skill_unload_result("unload-pdf", ["pdf"]), + ] + + changed = await strategy(messages) + + assert changed is False + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages) + + +async def test_unload_skills_compaction_preserves_group_with_search_tool_call() -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + messages = [ + Message( + role="assistant", + contents=[ + Content.from_function_call( + call_id="load-pdf", + name="load_skill", + arguments={"skill_name": "pdf"}, + ), + Content.from_search_tool_call( + call_id="search-1", + tool_name="web_search", + arguments={"query": "pdf"}, + ), + ], + ), + Message( + role="tool", + contents=[ + Content.from_function_result(call_id="load-pdf", result="pdf instructions"), + Content.from_search_tool_result( + call_id="search-1", + tool_name="web_search", + result="search result", + ), + ], + ), + _assistant_named_function_call("unload-pdf", strategy.tool.name, {"skill_name": "pdf"}), + _skill_unload_result("unload-pdf", ["pdf"]), + ] + + changed = await strategy(messages) + + assert changed is False + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages) + + +async def test_unload_skills_compaction_preserves_reused_id_group_with_unresolved_call() -> None: + strategy = AgentTriggeredCompactionStrategy(UnloadSkillsCompactionStrategy()) + messages = [ + Message( + role="assistant", + contents=[ + Content.from_function_call( + id="load-pdf", + call_id="shared", + name="load_skill", + arguments={"skill_name": "pdf"}, + ), + Content.from_function_call( + id="load-excel", + call_id="shared", + name="load_skill", + arguments={"skill_name": "excel"}, + ), + ], + ), + Message( + role="tool", + contents=[ + Content.from_function_result( + id="load-pdf", + call_id="shared", + result="pdf instructions", + ) + ], + ), + _assistant_named_function_call("unload-all", strategy.tool.name), + _skill_unload_result("unload-all", []), + ] + + changed = await strategy(messages) + + assert changed is False + assert all(message.additional_properties.get(EXCLUDED_KEY) is not True for message in messages) + + async def test_selective_tool_call_strategy_excludes_older_tool_groups() -> None: messages = [ Message(role="user", contents=["u"]), @@ -1673,6 +2313,42 @@ async def test_apply_compaction_logs_changed_context_without_content(caplog: Any # --- ToolResultCompactionStrategy tests --- +async def test_tool_result_compaction_preserves_hosted_search_result() -> None: + messages = [ + Message(role="user", contents=["find docs"]), + Message( + role="assistant", + contents=[ + Content.from_search_tool_call( + call_id="search-1", + tool_name="web_search", + arguments={"query": "agent framework"}, + ) + ], + ), + Message( + role="tool", + contents=[ + Content.from_search_tool_result( + call_id="search-1", + tool_name="web_search", + result="Found the Agent Framework documentation.", + ) + ], + ), + Message(role="assistant", contents=["done"]), + ] + annotate_message_groups(messages) + + changed = await ToolResultCompactionStrategy(keep_last_tool_call_groups=0)(messages) + + assert changed is True + summaries = [message.text or "" for message in messages if (message.text or "").startswith("[Tool results:")] + assert len(summaries) == 1 + assert "web_search: Found the Agent Framework documentation." in summaries[0] + assert "no results" not in summaries[0] + + async def test_tool_result_compaction_collapses_old_groups_into_summary() -> None: """Old tool-call groups are collapsed into summary messages, newest kept.""" messages = [ diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index f185504d2e3..4b8a34dc829 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -7956,6 +7956,7 @@ def test_replace_approval_contents_with_results_correlates_reused_call_id_pendin def test_replace_approval_contents_with_results_resolves_rejected_request() -> None: """A rejected pending call produces one terminal result.""" from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results + from agent_framework._types import APPROVAL_REJECTION_RESULT_KEY function_call, request, _ = _build_approved_tool_roundtrip( call_id="call_rejected", @@ -7977,6 +7978,7 @@ def test_replace_approval_contents_with_results_resolves_rejected_request() -> N results = [content for message in messages for content in message.contents if content.type == "function_result"] assert len(results) == 1 assert results[0].result == "Error: Tool call invocation was rejected by user." + assert results[0].additional_properties[APPROVAL_REJECTION_RESULT_KEY] is True assert resolved_contents == results diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index a5c18ebda85..358855267d0 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -601,6 +601,11 @@ class RawOpenAIChatClient( INJECTABLE: ClassVar[set[str]] = {"client"} STORES_BY_DEFAULT: ClassVar[bool] = True + SERVICE_CONTINUATION_OPTION_NAMES: ClassVar[tuple[str, ...]] = ( + "conversation_id", + "previous_response_id", + "conversation", + ) SUPPORTS_RICH_FUNCTION_OUTPUT: ClassVar[bool] = True _FEATURE_USAGE_INDEX: ClassVar[int | None] = FeatureIndex.OPENAI 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 15cadc9d9b9..3b94aa4d912 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -16,6 +16,7 @@ import pytest from agent_framework import ( Agent, + AgentTriggeredCompactionStrategy, Annotation, ChatOptions, ChatResponse, @@ -30,6 +31,7 @@ SupportsImageGenerationTool, SupportsMCPTool, SupportsWebSearchTool, + TruncationStrategy, WorkflowBuilder, tool, ) @@ -89,6 +91,43 @@ ) +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize( + ("option_name", "option_value"), + [ + ("previous_response_id", "resp_existing"), + ("conversation", "conv_existing"), + ], +) +async def test_agent_triggered_compaction_rejects_native_service_continuation_aliases( + stream: bool, + option_name: str, + option_value: str, +) -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + strategy = AgentTriggeredCompactionStrategy(TruncationStrategy(max_n=2, compact_to=1)) + options = cast(OpenAIChatOptions, {"store": False, option_name: option_value}) + + with pytest.raises(ChatClientInvalidRequestException, match="service-managed continuation"): + if stream: + response_stream = client.get_response( + [Message(role="user", contents=["Hello"])], + stream=True, + options=options, + compaction_strategy=strategy, + ) + async for _ in response_stream: + pass + else: + response = client.get_response( + [Message(role="user", contents=["Hello"])], + stream=False, + options=options, + compaction_strategy=strategy, + ) + _ = await response + + class OutputStruct(BaseModel): """A structured output for testing purposes.""" diff --git a/python/samples/02-agents/compaction/README.md b/python/samples/02-agents/compaction/README.md index 779893eefeb..420e3791e2c 100644 --- a/python/samples/02-agents/compaction/README.md +++ b/python/samples/02-agents/compaction/README.md @@ -11,6 +11,8 @@ This folder demonstrates context compaction patterns introduced by ADR-0019. - `custom.py` — defines a custom strategy implementing the `CompactionStrategy` protocol. - `tiktoken_tokenizer.py` — shows a `TokenizerProtocol` implementation backed by `tiktoken`. - `compaction_provider.py` — uses `CompactionProvider` with an agent and `InMemoryHistoryProvider`. +- `agent_triggered_compaction.py` — lets an agent trigger in-run compaction to unload one skill's context. +- `agent_triggered_compaction_basic.py` — demonstrates the generic wrapper with an existing tool-call strategy. Run samples with: @@ -22,6 +24,8 @@ uv run samples/02-agents/compaction/agent_client_overrides.py uv run samples/02-agents/compaction/custom.py uv run samples/02-agents/compaction/tiktoken_tokenizer.py uv run samples/02-agents/compaction/compaction_provider.py # requires OPENAI_API_KEY +uv run samples/02-agents/compaction/agent_triggered_compaction.py # requires Foundry configuration +uv run samples/02-agents/compaction/agent_triggered_compaction_basic.py # requires Foundry configuration ``` ## Security Considerations diff --git a/python/samples/02-agents/compaction/agent_triggered_compaction.py b/python/samples/02-agents/compaction/agent_triggered_compaction.py new file mode 100644 index 00000000000..031aef502ff --- /dev/null +++ b/python/samples/02-agents/compaction/agent_triggered_compaction.py @@ -0,0 +1,167 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import os +from itertools import count +from textwrap import dedent + +from agent_framework import ( + Agent, + AgentTriggeredCompactionStrategy, + CompactionStrategy, + InlineSkill, + Message, + SkillFrontmatter, + SkillsProvider, + UnloadSkillsCompactionStrategy, + included_messages, +) +from agent_framework.foundry import FoundryChatClient +from azure.identity import AzureCliCredential +from dotenv import load_dotenv + +load_dotenv() + +"""This sample demonstrates agent-triggered in-run compaction. + +The agent follows three steps in one function-calling run: + +1. Call ``load_skill`` to load a planning skill. +2. Use the skill, then call ``unload_skill``. +3. Continue without the loaded skill body in model context. + +``AgentTriggeredCompactionStrategy`` controls when compaction runs. +``UnloadSkillsCompactionStrategy`` controls what context is removed. +The trigger tool is registered explicitly alongside the strategy. + +A small inspection wrapper prints every model input after compaction so the +skill body is visibly present after ``load_skill`` and absent after +``unload_skill``. + +Environment variables: + FOUNDRY_PROJECT_ENDPOINT — Microsoft Foundry project endpoint. + FOUNDRY_MODEL — Model deployment name. + +Run with: + uv run samples/02-agents/compaction/agent_triggered_compaction.py +""" + + +MODEL_CALLS = count(1) + + +class InspectingCompactionStrategy: + """Print the projected messages after applying another strategy.""" + + def __init__(self, strategy: CompactionStrategy) -> None: + self.strategy = strategy + + async def __call__(self, messages: list[Message]) -> bool: + changed = await self.strategy(messages) + visible_messages = included_messages(messages) + + print(f"\n--- Model call {next(MODEL_CALLS)} ---") + print(f"Visible messages: {len(visible_messages)}") + for message in visible_messages: + parts: list[str] = [] + for content in message.contents: + if content.type == "text": + text = (content.text or "").replace("\n", " ") + parts.append(text[:100]) + elif content.type == "function_call": + parts.append(f"{content.name}({content.arguments})") + elif content.type == "function_result": + result = str(content.result).replace("\n", " ") + parts.append(f"result={result[:100]}") + else: + parts.append(content.type) + print(f"[{message.role}] {' | '.join(parts)}") + return changed + + +planning_skill = InlineSkill( + frontmatter=SkillFrontmatter( + name="planning", + description="Create a concise implementation plan before execution.", + ), + instructions=dedent("""\ + # Planning Skill + + Before implementation: + 1. Identify the required behavior. + 2. List the smallest code changes. + 3. Define the targeted validation. + """), +) + + +async def main() -> None: + # 1. Wrap the skill strategy. The wrapper automatically adopts its + # unload_skill trigger and exposes that single tool. + triggered_compaction = AgentTriggeredCompactionStrategy( + UnloadSkillsCompactionStrategy(), + ) + + # 2. Use the standard Foundry chat client. Run `az login` before starting. + client = FoundryChatClient( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + model=os.environ["FOUNDRY_MODEL"], + credential=AzureCliCredential(), + ) + + # 3. The trigger tool is registered explicitly. SkillsProvider contributes + # load_skill, while disabling approval keeps this sample automatic. + agent = Agent( + client=client, + name="PlanningAgent", + instructions=dedent("""\ + If the current context contains a completed unload_skill result, the planning + skill is already unloaded: do not call any tools, return the plan, and say + that execution can begin. + + Otherwise, follow this sequence exactly: + 1. Call load_skill with skill_name="planning". + 2. Use the loaded instructions to create a concise plan. + 3. Call unload_skill with skill_name="planning". + + Do not call load_skill and unload_skill in parallel. + """), + tools=[triggered_compaction.tool], + context_providers=[ + SkillsProvider( + planning_skill, + disable_load_skill_approval=True, + ) + ], + compaction_strategy=InspectingCompactionStrategy(triggered_compaction), + default_options={"store": False}, + ) + + # 4. The final model call printed by the middleware no longer contains the + # load_skill call or its large planning-skill result. + result = await agent.run("Plan a small implementation, unload the planning skill, then move to execution.") + print(f"\nAgent: {result.messages[-1].text}") + + +if __name__ == "__main__": + asyncio.run(main()) + +""" +Example output (model wording varies): + +--- Model call 1 --- +[user] Plan a small implementation, unload the planning skill, then move to execution. + +--- Model call 2 --- +[user] Plan a small implementation, unload the planning skill, then move to execution. +[assistant] load_skill({"skill_name":"planning"}) +[tool] result=planning Create a concise implementation plan... + +--- Model call 3 --- +Visible messages: 3 +[user] Plan a small implementation, unload the planning skill, then move to execution. +[assistant] unload_skill({"skill_name":"planning"}) +[tool] result=Skill(s) 'planning' will be unloaded before the next model call. + +Agent: [The generated plan, followed by confirmation that execution can begin.] +""" diff --git a/python/samples/02-agents/compaction/agent_triggered_compaction_basic.py b/python/samples/02-agents/compaction/agent_triggered_compaction_basic.py new file mode 100644 index 00000000000..8ed2e05b59f --- /dev/null +++ b/python/samples/02-agents/compaction/agent_triggered_compaction_basic.py @@ -0,0 +1,140 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import os +from itertools import count + +from agent_framework import ( + Agent, + AgentTriggeredCompactionStrategy, + CompactionStrategy, + Message, + ToolResultCompactionStrategy, + included_messages, + tool, +) +from agent_framework.foundry import FoundryChatClient +from azure.identity import AzureCliCredential +from dotenv import load_dotenv + +load_dotenv() + +"""This sample demonstrates generic agent-triggered compaction. + +The agent first calls ``collect_context`` to produce a deliberately large tool +result. It then calls ``compact_context``. Before the next model call, +``ToolResultCompactionStrategy`` replaces the older context-producing tool +group with a compact summary, and the wrapper removes the completed compaction +trigger group. + +Environment variables: + FOUNDRY_PROJECT_ENDPOINT — Microsoft Foundry project endpoint. + FOUNDRY_MODEL — Model deployment name. + +Run with: + uv run samples/02-agents/compaction/agent_triggered_compaction_basic.py +""" + + +MODEL_CALLS = count(1) + + +@tool(approval_mode="never_require") +def collect_context(topic: str) -> str: + """Collect intentionally verbose context for a topic.""" + return "\n".join( + f"Exploration note {index}: {topic} should remain simple, testable, and observable." for index in range(1, 21) + ) + + +class InspectingCompactionStrategy: + """Print the projected messages after applying another strategy.""" + + def __init__(self, strategy: CompactionStrategy) -> None: + self.strategy = strategy + + async def __call__(self, messages: list[Message]) -> bool: + changed = await self.strategy(messages) + visible_messages = included_messages(messages) + + print(f"\n--- Model call {next(MODEL_CALLS)} ---") + print(f"Visible messages: {len(visible_messages)}") + for message in visible_messages: + parts: list[str] = [] + for content in message.contents: + if content.type == "text": + text = (content.text or "").replace("\n", " ") + parts.append(text[:100]) + elif content.type == "function_call": + parts.append(f"{content.name}({content.arguments})") + elif content.type == "function_result": + result = str(content.result).replace("\n", " ") + parts.append(f"result={result[:100]}") + else: + parts.append(content.type) + print(f"[{message.role}] {' | '.join(parts)}") + return changed + + +async def main() -> None: + # 1. Wrap an existing strategy. The trigger does not count toward the + # keep-last setting, so 0 compacts every workload tool-call group. + # A summary remains, allowing this sample to remove the trigger too. + triggered_compaction = AgentTriggeredCompactionStrategy( + ToolResultCompactionStrategy(keep_last_tool_call_groups=0), + exclude_trigger=True, + ) + + # 2. Use Foundry with store=False so Agent Framework owns the full in-run + # message list. Run `az login` before starting. + client = FoundryChatClient( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + model=os.environ["FOUNDRY_MODEL"], + credential=AzureCliCredential(), + ) + + # 3. Register both the ordinary context-producing tool and the explicit + # compaction trigger tool. + agent = Agent( + client=client, + name="ResearchAgent", + instructions=( + "If the current context contains an assistant message starting with '[Tool results:', " + "compaction is complete: do not call any tools and answer the user in one sentence. " + "Otherwise, call collect_context once and then call compact_context. " + "Do not call the tools in parallel." + ), + tools=[collect_context, triggered_compaction.tool], + compaction_strategy=InspectingCompactionStrategy(triggered_compaction), + default_options={"store": False}, + ) + + # 4. The third model input contains a compact tool-result summary, but + # neither of the original tool-call groups. + result = await agent.run("Identify the main implementation priorities.") + print(f"\nAgent: {result.messages[-1].text}") + + +if __name__ == "__main__": + asyncio.run(main()) + +""" +Example output (model wording varies): + +--- Model call 1 --- +Visible messages: 1 +[user] Identify the main implementation priorities. + +--- Model call 2 --- +Visible messages: 3 +[user] Identify the main implementation priorities. +[assistant] collect_context({"topic":"implementation priorities"}) +[tool] result=Exploration note 1: implementation priorities should remain simple, testable... + +--- Model call 3 --- +Visible messages: 2 +[user] Identify the main implementation priorities. +[assistant] [Tool results: collect_context: Exploration note 1: implementation priorities...] + +Agent: Keep the implementation simple, testable, and observable. +"""