Multi-Agent System

This chapter introduces the Multi-Agent System.

A Multi-Agent System accomplishes 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 achieve.

Multi-Agent Architecture Basics

The core challenges of a Multi-Agent system include three aspects.

Division of labor: how to reasonably assign tasks to different Agents.

Communication: how information is passed between Agents.

Coordination: how to ensure the actions of multiple Agents are consistent and efficient.

Basic Architecture Patterns

Hierarchical Architecture

The hierarchical architecture adopts a tree structure, with a master Agent responsible for scheduling.

The master Agent (Orchestrator) is responsible for task decomposition and result integration.

The sub Agent (Subagent) is responsible for executing specific tasks.

Orchestrator (调度者)
    |
    +-- Agent A (专家A)
    +-- Agent B (专家B)
    +-- Agent C (专家C)

Peer-to-Peer Architecture

In the peer-to-peer architecture, all Agents are equal and communicate and collaborate 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 responsibilities.

Collaboration mechanisms ensure that multiple Agents can work in coordination.

Common Role Definitions

Planner: Break down complex tasks and formulate execution plans.

Executor: Call tools and perform specific operations.

Reviewer: Evaluate results and provide improvement suggestions.

Coordinator: Manage communication and task allocation among Agents.

Code Implementation

Basic Implementation of a Multi-Agent System

class Agent:
"""
Base 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 the prompt
prompt = self.build_prompt(task, context)

# Generate response
response = self.llm.generate(prompt)

# If a tool call is needed
if response.needs_tool_call:
return self.execute_tool(response)

return response.content

def build_prompt(self, task, context):
"""Build the Agent's prompt"""
return f"Role: {self.role}\nTask: {task}\nContext: {context}"

def execute_tool(self, response):
"""Execute tool call"""
tool_name = response.tool_name
tool_input = response.tool_input

# Find the corresponding tool from the tools list
tool = self.get_tool(tool_name)

# Execute the 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 the role name
self.agents = {a.role: a for a in agents}
# Coordinator Agent
self.coordinator = coordinator

def solve(self, task):
"""
Solve complex tasks
Completed by coordinating multiple Agents to collaborate
"""
# Step 1: The coordinator analyzes the task and determines which Agents need to participate
plan = self.coordinator.decompose(task)

# Step 2: Distribute tasks to the corresponding Agents
results = {}
for role, subtask in plan.subtasks.items():
# Get the Agent with the corresponding role
agent = self.agents.get(role)
if agent:
# Execute the subtask
results[role] = agent.execute(subtask)

# Step 3: The coordinator integrates the 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 the task
Analyze the task, determine which Agents are needed and how to divide the work
"""
prompt = f"""
Analyze the following task, break it down into subtasks, and assign them to appropriate Agents.

Task: {task}

Please output:
1. Which Agents are needed (roles)
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 the results of multiple Agents
"""
prompt = f"""
Integrate the execution results of the following Agents to generate the final answer.

Result list:
{results}

Please integrate and output the 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 dialogue systems.

AutoGen greatly simplifies the development complexity of multi-Agent systems.

Core Concepts

AssistantAgent: An intelligent assistant Agent that can call tools.

UserProxyAgent: Represents user behavior, can execute code and tool calls.

GroupChat: Supports group chat collaboration between multiple Agents.

GroupChatManager: Manages Agent interactions in group chat.

Code Examples

Basic Usage of AutoGen

# Import the 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 variables in actual use
"temperature": 0.7
}

# ==================== Create a single Agent ====================
# Create an 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 a user proxy Agent
# human_input_mode="NEVER" means no manual input is needed
# max_consecutive_auto_reply=10 means up to 10 consecutive automatic replies
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=10
)

# ==================== Single Agent conversation ====================
# Start the conversation
user_proxy.initiate_chat(
assistant,
message="Help me write a quicksort 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 a group chat
group_chat = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10 # Up to 10 conversation rounds
)

# Create a group chat manager
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config
)

# Start the group chat
user_proxy.initiate_chat(
manager,
message="Write a quicksort algorithm and review the code"
)

More Complex AutoGen Example

AutoGen Tool Calling

import autogen
from autogen import AssistantAgent, UserProxyAgent

# Define the tool list
# These tools will be registered in the Agent's tool list
tools = [
{
"name": "calculate",
"description": "Perform mathematical calculations",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "The mathematical expression to calculate, e.g., '2+3*5'"
}
},
"required": ["expression"]
}
},
{
"name": "search_web",
"description": "Search web information",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search keyword"
}
},
"required": ["query"]
}
}
]

# Create an Agent with tool-calling capability
assistant = AssistantAgent(
name="assistant",
system_message="""
You are a helpful assistant.
When you need to calculate or look up information, you can call relevant tools.
""",
llm_config={
"model": "gpt-4",
"api_key": "your-api-key"
},
tools=tools # Register tools
)

# Create a user proxy
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER"
)

# Start the conversation, the Agent will automatically call tools
user_proxy.initiate_chat(
assistant,
message="What is (15 + 25) * 2?"
)

# Example reply:
# The Agent might reply:
# "Let me calculate it: (15 + 25) * 2 = 40 * 2 = 80"
# Or directly call the calculate tool to get the 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 a standard format for communication between Agents.

It supports agent discovery, task collaboration, and state synchronization.

Core features

Agent Discovery: Agents can discover the capabilities and services of other Agents.

Task Collaboration: Multiple Agents can collaboratively complete complex tasks.

State Synchronization: Agents can synchronize state information with each other.

MCP (Model Context Protocol) Protocol

MCP is an open standard protocol.

It enables AI models to securely connect with external tools and data sources.

MCP is a standard protocol for tool integration, similar to the role of a USB interface.

Core Components

Host: The host environment for running AI applications.

Client: The client that establishes connections with MCP servers.

Server: The server that provides tools and resources.

MCP Tool Definition Example

MCP Tool Definition

{
"name": "filesystem",
"description": "File system operations 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": ["path"]
}
},
{
"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 addresses communication between Agents, while MCP addresses the connection between Agents and external tools. The two can be used together to build a complete multi-agent system.

Master-Slave Agent Pattern

Master-subagent mode is a classic multi-agent architecture pattern.

The master Agent (Orchestrator) is responsible for task decomposition and result integration.

The subagent (Subagent) is responsible for executing specific tasks.

Applicable Scenarios

Scenarios where tasks can be clearly decomposed into independent subtasks.

Scenarios requiring Agents with different expertise to collaborate.

Complex workflows requiring central coordination and control.

Code Implementation

Implementation of the Master-Slave Agent Pattern

class OrchestratorAgent:
"""
Main orchestration Agent (Master Agent)
Responsible for task decomposition, assignment, 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):
"""
Process user requests
Complete orchestration flow
"""
# ==================== Analysis Phase ====================
# Analyze the request to determine whether collaboration among multiple Agents is needed
plan = self.llm.plan(request)

if len(plan.subtasks) == 1:
# Single task, directly dispatch to the corresponding Agent
target_role = plan.target_role
return self.subagents[target_role].execute(plan.subtasks[0])

# ==================== Multi-Agent Collaboration Phase ====================
# Create task queue
task_queue = self.create_task_queue(plan.subtasks)

# Store completed results
results = []

# Processing loop until all tasks are complete
while task_queue:
# Get tasks that can be executed in parallel
parallel_tasks = self.get_parallel_tasks(task_queue)

# Execute in parallel
parallel_results = self.execute_in_parallel(parallel_tasks)
results.extend(parallel_results)

# Update task queue
# Handle dependencies, remove completed tasks, add new tasks
task_queue = self.update_queue(task_queue, parallel_results)

# ==================== Integration Phase ====================
# Integrate all results
return self.integrate_results(results)

def create_task_queue(self, subtasks):
"""Create task queue"""
return TaskQueue(subtasks)

def get_parallel_tasks(self, task_queue):
"""
Get tasks that can be executed in parallel
Exclude tasks with dependencies
"""
ready_tasks = []
for task in task_queue.tasks:
# Check whether all task dependencies are satisfied
if all(dep in task_queue.completed for dep in task.dependencies):
ready_tasks.append(task)
return ready_tasks

def execute_in_parallel(self, tasks):
"""Execute multiple tasks in parallel"""
import concurrent.futures

results = []
with concurrent.futures.ThreadPoolExecutor() as executor:
# Submit all tasks
future_to_task = {
executor.submit(self.execute_task, task): task
for task in tasks
}

# Collect results
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append(TaskResult(task=task, error=str(e)))

return results

def execute_task(self, task):
"""Execute a single task"""
agent = self.subagents.get(task.role)
if not agent:
return TaskResult(task=task, error=f"Unknown role: {task.role}")

return agent.execute(task.description)

def update_queue(self, task_queue, completed_results):
"""Update task queue"""
# Mark completed tasks as complete
for result in completed_results:
task_queue.completed.add(result.task.id)

# Remove completed tasks
task_queue.tasks = [
t for t in task_queue.tasks
if t.id not in task_queue.completed
]

return task_queue

def integrate_results(self, results):
"""
Integrate results from multiple Agents
Generate final answer
"""
prompt = f"""
Integrate the execution results of the following Agents and generate a final answer.

Result list:
{results}

Please integrate into a complete, coherent answer.
"""
return self.llm.generate(prompt)

class Subagent:
"""From the Agent base class"""

def __init__(self, role, llm, tools=None):
self.role = role
self.llm = llm
self.tools = tools or []

def execute(self, task):
"""Specific logic for executing the task"""
raise NotImplementedError

Pros and Cons of the Master-Slave Pattern

Advantages	Disadvantages
Clear structure, easy to understand and implement	The master Agent becomes a single point of failure
Centralized coordination, easy to control	The master Agent may become a performance bottleneck
Suitable for well-layered tasks	Low flexibility, difficult to adjust dynamically
Convenient debugging, each subtask can be traced	Not suitable for scenarios with frequent peer-to-peer interaction

Chapter Summary

This chapter introduces the core concepts and implementation methods of multi-agent systems.

Multi-Agent Architecture FundamentalsIntroduced two basic modes: hierarchical architecture and peer architecture.

Role Division and CollaborationImprove system efficiency through clear role definitions.

AutoGen FrameworkProvides a programming framework that simplifies multi-agent development.

A2A and MCP ProtocolsThey are standard protocols for Agent communication and tool integration.

Master-Slave Agent PatternSuitable for complex tasks with a clear hierarchical structure.

Choosing the appropriate multi-agent architecture depends on the specific scenario.

For simple scenarios, a single Agent is sufficient to meet the requirements.

For complex scenarios, it is necessary to choose an appropriate architecture pattern based on the characteristics of the task.

Other extensions

Agent Tools

Python Implementing Intelligent Agents

Multi-Agent System

Multi-Agent Architecture Basics

Basic Architecture Patterns

Hierarchical Architecture

Peer-to-Peer Architecture

Role Division and Collaboration

Common Role Definitions

Code Implementation

Basic Implementation of a Multi-Agent System

AutoGen Framework

Core Concepts

Code Examples

Basic Usage of AutoGen

More Complex AutoGen Example

AutoGen Tool Calling

A2A and MCP Protocols

A2A (Agent-to-Agent) Protocol

Core features

MCP (Model Context Protocol) Protocol

Core Components

MCP Tool Definition Example

MCP Tool Definition

Master-Slave Agent Pattern

Applicable Scenarios

Code Implementation

Implementation of the Master-Slave Agent Pattern

Pros and Cons of the Master-Slave Pattern

Chapter Summary