第 3 篇：让 Agent 学会分工，LangGraph 构建多 Agent系统

系列简介从零搭建一个多 Agent AI 助手覆盖原理、实现、部署全链路。不讲空话每篇都有可运行的代码。项目地址https://github.com/CodeMomentYY/LangGraph-Agent本篇目标用 LangGraph 搭建一个多 Agent 协作系统支持意图识别、动态路由、串行/并行执行。前言大家好我是一名前端工程师。都说前端“已死”那与其担心被 AI 替代不如打入敌人内部于是我开始折腾 Agent 开发。折腾下来发现Agent 的核心不是算法而是“工程能力”怎么设计架构、怎么串联服务、怎么把 LLM 的能力落地成产品。这些恰好是我们擅长的事。这个系列记录我从零搭建多 Agent 系统的完整过程。只聊技术知识和设计思路代码交给 AI 写。如果你也想从应用层切入 AI希望这个系列对你有帮助。读完本篇你将学到为什么需要多 Agent单 Agent 的瓶颈在哪LangGraph 是什么为什么选它多 Agent 架构设计Dispatcher 专业 Agent串行和并行两种执行模式的实现背景与动机上两篇我们用 30 行代码实现了单 Agent也了解了三种范式。但当你想让 Agent 同时具备“查天气”、“写文章”、“闲聊”这些能力时那么问题来了全塞到一个 Agent 里Prompt 会越来越臃肿。你得在一个 System Prompt 里描述所有工具、所有规则、所有场景。LLM 看到一大坨指令容易搞混该调工具的时候不调不该调的时候乱调。这就像一个人同时当厨师、服务员、收银员——忙得过来但质量堪忧。解决方案很自然分工。让不同的 Agent 各管一摊再加一个“调度员”决定把任务交给谁。但分工带来了新问题多个 Agent 之间怎么传递数据执行顺序怎么控制什么条件走哪条路如果还用前两篇的原生写法手写 for 循环 if-else代码会迅速变成一团意大利面。这就是为什么我们需要一个框架来管理流程——不是因为原生写法不行而是当节点多了、路由复杂了框架能帮你把“流程控制”这件事做得更清晰。核心概念单 Agent vs 多 Agent我们看一下两者对比单 Agent多 AgentPrompt一个巨大的 System Prompt每个 Agent 一个精简 Prompt职责什么都干各司其职调试出错不知道哪个环节的问题哪个 Agent 出错一目了然扩展加功能就改 Prompt越改越乱加功能就加一个新 Agent速度还行多了 Dispatcher 的开销设计多 Agent 架构架构图三个专业 Agenttools 路径需要调工具的任务查天气、获取网页、查语雀文档内部是 ReAct 循环writer 路径写作类任务邮件、文案、翻译纯 LLM 生成temperature 调高一点增加创造力chat 路径闲聊和知识问答纯 LLM 对话会从知识库检索相关内容RAG使用 LangGraph 来构建为什么选 LangGraph 而不是 LangChain很多人分不清 LangChain 和 LangGraph 的关系。简单来说LangChain工具箱。提供了调 LLM、解析输出、连接向量库等基础能力像是一堆零件。LangGraph施工图。在 LangChain 的零件基础上定义了谁先执行、谁后执行、什么条件走哪条路的流程控制。感兴趣的可以看看这篇文章https://www.datacamp.com/tutorial/langchain-vs-langgraph-vs-langsmith-vs-langflowLangChain 本身也能做 AgentAgentExecutor但它的问题是流程是黑盒的。你把工具和 Prompt 丢进去它内部自己循环你很难控制“什么时候该停”、“中间结果怎么传递”、“多个 Agent 怎么协作”。LangGraph 把流程显式化了——你画一张图节点是什么、边怎么连、条件是什么一目了然。对于多 Agent 协作这种需要精确控制流程的场景LangGraph 比 LangChain 的 AgentExecutor 好用得多。LangChain AgentExecutorLangGraph流程控制黑盒框架内部循环白盒你画图定义多 Agent不原生支持天然支持多节点条件路由难实现一行代码状态管理手动传递自动维护调试难看不到中间过程易逐节点 stream适合场景简单单 Agent复杂多 Agent 协作其实在项目里LangChain 的东西还是在用的消息格式、工具定义、OpenAI SDK只是不用它的 AgentExecutor改用 LangGraph 来编排流程。动手实现LangGraph 基础用法很多小伙伴不熟悉 LangGraph所以在搭建多 Agent 之前先看一个最简单的 LangGraph 示例理解它的三个核心概念State状态、Node节点、Edge边。其实就是把代码调度更具体化了像是在画“流程图”。fromlanggraph.graphimportStateGraph,ENDfromtyping_extensionsimportTypedDict# 1. 定义状态所有节点共享的数据classMyState(TypedDict):message:strcount:int# 2. 定义节点就是普通函数接收 state返回要更新的字段defsay_hello(state):return{message:f你好你已经来了{state[count]}次}defadd_count(state):return{count:state[count]1}# 3. 构建图注册节点 连接边graphStateGraph(MyState)graph.add_node(counter,add_count)graph.add_node(greeter,say_hello)graph.set_entry_point(counter)# 入口graph.add_edge(counter,greeter)# counter → greetergraph.add_edge(greeter,END)# greeter → 结束appgraph.compile()# 4. 运行resultapp.invoke({message:,count:0})print(result)# {message: 你好你已经来了 1 次, count: 1}就这么简单定义状态 → 写几个函数 → 用边连起来 → Compile → Invoke调用。LangGraph 真正强大的地方在于条件边——根据状态动态决定下一步走哪# 条件边根据 count 决定走哪条路defshould_continue(state):ifstate[count]3:returndonereturnagaingraph.add_conditional_edges(greeter,should_continue,{again:counter,# count 3 → 回到 counter循环done:END,# count 3 → 结束})有了条件边就能实现循环、分支、路由——这正是多 Agent 系统需要的能力。Step 1定义状态LangGraph 的核心是 State——所有节点共享的数据结构。每个节点读取 state、处理、写回 state。fromtypingimportAnnotated,Sequencefromtyping_extensionsimportTypedDictfromlangchain_core.messagesimportBaseMessageimportoperatorclassAgentState(TypedDict):所有节点共享的状态# 对话历史自动追加messages:Annotated[Sequence[BaseMessage],operator.add]# 意图列表dispatcher 填写intents:list[str]# 执行模式sequential 或 parallelmode:str# 当前执行到第几个意图current_step:int关键设计messages用operator.add标注意味着每个节点返回的消息会追加到列表里而不是覆盖这样对话历史自动累积。Step 2实现 Dispatcher意图分类Dispatcher 是整个系统的入口负责判断用户想干什么DISPATCHER_PROMPT你是一个意图分类器。 分类规则 - tools需要调用工具的请求查天气、获取网页、查文档 - writer写作类写邮件、文案、翻译 - chat其他闲聊、知识问答 如果需要多个步骤按顺序列出。 回复格式意图1,意图2|模式sequential 或 parallel 示例 - 上海天气 → tools|sequential - 查天气然后写文案 → tools,writer|sequential - 翻译hello顺便查天气 → tools,writer|parallel defdispatcher_node(state):last_user_msg...# 取最后一条用户消息responseinvoke_llm([SystemMessage(DISPATCHER_PROMPT),...])# 解析intents modeintents,modeparse_response(response)return{intents:intents,mode:mode,current_step:0}Dispatcher 的输出决定了后续走哪条路。注意它支持多意图。比如“查天气然后写文案”会被拆成[tools, writer]串行执行。Step 3构建 LangGraph 图这是最关键的部分——把所有节点和路由规则组装成一张图我们先看流程设计最终代码实现fromlanggraph.graphimportStateGraph,ENDdefbuild_graph():graphStateGraph(AgentState)# 注册节点graph.add_node(dispatcher,dispatcher_node)graph.add_node(mode_router,lambdastate:{})# 路由跳板graph.add_node(step_router,lambdastate:{})# 串行步骤跳板graph.add_node(router,router_node)# 工具决策graph.add_node(tool_executor,tool_executor_node)graph.add_node(writer_agent,writer_agent_node)graph.add_node(chat_agent,chat_agent_node)graph.add_node(advance_step,advance_step)# 推进到下一步graph.add_node(parallel_executor,parallel_executor_node)# 入口graph.set_entry_point(dispatcher)# dispatcher → mode_routergraph.add_edge(dispatcher,mode_router)# 根据 mode 分流graph.add_conditional_edges(mode_router,route_mode,{sequential:step_router,parallel:parallel_executor,})# 串行根据当前 intent 路由graph.add_conditional_edges(step_router,route_current_step,{tools:router,writer:writer_agent,chat:chat_agent,done:END,})# tools 路径ReAct 循环graph.add_conditional_edges(router,should_use_tools,{tools:tool_executor,next:advance_step,})graph.add_edge(tool_executor,router)# writer/chat 完成后推进graph.add_edge(writer_agent,advance_step)graph.add_edge(chat_agent,advance_step)# 检查是否还有下一步graph.add_conditional_edges(advance_step,has_next_step,{continue:step_router,done:END,})# 并行路径graph.add_edge(parallel_executor,END)returngraph.compile()Step 4串行 vs 并行串行有依赖关系的任务。“查天气然后写文案”——writer 需要天气结果才能写。并行互不依赖的任务。“翻译hello world顺便查天气”——两个任务没关系。并行模式用ThreadPoolExecutor同时执行多个 Agent最后让 LLM 把结果整合成一个连贯的回答。Step 5验证效果# 单意图curl-XPOST /api/chat-d{message: 你好}# → dispatcher: chat → chat_agent → 你好有什么可以帮你的# 单意图 工具curl-XPOST /api/chat-d{message: 上海天气}# → dispatcher: tools → router → get_weather → router → 上海晴天24°C# 多意图串行curl-XPOST /api/chat-d{message: 查上海天气写个朋友圈文案}# → dispatcher: [tools, writer] sequential# → tools查天气→ writer基于天气写文案→ 最终回复# 多意图并行curl-XPOST /api/chat-d{message: 翻译hello world顺便查北京天气}# → dispatcher: [tools, writer] parallel# → 并发执行 → 整合回复刨根问底序号问题1️⃣QDispatcher 的意图分类准确吗模糊场景怎么办A不一定准。帮我写个出行计划到底是 tools 还是 writer靠 Prompt 里的规则和示例来约束。模糊时兜底到 chat最安全的路径。后续可以加 few-shot 示例提升准确率。2️⃣Q并行模式怎么整合多个 Agent 的结果A用 LLM 整合。把多个 Agent 的输出拼在一起让 LLM 生成一个连贯的最终回答。类似于你有两段信息请合并成一段自然的回复。3️⃣QLangGraph 和直接写 Python 函数调用有什么区别A小项目没区别。但当图变复杂10 节点、多种条件路由、需要流式输出时LangGraph 的状态管理和条件边比手写 if-else 清晰得多。而且它内置了 stream 模式后面做 SSE 推送时直接用。本篇小结多 Agent 的核心是分工——Dispatcher 做路由专业 Agent 各管一摊LangGraph 用图来描述 Agent 的执行流程节点是函数边是条件支持串行有依赖和并行无依赖两种多意图执行模式dispatcher 的意图分类质量决定了整个系统的上限写在最后多 Agent 架构看起来复杂但核心思想很朴素把大问题拆成小问题让专业的人做专业的事。这和微服务架构的理念一模一样——单体应用拆成多个服务每个服务职责单一通过 API 协作。如果你做过前端的组件化拆分多 Agent 的设计思路你一定不陌生。下一篇预告核心服务架构搭好了接下来让它真正能完整跑起来——Web 界面 SSE 流式输出把 Agent 从命令行变成一个能用的产品。