diff --git a/python/packages/main/agent_framework/workflow/__init__.py b/python/packages/main/agent_framework/workflow/__init__.py index db8bccbcb2..cb8cd56fcf 100644 --- a/python/packages/main/agent_framework/workflow/__init__.py +++ b/python/packages/main/agent_framework/workflow/__init__.py @@ -56,6 +56,7 @@ _IMPORTS = [ "PlanReviewRequest", "RequestInfoEvent", "StandardMagenticManager", + "ConcurrentBuilder", ] diff --git a/python/packages/main/agent_framework/workflow/__init__.pyi b/python/packages/main/agent_framework/workflow/__init__.pyi index 63f301775f..ea88728525 100644 --- a/python/packages/main/agent_framework/workflow/__init__.pyi +++ b/python/packages/main/agent_framework/workflow/__init__.pyi @@ -8,6 +8,7 @@ from agent_framework_workflow import ( AgentRunUpdateEvent, Case, CheckpointStorage, + ConcurrentBuilder, Default, Executor, ExecutorCompletedEvent, @@ -56,6 +57,7 @@ __all__ = [ "AgentRunUpdateEvent", "Case", "CheckpointStorage", + "ConcurrentBuilder", "Default", "Executor", "ExecutorCompletedEvent", diff --git a/python/packages/workflow/agent_framework_workflow/__init__.py b/python/packages/workflow/agent_framework_workflow/__init__.py index 25fe4591de..5d3e1352fb 100644 --- a/python/packages/workflow/agent_framework_workflow/__init__.py +++ b/python/packages/workflow/agent_framework_workflow/__init__.py @@ -9,6 +9,7 @@ from ._checkpoint import ( InMemoryCheckpointStorage, WorkflowCheckpoint, ) +from ._concurrent import ConcurrentBuilder from ._const import ( DEFAULT_MAX_ITERATIONS, ) @@ -93,6 +94,7 @@ __all__ = [ "AgentRunUpdateEvent", "Case", "CheckpointStorage", + "ConcurrentBuilder", "Default", "EdgeDuplicationError", "Executor", diff --git a/python/packages/workflow/agent_framework_workflow/_concurrent.py b/python/packages/workflow/agent_framework_workflow/_concurrent.py new file mode 100644 index 0000000000..14feb6bc8e --- /dev/null +++ b/python/packages/workflow/agent_framework_workflow/_concurrent.py @@ -0,0 +1,304 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import inspect +import logging +from collections.abc import Callable, Sequence +from typing import Any + +from agent_framework import AgentProtocol, ChatMessage, Role + +from ._events import WorkflowCompletedEvent +from ._executor import AgentExecutorRequest, AgentExecutorResponse, Executor, handler +from ._workflow import Workflow, 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 should be AgentProtocol instances or Executors. +- A custom aggregator can be provided as: + - an Executor instance (it should handle list[AgentExecutorResponse] and add a WorkflowCompletedEvent), or + - a callback function with signature: + def cb(results: list[AgentExecutorResponse]) -> Any | None + def cb(results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None + If the callback returns a non-None value, it is sent as the data of a WorkflowCompletedEvent. + If it returns None, the callback may have already emitted a completion event via ctx. +""" + + +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=[ChatMessage(Role.USER, text=prompt)], should_respond=True) + await ctx.send_message(request) + + @handler + async def from_message(self, message: ChatMessage, ctx: WorkflowContext[AgentExecutorRequest]) -> None: # type: ignore[name-defined] + request = AgentExecutorRequest(messages=[message], should_respond=True) + await ctx.send_message(request) + + @handler + async def from_messages(self, messages: list[ChatMessage], ctx: WorkflowContext[AgentExecutorRequest]) -> None: # type: ignore[name-defined] + request = AgentExecutorRequest(messages=list(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_run_response.messages). + - Avoids duplicating the same user message per agent. + """ + + @handler + async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> 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: Role) -> 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 = getattr(role, "value", None) + if role_str is None: + role_str = str(role) + role_str = role_str.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_run_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', '')}: " + 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, Role.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, Role.ASSISTANT)), None) + if final_assistant is None: + final_assistant = next((m for m in reversed(conv) if _is_role(m, Role.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', '')}; 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.add_event(WorkflowCompletedEvent(data=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 wrapped in a WorkflowCompletedEvent. + """ + + def __init__(self, callback: Callable[..., Any], id: str | None = None) -> None: + super().__init__(id) + self._callback = callback + self._param_count = len(inspect.signature(callback).parameters) + + @handler + async def aggregate(self, results: list[AgentExecutorResponse], ctx: WorkflowContext[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.add_event(WorkflowCompletedEvent(ret)) + + +class ConcurrentBuilder: + r"""High-level builder for concurrent agent workflows. + + - `participants([...])` accepts a list of AgentProtocol (recommended) or Executor. + - `build()` wires: dispatcher -> fan-out -> participants -> fan-in -> aggregator. + - `with_custom_aggregator(...)` overrides the default aggregator with an Executor or callback. + + Usage: + ```python + from agent_framework.workflow import ConcurrentBuilder + + # Minimal: use default aggregator (returns list[ChatMessage]) + workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).build() + + + # Custom aggregator via callback (sync or async). The callback receives + # list[AgentExecutorResponse] and its return value becomes + # WorkflowCompletedEvent.data + def summarize(results): + return " | ".join(r.agent_run_response.messages[-1].text for r in results) + + + workflow = ConcurrentBuilder().participants([agent1, agent2, agent3]).with_custom_aggregator(summarize).build() + ``` + """ + + def __init__(self) -> None: + self._participants: list[AgentProtocol | Executor] = [] + self._aggregator: Executor | None = None + + 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. + + Raises: + ValueError: if `participants` is empty or contains duplicates + TypeError: if any entry is not AgentProtocol or Executor + + Example: + ```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 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 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 add a + `WorkflowCompletedEvent` to the context. + - Callback: sync or async callable with one of the signatures: + `(results: list[AgentExecutorResponse]) -> Any | None` or + `(results: list[AgentExecutorResponse], ctx: WorkflowContext[Any]) -> Any | None`. + If the callback returns a non-None value, it becomes the + `WorkflowCompletedEvent.data`. + + Example: + ```python + # Callback-based aggregator (string result) + async def summarize(results): + return " | ".join(r.agent_run_response.messages[-1].text for r in results) + + + wf = ConcurrentBuilder().participants([a1, a2, a3]).with_custom_aggregator(summarize).build() + ``` + """ + 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 build(self) -> Workflow: + r"""Build and validate the concurrent workflow. + + Wiring pattern: + - Dispatcher (internal) fans out the input to all `participants` + - Fan-in aggregator collects `AgentExecutorResponse` objects + - Aggregator emits a `WorkflowCompletedEvent` with either: + - list[ChatMessage] (default aggregator: one user + one assistant per agent) + - custom payload from the provided callback/executor + + Returns: + Workflow: a ready-to-run workflow instance + + Raises: + ValueError: if no participants were defined + + Example: + ```python + workflow = ConcurrentBuilder().participants([agent1, agent2]).build() + ``` + """ + if not self._participants: + raise ValueError("No participants provided. Call .participants([...]) first.") + + dispatcher = _DispatchToAllParticipants(id="dispatcher") + aggregator = self._aggregator or _AggregateAgentConversations(id="aggregator") + + builder = WorkflowBuilder() + return ( + builder.set_start_executor(dispatcher) + .add_fan_out_edges(dispatcher, list(self._participants)) + .add_fan_in_edges(list(self._participants), aggregator) + .build() + ) diff --git a/python/packages/workflow/tests/test_concurrent.py b/python/packages/workflow/tests/test_concurrent.py new file mode 100644 index 0000000000..25becca7cd --- /dev/null +++ b/python/packages/workflow/tests/test_concurrent.py @@ -0,0 +1,126 @@ +# Copyright (c) Microsoft. All rights reserved. + +from typing import Any, cast + +import pytest +from agent_framework import AgentRunResponse, ChatMessage, Role + +from agent_framework_workflow import ( + AgentExecutorRequest, + AgentExecutorResponse, + ConcurrentBuilder, + Executor, + WorkflowCompletedEvent, + WorkflowContext, + handler, +) + + +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 = AgentRunResponse(messages=ChatMessage(Role.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]) + + +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: WorkflowCompletedEvent | None = None + async for ev in wf.run_stream("prompt: hello world"): + if isinstance(ev, WorkflowCompletedEvent): + completed = ev + break + + assert completed is not None + assert isinstance(completed.data, list) + messages: list[ChatMessage] = cast(list[ChatMessage], completed.data) # type: ignore + + # Expect one user message + one assistant message per participant + assert len(messages) == 1 + 3 + assert messages[0].role == 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 == 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_run_response.messages + texts.append(msgs[-1].text if msgs else "") + return " | ".join(sorted(texts)) + + wf = ConcurrentBuilder().participants([e1, e2]).with_aggregator(summarize).build() + + completed: WorkflowCompletedEvent | None = None + async for ev in wf.run_stream("prompt: custom"): + if isinstance(ev, WorkflowCompletedEvent): + completed = ev + break + + assert completed is not None + # Custom aggregator returns a string payload + assert isinstance(completed.data, str) + assert completed.data == "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_run_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: WorkflowCompletedEvent | None = None + async for ev in wf.run_stream("prompt: custom sync"): + if isinstance(ev, WorkflowCompletedEvent): + completed = ev + break + + assert completed is not None + assert isinstance(completed.data, str) + assert completed.data == "One | Two" diff --git a/python/samples/getting_started/workflow/README.md b/python/samples/getting_started/workflow/README.md index 744ad3c9c0..9fa42bc209 100644 --- a/python/samples/getting_started/workflow/README.md +++ b/python/samples/getting_started/workflow/README.md @@ -78,6 +78,9 @@ Once comfortable with these, explore the rest of the samples below. ### orchestration | Sample | File | Concepts | |---|---|---| +| Concurrent Orchestration (Default Aggregator) | [orchestration/concurrent_agents.py](./orchestration/concurrent_agents.py) | Fan-out to multiple agents; fan-in with default aggregator returning combined ChatMessages | +| Concurrent Orchestration (Custom Aggregator) | [orchestration/concurrent_custom_aggregator.py](./orchestration/concurrent_custom_aggregator.py) | Override aggregator via callback; summarize results with an LLM | +| Concurrent Orchestration (Custom Agent Executors) | [orchestration/concurrent_custom_agent_executors.py](./orchestration/concurrent_custom_agent_executors.py) | Child executors own ChatAgents; concurrent fan-out/fan-in via ConcurrentBuilder | | Magentic Workflow (Multi-Agent) | [orchestration/magentic.py](./orchestration/magentic.py) | Orchestrate multiple agents with Magentic manager and streaming | | Magentic + Human Plan Review | [orchestration/magentic_human_plan_update.py](./orchestration/magentic_human_plan_update.py) | Human reviews/updates the plan before execution | diff --git a/python/samples/getting_started/workflow/orchestration/concurrent_agents.py b/python/samples/getting_started/workflow/orchestration/concurrent_agents.py new file mode 100644 index 0000000000..b8e165c5b4 --- /dev/null +++ b/python/samples/getting_started/workflow/orchestration/concurrent_agents.py @@ -0,0 +1,131 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from typing import Any + +from agent_framework import ChatMessage +from agent_framework.azure import AzureChatClient +from agent_framework.workflow import ConcurrentBuilder, WorkflowCompletedEvent +from azure.identity import AzureCliCredential + +""" +Sample: Concurrent fan-out/fan-in (agent-only API) with default aggregator + +Build a high-level concurrent workflow using ConcurrentBuilder and three domain agents. +The default dispatcher fans out the same user prompt to all agents in parallel. +The default aggregator fans in their results and emits a WorkflowCompletedEvent whose +data is a list[ChatMessage] representing the concatenated conversations from all agents. + +Demonstrates: +- Minimal wiring with ConcurrentBuilder().participants([...]).build() +- Fan-out to multiple agents, fan-in aggregation of final ChatMessages +- Streaming of AgentRunEvent for simple progress visibility + +Prerequisites: +- Azure OpenAI access configured for AzureChatClient (use az login + env vars) +- Familiarity with Workflow events (AgentRunEvent, WorkflowCompletedEvent) +""" + + +async def main() -> None: + # 1) Create three domain agents using AzureChatClient + chat_client = AzureChatClient(credential=AzureCliCredential()) + + researcher = chat_client.create_agent( + instructions=( + "You're an expert market and product researcher. Given a prompt, provide concise, factual insights," + " opportunities, and risks." + ), + name="researcher", + ) + + marketer = chat_client.create_agent( + instructions=( + "You're a creative marketing strategist. Craft compelling value propositions and target messaging" + " aligned to the prompt." + ), + name="marketer", + ) + + legal = chat_client.create_agent( + instructions=( + "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns" + " based on the prompt." + ), + name="legal", + ) + + # 2) Build a concurrent workflow + # Participants are either Agents (type of AgentProtocol) or Executors + workflow = ConcurrentBuilder().participants([researcher, marketer, legal]).build() + + # 3) Run with a single prompt, stream progress, and pretty-print the final combined messages + completion: WorkflowCompletedEvent | None = None + async for event in workflow.run_stream("We are launching a new budget-friendly electric bike for urban commuters."): + if isinstance(event, WorkflowCompletedEvent): + completion = event + + if completion: + print("===== Final Aggregated Conversation (messages) =====") + messages: list[ChatMessage] | Any = completion.data + for i, msg in enumerate(messages, start=1): + name = msg.author_name if msg.author_name else "user" + print(f"{'-' * 60}\n\n{i:02d} [{name}]:\n{msg.text}") + + """ + Sample Output: + + ===== Final Aggregated Conversation (messages) ===== + ------------------------------------------------------------ + + 01 [user]: + We are launching a new budget-friendly electric bike for urban commuters. + ------------------------------------------------------------ + + 02 [researcher]: + **Insights:** + + - **Target Demographic:** Urban commuters seeking affordable, eco-friendly transport; + likely to include students, young professionals, and price-sensitive urban residents. + - **Market Trends:** E-bike sales are growing globally, with increasing urbanization, + higher fuel costs, and sustainability concerns driving adoption. + - **Competitive Landscape:** Key competitors include brands like Rad Power Bikes, Aventon, + Lectric, and domestic budget-focused manufacturers in North America, Europe, and Asia. + - **Feature Expectations:** Customers expect reliability, ease-of-use, theft protection, + lightweight design, sufficient battery range for daily city commutes (typically 25-40 miles), + and low-maintenance components. + + **Opportunities:** + + - **First-time Buyers:** Capture newcomers to e-biking by emphasizing affordability, ease of + operation, and cost savings vs. public transit/car ownership. + ... + ------------------------------------------------------------ + + 03 [marketer]: + **Value Proposition:** + "Empowering your city commute: Our new electric bike combines affordability, reliability, and + sustainable design—helping you conquer urban journeys without breaking the bank." + + **Target Messaging:** + + *For Young Professionals:* + ... + ------------------------------------------------------------ + + 04 [legal]: + **Constraints, Disclaimers, & Policy Concerns for Launching a Budget-Friendly Electric Bike for Urban Commuters:** + + **1. Regulatory Compliance** + - Verify that the electric bike meets all applicable federal, state, and local regulations + regarding e-bike classification, speed limits, power output, and safety features. + - Ensure necessary certifications (e.g., UL certification for batteries, CE markings if sold internationally) are obtained. + + **2. Product Safety** + - Include consumer safety warnings regarding use, battery handling, charging protocols, and age restrictions. + ... + """ # noqa: E501 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/samples/getting_started/workflow/orchestration/concurrent_custom_agent_executors.py b/python/samples/getting_started/workflow/orchestration/concurrent_custom_agent_executors.py new file mode 100644 index 0000000000..067489dee2 --- /dev/null +++ b/python/samples/getting_started/workflow/orchestration/concurrent_custom_agent_executors.py @@ -0,0 +1,175 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from typing import Any + +from agent_framework import ChatAgent, ChatMessage +from agent_framework.azure import AzureChatClient +from agent_framework.workflow import ( + AgentExecutorRequest, + AgentExecutorResponse, + ConcurrentBuilder, + Executor, + WorkflowCompletedEvent, + WorkflowContext, + handler, +) +from azure.identity import AzureCliCredential + +""" +Sample: Concurrent Orchestration with Custom Agent Executors + +This sample shows a concurrent fan-out/fan-in pattern using child Executor classes +that each own their ChatAgent. The executors accept AgentExecutorRequest inputs +and emit AgentExecutorResponse outputs, which allows reuse of the high-level +ConcurrentBuilder API and the default aggregator. + +Demonstrates: +- Executors that create their ChatAgent in __init__ (via AzureChatClient) +- A @handler that converts AgentExecutorRequest -> AgentExecutorResponse +- ConcurrentBuilder().participants([...]) to build fan-out/fan-in +- Default aggregator returning list[ChatMessage] (one user + one assistant per agent) + +Prerequisites: +- Azure OpenAI configured for AzureChatClient (az login + required env vars) +""" + + +class ResearcherExec(Executor): + agent: ChatAgent + + def __init__(self, chat_client: AzureChatClient, id: str = "researcher"): + agent = chat_client.create_agent( + instructions=( + "You're an expert market and product researcher. Given a prompt, provide concise, factual insights," + " opportunities, and risks." + ), + name=id, + ) + super().__init__(agent=agent, id=id) + + @handler + async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None: + response = await self.agent.run(request.messages) + full_conversation = list(request.messages) + list(response.messages) + await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation)) + + +class MarketerExec(Executor): + agent: ChatAgent + + def __init__(self, chat_client: AzureChatClient, id: str = "marketer"): + agent = chat_client.create_agent( + instructions=( + "You're a creative marketing strategist. Craft compelling value propositions and target messaging" + " aligned to the prompt." + ), + name=id, + ) + super().__init__(agent=agent, id=id) + + @handler + async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None: + response = await self.agent.run(request.messages) + full_conversation = list(request.messages) + list(response.messages) + await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation)) + + +class LegalExec(Executor): + agent: ChatAgent + + def __init__(self, chat_client: AzureChatClient, id: str = "legal"): + agent = chat_client.create_agent( + instructions=( + "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns" + " based on the prompt." + ), + name=id, + ) + super().__init__(agent=agent, id=id) + + @handler + async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None: + response = await self.agent.run(request.messages) + full_conversation = list(request.messages) + list(response.messages) + await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation)) + + +async def main() -> None: + chat_client = AzureChatClient(credential=AzureCliCredential()) + + researcher = ResearcherExec(chat_client) + marketer = MarketerExec(chat_client) + legal = LegalExec(chat_client) + + workflow = ConcurrentBuilder().participants([researcher, marketer, legal]).build() + + completion: WorkflowCompletedEvent | None = None + async for event in workflow.run_stream("We are launching a new budget-friendly electric bike for urban commuters."): + if isinstance(event, WorkflowCompletedEvent): + completion = event + + if completion: + print("===== Final Aggregated Conversation (messages) =====") + messages: list[ChatMessage] | Any = completion.data + for i, msg in enumerate(messages, start=1): + name = msg.author_name if msg.author_name else "user" + print(f"{'-' * 60}\n\n{i:02d} [{name}]:\n{msg.text}") + + """ + Sample Output: + + ===== Final Aggregated Conversation (messages) ===== + ------------------------------------------------------------ + + 01 [user]: + We are launching a new budget-friendly electric bike for urban commuters. + ------------------------------------------------------------ + + 02 [researcher]: + **Insights:** + + - **Target Demographic:** Urban commuters seeking affordable, eco-friendly transport; + likely to include students, young professionals, and price-sensitive urban residents. + - **Market Trends:** E-bike sales are growing globally, with increasing urbanization, + higher fuel costs, and sustainability concerns driving adoption. + - **Competitive Landscape:** Key competitors include brands like Rad Power Bikes, Aventon, + Lectric, and domestic budget-focused manufacturers in North America, Europe, and Asia. + - **Feature Expectations:** Customers expect reliability, ease-of-use, theft protection, + lightweight design, sufficient battery range for daily city commutes (typically 25-40 miles), + and low-maintenance components. + + **Opportunities:** + + - **First-time Buyers:** Capture newcomers to e-biking by emphasizing affordability, ease of + operation, and cost savings vs. public transit/car ownership. + ... + ------------------------------------------------------------ + + 03 [marketer]: + **Value Proposition:** + "Empowering your city commute: Our new electric bike combines affordability, reliability, and + sustainable design—helping you conquer urban journeys without breaking the bank." + + **Target Messaging:** + + *For Young Professionals:* + ... + ------------------------------------------------------------ + + 04 [legal]: + **Constraints, Disclaimers, & Policy Concerns for Launching a Budget-Friendly Electric Bike for Urban Commuters:** + + **1. Regulatory Compliance** + - Verify that the electric bike meets all applicable federal, state, and local regulations + regarding e-bike classification, speed limits, power output, and safety features. + - Ensure necessary certifications (e.g., UL certification for batteries, CE markings if sold internationally) are obtained. + + **2. Product Safety** + - Include consumer safety warnings regarding use, battery handling, charging protocols, and age restrictions. + ... + """ # noqa: E501 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/samples/getting_started/workflow/orchestration/concurrent_custom_aggregator.py b/python/samples/getting_started/workflow/orchestration/concurrent_custom_aggregator.py new file mode 100644 index 0000000000..73f95388c6 --- /dev/null +++ b/python/samples/getting_started/workflow/orchestration/concurrent_custom_aggregator.py @@ -0,0 +1,125 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +from typing import Any + +from agent_framework import ChatMessage, Role +from agent_framework.azure import AzureChatClient +from agent_framework.workflow import ConcurrentBuilder, WorkflowCompletedEvent +from azure.identity import AzureCliCredential + +""" +Sample: Concurrent Orchestration with Custom Aggregator + +Build a concurrent workflow with ConcurrentBuilder that fans out one prompt to +multiple domain agents and fans in their responses. Override the default +aggregator with a custom async callback that uses AzureChatClient.get_response() +to synthesize a concise, consolidated summary from the experts' outputs. + +Demonstrates: +- ConcurrentBuilder().participants([...]).with_custom_aggregator(callback) +- Fan-out to agents and fan-in at an aggregator +- Aggregation implemented via an LLM call (chat_client.get_response) +- WorkflowCompletedEvent carrying the synthesized summary string + +Prerequisites: +- Azure OpenAI configured for AzureChatClient (az login + required env vars) +""" + + +async def main() -> None: + chat_client = AzureChatClient(credential=AzureCliCredential()) + + researcher = chat_client.create_agent( + instructions=( + "You're an expert market and product researcher. Given a prompt, provide concise, factual insights," + " opportunities, and risks." + ), + name="researcher", + ) + marketer = chat_client.create_agent( + instructions=( + "You're a creative marketing strategist. Craft compelling value propositions and target messaging" + " aligned to the prompt." + ), + name="marketer", + ) + legal = chat_client.create_agent( + instructions=( + "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns" + " based on the prompt." + ), + name="legal", + ) + + # Define a custom aggregator callback that uses the chat client to summarize + async def summarize_results(results: list[Any]) -> str: + # Extract one final assistant message per agent + expert_sections: list[str] = [] + for r in results: + try: + messages = getattr(r.agent_run_response, "messages", []) + final_text = messages[-1].text if messages and hasattr(messages[-1], "text") else "(no content)" + expert_sections.append(f"{getattr(r, 'executor_id', 'expert')}:\n{final_text}") + except Exception as e: + expert_sections.append(f"{getattr(r, 'executor_id', 'expert')}: (error: {type(e).__name__}: {e})") + + # Ask the model to synthesize a concise summary of the experts' outputs + system_msg = ChatMessage( + Role.SYSTEM, + text=( + "You are a helpful assistant that consolidates multiple domain expert outputs " + "into one cohesive, concise summary with clear takeaways. Keep it under 200 words." + ), + ) + user_msg = ChatMessage(Role.USER, text="\n\n".join(expert_sections)) + + response = await chat_client.get_response([system_msg, user_msg]) + # Return the model's final assistant text as the completion result + return response.messages[-1].text if response.messages else "" + + # Build with a custom aggregator callback function + # - participants([...]) accepts AgentProtocol (agents) or Executor instances. + # Each participant becomes a parallel branch (fan-out) from an internal dispatcher. + # - with_aggregator(...) overrides the default aggregator: + # • Default aggregator -> returns list[ChatMessage] (one user + one assistant per agent) + # • Custom callback -> return value becomes WorkflowCompletedEvent.data (string here) + # The callback can be sync or async; it receives list[AgentExecutorResponse]. + workflow = ( + ConcurrentBuilder().participants([researcher, marketer, legal]).with_aggregator(summarize_results).build() + ) + + completion: WorkflowCompletedEvent | None = None + async for event in workflow.run_stream("We are launching a new budget-friendly electric bike for urban commuters."): + if isinstance(event, WorkflowCompletedEvent): + completion = event + + if completion: + print("===== Final Consolidated Output =====") + print(completion.data) + + """ + Sample Output: + + ===== Final Consolidated Output ===== + Urban e-bike demand is rising rapidly due to eco-awareness, urban congestion, and high fuel costs, + with market growth projected at a ~10% CAGR through 2030. Key customer concerns are affordability, + easy maintenance, convenient charging, compact design, and theft protection. Differentiation opportunities + include integrating smart features (GPS, app connectivity), offering subscription or leasing options, and + developing portable, space-saving designs. Partnering with local governments and bike shops can boost visibility. + + Risks include price wars eroding margins, regulatory hurdles, battery quality concerns, and heightened expectations + for after-sales support. Accurate, substantiated product claims and transparent marketing (with range disclaimers) + are essential. All e-bikes must comply with local and federal regulations on speed, wattage, safety certification, + and labeling. Clear warranty, safety instructions (especially regarding batteries), and inclusive, accessible + marketing are required. For connected features, data privacy policies and user consents are mandatory. + + Effective messaging should target young professionals, students, eco-conscious commuters, and first-time buyers, + emphasizing affordability, convenience, and sustainability. Slogan suggestion: “Charge Ahead—City Commutes Made + Affordable.” Legal review in each target market, compliance vetting, and robust customer support policies are + critical before launch. + """ + + +if __name__ == "__main__": + asyncio.run(main())