YouTipMulti Agent System
This chapter introduces the Multi-Agent System.
A Multi-Agent System completes complex tasks through the collaboration of multiple Agents.
This is a key technology for building large-scale AI systems, enabling complex functions that a single Agent cannot accomplish.
* * *
## Multi-Agent Architecture Basics
The core challenges of multi-agent systems include three aspects.
Division of labor: How to reasonably distribute tasks among different Agents.
Communication: How Agents pass information to each other.
Coordination: How to ensure that multiple Agents act consistently and efficiently.
### Basic Architecture Patterns
#### Hierarchical Architecture
Hierarchical architecture adopts a tree structure with a main Agent responsible for scheduling.
The Orchestrator is responsible for task decomposition and result integration.
Subagents are responsible for executing specific tasks.
Orchestrator (Scheduler) | +-- Agent A (Expert A) +-- Agent B (Expert B) +-- Agent C (Expert C)
#### Peer-to-Peer Architecture
In peer-to-peer architecture, all Agents have equal status and communicate directly.
Suitable for scenarios where Agents need frequent peer-to-peer interaction.
Agent A <--> Agent B | | v v Agent C <--> Agent D
* * *
## Role Division and Collaboration
An effective multi-agent system requires clear role definitions.
Each Agent should have its own expertise and scope of responsibility.
Collaboration mechanisms ensure that multiple Agents can coordinate their work.
### Common Role Definitions
Planner: Decomposes complex tasks and formulates execution plans.
Executor: Calls tools and executes specific operations.
Reviewer: Evaluates results and proposes improvement suggestions.
Coordinator: Manages communication and task allocation among Agents.
### Code Implementation
## Multi-Agent System Basic Implementation
class Agent:
"""
Basic Agent class
Each Agent has its own role, tools, and execution logic
"""
def __init__ (self, role, llm, tools):
# Agent's role name
self.role= role
# Associated language model
self.llm= llm
# Available tools list
self.tools= tools
def execute(self, task, context=None):
"""
Execute task
:param task: Task description
:param context: Context information
:return: Execution result
"""
# Build prompt
prompt =self.build_prompt(task, context)
# Generate response
response =self.llm.generate(prompt)
# If tool call is needed
if response.needs_tool_call:
return self.execute_tool(response)
return response.content
def build_prompt(self, task, context):
"""Build Agent's prompt"""
return f"Role: {self.role}\n Task: {task}\n Context: {context}"
def execute_tool(self, response):
"""Execute tool call"""
tool_name = response.tool_name
tool_input = response.tool_input
# Find corresponding tool from tools list
tool =self.get_tool(tool_name)
# Execute tool
return tool.execute(tool_input)
class MultiAgentSystem:
"""
Multi-Agent collaboration system
Responsible for managing collaboration among multiple Agents
"""
def __init__ (self, agents, coordinator):
# Agent dictionary, key is role name
self.agents={a.role: a for a in agents}
# Coordinator Agent
self.coordinator= coordinator
def solve(self, task):
"""
Solve complex task
Through coordinating multiple Agents to collaborate
"""
# Step 1: Coordinator analyzes task and decides which Agents to involve
plan =self.coordinator.decompose(task)
# Step 2: Distribute tasks to corresponding Agents
results ={}
for role, subtask in plan.subtasks.items():
# Get Agent of corresponding role
agent =self.agents.get(role)
if agent:
# Execute subtask
results= agent.execute(subtask)
# Step 3: Coordinator integrates results
final_result =self.coordinator.integrate(results)
return final_result
class CoordinatorAgent:
"""Coordinator Agent"""
def __init__ (self, llm):
self.llm= llm
def decompose(self, task):
"""
Decompose task
Analyze task, decide which Agents are needed and how to divide work
"""
prompt = f"""
Analyze the following task, decompose into subtasks and assign to appropriate Agents.
Task: {task}
Please output:
1. Which Agents (roles) are needed
2. What subtask each Agent is responsible for
3. Dependencies between subtasks
Output format: JSON
"""
response =self.llm.generate(prompt)
return Plan.parse(response)
def integrate(self, results):
"""
Integrate results from multiple Agents
"""
prompt = f"""
Integrate the following execution results from Agents and generate final answer.
Result list:
{results}
Please integrate and output final result.
"""
return self.llm.generate(prompt)
* * *
## AutoGen Framework
AutoGen is a multi-agent programming framework developed by Microsoft.
It provides a concise API for building multi-agent conversation systems.
AutoGen greatly simplifies the development complexity of multi-agent systems.
### Core Concepts
AssistantAgent: An intelligent assistant Agent capable of calling tools.
UserProxyAgent: Represents user behavior, can execute code and tool calls.
GroupChat: Supports group chat collaboration among multiple Agents.
GroupChatManager: Manages Agent interactions in group chat.
### Code Example
## AutoGen Basic Usage
# Import autogen framework
import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
# ==================== Configure LLM ====================
# Create a configuration dictionary
llm_config ={
"model": "gpt-4",
"api_key": "your-api-key",# Read from environment variable in actual use
"temperature": 0.7
}
# ==================== Create Single Agent ====================
# Create assistant Agent
assistant = AssistantAgent(
name="assistant",
system_message="""
You are a helpful Python programming assistant.
You can help users write, debug, and optimize code.
""",
llm_config=llm_config
)
# Create user proxy Agent
# human_input_mode="NEVER" means no human input required
# max_consecutive_auto_reply=10 means maximum 10 consecutive auto replies
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)
# ==================== Single Agent Conversation ====================
# Start conversation
user_proxy.initiate_chat(
assistant,
message="Help me write a quick sort algorithm"
)
# ==================== Multi-Agent Group Chat ====================
# Create multiple Agents
coder = AssistantAgent(
name="coder",
system_message="You are a Python programming expert, responsible for writing code.",
llm_config=llm_config
)
reviewer = AssistantAgent(
name="reviewer",
system_message="You are a code review expert, responsible for reviewing code quality.",
llm_config=llm_config
)
# Create group chat
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10# Maximum 10 rounds of conversation
)
# Create group chat manager
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)
# Start group chat
user_proxy.initiate_chat(
manager,
message="Write a quick sort algorithm and review the code"
)
### More Complex AutoGen Example
## AutoGen Tool Calling
import autogen
from autogen import AssistantAgent, UserProxyAgent
# Define tools list
# These tools will be registered in the Agent's tools list
tools =[
{
"name": "calculate",
"description": "Perform mathematical calculation",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Mathematical expression to calculate, e.g. '2+3*5'"
}
},
"required":
}
},
{
"name": "search_web",
"description": "Search web information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search keywords"
}
},
"required":
}
}
]
# Create Agent with tool calling capability
assistant = AssistantAgent(
name="assistant",
system_message="""
You are a helpful assistant.
When calculation or information lookup is needed, you can call relevant tools.
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # Register tools
)
# Create user proxy
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)
# Start conversation, Agent will automatically call tools
user_proxy.initiate_chat(
assistant,
message="Calculate what is (15 + 25) * 2?"
)
# Example response:
# Agent may reply:
# "Let me calculate: (15 + 25) * 2 = 40 * 2 = 80"
# Or directly call calculate tool to get result
* * *
## A2A and MCP Protocols
Multi-agent systems require standardized communication protocols.
A2A and MCP are two important protocol standards.
### A2A (Agent-to-Agent) Protocol
The A2A protocol defines the standard format for communication between Agents.
It supports agent discovery, task collaboration, and state synchronization.
#### Core Functions
Agent Discovery: Agents can discover the capabilities and services of other Agents.
Task Collaboration: Multiple Agents can collaborate to complete complex tasks.
State Synchronization: Agents can synchronize state information with each other.
### MCP (Model Context Protocol)
MCP is an open standard protocol.
Enables AI models to securely connect with external tools and data sources.
MCP is the standard protocol for tool access, similar to the role of a USB interface.
#### Core Components
Host: The host environment running the AI application.
Client: The client that establishes connection with the MCP server.
Server: The server that provides tools and resources.
#### MCP Tool Definition Example
## MCP Tool Definition
{
"name": "filesystem",
"description": "File system operation tool",
"version": "1.0.0",
"tools": [
{
"name": "read_file",
"description": "Read file content",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path, required"
},
"encoding": {
"type": "string",
"description": "File encoding, optional, default is utf-8"
}
},
"required":
}
},
{
"name": "write_file",
"description": "Write file content",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path, required"
},
"content": {
"type": "string",
"description": "File content, required"
}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "List directory contents",
"inputSchema": {
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path"
}
}
}
}
],
"resources": [
{
"uri": "file://config",
"description": "Configuration file"
}
]
}
> Note: A2A and MCP solve different problems. A2A solves the communication problem between Agents, while MCP solves the connection problem between Agents and external tools. The two can be combined to build a complete multi-agent system.
* * *
## Orchestrator-Subagent Pattern
The orchestrator-subagent pattern is a classic multi-agent architecture pattern.
The Orchestrator is responsible for task decomposition and result integration.
Subagents are responsible for executing specific tasks.
### Applicable Scenarios
Scenarios where tasks can be clearly decomposed into independent subtasks.
Scenarios requiring collaboration among Agents with different professional skills.
Complex workflows requiring central coordination and control.
### Code Implementation
## Orchestrator-Subagent Pattern Implementation
class OrchestratorAgent:
"""
Orchestrator Agent (Main Agent)
Responsible for task decomposition, distribution, and result integration
"""
def __init__ (self, llm, subagents):
# Language model
self.llm= llm
# Subagent dictionary
self.subagents={a.role: a for a in subagents}
def handle_request(self, request):
"""
Handle user request
Complete orchestration workflow
"""
# ==================== Analysis Phase ====================
# Analyze request, decide whether multiple Agents need to collaborate
plan =self.llm.plan(request)
if len(plan.subtasks)==1:
# Single task, directly distribute to corresponding Agent
target_role = plan.target_role
return self.subagents[targ