agent/agent-framework
2
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

[BREAKING] Python: Standardize orchestration outputs as list of ChatMessage. Allow agent as group chat manager. (#2291)

Browse Source 
* Standardize orchestration outputs as list of chatmessage. Add chat options to group chat prompt manager

* refactor group chat

* Improve group chat manager

* README Update

* Cleanup

* Add comment

* More cleanup

* Standardize termination condition for group chat

* Improvements on termination logic

* Fix tests

* Fix new line

* PR feedback

* Update ChatKit based on OpenAI type change

* Raise error if response format is not expected type

* Only one starting executor required. Add tests.

* Add magentic start executor test
This commit is contained in:
Evan Mattson
2025-11-26 16:51:04 +09:00
committed by GitHub
Unverified
parent ed53ba158b
commit 907d79ab3c
19 changed files with 1724 additions and 573 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/chatkit/agent_framework_chatkit/_converter.py
+8 -2
View File
@@ -27,6 +27,7 @@ from chatkit.types import (
EndOfTurnItem,
HiddenContextItem,
ImageAttachment,
SDKHiddenContextItem,
TaskItem,
ThreadItem,
UserMessageItem,
@@ -180,8 +181,10 @@ class ThreadItemConverter:
# Subclasses can override this method to provide custom handling
return None
def hidden_context_to_input(self, item: HiddenContextItem) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit HiddenContextItem to Agent Framework ChatMessage(s).
def hidden_context_to_input(
self, item: HiddenContextItem | SDKHiddenContextItem
) -> ChatMessage | list[ChatMessage] | None:
"""Convert a ChatKit HiddenContextItem or SDKHiddenContextItem to Agent Framework ChatMessage(s).
This method is called internally by `to_agent_input()`. Override this method
to customize how hidden context is converted.
@@ -522,6 +525,9 @@ class ThreadItemConverter:
case HiddenContextItem():
out = self.hidden_context_to_input(item) or []
return out if isinstance(out, list) else [out]
case SDKHiddenContextItem():
out = self.hidden_context_to_input(item) or []
return out if isinstance(out, list) else [out]
case _:
assert_never(item)
python/packages/chatkit/pyproject.toml
+1 -1
View File
@@ -23,7 +23,7 @@ classifiers = [
]
dependencies = [
"agent-framework-core",
"openai-chatkit>=1.1.0,<2.0.0",
"openai-chatkit>=1.4.0,<2.0.0",
]
[tool.uv]
python/packages/core/agent_framework/_workflows/__init__.py
+4
View File
@@ -61,6 +61,8 @@ from ._group_chat import (
GroupChatDirective,
GroupChatStateSnapshot,
ManagerDirectiveModel,
ManagerSelectionRequest,
ManagerSelectionResponse,
)
from ._handoff import HandoffBuilder, HandoffUserInputRequest
from ._magentic import (
@@ -147,6 +149,8 @@ __all__ = [
"MagenticPlanReviewReply",
"MagenticPlanReviewRequest",
"ManagerDirectiveModel",
"ManagerSelectionRequest",
"ManagerSelectionResponse",
"Message",
"OrchestrationState",
"RequestInfoEvent",
python/packages/core/agent_framework/_workflows/_group_chat.py
+647 -213
View File
File diff suppressed because it is too large Load Diff
python/packages/core/agent_framework/_workflows/_handoff.py
+3 -2
View File
@@ -1424,6 +1424,7 @@ class HandoffBuilder:
prompt=self._request_prompt,
id="handoff-user-input",
)
builder = WorkflowBuilder(name=self._name, description=self._description).set_start_executor(input_node)
specialist_aliases = {alias: exec_id for alias, exec_id in self._aliases.items() if exec_id in specialists}
@@ -1440,6 +1441,7 @@ class HandoffBuilder:
wiring = _GroupChatConfig(
manager=None,
manager_participant=None,
manager_name=self._starting_agent_id,
participants=participant_specs,
max_rounds=None,
@@ -1453,14 +1455,13 @@ class HandoffBuilder:
orchestrator_factory=_handoff_orchestrator_factory,
interceptors=(),
checkpoint_storage=self._checkpoint_storage,
builder=WorkflowBuilder(name=self._name, description=self._description),
builder=builder,
return_builder=True,
)
if not isinstance(result, tuple):
raise TypeError("Expected tuple from assemble_group_chat_workflow with return_builder=True")
builder, coordinator = result
builder = builder.set_start_executor(input_node)
builder = builder.add_edge(input_node, starting_executor)
builder = builder.add_edge(coordinator, user_gateway)
builder = builder.add_edge(user_gateway, coordinator)
python/packages/core/agent_framework/_workflows/_magentic.py
+24 -23
View File
@@ -961,7 +961,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _emit_orchestrator_message(
self,
ctx: WorkflowContext[Any, ChatMessage],
ctx: WorkflowContext[Any, list[ChatMessage]],
message: ChatMessage,
kind: str,
) -> None:
@@ -1110,7 +1110,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
self,
message: _MagenticStartMessage,
context: WorkflowContext[
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, ChatMessage
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, list[ChatMessage]
],
) -> None:
"""Handle the initial start message to begin orchestration."""
@@ -1145,7 +1145,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
# Start the inner loop
ctx2 = cast(
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
context,
)
await self._run_inner_loop(ctx2)
@@ -1155,7 +1155,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
self,
task_text: str,
context: WorkflowContext[
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, ChatMessage
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, list[ChatMessage]
],
) -> None:
await self.handle_start_message(_MagenticStartMessage.from_string(task_text), context)
@@ -1165,7 +1165,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
self,
task_message: ChatMessage,
context: WorkflowContext[
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, ChatMessage
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, list[ChatMessage]
],
) -> None:
await self.handle_start_message(_MagenticStartMessage(task_message), context)
@@ -1175,7 +1175,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
self,
conversation: list[ChatMessage],
context: WorkflowContext[
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, ChatMessage
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, list[ChatMessage]
],
) -> None:
await self.handle_start_message(_MagenticStartMessage(conversation), context)
@@ -1184,7 +1184,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def handle_response_message(
self,
message: _MagenticResponseMessage,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Handle responses from agents."""
if getattr(self, "_terminated", False):
@@ -1216,7 +1216,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
response: _MagenticPlanReviewReply,
context: WorkflowContext[
# may broadcast ledger next, or ask for another round of review
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, ChatMessage
_MagenticResponseMessage | _MagenticRequestMessage | _MagenticPlanReviewRequest, list[ChatMessage]
],
) -> None:
if getattr(self, "_terminated", False):
@@ -1262,7 +1262,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
# Enter the normal coordination loop
ctx2 = cast(
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
context,
)
await self._run_inner_loop(ctx2)
@@ -1289,7 +1289,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
self._context.chat_history.append(self._task_ledger)
# No further review requests; proceed directly into coordination
ctx2 = cast(
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
context,
)
await self._run_inner_loop(ctx2)
@@ -1324,7 +1324,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _run_outer_loop(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Run the outer orchestration loop - planning phase."""
if self._context is None:
@@ -1347,7 +1347,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _run_inner_loop(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Run the inner orchestration loop. Coordination phase. Serialized with a lock."""
if self._context is None or self._task_ledger is None:
@@ -1357,7 +1357,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _run_inner_loop_helper(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Run inner loop with exclusive access."""
# Narrow optional context for the remainder of this method
@@ -1442,7 +1442,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _reset_and_replan(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Reset context and replan."""
if self._context is None:
@@ -1468,7 +1468,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
async def _prepare_final_answer(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> None:
"""Prepare the final answer using the manager."""
if self._context is None:
@@ -1478,11 +1478,11 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
final_answer = await self._manager.prepare_final_answer(self._context.clone(deep=True))
# Emit a completed event for the workflow
await context.yield_output(final_answer)
await context.yield_output([final_answer])
async def _check_within_limits_or_complete(
self,
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, ChatMessage],
context: WorkflowContext[_MagenticResponseMessage | _MagenticRequestMessage, list[ChatMessage]],
) -> bool:
"""Check if orchestrator is within operational limits."""
if self._context is None:
@@ -1509,7 +1509,7 @@ class MagenticOrchestratorExecutor(BaseGroupChatOrchestrator):
)
# Yield the partial result and signal completion
await context.yield_output(partial_result)
await context.yield_output([partial_result])
return False
return True
@@ -2283,21 +2283,22 @@ class MagenticWorkflow:
return
# At this point, checkpoint is guaranteed to be WorkflowCheckpoint
executor_states: dict[str, Any] = checkpoint.shared_state.get(EXECUTOR_STATE_KEY, {})
executor_states = cast(dict[str, Any], checkpoint.shared_state.get(EXECUTOR_STATE_KEY, {}))
orchestrator_id = getattr(orchestrator, "id", "")
orchestrator_state = executor_states.get(orchestrator_id)
orchestrator_state = cast(Any, executor_states.get(orchestrator_id))
if orchestrator_state is None:
orchestrator_state = executor_states.get("magentic_orchestrator")
orchestrator_state = cast(Any, executor_states.get("magentic_orchestrator"))
if not isinstance(orchestrator_state, dict):
return
context_payload = orchestrator_state.get("magentic_context")
orchestrator_state_dict = cast(dict[str, Any], orchestrator_state)
context_payload = cast(Any, orchestrator_state_dict.get("magentic_context"))
if not isinstance(context_payload, dict):
return
context_dict = cast(dict[str, Any], context_payload)
restored_participants = context_dict.get("participant_descriptions")
restored_participants = cast(Any, context_dict.get("participant_descriptions"))
if not isinstance(restored_participants, dict):
return
python/packages/core/agent_framework/_workflows/_orchestrator_helpers.py
+4
View File
@@ -186,6 +186,10 @@ class ParticipantRegistry:
"""Check if a participant is registered."""
return name in self._participant_entry_ids
def is_participant_registered(self, name: str) -> bool:
"""Check if a participant is registered (alias for is_registered for compatibility)."""
return self.is_registered(name)
def all_participants(self) -> set[str]:
"""Get all registered participant names."""
return set(self._participant_entry_ids.keys())
python/packages/core/tests/workflow/test_group_chat.py
+366 -100
View File
@@ -1,9 +1,10 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import AsyncIterable, Callable
from typing import Any
from typing import Any, cast
import pytest
from pydantic import BaseModel
from agent_framework import (
MAGENTIC_EVENT_TYPE_AGENT_DELTA,
@@ -14,6 +15,7 @@ from agent_framework import (
AgentThread,
BaseAgent,
ChatMessage,
Executor,
GroupChatBuilder,
GroupChatDirective,
GroupChatStateSnapshot,
@@ -23,21 +25,27 @@ from agent_framework import (
Role,
TextContent,
Workflow,
WorkflowContext,
WorkflowOutputEvent,
handler,
)
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
from agent_framework._workflows._group_chat import (
GroupChatOrchestratorExecutor,
ManagerSelectionResponse,
_default_orchestrator_factory, # type: ignore
_default_participant_factory, # type: ignore
_GroupChatConfig, # type: ignore
_PromptBasedGroupChatManager, # type: ignore
_SpeakerSelectorAdapter, # type: ignore
assemble_group_chat_workflow,
)
from agent_framework._workflows._magentic import (
_MagenticProgressLedger, # type: ignore
_MagenticProgressLedgerItem, # type: ignore
_MagenticStartMessage, # type: ignore
)
from agent_framework._workflows._participant_utils import GroupChatParticipantSpec
from agent_framework._workflows._workflow_builder import WorkflowBuilder
class StubAgent(BaseAgent):
@@ -70,6 +78,73 @@ class StubAgent(BaseAgent):
return _stream()
class StubManagerAgent(BaseAgent):
def __init__(self) -> None:
super().__init__(name="manager_agent", description="Stub manager")
self._call_count = 0
async def run(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentRunResponse: # type: ignore[override]
if self._call_count == 0:
self._call_count += 1
payload = {"selected_participant": "agent", "finish": False, "final_message": None}
return AgentRunResponse(
messages=[
ChatMessage(
role=Role.ASSISTANT,
text='{"selected_participant": "agent", "finish": false}',
author_name=self.name,
)
],
value=payload,
)
payload = {"selected_participant": None, "finish": True, "final_message": "agent manager final"}
return AgentRunResponse(
messages=[
ChatMessage(
role=Role.ASSISTANT,
text='{"finish": true, "final_message": "agent manager final"}',
author_name=self.name,
)
],
value=payload,
)
def run_stream(
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentRunResponseUpdate]: # type: ignore[override]
if self._call_count == 0:
self._call_count += 1
async def _stream_initial() -> AsyncIterable[AgentRunResponseUpdate]:
yield AgentRunResponseUpdate(
contents=[TextContent(text='{"selected_participant": "agent", "finish": false}')],
role=Role.ASSISTANT,
author_name=self.name,
)
return _stream_initial()
async def _stream_final() -> AsyncIterable[AgentRunResponseUpdate]:
yield AgentRunResponseUpdate(
contents=[TextContent(text='{"finish": true, "final_message": "agent manager final"}')],
role=Role.ASSISTANT,
author_name=self.name,
)
return _stream_final()
def make_sequence_selector() -> Callable[[GroupChatStateSnapshot], Any]:
state_counter = {"value": 0}
@@ -123,6 +198,22 @@ class StubMagenticManager(MagenticManagerBase):
return ChatMessage(role=Role.ASSISTANT, text="final", author_name="magentic_manager")
class PassthroughExecutor(Executor):
@handler
async def forward(self, message: Any, ctx: WorkflowContext[Any]) -> None:
await ctx.send_message(message)
class CountingWorkflowBuilder(WorkflowBuilder):
def __init__(self) -> None:
super().__init__()
self.start_calls = 0
def set_start_executor(self, executor: Any) -> "CountingWorkflowBuilder":
self.start_calls += 1
return cast("CountingWorkflowBuilder", super().set_start_executor(executor))
async def test_group_chat_builder_basic_flow() -> None:
selector = make_sequence_selector()
alpha = StubAgent("alpha", "ack from alpha")
@@ -130,21 +221,23 @@ async def test_group_chat_builder_basic_flow() -> None:
workflow = (
GroupChatBuilder()
.select_speakers(selector, display_name="manager", final_message="done")
.set_select_speakers_func(selector, display_name="manager", final_message="done")
.participants(alpha=alpha, beta=beta)
.build()
)
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("coordinate task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert len(outputs) == 1
assert outputs[0].text == "done"
assert outputs[0].author_name == "manager"
assert len(outputs[0]) >= 1
# The final message should be "done" from the manager
assert outputs[0][-1].text == "done"
assert outputs[0][-1].author_name == "manager"
async def test_magentic_builder_returns_workflow_and_runs() -> None:
@@ -169,11 +262,13 @@ async def test_magentic_builder_returns_workflow_and_runs() -> None:
agent_event_count += 1
if isinstance(event, WorkflowOutputEvent):
msg = event.data
if isinstance(msg, ChatMessage):
outputs.append(msg)
if isinstance(msg, list):
outputs.append(cast(list[ChatMessage], msg))
assert outputs, "Expected a final output message"
final = outputs[-1]
conversation = outputs[-1]
assert len(conversation) >= 1
final = conversation[-1]
assert final.text == "final"
assert final.author_name == "magentic_manager"
assert orchestrator_event_count > 0, "Expected orchestrator events to be emitted"
@@ -187,7 +282,7 @@ async def test_group_chat_as_agent_accepts_conversation() -> None:
workflow = (
GroupChatBuilder()
.select_speakers(selector, display_name="manager", final_message="done")
.set_select_speakers_func(selector, display_name="manager", final_message="done")
.participants(alpha=alpha, beta=beta)
.build()
)
@@ -239,7 +334,7 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="participants must be configured before build"):
builder.build()
@@ -250,10 +345,10 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="already has a manager configured"):
builder.select_speakers(selector)
builder.set_select_speakers_func(selector)
def test_empty_participants_raises_error(self) -> None:
"""Test that empty participants list raises ValueError."""
@@ -261,7 +356,7 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="participants cannot be empty"):
builder.participants([])
@@ -274,7 +369,7 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="Duplicate participant name 'test'"):
builder.participants([agent1, agent2])
@@ -302,7 +397,7 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="must define a non-empty 'name' attribute"):
builder.participants([agent])
@@ -314,11 +409,53 @@ class TestGroupChatBuilder:
def selector(state: GroupChatStateSnapshot) -> str | None:
return None
builder = GroupChatBuilder().select_speakers(selector)
builder = GroupChatBuilder().set_select_speakers_func(selector)
with pytest.raises(ValueError, match="participant names must be non-empty strings"):
builder.participants({"": agent})
def test_assemble_group_chat_respects_existing_start_executor(self) -> None:
"""Ensure assemble_group_chat_workflow does not override preconfigured start executor."""
async def manager(_: GroupChatStateSnapshot) -> GroupChatDirective:
return GroupChatDirective(finish=True)
builder = CountingWorkflowBuilder()
entry = PassthroughExecutor(id="entry")
builder = builder.set_start_executor(entry)
participant = PassthroughExecutor(id="participant")
participant_spec = GroupChatParticipantSpec(
name="participant",
participant=participant,
description="participant",
)
wiring = _GroupChatConfig(
manager=manager,
manager_participant=None,
manager_name="manager",
participants={"participant": participant_spec},
max_rounds=None,
termination_condition=None,
participant_aliases={},
participant_executors={"participant": participant},
)
result = assemble_group_chat_workflow(
wiring=wiring,
participant_factory=_default_participant_factory,
orchestrator_factory=_default_orchestrator_factory,
builder=builder,
return_builder=True,
)
assert isinstance(result, tuple)
assembled_builder, _ = result
assert assembled_builder is builder
assert builder.start_calls == 1
assert assembled_builder._start_executor is entry # type: ignore
class TestGroupChatOrchestrator:
"""Tests for GroupChatOrchestratorExecutor core functionality."""
@@ -336,25 +473,116 @@ class TestGroupChatOrchestrator:
workflow = (
GroupChatBuilder()
.select_speakers(selector)
.set_select_speakers_func(selector)
.participants([agent])
.with_max_rounds(2) # Limit to 2 rounds
.build()
)
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
# Should have terminated due to max_rounds, expect at least one output
assert len(outputs) >= 1
# The final message should be about round limit
final_output = outputs[-1]
# The final message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "round limit" in final_output.text.lower()
async def test_termination_condition_halts_conversation(self) -> None:
"""Test that a custom termination condition stops the workflow."""
def selector(state: GroupChatStateSnapshot) -> str | None:
return "agent"
def termination_condition(conversation: list[ChatMessage]) -> bool:
replies = [msg for msg in conversation if msg.role == Role.ASSISTANT and msg.author_name == "agent"]
return len(replies) >= 2
agent = StubAgent("agent", "response")
workflow = (
GroupChatBuilder()
.set_select_speakers_func(selector)
.participants([agent])
.with_termination_condition(termination_condition)
.build()
)
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert outputs, "Expected termination to yield output"
conversation = outputs[-1]
agent_replies = [msg for msg in conversation if msg.author_name == "agent" and msg.role == Role.ASSISTANT]
assert len(agent_replies) == 2
final_output = conversation[-1]
assert final_output.author_name == "manager"
assert "termination condition" in final_output.text.lower()
async def test_termination_condition_uses_manager_final_message(self) -> None:
"""Test that manager-provided final message is used on termination."""
async def selector(state: GroupChatStateSnapshot) -> str | None:
return None
agent = StubAgent("agent", "response")
final_text = "manager summary on termination"
workflow = (
GroupChatBuilder()
.set_select_speakers_func(selector, final_message=final_text)
.participants([agent])
.with_termination_condition(lambda _: True)
.build()
)
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert outputs, "Expected termination to yield output"
conversation = outputs[-1]
assert conversation[-1].text == final_text
assert conversation[-1].author_name == "manager"
async def test_termination_condition_agent_manager_finalizes(self) -> None:
"""Test that agent-based manager can provide final message on termination."""
manager = StubManagerAgent()
worker = StubAgent("agent", "response")
workflow = (
GroupChatBuilder()
.set_manager(manager, display_name="Manager")
.participants([worker])
.with_termination_condition(lambda conv: any(msg.author_name == "agent" for msg in conv))
.build()
)
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert outputs, "Expected termination to yield output"
conversation = outputs[-1]
assert conversation[-1].text == "agent manager final"
assert conversation[-1].author_name == "Manager"
async def test_unknown_participant_error(self) -> None:
"""Test that _apply_directive raises error for unknown participants."""
@@ -363,7 +591,7 @@ class TestGroupChatOrchestrator:
agent = StubAgent("agent", "response")
workflow = GroupChatBuilder().select_speakers(selector).participants([agent]).build()
workflow = GroupChatBuilder().set_select_speakers_func(selector).participants([agent]).build()
with pytest.raises(ValueError, match="Manager selected unknown participant 'unknown_agent'"):
async for _ in workflow.run_stream("test task"):
@@ -379,7 +607,7 @@ class TestGroupChatOrchestrator:
agent = StubAgent("agent", "response")
# The _SpeakerSelectorAdapter will catch this and raise TypeError
workflow = GroupChatBuilder().select_speakers(bad_selector).participants([agent]).build() # type: ignore
workflow = GroupChatBuilder().set_select_speakers_func(bad_selector).participants([agent]).build() # type: ignore
# This should raise a TypeError because selector doesn't return str or None
with pytest.raises(TypeError, match="must return a participant name \\(str\\) or None"):
@@ -394,7 +622,7 @@ class TestGroupChatOrchestrator:
agent = StubAgent("agent", "response")
workflow = GroupChatBuilder().select_speakers(selector).participants([agent]).build()
workflow = GroupChatBuilder().set_select_speakers_func(selector).participants([agent]).build()
with pytest.raises(ValueError, match="requires at least one chat message"):
async for _ in workflow.run_stream([]):
@@ -529,69 +757,76 @@ class TestCheckpointing:
storage = InMemoryCheckpointStorage()
workflow = (
GroupChatBuilder().select_speakers(selector).participants([agent]).with_checkpointing(storage).build()
GroupChatBuilder()
.set_select_speakers_func(selector)
.participants([agent])
.with_checkpointing(storage)
.build()
)
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test task"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert len(outputs) == 1 # Should complete normally
class TestPromptBasedManager:
"""Tests for _PromptBasedGroupChatManager."""
class TestAgentManagerConfiguration:
"""Tests for agent-based manager configuration."""
async def test_manager_with_missing_next_agent_raises_error(self) -> None:
"""Test that manager directive without next_agent raises RuntimeError."""
async def test_set_manager_configures_response_format(self) -> None:
"""Ensure ChatAgent managers receive default ManagerSelectionResponse formatting."""
from unittest.mock import MagicMock
class MockChatClient:
async def get_response(self, messages: Any, response_format: Any = None) -> Any:
# Return response that has finish=False but no next_agent
class MockResponse:
def __init__(self) -> None:
self.value = {"finish": False, "next_agent": None}
self.messages: list[Any] = []
from agent_framework import ChatAgent
return MockResponse()
chat_client = MagicMock()
manager_agent = ChatAgent(chat_client=chat_client, name="Coordinator")
assert manager_agent.chat_options.response_format is None
manager = _PromptBasedGroupChatManager(MockChatClient()) # type: ignore
worker = StubAgent("worker", "response")
state = {
"participants": {"agent": "desc"},
"task": ChatMessage(role=Role.USER, text="test"),
"conversation": (),
}
builder = GroupChatBuilder().set_manager(manager_agent).participants([worker])
with pytest.raises(RuntimeError, match="missing next_agent while finish is False"):
await manager(state)
assert manager_agent.chat_options.response_format is ManagerSelectionResponse
assert builder._manager_participant is manager_agent # type: ignore[attr-defined]
async def test_manager_with_unknown_participant_raises_error(self) -> None:
"""Test that manager selecting unknown participant raises RuntimeError."""
async def test_set_manager_accepts_agent_manager(self) -> None:
"""Verify agent-based manager can be set and workflow builds."""
from unittest.mock import MagicMock
class MockChatClient:
async def get_response(self, messages: Any, response_format: Any = None) -> Any:
# Return response selecting unknown participant
class MockResponse:
def __init__(self) -> None:
self.value = {"finish": False, "next_agent": "unknown"}
self.messages: list[Any] = []
from agent_framework import ChatAgent
return MockResponse()
chat_client = MagicMock()
manager_agent = ChatAgent(chat_client=chat_client, name="Coordinator")
worker = StubAgent("worker", "response")
manager = _PromptBasedGroupChatManager(MockChatClient()) # type: ignore
builder = GroupChatBuilder().set_manager(manager_agent, display_name="Orchestrator")
builder = builder.participants([worker]).with_max_rounds(1)
state = {
"participants": {"agent": "desc"},
"task": ChatMessage(role=Role.USER, text="test"),
"conversation": (),
}
assert builder._manager_participant is manager_agent # type: ignore[attr-defined]
assert "worker" in builder._participants # type: ignore[attr-defined]
with pytest.raises(RuntimeError, match="Manager selected unknown participant 'unknown'"):
await manager(state)
async def test_set_manager_rejects_custom_response_format(self) -> None:
"""Reject custom response_format on ChatAgent managers."""
from unittest.mock import MagicMock
from agent_framework import ChatAgent
class CustomResponse(BaseModel):
value: str
chat_client = MagicMock()
manager_agent = ChatAgent(chat_client=chat_client, name="Coordinator", response_format=CustomResponse)
worker = StubAgent("worker", "response")
with pytest.raises(ValueError, match="response_format must be ManagerSelectionResponse"):
GroupChatBuilder().set_manager(manager_agent).participants([worker])
assert manager_agent.chat_options.response_format is CustomResponse
class TestFactoryFunctions:
@@ -599,9 +834,9 @@ class TestFactoryFunctions:
def test_default_orchestrator_factory_without_manager_raises_error(self) -> None:
"""Test that default factory requires manager to be set."""
config = _GroupChatConfig(manager=None, manager_name="test", participants={})
config = _GroupChatConfig(manager=None, manager_participant=None, manager_name="test", participants={})
with pytest.raises(RuntimeError, match="requires a manager to be set"):
with pytest.raises(RuntimeError, match="requires a manager to be configured"):
_default_orchestrator_factory(config)
@@ -619,14 +854,14 @@ class TestConversationHandling:
agent = StubAgent("agent", "response")
workflow = GroupChatBuilder().select_speakers(selector).participants([agent]).build()
workflow = GroupChatBuilder().set_select_speakers_func(selector).participants([agent]).build()
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test string"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert len(outputs) == 1
@@ -641,14 +876,14 @@ class TestConversationHandling:
agent = StubAgent("agent", "response")
workflow = GroupChatBuilder().select_speakers(selector).participants([agent]).build()
workflow = GroupChatBuilder().set_select_speakers_func(selector).participants([agent]).build()
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream(task_message):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert len(outputs) == 1
@@ -667,14 +902,14 @@ class TestConversationHandling:
agent = StubAgent("agent", "response")
workflow = GroupChatBuilder().select_speakers(selector).participants([agent]).build()
workflow = GroupChatBuilder().set_select_speakers_func(selector).participants([agent]).build()
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream(conversation):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
assert len(outputs) == 1
@@ -695,23 +930,25 @@ class TestRoundLimitEnforcement:
workflow = (
GroupChatBuilder()
.select_speakers(selector)
.set_select_speakers_func(selector)
.participants([agent])
.with_max_rounds(1) # Very low limit
.build()
)
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
# Should have at least one output (the round limit message)
assert len(outputs) >= 1
# The last message should be about round limit
final_output = outputs[-1]
# The last message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "round limit" in final_output.text.lower()
async def test_round_limit_in_ingest_participant_message(self) -> None:
@@ -728,23 +965,25 @@ class TestRoundLimitEnforcement:
workflow = (
GroupChatBuilder()
.select_speakers(selector)
.set_select_speakers_func(selector)
.participants([agent])
.with_max_rounds(1) # Hit limit after first response
.build()
)
outputs: list[ChatMessage] = []
outputs: list[list[ChatMessage]] = []
async for event in workflow.run_stream("test"):
if isinstance(event, WorkflowOutputEvent):
data = event.data
if isinstance(data, ChatMessage):
outputs.append(data)
if isinstance(data, list):
outputs.append(cast(list[ChatMessage], data))
# Should have at least one output (the round limit message)
assert len(outputs) >= 1
# The last message should be about round limit
final_output = outputs[-1]
# The last message in the conversation should be about round limit
conversation = outputs[-1]
assert len(conversation) >= 1
final_output = conversation[-1]
assert "round limit" in final_output.text.lower()
@@ -758,12 +997,12 @@ async def test_group_chat_checkpoint_runtime_only() -> None:
agent_b = StubAgent("agentB", "Reply from B")
selector = make_sequence_selector()
wf = GroupChatBuilder().participants([agent_a, agent_b]).select_speakers(selector).build()
wf = GroupChatBuilder().participants([agent_a, agent_b]).set_select_speakers_func(selector).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("runtime checkpoint test", checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
baseline_output = cast(list[ChatMessage], ev.data) if isinstance(ev.data, list) else None
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
@@ -794,7 +1033,7 @@ async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
wf = (
GroupChatBuilder()
.participants([agent_a, agent_b])
.select_speakers(selector)
.set_select_speakers_func(selector)
.with_checkpointing(buildtime_storage)
.build()
)
@@ -802,7 +1041,7 @@ async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("override test", checkpoint_storage=runtime_storage):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
baseline_output = cast(list[ChatMessage], ev.data) if isinstance(ev.data, list) else None
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
@@ -816,3 +1055,30 @@ async def test_group_chat_checkpoint_runtime_overrides_buildtime() -> None:
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
class _StubExecutor(Executor):
"""Minimal executor used to satisfy workflow wiring in tests."""
def __init__(self, id: str) -> None:
super().__init__(id=id)
@handler
async def handle(self, message: object, ctx: WorkflowContext[ChatMessage]) -> None:
await ctx.yield_output(message)
def test_set_manager_builds_with_agent_manager() -> None:
"""GroupChatBuilder should build when using an agent-based manager."""
manager = _StubExecutor("manager_executor")
participant = _StubExecutor("participant_executor")
workflow = (
GroupChatBuilder().set_manager(manager, display_name="Moderator").participants({"worker": participant}).build()
)
orchestrator = workflow.get_start_executor()
assert isinstance(orchestrator, GroupChatOrchestratorExecutor)
assert orchestrator._is_manager_agent()
python/packages/core/tests/workflow/test_handoff.py
+36
View File
@@ -23,7 +23,22 @@ from agent_framework import (
WorkflowOutputEvent,
)
from agent_framework._mcp import MCPTool
from agent_framework._workflows import _handoff as handoff_module # type: ignore
from agent_framework._workflows._handoff import _clone_chat_agent # type: ignore[reportPrivateUsage]
from agent_framework._workflows._workflow_builder import WorkflowBuilder
class _CountingWorkflowBuilder(WorkflowBuilder):
created: list["_CountingWorkflowBuilder"] = []
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.start_calls = 0
_CountingWorkflowBuilder.created.append(self)
def set_start_executor(self, executor: Any) -> "_CountingWorkflowBuilder": # type: ignore[override]
self.start_calls += 1
return cast("_CountingWorkflowBuilder", super().set_start_executor(executor))
@dataclass
@@ -478,6 +493,27 @@ async def test_return_to_previous_enabled():
assert len(specialist_a.calls) == 2, "Specialist A should handle follow-up with return_to_previous enabled"
def test_handoff_builder_sets_start_executor_once(monkeypatch: pytest.MonkeyPatch) -> None:
"""Ensure HandoffBuilder.build sets the start executor only once when assembling the workflow."""
_CountingWorkflowBuilder.created.clear()
monkeypatch.setattr(handoff_module, "WorkflowBuilder", _CountingWorkflowBuilder)
coordinator = _RecordingAgent(name="coordinator")
specialist = _RecordingAgent(name="specialist")
workflow = (
HandoffBuilder(participants=[coordinator, specialist])
.set_coordinator("coordinator")
.with_termination_condition(lambda conv: len(conv) > 0)
.build()
)
assert workflow is not None
assert _CountingWorkflowBuilder.created, "Expected CountingWorkflowBuilder to be instantiated"
builder = _CountingWorkflowBuilder.created[-1]
assert builder.start_calls == 1, "set_start_executor should be invoked exactly once"
async def test_tool_choice_preserved_from_agent_config():
"""Verify that agent-level tool_choice configuration is preserved and not overridden."""
from unittest.mock import AsyncMock
python/packages/core/tests/workflow/test_magentic.py
+44 -7
View File
@@ -33,6 +33,7 @@ from agent_framework import (
WorkflowStatusEvent,
handler,
)
from agent_framework._workflows import _group_chat as group_chat_module # type: ignore
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
from agent_framework._workflows._magentic import ( # type: ignore[reportPrivateUsage]
MagenticAgentExecutor,
@@ -42,6 +43,7 @@ from agent_framework._workflows._magentic import ( # type: ignore[reportPrivate
_MagenticProgressLedgerItem, # type: ignore
_MagenticStartMessage, # type: ignore
)
from agent_framework._workflows._workflow_builder import WorkflowBuilder
if sys.version_info >= (3, 12):
from typing import override
@@ -162,6 +164,19 @@ class FakeManager(MagenticManagerBase):
return ChatMessage(role=Role.ASSISTANT, text="FINAL", author_name="magentic_manager")
class _CountingWorkflowBuilder(WorkflowBuilder):
created: list["_CountingWorkflowBuilder"] = []
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.start_calls = 0
_CountingWorkflowBuilder.created.append(self)
def set_start_executor(self, executor: Any) -> "_CountingWorkflowBuilder": # type: ignore[override]
self.start_calls += 1
return cast("_CountingWorkflowBuilder", super().set_start_executor(executor))
async def test_standard_manager_plan_and_replan_combined_ledger():
manager = FakeManager(max_round_count=10, max_stall_count=3, max_reset_count=2)
ctx = MagenticContext(
@@ -210,7 +225,7 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
assert req_event is not None
completed = False
output: ChatMessage | None = None
output: list[ChatMessage] | None = None
async for ev in wf.send_responses_streaming(
responses={req_event.request_id: MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.APPROVE)}
):
@@ -222,7 +237,8 @@ async def test_magentic_workflow_plan_review_approval_to_completion():
break
assert completed
assert output is not None
assert isinstance(output, ChatMessage)
assert isinstance(output, list)
assert all(isinstance(msg, ChatMessage) for msg in output)
async def test_magentic_plan_review_approve_with_comments_replans_and_proceeds():
@@ -300,8 +316,10 @@ async def test_magentic_orchestrator_round_limit_produces_partial_result():
output_event = next((e for e in events if isinstance(e, WorkflowOutputEvent)), None)
assert output_event is not None
data = output_event.data
assert isinstance(data, ChatMessage)
assert data.role == Role.ASSISTANT
assert isinstance(data, list)
assert all(isinstance(msg, ChatMessage) for msg in data)
assert len(data) > 0
assert data[-1].role == Role.ASSISTANT
async def test_magentic_checkpoint_resume_round_trip():
@@ -374,6 +392,23 @@ class _DummyExec(Executor):
pass
def test_magentic_builder_sets_start_executor_once(monkeypatch: pytest.MonkeyPatch) -> None:
"""Ensure MagenticBuilder wiring sets the start executor only once."""
_CountingWorkflowBuilder.created.clear()
monkeypatch.setattr(group_chat_module, "WorkflowBuilder", _CountingWorkflowBuilder)
manager = FakeManager()
workflow = (
MagenticBuilder().participants(agentA=_DummyExec("agentA")).with_standard_manager(manager=manager).build()
)
assert workflow is not None
assert _CountingWorkflowBuilder.created, "Expected CountingWorkflowBuilder to be instantiated"
builder = _CountingWorkflowBuilder.created[-1]
assert builder.start_calls == 1, "set_start_executor should be called exactly once"
async def test_magentic_agent_executor_on_checkpoint_save_and_restore_roundtrip():
backing_executor = _DummyExec("backing")
agent_exec = MagenticAgentExecutor(backing_executor, "agentA")
@@ -746,9 +781,11 @@ async def test_magentic_stall_and_reset_successfully():
assert idle_status is not None
output_event = next((e for e in events if isinstance(e, WorkflowOutputEvent)), None)
assert output_event is not None
assert isinstance(output_event.data, ChatMessage)
assert output_event.data.text is not None
assert output_event.data.text == "re-ledger"
assert isinstance(output_event.data, list)
assert all(isinstance(msg, ChatMessage) for msg in output_event.data)
assert len(output_event.data) > 0
assert output_event.data[-1].text is not None
assert output_event.data[-1].text == "re-ledger"
async def test_magentic_checkpoint_runtime_only() -> None: