mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Remove warnings from workflow builder on not using factories (#2808)
* Revert concurrent * Fix comments
This commit is contained in:
Tao Chen
committed by
GitHub
Unverified
GitHub
parent
e008144187
commit
e0ff153ee9
@@ -3,7 +3,6 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import logging
|
||||
import uuid
|
||||
from collections.abc import Callable, Sequence
|
||||
from typing import Any
|
||||
|
||||
@@ -30,7 +29,8 @@ parallel workflow with:
|
||||
- a default aggregator that combines all agent conversations and completes the workflow
|
||||
|
||||
Notes:
|
||||
- Participants should be AgentProtocol instances or Executors.
|
||||
- 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
|
||||
@@ -396,7 +396,7 @@ class ConcurrentBuilder:
|
||||
| Callable[[list[AgentExecutorResponse]], Any]
|
||||
| Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any],
|
||||
) -> "ConcurrentBuilder":
|
||||
r"""Override the default aggregator with an executor, an executor factory, or a callback.
|
||||
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:
|
||||
@@ -521,52 +521,30 @@ class ConcurrentBuilder:
|
||||
)
|
||||
)
|
||||
|
||||
builder = WorkflowBuilder()
|
||||
participants: list[Executor | AgentProtocol] = []
|
||||
if self._participant_factories:
|
||||
# Register executors/agents to avoid warnings from the workflow builder
|
||||
# if factories are provided instead of direct instances. This doesn't
|
||||
# break the factory pattern since the concurrent builder still creates
|
||||
# new instances per workflow build.
|
||||
factory_names: list[str] = []
|
||||
# Resolve the participant factories now. This doesn't break the factory pattern
|
||||
# since the Concurrent builder still creates new instances per workflow build.
|
||||
for factory in self._participant_factories:
|
||||
factory_name = uuid.uuid4().hex
|
||||
factory_names.append(factory_name)
|
||||
instance = factory()
|
||||
if isinstance(instance, Executor):
|
||||
builder.register_executor(lambda executor=instance: executor, name=factory_name) # type: ignore[misc]
|
||||
else:
|
||||
builder.register_agent(lambda agent=instance: agent, name=factory_name) # type: ignore[misc]
|
||||
# Register the dispatcher and the aggregator
|
||||
builder.register_executor(lambda: dispatcher, name="dispatcher")
|
||||
builder.register_executor(lambda: aggregator, name="aggregator")
|
||||
|
||||
builder.set_start_executor("dispatcher")
|
||||
builder.add_fan_out_edges("dispatcher", factory_names)
|
||||
if self._request_info_enabled:
|
||||
# Insert interceptor between fan-in and aggregator
|
||||
# participants -> fan-in -> interceptor -> aggregator
|
||||
builder.register_executor(
|
||||
lambda: RequestInfoInterceptor(executor_id="request_info"),
|
||||
name="request_info_interceptor",
|
||||
)
|
||||
builder.add_fan_in_edges(factory_names, "request_info_interceptor")
|
||||
builder.add_edge("request_info_interceptor", "aggregator")
|
||||
else:
|
||||
# Direct fan-in to aggregator
|
||||
builder.add_fan_in_edges(factory_names, "aggregator")
|
||||
p = factory()
|
||||
participants.append(p)
|
||||
else:
|
||||
builder.set_start_executor(dispatcher)
|
||||
builder.add_fan_out_edges(dispatcher, self._participants)
|
||||
participants = self._participants
|
||||
|
||||
builder = WorkflowBuilder()
|
||||
builder.set_start_executor(dispatcher)
|
||||
builder.add_fan_out_edges(dispatcher, participants)
|
||||
|
||||
if self._request_info_enabled:
|
||||
# Insert interceptor between fan-in and aggregator
|
||||
# participants -> fan-in -> interceptor -> aggregator
|
||||
request_info_interceptor = RequestInfoInterceptor(executor_id="request_info")
|
||||
builder.add_fan_in_edges(participants, request_info_interceptor)
|
||||
builder.add_edge(request_info_interceptor, aggregator)
|
||||
else:
|
||||
# Direct fan-in to aggregator
|
||||
builder.add_fan_in_edges(participants, aggregator)
|
||||
|
||||
if self._request_info_enabled:
|
||||
# Insert interceptor between fan-in and aggregator
|
||||
# participants -> fan-in -> interceptor -> aggregator
|
||||
request_info_interceptor = RequestInfoInterceptor(executor_id="request_info")
|
||||
builder.add_fan_in_edges(self._participants, request_info_interceptor)
|
||||
builder.add_edge(request_info_interceptor, aggregator)
|
||||
else:
|
||||
# Direct fan-in to aggregator
|
||||
builder.add_fan_in_edges(self._participants, aggregator)
|
||||
if self._checkpoint_storage is not None:
|
||||
builder = builder.with_checkpointing(self._checkpoint_storage)
|
||||
|
||||
|
||||
@@ -4,7 +4,8 @@
|
||||
|
||||
This module provides a high-level, agent-focused API to assemble a sequential
|
||||
workflow where:
|
||||
- Participants are a sequence of AgentProtocol instances or Executors
|
||||
- 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
|
||||
@@ -15,7 +16,7 @@ Typical wiring:
|
||||
|
||||
Notes:
|
||||
- Participants can mix AgentProtocol and Executor objects
|
||||
- Agents are auto-wrapped by WorkflowBuilder as AgentExecutor
|
||||
- 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
|
||||
@@ -252,7 +253,7 @@ class SequentialBuilder:
|
||||
if not self._participants and not self._participant_factories:
|
||||
raise ValueError(
|
||||
"No participants or participant factories provided to the builder. "
|
||||
"Use .participants([...]) or .ss([...])."
|
||||
"Use .participants([...]) or .register_participants([...])."
|
||||
)
|
||||
|
||||
if self._participants and self._participant_factories:
|
||||
@@ -273,6 +274,8 @@ class SequentialBuilder:
|
||||
|
||||
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)
|
||||
|
||||
@@ -302,7 +302,8 @@ class WorkflowBuilder:
|
||||
|
||||
If multiple names are provided, the same factory function will be registered under each name.
|
||||
|
||||
...code-block:: python
|
||||
.. code-block:: python
|
||||
|
||||
from agent_framework import WorkflowBuilder, Executor, WorkflowContext, handler
|
||||
|
||||
|
||||
@@ -315,7 +316,7 @@ class WorkflowBuilder:
|
||||
# Register the same executor factory under multiple names
|
||||
workflow = (
|
||||
WorkflowBuilder()
|
||||
.register_executor(lambda: CustomExecutor(id="logger"), name=["ExecutorA", "ExecutorB"])
|
||||
.register_executor(lambda: LoggerExecutor(id="logger"), name=["ExecutorA", "ExecutorB"])
|
||||
.set_start_executor("ExecutorA")
|
||||
.add_edge("ExecutorA", "ExecutorB")
|
||||
.build()
|
||||
@@ -456,7 +457,12 @@ class WorkflowBuilder:
|
||||
source: The source executor or registered name of the source factory for the edge.
|
||||
target: The target executor or registered name of the target factory for the edge.
|
||||
condition: An optional condition function that determines whether the edge
|
||||
should be traversed based on the message type.
|
||||
should be traversed based on the message.
|
||||
|
||||
Note: If instances are provided for both source and target, they will be shared across
|
||||
all workflow instances created from the built Workflow. To avoid this, consider
|
||||
registering the executors and agents using `register_executor` and `register_agent`
|
||||
and referencing them by factory name for lazy initialization instead.
|
||||
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
@@ -505,17 +511,13 @@ class WorkflowBuilder:
|
||||
.build()
|
||||
)
|
||||
"""
|
||||
if not isinstance(source, str) or not isinstance(target, str):
|
||||
logger.warning(
|
||||
"Adding an edge with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instances. "
|
||||
"Consider using a registered name for lazy initialization instead."
|
||||
)
|
||||
|
||||
if (isinstance(source, str) and not isinstance(target, str)) or (
|
||||
not isinstance(source, str) and isinstance(target, str)
|
||||
):
|
||||
raise ValueError("Both source and target must be either names (str) or Executor/AgentProtocol instances.")
|
||||
raise ValueError(
|
||||
"Both source and target must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if isinstance(source, str) and isinstance(target, str):
|
||||
# Both are names; defer resolution to build time
|
||||
@@ -547,6 +549,11 @@ class WorkflowBuilder:
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
|
||||
Note: If instances are provided for source and targets, they will be shared across
|
||||
all workflow instances created from the built Workflow. To avoid this, consider
|
||||
registering the executors and agents using `register_executor` and `register_agent`
|
||||
and referencing them by factory name for lazy initialization instead.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
|
||||
@@ -583,17 +590,13 @@ class WorkflowBuilder:
|
||||
.build()
|
||||
)
|
||||
"""
|
||||
if not isinstance(source, str) or any(not isinstance(t, str) for t in targets):
|
||||
logger.warning(
|
||||
"Adding fan-out edges with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instances. "
|
||||
"Consider using registered names for lazy initialization instead."
|
||||
)
|
||||
|
||||
if (isinstance(source, str) and not all(isinstance(t, str) for t in targets)) or (
|
||||
not isinstance(source, str) and any(isinstance(t, str) for t in targets)
|
||||
):
|
||||
raise ValueError("Both source and targets must be either names (str) or Executor/AgentProtocol instances.")
|
||||
raise ValueError(
|
||||
"Both source and targets must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if isinstance(source, str) and all(isinstance(t, str) for t in targets):
|
||||
# Both are names; defer resolution to build time
|
||||
@@ -624,7 +627,7 @@ class WorkflowBuilder:
|
||||
Each condition function will be evaluated in order, and the first one that returns True
|
||||
will determine which target executor receives the message.
|
||||
|
||||
The last case (the default case) will receive messages that fall through all conditions
|
||||
The default case (if provided) will receive messages that fall through all conditions
|
||||
(i.e., no condition matched).
|
||||
|
||||
Args:
|
||||
@@ -634,6 +637,11 @@ class WorkflowBuilder:
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
|
||||
Note: If instances are provided for source and case targets, they will be shared across
|
||||
all workflow instances created from the built Workflow. To avoid this, consider
|
||||
registering the executors and agents using `register_executor` and `register_agent`
|
||||
and referencing them by factory name for lazy initialization instead.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
|
||||
@@ -681,18 +689,12 @@ class WorkflowBuilder:
|
||||
.build()
|
||||
)
|
||||
"""
|
||||
if not isinstance(source, str) or not all(isinstance(case.target, str) for case in cases):
|
||||
logger.warning(
|
||||
"Adding a switch-case edge group with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instance. "
|
||||
"Consider using a registered name for lazy initialization instead."
|
||||
)
|
||||
|
||||
if (isinstance(source, str) and not all(isinstance(case.target, str) for case in cases)) or (
|
||||
not isinstance(source, str) and any(isinstance(case.target, str) for case in cases)
|
||||
):
|
||||
raise ValueError(
|
||||
"Both source and case targets must be either names (str) or Executor/AgentProtocol instances."
|
||||
"Both source and case targets must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if isinstance(source, str) and all(isinstance(case.target, str) for case in cases):
|
||||
@@ -736,12 +738,16 @@ class WorkflowBuilder:
|
||||
source: The source executor or registered name of the source factory for the edge group.
|
||||
targets: A list of target executors or registered names of the target factories for the edges.
|
||||
selection_func: A function that selects target executors for messages.
|
||||
Takes (message, list[executor_id or registered target names]) and
|
||||
returns list[executor_id or registered target names].
|
||||
Takes (message, list[executor_id]) and returns list[executor_id].
|
||||
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
|
||||
Note: If instances are provided for source and targets, they will be shared across
|
||||
all workflow instances created from the built Workflow. To avoid this, consider
|
||||
registering the executors and agents using `register_executor` and `register_agent`
|
||||
and referencing them by factory name for lazy initialization instead.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
|
||||
@@ -795,17 +801,13 @@ class WorkflowBuilder:
|
||||
.build()
|
||||
)
|
||||
"""
|
||||
if not isinstance(source, str) or any(not isinstance(t, str) for t in targets):
|
||||
logger.warning(
|
||||
"Adding fan-out edges with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instances. "
|
||||
"Consider using registered names for lazy initialization instead."
|
||||
)
|
||||
|
||||
if (isinstance(source, str) and not all(isinstance(t, str) for t in targets)) or (
|
||||
not isinstance(source, str) and any(isinstance(t, str) for t in targets)
|
||||
):
|
||||
raise ValueError("Both source and targets must be either names (str) or Executor/AgentProtocol instances.")
|
||||
raise ValueError(
|
||||
"Both source and targets must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if isinstance(source, str) and all(isinstance(t, str) for t in targets):
|
||||
# Both are names; defer resolution to build time
|
||||
@@ -848,6 +850,11 @@ class WorkflowBuilder:
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
|
||||
Note: If instances are provided for sources and target, they will be shared across
|
||||
all workflow instances created from the built Workflow. To avoid this, consider
|
||||
registering the executors and agents using `register_executor` and `register_agent`
|
||||
and referencing them by factory name for lazy initialization instead.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
|
||||
@@ -879,17 +886,13 @@ class WorkflowBuilder:
|
||||
.build()
|
||||
)
|
||||
"""
|
||||
if not all(isinstance(s, str) for s in sources) or not isinstance(target, str):
|
||||
logger.warning(
|
||||
"Adding fan-in edges with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instances. "
|
||||
"Consider using registered names for lazy initialization instead."
|
||||
)
|
||||
|
||||
if (all(isinstance(s, str) for s in sources) and not isinstance(target, str)) or (
|
||||
not all(isinstance(s, str) for s in sources) and isinstance(target, str)
|
||||
):
|
||||
raise ValueError("Both sources and target must be either names (str) or Executor/AgentProtocol instances.")
|
||||
raise ValueError(
|
||||
"Both sources and target must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if all(isinstance(s, str) for s in sources) and isinstance(target, str):
|
||||
# Both are names; defer resolution to build time
|
||||
@@ -911,7 +914,7 @@ class WorkflowBuilder:
|
||||
The output of each executor in the chain will be sent to the next executor in the chain.
|
||||
The input types of each executor must be compatible with the output types of the previous executor.
|
||||
|
||||
Circles in the chain are not allowed, meaning the chain cannot have two executors with the same ID.
|
||||
Cycles in the chain are not allowed, meaning an executor cannot appear more than once in the chain.
|
||||
|
||||
Args:
|
||||
executors: A list of executors or registered names of the executor factories to chain together.
|
||||
@@ -919,6 +922,11 @@ class WorkflowBuilder:
|
||||
Returns:
|
||||
Self: The WorkflowBuilder instance for method chaining.
|
||||
|
||||
Note: If executor instances are provided, they will be shared across all workflow instances created
|
||||
from the built Workflow. To avoid this, consider registering the executors and agents using
|
||||
`register_executor` and `register_agent` and referencing them by factory name for lazy
|
||||
initialization instead.
|
||||
|
||||
Example:
|
||||
.. code-block:: python
|
||||
|
||||
@@ -958,16 +966,10 @@ class WorkflowBuilder:
|
||||
if len(executors) < 2:
|
||||
raise ValueError("At least two executors are required to form a chain.")
|
||||
|
||||
if not all(isinstance(e, str) for e in executors):
|
||||
logger.warning(
|
||||
"Adding a chain with Executor or AgentProtocol instances directly is not recommended, "
|
||||
"because workflow instances created from the builder will share the same executor/agent instances. "
|
||||
"Consider using registered names for lazy initialization instead."
|
||||
)
|
||||
|
||||
if not all(isinstance(e, str) for e in executors) and any(isinstance(e, str) for e in executors):
|
||||
raise ValueError(
|
||||
"All executors in the chain must be either names (str) or Executor/AgentProtocol instances."
|
||||
"All executors in the chain must be either registered factory names (str) "
|
||||
"or Executor/AgentProtocol instances."
|
||||
)
|
||||
|
||||
if all(isinstance(e, str) for e in executors):
|
||||
@@ -976,7 +978,7 @@ class WorkflowBuilder:
|
||||
self.add_edge(executors[i], executors[i + 1])
|
||||
return self
|
||||
|
||||
# Both are Executor/AgentProtocol instances; wrap and add now
|
||||
# All are Executor/AgentProtocol instances; wrap and add now
|
||||
# Wrap each candidate first to ensure stable IDs before adding edges
|
||||
wrapped: list[Executor] = [self._maybe_wrap_agent(e) for e in executors] # type: ignore[arg-type]
|
||||
for i in range(len(wrapped) - 1):
|
||||
|
||||
@@ -63,7 +63,7 @@ def test_concurrent_builder_rejects_duplicate_executors_from_factories() -> None
|
||||
return _FakeAgentExec("dup", "B") # same executor id
|
||||
|
||||
builder = ConcurrentBuilder().register_participants([create_dup1, create_dup2])
|
||||
with pytest.raises(ValueError, match="Executor with ID 'dup' has already been created."):
|
||||
with pytest.raises(ValueError, match="Duplicate executor ID 'dup' detected in workflow."):
|
||||
builder.build()
|
||||
|
||||
|
||||
|
||||
@@ -497,7 +497,13 @@ def test_mixing_eager_and_lazy_initialization_error():
|
||||
builder.register_executor(lambda: MockExecutor(id="Lazy"), name="Lazy")
|
||||
|
||||
# Mixing eager and lazy should raise an error during add_edge
|
||||
with pytest.raises(ValueError, match="Both source and target must be either names"):
|
||||
with pytest.raises(
|
||||
ValueError,
|
||||
match=(
|
||||
r"Both source and target must be either registered factory names \(str\) "
|
||||
r"or Executor/AgentProtocol instances\."
|
||||
),
|
||||
):
|
||||
builder.add_edge(eager_executor, "Lazy")
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user