LangGraph多智能体实战：构建研究-分析-写作-审核自动化流程

最近我在做一个自动化行业分析报告的项目核心是用 LangGraph 编排一组智能体让它们像一个小团队一样协同完成研究-分析-写作-审核的完整闭环。这个工作流最开始是我为了应付大量同质化报告而想出来的后来发现它完全可以迁移到方案撰写、竞品分析、技术文档沉淀甚至课程内容生产上。今天就把这套多智能体系统的设计思路、核心代码和调试经验完整记录下来给正在学 LangGraph 或者准备用多智能体重构业务流程的朋友一个参考。如果你是第一次接触 LangGraph我可以先说结论它本质上是一个给 LLM 应用用的状态机 图编排框架。你可以把流程画成一张有向图每个节点就是一个智能体动作比如搜索、推理、写文章节点之间通过状态传递数据边则决定了下一步去哪。相比单纯用 prompt 堆复杂任务研究-分析-写作-审核这种多智能体协作模式最大的优势是把一项复杂任务拆成了多个可验证的环节每个环节的职责单一、输出干净出了问题也能精准定位。接下来的内容会围绕这四个角色的设计展开包括角色 prompt、节点函数、状态定义、流程控制以及一些真实运行中会遇到的问题。1. 为什么需要研究-分析-写作-审核这样的多智能体团队1.1 单一模型跑完整流程的瓶颈在哪先讲个很常见的操作在 LibreChat、FastGPT 这类平台上大家倾向于一个会话里让 AI 完成所有事prompt 里写请根据我的主题先研究一下市场情况然后分析竞品最后写一份报告。作为一次性问答这没问题但放到生产环境里就会暴露出三个硬伤。第一token 上下文是有限的。一篇正经的行业分析报告可能需要看几十篇网页、总结大量数据如果所有原始资料都塞进同一个上下文还没到写正文阶段上下文就已经爆了。第二不同环节的 prompt 要求是不同的混在一个系统 prompt 里会互相干扰。比如做研究这个环节希望模型尽可能多地抓取信息不要过早下结论但写文章环节正好相反要收敛、要批判、要有明确观点。这两种指令混在一起模型很容易处于既要又要的拧巴状态。第三也是最实际的就是调试成本。当输出质量不行你根本不知道是哪个环节出错了——是资料没抓到还是分析跑偏了只能盲猜。多智能体团队解决的正是这三个问题。每个智能体只负责一个环节上下文里只需要装入它该处理的数据prompt 也干净。研究智能体拿到的是用户问题 搜索能力分析智能体拿到的是研究原始素材 分析指令写作智能体拿到的是分析结论 写作指令。这样每条链路清晰每个节点的输入输出都可检查出问题直接定位到对应节点。1.2 LangGraph 选型理由它凭什么能撑起团队协作选 LangGraph 而不是自己用 Python 写一个 while 循环调 API或者用简单的 LangChain 链是经过对比的。已经有 LangChain 经验的人上手 LangGraph 代价不高真正让我留下的是它对可控流程的支持。我的需求里有两个硬指标一是流程必须能循环。审核不过就要打回给写作写作改完再送审可能要来回几次。LangChain 的链式调用是线性 flow要实现循环得自己写状态管理比较别扭。二是需要人工介入。研究完一轮之后我想让人看一眼素材够不够不够就让研究员继续搜。LangGraph 提供的interrupt机制可以在图运行到某个节点前停下来把控制权交给人人决定继续还是修改输入。这两点在 LangGraph 里都是原生能力不用自己造轮子。还有一个很实际的原因LangGraph 的图结构非常直观。我可以在 notebook 里用graph.draw_mermaid()把流程图画出来给团队或客户看的时候不用解释这是一个多智能体系统只用说你看先是研究员干活然后分析师再然后写作、审核审核不过就返工。这种可视化能力在项目沟通里省了太多口水。注意如果你只是做一次性生成、对流程没有分叉和循环需求那直接用普通 LangChain 链就够引入 LangGraph 反而增加复杂度。多智能体编排不是银弹它适合的是流程自身有纠错、有分支、有状态累积的场景。我建议先画出流程数数有没有环有环再考虑图编排。2. 团队角色与整体状态设计2.1 四个角色的职责边界与协作契约搭建团队之前首先要定义清楚每个角色干什么、不干什么、输出什么。边界模糊是多智能体系统的大忌如果研究智能体忍不住写结论写作智能体又忍不住做分析整个系统会退化成一个巨大的、失控的聊天室。我把流程拆成了这样研究员Researcher输入任务主题输出一组检索到的资料摘要。它需要调用搜索工具把网页内容按标题、来源、原文要点、数据点整理成结构化列表。它不需要判断这些资料够不够也不需要给出观点。分析师Analyst输入研究素材输出主题分析结果包括核心结论、关键论据、反对观点、数据图表建议。它要做的是把零碎信息变成逻辑骨架。写作员Writer输入分析结果输出完整文章初稿。它负责把骨架填充成可读的文字注意表达、节奏、引用标注但不要引入新观点。审核员Reviewer输入初稿和分析结果输出审核结论。它检查逻辑是否连贯、论据是否充分、有没有明显事实错误、格式是否达标最后给出通过或打回意见。每个角色的交互契约就是输入字段 输出字段。我建议在状态定义里为每个角色设计独立的输入输出字段而不是让它们全部读写同一个大字段否则容易踩到相互覆盖的坑。这个过程有点像团队里的接口设计字段名就是接口签名一旦约定好各个节点的改动可以互相隔离。2.2 用 StateSchema 管理协作的白板在 LangGraph 里状态State就是整个团队共用的白板所有节点写的产出都累积在这块白板上。代码里我们通常定义一个TypedDict来描述白板结构。我用的字段是这样划分的from typing import TypedDict, Annotated, List, Optional import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver class ResearchItem(TypedDict): title: str url: str content: str key_points: List[str] class TeamState(TypedDict): task: str research_items: Annotated[List[ResearchItem], operator.add] analysis: Optional[str] outline: Optional[str] draft: Optional[str] review_result: Optional[dict] iteration: Annotated[int, operator.add]细说一下字段设计的思考。task是团队唯一不变的输入用户可以反复修改它并重新触发。research_items使用了Annotated[List[ResearchItem], operator.add]这是 LangGraph 里非常实用的小技巧——operator.add会把多个 reducer 追加的结果合并而不是覆盖。也就是说如果研究员节点被调用了两次比如第一次搜索不够第二次补充搜索新结果会追加到旧结果后面不会把第一次搜到的素材清空。如果你的系统里不需要跨节点累积数据用普通字段直接覆盖就行。iteration我用它记录重写的轮数。之所以单独加一个累积整数是因为审核不通过就回到写作节点重写这个循环可能重复很多次。如果没有轮数上限模型有可能无限循环下去烧掉你账户里大量 token 后输出一堆越来越差的内容。我后面会讲到怎么用它做熔断。实战心得状态字段不要舍不得定义。很多人图省事用一个messages字段传递所有内容结果就是调试时你根本不知道当前白板上到底是分析结果还是草稿。分字段存储每个节点只读自己关心的字段、只写自己负责的字段这是整个系统不混乱的根基。3. 把流程编译成图核心节点实现思路3.1 研究员节点调用搜索工具收集素材研节点的核心是工具调用 摘要整理。我的做法是给研究员绑定一个搜索工具如果你用的是 OpenAI 风格接口就是 function calling如果你用通用模型可以用tool参数传入工具 schema。整个节点的大致结构如下from langchain_openai import ChatOpenAI from langchain_core.tools import Tool def search_engine(query: str) - str: 调用搜索服务返回前几条结果的摘要文本。 # 这里替换成你自己的搜索 API / 数据库查询 return results_text research_llm ChatOpenAI(modelgpt-4o, temperature0.2) research_llm_with_tools research_llm.bind_tools([search_engine]) def research_node(state: TeamState) - dict: task state[task] # 构造给研究员模型的指令 prompt f你是研究助理。你的任务是围绕主题找到有信息量、有可信度的资料。 请调用search_engine搜索相关内容并输出一个JSON列表每个元素包含title、url、content、key_points。 content是对原文的压缩摘要不要保留HTML标签key_points最多3条。 主题{task} resp research_llm_with_tools.invoke(prompt) # 解析模型返回的JSON封装成ResearchItem new_items parse_items(resp.content) return {research_items: new_items}这里有一个值得琢磨的点研究员节点要不要用记忆我的选择是不用。这个节点每次运行时系统 prompt 只包含任务和少量指令不携带之前对话历史。原因是搜索任务是无状态的每次搜索都是独立的强制模型记住历史反而容易让它漏搜。如果你做的是需要连续追踪某个话题的消息流式分析那另当别论。另外关于搜索工具的数量和质量我建议一开始用一个别让模型自己挑一堆工具。搜索这个动作是最容易被反复调用的如果给的搜索 tool 有多个比如网页搜索、新闻搜索、学术搜索模型就会陷入选择困难输出也不稳定。只给一个 search_engine让它内部自己去决定关键词和筛选方式。3.2 分析师节点把素材提炼成大纲和论据研究节点跑完state[research_items]里可能有十几条素材。这时候如果把全部素材原样丢给分析师会非常考验 token 规模而且分析师输出的质量跟素材结构关系很大。所以我一般会在进入分析节点前先在代码里做一次素材预筛把素材按点击数、来源权威度、时效性排序截取 Top N 条送到分析师那里。分析节点 prompt 我写得比较克制重点明确要它输出结论、论据、反驳点、缺失信息四块。因为后续写作员要根据这个结构来写文章所以分析结果必须结构强而不能是一团自然语言。analysis_llm ChatOpenAI(modelgpt-4o, temperature0.4) def analysis_node(state: TeamState) - dict: tasks 你的任务\n tasks - 提取3~5个核心结论\n tasks - 每个结论给出至少2条支撑论据论据必须引用研究素材中的具体信息\n tasks - 针对每个结论给出1个可能的反方观点\n tasks - 明确指出当前素材中缺失的、会影响报告质量的信息\n tasks - 输出JSON格式{core_conclusion, evidences, counter_arguments, missing_info} snippets \n\n.join( f[{i1}]{item[title]}\n{item[content]} for i, item in state[research_items] ) resp analysis_llm.invoke(f{tasks}\n\n素材\n{snippets}) analysis json.loads(resp.content) # 这里还应该根据 analysis 自动生成大纲 outline_prompt f根据以下分析结果生成一篇报告的二级大纲{analysis} outline outline_llm.invoke(outline_prompt).content return {analysis: analysis, outline: outline}注意我在同一个节点里干了分析 生成大纲两件紧耦合的事。有人会问为什么不拆成两个节点我的理由是大纲完全依赖分析结论如果分两个节点状态里就得多传一个中间字段没什么收益反而增加状态复杂度。节点粒度不是越细越好你自己画流程的时候想清楚哪些动作可以原子化就行。站在实战角度我更推荐把分析这一节点的输出做严格校验。比如检查missing_info是否有内容如果有可以走一条条件边回到研究员节点让它针对缺失信息再搜一轮。这种研究-分析-补充研究的循环可以明显提高最终报告的质量代价就是多几次工具调用。3.3 写作员节点基于大纲生成成稿写作员节点看起来最简单其实最容易翻车。很多人的 prompt 里写着请根据大纲写一篇8000字的文章然后模型就输出一篇语焉不详、全是空话的文章。写作员真正要的是素材 大纲 风格要求三者缺一不可。我的写作节点 prompt 结构指明角色你是一位资深行业分析师写作风格偏理性、克制、用数据说话。给出写作要求字数范围、章节结构、是否需要小标题、引用方式。给出素材来自state[research_items]的前 N 条和state[analysis]。给出大纲来自state[outline]。要求输出纯 Markdown 格式且必须严格遵循大纲的层级顺序。writer_llm ChatOpenAI(modelgpt-4o, temperature0.7) def writing_node(state: TeamState) - dict: material \n.join( f【{item[title]}】{item[content]} for item in state[research_items][:5] ) prompt f 请根据下面的材料和分析结论遵循大纲写一篇结构完整的行业分析报告。 - 风格客观、简洁多使用具体的数字和事实避免空洞的形容。 - 结构严格按大纲的二级标题展开每个小节不少于300字。 - 引用文中引用来源时用【来源N】标注。 - 输出完整 Markdown 正文不要额外说明。 材料 {material} 分析结论 {state[analysis]} 大纲 {state[outline]} draft writer_llm.invoke(prompt).content return {draft: draft, iteration: 1}这里用iteration: 1表示这是第几轮重写。注意因为我在状态定义里给iteration加了operator.add返回 1 表示这一轮新增了一轮具体累计值由框架负责增加。如果你不希望模型参与第几轮重写的逻辑就完全可以在节点里写死返回 1框架会自动把它累加到通行状态里。写草稿时的体验是温度稍微调高一点0.7会让表达更自然。但由于写作环节涉及事实引用温度太高又可能导致模型把素材里的数据编得更顺滑——这是我的一个教训后来我要求模型在引用数据时必须优先使用材料里的原数值如果材料中没有该数据必须明确写数据缺失而不是自己编一个。3.4 审核员节点质量把关与条件跳转审核员是团队里的质量守门员它直接决定系统是顺利收尾还是进入返工循环。我设计的审核结果分三层pass、minor_issues、major_issues。只分两级很容易让模型偷懒全都不通过或全都通过三级则能让流程更自然地分流。reviewer_llm ChatOpenAI(modelgpt-4o, temperature0.1) def review_node(state: TeamState) - dict: review_prompt f 你是责任编辑。请对以下草稿进行严格审核。 重点检查 1. 结论是否有材料支撑材料中是否有直接依据 2. 逻辑是否连贯论据是否充足 3. 格式是否符合要求二级标题、引用标注 4. 是否有明显事实错误或猜测性内容。 请输出JSON {{verdict: pass | minor_issues | major_issues, comments: 给写作员的修改意见用中文}} 草稿 {state[draft]} review json.loads(reviewer_llm.invoke(review_prompt).content) return {review_result: review}审核节点返回时我不直接修改state[draft]而是把review_result写入状态。下一步怎么走交给条件边决定。这让审核逻辑和流程控制分离之后就算要增加人工审核、修改打回逻辑也只需要改边不需要动审核节点本身。写到这里你可以发现整个团队的核心就是把复杂任务拆解成四个可测试的原子操作。每一个节点都可以单独喂测试数据跑一次观察输出是否符合预期然后再把它们串成图。这个单节点可验证的特质是我选择多智能体方案最重要的理由之一。4. 流程控制串行、并行与人工审核4.1 条件边的写法与触发逻辑有了节点接下来就需要把它们连接起来。LangGraph 的边有三种普通边、条件边、循环边。我这条流程的主干是一条普通边从研究到分析到写作再到审核。但从审核之后就开始分支了。def should_go_to_end(state: TeamState) - str: verdict state[review_result][verdict] if verdict pass: return accepted return revise graph StateGraph(TeamState) graph.add_node(researcher, research_node) graph.add_node(analyst, analysis_node) graph.add_node(writer, writing_node) graph.add_node(reviewer, review_node) graph.add_edge(START, researcher) graph.add_edge(researcher, analyst) graph.add_edge(analyst, writer) graph.add_edge(writer, reviewer) graph.add_conditional_edges( reviewer, should_go_to_end, { accepted: END, revise: writer, }, )这段代码里最关键的就是add_conditional_edges。它接收三个参数源节点、条件函数、映射字典。条件函数读取当前状态里的某个字段返回一个字符串 key框架根据映射字典决定去哪个节点。我在这里把minor_issues和major_issues都映射到writer——它们区别只是修改量不同但都要回到写作节点重写。如果你想控制更细致比如minor_issues走一个轻量修补节点major_issues走重新分析节点那在字典里给不同 key 配不同目标即可。一个特别容易踩的坑是条件函数返回的 key 必须和映射字典里的 key 严格匹配多了少了都会在运行时抛 KeyError。我在调试阶段就因为 key 大小写不一致卡了十分钟。建议把条件函数返回值定义成枚举或常量避免手写字符串带来的低级错误。4.2 人工审核打断与恢复虽然我们有审核员但在严肃场景下完全把通过的决定权交给模型仍然不让人放心。我的方案是加一个人工审核环节放在 LLM 审核之后、最终输出之前。人工审核使用 LangGraph 的interrupt机制。LangGraph 的interrupt是这样工作的编译图的时候你可以指定在某一个节点前中断运行时图执行到那个节点之前会暂停把控制权交还给主程序。主程序可以把状态比如草稿、审核意见展示给用户等用户给出指令通过/修改意见再继续运行图。用代码说更直白from langgraph.types import interrupt, Command def human_review_node(state: TeamState) - dict: # 主动暂停等待外部指令 feedback interrupt({ draft: state[draft], review_result: state[review_result], question: 请人工确认草稿是否通过或输入修改意见。 }) # feedback 是外部传入的dict形如 {action: approve} 或 {action: revise, comments: ...} return { draft: state[draft], review_result: state[review_result], }然后把human_review_node接到条件边之后graph.add_node(human_review, human_review_node) graph.add_conditional_edges( reviewer, should_go_to_end, { accepted: human_review, revise: writer, }, ) graph.add_edge(human_review, END)但要注意这个写法只在图运行在支持checkpointer的模式下才有效。interrupt依赖检查点机制必须在编译时传入MemorySaver或数据库检查点否则会出现暂停后恢复时状态丢失的问题。checkpointer MemorySaver() app graph.compile(checkpointercheckpointer, interrupt_before[human_review])使用app.stream()或app.invoke()时如果图在human_review节点前被打断返回值里会包含一个__interrupt__字段。你拿到它后把草稿展示给用户让用户填写意见再通过app.invoke(Command(resume{action: approve}), config)让图继续跑。这个过程在真实产品里可以对接一个前端审批界面。这里你可以看到 LangGraph 的多智能体编排能力有多么贴近真实业务它允许自动系统与人工审批混流而不是一锤子跑到底。我后面的生产系统里所有对外发布的内容都会经过这一道人工审核机器负责出稿人负责拍板。4.3 并行节点什么时候用不用多智能体往往让人联想到同时干活。我在这个项目里其实没有使用并行因为每个环节都依赖前一个环节的输出先有研究素材才能分析先有分析才能写作。唯一可以考虑并行的场景是研究节点内部分多个搜索 agent 同时搜不同子主题但那已经是对单个节点内的优化了。LangGraph 也提供了发送到多个节点的并行机制比如create_branch或者一个节点映射到多个目标但并行会引入状态合并的复杂度。如果你的任务里几个智能体互不依赖再考虑并行否则请老老实实按依赖顺序串行执行。多智能体的价值是分工清晰而不是并发快。5. 实战中的坑与排查经验5.1 状态字段污染与消息累积这个坑几乎是每个 LaangGraph 新手都会踩的。因为 LangGraph 的状态机制里messages字段用operator.add累积时会把所有历史消息堆在一个数组里每次节点调用都往数组里塞入新的 AIMessage、ToolMessage。久而久之白板里积累了大量无用消息导致后续节点看到的上下文越来越长响应越来越慢甚至把重要字段淹没。我的建议是如果节点的输入输出不需要保留历史对话就不要在状态里定义messages累积字段。让每个节点直接读取自己关心的业务字段。比如研究员节点只需要task和上一轮missing_info那就只读这两个字段别的字段一概不管。万一你又确实需要记忆比如让写作员记住上一轮审核意见那么可以单独定义revision_comments: Optional[str]字段每次审核节点返回时覆盖该字段。这样写作节点在重写时拿到的是最新的修改意见不会把两轮意见混在一起。5.2 token 超限与上下文裁剪多智能体系统最大的消耗来自节点之间的数据传递。我实测过当研究素材有 15 条、每条摘要 1000 字时分析师节点的输入筐就会接近 2 万 token这还不算任务指令本身。如果再用 gpt-4o 这类模型成本会直线上升。解决办法有三个控制上游产出研究员节点在返回research_items前先对每条素材做要点压缩只保留关键数据和结论砍掉过程性描述。控制下游输入分析师节点可以做一个take_top_k的函数根据state[research_items]的长度只取前 6 条素材作为输入。设置每步 token 上限如果你用的是可控模型在 LLM 初始化时给max_tokens设置一个合理值。比如写作节点设置max_tokens4000至少保证长文能输出完分析师节点设置max_tokens2000防止模型输出超长 JSON 导致解析失败。注意LangGraph 本身不会自动帮你裁剪状态。编译图时你可以写一个 util 预处理节点把所有会传给大模型的字段显式截断、清理后再喂进去。千万不要把原始状态直接塞给模型。5.3 循环死锁与最大步数设置审核不过-重写循环是个好东西但会给图引入环环就意味着可能死循环。我在测试阶段就碰到过一次审核员死活不通过写作员改了几轮还是一样的问题最后图一直在writer - reviewer - writer之间打转直到我把 max iterations 耗尽。LangGraph 的recursion_limit是自救的好帮手。在调用app.invoke时传入config{recursion_limit: 15}框架会在总步数超过 15 时抛出GraphRecursionError强制停止循环。但只有这个还不够你需要在业务层面也做控制。我的实现是在审核节点里检查state[iteration]。如果iteration 3那就改成 verdict 直接为pass或者返回major_issues并且要求下一步进入人工评审。这样即使自动审核一直不满意系统也会在第三轮强制收尾然后由人来决定是否继续。代码并不复杂def review_node(state: TeamState) - dict: # 其他审核逻辑 if state[iteration] 3: return { review_result: { verdict: pass, comments: f已达到最大轮次{state[iteration]}自动通过并转人工审核。 } } return {review_result: review}这种做法比纯粹依赖recursion_limit更安全因为它不是粗暴地中断整个流程而是给流程一个体面的出口。5.4 条件函数返回不一致引发的 KeyError前面提到过条件函数返回的 key 必须和映射字典的 key 对应。还有一个更隐蔽的问题是条件函数在运行时会读取state[review_result]但这个字段可能压根不存在比如你在测试时直接用graph.invoke({task: xxx})没有走完前面的节点那么审核节点之前的条件边就会拿不到review_result字段。解决办法是在条件函数里使用state.get(review_result)并给一个默认值def should_go_to_end(state: TeamState) - str: result state.get(review_result) or {verdict: pass} return accepted if result[verdict] pass else revise这样即使状态缺失条件函数也能优先流向安全的分支不至于直接抛异常。调试流程的时候也可以用这种方式快速跳过无用节点方便配合 LangGraph 的 LangSmith 追踪查看流程时序。最后再分享一个小技巧如果你的图逻辑越来越复杂建议启用 LangGraph 的调试模式。每一步节点执行前后状态的变化、LLM 的输出、token 消耗量都会被记录到调用链上几乎可以还原现场。这套研究-分析-写作-审核团队已经在我多个项目里跑了几百个真实任务最大的体会就是多智能体编排的重点不是让智能体有意识地协作而是用清晰的流程边界和严格的状态契约把每个人的能力限制在合适的范围内。模型还是那个模型换一种组织方式输出质量就能差出一个量级。如果你想让自己手头的单轮生成任务升级成可控制的自动化流水线完全可以沿着这套思路先画图再定状态最后用 LangGraph 把它们接起来。