Implementation:Microsoft Agent framework ConcurrentBuilder With Aggregator

Knowledge Sources	Microsoft Agent Framework
Domains	Agent_Architecture, Multi_Agent_Systems
Last Updated	2026-02-11 17:00 GMT

Overview

The ConcurrentBuilder.with_aggregator() method allows developers to override the default aggregation behavior of a concurrent workflow by supplying a custom Executor instance or a callback function that synthesizes the list of parallel agent responses into a unified output.

Description

The with_aggregator() method is a builder-pattern method on ConcurrentBuilder that accepts an aggregation strategy and returns the builder instance for method chaining. The aggregator is invoked after all concurrent agent participants have completed execution, receiving the full list of AgentExecutorResponse objects.

The method supports two aggregator forms:

1. Executor instance: A subclass of Executor with a handler method that receives list[AgentExecutorResponse] and a WorkflowContext. The handler uses ctx.yield_output() to emit the aggregated result.
2. Callable (sync or async): A function matching one of two signatures:
   • (results: list[AgentExecutorResponse]) -> Any | None
   • (results: list[AgentExecutorResponse], ctx: WorkflowContext) -> Any | None

   If the callable returns a non-None value, that value becomes the workflow output. Internally, callable aggregators are wrapped in a _CallbackAggregator adapter.

The method enforces a single-assignment constraint: calling with_aggregator() more than once on the same builder raises a ValueError, ensuring each concurrent workflow has exactly one aggregation strategy.

Usage

Import ConcurrentBuilder from the agent_framework.orchestrations package. After constructing a builder with a list of participants, call .with_aggregator() with either an Executor instance or a callable, then call .build() to produce the final workflow.

Code Reference

Source Location

• Repository: agent-framework
• File: python/packages/orchestrations/agent_framework_orchestrations/_concurrent.py
• Lines: L266-321

Signature

def with_aggregator(
    self,
    aggregator: Executor
    | Callable[[list[AgentExecutorResponse]], Any]
    | Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any],
) -> "ConcurrentBuilder":

Import

from agent_framework.orchestrations import ConcurrentBuilder

I/O Contract

Inputs

Parameter	Type	Required	Description
aggregator	Callable[[list[AgentExecutorResponse]], Any] | Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any]	Yes	The custom aggregation strategy. An Executor instance is used directly. A callable is wrapped in an internal _CallbackAggregator adapter. Must be provided exactly once per builder instance.

Output

Type	Description
ConcurrentBuilder	Returns self to support fluent method chaining (e.g., .with_aggregator(...).build()).

Errors

Exception	Condition
ValueError	Raised if with_aggregator() has already been called on this builder instance. Each builder supports exactly one aggregator assignment.
TypeError	Raised if the aggregator argument is neither an Executor instance nor a callable.

Execution Flow

The with_aggregator() method implements the following sequence:

1. Guard Check: If self._aggregator is already set (not None), raise a ValueError to prevent double-assignment.
2. Type Dispatch:
   • If aggregator is an Executor instance, assign it directly to self._aggregator.
   • If aggregator is callable, wrap it in a _CallbackAggregator and assign the wrapper to self._aggregator.
   • Otherwise, raise a TypeError.
3. Return Self: Return the builder instance to enable fluent chaining.

def with_aggregator(self, aggregator):
    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

Usage Examples

Simple Callback Aggregator

from agent_framework.orchestrations import ConcurrentBuilder

async def summarize(results: list[AgentExecutorResponse]) -> str:
    return " | ".join(r.agent_response.messages[-1].text for r in results)

workflow = ConcurrentBuilder(participants=[a1, a2]).with_aggregator(summarize).build()

The summarize callback receives the list of all concurrent agent responses and joins their final messages with a pipe separator. Because it returns a non-None string, that string becomes the workflow output.

Executor-Based Aggregator With Streaming

from agent_framework.orchestrations import ConcurrentBuilder, Executor, handler

class CustomAggregator(Executor):
    @handler
    async def aggregate(
        self,
        results: list[AgentExecutorResponse],
        ctx: WorkflowContext,
    ) -> None:
        combined = " | ".join(r.agent_response.messages[-1].text for r in results)
        await ctx.yield_output(combined)

workflow = (
    ConcurrentBuilder(participants=[a1, a2, a3])
    .with_aggregator(CustomAggregator())
    .build()
)

The CustomAggregator Executor provides full access to the WorkflowContext, enabling streaming of aggregated output via ctx.yield_output().

Callback Aggregator With WorkflowContext

from agent_framework.orchestrations import ConcurrentBuilder

async def summarize_with_ctx(
    results: list[AgentExecutorResponse],
    ctx: WorkflowContext[Never, str],
) -> None:
    combined = " | ".join(r.agent_response.messages[-1].text for r in results)
    await ctx.yield_output(combined)

workflow = ConcurrentBuilder(participants=[a1, a2]).with_aggregator(summarize_with_ctx).build()

A callback can also accept a WorkflowContext as its second parameter for advanced scenarios like streaming output without implementing a full Executor subclass.

Consensus Voting Aggregator

from collections import Counter
from agent_framework.orchestrations import ConcurrentBuilder

async def majority_vote(results: list[AgentExecutorResponse]) -> str:
    answers = [r.agent_response.messages[-1].text.strip() for r in results]
    most_common, _ = Counter(answers).most_common(1)[0]
    return most_common

workflow = (
    ConcurrentBuilder(participants=[agent_a, agent_b, agent_c])
    .with_aggregator(majority_vote)
    .build()
)

This aggregator selects the most frequently occurring answer from three parallel agents, implementing a simple majority-vote consensus strategy.

Related Pages

Implements Principle

• Principle:Microsoft_Agent_framework_Response_Aggregation

Sources

Type	Name	URL
Repo	Microsoft Agent Framework	https://github.com/microsoft/agent-framework