Principle:Microsoft Agent framework Response Aggregation

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

Overview

A configuration pattern for customizing how concurrent agent responses are combined into a unified output, by overriding the default aggregation behavior of the ConcurrentBuilder with a custom Executor or callback function via the with_aggregator() method.

Description

Response Aggregation addresses the need for flexible post-processing when multiple agents execute in parallel within a concurrent workflow. By default, the ConcurrentBuilder collects all parallel agent responses and returns them as-is. The with_aggregator() method allows developers to inject a custom aggregation strategy that synthesizes the raw list of AgentExecutorResponse objects into a single, meaningful output.

The aggregator can be supplied in two forms:

1. Executor-based: A full Executor subclass with a handler method that receives the list of responses and a WorkflowContext, enabling complex aggregation logic including streaming intermediate outputs via ctx.yield_output().
2. Callback-based: A simple sync or async callable that accepts the list of responses (and optionally a WorkflowContext) and returns a synthesized value. If the callback returns a non-None value, that value becomes the workflow's final output.

This separation of the aggregation policy from the execution mechanism allows developers to declaratively control output synthesis at workflow-construction time, without modifying the underlying concurrent execution logic.

Usage

Use this principle when building concurrent (fan-out) workflows where the raw list of individual agent responses is insufficient and a synthesized, merged, or summarized output is required. Common scenarios include:

• Summarization: Combining multiple specialist agent outputs into a single narrative summary.
• Voting / Consensus: Selecting the most common or highest-confidence answer from multiple agents.
• Structured Merging: Assembling partial results from multiple agents into a unified structured object (e.g., combining a research agent's findings with an analysis agent's conclusions).
• Filtering: Discarding low-quality or irrelevant agent responses before returning the final output.

Theoretical Basis

The Response Aggregation pattern implements the Fan-Out / Fan-In concurrency model, where a single input is distributed to multiple parallel workers (fan-out) and their outputs are collected and combined by an aggregator (fan-in). The with_aggregator() method provides the fan-in customization point.

This follows the Strategy Pattern: the aggregation strategy is selected at workflow-construction time and injected into the builder, allowing the concurrent execution engine to remain agnostic to how results are synthesized. The builder enforces a single-assignment constraint -- calling with_aggregator() more than once raises a ValueError -- ensuring a clear, unambiguous aggregation policy per workflow.

The dual-form interface (Executor vs. callback) follows the Progressive Disclosure principle:

1. Simple use cases are served by a plain callback function, minimizing boilerplate.
2. Advanced use cases (streaming partial aggregations, accessing workflow context) are served by the full Executor interface.

# Conceptual flow (pseudocode)
responses: list[AgentExecutorResponse] = await run_all_agents_concurrently(participants, input)

if custom_aggregator is not None:
    output = await custom_aggregator.handle(responses, ctx)
else:
    output = responses  # default: return raw list

I/O Contract

Configuration Inputs

Parameter	Type	Required	Description
aggregator	Callable[[list[AgentExecutorResponse]], Any] | Callable[[list[AgentExecutorResponse], WorkflowContext[Never, Any]], Any]	Yes	The custom aggregation strategy to apply to the collected parallel agent responses. Accepts an Executor instance for full control (including streaming via ctx.yield_output()), or a sync/async callable for simple transformations. If the callable returns a non-None value, it becomes the workflow output.

Runtime Outputs

Output	Type	Description
Aggregated result	Any	The synthesized output produced by the custom aggregator. The type depends on the aggregator implementation: it may be a string, a structured object, or None if the aggregator yields output via the WorkflowContext instead of returning a value.

Usage Examples

Callback-Based Aggregation (String Concatenation)

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()

Executor-Based Aggregation (Streaming Output)

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()

Callback-Based Aggregation 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()

Related Pages

Implemented By

• Implementation:Microsoft_Agent_framework_ConcurrentBuilder_With_Aggregator

Sources

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