diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index eb81f81572e..43540d36c18 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -314,6 +314,7 @@ def __init__( id: str | None = None, context_mode: Literal["full", "last_agent", "custom"] | None = None, context_filter: Callable[[list[Message]], list[Message]] | None = None, + ensure_trailing_user_turn: bool = False, ): """Initialize the executor with a unique identifier. @@ -332,7 +333,12 @@ def __init__( context_filter: A function that takes the full conversation (list of Messages) as input and returns a filtered list of Messages to be used as context for the agent run. This is required if context_mode is set to "custom". + ensure_trailing_user_turn: If True, appends a synthetic user continuation message + ("please continue.") when the cache ends on an assistant message. Some + chat-completions providers return empty text when the prompt ends on + ``assistant``. Defaults to False to preserve existing behavior. """ + self._ensure_trailing_user_turn = ensure_trailing_user_turn # Prefer provided id; else use agent.name if present; else generate deterministic prefix exec_id = id or resolve_agent_id(agent) if not exec_id: @@ -746,14 +752,21 @@ async def _run_agent_and_emit( containing incremental updates (streaming mode) or a single output event (type='output') containing the complete response (non-streaming mode). """ + invocation_messages = self._build_invocation_messages() self._session.state[_WORKFLOW_DEFER_COMPUTER_FUNCTION_RESULTS_KEY] = True try: if ctx.is_streaming(): # Streaming mode: emit incremental updates - response = await self._run_agent_streaming(cast(WorkflowContext[Never, AgentResponseUpdate], ctx)) + response = await self._run_agent_streaming( + cast(WorkflowContext[Never, AgentResponseUpdate], ctx), + messages=invocation_messages, + ) else: # Non-streaming mode: use run() and emit single event - response = await self._run_agent(cast(WorkflowContext[Never, AgentResponse], ctx)) + response = await self._run_agent( + cast(WorkflowContext[Never, AgentResponse], ctx), + messages=invocation_messages, + ) finally: self._session.state.pop(_WORKFLOW_DEFER_COMPUTER_FUNCTION_RESULTS_KEY, None) @@ -771,11 +784,39 @@ async def _run_agent_and_emit( await ctx.send_message(agent_response) self._cache.clear() - async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentResponse | None: + def _build_invocation_messages(self) -> list[Message]: + """Return an invocation-only copy of the cache with trailing user turn if needed. + + The copy — not ``self._cache`` — is what gets passed to the agent's + ``run()`` call, so the synthetic "please continue." message never + persists in the cache or ``full_conversation``. Only the invocation + that needs the trailing user turn sees it; downstream agents in a + multi-agent chain are unaffected. + """ + messages = list(self._cache) + if self._ensure_trailing_user_turn and messages and messages[-1].role == "assistant": + logger.debug( + "AgentExecutor %s: cache ends on an assistant message; appending a synthetic " + "continuation turn to the invocation-only copy to avoid empty completions from " + "providers that require a trailing user message.", + self.id, + ) + messages.append(Message(role="user", contents=["please continue."])) + return messages + + async def _run_agent( + self, + ctx: WorkflowContext[Never, AgentResponse], + *, + messages: list[Message] | None = None, + ) -> AgentResponse | None: """Execute the underlying agent in non-streaming mode. Args: ctx: The workflow context for emitting events. + messages: Invocation-only message list. If provided, used instead + of ``self._cache`` so the synthetic trailing user turn does + not persist in the cache. Returns: The complete AgentResponse, or None if waiting for user input. @@ -785,7 +826,8 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs) tools = ctx.get_runtime_tools() - if not self._cache: + cache = messages if messages is not None else self._cache + if not cache: logger.warning( "AgentExecutor %s: Running agent with empty message cache. " "This could lead to service error for some LLM providers.", @@ -801,7 +843,7 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR } if tools is not None and self._accepts_runtime_tools: run_kwargs["tools"] = tools - response = await run_agent(self._cache, **run_kwargs) + response = await run_agent(cache, **run_kwargs) # Handle any user input requests if response.user_input_requests: @@ -826,11 +868,19 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR await ctx.yield_output(response) return response - async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUpdate]) -> AgentResponse | None: + async def _run_agent_streaming( + self, + ctx: WorkflowContext[Never, AgentResponseUpdate], + *, + messages: list[Message] | None = None, + ) -> AgentResponse | None: """Execute the underlying agent in streaming mode and collect the full response. Args: ctx: The workflow context for emitting events. + messages: Invocation-only message list. If provided, used instead + of ``self._cache`` so the synthetic trailing user turn does + not persist in the cache. Returns: The complete AgentResponse, or None if waiting for user input. @@ -840,7 +890,8 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs) tools = ctx.get_runtime_tools() - if not self._cache: + cache = messages if messages is not None else self._cache + if not cache: logger.warning( "AgentExecutor %s: Running agent with empty message cache. " "This could lead to service error for some LLM providers.", @@ -859,7 +910,7 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp } if tools is not None and self._accepts_runtime_tools: run_kwargs["tools"] = tools - stream = run_agent_stream(self._cache, **run_kwargs) + stream = run_agent_stream(cache, **run_kwargs) async for update in stream: updates.append(update) if any(content.type == "computer_tool_call" and content.user_input_request for content in update.contents): diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index 7448e0d66f2..e56db14930a 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/python/packages/core/tests/workflow/test_agent_executor.py @@ -576,6 +576,130 @@ def only_user_messages(msgs: list[Message]) -> list[Message]: assert seen[0].role == "user" and "hello" in (seen[0].text or "") +def test_ensure_trailing_user_turn_defaults_to_false() -> None: + """ensure_trailing_user_turn should default to False, preserving existing behavior.""" + agent = _CountingAgent(id="a", name="A") + executor = AgentExecutor(agent) + assert executor._ensure_trailing_user_turn is False # pyright: ignore[reportPrivateUsage] + + +def test_build_invocation_messages_noop_when_flag_disabled() -> None: + """When the flag is off, the returned messages match the cache without modification.""" + agent = _CountingAgent(id="a", name="A") + executor = AgentExecutor(agent, ensure_trailing_user_turn=False) + executor._cache = [ + Message(role="user", contents=["go"]), + Message(role="assistant", contents=["W-OK"]), + ] + messages = executor._build_invocation_messages() + assert len(messages) == 2 + assert messages[-1].role == "assistant" + assert len(executor._cache) == 2 + + +def test_build_invocation_messages_noop_when_cache_not_assistant_ending() -> None: + """When the flag is on but the cache already ends on user/tool, nothing is appended.""" + agent = _CountingAgent(id="a", name="A") + executor = AgentExecutor(agent, ensure_trailing_user_turn=True) + executor._cache = [ + Message(role="user", contents=["go"]), + Message(role="assistant", contents=["W-OK"]), + Message(role="user", contents=["continue"]), + ] + messages = executor._build_invocation_messages() + assert len(messages) == 3 + assert messages[-1].role == "user" + assert len(executor._cache) == 3 + + +def test_build_invocation_messages_appends_synthetic_user_turn_when_enabled() -> None: + """When the flag is on and cache ends on assistant, a synthetic user turn is + appended to the returned copy — but the original cache is not mutated.""" + agent = _CountingAgent(id="a", name="A") + executor = AgentExecutor(agent, ensure_trailing_user_turn=True) + executor._cache = [ + Message(role="user", contents=["go"]), + Message(role="assistant", contents=["W-OK"]), + ] + messages = executor._build_invocation_messages() + assert len(messages) == 3 + assert messages[-1].role == "user" + assert "please continue" in (messages[-1].text or "") + assert len(executor._cache) == 2 + assert executor._cache[-1].role == "assistant" + + +def test_build_invocation_messages_noop_when_cache_empty() -> None: + """An empty cache must not raise or append anything, even with the flag enabled.""" + agent = _CountingAgent(id="a", name="A") + executor = AgentExecutor(agent, ensure_trailing_user_turn=True) + assert executor._cache == [] + messages = executor._build_invocation_messages() + assert messages == [] + assert executor._cache == [] + + +async def test_synthetic_turn_does_not_leak_to_third_agent() -> None: + """The synthetic 'please continue.' must not appear in the third agent's + context — it is invocation-only and must not persist in full_conversation.""" + a1 = _MessageCapturingAgent(id="first", name="First", reply_text="first reply") + a2 = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply") + a3 = _MessageCapturingAgent(id="third", name="Third", reply_text="third reply") + exec_a = AgentExecutor(a1, id="exec_a") + exec_b = AgentExecutor(a2, id="exec_b", context_mode="full", ensure_trailing_user_turn=True) + exec_c = AgentExecutor(a3, id="exec_c", context_mode="full") + wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).add_edge(exec_b, exec_c).build() + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + seen = a3.last_messages + assert len(seen) == 3 + for msg in seen: + assert "please continue" not in (msg.text or "") + + +async def test_from_response_ensure_trailing_user_turn_end_to_end_full_mode() -> None: + """chained full mode context should get a synthetic trailing user turn before the second agent runs, + when the flag is enabled.""" + + first = _MessageCapturingAgent(id="first", name="First", reply_text="first reply") + second = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply") + + exec_a = AgentExecutor(first, id="exec_a") + exec_b = AgentExecutor(second, id="exec_b", context_mode="full", ensure_trailing_user_turn=True) + + wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).build() + + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = second.last_messages + assert len(seen) == 3 + assert seen[0].role == "user" and "hello" in (seen[0].text or "") + assert seen[1].role == "assistant" and "first reply" in (seen[1].text or "") + assert seen[2].role == "user" + + +async def test_from_response_without_ensure_trailing_user_turn_still_ends_on_assistant() -> None: + """Default (flag off) behavior is unchanged: chained context still ends on assistant.""" + first = _MessageCapturingAgent(id="first", name="First", reply_text="first reply") + second = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply") + + exec_a = AgentExecutor(first, id="exec_a") + exec_b = AgentExecutor(second, id="exec_b", context_mode="full") + + wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).build() + + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = second.last_messages + assert len(seen) == 2 + assert seen[-1].role == "assistant" + + async def test_checkpoint_save_does_not_include_context_mode() -> None: """on_checkpoint_save should not include context_mode in the saved state.""" agent = _CountingAgent(id="a", name="A") diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py b/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py index 6c9bcdcb112..eca0a72e12b 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py @@ -199,6 +199,7 @@ def __init__( context_mode: Literal["full", "last_agent", "custom"] | None = None, *, allow_direct_output: bool = False, + ensure_trailing_user_turn: bool = False, ) -> None: """Initialize the AgentApprovalExecutor. @@ -210,8 +211,13 @@ def __init__( downstream participant). Set this when this executor is the workflow's terminator — so the user-approved final response surfaces as a workflow ``output`` event. + ensure_trailing_user_turn: If True, appends a synthetic user continuation + message when the cache ends on an assistant message. Some chat-completions + providers return empty text when the prompt ends on ``assistant``. + Defaults to False. """ self._context_mode: Literal["full", "last_agent", "custom"] | None = context_mode + self._ensure_trailing_user_turn = ensure_trailing_user_turn self._description = agent.description super().__init__( @@ -235,6 +241,7 @@ def _build_workflow(self, agent: SupportsAgentRun, *, terminal: bool) -> Workflo agent_executor = AgentExecutor( agent, context_mode=self._context_mode, + ensure_trailing_user_turn=self._ensure_trailing_user_turn, ) request_info_cls = _TerminalAgentRequestInfoExecutor if terminal else AgentRequestInfoExecutor request_info_executor = request_info_cls(id="agent_request_info_executor") diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py b/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py index a92353998fb..d7147889150 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_sequential.py @@ -105,6 +105,7 @@ def __init__( chain_only_agent_responses: bool = False, output_from: Sequence[_ParticipantOutputSpecifier] | Literal["all"] | None = cast(Any, UNSET), intermediate_output_from: _ParticipantIntermediateOutputSelection = None, + ensure_trailing_user_turn: bool = False, ) -> None: """Initialize the SequentialBuilder. @@ -120,6 +121,11 @@ def __init__( intermediate_output_from: Optional participant names or instances whose ``yield_output`` calls surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant not selected by ``output_from``. Unlisted participant outputs are hidden. + ensure_trailing_user_turn: If True, whenever a participant would be run with a + conversation ending on an assistant message (e.g. chaining into the next agent), + a synthetic user continuation message is appended first. Some chat-completions + providers return empty text when the prompt ends on ``assistant``. Defaults to + False to preserve existing behavior. """ self._name = name or DEFAULT_WORKFLOW_NAME self._participants: list[SupportsAgentRun | Executor] = [] @@ -129,6 +135,7 @@ def __init__( self._request_info_filter: set[str] | None = None self._output_from = _coalesce_output_from(output_from=output_from) self._intermediate_output_from = _coerce_intermediate_output_from(intermediate_output_from) + self._ensure_trailing_user_turn: bool = ensure_trailing_user_turn self._set_participants(participants) @@ -223,10 +230,17 @@ def _resolve_participants(self) -> list[Executor]: p, context_mode=context_mode, allow_direct_output=(idx == last_idx), + ensure_trailing_user_turn=self._ensure_trailing_user_turn, ) ) else: - executors.append(AgentExecutor(p, context_mode=context_mode)) + executors.append( + AgentExecutor( + p, + context_mode=context_mode, + ensure_trailing_user_turn=self._ensure_trailing_user_turn, + ) + ) else: raise TypeError(f"Participants must be SupportsAgentRun or Executor instances. Got {type(p).__name__}.") diff --git a/python/packages/orchestrations/tests/test_orchestration_request_info.py b/python/packages/orchestrations/tests/test_orchestration_request_info.py index bbbaaaabd99..e3f8c969338 100644 --- a/python/packages/orchestrations/tests/test_orchestration_request_info.py +++ b/python/packages/orchestrations/tests/test_orchestration_request_info.py @@ -257,6 +257,38 @@ def test_propagate_request_enabled(self): assert executor._propagate_request is True # type: ignore + def test_ensure_trailing_user_turn_defaults_to_false(self): + """ensure_trailing_user_turn should default to False, preserving existing behavior.""" + agent = _TestAgent(id="test_id", name="test_agent", description="test description") + executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent)) + assert executor._ensure_trailing_user_turn is False # type: ignore + + def test_ensure_trailing_user_turn_stored_on_executor(self): + """ensure_trailing_user_turn=True should be stored on the AgentApprovalExecutor instance.""" + agent = _TestAgent(id="test_id", name="test_name", description="test description") + executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent), ensure_trailing_user_turn=True) + assert executor._ensure_trailing_user_turn is True # type: ignore + + def test_ensure_trailing_user_turn_propagated_to_inner_agent_executor(self): + """ensure_trailing_user_turn must be threaded through to the inner AgentExecutor + built by _build_workflow, since that inner executor is what actually runs the agent. + + without this propagation, the flag would be accepted + by AgentApprovalExecutor but silently dropped, so request-info-wrapped agents + would never get the synthetic trailing user turn. + """ + agent = _TestAgent(id="test_id", name="test_agent", description="Test description") + executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent), ensure_trailing_user_turn=True) + inner_agent_executor = executor.workflow.get_start_executor() + assert inner_agent_executor._ensure_trailing_user_turn is True # type: ignore + + def test_ensure_trailing_user_turn_false_propagated_to_inner_agent_executor(self): + """When left at the default, the inner AgentExecutor must also default to False.""" + agent = _TestAgent(id="test_id", name="test_agent", description="Test description") + executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent)) + inner_agent_executor = executor.workflow.get_start_executor() + assert inner_agent_executor._ensure_trailing_user_turn is False # type: ignore + def _build_with_request_info(builder_name: str, agents: list[str]) -> Any: from agent_framework.orchestrations import ConcurrentBuilder, GroupChatBuilder, SequentialBuilder diff --git a/python/packages/orchestrations/tests/test_sequential.py b/python/packages/orchestrations/tests/test_sequential.py index 4e4006456ed..2fe718e76a1 100644 --- a/python/packages/orchestrations/tests/test_sequential.py +++ b/python/packages/orchestrations/tests/test_sequential.py @@ -551,6 +551,103 @@ async def test_chain_only_agent_responses_three_agents() -> None: assert a3.last_messages[0].role == "assistant" and "A2 reply" in (a3.last_messages[0].text or "") +def test_sequential_builder_ensure_trailing_user_turn_defaults_to_false() -> None: + """ensure_trailing_user_turn should default to False, preserving existing behavior.""" + a1 = _EchoAgent(id="agent1", name="A1") + builder = SequentialBuilder(participants=[a1]) + assert builder._ensure_trailing_user_turn is False # type: ignore[attr-defined] + + +async def test_sequential_ensure_trailing_user_turn_disabled_second_agent_sees_assistant_ending() -> None: + """Default(flag off) second agent context still ends on first agent assistant message. + with the default setting the second participant receives a conversation ending on `assistant`""" + + a1 = _CapturingAgent(id="agent1", name="A1", reply_text="A1 reply") + a2 = _CapturingAgent(id="agent2", name="A2", reply_text="A2 reply") + + wf = SequentialBuilder(participants=[a1, a2]).build() + + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = a2.last_messages + assert len(seen) == 2 + assert seen[-1].role == "assistant" and "A1 reply" in (seen[-1].text or "") + + +async def test_sequential_ensure_trailing_user_enabled_appends_synthetic_user_message() -> None: + """with ensure_trailing_user_turn=True, the second agent receives a synthetic trailing + user message appended after the first agent assistant reply.""" + + a1 = _CapturingAgent(id="agent1", name="A1", reply_text="A1 reply") + a2 = _CapturingAgent(id="agent2", name="A2", reply_text="A2 reply") + + wf = SequentialBuilder(participants=[a1, a2], ensure_trailing_user_turn=True).build() + + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = a2.last_messages + assert len(seen) == 3 + assert seen[0].role == "user" and "hello" in (seen[0].text or "") + assert seen[1].role == "assistant" and "A1 reply" in (seen[1].text or "") + assert seen[2].role == "user" + + +async def test_sequential_ensure_trailing_user_turn_combined_with_chain_only_agent_responses() -> None: + """`chain_only_agent_responses = True` still produces an assistant-ending list + (`[assistant(writer)]`), so the synthetic turn must be appended there too. + """ + + a1 = _CapturingAgent(id="agent1", name="A1", reply_text="A1 reply") + a2 = _CapturingAgent(id="agent2", name="A2", reply_text="A2 reply") + + wf = SequentialBuilder( + participants=[a1, a2], + chain_only_agent_responses=True, + ensure_trailing_user_turn=True, + ).build() + + async for ev in wf.run("hello", stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = a2.last_messages + assert len(seen) == 2 + assert seen[0].role == "assistant" and "A1 reply" in (seen[0].text or "") + assert seen[1].role == "user" + + +async def test_sequential_ensure_trailing_user_turn_with_request_info_wrapped_agent() -> None: + """ensure_trailing_user_turn must also be honored for agents wrapped via with_request_info() + (i.e. threaded through AgentApprovalExecutor -> its inner AgentExecutor). + """ + from agent_framework_orchestrations._orchestration_request_info import AgentRequestInfoResponse + + a1 = _CapturingAgent(id="agent1", name="A1", reply_text="A1 reply") + a2 = _CapturingAgent(id="agent2", name="A2", reply_text="A2 reply") + + wf = SequentialBuilder(participants=[a1, a2], ensure_trailing_user_turn=True).with_request_info(agents=[a1]).build() + + request_events: list[Any] = [] + async for ev in wf.run("hello", stream=True): + if ev.type == "request_info" and isinstance(ev.data, AgentExecutorResponse): + request_events.append(ev) + + assert request_events + responses = {req.request_id: AgentRequestInfoResponse.approve() for req in request_events} + + async for ev in wf.run(stream=True, responses=responses): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + seen = a2.last_messages + assert len(seen) == 3 + assert seen[-1].role == "user" + + # --------------------------------------------------------------------------- # with_request_info tests # ---------------------------------------------------------------------------