Principle:Microsoft Agent framework Custom Aggregation Pattern

Overview

The Custom Aggregation Pattern defines how users implement their own aggregation logic for concurrent workflows in the Microsoft Agent framework. When multiple agents execute in parallel, each produces an AgentExecutorResponse. The aggregation pattern provides a user-defined callback mechanism that receives the complete list of these response objects and returns a single, synthesized result.

This is a user-defined pattern, not a built-in library API. The framework provides the execution infrastructure for running agents concurrently and collecting their responses; the user supplies the aggregation logic that determines how those responses are combined.

Problem Statement

In multi-agent workflows, parallel execution of agents produces multiple independent results. Without a structured aggregation step, the caller must manually collect, iterate, and combine these results every time. The Custom Aggregation Pattern establishes a consistent interface for defining reusable aggregation callbacks that the workflow engine invokes automatically once all parallel agents have completed.

Pattern Description

The pattern consists of three elements:

1. Parallel agent execution -- The workflow engine dispatches work to multiple agents concurrently.
2. Response collection -- The engine collects an AgentExecutorResponse from each agent upon completion.
3. User-defined aggregation callback -- A callback function supplied by the user receives the full list of AgentExecutorResponse objects and returns a synthesized result of any type.

Callback Signatures

Users may define their aggregator callback using one of two supported signatures:

# Option 1: Simple callback -- receives only the list of results
async def aggregator(results: list[AgentExecutorResponse]) -> Any:
    ...

# Option 2: Context-aware callback -- also receives the workflow context
async def aggregator(results: list[AgentExecutorResponse], ctx: WorkflowContext) -> Any:
    ...

Option 1 is suitable for stateless aggregation where the callback only needs the agent responses themselves. Option 2 provides access to the WorkflowContext, which is useful when the aggregation logic depends on workflow-level metadata, configuration, or shared state.

AgentExecutorResponse Fields

Each AgentExecutorResponse object in the results list exposes the following fields:

Field	Type	Description
.executor_id	str	Unique identifier of the agent executor that produced this response.
.agent_response	AgentResponse	The structured response object returned by the agent, containing the output text and any additional metadata.
.full_conversation	list[Message]	The complete conversation history between the workflow engine and the agent during this execution cycle.

Design Principles

User Ownership of Aggregation Logic

The framework deliberately does not prescribe how results should be combined. Users retain full control over:

• Filtering -- Ignoring responses from certain agents based on content or executor identity.
• Ranking -- Ordering responses by quality, confidence, or relevance before merging.
• Summarization -- Condensing multiple verbose responses into a compact synthesis.
• Voting / Consensus -- Selecting the majority answer from multiple agents.
• Structured output -- Returning typed objects, dictionaries, or domain-specific data rather than plain text.

Composability

Aggregator callbacks are ordinary Python async functions. They can be composed, chained, or wrapped with decorators. A library of reusable aggregators can be built and shared across projects without any framework-specific registration.

Usage Guidelines

1. Keep aggregators focused -- Each callback should perform a single, well-defined aggregation strategy.
2. Handle partial failures -- Check for error states within individual AgentExecutorResponse objects before accessing their fields.
3. Prefer the simple signature -- Use the two-argument form (results, ctx) only when the aggregation logic genuinely requires workflow context.
4. Return explicit types -- Annotate the return type of the aggregator to improve downstream type safety.

Example

A basic aggregator that concatenates all agent outputs into a labeled summary:

async def custom_aggregator(results: list[AgentExecutorResponse]) -> str:
    summary_parts = []
    for r in results:
        agent_name = r.executor_id
        response_text = r.agent_response.text
        summary_parts.append(f"[{agent_name}]: {response_text}")
    return "\n\n".join(summary_parts)

Related

• Implementation:Microsoft_Agent_framework_Custom_Aggregator_Callback -- The implementation-level documentation for custom aggregator callbacks.