diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md index 6f6913b71d..efe220a475 100644 --- a/python/packages/core/AGENTS.md +++ b/python/packages/core/AGENTS.md @@ -309,6 +309,7 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID - **`todos_remaining(*, looping_modes=None)`** / **`todos_remaining_message`** - Helper factories for todo-driven loops (the Python counterpart of .NET's `TodoCompletionLoopEvaluator`), designed for `create_harness_agent` but usable with any agent that registers a `TodoProvider` via `context_providers`. They resolve the `TodoProvider`/`AgentModeProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `loop_should_continue`/`loop_next_message`. `todos_remaining` returns a `should_continue` predicate that loops while any todo is open; pass `looping_modes=[...]` to gate looping to specific operating modes (case-insensitive; honors the `AgentModeProvider`'s `source_id`/`available_modes`), `looping_modes=None` (default) applies in every mode, and an empty sequence raises `ValueError`. `todos_remaining_message` is a `next_message` callable that lists the still-open todo titles and tells the agent to finish them, returning `None` when the session/agent/provider is unavailable or nothing is open (in which case the middleware's default `None` handling applies: reuse the previous iteration's messages verbatim under the default `fresh_context=False`, or `DEFAULT_NEXT_MESSAGE` only when `fresh_context=True`). - **`background_tasks_running()`** / **`background_tasks_running_message`** - Helper factories for background-agent-driven loops, mirroring the `todos_remaining` pair. They resolve the `BackgroundAgentsProvider` from the *running agent* (`agent.context_providers`, via `_resolve_context_provider`) rather than taking the provider as an argument, so they can be wired directly into `create_harness_agent`'s `loop_should_continue`/`loop_next_message`. `background_tasks_running` returns a `should_continue` predicate that loops while the provider's persisted state shows any task with `status == RUNNING` (pair it with `max_iterations` so the loop is bounded even if a task's persisted status is never refreshed). `background_tasks_running_message` is a `next_message` callable that lists the still-running tasks (`# (): `) and tells the agent to wait for them to finish and retrieve their results, returning `None` when the session/agent/provider is unavailable or no task is running. - **Approval escape hatch** - `_has_pending_approval_request(result)` checks whether an iteration's response carries a pending tool-approval request (any content with `type == "function_approval_request"`). Both the streaming and non-streaming loops stop and return that response to the caller *before* evaluating `should_continue`/`max_iterations` or injecting `next_message`, so the loop is HITL-safe even when wrapped outermost around a `ToolApprovalMiddleware` (mirrors the C# `LoopAgent`'s `HasPendingApprovalRequests`). + - **Signal protocol** (`_harness/_loop.py`) - `AgentLoopMiddleware` stamps `loop_exit_reason` in `AgentResponse.additional_properties` for framework-owned stops (`iteration_cap_reached` when the cap fires, overwriting any pre-existing value; `completed` when the judge from `with_judge` decides the task is answered). A `should_continue` callback may also stamp `loop_exit_reason` directly before returning `False`. See `python/samples/02-agents/harness/signal_protocol.py` for a complete `TASK_COMPLETE:` / `NEED_INPUT:` signal-token parsing example. - **Harness integration** - `create_harness_agent` enables the loop when a `loop_should_continue` callable is passed; it prepends `AgentLoopMiddleware(loop_should_continue, max_iterations=loop_max_iterations, next_message=loop_next_message)` ahead of `ToolApprovalMiddleware` so the loop is the outermost middleware (each iteration is a full agent run including tool approval, and the escape hatch hands pending approvals back to the caller). `loop_next_message` and `loop_max_iterations` only take effect together with `loop_should_continue` (with no `loop_should_continue` there is no loop, so they are ignored); `loop_max_iterations` defaults to the loop's default cap (`None` → unbounded). ### Workflows (`_workflows/`) diff --git a/python/packages/core/agent_framework/_harness/_loop.py b/python/packages/core/agent_framework/_harness/_loop.py index b4617a23e0..10baaafc59 100644 --- a/python/packages/core/agent_framework/_harness/_loop.py +++ b/python/packages/core/agent_framework/_harness/_loop.py @@ -23,6 +23,12 @@ response into ``JudgeVerdict``. The loop continues while the answer is "no". This is a convenience wrapper that builds an async ``should_continue`` predicate, so it is a special case of (1). +When ``max_iterations`` fires the loop stamps ``additional_properties["loop_exit_reason"] = +"iteration_cap_reached"`` on the response; the judge (via :meth:`AgentLoopMiddleware.with_judge`) +stamps ``"completed"`` when it decides the request was answered. Callers read the exit reason +directly from ``response.additional_properties["loop_exit_reason"]``. A custom ``should_continue`` +callback may also stamp that key itself before returning ``False``. + In every case, the input for the next iteration is controlled by the ``next_message`` callable. """ @@ -131,6 +137,10 @@ class JudgeVerdict(BaseModel): DEFAULT_JUDGE_MAX_ITERATIONS = 5 +# Key used in AgentResponse.additional_properties to surface the loop exit reason. +LOOP_EXIT_REASON_KEY: str = "loop_exit_reason" + + # A callable invoked between iterations. It always receives the loop keyword arguments # (``iteration``, ``last_result``, ``messages``, ``original_messages``, ``session``, ``agent``, # ``progress``, ``feedback``). Callers declare only the keywords they need plus ``**kwargs`` to @@ -210,6 +220,8 @@ async def _judge( answered = False if JUDGE_VERDICT_MORE in text else JUDGE_VERDICT_DONE in text reasoning = response.text.strip() # Continue looping while the request is not yet answered, surfacing the reasoning as feedback. + if answered: + last_result.additional_properties[LOOP_EXIT_REASON_KEY] = "completed" return (not answered), (reasoning or None) def _next_message(*, feedback: str | None = None, **kwargs: Any) -> AgentRunInputs: @@ -538,6 +550,7 @@ async def _process_non_streaming( snapshot: dict[str, Any] | None, ) -> None: iteration = 0 + cap_fired = False work_iterations = 0 progress: list[str] = [] # Aggregated transcript across iterations: each iteration's response messages plus the @@ -560,6 +573,9 @@ async def _process_non_streaming( f"got {type(result).__name__}." ) + # Clear any loop_exit_reason stamped by an inner agent or nested loop so the outer + # loop's own exit-reason decision is not masked by a stale inner value. + result.additional_properties.pop(LOOP_EXIT_REASON_KEY, None) final_result = result aggregated.extend(result.messages) if result.usage_details is not None: @@ -584,6 +600,8 @@ async def _process_non_streaming( # Decide whether to stop and capture any feedback from should_continue first, so the # feedback is available to both the progress and next-message callables this iteration. stop, feedback = await self._evaluate_stop(loop_kwargs, work_iterations) + if stop and self.max_iterations is not None and work_iterations >= self.max_iterations: + cap_fired = True loop_kwargs = self._build_loop_kwargs( context=context, iteration=iteration, @@ -617,6 +635,9 @@ async def _process_non_streaming( finally: context.options.pop(_LOOP_ITERATION_TOKEN_KEY, None) + if cap_fired: + final_result.additional_properties.setdefault(LOOP_EXIT_REASON_KEY, "iteration_cap_reached") + if not self.return_final_only: context.result = self._aggregate_response(final_result, aggregated, aggregated_usage) await self._fire_turn_scoped_after_providers( @@ -638,6 +659,7 @@ def _process_streaming( async def _generator() -> Any: iteration = 0 + cap_fired = False work_iterations = 0 progress: list[str] = [] stamped_options = dict(context.options) if context.options is not None else {} @@ -658,6 +680,9 @@ async def _generator() -> Any: yield update holder["final"] = await inner.get_final_response() + # Clear any loop_exit_reason stamped by an inner agent or nested loop so the outer + # loop's own exit-reason decision is not masked by a stale inner value. + holder["final"].additional_properties.pop(LOOP_EXIT_REASON_KEY, None) except MiddlewareTermination: # The pipeline's MiddlewareTermination suppression is no longer active once # process() has returned (the stream is consumed lazily), so a termination @@ -686,6 +711,8 @@ async def _generator() -> Any: # Decide whether to stop and capture any feedback from should_continue first, so the # feedback is available to both the progress and next-message callables this iteration. stop, feedback = await self._evaluate_stop(loop_kwargs, work_iterations) + if stop and self.max_iterations is not None and work_iterations >= self.max_iterations: + cap_fired = True loop_kwargs = self._build_loop_kwargs( context=context, iteration=iteration, @@ -721,6 +748,8 @@ async def _generator() -> Any: yield self._message_to_update(message) finally: context.options.pop(_LOOP_ITERATION_TOKEN_KEY, None) + if cap_fired and holder["final"] is not None: + holder["final"].additional_properties.setdefault(LOOP_EXIT_REASON_KEY, "iteration_cap_reached") await self._fire_turn_scoped_after_providers(context, holder["final"], original_messages) def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse: diff --git a/python/packages/core/tests/core/test_harness_loop.py b/python/packages/core/tests/core/test_harness_loop.py index 3b87544d80..aaa7686116 100644 --- a/python/packages/core/tests/core/test_harness_loop.py +++ b/python/packages/core/tests/core/test_harness_loop.py @@ -40,6 +40,7 @@ DEFAULT_JUDGE_MAX_ITERATIONS, DEFAULT_MAX_ITERATIONS, DEFAULT_NEXT_MESSAGE, + LOOP_EXIT_REASON_KEY, AgentLoopMiddleware, ) @@ -1697,3 +1698,151 @@ async def test_concurrent_same_agent_run_during_stream_pause_is_not_suppressed() # once for the loop boundary, once for the concurrent run's own boundary assert turn_scoped.after_calls == 2 + + +# --- Framework stop-reason tests --- + + +async def test_loop_sets_iteration_cap_reason_in_additional_properties() -> None: + client = RecordingChatClient() + agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)]) + + response = await agent.run("task") + + assert client.call_count == 2 + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached" + + +async def test_loop_sets_iteration_cap_reason_streaming() -> None: + client = RecordingChatClient() + agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)]) + + stream = agent.run("task", stream=True) + async for _ in stream: + pass + response = await stream.get_final_response() + + assert client.call_count == 2 + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached" + + +class StaleExitReasonChatClient(RecordingChatClient): + """Stamps a stale ``loop_exit_reason`` on every response, emulating an inner provider or nested loop.""" + + def _inner_get_response( + self, + *, + messages: Sequence[Message], + stream: bool = False, + options: Mapping[str, Any], + **kwargs: Any, + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + inner = super()._inner_get_response(messages=messages, stream=stream, options=options, **kwargs) + if isinstance(inner, ResponseStream): + + async def _gen() -> AsyncIterable[ChatResponseUpdate]: + async for update in inner: + update.additional_properties = {LOOP_EXIT_REASON_KEY: "stale_reason"} + yield update + + return ResponseStream(_gen(), finalizer=ChatResponse.from_updates) + + async def _get() -> ChatResponse: + response = await inner + response.additional_properties = {LOOP_EXIT_REASON_KEY: "stale_reason"} + return response + + return _get() + + +async def test_loop_cap_overwrites_pre_existing_exit_reason() -> None: + client = StaleExitReasonChatClient() + agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)]) + + response = await agent.run("task") + + assert client.call_count == 2 + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached" + + +async def test_loop_cap_overwrites_pre_existing_exit_reason_streaming() -> None: + client = StaleExitReasonChatClient() + agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: True, max_iterations=2)]) + + stream = agent.run("task", stream=True) + async for _ in stream: + pass + response = await stream.get_final_response() + + assert client.call_count == 2 + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached" + + +async def test_judge_sets_completed_in_additional_properties() -> None: + """Judge marks the response as completed when the task is answered.""" + agent_client = RecordingChatClient(texts=["The task is fully complete."]) + judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True) + agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)]) + + response = await agent.run("do the task") + + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed" + + +async def test_judge_sets_completed_in_additional_properties_streaming() -> None: + agent_client = RecordingChatClient(texts=["The task is fully complete."]) + judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True) + agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)]) + + stream = agent.run("do the task", stream=True) + async for _ in stream: + pass + response = await stream.get_final_response() + + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed" + + +async def test_judge_sets_completed_only_once_answered() -> None: + """Judge does not stamp completed until the task is actually answered.""" + agent_client = RecordingChatClient(texts=["Still working...", "Now done."]) + judge_client = RecordingChatClient( + texts=['{"answered": false, "reasoning": "needs more"}', '{"answered": true, "reasoning": "done"}'], + honor_response_format=True, + ) + agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)]) + + response = await agent.run("do the task") + + assert agent_client.call_count == 2 + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed" + + +async def test_judge_does_not_set_completed_when_cap_fires() -> None: + agent_client = RecordingChatClient() + judge_client = RecordingChatClient(texts=['{"answered": false}'] * 2, honor_response_format=True) + agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client, max_iterations=2)]) + + response = await agent.run("do the task") + + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "iteration_cap_reached" + + +async def test_stale_inner_exit_reason_is_cleared_before_outer_evaluation() -> None: + """A stale inner loop_exit_reason must not leak into, or block, the outer loop's decision.""" + agent_client = StaleExitReasonChatClient(texts=["The task is fully complete."]) + judge_client = RecordingChatClient(texts=['{"answered": true, "reasoning": "done"}'], honor_response_format=True) + agent = Agent(client=agent_client, middleware=[AgentLoopMiddleware.with_judge(judge_client)]) + + response = await agent.run("do the task") + + assert response.additional_properties.get(LOOP_EXIT_REASON_KEY) == "completed" + + +async def test_stale_inner_exit_reason_is_cleared_when_loop_stops_without_reason() -> None: + """If the outer loop stops without stamping a reason, the inner stale value must not survive.""" + client = StaleExitReasonChatClient() + agent = Agent(client=client, middleware=[AgentLoopMiddleware(lambda **_: False, max_iterations=5)]) + + response = await agent.run("task") + + assert LOOP_EXIT_REASON_KEY not in response.additional_properties diff --git a/python/samples/02-agents/harness/signal_protocol.py b/python/samples/02-agents/harness/signal_protocol.py new file mode 100644 index 0000000000..54f9e80e32 --- /dev/null +++ b/python/samples/02-agents/harness/signal_protocol.py @@ -0,0 +1,140 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Signal-protocol sample: agent signals task completion via TASK_COMPLETE: / NEED_INPUT:. + +This sample shows how to implement a signal protocol as a ``should_continue`` callback +for :class:`~agent_framework.AgentLoopMiddleware`. The parsing logic lives here in the +application rather than in the framework core, which lets you customize the tokens, the +matching rules, and the exit-reason values for your own use case. + +How it works: +1. The agent emits ``TASK_COMPLETE: `` or ``NEED_INPUT: `` at the + start of a line when it wants to stop. +2. The ``signal_should_continue`` callback detects these tokens and sets + ``response.additional_properties["loop_exit_reason"]`` before returning ``False``. +3. The caller reads the exit reason from ``response.additional_properties`` directly. + +Iteration-cap precedence: + ``AgentLoopMiddleware`` checks ``max_iterations`` before calling ``should_continue``. + If the agent emits ``TASK_COMPLETE:`` on the very last allowed iteration, the signal is + NOT detected because the cap fires first and the middleware stamps + ``"iteration_cap_reached"``. Handle that value as a distinct outcome and inspect the + response text if needed. + +The demo at the bottom uses a scripted in-memory chat client, so it runs without an LLM. +""" + +from __future__ import annotations + +import asyncio +import re +import warnings +from collections.abc import Awaitable, Mapping, Sequence +from typing import Any + +# The harness is experimental; silence the warning for this sample. +warnings.filterwarnings("ignore", message=r"\[HARNESS\]") + +from agent_framework import ( # noqa: E402 + Agent, + AgentLoopMiddleware, + AgentResponse, + BaseChatClient, + ChatOptions, + ChatResponse, + Message, +) + +TASK_COMPLETE_TOKEN = "TASK_COMPLETE:" # noqa: S105 +NEED_INPUT_TOKEN = "NEED_INPUT:" # noqa: S105 +LOOP_EXIT_REASON_KEY = "loop_exit_reason" + +# A token only counts at the start of a line (leading spaces/tabs allowed), so a +# mid-sentence or backtick-quoted mention is not a signal. +_TASK_COMPLETE_RE = re.compile(rf"^[ \t]*{re.escape(TASK_COMPLETE_TOKEN)}(?P[^\n]*)", re.MULTILINE) +_NEED_INPUT_RE = re.compile(rf"^[ \t]*{re.escape(NEED_INPUT_TOKEN)}(?P[^\n]*)", re.MULTILINE) + + +def _find_signal(response: AgentResponse) -> tuple[str, str] | None: + """Scan assistant-role messages for a terminal signal token. + + Only assistant messages are scanned: user-role nudges injected by the loop between + iterations must not trigger false positives. ``TASK_COMPLETE:`` wins over ``NEED_INPUT:``. + + Returns: + ``(exit_reason, text_after_token)`` or ``None`` when no signal is present. + """ + text = "\n".join(m.text or "" for m in response.messages if m.role == "assistant") + for pattern, reason in ((_TASK_COMPLETE_RE, "completed"), (_NEED_INPUT_RE, "need_input")): + match = pattern.search(text) + if match is not None: + return reason, match.group("text").strip() + return None + + +def signal_should_continue(*, last_result: AgentResponse, **kwargs: Any) -> tuple[bool, str | None]: + """``should_continue`` callback for :class:`~agent_framework.AgentLoopMiddleware`. + + Returns ``(False, text)`` when the agent emits a terminal signal and ``(True, None)`` + otherwise. On a signal it also stamps ``last_result.additional_properties["loop_exit_reason"]`` + so callers can read the exit reason without re-parsing the response text. + """ + signal = _find_signal(last_result) + if signal is None: + return True, None + reason, text = signal + last_result.additional_properties.setdefault(LOOP_EXIT_REASON_KEY, reason) + return False, text or None + + +class ScriptedChatClient(BaseChatClient[ChatOptions[None]]): + """Minimal chat client that replays scripted texts, then echoes a filler reply.""" + + def __init__(self, texts: list[str]) -> None: + super().__init__() + self._texts = list(texts) + + def _inner_get_response( # type: ignore[override] + self, *, messages: Sequence[Message], stream: bool = False, options: Mapping[str, Any], **kwargs: Any + ) -> Awaitable[ChatResponse]: + async def _get() -> ChatResponse: + text = self._texts.pop(0) if self._texts else "still working" + return ChatResponse(messages=Message(role="assistant", contents=[text])) + + return _get() + + +async def main() -> None: + # Scenario 1: the agent signals completion on the first pass. + agent = Agent( + client=ScriptedChatClient(["TASK_COMPLETE: all steps finished"]), + middleware=[AgentLoopMiddleware(signal_should_continue, max_iterations=10)], + ) + response = await agent.run("complete the task") + reason = response.additional_properties.get(LOOP_EXIT_REASON_KEY) + print(f"Exit reason: {reason}") + assert reason == "completed" + + # Scenario 2: the agent asks the user a question. + agent = Agent( + client=ScriptedChatClient(["NEED_INPUT: which file should I edit?"]), + middleware=[AgentLoopMiddleware(signal_should_continue, max_iterations=10)], + ) + response = await agent.run("edit the file") + reason = response.additional_properties.get(LOOP_EXIT_REASON_KEY) + print(f"Exit reason: {reason}") + assert reason == "need_input" + + # Scenario 3: the iteration cap fires before the agent signals anything. + agent = Agent( + client=ScriptedChatClient(["still working"] * 3), + middleware=[AgentLoopMiddleware(signal_should_continue, max_iterations=2)], + ) + response = await agent.run("task that takes too long") + reason = response.additional_properties.get(LOOP_EXIT_REASON_KEY) + print(f"Exit reason: {reason}") + assert reason == "iteration_cap_reached" + + +if __name__ == "__main__": + asyncio.run(main())