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: Move orchestrations to dedicated package (#3685)

Browse Source 
* Move orchestrations to dedicated package

* Merge main

* Fix markdown links

* Fix links
This commit is contained in:
Evan Mattson
2026-02-05 12:05:13 +09:00
committed by GitHub
Unverified
parent 10afb86213
commit 0daa7700c6
47 changed files with 1197 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/core/README.md
+1 -4
View File
@@ -213,10 +213,7 @@ if __name__ == "__main__":
asyncio.run(main())
```
**Note**: GroupChat, Sequential, and Concurrent orchestrations are available today. See examples in:
- [python/samples/getting_started/workflows/orchestration/](https://github.com/microsoft/agent-framework/tree/main/python/samples/getting_started/workflows/orchestration)
- [group_chat_simple_selector.py](https://github.com/microsoft/agent-framework/blob/main/python/samples/getting_started/workflows/orchestration/group_chat_simple_selector.py)
- [group_chat_prompt_based_manager.py](https://github.com/microsoft/agent-framework/blob/main/python/samples/getting_started/workflows/orchestration/group_chat_prompt_based_manager.py)
**Note**: Sequential, Concurrent, Group Chat, Handoff, and Magentic orchestrations are available. See examples in [orchestration samples](../../samples/getting_started/orchestrations).
## More Examples & Samples
python/packages/core/agent_framework/_workflows/__init__.py
-50
View File
@@ -20,7 +20,6 @@ from ._checkpoint import (
WorkflowCheckpoint,
)
from ._checkpoint_summary import WorkflowCheckpointSummary, get_checkpoint_summary
from ._concurrent import ConcurrentBuilder
from ._const import (
DEFAULT_MAX_ITERATIONS,
)
@@ -66,30 +65,6 @@ from ._executor import (
handler,
)
from ._function_executor import FunctionExecutor, executor
from ._group_chat import (
AgentBasedGroupChatOrchestrator,
GroupChatBuilder,
GroupChatState,
)
from ._handoff import HandoffAgentUserRequest, HandoffBuilder, HandoffSentEvent
from ._magentic import (
ORCH_MSG_KIND_INSTRUCTION,
ORCH_MSG_KIND_NOTICE,
ORCH_MSG_KIND_TASK_LEDGER,
ORCH_MSG_KIND_USER_TASK,
MagenticBuilder,
MagenticContext,
MagenticManagerBase,
MagenticOrchestrator,
MagenticOrchestratorEvent,
MagenticOrchestratorEventType,
MagenticPlanReviewRequest,
MagenticPlanReviewResponse,
MagenticProgressLedger,
MagenticProgressLedgerItem,
MagenticResetSignal,
StandardMagenticManager,
)
from ._orchestration_request_info import AgentRequestInfoResponse
from ._orchestration_state import OrchestrationState
from ._request_info_mixin import response_handler
@@ -99,7 +74,6 @@ from ._runner_context import (
Message,
RunnerContext,
)
from ._sequential import SequentialBuilder
from ._validation import (
EdgeDuplicationError,
GraphConnectivityError,
@@ -120,11 +94,6 @@ from ._workflow_executor import (
__all__ = [
"DEFAULT_MAX_ITERATIONS",
"ORCH_MSG_KIND_INSTRUCTION",
"ORCH_MSG_KIND_NOTICE",
"ORCH_MSG_KIND_TASK_LEDGER",
"ORCH_MSG_KIND_USER_TASK",
"AgentBasedGroupChatOrchestrator",
"AgentExecutor",
"AgentExecutorRequest",
"AgentExecutorResponse",
@@ -132,7 +101,6 @@ __all__ = [
"BaseGroupChatOrchestrator",
"Case",
"CheckpointStorage",
"ConcurrentBuilder",
"Default",
"Edge",
"EdgeCondition",
@@ -147,35 +115,17 @@ __all__ = [
"FileCheckpointStorage",
"FunctionExecutor",
"GraphConnectivityError",
"GroupChatBuilder",
"GroupChatRequestMessage",
"GroupChatRequestSentEvent",
"GroupChatResponseReceivedEvent",
"GroupChatState",
"HandoffAgentUserRequest",
"HandoffBuilder",
"HandoffSentEvent",
"InMemoryCheckpointStorage",
"InProcRunnerContext",
"MagenticBuilder",
"MagenticContext",
"MagenticManagerBase",
"MagenticOrchestrator",
"MagenticOrchestratorEvent",
"MagenticOrchestratorEventType",
"MagenticPlanReviewRequest",
"MagenticPlanReviewResponse",
"MagenticProgressLedger",
"MagenticProgressLedgerItem",
"MagenticResetSignal",
"Message",
"OrchestrationState",
"RequestInfoEvent",
"Runner",
"RunnerContext",
"SequentialBuilder",
"SingleEdgeGroup",
"StandardMagenticManager",
"SubWorkflowRequestMessage",
"SubWorkflowResponseMessage",
"SuperStepCompletedEvent",
python/packages/core/agent_framework/_workflows/_concurrent.py
-592
View File
@@ -1,592 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import inspect
import logging
from collections.abc import Callable, Sequence
from typing import Any
from typing_extensions import Never
from agent_framework import AgentProtocol, ChatMessage
from ._agent_executor import AgentExecutor, AgentExecutorRequest, AgentExecutorResponse
from ._agent_utils import resolve_agent_id
from ._checkpoint import CheckpointStorage
from ._executor import Executor, handler
from ._message_utils import normalize_messages_input
from ._orchestration_request_info import AgentApprovalExecutor
from ._workflow import Workflow
from ._workflow_builder import WorkflowBuilder
from ._workflow_context import WorkflowContext
logger = logging.getLogger(__name__)
"""Concurrent builder for agent-only fan-out/fan-in workflows.
This module provides a high-level, agent-focused API to quickly assemble a
parallel workflow with:
- a default dispatcher that broadcasts the input to all agent participants
- a default aggregator that combines all agent conversations and completes the workflow
Notes:
- Participants can be provided as AgentProtocol or Executor instances via `.participants()`,
or as factories returning AgentProtocol or Executor via `.register_participants()`.
- A custom aggregator can be provided as:
- an Executor instance (it should handle list[AgentExecutorResponse],
yield output), or
- a callback function with signature:
def cb(results: list[AgentExecutorResponse]) -> Any | None
def cb(results: list[AgentExecutorResponse], ctx: WorkflowContext) -> Any | None
The callback is wrapped in _CallbackAggregator.
If the callback returns a non-None value, _CallbackAggregator yields that as output.
If it returns None, the callback may have already yielded an output via ctx, so no further action is taken.
"""
class _DispatchToAllParticipants(Executor):
"""Broadcasts input to all downstream participants (via fan-out edges)."""
@handler
async def from_request(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
# No explicit target: edge routing delivers to all connected participants.
await ctx.send_message(request)
@handler
async def from_str(self, prompt: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
request = AgentExecutorRequest(messages=normalize_messages_input(prompt), should_respond=True)
await ctx.send_message(request)
@handler
async def from_message(self, message: ChatMessage, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
request = AgentExecutorRequest(messages=normalize_messages_input(message), should_respond=True)
await ctx.send_message(request)
@handler
async def from_messages(
self,
messages: list[str | ChatMessage],
ctx: WorkflowContext[AgentExecutorRequest],
) -> None:
request = AgentExecutorRequest(messages=normalize_messages_input(messages), should_respond=True)
await ctx.send_message(request)
class _AggregateAgentConversations(Executor):
"""Aggregates agent responses and completes with combined ChatMessages.
Emits a list[ChatMessage] shaped as:
[ single_user_prompt?, agent1_final_assistant, agent2_final_assistant, ... ]
- Extracts a single user prompt (first user message seen across results).
- For each result, selects the final assistant message (prefers agent_response.messages).
- Avoids duplicating the same user message per agent.
"""
@handler
async def aggregate(
self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, list[ChatMessage]]
) -> None:
if not results:
logger.error("Concurrent aggregator received empty results list")
raise ValueError("Aggregation failed: no results provided")
def _is_role(msg: Any, role: str) -> bool:
r = getattr(msg, "role", None)
if r is None:
return False
# Normalize both r and role to lowercase strings for comparison
r_str = str(r).lower() if isinstance(r, str) or hasattr(r, "__str__") else r
role_str = str(role).lower()
return r_str == role_str
prompt_message: ChatMessage | None = None
assistant_replies: list[ChatMessage] = []
for r in results:
resp_messages = list(getattr(r.agent_response, "messages", []) or [])
conv = r.full_conversation if r.full_conversation is not None else resp_messages
logger.debug(
f"Aggregating executor {getattr(r, 'executor_id', '<unknown>')}: "
f"{len(resp_messages)} response msgs, {len(conv)} conversation msgs"
)
# Capture a single user prompt (first encountered across any conversation)
if prompt_message is None:
found_user = next((m for m in conv if _is_role(m, "user")), None)
if found_user is not None:
prompt_message = found_user
# Pick the final assistant message from the response; fallback to conversation search
final_assistant = next((m for m in reversed(resp_messages) if _is_role(m, "assistant")), None)
if final_assistant is None:
final_assistant = next((m for m in reversed(conv) if _is_role(m, "assistant")), None)
if final_assistant is not None:
assistant_replies.append(final_assistant)
else:
logger.warning(
f"No assistant reply found for executor {getattr(r, 'executor_id', '<unknown>')}; skipping"
)
if not assistant_replies:
logger.error(f"Aggregation failed: no assistant replies found across {len(results)} results")
raise RuntimeError("Aggregation failed: no assistant replies found")
output: list[ChatMessage] = []
if prompt_message is not None:
output.append(prompt_message)
else:
logger.warning("No user prompt found in any conversation; emitting assistants only")
output.extend(assistant_replies)
await ctx.yield_output(output)
class _CallbackAggregator(Executor):
"""Wraps a Python callback as an aggregator.
Accepts either an async or sync callback with one of the signatures:
- (results: list[AgentExecutorResponse]) -> Any | None
- (results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None
Notes:
- Async callbacks are awaited directly.
- Sync callbacks are executed via asyncio.to_thread to avoid blocking the event loop.
- If the callback returns a non-None value, it is yielded as an output.
"""
def __init__(self, callback: Callable[..., Any], id: str | None = None) -> None:
derived_id = getattr(callback, "__name__", "") or ""
if not derived_id or derived_id == "<lambda>":
derived_id = f"{type(self).__name__}_unnamed"
super().__init__(id or derived_id)
self._callback = callback
self._param_count = len(inspect.signature(callback).parameters)
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, Any]) -> None:
# Call according to provided signature, always non-blocking for sync callbacks
if self._param_count >= 2:
if inspect.iscoroutinefunction(self._callback):
ret = await self._callback(results, ctx) # type: ignore[misc]
else:
ret = await asyncio.to_thread(self._callback, results, ctx)
else:
if inspect.iscoroutinefunction(self._callback):
ret = await self._callback(results) # type: ignore[misc]
else:
ret = await asyncio.to_thread(self._callback, results)
# If the callback returned a value, finalize the workflow with it
if ret is not None:
await ctx.yield_output(ret)
class ConcurrentBuilder:
r"""High-level builder for concurrent agent workflows.
- `participants([...])` accepts a list of AgentProtocol (recommended) or Executor.
- `register_participants([...])` accepts a list of factories for AgentProtocol (recommended)
or Executor factories
- `build()` wires: dispatcher -> fan-out -> participants -> fan-in -> aggregator.
- `with_aggregator(...)` overrides the default aggregator with an Executor or callback.
- `register_aggregator(...)` accepts a factory for an Executor as custom aggregator.
Usage:
.. code-block:: python
from agent_framework import ConcurrentBuilder
# Minimal: use default aggregator (returns list[ChatMessage])
workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).build()
# With agent factories
workflow = ConcurrentBuilder().register_participants([create_agent1, create_agent2, create_agent3]).build()
# Custom aggregator via callback (sync or async). The callback receives
# list[AgentExecutorResponse] and its return value becomes the workflow's output.
def summarize(results: list[AgentExecutorResponse]) -> str:
return " | ".join(r.agent_response.messages[-1].text for r in results)
workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_aggregator(summarize).build()
# Custom aggregator via a factory
class MyAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
await ctx.yield_output(" | ".join(r.agent_response.messages[-1].text for r in results))
workflow = (
ConcurrentBuilder()
.register_participants([create_agent1, create_agent2, create_agent3])
.register_aggregator(lambda: MyAggregator(id="my_aggregator"))
.build()
)
# Enable checkpoint persistence so runs can resume
workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_checkpointing(storage).build()
# Enable request info before aggregation
workflow = ConcurrentBuilder().participants([agent1, agent2]).with_request_info().build()
"""
def __init__(self) -> None:
self._participants: list[AgentProtocol | Executor] = []
self._participant_factories: list[Callable[[], AgentProtocol | Executor]] = []
self._aggregator: Executor | None = None
self._aggregator_factory: Callable[[], Executor] | None = None
self._checkpoint_storage: CheckpointStorage | None = None
self._request_info_enabled: bool = False
self._request_info_filter: set[str] | None = None
self._intermediate_outputs: bool = False
def register_participants(
self,
participant_factories: Sequence[Callable[[], AgentProtocol | Executor]],
) -> "ConcurrentBuilder":
r"""Define the parallel participants for this concurrent workflow.
Accepts factories (callables) that return AgentProtocol instances (e.g., created
by a chat client) or Executor instances. Each participant created by a factory
is wired as a parallel branch using fan-out edges from an internal dispatcher.
Args:
participant_factories: Sequence of callables returning AgentProtocol or Executor instances
Raises:
ValueError: if `participant_factories` is empty or `.participants()`
or `.register_participants()` were already called
Example:
.. code-block:: python
def create_researcher() -> ChatAgent:
return ...
def create_marketer() -> ChatAgent:
return ...
def create_legal() -> ChatAgent:
return ...
class MyCustomExecutor(Executor): ...
wf = ConcurrentBuilder().register_participants([create_researcher, create_marketer, create_legal]).build()
# Mixing agent(s) and executor(s) is supported
wf2 = ConcurrentBuilder().register_participants([create_researcher, MyCustomExecutor]).build()
"""
if self._participants:
raise ValueError("Cannot mix .participants() and .register_participants() in the same builder instance.")
if self._participant_factories:
raise ValueError("register_participants() has already been called on this builder instance.")
if not participant_factories:
raise ValueError("participant_factories cannot be empty")
self._participant_factories = list(participant_factories)
return self
def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "ConcurrentBuilder":
r"""Define the parallel participants for this concurrent workflow.
Accepts AgentProtocol instances (e.g., created by a chat client) or Executor
instances. Each participant is wired as a parallel branch using fan-out edges
from an internal dispatcher.
Args:
participants: Sequence of AgentProtocol or Executor instances
Raises:
ValueError: if `participants` is empty, contains duplicates, or `.register_participants()`
or `.participants()` were already called
TypeError: if any entry is not AgentProtocol or Executor
Example:
.. code-block:: python
wf = ConcurrentBuilder().participants([researcher_agent, marketer_agent, legal_agent]).build()
# Mixing agent(s) and executor(s) is supported
wf2 = ConcurrentBuilder().participants([researcher_agent, my_custom_executor]).build()
"""
if self._participant_factories:
raise ValueError("Cannot mix .participants() and .register_participants() in the same builder instance.")
if self._participants:
raise ValueError("participants() has already been called on this builder instance.")
if not participants:
raise ValueError("participants cannot be empty")
# Defensive duplicate detection
seen_agent_ids: set[int] = set()
seen_executor_ids: set[str] = set()
for p in participants:
if isinstance(p, Executor):
if p.id in seen_executor_ids:
raise ValueError(f"Duplicate executor participant detected: id '{p.id}'")
seen_executor_ids.add(p.id)
elif isinstance(p, AgentProtocol):
pid = id(p)
if pid in seen_agent_ids:
raise ValueError("Duplicate agent participant detected (same agent instance provided twice)")
seen_agent_ids.add(pid)
else:
raise TypeError(f"participants must be AgentProtocol or Executor instances; got {type(p).__name__}")
self._participants = list(participants)
return self
def register_aggregator(self, aggregator_factory: Callable[[], Executor]) -> "ConcurrentBuilder":
r"""Define a custom aggregator for this concurrent workflow.
Accepts a factory (callable) that returns an Executor instance. The executor
should handle `list[AgentExecutorResponse]` and yield output using `ctx.yield_output(...)`.
Args:
aggregator_factory: Callable that returns an Executor instance
Example:
.. code-block:: python
class MyCustomExecutor(Executor): ...
wf = (
ConcurrentBuilder()
.register_participants([create_researcher, create_marketer, create_legal])
.register_aggregator(lambda: MyCustomExecutor(id="my_aggregator"))
.build()
)
"""
if self._aggregator is not None:
raise ValueError(
"Cannot mix .with_aggregator(...) and .register_aggregator(...) in the same builder instance."
)
if self._aggregator_factory is not None:
raise ValueError("register_aggregator() has already been called on this builder instance.")
self._aggregator_factory = aggregator_factory
return self
def with_aggregator(
self,
aggregator: Executor
| Callable[[list[AgentExecutorResponse]], Any]
| Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any],
) -> "ConcurrentBuilder":
r"""Override the default aggregator with an executor or a callback.
- Executor: must handle `list[AgentExecutorResponse]` and yield output using `ctx.yield_output(...)`
- Callback: sync or async callable with one of the signatures:
`(results: list[AgentExecutorResponse]) -> Any | None` or
`(results: list[AgentExecutorResponse], ctx: WorkflowContext) -> Any | None`.
If the callback returns a non-None value, it becomes the workflow's output.
Args:
aggregator: Executor instance, or callback function
Example:
.. code-block:: python
# Executor-based aggregator
class CustomAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext) -> None:
await ctx.yield_output(" | ".join(r.agent_response.messages[-1].text for r in results))
wf = ConcurrentBuilder().participants([a1, a2, a3]).with_aggregator(CustomAggregator()).build()
# Callback-based aggregator (string result)
async def summarize(results: list[AgentExecutorResponse]) -> str:
return " | ".join(r.agent_response.messages[-1].text for r in results)
wf = ConcurrentBuilder().participants([a1, a2, a3]).with_aggregator(summarize).build()
# Callback-based aggregator (yield result)
async def summarize(results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
await ctx.yield_output(" | ".join(r.agent_response.messages[-1].text for r in results))
wf = ConcurrentBuilder().participants([a1, a2, a3]).with_aggregator(summarize).build()
"""
if self._aggregator_factory is not None:
raise ValueError(
"Cannot mix .with_aggregator(...) and .register_aggregator(...) in the same builder instance."
)
if self._aggregator is not None:
raise ValueError("with_aggregator() has already been called on this builder instance.")
if isinstance(aggregator, Executor):
self._aggregator = aggregator
elif callable(aggregator):
self._aggregator = _CallbackAggregator(aggregator)
else:
raise TypeError("aggregator must be an Executor or a callable")
return self
def with_checkpointing(self, checkpoint_storage: CheckpointStorage) -> "ConcurrentBuilder":
"""Enable checkpoint persistence using the provided storage backend.
Args:
checkpoint_storage: CheckpointStorage instance for persisting workflow state
"""
self._checkpoint_storage = checkpoint_storage
return self
def with_request_info(
self,
*,
agents: Sequence[str | AgentProtocol] | None = None,
) -> "ConcurrentBuilder":
"""Enable request info after agent participant responses.
This enables human-in-the-loop (HIL) scenarios for the sequential orchestration.
When enabled, the workflow pauses after each agent participant runs, emitting
a RequestInfoEvent that allows the caller to review the conversation and optionally
inject guidance for the agent participant to iterate. The caller provides input via
the standard response_handler/request_info pattern.
Simulated flow with HIL:
Input -> [Agent Participant <-> Request Info] -> [Agent Participant <-> Request Info] -> ...
Note: This is only available for agent participants. Executor participants can incorporate
request info handling in their own implementation if desired.
Args:
agents: Optional list of agents names or agent factories to enable request info for.
If None, enables HIL for all agent participants.
Returns:
Self for fluent chaining
"""
from ._orchestration_request_info import resolve_request_info_filter
self._request_info_enabled = True
self._request_info_filter = resolve_request_info_filter(list(agents) if agents else None)
return self
def with_intermediate_outputs(self) -> "ConcurrentBuilder":
"""Enable intermediate outputs from agent participants before aggregation.
When enabled, the workflow returns each agent participant's response or yields
streaming updates as they become available. The output of the aggregator will
always be available as the final output of the workflow.
Returns:
Self for fluent chaining
"""
self._intermediate_outputs = True
return self
def _resolve_participants(self) -> list[Executor]:
"""Resolve participant instances into Executor objects."""
if not self._participants and not self._participant_factories:
raise ValueError("No participants provided. Call .participants() or .register_participants() first.")
# We don't need to check if both are set since that is handled in the respective methods
participants: list[Executor | AgentProtocol] = []
if self._participant_factories:
# Resolve the participant factories now. This doesn't break the factory pattern
# since the Sequential builder still creates new instances per workflow build.
for factory in self._participant_factories:
p = factory()
participants.append(p)
else:
participants = self._participants
executors: list[Executor] = []
for p in participants:
if isinstance(p, Executor):
executors.append(p)
elif isinstance(p, AgentProtocol):
if self._request_info_enabled and (
not self._request_info_filter or resolve_agent_id(p) in self._request_info_filter
):
# Handle request info enabled agents
executors.append(AgentApprovalExecutor(p))
else:
executors.append(AgentExecutor(p))
else:
raise TypeError(f"Participants must be AgentProtocol or Executor instances. Got {type(p).__name__}.")
return executors
def build(self) -> Workflow:
r"""Build and validate the concurrent workflow.
Wiring pattern:
- Dispatcher (internal) fans out the input to all `participants`
- Fan-in collects `AgentExecutorResponse` objects from all participants
- If request info is enabled, the orchestration emits a request info event with outputs from all participants
before sending the outputs to the aggregator
- Aggregator yields output and the workflow becomes idle. The output is either:
- list[ChatMessage] (default aggregator: one user + one assistant per agent)
- custom payload from the provided aggregator
Returns:
Workflow: a ready-to-run workflow instance
Raises:
ValueError: if no participants were defined
Example:
.. code-block:: python
workflow = ConcurrentBuilder().participants([agent1, agent2]).build()
"""
# Internal nodes
dispatcher = _DispatchToAllParticipants(id="dispatcher")
aggregator = (
self._aggregator
if self._aggregator is not None
else (
self._aggregator_factory()
if self._aggregator_factory is not None
else _AggregateAgentConversations(id="aggregator")
)
)
# Resolve participants and participant factories to executors
participants: list[Executor] = self._resolve_participants()
builder = WorkflowBuilder()
builder.set_start_executor(dispatcher)
# Fan-out for parallel execution
builder.add_fan_out_edges(dispatcher, participants)
# Direct fan-in to aggregator
builder.add_fan_in_edges(participants, aggregator)
if not self._intermediate_outputs:
# Constrain output to aggregator only
builder = builder.with_output_from([aggregator])
if self._checkpoint_storage is not None:
builder = builder.with_checkpointing(self._checkpoint_storage)
return builder.build()
python/packages/core/agent_framework/_workflows/_group_chat.py
-1016
View File
File diff suppressed because it is too large Load Diff
python/packages/core/agent_framework/_workflows/_handoff.py
-1229
View File
File diff suppressed because it is too large Load Diff
python/packages/core/agent_framework/_workflows/_magentic.py
-2003
View File
File diff suppressed because it is too large Load Diff
python/packages/core/agent_framework/_workflows/_sequential.py
-329
View File
@@ -1,329 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
"""Sequential builder for agent/executor workflows with shared conversation context.
This module provides a high-level, agent-focused API to assemble a sequential
workflow where:
- Participants can be provided as AgentProtocol or Executor instances via `.participants()`,
or as factories returning AgentProtocol or Executor via `.register_participants()`
- A shared conversation context (list[ChatMessage]) is passed along the chain
- Agents append their assistant messages to the context
- Custom executors can transform or summarize and return a refined context
- The workflow finishes with the final context produced by the last participant
Typical wiring:
input -> _InputToConversation -> participant1 -> (agent? -> _ResponseToConversation) -> ... -> participantN -> _EndWithConversation
Notes:
- Participants can mix AgentProtocol and Executor objects
- Agents are auto-wrapped by WorkflowBuilder as AgentExecutor (unless already wrapped)
- AgentExecutor produces AgentExecutorResponse; _ResponseToConversation converts this to list[ChatMessage]
- Non-agent executors must define a handler that consumes `list[ChatMessage]` and sends back
the updated `list[ChatMessage]` via their workflow context
Why include the small internal adapter executors?
- Input normalization ("input-conversation"): ensures the workflow always starts with a
`list[ChatMessage]` regardless of whether callers pass a `str`, a single `ChatMessage`,
or a list. This keeps the first hop strongly typed and avoids boilerplate in participants.
- Agent response adaptation ("to-conversation:<participant>"): agents (via AgentExecutor)
emit `AgentExecutorResponse`. The adapter converts that to a `list[ChatMessage]`
using `full_conversation` so original prompts aren't lost when chaining.
- Result output ("end"): yields the final conversation list and the workflow becomes idle
giving a consistent terminal payload shape for both agents and custom executors.
These adapters are first-class executors by design so they are type-checked at edges,
observable (ExecutorInvoke/Completed events), and easily testable/reusable. Their IDs are
deterministic and self-describing (for example, "to-conversation:writer") to reduce event-log
confusion and to mirror how the concurrent builder uses explicit dispatcher/aggregator nodes.
""" # noqa: E501
import logging
from collections.abc import Callable, Sequence
from typing import Any
from agent_framework import AgentProtocol, ChatMessage
from ._agent_executor import (
AgentExecutor,
AgentExecutorResponse,
)
from ._agent_utils import resolve_agent_id
from ._checkpoint import CheckpointStorage
from ._executor import (
Executor,
handler,
)
from ._message_utils import normalize_messages_input
from ._orchestration_request_info import AgentApprovalExecutor
from ._workflow import Workflow
from ._workflow_builder import WorkflowBuilder
from ._workflow_context import WorkflowContext
logger = logging.getLogger(__name__)
class _InputToConversation(Executor):
"""Normalizes initial input into a list[ChatMessage] conversation."""
@handler
async def from_str(self, prompt: str, ctx: WorkflowContext[list[ChatMessage]]) -> None:
await ctx.send_message(normalize_messages_input(prompt))
@handler
async def from_message(self, message: ChatMessage, ctx: WorkflowContext[list[ChatMessage]]) -> None:
await ctx.send_message(normalize_messages_input(message))
@handler
async def from_messages(self, messages: list[str | ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
await ctx.send_message(normalize_messages_input(messages))
class _EndWithConversation(Executor):
"""Terminates the workflow by emitting the final conversation context."""
@handler
async def end_with_messages(
self,
conversation: list[ChatMessage],
ctx: WorkflowContext[Any, list[ChatMessage]],
) -> None:
"""Handler for ending with a list of ChatMessage.
This is used when the last participant is a custom executor.
"""
await ctx.yield_output(list(conversation))
@handler
async def end_with_agent_executor_response(
self,
response: AgentExecutorResponse,
ctx: WorkflowContext[Any, list[ChatMessage] | None],
) -> None:
"""Handle case where last participant is an agent.
The agent is wrapped by AgentExecutor and emits AgentExecutorResponse.
"""
await ctx.yield_output(response.full_conversation)
class SequentialBuilder:
r"""High-level builder for sequential agent/executor workflows with shared context.
- `participants([...])` accepts a list of AgentProtocol (recommended) or Executor instances
- `register_participants([...])` accepts a list of factories for AgentProtocol (recommended)
or Executor factories
- Executors must define a handler that consumes list[ChatMessage] and sends out a list[ChatMessage]
- The workflow wires participants in order, passing a list[ChatMessage] down the chain
- Agents append their assistant messages to the conversation
- Custom executors can transform/summarize and return a list[ChatMessage]
- The final output is the conversation produced by the last participant
Usage:
.. code-block:: python
from agent_framework import SequentialBuilder
# With agent instances
workflow = SequentialBuilder().participants([agent1, agent2, summarizer_exec]).build()
# With agent factories
workflow = (
SequentialBuilder().register_participants([create_agent1, create_agent2, create_summarizer_exec]).build()
)
# Enable checkpoint persistence
workflow = SequentialBuilder().participants([agent1, agent2]).with_checkpointing(storage).build()
# Enable request info for mid-workflow feedback (pauses before each agent)
workflow = SequentialBuilder().participants([agent1, agent2]).with_request_info().build()
# Enable request info only for specific agents
workflow = (
SequentialBuilder()
.participants([agent1, agent2, agent3])
.with_request_info(agents=[agent2]) # Only pause before agent2
.build()
)
"""
def __init__(self) -> None:
self._participants: list[AgentProtocol | Executor] = []
self._participant_factories: list[Callable[[], AgentProtocol | Executor]] = []
self._checkpoint_storage: CheckpointStorage | None = None
self._request_info_enabled: bool = False
self._request_info_filter: set[str] | None = None
self._intermediate_outputs: bool = False
def register_participants(
self,
participant_factories: Sequence[Callable[[], AgentProtocol | Executor]],
) -> "SequentialBuilder":
"""Register participant factories for this sequential workflow."""
if self._participants:
raise ValueError("Cannot mix .participants() and .register_participants() in the same builder instance.")
if self._participant_factories:
raise ValueError("register_participants() has already been called on this builder instance.")
if not participant_factories:
raise ValueError("participant_factories cannot be empty")
self._participant_factories = list(participant_factories)
return self
def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "SequentialBuilder":
"""Define the ordered participants for this sequential workflow.
Accepts AgentProtocol instances (auto-wrapped as AgentExecutor) or Executor instances.
Raises if empty or duplicates are provided for clarity.
"""
if self._participant_factories:
raise ValueError("Cannot mix .participants() and .register_participants() in the same builder instance.")
if self._participants:
raise ValueError("participants() has already been called on this builder instance.")
if not participants:
raise ValueError("participants cannot be empty")
# Defensive duplicate detection
seen_agent_ids: set[int] = set()
seen_executor_ids: set[str] = set()
for p in participants:
if isinstance(p, Executor):
if p.id in seen_executor_ids:
raise ValueError(f"Duplicate executor participant detected: id '{p.id}'")
seen_executor_ids.add(p.id)
else:
# Treat non-Executor as agent-like (AgentProtocol). Structural checks can be brittle at runtime.
pid = id(p)
if pid in seen_agent_ids:
raise ValueError("Duplicate agent participant detected (same agent instance provided twice)")
seen_agent_ids.add(pid)
self._participants = list(participants)
return self
def with_checkpointing(self, checkpoint_storage: CheckpointStorage) -> "SequentialBuilder":
"""Enable checkpointing for the built workflow using the provided storage."""
self._checkpoint_storage = checkpoint_storage
return self
def with_request_info(
self,
*,
agents: Sequence[str | AgentProtocol] | None = None,
) -> "SequentialBuilder":
"""Enable request info after agent participant responses.
This enables human-in-the-loop (HIL) scenarios for the sequential orchestration.
When enabled, the workflow pauses after each agent participant runs, emitting
a RequestInfoEvent that allows the caller to review the conversation and optionally
inject guidance for the agent participant to iterate. The caller provides input via
the standard response_handler/request_info pattern.
Simulated flow with HIL:
Input -> [Agent Participant <-> Request Info] -> [Agent Participant <-> Request Info] -> ...
Note: This is only available for agent participants. Executor participants can incorporate
request info handling in their own implementation if desired.
Args:
agents: Optional list of agents names or agent factories to enable request info for.
If None, enables HIL for all agent participants.
Returns:
Self for fluent chaining
"""
from ._orchestration_request_info import resolve_request_info_filter
self._request_info_enabled = True
self._request_info_filter = resolve_request_info_filter(list(agents) if agents else None)
return self
def with_intermediate_outputs(self) -> "SequentialBuilder":
"""Enable intermediate outputs from agent participants.
When enabled, the workflow returns each agent participant's response or yields
streaming updates as they become available. The output of the last participant
will always be available as the final output of the workflow.
Returns:
Self for fluent chaining
"""
self._intermediate_outputs = True
return self
def _resolve_participants(self) -> list[Executor]:
"""Resolve participant instances into Executor objects."""
if not self._participants and not self._participant_factories:
raise ValueError("No participants provided. Call .participants() or .register_participants() first.")
# We don't need to check if both are set since that is handled in the respective methods
participants: list[Executor | AgentProtocol] = []
if self._participant_factories:
# Resolve the participant factories now. This doesn't break the factory pattern
# since the Sequential builder still creates new instances per workflow build.
for factory in self._participant_factories:
p = factory()
participants.append(p)
else:
participants = self._participants
executors: list[Executor] = []
for p in participants:
if isinstance(p, Executor):
executors.append(p)
elif isinstance(p, AgentProtocol):
if self._request_info_enabled and (
not self._request_info_filter or resolve_agent_id(p) in self._request_info_filter
):
# Handle request info enabled agents
executors.append(AgentApprovalExecutor(p))
else:
executors.append(AgentExecutor(p))
else:
raise TypeError(f"Participants must be AgentProtocol or Executor instances. Got {type(p).__name__}.")
return executors
def build(self) -> Workflow:
"""Build and validate the sequential workflow.
Wiring pattern:
- _InputToConversation normalizes the initial input into list[ChatMessage]
- For each participant in order:
- If Agent (or AgentExecutor): pass conversation to the agent, then optionally
route through a request info interceptor, then convert response to conversation
via _ResponseToConversation
- Else (custom Executor): pass conversation directly to the executor
- _EndWithConversation yields the final conversation and the workflow becomes idle
"""
# Internal nodes
input_conv = _InputToConversation(id="input-conversation")
end = _EndWithConversation(id="end")
# Resolve participants and participant factories to executors
participants: list[Executor] = self._resolve_participants()
builder = WorkflowBuilder()
builder.set_start_executor(input_conv)
# Start of the chain is the input normalizer
prior: Executor | AgentProtocol = input_conv
for p in participants:
builder.add_edge(prior, p)
prior = p
# Terminate with the final conversation
builder.add_edge(prior, end)
if not self._intermediate_outputs:
# Constrain output to end only
builder = builder.with_output_from([end])
if self._checkpoint_storage is not None:
builder = builder.with_checkpointing(self._checkpoint_storage)
return builder.build()
python/packages/core/agent_framework/orchestrations/__init__.py
+61
View File
@@ -0,0 +1,61 @@
# Copyright (c) Microsoft. All rights reserved.
import importlib
from typing import Any
IMPORT_PATH = "agent_framework_orchestrations"
PACKAGE_NAME = "agent-framework-orchestrations"
_IMPORTS = [
"__version__",
# Sequential
"SequentialBuilder",
# Concurrent
"ConcurrentBuilder",
# Handoff
"HandoffAgentExecutor",
"HandoffAgentUserRequest",
"HandoffBuilder",
"HandoffConfiguration",
"HandoffSentEvent",
# Group Chat
"AgentBasedGroupChatOrchestrator",
"AgentOrchestrationOutput",
"GroupChatBuilder",
"GroupChatOrchestrator",
"GroupChatSelectionFunction",
"GroupChatState",
# Magentic
"MAGENTIC_MANAGER_NAME",
"ORCH_MSG_KIND_INSTRUCTION",
"ORCH_MSG_KIND_NOTICE",
"ORCH_MSG_KIND_TASK_LEDGER",
"ORCH_MSG_KIND_USER_TASK",
"MagenticAgentExecutor",
"MagenticBuilder",
"MagenticContext",
"MagenticManagerBase",
"MagenticOrchestrator",
"MagenticOrchestratorEvent",
"MagenticOrchestratorEventType",
"MagenticPlanReviewRequest",
"MagenticPlanReviewResponse",
"MagenticProgressLedger",
"MagenticProgressLedgerItem",
"MagenticResetSignal",
"StandardMagenticManager",
]
def __getattr__(name: str) -> Any:
if name in _IMPORTS:
try:
return getattr(importlib.import_module(IMPORT_PATH), name)
except ModuleNotFoundError as exc:
raise ModuleNotFoundError(
f"The '{PACKAGE_NAME}' package is not installed, please do `pip install {PACKAGE_NAME}`"
) from exc
raise AttributeError(f"Module {IMPORT_PATH} has no attribute {name}.")
def __dir__() -> list[str]:
return _IMPORTS
python/packages/core/agent_framework/orchestrations/__init__.pyi
+141
View File
@@ -0,0 +1,141 @@
# Copyright (c) Microsoft. All rights reserved.
# Type stubs for lazy-loaded orchestrations module
# These re-export types from agent_framework_orchestrations
from agent_framework_orchestrations import (
# Magentic
MAGENTIC_MANAGER_NAME as MAGENTIC_MANAGER_NAME,
)
from agent_framework_orchestrations import (
ORCH_MSG_KIND_INSTRUCTION as ORCH_MSG_KIND_INSTRUCTION,
)
from agent_framework_orchestrations import (
ORCH_MSG_KIND_NOTICE as ORCH_MSG_KIND_NOTICE,
)
from agent_framework_orchestrations import (
ORCH_MSG_KIND_TASK_LEDGER as ORCH_MSG_KIND_TASK_LEDGER,
)
from agent_framework_orchestrations import (
ORCH_MSG_KIND_USER_TASK as ORCH_MSG_KIND_USER_TASK,
)
from agent_framework_orchestrations import (
# Group Chat
AgentBasedGroupChatOrchestrator as AgentBasedGroupChatOrchestrator,
)
from agent_framework_orchestrations import (
AgentOrchestrationOutput as AgentOrchestrationOutput,
)
from agent_framework_orchestrations import (
# Concurrent
ConcurrentBuilder as ConcurrentBuilder,
)
from agent_framework_orchestrations import (
GroupChatBuilder as GroupChatBuilder,
)
from agent_framework_orchestrations import (
GroupChatOrchestrator as GroupChatOrchestrator,
)
from agent_framework_orchestrations import (
GroupChatSelectionFunction as GroupChatSelectionFunction,
)
from agent_framework_orchestrations import (
GroupChatState as GroupChatState,
)
from agent_framework_orchestrations import (
# Handoff
HandoffAgentExecutor as HandoffAgentExecutor,
)
from agent_framework_orchestrations import (
HandoffAgentUserRequest as HandoffAgentUserRequest,
)
from agent_framework_orchestrations import (
HandoffBuilder as HandoffBuilder,
)
from agent_framework_orchestrations import (
HandoffConfiguration as HandoffConfiguration,
)
from agent_framework_orchestrations import (
HandoffSentEvent as HandoffSentEvent,
)
from agent_framework_orchestrations import (
MagenticAgentExecutor as MagenticAgentExecutor,
)
from agent_framework_orchestrations import (
MagenticBuilder as MagenticBuilder,
)
from agent_framework_orchestrations import (
MagenticContext as MagenticContext,
)
from agent_framework_orchestrations import (
MagenticManagerBase as MagenticManagerBase,
)
from agent_framework_orchestrations import (
MagenticOrchestrator as MagenticOrchestrator,
)
from agent_framework_orchestrations import (
MagenticOrchestratorEvent as MagenticOrchestratorEvent,
)
from agent_framework_orchestrations import (
MagenticOrchestratorEventType as MagenticOrchestratorEventType,
)
from agent_framework_orchestrations import (
MagenticPlanReviewRequest as MagenticPlanReviewRequest,
)
from agent_framework_orchestrations import (
MagenticPlanReviewResponse as MagenticPlanReviewResponse,
)
from agent_framework_orchestrations import (
MagenticProgressLedger as MagenticProgressLedger,
)
from agent_framework_orchestrations import (
MagenticProgressLedgerItem as MagenticProgressLedgerItem,
)
from agent_framework_orchestrations import (
MagenticResetSignal as MagenticResetSignal,
)
from agent_framework_orchestrations import (
# Sequential
SequentialBuilder as SequentialBuilder,
)
from agent_framework_orchestrations import (
StandardMagenticManager as StandardMagenticManager,
)
from agent_framework_orchestrations import (
__version__ as __version__,
)
__all__ = [
"MAGENTIC_MANAGER_NAME",
"ORCH_MSG_KIND_INSTRUCTION",
"ORCH_MSG_KIND_NOTICE",
"ORCH_MSG_KIND_TASK_LEDGER",
"ORCH_MSG_KIND_USER_TASK",
"AgentBasedGroupChatOrchestrator",
"AgentOrchestrationOutput",
"ConcurrentBuilder",
"GroupChatBuilder",
"GroupChatOrchestrator",
"GroupChatSelectionFunction",
"GroupChatState",
"HandoffAgentExecutor",
"HandoffAgentUserRequest",
"HandoffBuilder",
"HandoffConfiguration",
"HandoffSentEvent",
"MagenticAgentExecutor",
"MagenticBuilder",
"MagenticContext",
"MagenticManagerBase",
"MagenticOrchestrator",
"MagenticOrchestratorEvent",
"MagenticOrchestratorEventType",
"MagenticPlanReviewRequest",
"MagenticPlanReviewResponse",
"MagenticProgressLedger",
"MagenticProgressLedgerItem",
"MagenticResetSignal",
"SequentialBuilder",
"StandardMagenticManager",
"__version__",
]
python/packages/core/pyproject.toml
+1
View File
@@ -55,6 +55,7 @@ all = [
"agent-framework-lab",
"agent-framework-mem0",
"agent-framework-ollama",
"agent-framework-orchestrations",
"agent-framework-purview",
"agent-framework-redis",
]
python/packages/core/tests/workflow/test_agent_executor.py
+1 -1
View File
@@ -12,13 +12,13 @@ from agent_framework import (
ChatMessage,
ChatMessageStore,
Content,
SequentialBuilder,
WorkflowOutputEvent,
WorkflowRunState,
WorkflowStatusEvent,
)
from agent_framework._workflows._agent_executor import AgentExecutorResponse
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
from agent_framework.orchestrations import SequentialBuilder
class _CountingAgent(BaseAgent):
python/packages/core/tests/workflow/test_concurrent.py
-550
View File
@@ -1,550 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from typing import Any, cast
import pytest
from typing_extensions import Never
from agent_framework import (
AgentExecutorRequest,
AgentExecutorResponse,
AgentResponse,
ChatMessage,
ConcurrentBuilder,
Executor,
WorkflowContext,
WorkflowOutputEvent,
WorkflowRunState,
WorkflowStatusEvent,
handler,
)
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
class _FakeAgentExec(Executor):
"""Test executor that mimics an agent by emitting an AgentExecutorResponse.
It takes the incoming AgentExecutorRequest, produces a single assistant message
with the configured reply text, and sends an AgentExecutorResponse that includes
full_conversation (the original user prompt followed by the assistant message).
"""
def __init__(self, id: str, reply_text: str) -> None:
super().__init__(id)
self._reply_text = reply_text
@handler
async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
response = AgentResponse(messages=ChatMessage("assistant", text=self._reply_text))
full_conversation = list(request.messages) + list(response.messages)
await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))
def test_concurrent_builder_rejects_empty_participants() -> None:
with pytest.raises(ValueError):
ConcurrentBuilder().participants([])
def test_concurrent_builder_rejects_duplicate_executors() -> None:
a = _FakeAgentExec("dup", "A")
b = _FakeAgentExec("dup", "B") # same executor id
with pytest.raises(ValueError):
ConcurrentBuilder().participants([a, b])
def test_concurrent_builder_rejects_duplicate_executors_from_factories() -> None:
"""Test that duplicate executor IDs from factories are detected at build time."""
def create_dup1() -> Executor:
return _FakeAgentExec("dup", "A")
def create_dup2() -> Executor:
return _FakeAgentExec("dup", "B") # same executor id
builder = ConcurrentBuilder().register_participants([create_dup1, create_dup2])
with pytest.raises(ValueError, match="Duplicate executor ID 'dup' detected in workflow."):
builder.build()
def test_concurrent_builder_rejects_mixed_participants_and_factories() -> None:
"""Test that mixing .participants() and .register_participants() raises an error."""
# Case 1: participants first, then register_participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
(
ConcurrentBuilder()
.participants([_FakeAgentExec("a", "A")])
.register_participants([lambda: _FakeAgentExec("b", "B")])
)
# Case 2: register_participants first, then participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
(
ConcurrentBuilder()
.register_participants([lambda: _FakeAgentExec("a", "A")])
.participants([_FakeAgentExec("b", "B")])
)
def test_concurrent_builder_rejects_multiple_calls_to_participants() -> None:
"""Test that multiple calls to .participants() raises an error."""
with pytest.raises(ValueError, match=r"participants\(\) has already been called"):
(ConcurrentBuilder().participants([_FakeAgentExec("a", "A")]).participants([_FakeAgentExec("b", "B")]))
def test_concurrent_builder_rejects_multiple_calls_to_register_participants() -> None:
"""Test that multiple calls to .register_participants() raises an error."""
with pytest.raises(ValueError, match=r"register_participants\(\) has already been called"):
(
ConcurrentBuilder()
.register_participants([lambda: _FakeAgentExec("a", "A")])
.register_participants([lambda: _FakeAgentExec("b", "B")])
)
async def test_concurrent_default_aggregator_emits_single_user_and_assistants() -> None:
# Three synthetic agent executors
e1 = _FakeAgentExec("agentA", "Alpha")
e2 = _FakeAgentExec("agentB", "Beta")
e3 = _FakeAgentExec("agentC", "Gamma")
wf = ConcurrentBuilder().participants([e1, e2, e3]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("prompt: hello world"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(list[ChatMessage], ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
messages: list[ChatMessage] = output
# Expect one user message + one assistant message per participant
assert len(messages) == 1 + 3
assert messages[0].role == "user"
assert "hello world" in messages[0].text
assistant_texts = {m.text for m in messages[1:]}
assert assistant_texts == {"Alpha", "Beta", "Gamma"}
assert all(m.role == "assistant" for m in messages[1:])
async def test_concurrent_custom_aggregator_callback_is_used() -> None:
# Two synthetic agent executors for brevity
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
async def summarize(results: list[AgentExecutorResponse]) -> str:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_response.messages
texts.append(msgs[-1].text if msgs else "")
return " | ".join(sorted(texts))
wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: custom"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
# Custom aggregator returns a string payload
assert isinstance(output, str)
assert output == "One | Two"
async def test_concurrent_custom_aggregator_sync_callback_is_used() -> None:
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
# Sync callback with ctx parameter (should run via asyncio.to_thread)
def summarize_sync(results: list[AgentExecutorResponse], _ctx: WorkflowContext[Any]) -> str: # type: ignore[unused-argument]
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_response.messages
texts.append(msgs[-1].text if msgs else "")
return " | ".join(sorted(texts))
wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize_sync).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: custom sync"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One | Two"
def test_concurrent_custom_aggregator_uses_callback_name_for_id() -> None:
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
def summarize(results: list[AgentExecutorResponse]) -> str: # type: ignore[override]
return str(len(results))
wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize).build()
assert "summarize" in wf.executors
aggregator = wf.executors["summarize"]
assert aggregator.id == "summarize"
async def test_concurrent_with_aggregator_executor_instance() -> None:
"""Test with_aggregator using an Executor instance (not factory)."""
class CustomAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" & ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
aggregator_instance = CustomAggregator(id="instance_aggregator")
wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(aggregator_instance).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: instance test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One & Two"
async def test_concurrent_with_aggregator_executor_factory() -> None:
"""Test with_aggregator using an Executor factory."""
class CustomAggregator(Executor):
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" | ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
wf = (
ConcurrentBuilder()
.participants([e1, e2])
.register_aggregator(lambda: CustomAggregator(id="custom_aggregator"))
.build()
)
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: factory test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One | Two"
async def test_concurrent_with_aggregator_executor_factory_with_default_id() -> None:
"""Test with_aggregator using an Executor class directly as factory (with default __init__ parameters)."""
class CustomAggregator(Executor):
def __init__(self, id: str = "default_aggregator") -> None:
super().__init__(id)
@handler
async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
texts: list[str] = []
for r in results:
msgs: list[ChatMessage] = r.agent_response.messages
texts.append(msgs[-1].text if msgs else "")
await ctx.yield_output(" | ".join(sorted(texts)))
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
wf = ConcurrentBuilder().participants([e1, e2]).register_aggregator(CustomAggregator).build()
completed = False
output: str | None = None
async for ev in wf.run_stream("prompt: factory test"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(str, ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, str)
assert output == "One | Two"
def test_concurrent_builder_rejects_multiple_calls_to_with_aggregator() -> None:
"""Test that multiple calls to .with_aggregator() raises an error."""
def summarize(results: list[AgentExecutorResponse]) -> str: # type: ignore[override]
return str(len(results))
with pytest.raises(ValueError, match=r"with_aggregator\(\) has already been called"):
(ConcurrentBuilder().with_aggregator(summarize).with_aggregator(summarize))
def test_concurrent_builder_rejects_multiple_calls_to_register_aggregator() -> None:
"""Test that multiple calls to .register_aggregator() raises an error."""
class CustomAggregator(Executor):
pass
with pytest.raises(ValueError, match=r"register_aggregator\(\) has already been called"):
(
ConcurrentBuilder()
.register_aggregator(lambda: CustomAggregator(id="agg1"))
.register_aggregator(lambda: CustomAggregator(id="agg2"))
)
async def test_concurrent_checkpoint_resume_round_trip() -> None:
storage = InMemoryCheckpointStorage()
participants = (
_FakeAgentExec("agentA", "Alpha"),
_FakeAgentExec("agentB", "Beta"),
_FakeAgentExec("agentC", "Gamma"),
)
wf = ConcurrentBuilder().participants(list(participants)).with_checkpointing(storage).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("checkpoint concurrent"):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_participants = (
_FakeAgentExec("agentA", "Alpha"),
_FakeAgentExec("agentB", "Beta"),
_FakeAgentExec("agentC", "Gamma"),
)
wf_resume = ConcurrentBuilder().participants(list(resumed_participants)).with_checkpointing(storage).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_concurrent_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
storage = InMemoryCheckpointStorage()
agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf = ConcurrentBuilder().participants(agents).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]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf_resume = ConcurrentBuilder().participants(resumed_agents).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
async def test_concurrent_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
agents = [_FakeAgentExec(id="agent1", reply_text="A1"), _FakeAgentExec(id="agent2", reply_text="A2")]
wf = ConcurrentBuilder().participants(agents).with_checkpointing(buildtime_storage).build()
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]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
def test_concurrent_builder_rejects_empty_participant_factories() -> None:
with pytest.raises(ValueError):
ConcurrentBuilder().register_participants([])
async def test_concurrent_builder_reusable_after_build_with_participants() -> None:
"""Test that the builder can be reused to build multiple identical workflows with participants()."""
e1 = _FakeAgentExec("agentA", "One")
e2 = _FakeAgentExec("agentB", "Two")
builder = ConcurrentBuilder().participants([e1, e2])
builder.build()
assert builder._participants[0] is e1 # type: ignore
assert builder._participants[1] is e2 # type: ignore
assert builder._participant_factories == [] # type: ignore
async def test_concurrent_builder_reusable_after_build_with_factories() -> None:
"""Test that the builder can be reused to build multiple workflows with register_participants()."""
call_count = 0
def create_agent_executor_a() -> Executor:
nonlocal call_count
call_count += 1
return _FakeAgentExec("agentA", "One")
def create_agent_executor_b() -> Executor:
nonlocal call_count
call_count += 1
return _FakeAgentExec("agentB", "Two")
builder = ConcurrentBuilder().register_participants([create_agent_executor_a, create_agent_executor_b])
# Build the first workflow
wf1 = builder.build()
assert builder._participants == [] # type: ignore
assert len(builder._participant_factories) == 2 # type: ignore
assert call_count == 2
# Build the second workflow
wf2 = builder.build()
assert call_count == 4
# Verify that the two workflows have different executor instances
assert wf1.executors["agentA"] is not wf2.executors["agentA"]
assert wf1.executors["agentB"] is not wf2.executors["agentB"]
async def test_concurrent_with_register_participants() -> None:
"""Test workflow creation using register_participants with factories."""
def create_agent1() -> Executor:
return _FakeAgentExec("agentA", "Alpha")
def create_agent2() -> Executor:
return _FakeAgentExec("agentB", "Beta")
def create_agent3() -> Executor:
return _FakeAgentExec("agentC", "Gamma")
wf = ConcurrentBuilder().register_participants([create_agent1, create_agent2, create_agent3]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("test prompt"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = cast(list[ChatMessage], ev.data)
if completed and output is not None:
break
assert completed
assert output is not None
messages: list[ChatMessage] = output
# Expect one user message + one assistant message per participant
assert len(messages) == 1 + 3
assert messages[0].role == "user"
assert "test prompt" in messages[0].text
assistant_texts = {m.text for m in messages[1:]}
assert assistant_texts == {"Alpha", "Beta", "Gamma"}
assert all(m.role == "assistant" for m in messages[1:])
python/packages/core/tests/workflow/test_full_conversation.py
+1 -1
View File
@@ -16,13 +16,13 @@ from agent_framework import (
ChatMessage,
Content,
Executor,
SequentialBuilder,
WorkflowBuilder,
WorkflowContext,
WorkflowRunState,
WorkflowStatusEvent,
handler,
)
from agent_framework.orchestrations import SequentialBuilder
class _SimpleAgent(BaseAgent):
python/packages/core/tests/workflow/test_group_chat.py
-1334
View File
File diff suppressed because it is too large Load Diff
python/packages/core/tests/workflow/test_handoff.py
-709
View File
@@ -1,709 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import AsyncIterable
from typing import Any, cast
from unittest.mock import AsyncMock, MagicMock
import pytest
from agent_framework import (
ChatAgent,
ChatMessage,
ChatResponse,
ChatResponseUpdate,
Content,
HandoffAgentUserRequest,
HandoffBuilder,
RequestInfoEvent,
WorkflowEvent,
WorkflowOutputEvent,
resolve_agent_id,
use_function_invocation,
)
@use_function_invocation
class MockChatClient:
"""Mock chat client for testing handoff workflows."""
additional_properties: dict[str, Any]
def __init__(
self,
name: str,
*,
handoff_to: str | None = None,
) -> None:
"""Initialize the mock chat client.
Args:
name: The name of the agent using this chat client.
handoff_to: The name of the agent to hand off to, or None for no handoff.
This is hardcoded for testing purposes so that the agent always attempts to hand off.
"""
self._name = name
self._handoff_to = handoff_to
self._call_index = 0
async def get_response(self, messages: Any, **kwargs: Any) -> ChatResponse:
contents = _build_reply_contents(self._name, self._handoff_to, self._next_call_id())
reply = ChatMessage(
role="assistant",
contents=contents,
)
return ChatResponse(messages=reply, response_id="mock_response")
def get_streaming_response(self, messages: Any, **kwargs: Any) -> AsyncIterable[ChatResponseUpdate]:
async def _stream() -> AsyncIterable[ChatResponseUpdate]:
contents = _build_reply_contents(self._name, self._handoff_to, self._next_call_id())
yield ChatResponseUpdate(contents=contents, role="assistant")
return _stream()
def _next_call_id(self) -> str | None:
if not self._handoff_to:
return None
call_id = f"{self._name}-handoff-{self._call_index}"
self._call_index += 1
return call_id
def _build_reply_contents(
agent_name: str,
handoff_to: str | None,
call_id: str | None,
) -> list[Content]:
contents: list[Content] = []
if handoff_to and call_id:
contents.append(
Content.from_function_call(
call_id=call_id, name=f"handoff_to_{handoff_to}", arguments={"handoff_to": handoff_to}
)
)
text = f"{agent_name} reply"
contents.append(Content.from_text(text=text))
return contents
class MockHandoffAgent(ChatAgent):
"""Mock agent that can hand off to another agent."""
def __init__(
self,
*,
name: str,
handoff_to: str | None = None,
) -> None:
"""Initialize the mock handoff agent.
Args:
name: The name of the agent.
handoff_to: The name of the agent to hand off to, or None for no handoff.
This is hardcoded for testing purposes so that the agent always attempts to hand off.
"""
super().__init__(chat_client=MockChatClient(name, handoff_to=handoff_to), name=name, id=name)
async def _drain(stream: AsyncIterable[WorkflowEvent]) -> list[WorkflowEvent]:
return [event async for event in stream]
async def test_handoff():
"""Test that agents can hand off to each other."""
# `triage` hands off to `specialist`, who then hands off to `escalation`.
# `escalation` has no handoff, so the workflow should request user input to continue.
triage = MockHandoffAgent(name="triage", handoff_to="specialist")
specialist = MockHandoffAgent(name="specialist", handoff_to="escalation")
escalation = MockHandoffAgent(name="escalation")
# Without explicitly defining handoffs, the builder will create connections
# between all agents.
workflow = (
HandoffBuilder(participants=[triage, specialist, escalation])
.with_start_agent(triage)
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == "user") >= 2)
.build()
)
# Start conversation - triage hands off to specialist then escalation
# escalation won't trigger a handoff, so the response from it will become
# a request for user input because autonomous mode is not enabled by default.
events = await _drain(workflow.run_stream("Need technical support"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
assert len(requests) == 1
request = requests[0]
assert isinstance(request.data, HandoffAgentUserRequest)
assert request.source_executor_id == escalation.name
async def test_autonomous_mode_yields_output_without_user_request():
"""Ensure autonomous interaction mode yields output without requesting user input."""
triage = MockHandoffAgent(name="triage", handoff_to="specialist")
specialist = MockHandoffAgent(name="specialist")
workflow = (
HandoffBuilder(participants=[triage, specialist])
.with_start_agent(triage)
# Since specialist has no handoff, the specialist will be generating normal responses.
# With autonomous mode, this should continue until the termination condition is met.
.with_autonomous_mode(
agents=[specialist],
turn_limits={resolve_agent_id(specialist): 1},
)
# This termination condition ensures the workflow runs through both agents.
# First message is the user message to triage, second is triage's response, which
# is a handoff to specialist, third is specialist's response that should not request
# user input due to autonomous mode. Fourth message will come from the specialist
# again and will trigger termination.
.with_termination_condition(lambda conv: len(conv) >= 4)
.build()
)
events = await _drain(workflow.run_stream("Package arrived broken"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert not requests, "Autonomous mode should not request additional user input"
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs, "Autonomous mode should yield a workflow output"
final_conversation = outputs[-1].data
assert isinstance(final_conversation, list)
conversation_list = cast(list[ChatMessage], final_conversation)
assert any(msg.role == "assistant" and (msg.text or "").startswith("specialist reply") for msg in conversation_list)
async def test_autonomous_mode_resumes_user_input_on_turn_limit():
"""Autonomous mode should resume user input request when turn limit is reached."""
triage = MockHandoffAgent(name="triage", handoff_to="worker")
worker = MockHandoffAgent(name="worker")
workflow = (
HandoffBuilder(participants=[triage, worker])
.with_start_agent(triage)
.with_autonomous_mode(agents=[worker], turn_limits={resolve_agent_id(worker): 2})
.with_termination_condition(lambda conv: False)
.build()
)
events = await _drain(workflow.run_stream("Start"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests and len(requests) == 1, "Turn limit should force a user input request"
assert requests[0].source_executor_id == worker.name
def test_build_fails_without_start_agent():
"""Verify that build() raises ValueError when with_start_agent() was not called."""
triage = MockHandoffAgent(name="triage")
specialist = MockHandoffAgent(name="specialist")
with pytest.raises(ValueError, match=r"Must call with_start_agent\(...\) before building the workflow."):
HandoffBuilder(participants=[triage, specialist]).build()
def test_build_fails_without_participants():
"""Verify that build() raises ValueError when no participants are provided."""
with pytest.raises(
ValueError, match=r"No participants provided\. Call \.participants\(\) or \.register_participants\(\) first."
):
HandoffBuilder().build()
async def test_handoff_async_termination_condition() -> None:
"""Test that async termination conditions work correctly."""
termination_call_count = 0
async def async_termination(conv: list[ChatMessage]) -> bool:
nonlocal termination_call_count
termination_call_count += 1
user_count = sum(1 for msg in conv if msg.role == "user")
return user_count >= 2
coordinator = MockHandoffAgent(name="coordinator", handoff_to="worker")
worker = MockHandoffAgent(name="worker")
workflow = (
HandoffBuilder(participants=[coordinator, worker])
.with_start_agent(coordinator)
.with_termination_condition(async_termination)
.build()
)
events = await _drain(workflow.run_stream("First user message"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage("user", ["Second user message"])]})
)
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert len(outputs) == 1
final_conversation = outputs[0].data
assert isinstance(final_conversation, list)
final_conv_list = cast(list[ChatMessage], final_conversation)
user_messages = [msg for msg in final_conv_list if msg.role == "user"]
assert len(user_messages) == 2
assert termination_call_count > 0
async def test_tool_choice_preserved_from_agent_config():
"""Verify that agent-level tool_choice configuration is preserved and not overridden."""
# Create a mock chat client that records the tool_choice used
recorded_tool_choices: list[Any] = []
async def mock_get_response(messages: Any, options: dict[str, Any] | None = None, **kwargs: Any) -> ChatResponse:
if options:
recorded_tool_choices.append(options.get("tool_choice"))
return ChatResponse(
messages=[ChatMessage("assistant", ["Response"])],
response_id="test_response",
)
mock_client = MagicMock()
mock_client.get_response = AsyncMock(side_effect=mock_get_response)
# Create agent with specific tool_choice configuration via default_options
agent = ChatAgent(
chat_client=mock_client,
name="test_agent",
default_options={"tool_choice": {"mode": "required"}}, # type: ignore
)
# Run the agent
await agent.run("Test message")
# Verify tool_choice was preserved
assert len(recorded_tool_choices) > 0, "No tool_choice recorded"
last_tool_choice = recorded_tool_choices[-1]
assert last_tool_choice is not None, "tool_choice should not be None"
assert last_tool_choice == {"mode": "required"}, f"Expected 'required', got {last_tool_choice}"
# region Participant Factory Tests
def test_handoff_builder_rejects_empty_participant_factories():
"""Test that HandoffBuilder rejects empty participant_factories dictionary."""
# Empty factories are rejected immediately when calling participant_factories()
with pytest.raises(ValueError, match=r"participant_factories cannot be empty"):
HandoffBuilder().register_participants({})
with pytest.raises(
ValueError, match=r"No participants provided\. Call \.participants\(\) or \.register_participants\(\) first\."
):
HandoffBuilder(participant_factories={}).build()
def test_handoff_builder_rejects_mixing_participants_and_factories():
"""Test that mixing participants and participant_factories in __init__ raises an error."""
triage = MockHandoffAgent(name="triage")
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participants=[triage], participant_factories={"triage": lambda: triage})
def test_handoff_builder_rejects_mixing_participants_and_participant_factories_methods():
"""Test that mixing .participants() and .participant_factories() raises an error."""
triage = MockHandoffAgent(name="triage")
# Case 1: participants first, then participant_factories
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participants=[triage]).register_participants({
"specialist": lambda: MockHandoffAgent(name="specialist")
})
# Case 2: participant_factories first, then participants
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(participant_factories={"triage": lambda: triage}).participants([
MockHandoffAgent(name="specialist")
])
# Case 3: participants(), then participant_factories()
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder().participants([triage]).register_participants({
"specialist": lambda: MockHandoffAgent(name="specialist")
})
# Case 4: participant_factories(), then participants()
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder().register_participants({"triage": lambda: triage}).participants([
MockHandoffAgent(name="specialist")
])
# Case 5: mix during initialization
with pytest.raises(ValueError, match="Cannot mix .participants"):
HandoffBuilder(
participants=[triage], participant_factories={"specialist": lambda: MockHandoffAgent(name="specialist")}
)
def test_handoff_builder_rejects_multiple_calls_to_participant_factories():
"""Test that multiple calls to .participant_factories() raises an error."""
with pytest.raises(
ValueError, match=r"register_participants\(\) has already been called on this builder instance."
):
(
HandoffBuilder()
.register_participants({"agent1": lambda: MockHandoffAgent(name="agent1")})
.register_participants({"agent2": lambda: MockHandoffAgent(name="agent2")})
)
def test_handoff_builder_rejects_multiple_calls_to_participants():
"""Test that multiple calls to .participants() raises an error."""
with pytest.raises(ValueError, match="participants have already been assigned"):
(
HandoffBuilder()
.participants([MockHandoffAgent(name="agent1")])
.participants([MockHandoffAgent(name="agent2")])
)
def test_handoff_builder_rejects_instance_coordinator_with_factories():
"""Test that using an agent instance for set_coordinator when using factories raises an error."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage")
def create_specialist() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist")
# Create an agent instance
coordinator_instance = MockHandoffAgent(name="coordinator")
with pytest.raises(ValueError, match=r"Call participants\(\.\.\.\) before with_start_agent\(\.\.\.\)"):
(
HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).with_start_agent(coordinator_instance) # Instance, not factory name
)
def test_handoff_builder_rejects_factory_name_coordinator_with_instances():
"""Test that using a factory name for set_coordinator when using instances raises an error."""
triage = MockHandoffAgent(name="triage")
specialist = MockHandoffAgent(name="specialist")
with pytest.raises(ValueError, match=r"Call register_participants\(...\) before with_start_agent\(...\)"):
(
HandoffBuilder(participants=[triage, specialist]).with_start_agent(
"triage"
) # String factory name, not instance
)
def test_handoff_builder_rejects_mixed_types_in_add_handoff_source():
"""Test that add_handoff rejects factory name source with instance-based participants."""
triage = MockHandoffAgent(name="triage")
specialist = MockHandoffAgent(name="specialist")
with pytest.raises(TypeError, match="Cannot mix factory names \\(str\\) and AgentProtocol.*instances"):
(
HandoffBuilder(participants=[triage, specialist])
.with_start_agent(triage)
.add_handoff("triage", [specialist]) # String source with instance participants
)
def test_handoff_builder_accepts_all_factory_names_in_add_handoff():
"""Test that add_handoff accepts all factory names when using participant_factories."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage")
def create_specialist_a() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_a")
def create_specialist_b() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_b")
# This should work - all strings with participant_factories
builder = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.with_start_agent("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
def test_handoff_builder_accepts_all_instances_in_add_handoff():
"""Test that add_handoff accepts all instances when using participants."""
triage = MockHandoffAgent(name="triage", handoff_to="specialist_a")
specialist_a = MockHandoffAgent(name="specialist_a")
specialist_b = MockHandoffAgent(name="specialist_b")
# This should work - all instances with participants
builder = (
HandoffBuilder(participants=[triage, specialist_a, specialist_b])
.with_start_agent(triage)
.add_handoff(triage, [specialist_a, specialist_b])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
async def test_handoff_with_participant_factories():
"""Test workflow creation using participant_factories."""
call_count = 0
def create_triage() -> MockHandoffAgent:
nonlocal call_count
call_count += 1
return MockHandoffAgent(name="triage", handoff_to="specialist")
def create_specialist() -> MockHandoffAgent:
nonlocal call_count
call_count += 1
return MockHandoffAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.with_start_agent("triage")
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == "user") >= 2)
.build()
)
# Factories should be called during build
assert call_count == 2
events = await _drain(workflow.run_stream("Need help"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Follow-up message
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage("user", ["More details"])]})
)
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs
async def test_handoff_participant_factories_reusable_builder():
"""Test that the builder can be reused to build multiple workflows with factories."""
call_count = 0
def create_triage() -> MockHandoffAgent:
nonlocal call_count
call_count += 1
return MockHandoffAgent(name="triage", handoff_to="specialist")
def create_specialist() -> MockHandoffAgent:
nonlocal call_count
call_count += 1
return MockHandoffAgent(name="specialist")
builder = HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).with_start_agent("triage")
# Build first workflow
wf1 = builder.build()
assert call_count == 2
# Build second workflow
wf2 = builder.build()
assert call_count == 4
# Verify that the two workflows have different agent instances
assert wf1.executors["triage"] is not wf2.executors["triage"]
assert wf1.executors["specialist"] is not wf2.executors["specialist"]
async def test_handoff_with_participant_factories_and_add_handoff():
"""Test that .add_handoff() works correctly with participant_factories."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage", handoff_to="specialist_a")
def create_specialist_a() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_a", handoff_to="specialist_b")
def create_specialist_b() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_b")
workflow = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.with_start_agent("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
.add_handoff("specialist_a", ["specialist_b"])
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == "user") >= 3)
.build()
)
# Start conversation - triage hands off to specialist_a
events = await _drain(workflow.run_stream("Initial request"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Verify specialist_a executor exists and was called
assert "specialist_a" in workflow.executors
# Second user message - specialist_a hands off to specialist_b
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage("user", ["Need escalation"])]})
)
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
# Verify specialist_b executor exists
assert "specialist_b" in workflow.executors
async def test_handoff_participant_factories_with_checkpointing():
"""Test checkpointing with participant_factories."""
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
storage = InMemoryCheckpointStorage()
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage", handoff_to="specialist")
def create_specialist() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.with_start_agent("triage")
.with_checkpointing(storage)
.with_termination_condition(lambda conv: sum(1 for m in conv if m.role == "user") >= 2)
.build()
)
# Run workflow and capture output
events = await _drain(workflow.run_stream("checkpoint test"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests
events = await _drain(
workflow.send_responses_streaming({requests[-1].request_id: [ChatMessage("user", ["follow up"])]})
)
outputs = [ev for ev in events if isinstance(ev, WorkflowOutputEvent)]
assert outputs, "Should have workflow output after termination condition is met"
# List checkpoints - just verify they were created
checkpoints = await storage.list_checkpoints()
assert checkpoints, "Checkpoints should be created during workflow execution"
def test_handoff_set_coordinator_with_factory_name():
"""Test that set_coordinator accepts factory name as string."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage")
def create_specialist() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist")
builder = HandoffBuilder(
participant_factories={"triage": create_triage, "specialist": create_specialist}
).with_start_agent("triage")
workflow = builder.build()
assert "triage" in workflow.executors
def test_handoff_add_handoff_with_factory_names():
"""Test that add_handoff accepts factory names as strings."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage", handoff_to="specialist_a")
def create_specialist_a() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_a")
def create_specialist_b() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist_b")
builder = (
HandoffBuilder(
participant_factories={
"triage": create_triage,
"specialist_a": create_specialist_a,
"specialist_b": create_specialist_b,
}
)
.with_start_agent("triage")
.add_handoff("triage", ["specialist_a", "specialist_b"])
)
workflow = builder.build()
assert "triage" in workflow.executors
assert "specialist_a" in workflow.executors
assert "specialist_b" in workflow.executors
async def test_handoff_participant_factories_autonomous_mode():
"""Test autonomous mode with participant_factories."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage", handoff_to="specialist")
def create_specialist() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist")
workflow = (
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.with_start_agent("triage")
.with_autonomous_mode(agents=["specialist"], turn_limits={"specialist": 1})
.build()
)
events = await _drain(workflow.run_stream("Issue"))
requests = [ev for ev in events if isinstance(ev, RequestInfoEvent)]
assert requests and len(requests) == 1
assert requests[0].source_executor_id == "specialist"
def test_handoff_participant_factories_invalid_coordinator_name():
"""Test that set_coordinator raises error for non-existent factory name."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage")
with pytest.raises(
ValueError, match="Start agent factory name 'nonexistent' is not in the participant_factories list"
):
(HandoffBuilder(participant_factories={"triage": create_triage}).with_start_agent("nonexistent").build())
def test_handoff_participant_factories_invalid_handoff_target():
"""Test that add_handoff raises error for non-existent target factory name."""
def create_triage() -> MockHandoffAgent:
return MockHandoffAgent(name="triage")
def create_specialist() -> MockHandoffAgent:
return MockHandoffAgent(name="specialist")
with pytest.raises(ValueError, match="Target factory name 'nonexistent' is not in the participant_factories list"):
(
HandoffBuilder(participant_factories={"triage": create_triage, "specialist": create_specialist})
.with_start_agent("triage")
.add_handoff("triage", ["nonexistent"])
.build()
)
# endregion Participant Factory Tests
python/packages/core/tests/workflow/test_magentic.py
-1301
View File
File diff suppressed because it is too large Load Diff
python/packages/core/tests/workflow/test_sequential.py
-454
View File
@@ -1,454 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import AsyncIterable
from typing import Any
import pytest
from agent_framework import (
AgentExecutorResponse,
AgentResponse,
AgentResponseUpdate,
AgentThread,
BaseAgent,
ChatMessage,
Content,
Executor,
SequentialBuilder,
TypeCompatibilityError,
WorkflowContext,
WorkflowOutputEvent,
WorkflowRunState,
WorkflowStatusEvent,
handler,
)
from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
class _EchoAgent(BaseAgent):
"""Simple agent that appends a single assistant message with its name."""
async def run( # type: ignore[override]
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AgentResponse:
return AgentResponse(messages=[ChatMessage("assistant", [f"{self.name} reply"])])
async def run_stream( # type: ignore[override]
self,
messages: str | ChatMessage | list[str] | list[ChatMessage] | None = None,
*,
thread: AgentThread | None = None,
**kwargs: Any,
) -> AsyncIterable[AgentResponseUpdate]:
# Minimal async generator with one assistant update
yield AgentResponseUpdate(contents=[Content.from_text(text=f"{self.name} reply")])
class _SummarizerExec(Executor):
"""Custom executor that summarizes by appending a short assistant message."""
@handler
async def summarize(self, agent_response: AgentExecutorResponse, ctx: WorkflowContext[list[ChatMessage]]) -> None:
conversation = agent_response.full_conversation or []
user_texts = [m.text for m in conversation if m.role == "user"]
agents = [m.author_name or m.role for m in conversation if m.role == "assistant"]
summary = ChatMessage("assistant", [f"Summary of users:{len(user_texts)} agents:{len(agents)}"])
await ctx.send_message(list(conversation) + [summary])
class _InvalidExecutor(Executor):
"""Invalid executor that does not have a handler that accepts a list of chat messages"""
@handler
async def summarize(self, conversation: list[str], ctx: WorkflowContext[list[ChatMessage]]) -> None:
pass
def test_sequential_builder_rejects_empty_participants() -> None:
with pytest.raises(ValueError):
SequentialBuilder().participants([])
def test_sequential_builder_rejects_empty_participant_factories() -> None:
with pytest.raises(ValueError):
SequentialBuilder().register_participants([])
def test_sequential_builder_rejects_mixing_participants_and_factories() -> None:
"""Test that mixing .participants() and .register_participants() raises an error."""
a1 = _EchoAgent(id="agent1", name="A1")
# Try .participants() then .register_participants()
with pytest.raises(ValueError, match="Cannot mix"):
SequentialBuilder().participants([a1]).register_participants([lambda: _EchoAgent(id="agent2", name="A2")])
# Try .register_participants() then .participants()
with pytest.raises(ValueError, match="Cannot mix"):
SequentialBuilder().register_participants([lambda: _EchoAgent(id="agent1", name="A1")]).participants([a1])
def test_sequential_builder_validation_rejects_invalid_executor() -> None:
"""Test that adding an invalid executor to the builder raises an error."""
with pytest.raises(TypeCompatibilityError):
SequentialBuilder().participants([_EchoAgent(id="agent1", name="A1"), _InvalidExecutor(id="invalid")]).build()
async def test_sequential_agents_append_to_context() -> None:
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder().participants([a1, a2]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("hello sequential"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = ev.data # type: ignore[assignment]
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, list)
msgs: list[ChatMessage] = output
assert len(msgs) == 3
assert msgs[0].role == "user" and "hello sequential" in msgs[0].text
assert msgs[1].role == "assistant" and (msgs[1].author_name == "A1" or True)
assert msgs[2].role == "assistant" and (msgs[2].author_name == "A2" or True)
assert "A1 reply" in msgs[1].text
assert "A2 reply" in msgs[2].text
async def test_sequential_register_participants_with_agent_factories() -> None:
"""Test that register_participants works with agent factories."""
def create_agent1() -> _EchoAgent:
return _EchoAgent(id="agent1", name="A1")
def create_agent2() -> _EchoAgent:
return _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder().register_participants([create_agent1, create_agent2]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("hello factories"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = ev.data
if completed and output is not None:
break
assert completed
assert output is not None
assert isinstance(output, list)
msgs: list[ChatMessage] = output
assert len(msgs) == 3
assert msgs[0].role == "user" and "hello factories" in msgs[0].text
assert msgs[1].role == "assistant" and "A1 reply" in msgs[1].text
assert msgs[2].role == "assistant" and "A2 reply" in msgs[2].text
async def test_sequential_with_custom_executor_summary() -> None:
a1 = _EchoAgent(id="agent1", name="A1")
summarizer = _SummarizerExec(id="summarizer")
wf = SequentialBuilder().participants([a1, summarizer]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("topic X"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = ev.data
if completed and output is not None:
break
assert completed
assert output is not None
msgs: list[ChatMessage] = output
# Expect: [user, A1 reply, summary]
assert len(msgs) == 3
assert msgs[0].role == "user"
assert msgs[1].role == "assistant" and "A1 reply" in msgs[1].text
assert msgs[2].role == "assistant" and msgs[2].text.startswith("Summary of users:")
async def test_sequential_register_participants_mixed_agents_and_executors() -> None:
"""Test register_participants with both agent and executor factories."""
def create_agent() -> _EchoAgent:
return _EchoAgent(id="agent1", name="A1")
def create_summarizer() -> _SummarizerExec:
return _SummarizerExec(id="summarizer")
wf = SequentialBuilder().register_participants([create_agent, create_summarizer]).build()
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("topic Y"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = ev.data
if completed and output is not None:
break
assert completed
assert output is not None
msgs: list[ChatMessage] = output
# Expect: [user, A1 reply, summary]
assert len(msgs) == 3
assert msgs[0].role == "user" and "topic Y" in msgs[0].text
assert msgs[1].role == "assistant" and "A1 reply" in msgs[1].text
assert msgs[2].role == "assistant" and msgs[2].text.startswith("Summary of users:")
async def test_sequential_checkpoint_resume_round_trip() -> None:
storage = InMemoryCheckpointStorage()
initial_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder().participants(list(initial_agents)).with_checkpointing(storage).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("checkpoint sequential"):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf_resume = SequentialBuilder().participants(list(resumed_agents)).with_checkpointing(storage).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_sequential_checkpoint_runtime_only() -> None:
"""Test checkpointing configured ONLY at runtime, not at build time."""
storage = InMemoryCheckpointStorage()
agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder().participants(list(agents)).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]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
resumed_agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf_resume = SequentialBuilder().participants(list(resumed_agents)).build()
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id, checkpoint_storage=storage):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data # type: ignore[assignment]
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_sequential_checkpoint_runtime_overrides_buildtime() -> None:
"""Test that runtime checkpoint storage overrides build-time configuration."""
import tempfile
with tempfile.TemporaryDirectory() as temp_dir1, tempfile.TemporaryDirectory() as temp_dir2:
from agent_framework._workflows._checkpoint import FileCheckpointStorage
buildtime_storage = FileCheckpointStorage(temp_dir1)
runtime_storage = FileCheckpointStorage(temp_dir2)
agents = (_EchoAgent(id="agent1", name="A1"), _EchoAgent(id="agent2", name="A2"))
wf = SequentialBuilder().participants(list(agents)).with_checkpointing(buildtime_storage).build()
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]
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
buildtime_checkpoints = await buildtime_storage.list_checkpoints()
runtime_checkpoints = await runtime_storage.list_checkpoints()
assert len(runtime_checkpoints) > 0, "Runtime storage should have checkpoints"
assert len(buildtime_checkpoints) == 0, "Build-time storage should have no checkpoints when overridden"
async def test_sequential_register_participants_with_checkpointing() -> None:
"""Test that checkpointing works with register_participants."""
storage = InMemoryCheckpointStorage()
def create_agent1() -> _EchoAgent:
return _EchoAgent(id="agent1", name="A1")
def create_agent2() -> _EchoAgent:
return _EchoAgent(id="agent2", name="A2")
wf = SequentialBuilder().register_participants([create_agent1, create_agent2]).with_checkpointing(storage).build()
baseline_output: list[ChatMessage] | None = None
async for ev in wf.run_stream("checkpoint with factories"):
if isinstance(ev, WorkflowOutputEvent):
baseline_output = ev.data
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
break
assert baseline_output is not None
checkpoints = await storage.list_checkpoints()
assert checkpoints
checkpoints.sort(key=lambda cp: cp.timestamp)
resume_checkpoint = next(
(cp for cp in checkpoints if (cp.metadata or {}).get("checkpoint_type") == "superstep"),
checkpoints[-1],
)
wf_resume = (
SequentialBuilder().register_participants([create_agent1, create_agent2]).with_checkpointing(storage).build()
)
resumed_output: list[ChatMessage] | None = None
async for ev in wf_resume.run_stream(checkpoint_id=resume_checkpoint.checkpoint_id):
if isinstance(ev, WorkflowOutputEvent):
resumed_output = ev.data
if isinstance(ev, WorkflowStatusEvent) and ev.state in (
WorkflowRunState.IDLE,
WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
):
break
assert resumed_output is not None
assert [m.role for m in resumed_output] == [m.role for m in baseline_output]
assert [m.text for m in resumed_output] == [m.text for m in baseline_output]
async def test_sequential_register_participants_factories_called_on_build() -> None:
"""Test that factories are called during build(), not during register_participants()."""
call_count = 0
def create_agent() -> _EchoAgent:
nonlocal call_count
call_count += 1
return _EchoAgent(id=f"agent{call_count}", name=f"A{call_count}")
builder = SequentialBuilder().register_participants([create_agent, create_agent])
# Factories should not be called yet
assert call_count == 0
wf = builder.build()
# Now factories should have been called
assert call_count == 2
# Run the workflow to ensure it works
completed = False
output: list[ChatMessage] | None = None
async for ev in wf.run_stream("test factories timing"):
if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
completed = True
elif isinstance(ev, WorkflowOutputEvent):
output = ev.data # type: ignore[assignment]
if completed and output is not None:
break
assert completed
assert output is not None
msgs: list[ChatMessage] = output
# Should have user message + 2 agent replies
assert len(msgs) == 3
async def test_sequential_builder_reusable_after_build_with_participants() -> None:
"""Test that the builder can be reused to build multiple identical workflows with participants()."""
a1 = _EchoAgent(id="agent1", name="A1")
a2 = _EchoAgent(id="agent2", name="A2")
builder = SequentialBuilder().participants([a1, a2])
# Build first workflow
builder.build()
assert builder._participants[0] is a1 # type: ignore
assert builder._participants[1] is a2 # type: ignore
assert builder._participant_factories == [] # type: ignore
async def test_sequential_builder_reusable_after_build_with_factories() -> None:
"""Test that the builder can be reused to build multiple workflows with register_participants()."""
call_count = 0
def create_agent1() -> _EchoAgent:
nonlocal call_count
call_count += 1
return _EchoAgent(id="agent1", name="A1")
def create_agent2() -> _EchoAgent:
nonlocal call_count
call_count += 1
return _EchoAgent(id="agent2", name="A2")
builder = SequentialBuilder().register_participants([create_agent1, create_agent2])
# Build first workflow - factories should be called
builder.build()
assert call_count == 2
assert builder._participants == [] # type: ignore
assert len(builder._participant_factories) == 2 # type: ignore
assert builder._participant_factories[0] is create_agent1 # type: ignore
assert builder._participant_factories[1] is create_agent2 # type: ignore
python/packages/core/tests/workflow/test_workflow_kwargs.py
+13 -9
View File
@@ -11,17 +11,19 @@ from agent_framework import (
AgentThread,
BaseAgent,
ChatMessage,
ConcurrentBuilder,
Content,
GroupChatBuilder,
GroupChatState,
HandoffBuilder,
SequentialBuilder,
WorkflowRunState,
WorkflowStatusEvent,
tool,
)
from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY
from agent_framework.orchestrations import (
ConcurrentBuilder,
GroupChatBuilder,
GroupChatState,
HandoffBuilder,
SequentialBuilder,
)
# Track kwargs received by tools during test execution
_received_kwargs: list[dict[str, Any]] = []
@@ -371,14 +373,15 @@ async def test_handoff_kwargs_flow_to_agents() -> None:
async def test_magentic_kwargs_flow_to_agents() -> None:
"""Test that kwargs flow to agents in a magentic workflow via MagenticAgentExecutor."""
from agent_framework import MagenticBuilder
from agent_framework._workflows._magentic import (
from agent_framework_orchestrations._magentic import (
MagenticContext,
MagenticManagerBase,
MagenticProgressLedger,
MagenticProgressLedgerItem,
)
from agent_framework.orchestrations import MagenticBuilder
# Create a mock manager that completes after one round
class _MockManager(MagenticManagerBase):
def __init__(self) -> None:
@@ -422,14 +425,15 @@ async def test_magentic_kwargs_flow_to_agents() -> None:
async def test_magentic_kwargs_stored_in_state() -> None:
"""Test that kwargs are stored in State when using MagenticWorkflow.run_stream()."""
from agent_framework import MagenticBuilder
from agent_framework._workflows._magentic import (
from agent_framework_orchestrations._magentic import (
MagenticContext,
MagenticManagerBase,
MagenticProgressLedger,
MagenticProgressLedgerItem,
)
from agent_framework.orchestrations import MagenticBuilder
class _MockManager(MagenticManagerBase):
def __init__(self) -> None:
super().__init__(max_stall_count=3, max_reset_count=None, max_round_count=1)