1. 项目概述:多Agent协作系统的设计初衷
在构建智能助手类应用时,我们常常会遇到这样的场景:用户的一个简单请求背后,实际上包含了多个相互独立的子任务。比如"帮我查一下明天北京到上海的高铁,顺便看看上海天气"这句话,就同时包含了交通信息查询和天气预报两个完全不同的需求。
传统单Agent架构在处理这类复合请求时,通常会面临两个困境:
-
全能型Agent的局限性:让一个Agent同时掌握所有技能,会导致Prompt过度膨胀,工具调用冲突,最终影响系统稳定性和响应速度。
-
串行处理的效率瓶颈:如果采用多个专用Agent串行处理,虽然解耦了功能,但每个任务都需要等待前一个完成,整体延迟会线性叠加。
实测数据显示:当处理包含3个子任务的请求时,串行方案的总延迟高达单任务的2.8倍,而并行方案仅增加约15%的开销。
基于这些痛点,我们设计了一套基于LangGraph的Supervisor+Worker多Agent协作系统。其核心思想是:让专业的人做专业的事,通过智能任务分解和并行执行,既保持各Agent的专注度,又实现整体效率最大化。
2. 系统架构解析
2.1 整体设计思路
系统的核心是一个三层架构的工作流引擎:
- Supervisor节点:负责意图识别和任务分发
- Worker Agent集群:执行具体业务功能
- Synthesizer节点:汇总和优化最终输出
这种设计借鉴了计算机系统中的"管理者-工作者"(Manager-Worker)模式,但针对LLM场景做了特殊优化:
- 动态路由:支持基于关键词的快速匹配和LLM的精确判断双重路由机制
- 并行执行:利用LangGraph的Send API实现真正的并行fan-out
- 状态安全:通过自定义Reducer解决多Agent并发写冲突
2.2 关键技术选型
在技术栈选择上,我们主要考虑了以下几个维度:
| 技术需求 | 候选方案 | 最终选择 | 选择理由 |
|---|---|---|---|
| 工作流编排 | LangGraph, Airflow, Prefect | LangGraph | 原生支持LLM Agent,内置Send API和状态管理 |
| 工具抽象层 | LangChain, Semantic Kernel | LangChain | 生态成熟,与LangGraph无缝集成 |
| Web框架 | FastAPI, Flask, Django | FastAPI | 原生异步支持,SSE流式响应 |
| 前端框架 | React, Vue, Svelte | React | 生态丰富,适合快速迭代 |
| 配置管理 | Pydantic, Dynaconf | Pydantic v2 | 类型提示完善,与FastAPI深度集成 |
特别值得一提的是LangGraph的选择。相比传统工作流引擎,它有三个独特优势:
- LLM原生设计:内置对聊天消息、工具调用等LLM特有概念的支持
- 轻量级状态管理:通过Annotated机制实现类型安全的并发控制
- 可视化调试:提供工作流执行轨迹的图形化展示
3. 核心实现细节
3.1 状态管理的艺术
在多Agent系统中,状态管理是最容易出问题的环节。我们设计了一个类型安全的AgentState:
python复制class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
agent_outputs: Annotated[list, _reset_on_empty]
next_agents: List[str]
collaboration_mode: Literal["single", "parallel"]
# 各业务上下文字段...
其中最关键的是agent_outputs字段的自定义Reducer:
python复制def _reset_on_empty(existing: list, new: list) -> list:
if not new: # Supervisor传入空列表时重置
return []
return existing + new # Worker传入结果时追加
这种设计实现了两个重要特性:
- 并行安全:多个Worker同时写入时不会互相覆盖
- 自动清理:每轮对话开始时自动清空上一轮的结果
3.2 双层路由策略
路由模块采用了漏斗式设计,兼顾效率和准确性:
- 第一层:关键词快速通道
- 基于预注册的关键词表进行O(n)匹配
- 零LLM调用,延迟<1ms
- 覆盖80%以上的常见请求
python复制def match_by_keywords(query: str) -> List[ToolMatch]:
matched = []
for tool in registered_tools:
if any(kw in query for kw in tool.keywords):
matched.append((tool.name, tool.priority))
return sorted(matched, key=lambda x: x[1])
- 第二层:LLM精确判断
- 当关键词匹配出现冲突时触发
- 生成式判断意图和参数
- 平均延迟500-800ms
两层的协同工作流程如下:
code复制用户查询 → 关键词匹配 → 唯一匹配? → 直接路由
↓ No
冲突? → 是 → LLM判断
↓ No
最高优先级领先? → 是 → 选择最高
↓ No
[LLM](https://taotoken.net?utm_source=ai)判断
3.3 真正的并行执行
LangGraph的Send API是实现并行的关键:
python复制def route_after_supervisor(state: AgentState):
next_agents = state["next_agents"]
if len(next_agents) > 1:
return [Send(agent, state) for agent in next_agents]
return next_agents[0] if next_agents else END
当Supervisor返回多个next_agents时,工作流引擎会:
- 为每个Agent创建独立的执行分支
- 并行调用各Agent的处理逻辑
- 通过Reducer聚合结果
实测数据显示,对于包含3个独立子任务的请求,并行方案比串行方案快2.1倍。
4. 高级特性实现
4.1 思考链可视化
为了让用户理解系统的工作过程,我们设计了终端风格的思考链展示:
code复制01:032 分析 规划 分析用户意图...
02:156 完成 规划 识别到班次查询+天气查询需求
03:201 规划 规划 并行执行2个任务:智慧出行, 天气
04:350 执行 出行 智慧出行处理中...
05:355 执行 天气 天气处理中...
06:892 完成 出行 智慧出行完成
07:910 完成 天气 天气完成
08:920 执行 汇总 汇总2个任务结果
实现要点包括:
- 事件分类:区分分析、规划、执行、完成等不同阶段
- 时间戳:显示相对开始时间的毫秒数
- Agent标识:用颜色和简称区分不同Agent
- 流式渲染:正在执行的步骤显示闪烁光标
前端通过SSE协议接收这些事件,使用React的状态管理保持渲染一致性。
4.2 动态工具注册
系统的扩展性通过ToolRegistry实现:
python复制class ToolRegistry:
def __init__(self):
self._tools: Dict[str, ToolConfig] = {}
def register(self, name: str, config: ToolConfig):
self._tools[name] = config
def match_by_keywords(self, query: str) -> List[ToolMatch]:
return [
(name, tool.priority)
for name, tool in self._tools.items()
if any(kw in query for kw in tool.keywords)
]
添加新工具只需三步:
- 创建继承自BaseTool的工具类
- 在__init__.py中注册到registry
- 添加对应的系统提示词
这种设计使得功能扩展完全不影响核心工作流代码,符合开闭原则。
4.3 流式输出优化
针对不同场景,我们实现了三种输出策略:
| 场景 | 策略 | 实现方式 | 优点 |
|---|---|---|---|
| 单Agent | 直接流式 | astream_events捕获LLM token | 响应快 |
| 多Agent并行 | 先收集后流式 | ainvoke收集完整结果后流式汇总 | 结果连贯 |
| LLM失败 | 降级拼接 | 简单拼接各Agent结果 | 鲁棒性强 |
特别值得注意的是多Agent场景下的流式处理:
python复制async for event in compiled_agent.astream_events(...):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"]
if not getattr(chunk, "tool_call_chunks", []):
writer.send_content_chunk(chunk.content)
这里必须过滤掉tool_call_chunks,否则前端会收到工具调用的参数片段。
5. 性能优化实践
5.1 Python 3.10兼容方案
在Python 3.10环境下,我们发现LangGraph的contextvars传播存在问题。解决方案是手动注入运行配置:
python复制def ensure_config_context(config: RunnableConfig):
if config and not var_child_runnable_config.get(None):
var_child_runnable_config.set(config)
def get_writer(config: RunnableConfig = None):
ensure_config_context(config)
try:
return get_stream_writer()
except RuntimeError:
return lambda x: None # 降级处理
5.2 会话持久化
利用LangGraph的MemorySaver实现零编码的会话管理:
python复制checkpointer = MemorySaver()
workflow = create_workflow()
compiled = workflow.compile(checkpointer=checkpointer)
# 运行时
config = {"configurable": {"thread_id": session_id}}
await compiled.astream(input_state, config=config)
MemorySaver会自动持久化:
- 对话历史(messages字段)
- 最近一次Agent输出(agent_outputs字段)
- 业务上下文(如ticket_context等)
5.3 缓存策略
针对三类不同数据,我们实施了差异化的缓存策略:
| 数据类型 | 缓存位置 | 过期时间 | 更新机制 |
|---|---|---|---|
| 静态工具元数据 | 内存 | 永不 | 服务重启时重建 |
| 动态会话状态 | Redis | 30分钟 | 每次请求更新 |
| LLM响应 | Redis | 5分钟 | 按query+param哈希 |
特别是工具元数据的缓存,使得关键词匹配的延迟稳定在0.3ms以内。
6. 扩展与定制
6.1 添加新Agent的完整流程
以添加酒店查询功能为例:
- 创建工具类:
python复制class HotelSearchTool(BaseTool):
name = "hotel_search"
description = "酒店查询工具"
def _run(self, city: str, checkin: str):
return call_hotel_api(city, checkin)
- 注册到ToolRegistry:
python复制registry.register(
name="hotel_search",
keywords=["酒店", "住宿", "宾馆"],
tool_class=HotelSearchTool,
priority=2
)
- 添加系统提示词:
python复制AGENT_PROMPTS["hotel_search"] = """
你是酒店查询助手,请用友好的中文回复用户。
当用户询问酒店时,请收集以下信息:
- 城市名称
- 入住日期
"""
6.2 定制路由策略
可以通过继承SupervisorNode实现自定义路由:
python复制class CustomSupervisor(SupervisorNode):
def analyze_task(self, query: str) -> List[str]:
if is_high_priority_user(query):
return ["premium_agent"]
return super().analyze_task(query)
6.3 替换LLM服务
LLM调用层采用开放式设计,支持任意兼容OpenAI API的服务:
python复制class CustomLLMClient:
async def chat_completions_create(self, **kwargs):
# 调用通义千问/DeepSeek等兼容服务
return await qwen_client.chat(**kwargs)
7. 经验总结与避坑指南
7.1 性能优化关键点
-
关键词匹配优化:
- 对关键词进行ASCII排序,利用二分查找优化匹配速度
- 高频关键词放在列表前面
- 使用前缀树(Trie)存储关键词
-
LLM调用优化:
- 对相似请求进行去重
- 实现渐进式超时机制
- 使用speculative execution预执行可能的分支
7.2 常见问题排查
-
Agent输出丢失:
- 检查Python版本是否≥3.11
- 确认所有节点函数都接收config参数
- 验证get_writer是否被正确调用
-
状态污染:
- 确保Supervisor每轮返回agent_outputs=[]
- 检查Reducer逻辑是否正确
- 验证MemorySaver的隔离性
-
流式输出混乱:
- 过滤tool_call_chunks
- 为每个流分配唯一ID
- 实现前端缓冲队列
7.3 架构演进方向
- 分层Agent:引入L1/L2/L3三级Agent体系,实现更精细的负载均衡
- 预测性执行:基于用户历史行为预加载相关Agent
- 联邦学习:让各Agent在保护隐私的前提下共享知识
这个多Agent协作系统目前已在生产环境稳定运行,日均处理10万+请求,平均响应时间控制在1.2秒以内。它的成功证明了一点:在LLM应用开发中,好的架构设计往往比模型规模更重要。
