基于LangGraph的多Agent并行协作系统设计与实现

1. 项目概述：基于LangGraph的多Agent协作系统

这个项目构建了一个面向旅行场景的多Agent协作系统，核心目标是解决复杂任务分解与并行处理的问题。当用户提出"帮我查一下明天北京到上海的高铁，顺便看看上海天气"这类复合请求时，传统单一Agent架构要么需要设计臃肿的Prompt，要么只能串行执行导致响应延迟。我们的系统通过LangGraph实现了真正的并行任务调度，将端到端响应时间从串行模式的T1+T2优化为并行模式的max(T1,T2)。

系统采用三层架构设计：

• Supervisor：负责意图识别和任务分解
• Worker Agents：专精于特定领域的任务执行
• Synthesizer：汇总多个Agent的执行结果

技术栈选择上，后端使用FastAPI提供SSE流式接口，LangGraph构建工作流引擎，LangChain封装工具调用，前端采用React+TypeScript+Vite实现终端风格的思考过程可视化。整个系统设计遵循"职责单一"原则，每个组件都保持高度内聚，平均代码量控制在100行以内，便于维护和扩展。

2. 核心架构设计解析

2.1 整体数据流设计

系统数据流经过精心设计，确保从用户请求到最终响应的每个环节都高效可靠：

code复制用户输入
  │
  ▼
FastAPI SSE接口(POST /api/v1/chat/stream)
  │
  ▼
LangGraph StateGraph工作流
  ├─ Supervisor节点（意图识别与路由）
  │   ├─ 关键词快速匹配层（零LLM开销）
  │   └─ LLM精确分析层（语义模糊时触发）
  │
  ├─ Worker Agent节点（并行执行）
  │   ├─ ticket_search：班次查询
  │   ├─ weather_search：天气查询
  │   ├─ kb_search：政策咨询
  │   └─ chitchat：自然语言闲聊
  │
  └─ Synthesizer节点（结果聚合）
      ├─ 单结果透传
      └─ 多结果LLM汇总
  │
  ▼
SSE事件流 → 前端渲染

这种架构的关键优势在于：

1. 并行处理能力：通过LangGraph的Send API实现真正的并行fan-out
2. 弹性扩展：新增Agent只需注册工具，无需修改工作流代码
3. 实时可视化：终端风格的思考链展示增强用户体验

2.2 状态管理设计

系统的核心状态通过AgentState类管理，采用TypedDict确保类型安全：

python复制class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]  # 对话历史
    agent_outputs: Annotated[list, _reset_on_empty]  # Agent执行结果
    next_agents: List[str]  # 路由决策结果
    collaboration_mode: str  # "single"|"parallel"
    # 各Agent业务上下文
    ticket_context: dict
    weather_context: dict
    consultation_context: dict
    final_response: str  # 最终响应

其中agent_outputs字段使用自定义Reducer解决并行写入问题：

python复制def _reset_on_empty(existing: list, new: list) -> list:
    """空列表重置，非空追加"""
    if not new:
        return []  # Supervisor传入[]时清空旧数据
    return existing + new  # Worker Agent结果安全聚合

这种设计相比传统加锁方案更优雅，将并发问题消化在状态定义层，业务代码无需关心线程安全问题。

3. 关键技术实现细节

3.1 Supervisor双层路由机制

路由策略直接影响系统响应速度和准确率。我们设计了"关键词快速通道+LLM精确补充"的双层漏斗：

python复制def _analyze_task(query: str) -> dict:
    registry = get_tool_registry()
    
    # 第一层：关键词匹配（零LLM开销）
    matched_tools = registry.match_by_keywords(query)
    if len(matched_tools) == 1:
        return {"mode": "single", "subtasks": [...]}
    
    # 第二层：LLM精确分析
    return _analyze_with_llm(query)

性能对比数据：

实测表明，80%以上的日常请求都能通过关键词匹配直接路由，大幅降低LLM调用开销。

3.2 真正的并行fan-out实现

LangGraph的Send API是本项目的核心技术亮点，实现代码如下：

python复制def route_after_supervisor(state: AgentState):
    next_agents = state.get("next_agents", [])
    
    if not next_agents:
        return END  # 无任务直接结束
        
    if len(next_agents) == 1:
        return next_agents[0]  # 单任务直连
        
    # 多任务并行fan-out
    return [Send(agent, state) for agent in next_agents]

工作流编译时将节点和边组装成有向无环图：

python复制def create_workflow() -> StateGraph:
    workflow = StateGraph(AgentState)
    
    # 添加三类节点
    workflow.add_node("supervisor", supervisor_node)
    for name in agents.keys():
        workflow.add_node(name, _make_agent_node(name))
    workflow.add_node("synthesizer", synthesizer_node)
    
    # 设置边和条件路由
    workflow.set_entry_point("supervisor")
    workflow.add_conditional_edges("supervisor", route_after_supervisor)
    for name in agents.keys():
        workflow.add_edge(name, "synthesizer")
    workflow.add_edge("synthesizer", END)
    
    return workflow

这种设计使得两个Agent并行时，系统延迟从T1+T2降低到max(T1,T2)，实测复合查询的响应时间缩短40%以上。

3.3 流式输出双模态策略

Worker Agent根据任务类型采用不同的输出策略：

python复制def _make_agent_node(agent_name: str, compiled_agent):
    async def node_fn(state: AgentState, config: RunnableConfig):
        writer = get_writer(config)
        is_single_agent = len(state["next_agents"]) == 1

        if is_single_agent:
            # 单Agent：流式推送每个token
            response = await _stream_agent(compiled_agent, writer)
        else:
            # 多Agent：收集完整结果供Synthesizer汇总
            result = await compiled_agent.ainvoke(...)
            response = _extract_response(result)
            
        return {
            "messages": [AIMessage(content=response)],
            "agent_outputs": [{"agent_name": agent_name, "response": response}]
        }
    return node_fn

流式实现的关键是捕获ReAct子图内部的LLM token：

python复制async def _stream_agent(compiled_agent, writer):
    response_text = ""
    async for event in compiled_agent.astream_events(...):
        if event["event"] == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content and not getattr(chunk, "tool_call_chunks", []):
                response_text += chunk.content
                writer({"type": "content_stream", "block": {
                    "delta": chunk.content,
                    "accumulated": response_text
                }})
    return response_text

关键细节：必须过滤tool_call_chunks，否则前端会收到工具调用的JSON碎片

4. 扩展与维护方案

4.1 插件化扩展机制

添加新Agent只需三步，无需修改工作流代码：

1. 创建LangChain Tool：

python复制class HotelSearchTool(BaseTool):
    name = "hotel_search"
    description = "酒店查询工具"
    args_schema = HotelSearchInput

    def _run(self, city: str, checkin: str) -> str:
        return f"{city}的酒店信息..."

2. 注册到ToolRegistry：

python复制registry.register(
    name="hotel_search",
    display_name="酒店",
    keywords=["酒店", "住宿"],
    tool_class=HotelSearchTool,
    priority=2
)

3. 添加系统提示词：

python复制AGENT_PROMPTS["hotel_search"] = "你是酒店查询助手..."

这种设计使得系统功能扩展变得非常简单，平均添加新功能只需30分钟开发时间。

4.2 配置管理方案

采用pydantic-settings统一管理配置：

python复制# backend/app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    LLM_API_KEY: str
    LLM_API_BASE: str = "https://dashscope.aliyuncs.com"
    LLM_MODEL: str = "qwen-max"
    
    class Config:
        env_file = ".env"

配置项通过环境变量注入，确保敏感信息不会硬编码在代码中。

5. 性能优化与问题排查

5.1 Python 3.10兼容性方案

遇到最棘手的问题是Python 3.10下contextvars传播异常：

python复制def ensure_config_context(config: RunnableConfig) -> None:
    """手动注入contextvar，解决Python 3.10兼容问题"""
    from langchain_core.runnables.config import var_child_runnable_config
    if config and not var_child_runnable_config.get(None):
        var_child_runnable_config.set(config)

def get_writer(config: RunnableConfig = None) -> Callable:
    """获取安全的stream writer"""
    if config:
        ensure_config_context(config)
    try:
        from langgraph.config import get_stream_writer
        return get_stream_writer()
    except RuntimeError:
        return lambda x: None  # 降级处理

这个方案确保在Python 3.10环境下也能正确获取stream writer。

5.2 常见问题排查指南

6. 部署与使用指南

6.1 环境准备

系统要求：

• Python 3.10+
• Node.js 18+
• Redis（可选，用于会话持久化）

安装依赖：

bash复制# 后端
cd backend
pip install -r requirements.txt

# 前端
cd frontend
npm install

6.2 配置说明

.env配置示例：

ini复制LLM_API_KEY=sk-xxxxxxxxxxxxxxxx
LLM_API_BASE=https://dashscope.aliyuncs.com
LLM_MODEL=qwen-max

# 外部服务
TICKET_API_URL=http://localhost:9001
QDRANT_URL=http://localhost:6333

6.3 启动命令

bash复制# 后端
python -m uvicorn app.main:app --reload --port 8002

# 前端
npm run dev

访问http://localhost:5173即可开始使用系统。

7. 项目总结与改进方向

这个多Agent协作系统通过LangGraph实现了真正的并行任务处理，核心创新点包括：

1. 双层路由策略平衡速度与准确率
2. Send API实现并行fan-out
3. 自定义Reducer解决状态共享问题
4. 插件化架构支持快速扩展

实测数据显示：

• 关键词路由节省80%的LLM调用
• 并行处理使复合查询延迟降低40%
• 新增Agent开发时间控制在30分钟内

未来改进方向：

1. 引入优先级调度机制
2. 增加Agent间通信能力
3. 优化资源利用率监控
4. 支持动态Agent加载

这个架构不仅适用于旅行场景，经过适当调整也可应用于客服、数据分析等领域，具有广泛的适用性。