Team Module

The team module provides classes for multi-agent collaboration.

Base Team Class

class pantheon.team.Team(agents: list[Agent | RemoteAgent])

Bases: ABC

__init__(agents: list[Agent | RemoteAgent])

async async_setup()

async gather_events()

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, **kwargs)

async chat(message: str | dict | None = None)

Chat with the team with a REPL interface.

Team Types

SequentialTeam

class pantheon.team.SequentialTeam(agents: list[Agent], connect_prompt: str | list[str] = 'Next:')

Bases: Team

Team that run agents in sequential order.

Agents execute one after another in sequence.

Example:

from pantheon.team import SequentialTeam
from pantheon.agent import Agent

researcher = Agent(name="researcher", instructions="Research topics")
writer = Agent(name="writer", instructions="Write summaries")

team = SequentialTeam([researcher, writer])
await team.chat("Research and summarize AI trends")

__init__(agents: list[Agent], connect_prompt: str | list[str] = 'Next:')

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, connect_prompt: str | list[str] | None = None, agent_kwargs: dict = {}, **final_kwargs)

SwarmTeam

class pantheon.team.SwarmTeam(agents: list[Agent | RemoteAgent])

Bases: Team

Team that run agents in handoff & routines patterns like OpenAI’s [Swarm framework](openai/swarm).

Agents can dynamically transfer control to each other.

Key Features:

• Dynamic handoffs between agents

• Agents decide when to transfer

• Flexible routing based on context

Example:

from pantheon.team import SwarmTeam

agent1 = Agent(name="Agent1", instructions="First agent")
agent2 = Agent(name="Agent2", instructions="Second agent")

@agent1.tool
def transfer_to_agent2():
    """Transfer control to Agent2."""
    return agent2

team = SwarmTeam([agent1, agent2])
await team.chat()

__init__(agents: list[Agent | RemoteAgent])

get_active_agent(memory: Memory) → Agent | RemoteAgent

set_active_agent(memory: Memory, agent_name: str)

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, memory: Memory | None = None, **kwargs)

SwarmCenterTeam

class pantheon.team.SwarmCenterTeam(triage: Agent, agents: list[Agent | RemoteAgent])

Bases: SwarmTeam

Swarm team that has a central triage agent that decides which agent to handoff to.

A central coordinator manages worker agents.

Structure:

• First agent is the coordinator

• Remaining agents are workers

• Coordinator distributes tasks

Example:

from pantheon.team import SwarmCenterTeam

coordinator = Agent(
    name="coordinator",
    instructions="Coordinate tasks between workers"
)
worker1 = Agent(name="worker1", instructions="Handle type A tasks")
worker2 = Agent(name="worker2", instructions="Handle type B tasks")

team = SwarmCenterTeam([coordinator, worker1, worker2])

__init__(triage: Agent, agents: list[Agent | RemoteAgent])

async add_agent(agent: Agent | RemoteAgent)

async remove_agent(agent: Agent | RemoteAgent)

async async_setup()

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, **kwargs)

MoATeam

class pantheon.team.MoATeam(proposers: list[Agent], aggregator: Agent, layers: int = 1, parallel: bool = True)

Bases: Team

Team that run agents in a MoA (Mixture-of-Agents) pattern.

Reference:

• [MoA: Mixure-of-Agents](https://arxiv.org/abs/2406.04692)

• [Self-MoA](https://arxiv.org/abs/2502.00674)

Mixture of Agents - multiple agents propose, final agent synthesizes.

Structure:

• All agents except last: Propose solutions

• Last agent: Synthesize best ideas

• Parallel proposal generation

Example:

from pantheon.team import MoATeam

expert1 = Agent(name="expert1", instructions="Provide approach 1")
expert2 = Agent(name="expert2", instructions="Provide approach 2")
synthesizer = Agent(
    name="synthesizer",
    instructions="Combine best ideas from all experts"
)

team = MoATeam([expert1, expert2, synthesizer])

AGGREGATION_TEMPLATE = "Below are responses from different AI models to the same query.\nPlease carefully analyze these responses and generate a final answer that is:\n- Most accurate and comprehensive\n- Best aligned with the user's instructions\n- Free from errors or inconsistencies\n\n### Query:\n{user_query}\n\n### Responses:\n{responses}\n\n### Final Answer:"

__init__(proposers: list[Agent], aggregator: Agent, layers: int = 1, parallel: bool = True)

get_aggregate_prompt(user_query: list[dict], responses: dict[str, AgentResponse]) → str

async run_proposers(input_, **proposer_kwargs) → dict[str, AgentResponse]

async run(msg: str | BaseModel | AgentResponse | list[str | BaseModel | dict] | AgentTransfer | VisionInput, proposer_kwargs: dict = {}, **aggregator_kwargs) → AgentResponse

Common Methods

All team types share these methods:

pantheon.team.chat(message: str | dict | None = None)

Start an interactive chat session with the team.

Parameters:

message (str | dict | None) – Optional initial message

pantheon.team.run(msg: AgentInput, **kwargs)

Execute a single task with the team.

Parameters:

msg (AgentInput) – Input message or data

Returns:

Team execution result

Return type:

AgentResponse

pantheon.team.events_queue

Async queue containing team events during execution.

Type:

asyncio.Queue

Team Selection Guide

Team Type	Best For	Example Use Cases
Sequential	Pipeline workflows	Research → Analysis → Report
Swarm	Dynamic routing	Customer support with escalation
SwarmCenter	Task distribution	Project management systems
MoA	Consensus building	Complex decision making

Advanced Examples

Customer Support Team

from pantheon.team import SwarmTeam
from pantheon.agent import Agent

# Create specialized support agents
general = Agent(
    name="general_support",
    instructions="Handle general inquiries, transfer when needed"
)

technical = Agent(
    name="technical_support",
    instructions="Handle technical issues"
)

billing = Agent(
    name="billing_support",
    instructions="Handle billing questions"
)

# Add transfer capabilities
@general.tool
def transfer_to_technical():
    """Transfer to technical support."""
    return technical

@general.tool
def transfer_to_billing():
    """Transfer to billing support."""
    return billing

# Back transfers
@technical.tool
def transfer_to_general():
    """Transfer back to general support."""
    return general

@billing.tool
def transfer_to_general():
    """Transfer back to general support."""
    return general

# Create support team
support_team = SwarmTeam([general, technical, billing])

# Start support session
await support_team.chat("I need help with my account")

Research and Writing Pipeline

from pantheon.team import SequentialTeam

# Create pipeline agents
researcher = Agent(
    name="researcher",
    instructions="Research topics thoroughly using available sources"
)

analyst = Agent(
    name="analyst",
    instructions="Analyze research findings and identify key insights"
)

writer = Agent(
    name="writer",
    instructions="Write clear, engaging content based on analysis"
)

editor = Agent(
    name="editor",
    instructions="Edit for clarity, grammar, and style"
)

# Create pipeline team
pipeline = SequentialTeam([researcher, analyst, writer, editor])

# Execute pipeline
result = await pipeline.run("Create article about quantum computing")

Consensus Building Team

from pantheon.team import MoATeam

# Domain experts
security_expert = Agent(
    name="security_expert",
    instructions="Evaluate from security perspective"
)

performance_expert = Agent(
    name="performance_expert",
    instructions="Evaluate from performance perspective"
)

ux_expert = Agent(
    name="ux_expert",
    instructions="Evaluate from user experience perspective"
)

# Synthesizer
architect = Agent(
    name="architect",
    instructions="Synthesize all perspectives into balanced solution"
)

# Create consensus team
design_team = MoATeam([
    security_expert,
    performance_expert,
    ux_expert,
    architect
])

# Get consensus on design
result = await design_team.run(
    "Design a new authentication system"
)