diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index b7787736fa..737ddc6919 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -3,7 +3,7 @@ import logging import sys from collections.abc import Awaitable, Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Literal, cast from typing_extensions import Never @@ -37,10 +37,15 @@ class AgentExecutorRequest: messages: A list of chat messages to be processed by the agent. should_respond: A flag indicating whether the agent should respond to the messages. If False, the messages will be saved to the executor's cache but not sent to the agent. + fallback_messages: Messages used only if the agent would otherwise run with an empty + cache. The cache is cleared after every response, so a caller that cannot know + whether earlier context survived can supply a fallback here instead of guessing. + Ignored whenever the cache holds anything, so valid context is never displaced. """ messages: list[Message] should_respond: bool = True + fallback_messages: list[Message] = field(default_factory=lambda: list[Message]()) @dataclass @@ -208,6 +213,8 @@ async def run( self._cache.extend(request.messages) if request.should_respond: + if not self._cache: + self._cache.extend(request.fallback_messages) await self._run_agent_and_emit(ctx) @handler diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index ccb1e9425b..d4ff6d9ef9 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/python/packages/core/tests/workflow/test_agent_executor.py @@ -7,6 +7,7 @@ from agent_framework import ( AgentExecutor, + AgentExecutorRequest, AgentResponse, AgentResponseUpdate, AgentRunInputs, @@ -867,3 +868,38 @@ async def test_agent_executor_request_info_uses_user_input_request_id() -> None: # endregion Tool approval emission + + +async def _run_request(agent: _MessageCapturingAgent, request: AgentExecutorRequest) -> None: + executor = AgentExecutor(agent, id="exec") + wf = WorkflowBuilder(start_executor=executor).build() + async for ev in wf.run(request, stream=True): + if ev.type == "status" and ev.state == WorkflowRunState.IDLE: + break + + +async def test_fallback_messages_used_when_cache_is_empty() -> None: + """A request carrying no messages must fall back rather than invoke the agent with nothing.""" + agent = _MessageCapturingAgent(id="a", name="A") + + await _run_request( + agent, + AgentExecutorRequest(messages=[], fallback_messages=[Message("user", ["carry on"])]), + ) + + assert [m.text for m in agent.last_messages] == ["carry on"] + + +async def test_fallback_messages_ignored_when_cache_has_content() -> None: + """Fallback messages must never displace real context.""" + agent = _MessageCapturingAgent(id="a", name="A") + + await _run_request( + agent, + AgentExecutorRequest( + messages=[Message("user", ["real task"])], + fallback_messages=[Message("user", ["carry on"])], + ), + ) + + assert [m.text for m in agent.last_messages] == ["real task"] diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py index 16061b726b..212ef9f0bf 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_base_group_chat_orchestrator.py @@ -446,6 +446,7 @@ async def _send_request_to_participant( ctx: WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], *, additional_instruction: str | None = None, + fallback_instruction: str | None = None, metadata: dict[str, Any] | None = None, ) -> None: """Send a request to a participant. @@ -459,6 +460,9 @@ async def _send_request_to_participant( ctx: Workflow context for message routing additional_instruction: Optional additional instruction for the participant. This can be used to provide guidance to steer the participant's response. + fallback_instruction: Optional instruction applied only if the agent would + otherwise run with no messages at all. Ignored for custom executors, which + keep no message cache and always receive the full context envelope. metadata: Optional metadata dict Raises: @@ -469,7 +473,13 @@ async def _send_request_to_participant( messages: list[Message] = [] if additional_instruction: messages.append(Message(role="user", contents=[additional_instruction])) - request = AgentExecutorRequest(messages=messages, should_respond=True) + request = AgentExecutorRequest( + messages=messages, + should_respond=True, + fallback_messages=( + [Message(role="user", contents=[fallback_instruction])] if fallback_instruction else [] + ), + ) await ctx.send_message(request, target_id=target) await ctx.add_event( WorkflowEvent( diff --git a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py index bbb61edae2..76a5e526ce 100644 --- a/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py +++ b/python/packages/orchestrations/agent_framework_orchestrations/_group_chat.py @@ -69,6 +69,13 @@ logger = logging.getLogger(__name__) +# Applied only when the selected agent would otherwise run with no messages at all. This +# happens when it just spoke (it is excluded from the broadcast) or when cleaning stripped +# every broadcast message, since AgentExecutor clears its cache after each response. Some +# agents (for example A2AAgent) reject empty input. AgentExecutor drops this whenever real +# context is present, so it never displaces the conversation. +_CONTINUATION_DEFAULT_INSTRUCTION = "Continue the conversation." + @dataclass(frozen=True) class GroupChatState: @@ -233,6 +240,7 @@ async def _handle_response( await self._send_request_to_participant( next_speaker, cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), + fallback_instruction=_CONTINUATION_DEFAULT_INSTRUCTION, ) self._increment_round() @@ -413,6 +421,7 @@ async def _handle_response( # If not terminating, next_speaker must be provided thus will not be None agent_orchestration_output.next_speaker, # type: ignore[arg-type] cast(WorkflowContext[AgentExecutorRequest | GroupChatRequestMessage], ctx), + fallback_instruction=_CONTINUATION_DEFAULT_INSTRUCTION, ) self._increment_round() diff --git a/python/packages/orchestrations/tests/test_group_chat.py b/python/packages/orchestrations/tests/test_group_chat.py index 935111ccaa..9034e7e7c1 100644 --- a/python/packages/orchestrations/tests/test_group_chat.py +++ b/python/packages/orchestrations/tests/test_group_chat.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import json from collections.abc import AsyncIterable, Callable, Sequence from typing import Any, cast @@ -14,14 +15,19 @@ ChatResponse, ChatResponseUpdate, Content, + Executor, Message, + WorkflowContext, WorkflowEvent, WorkflowRunState, + handler, ) from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage +from agent_framework._workflows._message_utils import normalize_messages_input from agent_framework.orchestrations import ( AgentRequestInfoResponse, GroupChatBuilder, + GroupChatRequestMessage, GroupChatState, MagenticContext, MagenticManagerBase, @@ -30,6 +36,10 @@ ) from agent_framework_orchestrations import BaseGroupChatOrchestrator +from agent_framework_orchestrations._base_group_chat_orchestrator import GroupChatResponseMessage +from agent_framework_orchestrations._group_chat import ( # pyright: ignore[reportPrivateUsage] + _CONTINUATION_DEFAULT_INSTRUCTION, +) class StubAgent(BaseAgent): @@ -1098,3 +1108,253 @@ def invalid_factory() -> Any: # endregion + +# region Empty-input regression (issue #7456) + + +class RecordingStubAgent(BaseAgent): + """Stub agent that records received messages and rejects empty input, like ``A2AAgent``.""" + + def __init__(self, agent_name: str, reply_text: str, **kwargs: Any) -> None: + super().__init__(name=agent_name, description=f"Recording stub agent {agent_name}", **kwargs) + self._reply_text = reply_text + self.received_messages: list[list[Message]] = [] + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + normalized = normalize_messages_input(messages) + self.received_messages.append(normalized) + if not normalized: + raise ValueError("At least one message is required when starting a new task (no continuation_token).") + if stream: + return self._run_stream_impl() + return self._run_impl() + + async def _run_impl(self) -> AgentResponse[Any]: + response = Message(role="assistant", contents=[self._reply_text], author_name=self.name) + return AgentResponse(messages=[response]) + + async def _run_stream_impl(self) -> AsyncIterable[AgentResponseUpdate]: + yield AgentResponseUpdate( + contents=[Content.from_text(text=self._reply_text)], role="assistant", author_name=self.name + ) + + +class RepeatSpeakerManagerAgent(Agent): + """Manager that selects the same participant twice in a row, then terminates.""" + + def __init__(self, speaker: str, selections: int = 2) -> None: + super().__init__(client=cast(Any, MockChatClient()), name="manager_agent", description="Repeat manager") + self._speaker = speaker + self._selections = selections + self._call_count = 0 + + async def run( # type: ignore[override] # ty: ignore[invalid-method-override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + session: AgentSession | None = None, + **kwargs: Any, + ) -> AgentResponse[Any]: + if self._call_count < self._selections: + self._call_count += 1 + payload: dict[str, Any] = { + "terminate": False, + "reason": "Selecting agent", + "next_speaker": self._speaker, + "final_message": None, + } + else: + payload = { + "terminate": True, + "reason": "Task complete", + "next_speaker": None, + "final_message": "manager final", + } + return AgentResponse[Any]( + messages=[Message(role="assistant", contents=[json.dumps(payload)], author_name=self.name)], + value=payload, + ) + + +@pytest.mark.parametrize("stream", [False, True]) +async def test_group_chat_consecutive_selection_sends_non_empty_messages(stream: bool) -> None: + """A participant re-selected right after it spoke must not be invoked with empty input.""" + agent = RecordingStubAgent("solo", "reply from solo") + + workflow = GroupChatBuilder( + participants=[agent], + max_rounds=2, + selection_func=lambda state: "solo", + ).build() + + if stream: + async for _ in workflow.run("kickoff", stream=True): + pass + else: + await workflow.run("kickoff") + + assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" + assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1]), ( + "Second invocation should carry the continuation instruction" + ) + + +async def test_agent_orchestrator_consecutive_selection_sends_non_empty_messages() -> None: + """Same guarantee when an orchestrator agent (LLM) re-selects the participant that just spoke.""" + agent = RecordingStubAgent("solo", "reply from solo") + + workflow = GroupChatBuilder( + participants=[agent], + orchestrator_agent=RepeatSpeakerManagerAgent("solo"), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(agent.received_messages) == 2, "Expected the participant to be invoked twice" + assert all(received for received in agent.received_messages), "Participant was invoked with empty messages" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in agent.received_messages[1]), ( + "Second invocation should carry the continuation instruction" + ) + + +async def test_group_chat_alternating_selection_has_no_continuation_instruction() -> None: + """When a different participant speaks each round, no continuation instruction is added.""" + alpha = RecordingStubAgent("alpha", "reply from alpha") + beta = RecordingStubAgent("beta", "reply from beta") + + workflow = GroupChatBuilder( + participants=[alpha, beta], + max_rounds=2, + selection_func=make_sequence_selector(), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + for participant in (alpha, beta): + for received in participant.received_messages: + assert received, "Participant was invoked with empty messages" + assert all(_CONTINUATION_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + + +class ToolOnlyStubAgent(BaseAgent): + """Stub agent whose response holds only a function call, so cleaning strips it entirely.""" + + def __init__(self, agent_name: str, **kwargs: Any) -> None: + super().__init__(name=agent_name, description=f"Tool-only stub agent {agent_name}", **kwargs) + + def run( # type: ignore[override] + self, + messages: str | Content | Message | Sequence[str | Content | Message] | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Any: + return self._run_stream_impl() if stream else self._run_impl() + + def _contents(self) -> list[Content]: + return [Content.from_function_call(call_id=f"call-{self.name}", name="do_work", arguments={})] + + async def _run_impl(self) -> AgentResponse[Any]: + return AgentResponse(messages=[Message(role="assistant", contents=self._contents(), author_name=self.name)]) + + async def _run_stream_impl(self) -> AsyncIterable[AgentResponseUpdate]: + yield AgentResponseUpdate(contents=self._contents(), role="assistant", author_name=self.name) + + +async def test_group_chat_tool_only_first_speaker_keeps_initial_task() -> None: + """An empty broadcast must not displace context the next speaker already holds. + + ``clean_conversation_for_handoff`` drops messages with no text, so a tool-only first + response broadcasts nothing. The next speaker still has the initial task cached and must + receive it, rather than a continuation instruction that would replace it. + """ + alpha = RecordingStubAgent("alpha", "reply from alpha") + beta = ToolOnlyStubAgent("beta") + + speakers = iter(["beta", "alpha"]) + + workflow = GroupChatBuilder( + participants=[alpha, beta], + max_rounds=2, + selection_func=lambda state: next(speakers), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(alpha.received_messages) == 1, "Expected alpha to be invoked once" + received = alpha.received_messages[0] + assert any("kickoff" in (message.text or "") for message in received), "Initial task was dropped" + assert all(_CONTINUATION_DEFAULT_INSTRUCTION not in (message.text or "") for message in received) + + +async def test_group_chat_empty_broadcast_after_earlier_speaker_sends_non_empty_messages() -> None: + """A non-consecutive speaker whose cache is empty must still receive messages. + + ``alpha`` speaks and its cache is cleared. ``beta`` then returns tool-only content, which + ``clean_conversation_for_handoff`` strips to nothing, so the broadcast back to ``alpha`` + adds nothing. Re-selecting ``alpha`` would otherwise invoke it with no messages at all. + """ + alpha = RecordingStubAgent("alpha", "reply from alpha") + beta = ToolOnlyStubAgent("beta") + + speakers = iter(["alpha", "beta", "alpha"]) + + workflow = GroupChatBuilder( + participants=[alpha, beta], + max_rounds=3, + selection_func=lambda state: next(speakers), + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(alpha.received_messages) == 2, "Expected alpha to be invoked twice" + assert all(received for received in alpha.received_messages), "Participant was invoked with empty messages" + assert any(_CONTINUATION_DEFAULT_INSTRUCTION in (message.text or "") for message in alpha.received_messages[1]) + + +class RecordingCustomExecutor(Executor): + """Custom executor participant that records the request envelopes it receives.""" + + def __init__(self, name: str) -> None: + super().__init__(name) + self.received_instructions: list[str | None] = [] + + @handler + async def _respond(self, message: GroupChatRequestMessage, ctx: WorkflowContext[GroupChatResponseMessage]) -> None: + self.received_instructions.append(message.additional_instruction) + await ctx.send_message( + GroupChatResponseMessage(message=Message(role="assistant", contents=["custom reply"], author_name=self.id)) + ) + + +async def test_group_chat_consecutive_custom_executor_gets_no_continuation_instruction() -> None: + """Custom executors have no agent cache, so re-selecting one must not add the instruction.""" + custom = RecordingCustomExecutor("custom") + + workflow = GroupChatBuilder( + participants=[custom], + max_rounds=2, + selection_func=lambda state: "custom", + ).build() + + async for _ in workflow.run("kickoff", stream=True): + pass + + assert len(custom.received_instructions) == 2, "Expected the participant to be invoked twice" + assert custom.received_instructions == [None, None] + + +# endregion