Python: Add factory pattern to concurrent orchestration builder

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

@@ -189,8 +189,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 +203,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 +224,65 @@ 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()` was 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 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 +291,7 @@ 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()` was called
             TypeError: if any entry is not AgentProtocol or Executor
 
         Example:
@@ -243,6 +303,11 @@ 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 not participants:
             raise ValueError("participants cannot be empty")
 
@@ -265,12 +330,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 +350,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 +424,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 +438,60 @@ 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))
+        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(self._participants), request_info_interceptor)
+            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(self._participants), aggregator)
+            builder.add_fan_in_edges(list(participants), aggregator)
 
         if self._checkpoint_storage is not None:
             builder = builder.with_checkpointing(self._checkpoint_storage)