SwarmCenter Team#
SwarmCenterTeam implements a hub-and-spoke pattern where a central triage agent routes tasks to specialized worker agents. Transfer tools are automatically generated.
Overview#
Key characteristics:
Central Triage Agent: One agent manages all task routing
Automatic Transfer Tools:
transfer_to_{agent_name}generated for each workerAutomatic Return Tools: Each worker gets
transfer_back_to_triageDynamic Agent Management: Add or remove workers at runtime
Basic Usage#
Creating a SwarmCenter Team#
from pantheon.team import SwarmCenterTeam
from pantheon.agent import Agent
# Create triage agent (the central router)
triage = Agent(
name="triage",
instructions="""You are a task router. Analyze requests and delegate to:
- researcher: For information gathering
- developer: For code implementation
- writer: For documentation"""
)
# Create specialized workers
researcher = Agent(
name="researcher",
instructions="Research information and provide findings."
)
developer = Agent(
name="developer",
instructions="Implement code solutions."
)
writer = Agent(
name="writer",
instructions="Create documentation."
)
# Create SwarmCenter team
team = SwarmCenterTeam(
triage=triage,
agents=[researcher, developer, writer]
)
# Run the team
result = await team.run("Create a Python function to parse JSON files")
Constructor Parameters#
Parameter |
Type |
Description |
|---|---|---|
|
Agent |
The central triage agent that routes tasks to workers. |
|
list[Agent | RemoteAgent] |
List of worker agents to receive delegated tasks. |
Auto-Generated Tools#
SwarmCenterTeam automatically generates transfer tools:
For the Triage Agent#
For each worker agent, a transfer tool is created:
# Auto-generated for triage agent:
def transfer_to_researcher():
"""Transfer to researcher."""
return researcher
def transfer_to_developer():
"""Transfer to developer."""
return developer
def transfer_to_writer():
"""Transfer to writer."""
return writer
The function name is derived from the agent’s name (spaces replaced with underscores, lowercased).
For Each Worker Agent#
Each worker automatically gets a tool to transfer back:
# Auto-generated for each worker:
def transfer_back_to_triage():
"""Transfer back to the triage agent."""
return triage
How It Works#
User Message
|
v
[Triage Agent]
|
+---> Analyzes request
|
+---> Calls transfer_to_{worker}()
| |
| v
| [Worker Agent]
| |
| +---> Processes task
| |
| +---> Calls transfer_back_to_triage()
| |
| v
+<----------------[Triage Agent]
|
+---> Returns final response
Dynamic Agent Management#
Add Agent at Runtime#
# Create initial team
team = SwarmCenterTeam(triage=triage, agents=[researcher])
# Add new agent later
new_agent = Agent(name="analyst", instructions="Analyze data.")
await team.add_agent(new_agent)
# Triage now has transfer_to_analyst() tool
# Analyst has transfer_back_to_triage() tool
Remove Agent at Runtime#
# Remove agent from team
await team.remove_agent(analyst)
# Triage no longer has transfer_to_analyst() tool
Run Method#
result = await team.run(
msg="Your task message",
**kwargs # Additional kwargs passed to agents
)
The run() method:
Calls
async_setup()to initialize transfer toolsDelegates to
SwarmTeam.run()for execution
Examples#
Development Team#
from pantheon.team import SwarmCenterTeam
from pantheon.agent import Agent
from pantheon.toolsets import FileManagerToolSet, PythonInterpreterToolSet
# Tech lead as triage
tech_lead = Agent(
name="tech_lead",
instructions="""Route development tasks:
- architect: System design and planning
- developer: Code implementation
- tester: Testing and QA"""
)
architect = Agent(
name="architect",
instructions="Design system architecture."
)
developer = Agent(
name="developer",
instructions="Implement code based on specifications."
)
await developer.toolset(FileManagerToolSet("files"))
await developer.toolset(PythonInterpreterToolSet("python"))
tester = Agent(
name="tester",
instructions="Write and run tests."
)
dev_team = SwarmCenterTeam(
triage=tech_lead,
agents=[architect, developer, tester]
)
result = await dev_team.run("Build a REST API for user management")
Research Team#
from pantheon.toolsets import WebToolSet
coordinator = Agent(
name="coordinator",
instructions="""Coordinate research:
- literature_reviewer: Academic paper analysis
- data_analyst: Data processing
- writer: Report writing"""
)
literature_reviewer = Agent(
name="literature_reviewer",
instructions="Review and summarize academic papers."
)
await literature_reviewer.toolset(WebToolSet("web"))
data_analyst = Agent(
name="data_analyst",
instructions="Analyze data and create visualizations."
)
writer = Agent(
name="writer",
instructions="Write comprehensive research reports."
)
research_team = SwarmCenterTeam(
triage=coordinator,
agents=[literature_reviewer, data_analyst, writer]
)
With Remote Agents#
SwarmCenterTeam supports RemoteAgent workers:
from pantheon.agent import RemoteAgent
# Remote agent fetches its info automatically
remote_specialist = RemoteAgent(service_id="specialist-service")
team = SwarmCenterTeam(
triage=triage,
agents=[local_agent, remote_specialist]
)
# await team.add_agent(remote_specialist) will call fetch_info()
Best Practices#
Clear Triage Instructions: Define routing rules in triage agent’s instructions
Distinct Worker Roles: Each worker should have specialized expertise
Descriptive Agent Names: Names become part of transfer function names
Return to Triage: Workers should transfer back when done
Appropriate Team Size: Keep to 3-6 workers for effective routing
SwarmTeam vs SwarmCenterTeam#
Feature |
SwarmTeam |
SwarmCenterTeam |
|---|---|---|
Transfer Tools |
Manual definition |
Auto-generated |
Structure |
Any-to-any transfers |
Hub-and-spoke pattern |
Central Agent |
Optional |
Required (triage) |
Return Pattern |
Custom |
Auto |
Use Case |
Flexible workflows |
Centralized task routing |