写在前面
当单个 Agent 的能力无法满足复杂任务需求时，多 Agent 协作成为了必然选择。
想象一下：一个研究助手 Agent 需要同时搜索网页、分析数据、生成报告——这些任务可能需要不同的专业能力。如果所有能力都塞进一个 Agent，不仅会导致"万能但平庸"的问题，还会带来高昂的推理成本。
多 Agent 协作的核心思想是：让专业的人做专业的事。但问题是：如何组织这些 Agent？让谁来指挥谁？Agent 之间如何通信？
本文将深入剖析两种主流的多 Agent 协作模式——Supervisor 模式和去中心化模式——的原理、实现和适用场景，并通过完整的代码示例帮助你构建生产级别的多 Agent 系统。
一、为什么需要多 Agent 协作？
1.1 单 Agent 的局限性
┌─────────────────────────────────────────────────────────────────────┐
│ 单 Agent 的能力边界 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 通用 Agent │ │
│ │ │ │
│ │ 搜索 ──▶ 50% 编程 ──▶ 60% 写作 ──▶ 55% │ │
│ │ 数学 ──▶ 40% 分析 ──▶ 50% 对话 ──▶ 70% │ │
│ │ │ │
│ │ 问题：每个能力都"够用"但不够"专业" │ │
│ │ 推理成本：单 Agent 处理复杂任务，成本高 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘单 Agent 面临的核心问题：
1.2 多 Agent 的优势
┌─────────────────────────────────────────────────────────────────────┐
│ 多 Agent 协作的优势 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 专业化分工 │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ 研究Agent│ │ 编程Agent│ │ 写作Agent│ │ │
│ │ │ 搜索 90% │ │ 编码 95% │ │ 文笔 90% │ │ │
│ │ │ 分析 80% │ │ Debug 90%│ │ 结构 85% │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ │ 每个 Agent 专注于自己的领域，达到专家级水平 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 成本优化 │ │
│ │ │ │
│ │ • 简单任务用小模型，专业任务用大模型 │ │
│ │ • 并行执行减少总等待时间 │ │
│ │ • 按需调用，避免资源浪费 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘1.3 多 Agent 协作的典型场景
二、Supervisor 模式：中央调度器分配任务
2.1 模式概述
┌─────────────────────────────────────────────────────────────────────┐
│ Supervisor 模式架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Supervisor │ │
│ │ (调度器) │ │
│ └──────┬───────┘ │
│ │ │
│ ┌───────────┼───────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ (搜索) │ │ (分析) │ │ (写作) │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ └───────────┴───────────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ 结果 │ │
│ │ 汇总 │ │
│ └───────────┘ │
│ │
│ 特点： │
│ • 单一 Supervisor 负责协调和决策 │
│ • Agent 之间不直接通信，都通过 Supervisor │
│ • 调度逻辑集中，易于理解和调试 │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 Supervisor 模式的核心概念
2.3 Supervisor 模式实现
"""
Supervisor 模式实现
基于 LangGraph 构建中央调度式多 Agent 系统
"""
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
import os
# === 配置 ===
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
# === 定义 Agent 状态 ===
class MultiAgentState(TypedDict):
"""
多 Agent 协作状态
包含所有 Agent 共享的信息
"""
# 用户请求
user_request: str
# 任务分解结果
subtasks: list = []
completed_subtasks: list = []
# 各 Agent 的结果
research_result: str = ""
analysis_result: str = ""
writing_result: str = ""
# 最终输出
final_output: str = ""
# 消息历史
messages: Annotated[list, add_messages]
# === 定义 Agent 工具 ===
class AgentTools:
"""Agent 工具集合"""
@staticmethod
def search_tool(query: str) -> str:
"""搜索工具（模拟）"""
return f"搜索结果：关于「{query}」的最新信息包括...（这是模拟的搜索结果）"
@staticmethod
def analysis_tool(data: str) -> str:
"""分析工具（模拟）"""
return f"分析结果：对「{data[:50]}...」的分析显示...（这是模拟的分析结果）"
@staticmethod
def writing_tool(content: str, style: str = "formal") -> str:
"""写作工具（模拟）"""
return f"根据要求撰写的{style}风格文档：{content[:50]}..."
# === 定义各专业 Agent ===
class ResearchAgent:
"""研究 Agent：负责信息检索"""
def __init__(self, llm: ChatOpenAI):
self.llm = llm
self.name = "ResearchAgent"
def execute(self, task: str) -> dict:
"""
执行研究任务
Args:
task: 研究任务描述
Returns:
研究结果字典
"""
# 调用搜索工具
search_results = AgentTools.search_tool(task)
# 使用 LLM 整理搜索结果
prompt = f"""
请整理以下搜索结果，提取关键信息：
搜索任务：{task}
搜索结果：
{search_results}
请输出结构化的信息摘要。
"""
response = self.llm.invoke(prompt)
return {
"agent": self.name,
"task": task,
"result": response.content,
"status": "completed"
}
class AnalysisAgent:
"""分析 Agent：负责数据分析和推理"""
def __init__(self, llm: ChatOpenAI):
self.llm = llm
self.name = "AnalysisAgent"
def execute(self, data: str, focus: str = "general") -> dict:
"""
执行分析任务
Args:
data: 待分析的数据
focus: 分析重点
Returns:
分析结果字典
"""
# 调用分析工具
analysis_result = AgentTools.analysis_tool(data)
# 使用 LLM 进行深度分析
prompt = f"""
基于以下数据和指定重点进行深度分析：
分析数据：
{data[:1000]}
分析重点：{focus}
请提供：
1. 主要发现
2. 关键洞察
3. 潜在问题或风险
"""
response = self.llm.invoke(prompt)
return {
"agent": self.name,
"focus": focus,
"result": response.content,
"status": "completed"
}
class WritingAgent:
"""写作 Agent：负责内容生成和文档撰写"""
def __init__(self, llm: ChatOpenAI):
self.llm = llm
self.name = "WritingAgent"
def execute(self, content: str, style: str = "formal", format: str = "report") -> dict:
"""
执行写作任务
Args:
content: 写作内容素材
style: 写作风格
format: 输出格式
Returns:
写作结果字典
"""
prompt = f"""
请根据以下素材撰写一篇{style}风格的{format}：
素材内容：
{content}
要求：
1. 结构清晰，逻辑连贯
2. 内容充实，有深度
3. 符合指定的风格
"""
response = self.llm.invoke(prompt)
return {
"agent": self.name,
"style": style,
"format": format,
"result": response.content,
"status": "completed"
}
# === Supervisor 实现 ===
class Supervisor:
"""
Supervisor：中央调度器
职责：
1. 理解用户请求
2. 分解任务为子任务
3. 分配子任务给合适的 Agent
4. 汇总结果并生成最终输出
"""
def __init__(self, llm: ChatOpenAI):
self.llm = llm
self.name = "Supervisor"
# 初始化专业 Agent
self.research_agent = ResearchAgent(llm)
self.analysis_agent = AnalysisAgent(llm)
self.writing_agent = WritingAgent(llm)
def decompose_task(self, user_request: str) -> list[dict]:
"""
任务分解
将用户请求分解为可执行的子任务
Args:
user_request: 用户原始请求
Returns:
子任务列表
"""
prompt = f"""
请分析以下用户请求，将其分解为具体的子任务：
用户请求：
{user_request}
子任务应该包括：
1. 任务类型（research/analysis/writing）
2. 任务描述
3. 任务参数
请以 JSON 数组格式输出。
"""
response = self.llm.invoke(prompt)
# 简单解析（实际应用中应使用更严格的 JSON 解析）
# 这里简化处理，直接返回一个结构
subtasks = [
{
"type": "research",
"description": f"搜索并整理关于「{user_request}」的信息",
"priority": 1
},
{
"type": "analysis",
"description": f"分析收集到的信息，提取关键洞察",
"priority": 2
},
{
"type": "writing",
"description": "撰写最终报告",
"priority": 3
}
]
return subtasks
def assign_task(self, task: dict, state: MultiAgentState) -> dict:
"""
任务分配
将子任务分配给合适的 Agent
Args:
task: 子任务
state: 当前状态
Returns:
执行结果
"""
task_type = task["type"]
if task_type == "research":
return self.research_agent.execute(task["description"])
elif task_type == "analysis":
# 分析需要研究结果作为输入
data = state.get("research_result", "")
return self.analysis_agent.execute(data, task.get("focus", "general"))
elif task_type == "writing":
# 写作需要研究和分析结果
content = f"研究结果：{state.get('research_result', '')}\n分析结果：{state.get('analysis_result', '')}"
return self.writing_agent.execute(content, task.get("style", "formal"), task.get("format", "report"))
else:
return {"error": f"Unknown task type: {task_type}"}
def aggregate_results(self, state: MultiAgentState) -> str:
"""
结果汇总
将各 Agent 的结果汇总为最终输出
Args:
state: 最终状态
Returns:
最终输出
"""
prompt = f"""
请将以下研究、分析和写作结果汇总为最终输出：
用户请求：{state['user_request']}
研究结果：
{state.get('research_result', 'N/A')}
分析结果：
{state.get('analysis_result', 'N/A')}
写作结果：
{state.get('writing_result', 'N/A')}
请输出一份完整、连贯的最终报告。
"""
response = self.llm.invoke(prompt)
return response.content
# === LangGraph 工作流节点 ===
def supervisor_decompose(state: MultiAgentState) -> dict:
"""Supervisor：任务分解节点"""
llm = ChatOpenAI(model="gpt-4", api_key=OPENAI_API_KEY)
supervisor = Supervisor(llm)
subtasks = supervisor.decompose_task(state["user_request"])
return {"subtasks": subtasks}
def research_node(state: MultiAgentState) -> dict:
"""研究 Agent 节点"""
llm = ChatOpenAI(model="gpt-4", api_key=OPENAI_API_KEY)
research_agent = ResearchAgent(llm)
# 获取研究任务
research_task = next((t for t in state["subtasks"] if t["type"] == "research"), None)
if research_task:
result = research_agent.execute(research_task["description"])
return {"research_result": result["result"]}
return {"research_result": ""}
def analysis_node(state: MultiAgentState) -> dict:
"""分析 Agent 节点"""
llm = ChatOpenAI(model="gpt-4", api_key=OPENAI_API_KEY)
analysis_agent = AnalysisAgent(llm)
result = analysis_agent.execute(
data=state.get("research_result", ""),
focus="general"
)
return {"analysis_result": result["result"]}
def writing_node(state: MultiAgentState) -> dict:
"""写作 Agent 节点"""
llm = ChatOpenAI(model="gpt-4", api_key=OPENAI_API_KEY)
writing_agent = WritingAgent(llm)
content = f"研究结果：{state.get('research_result', '')}\n分析结果：{state.get('analysis_result', '')}"
result = writing_agent.execute(content)
return {"writing_result": result["result"]}
def supervisor_aggregate(state: MultiAgentState) -> dict:
"""Supervisor：结果汇总节点"""
llm = ChatOpenAI(model="gpt-4", api_key=OPENAI_API_KEY)
supervisor = Supervisor(llm)
final_output = supervisor.aggregate_results(state)
return {"final_output": final_output}
# === 构建工作流 ===
def build_supervisor_workflow():
"""构建 Supervisor 模式工作流"""
workflow = StateGraph(MultiAgentState)
# 添加节点
workflow.add_node("supervisor_decompose", supervisor_decompose)
workflow.add_node("research", research_node)
workflow.add_node("analysis", analysis_node)
workflow.add_node("writing", writing_node)
workflow.add_node("supervisor_aggregate", supervisor_aggregate)
# 定义边
workflow.add_edge(START, "supervisor_decompose")
workflow.add_edge("supervisor_decompose", "research")
# 分析依赖研究结果
workflow.add_edge("research", "analysis")
# 写作依赖研究和分析结果
workflow.add_edge("analysis", "writing")
# 汇总
workflow.add_edge("writing", "supervisor_aggregate")
workflow.add_edge("supervisor_aggregate", END)
return workflow.compile()
# === 使用示例 ===
def main():
"""主函数"""
print("🚀 Supervisor 模式多 Agent 系统")
print("=" * 50)
# 构建工作流
app = build_supervisor_workflow()
# 定义初始状态
initial_state = {
"user_request": "请分析人工智能对软件工程行业的影响，包括就业、技能需求和未来趋势",
"subtasks": [],
"completed_subtasks": [],
"research_result": "",
"analysis_result": "",
"writing_result": "",
"final_output": "",
"messages": []
}
# 执行工作流
print("\n📋 开始执行任务...\n")
result = app.invoke(initial_state)
# 输出结果
print("\n" + "=" * 50)
print("📄 最终报告")
print("=" * 50)
print(result["final_output"])
print("\n" + "=" * 50)
print("📊 执行统计")
print("=" * 50)
print(f"分解任务数: {len(result['subtasks'])}")
print(f"完成任务数: {len(result['completed_subtasks'])}")
if __name__ == "__main__":
main()三、去中心化模式：Agent 之间自主协商
3.1 模式概述
┌─────────────────────────────────────────────────────────────────────┐
│ 去中心化模式架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ Agent A │◀───────────┐ │
│ └───┬─────┘ │ │
│ │ │ │
│ │ ┌──────────────┼──────────────┐ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Agent B │◀─────▶│ Agent C │◀─────▶│ Agent D │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │
│ │ ┌────────────────────────────────┐ │
│ └────▶│ 共享状态 │ │
│ │ (黑板 / 消息队列 / KV存储) │ │
│ └────────────────────────────────┘ │
│ │
│ 特点： │
│ • 无中央控制器，Agent 自主决策 │
│ • 通过共享状态或消息传递进行通信 │
│ • 高度可扩展，适合大规模系统 │
│ • 复杂行为从简单规则的交互中涌现 │
│ │
└─────────────────────────────────────────────────────────────────────┘3.2 去中心化模式的核心概念
3.3 去中心化模式实现
"""
去中心化模式实现
基于消息传递的多 Agent 协作系统
"""
from typing import TypedDict, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import uuid
from datetime import datetime
import asyncio
# === 消息定义 ===
class MessageType(Enum):
"""消息类型"""
REQUEST = "request" # 请求消息
RESPONSE = "response" # 响应消息
BROADCAST = "broadcast" # 广播消息
NEGOTIATION = "negotiation" # 协商消息
@dataclass
class Message:
"""Agent 间消息"""
id: str = field(default_factory=lambda: str(uuid.uuid4()))
sender: str = "" # 发送者 ID
receivers: list[str] = [] # 接收者列表（空表示广播）
type: MessageType = MessageType.REQUEST
content: dict = field(default_factory=dict)
timestamp: float = field(default_factory=datetime.now().timestamp)
reply_to: Optional[str] = None # 关联的消息 ID
@dataclass
class SharedState:
"""共享状态"""
data: dict = field(default_factory=dict)
version: int = 0
def get(self, key: str, default=None):
return self.data.get(key, default)
def set(self, key: str, value):
self.data[key] = value
self.version += 1
def update(self, updates: dict):
self.data.update(updates)
self.version += 1
# === Agent 基类 ===
class DecentralizedAgent:
"""
去中心化 Agent 基类
特点：
1. 既是请求者也是服务者
2. 通过消息队列通信
3. 维护本地状态和共享状态视图
"""
def __init__(
self,
agent_id: str,
capabilities: list[str],
message_queue: list[Message],
shared_state: SharedState
):
self.agent_id = agent_id
self.capabilities = capabilities # Agent 能提供的服务
self.message_queue = message_queue # 消息队列
self.shared_state = shared_state # 共享状态
self.local_memory: dict = {} # Agent 本地记忆
def send_message(
self,
receivers: list[str],
msg_type: MessageType,
content: dict,
reply_to: Optional[str] = None
) -> Message:
"""
发送消息
Args:
receivers: 接收者列表，空表示广播
msg_type: 消息类型
content: 消息内容
reply_to: 回复关联的消息 ID
Returns:
发送的消息
"""
message = Message(
sender=self.agent_id,
receivers=receivers,
type=msg_type,
content=content,
reply_to=reply_to
)
self.message_queue.append(message)
return message
def receive_messages(self, filter_func: Optional[Callable] = None) -> list[Message]:
"""
接收消息
Args:
filter_func: 消息过滤函数
Returns:
匹配的消息列表
"""
my_messages = [
msg for msg in self.message_queue
if (msg.receivers == [] or self.agent_id in msg.receivers)
and msg.sender != self.agent_id
]
if filter_func:
my_messages = [msg for msg in my_messages if filter_func(msg)]
return my_messages
def can_handle(self, request: dict) -> bool:
"""检查是否能处理请求"""
required_capability = request.get("required_capability")
return required_capability in self.capabilities if required_capability else True
def execute_capability(self, capability: str, params: dict) -> dict:
"""
执行能力（子类实现）
Args:
capability: 能力名称
params: 参数
Returns:
执行结果
"""
raise NotImplementedError
# === 具体 Agent 实现 ===
class CodeReviewAgent(DecentralizedAgent):
"""代码审查 Agent"""
def __init__(self, message_queue: list[Message], shared_state: SharedState):
super().__init__(
agent_id="code_review",
capabilities=["code_review", "bug_detection", "security_scan"],
message_queue=message_queue,
shared_state=shared_state
)
def execute_capability(self, capability: str, params: dict) -> dict:
"""执行代码审查能力"""
if capability == "code_review":
code = params.get("code", "")
return {
"status": "completed",
"findings": [
f"发现代码质量问题：{code[:50]}...",
"建议：添加文档注释",
"建议：提取公共函数"
],
"score": 7.5
}
elif capability == "bug_detection":
return {"status": "completed", "bugs": [], "risk_level": "low"}
else:
return {"status": "error", "message": f"Unknown capability: {capability}"}
class TestingAgent(DecentralizedAgent):
"""测试 Agent"""
def __init__(self, message_queue: list[Message], shared_state: SharedState):
super().__init__(
agent_id="testing",
capabilities=["unit_test", "integration_test", "coverage_analysis"],
message_queue=message_queue,
shared_state=shared_state
)
def execute_capability(self, capability: str, params: dict) -> dict:
"""执行测试能力"""
if capability == "unit_test":
code = params.get("code", "")
return {
"status": "completed",
"tests_generated": 10,
"coverage": 85.0,
"failed_tests": []
}
elif capability == "integration_test":
return {"status": "completed", "integration_points": 5, "all_passed": True}
else:
return {"status": "error", "message": f"Unknown capability: {capability}"}
class DocumentationAgent(DecentralizedAgent):
"""文档 Agent"""
def __init__(self, message_queue: list[Message], shared_state: SharedState):
super().__init__(
agent_id="documentation",
capabilities=["api_doc", "readme", "changelog"],
message_queue=message_queue,
shared_state=shared_state
)
def execute_capability(self, capability: str, params: dict) -> dict:
"""执行文档生成能力"""
if capability == "api_doc":
code = params.get("code", "")
return {
"status": "completed",
"documentation": f"API 文档已生成，包含 {code.count('def')} 个函数"
}
else:
return {"status": "error", "message": f"Unknown capability: {capability}"}
# === 协调器 ===
class DecentralizedCoordinator:
"""
去中心化协调器
负责：
1. 管理 Agent 注册
2. 消息路由
3. 冲突解决
"""
def __init__(self):
self.agents: dict[str, DecentralizedAgent] = {}
self.message_queue: list[Message] = []
self.shared_state = SharedState()
def register_agent(self, agent: DecentralizedAgent):
"""注册 Agent"""
self.agents[agent.agent_id] = agent
print(f"✅ Agent 注册: {agent.agent_id}")
print(f" 能力: {', '.join(agent.capabilities)}")
def find_agents(self, capability: str) -> list[DecentralizedAgent]:
"""查找具有特定能力的 Agent"""
return [
agent for agent in self.agents.values()
if capability in agent.capabilities
]
def route_request(self, requester: str, request: dict) -> dict:
"""
路由请求到合适的 Agent
使用简单的拍卖机制：多个 Agent 竞争执行
"""
capability = request.get("required_capability")
suitable_agents = self.find_agents(capability)
if not suitable_agents:
return {"status": "error", "message": f"No agent can handle {capability}"}
# 简单的路由策略：选择第一个合适的 Agent
# 实际应用中可以使用更复杂的策略（负载均衡、拍卖等）
selected_agent = suitable_agents[0]
# 发送请求消息
message = selected_agent.send_message(
receivers=[selected_agent.agent_id],
msg_type=MessageType.REQUEST,
content=request
)
# Agent 处理请求
result = selected_agent.execute_capability(capability, request.get("params", {}))
# 更新共享状态
self.shared_state.set(
f"{capability}_result",
{"agent": selected_agent.agent_id, "result": result}
)
return {
"status": "success",
"agent": selected_agent.agent_id,
"result": result
}
def broadcast(self, sender: str, content: dict):
"""
广播消息
Args:
sender: 发送者 ID
content: 消息内容
"""
message = Message(
sender=sender,
receivers=[], # 空表示广播
type=MessageType.BROADCAST,
content=content
)
self.message_queue.append(message)
def run_negotiation(self, topic: str, participants: list[str]) -> dict:
"""
运行协商协议
多 Agent 就某个话题达成共识
"""
# 收集各 Agent 的意见
opinions = []
for agent_id in participants:
agent = self.agents.get(agent_id)
if agent:
# Agent 表达意见
opinion = f"{agent_id} 的观点..."
opinions.append({"agent": agent_id, "opinion": opinion})
# 汇总意见，形成共识
consensus = {
"topic": topic,
"opinions": opinions,
"decision": "综合各方意见的最终决策",
"agreed_by": participants
}
# 广播共识
self.broadcast("coordinator", consensus)
return consensus
# === 使用示例 ===
def decentralized_demo():
"""去中心化模式演示"""
print("🌐 去中心化多 Agent 系统")
print("=" * 50)
# 创建协调器
coordinator = DecentralizedCoordinator()
# 创建和注册 Agent
code_review_agent = CodeReviewAgent(coordinator.message_queue, coordinator.shared_state)
testing_agent = TestingAgent(coordinator.message_queue, coordinator.shared_state)
documentation_agent = DocumentationAgent(coordinator.message_queue, coordinator.shared_state)
coordinator.register_agent(code_review_agent)
coordinator.register_agent(testing_agent)
coordinator.register_agent(documentation_agent)
print("\n" + "-" * 50)
print("📋 执行代码审查任务")
print("-" * 50)
# 路由请求
request = {
"required_capability": "code_review",
"params": {
"code": "def hello(): print('world')"
}
}
result = coordinator.route_request("user", request)
print(f"\n执行结果:")
print(f" 处理 Agent: {result['agent']}")
print(f" 状态: {result['result']['status']}")
print(f" 评分: {result['result'].get('score', 'N/A')}")
print(f" 发现: {result['result'].get('findings', [])[:2]}")
print("\n" + "-" * 50)
print("📊 共享状态")
print("-" * 50)
print(f" 版本: {coordinator.shared_state.version}")
print(f" 数据: {coordinator.shared_state.data}")
if __name__ == "__main__":
decentralized_demo()四、混合模式：层级式协作
4.1 模式概述
┌─────────────────────────────────────────────────────────────────────┐
│ 混合模式架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────┐ │
│ │ 顶层 │ │
│ │ Supervisor│ │
│ └─────┬─────┘ │
│ │ │
│ ┌─────────────────────┼─────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 子Supervisor│ │ 子Supervisor│ │ 子Supervisor│ │
│ │ (研究组) │ │ (开发组) │ │ (测试组) │ │
│ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │
│ │ │ │ │
│ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐ │
│ │ Agent A │ │ Agent D │ │ Agent G │ │
│ │ Agent B │ │ Agent E │ │ Agent H │ │
│ │ Agent C │ │ Agent F │ │ Agent I │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ 特点： │
│ • 层级结构：顶层 Supervisor + 多个子 Supervisor │
│ • 每个子 Supervisor 管理一组同类型 Agent │
│ • 结合了 Supervisor 的可控性和去中心化的灵活性 │
│ │
└─────────────────────────────────────────────────────────────────────┘4.2 混合模式实现
"""
混合模式实现
层级式多 Agent 协作系统
"""
from typing import TypedDict, Optional
from dataclasses import dataclass
@dataclass
class Task:
"""任务定义"""
id: str
type: str # research/development/testing
description: str
priority: int = 1
status: str = "pending" # pending/in_progress/completed/failed
assignee: Optional[str] = None
dependencies: list[str] = [] # 依赖的任务 ID
class HybridMultiAgentSystem:
"""
混合模式多 Agent 系统
结合 Supervisor 模式（顶层调度）和去中心化模式（团队内部协作）
"""
def __init__(self):
# Agent 池
self.agents: dict[str, dict] = {
"researcher_1": {"type": "research", "status": "idle", "capabilities": ["search", "analysis"]},
"researcher_2": {"type": "research", "status": "idle", "capabilities": ["search", "writing"]},
"developer_1": {"type": "development", "status": "idle", "capabilities": ["coding", "refactor"]},
"developer_2": {"type": "development", "status": "idle", "capabilities": ["coding", "debug"]},
"tester_1": {"type": "testing", "status": "idle", "capabilities": ["unit_test", "integration_test"]},
"tester_2": {"type": "testing", "status": "idle", "capabilities": ["security_test", "performance_test"]},
}
# 子 Supervisor（团队组长）
self.supervisors: dict[str, dict] = {
"research_lead": {
"team": ["researcher_1", "researcher_2"],
"tasks": []
},
"development_lead": {
"team": ["developer_1", "developer_2"],
"tasks": []
},
"testing_lead": {
"team": ["tester_1", "tester_2"],
"tasks": []
}
}
# 任务队列
self.task_queue: list[Task] = []
# 完成的任务
self.completed_tasks: list[Task] = []
def add_task(self, task: Task):
"""添加任务到队列"""
self.task_queue.append(task)
def top_supervisor_dispatch(self, task: Task) -> dict:
"""
顶层 Supervisor 分配任务
决定任务应该分配给哪个团队
"""
# 根据任务类型分配到对应的子 Supervisor
team_mapping = {
"research": "research_lead",
"development": "development_lead",
"testing": "testing_lead"
}
team_lead = team_mapping.get(task.type)
if team_lead:
self.supervisors[team_lead]["tasks"].append(task)
return {
"status": "dispatched",
"team": team_lead,
"task_id": task.id
}
return {"status": "error", "message": f"Unknown task type: {task.type}"}
def team_supervisor_assign(self, team_lead: str, task: Task) -> dict:
"""
子 Supervisor 分配任务给团队成员
使用简单的负载均衡策略
"""
team = self.supervisors[team_lead]["team"]
# 查找最空闲的 Agent
min_load = float('inf')
selected_agent = None
for agent_id in team:
if self.agents[agent_id]["status"] == "idle":
load = sum(
1 for t in self.task_queue + self._get_team_tasks(team_lead)
if t.assignee == agent_id and t.status == "in_progress"
)
if load < min_load:
min_load = load
selected_agent = agent_id
if selected_agent:
task.assignee = selected_agent
task.status = "in_progress"
self.agents[selected_agent]["status"] = "busy"
return {
"status": "assigned",
"agent": selected_agent,
"task_id": task.id
}
return {"status": "queued", "message": "No available agent"}
def _get_team_tasks(self, team_lead: str) -> list[Task]:
"""获取团队的任务列表"""
return self.supervisors[team_lead]["tasks"]
def agent_complete_task(self, agent_id: str, task_id: str, result: dict):
"""
Agent 完成任务
更新状态并检查依赖任务
"""
# 找到并更新任务
for task in self.task_queue:
if task.id == task_id:
task.status = "completed"
self.completed_tasks.append(task)
break
# 更新 Agent 状态
self.agents[agent_id]["status"] = "idle"
# 检查是否有依赖此任务的任务可以开始
self._check_dependent_tasks(task_id)
def _check_dependent_tasks(self, completed_task_id: str):
"""检查依赖任务是否可以开始"""
for task in self.task_queue:
if task.status == "pending" and completed_task_id in task.dependencies:
# 检查所有依赖是否都已完成
all_deps_complete = all(
t.status == "completed" for t in self.task_queue
if t.id in task.dependencies
)
if all_deps_complete:
# 重新分配任务
self.top_supervisor_dispatch(task)
def get_team_status(self) -> dict:
"""获取团队状态"""
status = {}
for agent_id, agent_info in self.agents.items():
team_lead = None
for lead, info in self.supervisors.items():
if agent_id in info["team"]:
team_lead = lead
break
status[agent_id] = {
**agent_info,
"team_lead": team_lead
}
return status
# === 使用示例 ===
def hybrid_demo():
"""混合模式演示"""
print("🏗️ 混合模式多 Agent 系统")
print("=" * 50)
system = HybridMultiAgentSystem()
# 添加任务
tasks = [
Task(id="t1", type="research", description="研究 AI 趋势", priority=1),
Task(id="t2", type="development", description="开发新功能", priority=2),
Task(id="t3", type="testing", description="测试新功能", priority=2, dependencies=["t2"]),
]
for task in tasks:
system.add_task(task)
print("\n📋 任务分配过程")
print("-" * 50)
# 顶层 Supervisor 分配
for task in tasks:
result = system.top_supervisor_dispatch(task)
print(f"任务 {task.id} ({task.type}): 分配到 {result['team']}")
# 子 Supervisor 分配
team_result = system.team_supervisor_assign(result['team'], task)
print(f" └─> 由 {team_result.get('agent', team_result.get('message'))} 执行")
print("\n" + "-" * 50)
print("👥 Agent 状态")
print("-" * 50)
status = system.get_team_status()
for agent_id, info in status.items():
print(f"{agent_id}: {info['status']} ({info['team_lead']})")
if __name__ == "__main__":
hybrid_demo()五、通信机制：消息传递 vs 共享状态
5.1 两种通信机制对比
┌─────────────────────────────────────────────────────────────────────┐
│ 通信机制对比 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────┐ ┌─────────────────────┐ │
│ │ 消息传递模式 │ │ 共享状态模式 │ │
│ ├─────────────────────┤ ├─────────────────────┤ │
│ │ │ │ │ │
│ │ Agent A ──MSG──▶ Agent B │ ┌─────────────┐ │ │
│ │ │ │ │ 共享状态 │ │ │
│ │ 特点： │ │ │ (Blackboard) │ │ │
│ │ • 点对点通信 │ │ └──────┬──────┘ │ │
│ │ • 解耦发送方和接收方│ │ │ │ │
│ │ • 异步处理 │ │ ┌────┴────┐ │ │
│ │ • 消息队列 │ │ ▼ ▼ ▼ │ │
│ │ │ │ A B C │ │
│ └─────────────────────┘ │ Agent们都访问 │ │
│ │ 同一状态空间 │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 直接读写共享数据 │ │
│ │ • 简单直接 │ │
│ │ • 需要同步机制 │ │
│ │ • 一致性挑战 │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘5.2 通信机制实现
"""
通信机制实现
支持消息传递和共享状态两种模式
"""
from typing import Callable, Optional
from collections import defaultdict
import threading
import time
# === 消息队列实现 ===
class MessageQueue:
"""
线程安全的异步消息队列
支持：
1. 点对点消息
2. 广播消息
3. 消息订阅
"""
def __init__(self):
self._queue: list[Message] = []
self._lock = threading.Lock()
self._subscribers: dict[str, list[Callable]] = defaultdict(list)
def publish(self, message: Message):
"""发布消息到队列"""
with self._lock:
self._queue.append(message)
# 通知订阅者
for callback in self._subscribers.get(message.type.value, []):
callback(message)
# 广播给所有订阅者
for callback in self._subscribers.get("*", []):
callback(message)
def subscribe(self, message_type: str, callback: Callable):
"""订阅特定类型的消息"""
self._subscribers[message_type].append(callback)
def get_messages(
self,
receiver: Optional[str] = None,
message_type: Optional[MessageType] = None
) -> list[Message]:
"""获取消息"""
with self._lock:
messages = self._queue.copy()
if receiver:
messages = [
m for m in messages
if m.receivers == [] or receiver in m.receivers
]
if message_type:
messages = [m for m in messages if m.type == message_type]
# 清除已获取的消息
with self._lock:
self._queue = [
m for m in self._queue
if m not in messages
]
return messages
# === 共享状态实现 ===
class ThreadSafeSharedState:
"""
线程安全的共享状态
支持：
1. 读写锁
2. 乐观锁（版本控制）
3. 观察者模式
"""
def __init__(self):
self._data: dict = {}
self._version: int = 0
self._lock = threading.RLock()
self._watchers: list[Callable] = []
def get(self, key: str, default=None):
"""读取值"""
with self._lock:
return self._data.get(key, default)
def set(self, key: str, value):
"""写入值"""
with self._lock:
old_value = self._data.get(key)
self._data[key] = value
self._version += 1
# 通知观察者
for watcher in self._watchers:
watcher(key, old_value, value)
def update(self, updates: dict, expected_version: Optional[int] = None) -> bool:
"""
原子性更新（乐观锁）
Args:
updates: 要更新的键值对
expected_version: 期望的版本号，如果提供则检查版本匹配
Returns:
是否更新成功
"""
with self._lock:
if expected_version is not None and self._version != expected_version:
return False # 版本冲突
self._data.update(updates)
self._version += 1
return True
def watch(self, callback: Callable):
"""注册观察者"""
self._watchers.append(callback)
def get_version(self) -> int:
"""获取当前版本"""
with self._lock:
return self._version
# === 混合通信实现 ===
class HybridCommunication:
"""
混合通信系统
同时支持消息传递和共享状态
"""
def __init__(self):
self.message_queue = MessageQueue()
self.shared_state = ThreadSafeSharedState()
def send_message(
self,
sender: str,
receivers: list[str],
msg_type: MessageType,
content: dict
):
"""发送消息"""
message = Message(
sender=sender,
receivers=receivers,
type=msg_type,
content=content
)
self.message_queue.publish(message)
def write_shared_state(self, agent_id: str, key: str, value):
"""写入共享状态"""
self.shared_state.set(key, {"value": value, "updated_by": agent_id})
def read_shared_state(self, key: str):
"""读取共享状态"""
return self.shared_state.get(key)
def broadcast_update(self, agent_id: str, updates: dict):
"""
广播状态更新
1. 更新共享状态
2. 发送广播消息通知其他 Agent
"""
for key, value in updates.items():
self.write_shared_state(agent_id, key, value)
self.send_message(
sender=agent_id,
receivers=[], # 广播
msg_type=MessageType.BROADCAST,
content={"updates": updates}
)
# === 使用示例 ===
def communication_demo():
"""通信机制演示"""
print("📨 混合通信机制演示")
print("=" * 50)
comm = HybridCommunication()
# 模拟 Agent A 发送消息
print("\n📤 Agent A 发送消息给 Agent B")
comm.send_message(
sender="Agent A",
receivers=["Agent B"],
msg_type=MessageType.REQUEST,
content={"task": "代码审查", "code": "def foo(): pass"}
)
# 模拟 Agent B 接收消息
messages = comm.message_queue.get_messages(receiver="Agent B")
for msg in messages:
print(f" 收到消息: {msg.content}")
# 模拟共享状态写入
print("\n📝 Agent A 更新共享状态")
comm.write_shared_state("Agent A", "current_task", "代码审查中")
comm.write_shared_state("Agent A", "progress", 50)
print(f" 当前任务: {comm.read_shared_state('current_task')}")
print(f" 进度: {comm.read_shared_state('progress')}%")
# 模拟广播
print("\n📢 Agent A 广播状态更新")
comm.broadcast_update("Agent A", {"status": "忙碌", "current_task": "代码审查"})
print(" 广播已发送")
if __name__ == "__main__":
communication_demo()六、任务分解与合并策略
6.1 任务分解策略
┌─────────────────────────────────────────────────────────────────────┐
│ 任务分解策略 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 输入：复杂任务 │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 分解引擎 │ │
│ ├─────────────────────────────────────────────────────────────┤ │
│ │ │ │
│ │ 策略1：功能分解 │ │
│ │ ├── 搜索子任务 ──▶ Agent: Research │ │
│ │ ├── 分析子任务 ──▶ Agent: Analysis │ │
│ │ └── 写作子任务 ──▶ Agent: Writing │ │
│ │ │ │
│ │ 策略2：数据分解 │ │
│ │ ├── 数据集A ──▶ Agent 1 │ │
│ │ ├── 数据集B ──▶ Agent 2 │ │
│ │ └── 数据集C ──▶ Agent 3 │ │
│ │ │ │
│ │ 策略3：时间分解 │ │
│ │ ├── 阶段1 ──▶ Agent (初步分析) │ │
│ │ ├── 阶段2 ──▶ Agent (深度分析) │ │
│ │ └── 阶段3 ──▶ Agent (最终总结) │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘6.2 任务合并策略
"""
任务分解与合并策略
"""
from typing import TypedDict, Optional
from dataclasses import dataclass
from enum import Enum
class MergeStrategy(Enum):
"""合并策略"""
CONCATENATE = "concatenate" # 简单拼接
SUMMARIZE = "summarize" # 摘要合并
CROSS_REFERENCE = "cross_reference" # 交叉引用
HIERARCHICAL = "hierarchical" # 层级合并
@dataclass
class SubTask:
"""子任务"""
id: str
description: str
agent_type: str
priority: int
dependencies: list[str] = None
result: Optional[str] = None
class TaskDecomposer:
"""
任务分解器
将复杂任务分解为可执行的子任务
"""
def __init__(self, llm=None):
self.llm = llm
def decompose_by_function(self, task: str) -> list[SubTask]:
"""
按功能分解任务
适用于：研究类、内容生成类任务
"""
subtasks = []
# 典型的功能分解模板
templates = {
"research": [
SubTask("search", "信息搜索与收集", "research", 1),
SubTask("analyze", "信息分析与整理", "analysis", 2, ["search"]),
SubTask("synthesize", "综合与总结", "writing", 3, ["analyze"]),
],
"development": [
SubTask("design", "设计架构与方案", "architect", 1),
SubTask("implement", "代码实现", "developer", 2, ["design"]),
SubTask("test", "测试验证", "tester", 3, ["implement"]),
SubTask("deploy", "部署上线", "devops", 4, ["test"]),
],
"analysis": [
SubTask("collect", "数据收集", "data_engineer", 1),
SubTask("clean", "数据清洗", "data_engineer", 2, ["collect"]),
SubTask("analyze", "数据分析", "analyst", 3, ["clean"]),
SubTask("visualize", "可视化", "analyst", 4, ["analyze"]),
]
}
# 根据任务关键词选择模板
if any(kw in task for kw in ["研究", "调查", "分析"]):
return templates["research"]
elif any(kw in task for kw in ["开发", "实现", "构建"]):
return templates["development"]
elif any(kw in task for kw in ["数据", "统计"]):
return templates["analysis"]
# 默认使用研究模板
return templates["research"]
def decompose_by_data(self, task: str, data_chunks: list) -> list[SubTask]:
"""
按数据分解任务
适用于：大规模数据处理任务
"""
subtasks = []
for i, chunk in enumerate(data_chunks):
subtask = SubTask(
id=f"process_{i}",
description=f"处理数据块 {i+1}/{len(data_chunks)}",
agent_type="data_processor",
priority=i + 1
)
subtasks.append(subtask)
# 添加汇总任务
subtasks.append(
SubTask(
id="aggregate",
description="汇总所有处理结果",
agent_type="aggregator",
priority=len(data_chunks) + 1,
dependencies=[f"process_{i}" for i in range(len(data_chunks))]
)
)
return subtasks
class TaskMerger:
"""
任务合并器
将多个子任务的结果合并为最终输出
"""
def __init__(self, llm=None):
self.llm = llm
def merge(
self,
results: list[dict],
strategy: MergeStrategy = MergeStrategy.CONCATENATE,
original_task: str = ""
) -> str:
"""
合并子任务结果
Args:
results: 子任务结果列表
strategy: 合并策略
original_task: 原始任务描述
Returns:
合并后的最终结果
"""
if not results:
return ""
if strategy == MergeStrategy.CONCATENATE:
return self._concatenate_merge(results)
elif strategy == MergeStrategy.SUMMARIZE:
return self._summarize_merge(results, original_task)
elif strategy == MergeStrategy.CROSS_REFERENCE:
return self._cross_reference_merge(results)
elif strategy == MergeStrategy.HIERARCHICAL:
return self._hierarchical_merge(results)
return str(results)
def _concatenate_merge(self, results: list[dict]) -> str:
"""简单拼接"""
merged = []
for r in results:
if isinstance(r, dict) and "result" in r:
merged.append(r["result"])
elif isinstance(r, str):
merged.append(r)
return "\n\n".join(merged)
def _summarize_merge(self, results: list[dict], task: str) -> str:
"""摘要合并"""
# 使用 LLM 进行智能摘要
# 这里简化处理
return f"基于{len(results)}个子任务结果，对「{task}」的综合分析已完成。"
def _cross_reference_merge(self, results: list[dict]) -> str:
"""交叉引用合并"""
# 识别结果间的关联
return "交叉引用分析完成，发现以下关联：..."
def _hierarchical_merge(self, results: list[dict]) -> str:
"""层级合并"""
# 按层级组织结果
sections = []
for i, r in enumerate(results):
sections.append(f"## 部分 {i+1}\n{r.get('result', str(r))}")
return "\n\n".join(sections)
# === 使用示例 ===
def task_management_demo():
"""任务分解与合并演示"""
print("📦 任务分解与合并演示")
print("=" * 50)
decomposer = TaskDecomposer()
merger = TaskMerger()
# 分解任务
task = "研究人工智能对就业市场的影响"
subtasks = decomposer.decompose_by_function(task)
print(f"\n📋 任务: {task}")
print("分解结果:")
for st in subtasks:
deps = f", 依赖: {st.dependencies}" if st.dependencies else ""
print(f" {st.id}. {st.description} ({st.agent_type}){deps}")
# 模拟执行
results = [
{"task_id": "search", "result": "搜索到了大量关于 AI 和就业的学术论文和报告"},
{"task_id": "analyze", "result": "分析显示 AI 对重复性工作影响最大，对创意工作影响较小"},
{"task_id": "synthesize", "result": "综合分析：AI 将改变就业结构，但不会导致大规模失业"},
]
# 合并结果
print("\n🔀 合并结果:")
merged = merger.merge(results, MergeStrategy.CONCATENATE)
print(merged)
if __name__ == "__main__":
task_management_demo()七、冲突检测与解决
7.1 冲突类型
┌─────────────────────────────────────────────────────────────────────┐
│ 多 Agent 系统中的冲突 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 冲突类型 │ │
│ │ │ │
│ │ 1. 资源冲突 │ │
│ │ Agent A ──▶ 占用 ──▶ 共享资源 │ │
│ │ Agent B ──▶ 等待 ──▶ 共享资源 │ │
│ │ │ │
│ │ 2. 结果冲突 │ │
│ │ Agent A ──▶ 结论A │ │
│ │ Agent B ──▶ 结论B ──▶ 相互矛盾！ │ │
│ │ │ │
│ │ 3. 目标冲突 │ │
│ │ Agent A ──▶ 目标1 │ │
│ │ Agent B ──▶ 目标2 ──▶ 目标冲突！ │ │
│ │ │ │
│ │ 4. 依赖冲突 │ │
│ │ Agent A ──▶ 需要 ──▶ 任务X │ │
│ │ Agent B ──▶ 也在等 ──▶ 任务X ──▶ 循环依赖！ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘7.2 冲突解决策略实现
"""
冲突检测与解决机制
"""
from typing import Optional
from dataclasses import dataclass
from enum import Enum
class ConflictType(Enum):
"""冲突类型"""
RESOURCE = "resource" # 资源冲突
RESULT = "result" # 结果冲突
GOAL = "goal" # 目标冲突
DEPENDENCY = "dependency" # 依赖冲突
class ConflictResolutionStrategy(Enum):
"""冲突解决策略"""
PRIORITY = "priority" # 优先级策略
ROLLBACK = "rollback" # 回滚策略
NEGOTIATION = "negotiation" # 协商策略
VOTING = "voting" # 投票策略
ARBITRATION = "arbitration" # 仲裁策略
@dataclass
class Conflict:
"""冲突定义"""
id: str
type: ConflictType
involved_agents: list[str]
description: str
timestamp: float
resolved: bool = False
resolution: Optional[str] = None
class ConflictDetector:
"""
冲突检测器
检测多 Agent 系统中的各种冲突
"""
def __init__(self):
self.conflicts: list[Conflict] = []
self.resource_locks: dict[str, str] = {} # resource -> holder
def detect_resource_conflict(
self,
agent_id: str,
resource: str
) -> Optional[Conflict]:
"""检测资源冲突"""
if resource in self.resource_locks:
holder = self.resource_locks[resource]
if holder != agent_id:
return Conflict(
id=f"conflict_{resource}_{agent_id}",
type=ConflictType.RESOURCE,
involved_agents=[holder, agent_id],
description=f"Agent {agent_id} 尝试访问已被 {holder} 占用的资源 {resource}",
timestamp=datetime.now().timestamp()
)
return None
def detect_result_conflict(
self,
agent_results: dict[str, str],
threshold: float = 0.3
) -> Optional[Conflict]:
"""检测结果冲突（基于语义相似度）"""
# 简化实现：检查是否有明显矛盾
results = list(agent_results.values())
# 检查是否有"但是"、"然而"等转折词表示冲突
for i, r1 in enumerate(results):
for r2 in results[i+1:]:
# 简单的冲突检测：检查是否有相反的结论
positive_markers = ["是", "可以", "应该", "会"]
negative_markers = ["不", "否", "不会", "不应该"]
has_positive = any(m in r1 for m in positive_markers)
has_negative = any(m in r2 for m in negative_markers)
if has_positive and has_negative:
return Conflict(
id=f"conflict_result_{i}",
type=ConflictType.RESULT,
involved_agents=list(agent_results.keys()),
description="Agent 结果存在矛盾",
timestamp=datetime.now().timestamp()
)
return None
def detect_dependency_conflict(
self,
tasks: list[dict]
) -> Optional[Conflict]:
"""检测依赖冲突"""
task_map = {t["id"]: t for t in tasks}
for task in tasks:
deps = task.get("dependencies", [])
for dep_id in deps:
if dep_id not in task_map:
continue
dep_task = task_map[dep_id]
dep_deps = dep_task.get("dependencies", [])
# 检查循环依赖
if task["id"] in dep_deps:
return Conflict(
id=f"conflict_dep_{task['id']}_{dep_id}",
type=ConflictType.DEPENDENCY,
involved_agents=[task.get("assignee"), dep_task.get("assignee")],
description=f"循环依赖：{task['id']} <-> {dep_id}",
timestamp=datetime.now().timestamp()
)
return None
class ConflictResolver:
"""
冲突解决器
提供多种冲突解决策略
"""
def __init__(self):
self.detector = ConflictDetector()
def resolve_with_priority(
self,
conflict: Conflict,
agent_priorities: dict[str, int]
) -> dict:
"""
优先级策略解决冲突
高优先级 Agent 获得资源
"""
# 找出优先级最高的 Agent
max_priority = -1
winner = None
for agent_id in conflict.involved_agents:
priority = agent_priorities.get(agent_id, 0)
if priority > max_priority:
max_priority = priority
winner = agent_id
return {
"strategy": ConflictResolutionStrategy.PRIORITY,
"winner": winner,
"loser": [a for a in conflict.involved_agents if a != winner],
"resolution": f"Agent {winner} 优先级最高，获得资源"
}
def resolve_with_rollback(
self,
conflict: Conflict,
rollback_plan: dict
) -> dict:
"""
回滚策略解决冲突
回滚到冲突发生前的状态
"""
return {
"strategy": ConflictResolutionStrategy.ROLLBACK,
"rollback_to": rollback_plan.get(conflict.id),
"resolution": "已回滚到冲突发生前的状态"
}
def resolve_with_negotiation(
self,
conflict: Conflict,
negotiation_prompt: str = ""
) -> dict:
"""
协商策略解决冲突
Agent 之间通过对话协商解决方案
"""
# 简化实现：直接要求重试
return {
"strategy": ConflictResolutionStrategy.NEGOTIATION,
"action": "retry_with_alternative",
"resolution": "Agent 将尝试其他方法或资源"
}
def resolve_with_arbitration(
self,
conflict: Conflict,
arbitrator: str = "system"
) -> dict:
"""
仲裁策略解决冲突
由仲裁者（如 Supervisor）做出最终决定
"""
# 简化实现：仲裁者做出决定
return {
"strategy": ConflictResolutionStrategy.ARBITRATION,
"arbitrator": arbitrator,
"decision": "基于综合评估的决定",
"resolution": "仲裁者已做出最终决定"
}
def resolve(self, conflict: Conflict, **kwargs) -> dict:
"""
统一解决接口
根据冲突类型和参数选择合适的解决策略
"""
strategy = kwargs.get("strategy", ConflictResolutionStrategy.PRIORITY)
if strategy == ConflictResolutionStrategy.PRIORITY:
return self.resolve_with_priority(conflict, kwargs.get("priorities", {}))
elif strategy == ConflictResolutionStrategy.ROLLBACK:
return self.resolve_with_rollback(conflict, kwargs.get("rollback_plan", {}))
elif strategy == ConflictResolutionStrategy.NEGOTIATION:
return self.resolve_with_negotiation(conflict)
elif strategy == ConflictResolutionStrategy.ARBITRATION:
return self.resolve_with_arbitration(conflict, kwargs.get("arbitrator", "system"))
return {"resolution": "未找到合适的解决策略"}
# === 使用示例 ===
from datetime import datetime
def conflict_demo():
"""冲突检测与解决演示"""
print("⚖️ 冲突检测与解决演示")
print("=" * 50)
detector = ConflictDetector()
resolver = ConflictResolver()
# 检测资源冲突
print("\n🔍 检测资源冲突")
detector.resource_locks["database"] = "Agent A"
conflict = detector.detect_resource_conflict("Agent B", "database")
if conflict:
print(f" 发现冲突: {conflict.description}")
# 解决冲突
resolution = resolver.resolve(
conflict,
strategy=ConflictResolutionStrategy.PRIORITY,
priorities={"Agent A": 1, "Agent B": 2}
)
print(f" 解决方案: {resolution['resolution']}")
print(f" 获胜者: {resolution.get('winner', 'N/A')}")
# 检测结果冲突
print("\n🔍 检测结果冲突")
results = {
"Agent A": "AI 会对就业产生负面影响，会导致大规模失业",
"Agent B": "AI 不会对就业产生负面影响，反而会创造更多机会"
}
conflict = detector.detect_result_conflict(results)
if conflict:
print(f" 发现冲突: {conflict.description}")
# 解决冲突
resolution = resolver.resolve(
conflict,
strategy=ConflictResolutionStrategy.ARBITRATION
)
print(f" 解决方案: {resolution['resolution']}")
if __name__ == "__main__":
conflict_demo()八、总结与对比
8.1 三种模式对比
┌─────────────────────────────────────────────────────────────────────┐
│ 多 Agent 协作模式对比 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┬─────────────┬─────────────┬─────────────────┐ │
│ │ 维度 │ Supervisor │ 去中心化 │ 混合模式 │ │
│ ├─────────────────┼─────────────┼─────────────┼─────────────────┤ │
│ │ 控制方式 │ 中央调度 │ 自主协商 │ 分层控制 │ │
│ │ 可扩展性 │ 中等 │ 高 │ 高 │ │
│ │ 复杂性 │ 低 │ 高 │ 中高 │ │
│ │ 可预测性 │ 高 │ 低 │ 中等 │ │
│ │ 容错性 │ 单点故障 │ 去中心化 │ 较好 │ │
│ │ 适用场景 │ 任务明确 │ 开放环境 │ 复杂大型系统 │ │
│ └─────────────────┴─────────────┴─────────────┴─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘8.2 选型建议
8.3 架构图回顾
┌─────────────────────────────────────────────────────────────────────┐
│ 多 Agent 协作架构总览 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 通信层 │ │
│ │ 消息队列 ────────────────────── 共享状态 │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌─────────────────────────────┼─────────────────────────────┐ │
│ │ 协调层 │ │
│ │ │ │
│ │ 冲突检测 ─────── 冲突解决 ─────── 任务分配 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌─────────────────────────────┼─────────────────────────────┐ │
│ │ Agent 层 │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Agent A │ │ Agent B │ │ Agent C │ │ Agent D │ │ │
│ │ │ 研究专家 │ │ 分析专家 │ │ 编码专家 │ │ 测试专家 │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘参考资料
本文首发于 2026 年 4 月 1 日，风格对标 PySuper（zhengxingtao.com）
评论区