Team Based Artchitectures
Multi-agent Systems
An agent uses an LLM to control application flow. However, as systems grow complex, using a single agent can become challenging. This is where multi-agent systems come in.
Why Use Multiple Agents?
- Simplicity: Break complex tasks into manageable pieces
- Expertise: Create specialized agents for specific tasks
- Better Control: Manage how agents work together
Common Multi-agent Patterns
-
Network Pattern
- Agents can communicate freely with every other agent
- Any agent can decide which other agent to call next
- Flexible but potentially complex to manage
-
Supervisor Pattern
- Central supervisor coordinates other agents
- Clear control flow through supervisor
- Better oversight and management
- Can be implemented through tool-calling
-
Hierarchical Pattern
- Supervisors of supervisors
- Allows for more complex control flows
- Suitable for large-scale systems
-
Custom Workflow Pattern
- Agents communicate with specific subset of agents
- Parts of flow are deterministic
- Limited decision-making about which agents to call next
We will build our agent using a supervisor based multi-agent pattern.
Supervisor-Based Agent Architecture
A team-based architecture typically consists of three main components:
-
Supervisor Agent
- Coordinates team activities
- Makes routing decisions
- Ensures task completion
-
Specialist Agents
- Handle domain-specific tasks
- Maintain focused expertise
- Provide detailed analysis
-
State Management System
- Maintains conversation context
- Tracks team progress
- Manages shared resources
Implementation Pattern
The code implements a team-based architecture using langgraph with three essential components that work together to process complex tasks:
1. Supervisor Agent
def create_supervisor_agent(llm: ChatOpenAI):
"""Creates a supervisor agent that orchestrates the team's activities.
The supervisor agent is the core decision-maker that:
1. Analyzes incoming queries to understand requirements
2. Determines which specialist agent is best suited for each subtask
3. Routes tasks to appropriate specialists
4. Monitors the overall progress of the task
5. Decides when enough information has been gathered
Args:
llm: Language model for decision making
Returns:
A supervisor agent configured with team coordination capabilities
"""
supervisor_prompt = """
Core responsibilities:
1. Analyze incoming queries to break down complex tasks
2. Determine what specific information is needed
3. Select the most appropriate specialist for each subtask
4. Monitor progress and ensure task completion
5. Decide when sufficient information has been gathered
"""
return create_team_supervisor(
llm=llm,
system_prompt=supervisor_prompt,
members=["SpecialistA", "SpecialistB"]
)
2. Specialist Agents
def create_specialist_agent(
llm: ChatOpenAI,
domain: str,
tools: List[Tool]
):
"""Creates a specialist agent with domain-specific expertise.
Specialist agents are focused experts that:
1. Handle specific types of tasks within their domain
2. Use specialized tools for their domain
3. Provide structured analysis and insights
4. Request clarification when needed
Args:
llm: Language model for domain-specific processing
domain: Area of expertise (e.g., "financial analysis", "market research")
tools: List of domain-specific tools available to this specialist
Returns:
A specialist agent configured for its specific domain
"""
system_prompt = f"""
You are specialized in {domain}.
When responding:
1. Use your domain-specific tools effectively
2. Provide clearly structured outputs
3. Explicitly request any missing information needed
"""
return create_agent(
llm=llm,
tools=tools,
system_prompt=system_prompt
)
3. State Management
class TeamState(TypedDict):
"""Manages the shared state and context for the entire team.
Attributes:
messages: List of all messages in the conversation history
team_members: List of available specialist agents
next: Identifier of the next agent to act
information_needed: List of missing information to be gathered
reasoning: Explanation for the current decision or action
"""
messages: List[BaseMessage]
team_members: List[str]
next: str
information_needed: List[str]
reasoning: str
Team Graph Implementation
The team graph orchestrates how all components work together:
def create_team_graph():
"""Creates a coordinated team of agents with defined interaction patterns.
The graph defines:
1. How agents communicate with each other
2. The flow of information between agents
3. Decision points for task routing
4. Conditions for task completion
Process Flow:
1. Supervisor receives task and analyzes requirements
2. Tasks are routed to appropriate specialists
3. Specialists process tasks and return results
4. Supervisor evaluates results and decides next steps
5. Process continues until task is complete
Returns:
A compiled graph ready for task processing
"""
# Initialize team members
specialist_a = create_specialist_agent(...)
specialist_b = create_specialist_agent(...)
supervisor = create_supervisor_agent(...)
# Create the coordination graph
graph = StateGraph(TeamState)
# Define team structure
graph.add_node("SpecialistA", specialist_a)
graph.add_node("SpecialistB", specialist_b)
graph.add_node("supervisor", supervisor)
# Define information flow
graph.add_edge("SpecialistA", "supervisor")
graph.add_edge("SpecialistB", "supervisor")
# Set up decision routing
graph.add_conditional_edges(
"supervisor",
lambda x: x["next"],
{
"SpecialistA": "SpecialistA",
"SpecialistB": "SpecialistB",
"FINISH": END
}
)
return graph.compile()