diff --git a/python/packages/orchestrations/README.md b/python/packages/orchestrations/README.md index bbdb63a118a..8c2b3c4bb26 100644 --- a/python/packages/orchestrations/README.md +++ b/python/packages/orchestrations/README.md @@ -51,6 +51,19 @@ workflow = ( ) ``` +By default, each user response returns to the agent that requested it. To have the start agent re-evaluate every user +turn, disable return-to-previous routing: + +```python +workflow = ( + HandoffBuilder() + .participants([triage, billing, support]) + .with_start_agent(triage) + .enable_return_to_previous(False) + .build() +) +``` + ### GroupChatBuilder Orchestrator-directed multi-agent conversations: diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py b/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py index 490e85708a3..aadcb8ad6df 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_handoff.py @@ -241,6 +241,7 @@ def __init__( *, agent_session: AgentSession | None = None, is_start_agent: bool = False, + user_response_target_id: str | None = None, termination_condition: TerminationCondition | None = None, autonomous_mode: bool = False, autonomous_mode_prompt: str | None = None, @@ -254,6 +255,8 @@ def __init__( agent_session: Optional AgentSession that manages the agent's execution context is_start_agent: Whether this agent is the starting agent in the handoff workflow. There can only be one starting agent in a handoff workflow. + user_response_target_id: Optional executor ID that should handle user responses. + Defaults to this executor. termination_condition: Optional callable that determines when to terminate the workflow autonomous_mode: Whether the agent should operate involve external systems after a response that does not trigger a handoff or before the turn @@ -271,6 +274,7 @@ def __init__( self._handoff_targets = {handoff.target_id for handoff in handoffs} self._termination_condition = termination_condition self._is_start_agent = is_start_agent + self._user_response_target_id = user_response_target_id # Autonomous mode members self._autonomous_mode = autonomous_mode @@ -510,8 +514,15 @@ async def handle_response( # Broadcast the user response to all other agents await self._broadcast_messages(response, ctx) - # Append the user response messages to the cache + # Keep this executor synchronized even when another agent handles the response. self._cache.extend(response) + if self._user_response_target_id is not None and self._user_response_target_id != self.id: + await ctx.send_message( + AgentExecutorRequest(messages=[], should_respond=True), + target_id=self._user_response_target_id, + ) + return + await self._run_agent_and_emit(ctx) async def _broadcast_messages( @@ -623,6 +634,8 @@ class HandoffBuilder: Agents can hand off to other agents using `.add_handoff()`. This provides a decentralized approach to multi-agent collaboration. Handoffs can be configured using `.add_handoff`. If none are specified, all agents can hand off to all others by default (making a mesh topology). + By default, user responses return to the agent that requested them. Use + `.enable_return_to_previous(False)` to route user responses through the start agent instead. Participants must be ``Agent`` instances. ``SupportsAgentRun`` protocol implementors that are not ``Agent`` subclasses are not supported because handoff workflows require cloning, @@ -690,6 +703,7 @@ def __init__( # Handoff related members self._handoff_config: dict[str, set[HandoffConfiguration]] = {} + self._return_to_previous: bool = True # Checkpoint related members self._checkpoint_storage: CheckpointStorage | None = checkpoint_storage @@ -864,6 +878,33 @@ def with_start_agent(self, agent: Agent) -> "HandoffBuilder": return self + def enable_return_to_previous(self, enabled: bool = True) -> "HandoffBuilder": + """Configure whether user responses return to the agent that requested them. + + Return-to-previous routing is enabled by default. Disable it to route each user response + through the configured start agent so that agent can re-evaluate the conversation before + handing off again. + + Args: + enabled: Whether user responses should return to the agent that requested them. + + Returns: + Self for method chaining. + + Example: + + .. code-block:: python + + workflow = ( + HandoffBuilder(participants=[triage, billing, support]) + .with_start_agent(triage) + .enable_return_to_previous(False) + .build() + ) + """ + self._return_to_previous = enabled + return self + def with_autonomous_mode( self, *, @@ -1155,6 +1196,7 @@ def _resolve_executors( agent=agent, handoffs=handoffs.get(resolved_id, []), is_start_agent=(id == self._start_id), + user_response_target_id=None if self._return_to_previous else self._start_id, termination_condition=self._termination_condition, autonomous_mode=autonomous_mode, autonomous_mode_prompt=self._autonomous_mode_prompts.get(id, None), diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py index 099219a18c0..781e64b73b9 100644 --- a/python/packages/orchestrations/tests/test_handoff.py +++ b/python/packages/orchestrations/tests/test_handoff.py @@ -84,6 +84,7 @@ def __init__( self._name = name self._handoff_to = handoff_to self._call_index = 0 + self.received_messages: list[list[Message]] = [] def _inner_get_response( self, @@ -93,6 +94,7 @@ def _inner_get_response( options: Mapping[str, Any], **kwargs: Any, ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + self.received_messages.append(list(messages)) if stream: return self._build_streaming_response(options=dict(options)) @@ -303,6 +305,71 @@ def _request_text(event: WorkflowEvent[Any]) -> str: return messages[-1].text or "" +@pytest.mark.parametrize( + ("return_to_previous", "expected_start_agent_calls"), + [ + param(None, 1, id="default"), + param(True, 1, id="enabled"), + param(False, 2, id="disabled"), + ], +) +async def test_handoff_user_response_routing( + return_to_previous: bool | None, + expected_start_agent_calls: int, +) -> None: + """User responses return to the prior agent unless configured to revisit the start agent.""" + triage_client = MockChatClient(name="triage", handoff_to="specialist") + specialist_client = MockChatClient(name="specialist") + triage = Agent( + id="triage", + name="triage", + client=triage_client, + require_per_service_call_history_persistence=True, + ) + specialist = Agent( + id="specialist", + name="specialist", + client=specialist_client, + require_per_service_call_history_persistence=True, + ) + builder = HandoffBuilder( + participants=_as_handoff_agents(triage, specialist), + termination_condition=lambda _: False, + ).with_start_agent(_as_handoff_agent(triage)) + if return_to_previous is True: + builder.enable_return_to_previous() + elif return_to_previous is False: + builder.enable_return_to_previous(False) + workflow = builder.build() + + first_events = await _drain(workflow.run("Need specialist help", stream=True)) + first_request = _latest_request_info_event(first_events) + assert first_request.source_executor_id == specialist.name + assert len(triage_client.received_messages) == 1 + assert len(specialist_client.received_messages) == 1 + + second_events = await _drain( + workflow.run( + stream=True, + responses={first_request.request_id: [Message(role="user", contents=["Additional details"])]}, + ) + ) + second_request = _latest_request_info_event(second_events) + + assert second_request.source_executor_id == specialist.name + assert len(triage_client.received_messages) == expected_start_agent_calls + assert len(specialist_client.received_messages) == 2 + if return_to_previous is False: + assert any( + message.role == "user" and message.text == "Additional details" + for message in triage_client.received_messages[-1] + ) + assert any( + message.role == "user" and message.text == "Additional details" + for message in specialist_client.received_messages[-1] + ) + + async def test_resume_keeps_prior_user_context_for_same_agent() -> None: """Ensure same-agent request_info resumes retain prior turn context.""" refund_agent = Agent(