Skip to content
Docs

Documentation navigation

Swarms

Multi-agent collaboration with designated final-output agents, per-agent final tools, and member-style registration.

#Swarms

Runnable examples of multiple agents working together. A Swarm holds a list of Agents and routes work between them: the user message goes in, the result of the designated final-output agent comes out.

#A three-agent swarm

This example builds a realistic investment-analysis swarm: a financial analyst, a risk assessor, and a director who synthesizes their findings into a final memo.

Python 
from maivn import Agent, Swarm
from maivn.messages import HumanMessage
from pydantic import BaseModel, Field

# Specialist 1: financial analysis
financial_analyst = Agent(
    name='Financial Analyst',
    system_prompt='Analyze financial data (revenue, EBITDA, margins). Focus on numerical trends.',
    api_key='...',
)

@financial_analyst.toolify(name='calculate_cagr')
def calculate_cagr(start_value: float, end_value: float, periods: int) -> float:
    """Compound Annual Growth Rate."""
    return (end_value / start_value) ** (1 / periods) - 1

# Specialist 2: risk assessment
risk_assessor = Agent(
    name='Risk Assessor',
    system_prompt='Evaluate market volatility, competition, and regulatory risk. Score 1-10.',
    api_key='...',
)

@risk_assessor.toolify(name='calculate_risk_score')
def calculate_risk_score(volatility: float, competition: float, regulation: float) -> float:
    return round(volatility * 0.4 + competition * 0.3 + regulation * 0.3, 1)

# Final output: investment director
investment_director = Agent(
    name='Investment Director',
    use_as_final_output=True,
    system_prompt=(
        'Synthesize the Financial Analyst and Risk Assessor reports into an '
        'Investment Memo. Always call generate_investment_memo as the final tool.'
    ),
    api_key='...',
)

@investment_director.toolify(name='generate_investment_memo', final_tool=True)
class InvestmentMemo(BaseModel):
    company_name: str
    financial_data: dict
    risk_data: dict
    recommendation: str  # Buy / Hold / Sell

swarm = Swarm(
    name='Investment Analysis Swarm',
    description='Comprehensive company evaluation and investment decisioning.',
    agents=[financial_analyst, risk_assessor, investment_director],
)

swarm.invoke(
    [HumanMessage(content=(
        'Evaluate "TechNova Solutions" for a $50M Series C investment. '
        'Revenue: [12.5, 18.2, 25.0, 38.5]M USD over 4 years. '
        'Net income: [1.2, 2.5, 4.8, 8.2]M USD. Provide a memo.'
    ))],
    force_final_tool=True,
)

A few things worth noticing:

  • use_as_final_output=True on investment_director tells the swarm that this agent's output is what the swarm returns.
  • final_tool=True on generate_investment_memo marks it as a terminal, structured result tool.
  • force_final_tool=True at the swarm level requires that path on every invocation where that keyword is supplied. Execution failures still need handling.

#Designating the final-output agent

A swarm needs to know which agent's output to return. Three ways:

  1. use_as_final_output=True on one agent (most explicit):

``text director = Agent(..., use_as_final_output=True) ``

  1. The agent owns the only final_tool — the swarm infers it.
  1. Let the swarm decide — no use_as_final_output and no force_final_tool. The orchestrator picks the most relevant agent based on the task. Useful for exploratory swarms.

#Per-agent final_tool ownership

Multiple agents can each own their own final_tool. The runtime keeps them straight. Mark one agent use_as_final_output=True to disambiguate.

Python 
@category_classifier.toolify(name='emit_category', final_tool=True)
class CategoryDecision(BaseModel):
    ticket_id: str
    product_area: str

@priority_scorer.toolify(name='emit_priority', final_tool=True)
class PriorityDecision(BaseModel):
    ticket_id: str
    priority: str  # P0 / P1 / P2 / P3

resolution_director = Agent(
    name='Resolution Director',
    use_as_final_output=True,   # this one's verdict is the swarm's answer
    system_prompt='Consume category + priority, then call emit_triage_decision.',
    api_key='...',
)

@resolution_director.toolify(name='emit_triage_decision', final_tool=True)
class TriageDecision(BaseModel):
    ticket_id: str
    product_area: str
    priority: str
    assigned_team: str
    next_action: str

triage_swarm = Swarm(
    name='Support Ticket Triage Swarm',
    agents=[category_classifier, priority_scorer, resolution_director],
)

If you forget to designate one final agent and multiple own a final_tool, swarm.validate_tool_configuration() raises so the misconfiguration surfaces before you invoke.

#@swarm.member — factory-style registration

Sometimes it's cleaner to build the agent inside a factory that gets registered with the swarm. The @swarm.member decorator runs the factory and threads agent dependencies through it:

Python 
swarm = Swarm(
    name='Member Planning Swarm',
    system_prompt='Coordinate the member agents. Editor produces the final brief.',
)

@swarm.toolify(name='load_launch_context')
def load_launch_context(market: str = 'healthcare') -> dict:
    return {'market': market, 'launch_window': 'Q3', 'constraints': ['HIPAA review']}

@swarm.member
@depends_on_tool(load_launch_context, arg_name='launch_context')
def research_agent() -> Agent:
    agent = Agent(name='Member Research Agent', system_prompt='...', api_key='...')

    @agent.toolify(name='compile_market_notes')
    def compile_market_notes() -> dict:
        return {'buyer': 'clinic operations leaders', 'risk': 'security review timing'}

    return agent

The strategy editor depends on the research agent's output:

Python 
strategy_editor = Agent(
    name='Member Strategy Editor',
    system_prompt='Use write_launch_brief as the final tool.',
    use_as_final_output=True,
    api_key='...',
)

@strategy_editor.toolify(name='write_launch_brief', final_tool=True)
class LaunchBrief(BaseModel):
    title: str
    market: str
    research_notes: dict     # filled in from the research agent
    recommended_motion: str
    next_steps: list[str]

# Register the editor with a required dependency on research_agent
strategy_editor = swarm.member.depends_on_agent(
    research_agent, arg_name='research_notes',
)(strategy_editor)

The member dependency supplies research context before the editor runs. It does not directly inject the LaunchBrief.research_notes model field; that field is filled during the editor's tool call. For direct field injection, declare the dependency on the final model itself.

#Swarm-level tools

Tools attached directly to the swarm run in the parent invocation. Use them for context loaders or explicit member dependencies. To let a member call one itself, also register the tool on that member with member.add_tool(...):

Python 
@swarm.toolify(name='load_account_context')
def load_account_context(account_id: str) -> dict:
    return {'tier': 'enterprise', 'mrr': 12000, 'csat': 4.6}

Any agent in the swarm can call this tool by name.

#Scaling up: bound resources per agent

For larger swarms, each agent can be bound to its own set of memory resources (PDFs, transcripts, internal docs). The runtime makes each agent's bound resources available in its scope. A member's outputs can still share information with the swarm, so this is not a confidentiality barrier between members. It is useful when you want specialists with distinct knowledge bases:

Python 
brand_strategist = Agent(
    name='Brand Strategist',
    system_prompt='You know the brand strategy deeply. Answer from BRAND.pdf.',
    api_key='...',
    resources=[{'file': 'business/BRAND.pdf', 'name': 'Brand Strategy',
                'mime_type': 'application/pdf', 'binding_type': 'agent'}],
)

valuation_analyst = Agent(
    name='Valuation Analyst',
    system_prompt='You know the financial model deeply. Answer from VALUATION.pdf.',
    api_key='...',
    resources=[{'file': 'business/VALUATION.pdf', 'name': 'Valuation Model',
                'mime_type': 'application/pdf', 'binding_type': 'agent'}],
)

bi_swarm = Swarm(
    name='Business Intelligence Swarm',
    agents=[brand_strategist, valuation_analyst],
)

See Memory for resource lifecycle, retrieval policies, and configuration.

#Async, streaming, batch

Both scopes support sync/async invocation, event streaming, and batching. Agent-only features such as structured_output are not available on Swarm:

Python 
swarm.invoke(messages)
swarm.stream(messages)        # token-level streaming
await swarm.ainvoke(messages)
async for event in swarm.astream(messages):
    handle(event)
swarm.batch([msgs_1, msgs_2, msgs_3])  # multiple invocations concurrently

See Batch & Scheduling for batch semantics.

#What's next

  • Multi-Agent guide — the deeper treatment of swarm orchestration, including final-output policy and resource binding.
  • Memory — bound resources per agent, retrieval, and skills/insights.
  • Agents & Tools — hook execution modes including swarm-wide hooks.