Implementation:Microsoft Autogen Swarm Init

Knowledge Sources	Microsoft AutoGen
Domains	Multi-Agent Systems, Swarm Intelligence, Team Orchestration, Workflow Architecture
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for assembling a swarm team where agent routing is driven by handoff messages, provided by Microsoft AutoGen.

Description

The Swarm class extends BaseGroupChat and creates a team where the next speaker is determined exclusively by HandoffMessage events rather than by a centralized selector. The constructor accepts a list of ChatAgent participants, with the first participant designated as the initial speaker.

During initialization, the constructor validates that:

All participants are instances of ChatAgent (inner teams are not supported).
The first participant is capable of producing HandoffMessage instances (checked via produced_message_types).

The orchestration is managed by an internal SwarmGroupChatManager that implements the speaker selection logic: it scans the message thread in reverse for the most recent HandoffMessage and sets the target as the next speaker. If no handoff is found, the current speaker continues.

The class also implements the Component protocol with SwarmConfig, supporting declarative serialization and deserialization of the entire team configuration.

Usage

Import and instantiate Swarm after creating specialized agents with handoff capabilities. Pass the agents as participants (order matters: the first agent speaks first), configure a termination condition, and optionally set max turns. Then execute with run() or run_stream().

Code Reference

Source Location

Repository: Microsoft AutoGen
File: python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_swarm_group_chat.py (Lines 233-265)

Signature

class Swarm(BaseGroupChat, Component[SwarmConfig]):
    def __init__(
        self,
        participants: List[ChatAgent],
        *,
        name: str | None = None,
        description: str | None = None,
        termination_condition: TerminationCondition | None = None,
        max_turns: int | None = None,
        runtime: AgentRuntime | None = None,
        custom_message_types: List[type[BaseAgentEvent | BaseChatMessage]] | None = None,
        emit_team_events: bool = False,
    ) -> None

Import

from autogen_agentchat.teams import Swarm

I/O Contract

Inputs

Name	Type	Required	Description
participants	List[ChatAgent]	Yes	The agents in the swarm. The first agent is the initial speaker and must be able to produce HandoffMessage. Order determines who speaks first.
name	None	No	Name of the team. Defaults to "Swarm". Must be unique if used as a nested team.
description	None	No	Description of the team. Defaults to "A team of agents.".
termination_condition	None	No	Condition that stops the swarm. Without one, the swarm runs indefinitely (unless max_turns is set).
max_turns	None	No	Maximum number of turns before stopping. Defaults to None (no limit).
runtime	None	No	Custom agent runtime. Defaults to None (uses a built-in SingleThreadedAgentRuntime).
custom_message_types	None	No	Custom message types used in the group chat. Required if agents produce non-standard message types.
emit_team_events	bool	No	Whether to emit team-level events in the output stream. Defaults to False.

Outputs

Name	Type	Description
Swarm instance	Swarm	A configured swarm team ready for execution via run() or run_stream(). Manages speaker transitions based on HandoffMessage events.

Usage Examples

Basic Example

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # Agent 1: initial speaker, can hand off to agent 2
    alice = AssistantAgent(
        "Alice",
        model_client=model_client,
        handoffs=["Bob"],
        system_message="You are Alice. Answer questions about yourself. For anything else, transfer to Bob.",
    )

    # Agent 2: handles tasks Alice cannot
    bob = AssistantAgent(
        "Bob",
        model_client=model_client,
        system_message="You are Bob. Your birthday is January 1st.",
    )

    # Assemble the swarm: Alice speaks first
    termination = MaxMessageTermination(5)
    team = Swarm(
        participants=[alice, bob],
        termination_condition=termination,
    )

    # Execute the swarm
    stream = team.run_stream(task="What is Bob's birthday?")
    async for message in stream:
        print(message)


asyncio.run(main())

Related Pages

Implements Principle

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment