Python: Add factory pattern to concurrent orchestration builder

python/packages/core/agent_framework/_workflows/_concurrent.py

@@ -3,6 +3,7 @@
 import asyncio
 import inspect
 import logging
+import uuid
 from collections.abc import Callable, Sequence
 from typing import Any
 
@@ -189,8 +190,10 @@ 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_custom_aggregator(...)` overrides the default aggregator with an Executor or callback.
+    - `with_aggregator(...)` overrides the default aggregator with an Executor or callback.
 
     Usage:
 
@@ -201,14 +204,17 @@ class 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):
+        def summarize(results: list[AgentExecutorResponse]) -> str:
             return " | ".join(r.agent_run_response.messages[-1].text for r in results)
 
 
-        workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_custom_aggregator(summarize).build()
+        workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_aggregator(summarize).build()
 
         # Enable checkpoint persistence so runs can resume
         workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_checkpointing(storage).build()
@@ -219,10 +225,69 @@ def summarize(results):
 
     def __init__(self) -> None:
         self._participants: list[AgentProtocol | Executor] = []
-        self._aggregator: Executor | None = None
+        self._participant_factories: list[Callable[[], AgentProtocol | Executor]] = []
+        self._aggregator: (
+            Executor
+            | Callable[[], Executor]
+            | Callable[[list[AgentExecutorResponse]], Any]
+            | Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any]
+            | None
+        ) = None
         self._checkpoint_storage: CheckpointStorage | None = None
         self._request_info_enabled: 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.
+
+        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.
 
@@ -231,7 +296,8 @@ def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "Con
         from an internal dispatcher.
 
         Raises:
-            ValueError: if `participants` is empty or contains duplicates
+            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:
@@ -243,6 +309,14 @@ def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "Con
             # 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")
 
@@ -265,12 +339,18 @@ def participants(self, participants: Sequence[AgentProtocol | Executor]) -> "Con
         self._participants = list(participants)
         return self
 
-    def with_aggregator(self, aggregator: Executor | Callable[..., 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(...)` and add a
-          output and the workflow becomes idle.
+    def with_aggregator(
+        self,
+        aggregator: Executor
+        | Callable[[], Executor]
+        | 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.
+
+        - Executor: must handle `list[AgentExecutorResponse]` and yield output using `ctx.yield_output(...)`
+        - Executor factory: callable returning an Executor instance that handles `list[AgentExecutorResponse]`
+          and yields 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`.
@@ -279,20 +359,44 @@ def with_aggregator(self, aggregator: Executor | Callable[..., Any]) -> "Concurr
         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_run_response.messages[-1].text for r in results))
+
+
+            wf = ConcurrentBuilder().participants([a1, a2, a3]).with_aggregator(CustomAggregator()).build()
+
+            # Factory-based aggregator
+            wf = (
+                ConcurrentBuilder()
+                .participants([a1, a2, a3])
+                .with_aggregator(lambda: CustomAggregator(id="custom_aggregator"))
+                .build()
+            )
+
 
             # Callback-based aggregator (string result)
-            async def summarize(results):
+            async def summarize(results: list[AgentExecutorResponse]) -> str:
                 return " | ".join(r.agent_run_response.messages[-1].text for r in results)
 
 
-            wf = ConcurrentBuilder().participants([a1, a2, a3]).with_custom_aggregator(summarize).build()
+            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_run_response.messages[-1].text for r in results))
+
+
+            wf = ConcurrentBuilder().participants([a1, a2, a3]).with_aggregator(summarize).build()
         """
-        if isinstance(aggregator, Executor):
+        if isinstance(aggregator, Executor) or callable(aggregator):
             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":
@@ -329,7 +433,7 @@ def build(self) -> Workflow:
             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 callback/executor
+          - custom payload from the provided aggregator
 
         Returns:
             Workflow: a ready-to-run workflow instance
@@ -343,25 +447,92 @@ def build(self) -> Workflow:
 
             workflow = ConcurrentBuilder().participants([agent1, agent2]).build()
         """
-        if not self._participants:
-            raise ValueError("No participants provided. Call .participants([...]) first.")
+        if not self._participants and not self._participant_factories:
+            raise ValueError(
+                "No participants provided. Call .participants([...]) or .register_participants([...]) first."
+            )
 
+        # Internal nodes
         dispatcher = _DispatchToAllParticipants(id="dispatcher")
-        aggregator = self._aggregator or _AggregateAgentConversations(id="aggregator")
+        if isinstance(self._aggregator, Executor):
+            # Case 1: Executor instance - use directly
+            aggregator = self._aggregator
+        elif callable(self._aggregator):
+            # Distinguish between an aggregator factory (could also be a class) and callback-based aggregator
+            if inspect.isclass(self._aggregator):
+                aggregator = self._aggregator()
+            else:
+                # Check the signature: factory has 0 params, callback has 1-2 params
+                sig = inspect.signature(self._aggregator)
+                param_count = len(sig.parameters)
+
+                # Case 2: Executor factory (no parameters) - call it to create the executor
+                # Case 3: Callback with parameters (1-2 params) - wrap in _CallbackAggregator
+                aggregator = self._aggregator() if param_count == 0 else _CallbackAggregator(self._aggregator)  # type: ignore
+
+            if not isinstance(aggregator, Executor):
+                raise TypeError(f"Aggregator factory must return an Executor; got {type(aggregator).__name__}")
+        else:
+            # Case 4: No custom aggregator provided - use the default one
+            aggregator = _AggregateAgentConversations(id="aggregator")
+
+        participants: list[Executor | AgentProtocol] = []
+        if self._participant_factories:
+            for factory in self._participant_factories:
+                p = factory()
+                if not isinstance(p, (AgentProtocol, Executor)):
+                    raise TypeError(
+                        f"Participant factory must return AgentProtocol or Executor; got {type(p).__name__}"
+                    )
+                participants.append(p)
+        else:
+            participants = self._participants
 
         builder = WorkflowBuilder()
-        builder.set_start_executor(dispatcher)
-        builder.add_fan_out_edges(dispatcher, list(self._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(list(self._participants), request_info_interceptor)
-            builder.add_edge(request_info_interceptor, aggregator)
+        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] = []
+            for p in participants:
+                factory_name = uuid.uuid4().hex
+                factory_names.append(factory_name)
+                if isinstance(p, Executor):
+                    builder.register_executor(lambda p=p: p, name=factory_name)
+                else:
+                    builder.register_agent(lambda p=p: p, name=factory_name)
+            # 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")
         else:
-            # Direct fan-in to aggregator
-            builder.add_fan_in_edges(list(self._participants), aggregator)
+            builder.set_start_executor(dispatcher)
+            builder.add_fan_out_edges(dispatcher, list(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(list(participants), request_info_interceptor)
+                builder.add_edge(request_info_interceptor, aggregator)
+            else:
+                # Direct fan-in to aggregator
+                builder.add_fan_in_edges(list(participants), aggregator)
 
         if self._checkpoint_storage is not None:
             builder = builder.with_checkpointing(self._checkpoint_storage)

python/packages/core/agent_framework/_workflows/_workflow_builder.py

@@ -1148,21 +1148,29 @@ def _resolve_edge_registry(self) -> tuple[Executor, list[Executor], list[EdgeGro
         if isinstance(self._start_executor, Executor):
             start_executor = self._start_executor
 
-        executors: dict[str, Executor] = {}
+        # Maps registered factory names to created executor instances for edge resolution
+        factory_name_to_instance: dict[str, Executor] = {}
+        # Maps executor IDs to created executor instances to prevent duplicates
+        executor_id_to_instance: dict[str, Executor] = {}
         deferred_edge_groups: list[EdgeGroup] = []
         for name, exec_factory in self._executor_registry.items():
             instance = exec_factory()
+            if instance.id in executor_id_to_instance:
+                raise ValueError(f"Executor with ID '{instance.id}' has already been registered.")
+            executor_id_to_instance[instance.id] = instance
+
             if isinstance(self._start_executor, str) and name == self._start_executor:
                 start_executor = instance
+
             # All executors will get their own internal edge group for receiving system messages
             deferred_edge_groups.append(InternalEdgeGroup(instance.id))  # type: ignore[call-arg]
-            executors[name] = instance
+            factory_name_to_instance[name] = instance
 
         def _get_executor(name: str) -> Executor:
             """Helper to get executor by the registered name. Raises if not found."""
-            if name not in executors:
-                raise ValueError(f"Executor with name '{name}' has not been registered.")
-            return executors[name]
+            if name not in factory_name_to_instance:
+                raise ValueError(f"Executor with factory name '{name}' has not been registered.")
+            return factory_name_to_instance[name]
 
         for registration in self._edge_registry:
             match registration:
@@ -1201,7 +1209,7 @@ def _get_executor(name: str) -> Executor:
         if start_executor is None:
             raise ValueError("Failed to resolve starting executor from registered factories.")
 
-        return start_executor, list(executors.values()), deferred_edge_groups
+        return start_executor, list(executor_id_to_instance.values()), deferred_edge_groups
 
     def build(self) -> Workflow:
         """Build and return the constructed workflow.