最近很多朋友问我同一个问题多智能体到底怎么上手看了不少概念文章一打开代码库还是懵。我觉得与其啃复杂框架不如先手搓一个“AI任务调度助手”把多智能体的核心思路跑通。这篇文章就用LangChain生态里最常用的工具搭一个能拆解任务、分派Agent、汇总结果的简单调度器。看完你就能理解多智能体的协作逻辑也能直接跑出一版可用的Demo。先说清楚这个Demo要解决什么问题。平时我们调大模型基本是一个Agent从头干到尾。比如让它写一份行业分析报告它又要联网查资料又要整理数据还要润色文字最后出来的内容常常顾此失彼。多智能体的思路是把一个大任务拆成多个子任务每个子任务交给一个专职Agent去处理——有负责调研的有负责写初稿的有负责审核的。调度器Orchestrator负责拆任务、发任务、收集结果。这种结构的好处是职责清晰每个Agent只干一件事模型就不容易“精神分裂”而且可以针对不同子任务选不同的模型或提示词。我选的实现路线是LangGraph不是LangChain早期的AgentExecutor。LangGraph是LangChain团队专门做Agent编排的库它把Agent流程定义成一张图节点是Agent的动作边是节点之间的流转状态是节点间传递的数据。这种“状态图”的方式特别适合调度场景因为你可以清楚看到任务流转到哪一步、状态里存了什么出问题也容易排查。1. 整体设计与思路拆解1.1 为什么多智能体反而比单个大模型好用单个大模型Agent不是不能用而是在处理复杂任务时有几个痛点。第一上下文容易被污染。一个Agent既要处理用户需求又要调用工具还要根据中间结果调整策略所有信息都堆在一个上下文窗口里跑着跑着就忘了前面的关键约束。第二提示词写起来非常纠结。想让一个Agent既懂“拆任务”又懂“调工具”又懂“汇总报告”这个系统提示词可能要上千字而且模型很容易顾此失彼。第三调试困难。任务哪一步出错了只能看一长串对话历史定位问题很痛苦。多智能体架构把这些问题拆开了。每个Agent只维护一小段上下文提示词短小专注一个环节出问题只看那个Agent的状态即可。像我这版调度助手就分了三个角色任务规划Agent负责拆解需求、执行Agent负责具体干活、质检Agent负责检查结果。三个Agent各司其职调度器只在它们之间传状态不干涉具体执行。这样跑起来之后我可以单独调试任何一个环节比如觉得写稿质量不好就只优化写稿Agent的提示词其他都不用动。1.2 选LangGraph做编排的理由如果你去搜“LangChain多智能体”会看到很多旧教程还在用initialize_agent或者AgentExecutor。坦白说那套东西做简单问答还行做调度就力不从心了。AgentExecutor是一个封装好的黑盒你很难在里面插入“先检查再决定下一步”的逻辑。后来LangChain团队推出LangGraph思路变成“把Agent流程画成图”这对调度场景是降维打击。LangGraph里有两个核心概念State状态和Graph图。State是一个全局数据对象所有节点都能读写它Graph就是节点和边的连接。调度器运行时从入口节点开始按边走到下一个节点直到走到END节点。每个节点可以是一个Python函数也可以是一个内置的Agent节点。节点执行完之后返回一个字典更新State。边可以是普通边无条件跳转也可以是条件边根据当前State内容决定跳转到哪个节点。这套表达能力和灵活性比AgentExecutor强太多了而且调试时能看到完整的运行轨迹。1.3 调度助手的架构长什么样我这版Demo的流程图用文字描述大概是这样的入口节点接收用户输入把任务文本存进State然后进入规划节点让大模型拆解任务输出一个子任务列表每个子任务包含任务ID、任务描述、负责角色接下来进入执行阶段每个子任务落到对应的Agent节点上执行所有子任务执行完后进入汇总节点把各Agent的输出整理成最终答复。为了让新手好理解我故意没做复杂的并行分发。实际项目中可以让多个Agent并行执行但在入门Demo里串行执行更容易看清每个节点的输入输出。我在设计时刻意加了一个“人工审核”节点规划节点拆解完任务后先不急着执行而是把任务清单返回给用户用户确认后再往下走。这样既演示了LangGraph的中断与恢复机制也防止模型拆出奇怪的任务列表浪费token。这个设计思路来自实际踩坑——模型自己拆任务很容易拆出不可执行的子任务加一道人工确认效果立刻不一样。2. 核心细节解析与实操要点2.1 这些概念先搞清楚再动手在写代码之前有几个LangGraph的高频概念需要先有感觉。第一个是State它决定节点间怎么传数据。你用TypedDict定义一个状态结构比如{input: str, tasks: list, results: dict}。所有节点函数接收一个state参数返回一个更新字典框架会自动合并。第二个是节点节点就是普通Python函数签名是(state) - dict。需要注意节点返回的字典不会覆盖整个State只会更新你指定的key。第三个是边。普通边用add_edge表示执行完节点A必然去节点B。条件边用add_conditional_edges你需要传一个路由函数函数根据当前State计算一个分支名框架根据分支名找到对应的目标节点。第四个是状态记忆LangGraph默认会把State传给每个节点但不会自动保存历史如果你需要多轮对话记忆要在State里加一个messages字段专门存历史消息。2.2 模型与工具怎么选型这版Demo不绑定云厂商我推荐用兼容OpenAI接口的本地推理服务比如你在本地启动一个模型服务然后通过ChatOpenAI(base_urlhttp://localhost:11434/v1, api_keynot-needed, modellocal-chat-model)来调用。这样做的原因是入门阶段调试频率很高如果每调一次都走API付费心理上压力大而且本地模型的响应延迟低调试体验更流畅。不过要注意本地模型的能力参差不齐规划Agent这种需要结构化输出的环节建议选指令跟随能力强的模型。工具方面执行Agent可以用内置的TavilySearchResults做联网检索也可以写一个简单的file_tool读写本地文件。我这版Demo里做了一个自定义工具叫now_time用来返回当前时间。你别小看这个工具很多模型对“当前时间”没有概念写日报类任务时总会瞎编日期。给Agent挂一个取当前时间的工具单位时间内出错的概率立刻降下来。这其实说明了一个道理多智能体系统里工具不在多而在准。2.3 结构化输出是调度器的生命线在多智能体调度里最怕的就是模型返回一段乱七八糟的文本下游节点没法解析。规划节点的输出必须是结构化的子任务清单否则没法分派。所以我会给规划节点用强格式约束——要么用Pydantic的with_structured_output要么在提示词里明确要求JSON格式再加一个JSON解析兜底函数。实测中提示词写得好不好直接决定解析成功率。我总结了一个比较稳的JSON提示词模板要求模型只输出JSON不要输出任何解释提供JSON示例字段用蛇形命名明确说“如果不符合条件返回空列表而不是报错”。除了模型输出节点之间的消息格式也要统一。我这版里每个Agent返回的output字段固定是字符串字符串汇总节点直接拼接。如果你想传更富的数据比如置信度、来源链接就用字典或Pydantic模型并写在State定义里。切忌一会儿返回字符串一会儿返回字典排查问题时会疯掉。3. 实操过程与核心环节实现3.1 环境安装与目录结构先创建一个干净的虚拟环境然后安装依赖。我用的是langgraph它会自动带起langchain-core等相关包。命令就一条pip install langgraph langchain-core langchain-openai如果你要用联网搜索工具再加一个tavily-python并设置API Key。不用工具也能跑Demo可以先用一个模拟工具替代。我的项目结构很简单只有两个文件state.py放状态定义agent.py放节点和构图逻辑。新手先不要学别人一上来搞十几个文件单个文件跑通再拆分。3.2 定义状态结构我这样定义Statefrom typing import TypedDict, List, Dict, Optional class SubTask(TypedDict): task_id: str desc: str owner: str status: str class AgentState(TypedDict): user_request: str tasks: List[SubTask] current_task: Optional[SubTask] results: Dict[str, str] error: Optional[str]这个结构里的核心是results它是一个字典key是子任务IDvalue是Agent执行后的文本结果。汇总节点只认这个字典不用关心中间过程。current_task字段是给执行节点用的每次从tasks里取一个未处理的任务放进去执行完再写回results。error字段用来记录错误信息方便条件边判断是否需要重试。3.3 写规划节点规划节点的核心工作是把用户请求变成子任务清单。我用LangChain的with_structured_output来做感觉比手动解析JSON稳得多。定义好子任务的数据结构from pydantic import BaseModel, Field class SubTaskOut(BaseModel): task_id: str Field(description任务编号例如 T01) desc: str Field(description任务描述) owner: str Field(description负责角色名称) class TaskPlan(BaseModel): tasks: List[SubTaskOut] Field(description子任务清单)接着写规划节点函数from langchain_openai import ChatOpenAI system_planner 你是一个项目规划助手。用户会给你一个完整需求你需要把它拆解为多个独立的子任务。 要求 1. 每个子任务只包含一个明确的动作。 2. 子任务数量控制在3到5个。 3. 角色限定为research 或 write 或 review。 4. 最终汇总不需要单独拆任务。 只输出结构化结果不允许额外解释。用户需求{request} def plan_node(state: AgentState) - dict: llm ChatOpenAI(modellocal-chat-model, temperature0) structured_llm llm.with_structured_output(TaskPlan) prompt system_planner.format(requeststate[user_request]) plan structured_llm.invoke(prompt) tasks [SubTask(task_idt.task_id, desct.desc, ownert.owner, statuspending) for t in plan.tasks] return {tasks: tasks}这里把temperature设为0是为了让模型减少发散输出更可预测。实际对比过temperature0.7时偶尔会拆出玄幻的任务描述0就很稳。3.4 写执行节点和质检节点执行节点不区分是research还是write统一走一个通用执行函数。判断标准就看owner字段如果是research就调用搜索工具如果是write就让模型基于已有资料写文本如果是review就检查文本长度和关键词。def execute_node(state: AgentState): if not state.get(tasks): return {error: no tasks} for t in state[tasks]: if t[status] pending: state[current_task] t if t[owner] research: output do_research(t[desc]) elif t[owner] write: output do_write(t[desc]) else: output do_review(t[desc]) state[results][t[task_id]] output t[status] done break return {results: state[results], tasks: state[tasks]}这样写有一个好处一次执行一个子任务下次再进入节点时继续处理下一个逻辑简单且不容易死循环。do_write其实就是一次普通的大模型生成调用do_review可以简单检查字数超过500字就算过。3.5 组装LangGraph并加入人工确认最关键的是构图。我用了StateGraph然后在规划节点之后加了一个条件边根据用户输入的“是否自动执行”标志决定是去执行还是返回给用户确认。如果你在终端环境跑用户确认要修改StateLangGraph提供了一个interrupt_before参数可以在进入执行节点前中断然后你手动修改State再恢复运行。from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver g StateGraph(AgentState) g.add_node(planner, plan_node) g.add_node(executor, execute_node) g.add_node(summarizer, sum_node) g.set_entry_point(planner) g.add_edge(planner, executor) g.add_conditional_edges(executor, check_remaining_tasks, {continue: executor, done: summarizer}) g.add_edge(summarizer, END) app g.compile(checkpointerMemorySaver())关键在于check_remaining_tasksdef check_remaining_tasks(state): if any(t[status] pending for t in state[tasks]): return continue return done这样执行节点会反复进入直到所有任务做完才进入汇总。如果你想要人工确认可以在compile时传interrupt_before[executor]然后在调用invoke后访问app.get_state(config)修改state后调用app.update_state(config, values)再继续invoke。这个机制我第一次用的时候觉得很绕后来发现它其实是LangGraph给我们留的“安全刹车”。3.6 总结节点与整体运行效果汇总节点就是把state[results]里的各段输出整合一下有时也会让大模型把多个结果转成一篇连贯的最终报告。我这里是自己拼接加简单润色因为初版要尽量少依赖大模型跑得快一些。def sum_node(state: AgentState): parts [state[results][t[task_id]] for t in state[tasks]] final_text \n\n.join(parts) return {final_answer: final_text}运行一个例子看看效果。用户输入“帮我整理一篇关于本地模型推理加速的调研短文”规划节点拆出来的任务大概是“搜集相关工具及性能数据”“撰写调研短文”“检查字数是否满足要求”。执行节点分别处理后汇总节点把三块内容拼起来。整体跑下来耗时可能比单个Agent长一点但每一段内容的质量是稳定的不会出现“前面还在讲推理加速后面突然扯到部署成本”这种跑题情况。4. 常见问题与排查技巧实录4.1 模型不遵守JSON输出指令这个问题我几乎每次做多智能体项目都会遇到。要么是模型输出里带着解释性文字要么是JSON字段名不对。解决办法有两个方向第一换更强指令跟随的模型如果本地模型达不到要求就在结构化输出后面加一个兜底解析函数用正则把第一个{到最后一个}切出来再json.loads第二在提示词里把“不遵守格式的后果”写清楚比如“任何非JSON输出都会被系统丢弃你将拿不到任何报酬”实测下来威慑力比“请遵守JSON格式”管用。4.2 调度器一直在执行节点里打转这个问题多半出在状态更新逻辑上。比如忘了把已处理的任务改成done或者check_remaining_tasks判断条件写反了。我排查时会在每个节点入口打印一下State的关键字段print(, state.get(current_task), state.get(results))确认任务状态是否更新。还有个隐蔽坑LangGraph的State更新是按key覆盖的如果你在节点里直接修改了传入的列表对象而没有返回新对象框架可能感知不到变化。所以返回时要显式返回tasks和results。4.3 多个Agent上下文互相污染如果你把多个Agent塞进同一个State里彼此历史消息混在一起很容易造成模型行为错乱。我用的隔离技巧是每个Agent维护自己的消息列表key比如research_messages、write_messages节点之间互不相干。汇总节点也不去读这些列表只读最后的结果字符串。这样设计虽然牺牲了一点信息传递效率但换来了调试时极大的清爽感。4.4 性能太慢怎么办多智能体一定比单Agent慢因为模型被调用了多次。如果你觉得Demo响应时间长优先看两点第一是不是本地模型跑在CPU上如果是换GPU或换个小一点的量化模型第二是不是子任务拆得太多控制在3个以内能显著提速。还有一个技巧是并行化模型支持并发的话把执行节点改造成线程池并发执行整体耗时能压缩到只比单任务慢一点。但入门阶段不推荐并行因为并发引入的竞态问题特别不好排查。5. 从Demo到真正可用的任务调度助手5.1 加一个简单记忆功能目前的Demo是无状态的用户每次提问都是全新上下文。实际使用时多轮对话很重要。LangGraph支持MemorySaver但那是给调度器存状态用的不是模型记忆。模型记忆需要你自己把历史对话传给模型。我这版Demo里在写Agent的系统提示词里插入了历史要点列表每次执行任务前把上几轮的关键结论传进去。注意不要传原始对话模型会分不清新旧任务传经过压缩的要点会更稳。5.2 接入团队协作场景这个调度助手稍加改造就能模拟一个“内容生产线”策划Agent出选题调研Agent找资料写作Agent出初稿校对Agent改病句。每个Agent可以属于不同租户或者独立服务LangGraph支持节点函数内部调用其他Agent服务。这样其实就变成了一个简单的“Agent工作流引擎”。我自己跑下来的经验是先把流程画在纸上再翻译成StateGraph顺序不要反。直接写代码的话很容易漏掉某条边。5.3 监控、容错与重试生产环境里Agent一定会失败。你可以给每个执行节点包一层try-except当某个子任务失败时把错误信息写入error字段条件边收到错误后走一个“修复节点”。修复节点的逻辑很简单让模型根据错误信息重写任务描述最多重试两次还不行就把任务标记为失败直接返回部分结果。实测这个机制能挽回大概三成意外报错——比如模型临时返回空内容重写任务描述后往往能正常执行。重试要限制次数否则像网络抖动那种错误会白白消耗大量时间。6. 写在最后的几点体会我自己做完这个Demo之后最大的感受是多智能体并没有想象中那么玄乎本质就是“把流程拆成节点让模型在每个节点做一件事”。难的不是LangGraph的API而是任务拆分的边界感。哪些事该让模型决策哪些事该用代码判定这个度需要反复调。建议你第一次跑的时候故意试几个模糊需求比如“帮我处理一下这个数据”看看规划节点怎么拆——它会提醒你模型不擅长处理信息不完整的请求所以调度器最好在入口做一个需求澄清的节点那又是下一个迭代了。最后分享一个小技巧如果你在论文或业务项目里要用多智能体不要一开始就上“多模型多工具多团队”的豪华配置先跑通一个三节点的最小闭环然后把你想加的功能逐个贴在边上测试。多智能体系统的问题绝大多数出在数据流上不在模型能力上。状态设计得清楚这个系统就成功了一大半。