LangGraph多代理实战：5个AI阶段编排调研报告生成系统

LangGraph实战系列写到第47篇今天聊一个避不开的话题多代理Multi-Agent。单智能体能做的事越来越不够用了稍微复杂一点的需求拆给多个AI分工协作效果往往好得多。但多代理不是简单地把几个Agent塞进一个循环里跑难点在于任务怎么拆、结果怎么合、流程怎么控。这套东西想落地LangGraph是我目前用下来最顺手的工作流框架没有之一。这次我用一个“5个AI阶段”的设计方式把一个完整的调研报告生成系统拆成五个不同职责的AI代理规划、采集、分析、写作、审核。整个流程用LangGraph串起来既能顺序执行也能在某些阶段并行分发任务。文章会把整个设计思路、核心概念、完整代码、踩坑经验都摊开来讲适合已经写过几个LangGraphdemo、但对多代理编排还不太熟的朋友。内容偏实战代码部分建议直接跟着跑一遍。1. 先把“多代理5个AI阶段”这套设计讲明白1.1 多代理不是堆Agent是在搭流程很多新手接触多代理第一反应是“多搞几个Agent丢进去让它们自己聊”。结果跑出来流程不受控输出不稳定token还烧得飞快。我的经验是多代理的核心不是Agent多而是流程清晰。Agent只是完成某个环节的执行者真正决定系统上限的是状态怎么流转、数据怎么传递、异常怎么处理。为什么需要多代理因为单个Agent在长任务里会“变形”。让它既要检索资料、又要分析数据、还要写报告它很容易在前半段认真、后半段敷衍甚至把事实编错。分阶段处理之后每个Agent只负责一件事职责边界清楚了Prompt可以写得非常具体输出质量能明显提高。这也是“5个AI阶段”设计的出发点把一个大任务拆成五个有清晰边界的子任务每个子任务交给专门的Agent。1.2 5个AI阶段的具体分工和流转逻辑这5个阶段我分别命名为任务规划、情报采集、数据分析、内容生成、质检修订。整体流程是流水线式的但中间可以出现并行分支。下面这张表把每个阶段的核心职责说清楚阶段Agent角色核心职责输入输出阶段一Task Planner任务拆解与方案生成用户需求子任务清单阶段二Researcher情报采集与检索子任务清单原始资料列表阶段三Analyst数据整理与分析原始资料结构化分析结果阶段四Writer内容组织与撰写分析结果报告初稿阶段五Reviewer质量审核与修订报告初稿终稿或修订意见阶段一的关键是“拆得开”。比如用户说“帮我调研一下新能源汽车充电桩的市场情况”Planner不能直接开写而是要把这个大问题拆成“市场规模、竞争格局、技术路线、政策环境、用户痛点”五个子问题。阶段二拿到子任务清单后可以并行地去搜索资料这里就是Send函数发挥作用的场景每个子问题对应一个Worker任务。阶段三把采集回来的资料做去重、归纳、提炼形成结构化结论。阶段四基于分析结果写报告阶段五做质检如果发现事实错误或逻辑不通就打回给阶段四重写形成循环。这个流水线的好处是每一级的输入和输出都是结构化的Agent之间不需要“自由聊天”而是通过消息缓冲传递状态。出了错也知道是哪个阶段的问题排查起来非常快。1.3 为什么选LangGraph而不是纯LangChainLangChain和LangGraph都是同一个生态里的东西但定位完全不同。LangChain偏重的是与模型、工具、向量库的交互能力它给你的是封装好的组件LangGraph偏重的是流程编排把Agent的每一步执行定义成图上的节点和边精确控制状态如何更新、什么时候跳转、什么时候终止。LangGraph底层直接把LangChain的工具链继承了所以你可以在图里继续用LangChain的模型封装和工具调用只是把原来的线性链换成了图状的工作流。实际项目中纯LangChain有个痛点Agent的ReAct循环是隐式的什么时候调工具、调完工具怎么回到推理全靠Prompt和模型自觉。LangGraph则把这些变成显式的节点和边什么时候该调用工具、工具返回后状态怎么合并开发者全部可控。而且LangGraph里有Checkpointer机制能做持久化和人工干预这对生产环境太重要了。可以这么理解LangChain提供的是零件LangGraph提供的是装配线和流水线控制系统。2. 动手前需要搞清楚的LangGraph核心概念2.1 一张图的骨架State、Node、Edge用LangGraph之前必须先把三个基本元素搞清楚。State是整个图运行时的“共享内存”通常用一个TypedDict定义。所有节点都能读State需要更新时通过节点返回值做部分更新。这个设计类似React里的状态管理每个节点只维护自己关心的一部分字段其他字段保持不变。State字段还可以声明Reducer比如用operator.add实现列表累加这样多个并行节点就能把结果追加到同一个字段里不会互相覆盖。Node是图上的一个执行单元本质上就是一个函数。函数签名的标准形式是def node_name(state: MyState) - dict输入当前State返回需要更新的字段字典。节点内部可以调用LLM、执行代码、调用外部API想做什么都行。Edge用来定义节点之间的流转关系分为普通边和条件边。普通边就是“A执行完必定去B”条件边则是根据State里的某个字段判断“走哪条分支”。整个图一旦编译完成就可以像调用函数一样传入初始状态去执行。LangGraph会自动处理节点间的依赖关系能并行的节点会尽量并行这比手动写for循环去调Agent要优雅得多。2.2 Send函数的真实用法和适用场景Send函数是LangGraph里让人又爱又恨的东西。官方文档里它的作用是“动态创建并行分支”但很多人第一次看到都有点懵它和invoke里传参到底有什么区别我举个具体例子。阶段二要做情报采集Planner拆出了5个子任务。如果用普通写法节点函数里写一个for循环依次调用5次检索这5个检索是串行的耗时是5次调用之和。如果用Send可以在图的执行过程中动态生成5个分支把每个子任务分别派发给同一个Worker节点LangGraph会并行调度这5个分支。关键区别在于for循环是同一个节点内部顺序执行Send则是图层面上的并行展开所有分支共享同一个State的Reducer逻辑最后把结果累加回State。Send的用法格式是Send(node_name, payload)其中node_name是要派发到的目标节点名payload就是传给这个节点的状态快照也可以只传一个包含目标节点所需字段的dict。这段代码通常写在条件边对应的路由函数里根据当前State的内容动态生成一个Send列表LangGraph看到这个列表就会为列表中的每一项启动一个独立执行分支。实用性上凡是“一对多分发”的场景都可以优先考虑Send多路情报检索、多个子任务并行验证、批量数据清洗。不过要注意所有并行分支最终都会回到同一个汇合节点汇合节点收到的状态是各分支更新后的合并结果所以State字段的Reducer必须提前定义好否则会出现写过的问题。2.3 图的状态合并与条件路由多代理工作流里状态合并是最容易被忽略的坑。默认情况下同一个字段在多个节点里都被写入时后写的覆盖先写的。但在并行场景比如多个Researcher分别返回raw_materials这个默认行为显然不够用。解决办法是给字段显式声明Reducer最常用的是operator.add它会把所有返回值当作列表元素依次追加进去。条件路由则决定了整个图的“灵活度”。前面提到的审核阶段到底该直接结束还是打回重写这需要看Reviewer的输出。我通常会在State里定义一个needs_revision布尔字段然后写一个路由函数读取这个字段返回下一步节点名。LangGraph执行到条件边时会自动调用这个函数并根据返回值跳转。结合递归深度控制就能把“最多修订几次”这类约束做得非常清晰。3. 实战搭一个5阶段多代理调研报告工作流3.1 环境准备与模型接入我的环境是Python 3.10以上直接pip安装。核心依赖如下pip install langgraph langchain langchain-openai如果要用DeepSeek或者其他的兼容OpenAI接口的模型也只需要在ChatOpenAI里改base_url和api_key。我这里以OpenAI风格接口为例但代码不绑定具体厂商。另外建议在环境变量里配好API Key别硬编码到代码里不然代码一旦提交到仓库就麻烦了。有一点经验可以分享多代理系统每个节点都在调用模型如果用gpt-4这类高配模型token消耗会非常快。实测下来规划、分析、审核这几个节点对推理能力要求高可以用强一点的模型采集节点主要做信息抽取和上下文总结用便宜一点的模型性价比更高。所以我在代码里让每个Agent持有自己的模型客户端方便替换。3.2 定义State并初始化图整个工作流的状态对象State是这个系统的骨骼我贴一下用TypedDict定义的状态结构以及申明Reducer的方式from typing import TypedDict, List, Annotated import operator class ReportState(TypedDict): user_demand: str task_list: List[str] raw_materials: Annotated[List[dict], operator.add] analysis_result: List[dict] draft: str final_report: str needs_revision: bool revision_count: int字段的语义很简单user_demand是用户最初的输入task_list是Planner拆出来的子任务清单raw_materials是所有Researcher返回的检索材料用operator.add做Reducer这样四个并行检索节点的结果会追加到一个列表而不是互相覆盖analysis_result是Analyst的结构化分析结果draft是Writer生成的初稿final_report是终稿needs_revision和revision_count配合控制审核是否通过以及最多修订次数。Reducer用Annotated[List[dict], operator.add]声明等于告诉LangGraph一旦有多个节点更新这个字段就把返回值逐个追加到已有列表的尾部。这个写法如果不熟悉建议先跑一个小demo验证一下后面踩坑的概率会小很多。3.3 实现阶段一到阶段五的Agent节点每个Agent本质就是一个Python函数内部调用语言模型。我这里把Prompt写出来方便看清楚每个阶段的分工。阶段一任务规划。这个节点的目标是拆解任务输出结构化的JSON数组。为了让模型输出稳定我使用了with_structured_output直接解析成JSON列表。from langchain_openai import ChatOpenAI planner_llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) def planner_node(state: ReportState): demand state[user_demand] prompt f你是资深调研专家。请把以下需求拆解为5个具体的调研子问题。 要求每个子问题必须具体、可独立检索输出JSON数组元素为字符串。 需求{demand} 只输出JSON不要额外解释。 resp planner_llm.with_structured_output() # 简化把返回对象转成list task_list resp.invoke(prompt)[tasks] return {task_list: task_list}这里with_structured_output()需要配合Pydantic模型或者直接解析为了不让示例过重我省略了Schema定义实际项目里建议定义一个TaskList的Pydantic类来约束格式。阶段二情报采集。这个节点相对特殊因为它是被Send并行派发的。每个分支拿到的是task_list里的一个子任务。这里的关键是节点函数签名保持不变但Send传进来的state只包含你指定的字段所以节点的输入字段要设计成与主State兼容。from langchain_core.output_parsers import StrOutputParser researcher_llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) def researcher_node(state: ReportState): task state[current_task] prompt f根据以下子任务检索并整理你已知的知识给出最相关的3条资料。 每条资料包含标题、来源、核心观点、关键数据。 子任务{task} 输出格式JSON数组。 resp researcher_llm.invoke(prompt) parser StrOutputParser() json_str parser.invoke(resp) materials parse_json_list(json_str) return {raw_materials: [{task: task, materials: materials}]}注意返回的是raw_materials列表因为Reducer是operator.add所以每个并行分支返回的[{task: ..., materials: ...}]会被追加到总列表里。这样主State里的raw_materials就会累积所有子任务的检索结果。阶段三数据分析。这个节点读取raw_materials把所有资料汇总提炼成几个结构化观点。analyst_llm ChatOpenAI(modelgpt-4o, temperature0.2) def analyst_node(state: ReportState): materials state[raw_materials] prompt f以下是初步检索材料请你做以下工作 1. 去重 2. 提炼出5条核心结论 3. 指出相互矛盾的资料。 材料{materials} 输出格式JSON对象包含scores、key_findings、conflicts。 resp analyst_llm.invoke(prompt) result parse_json(resp.content) return {analysis_result: result}阶段四内容生成。Writer读取analysis_result输出报告初稿。writer_llm ChatOpenAI(modelgpt-4o, temperature0.4) def writer_node(state: ReportState): analysis state[analysis_result] prompt f根据以下分析结果写一份1500字以上的调研报告。 要求结构清晰、有标题、有结论建议。 分析结果{analysis} 直接输出报告正文。 resp writer_llm.invoke(prompt) return {draft: resp.content}阶段五质检修订。这个节点比较有意思它名为Reviewer其实内部可以走两条逻辑如果质量好直接写final_report如果质量不好返回修订意见并标记needs_revisionTrue。reviewer_llm ChatOpenAI(modelgpt-4o, temperature0.0) def reviewer_node(state: ReportState): draft state[draft] revision_count state.get(revision_count, 0) if revision_count 3: return {final_report: draft, needs_revision: False} prompt f请审核以下报告检查事实准确性、逻辑连贯性、内容完整性。 如果问题较多请输出REVISE如果基本合格请输出ACCEPT。 报告{draft} 只输出ACCEPT或REVISE。 resp reviewer_llm.invoke(prompt) if ACCEPT in resp.content: return {final_report: draft, needs_revision: False} return {needs_revision: True, revision_count: revision_count 1}这样把所有阶段都实现成节点函数下一步就是把它们连成图。3.4 用Send并行分派检索任务Send在这里的作用非常关键。它负责在Planner之后把task_list里的每个子任务动态分派给researcher_node。我写了一个路由函数放在Planner到Researcher的条件边里。from langgraph.constants import Send def dispatch_researchers(state: ReportState): return [Send(researcher_node, {current_task: task}) for task in state[task_list]]这个函数的返回值不是字符串而是一个Send对象列表。LangGraph看到这个列表就会为每个Send对象启动一个独立的执行分支目标节点是researcher_node传入的状态是{current_task: task}这个局部状态。需要注意一点Send的payload虽然是局部状态但目标节点返回的字段会通过Reducer合并回主State。这里之所以能顺利累加到raw_materials正是因为它声明了operator.add。如果忘记写Reducer并行分支返回的新值会把之前的值整个覆盖掉结果只剩最后完成的那个分支的数据。这个设计的威力在于无论用户提了5个子任务还是20个子任务代码都不用改只是并行分支数量不同。相比串行循环响应时间可以压缩数倍。3.5 条件路由控制审核与修订审核阶段需要“要么打回重写要么直接结束”的分支逻辑这个场景用条件边实现。我在图里定义了一条从writer_node到reviewer_node的边再从reviewer_node连出一条条件边。def route_after_review(state: ReportState): if state.get(needs_revision): return writer_node return END当needs_revision为True时流程跳回writer_node重新写稿然后再次走审核。这里就需要防止无限循环。我在reviewer_node里加了revision_count的判断最多修订3次超过就直接接受初稿。这种“最大重试次数状态计数”的写法在生产环境几乎是标配。3.6 完整代码与运行效果把上面的函数串成图核心代码如下from langgraph.graph import StateGraph, START, END builder StateGraph(ReportState) builder.add_node(planner_node, planner_node) builder.add_node(researcher_node, researcher_node) builder.add_node(analyst_node, analyst_node) builder.add_node(writer_node, writer_node) builder.add_node(reviewer_node, reviewer_node) builder.add_edge(START, planner_node) builder.add_conditional_edges(planner_node, dispatch_researchers, [researcher_node]) builder.add_edge(researcher_node, analyst_node) builder.add_edge(analyst_node, writer_node) builder.add_edge(writer_node, reviewer_node) builder.add_conditional_edges(reviewer_node, route_after_review, {writer_node: writer_node, END: END}) graph builder.compile() result graph.invoke({user_demand: 调研一下新能源汽车充电桩的市场情况}) print(result[final_report])这里有一个细节add_conditional_edges的第三个参数需要映射路由函数返回值和实际节点名。返回END时图会终止返回writer_node时会跳回重写节点。LangGraph从0.2.x版本开始推荐用END代替之前版本的None来标记终止建议跟着新版本API走。第一次跑这个流程你会发现输出顺序不是固定的Planner先执行然后Researcher多个分支几乎同时执行接下来Analyst、Writer、Reviewer依次执行。如果打回重写Reviewer执行完之后会再出现一次Writer和Reviewer的执行轨迹。这种“可看见的执行路径”就是LangGraph相比普通Agent循环的最大优势每一步做了什么、用了哪份状态全都一目了然。4. 常见问题与排查技巧实录4.1 问题速查表多代理工作流第一次跑通之后紧接着就是各种问题。我整理了这段时间里最常遇到的几个按症状、原因、解决方案列了个表。症状常见原因解决方案并行分支返回的结果互相覆盖State字段未声明Reducer给字段加Annotated[list, operator.add]Send派发后目标节点拿不到字段Payload字段与节点函数使用的key不一致检查Send传入的payload键名是否与节点读取的键名完全一致审核后无限循环缺少修订次数上限在Reviewer节点维护revision_count超过阈值强制终止模型返回的JSON解析失败结构化输出没有强约束用with_structured_output配合Pydantic或增加解析兜底逻辑整个流程非常慢每个节点都用同一个高配模型区分推理型节点和抽取型节点用不同模型并行分支尽量用Send展开状态里传递的消息历史越来越长没有裁剪历史记录对超过轮数的历史做截断或只保留最终结论字段这些坑几乎每个LangGraph项目都会踩一遍。我印象最深的是Reducer问题看起来只是加一个operator.add的事但如果不理解它的机制排查起来会以为是自己并行逻辑写错了白白折腾大半天。4.2 我对Send函数踩过的坑作为LangGraph里最让人困惑的功能Send函数的坑值得单独拿出来讲。第一个坑是把Send当成了异步调用。很多人在路由函数里写Send(node, state)以为会像async那样立刻返回一个Future。实际上Send只是在构图时告诉运行时“这里要动态展开N个分支”真正的并行调度由LangGraph执行器完成。你不需要也不能手动去等待这些分支结束只要定义好汇合节点执行器会等所有分支完成后再继续。第二个坑是Send的payload设计。我曾经把整个主State都作为payload传给目标节点结果并行分支之间互相干扰因为每个分支拿到的是同一份State的引用某个分支修改了某个字段其他分支读到的值就变了。正确做法是payload只传目标节点需要的最小字段集比如{current_task: task}返回时再把增量字段合并回主State。这既是性能优化也是避免并发冲突的关键。第三个坑和Reducer有关。并行分支一旦都往同一个字段写数据如果不加operator.add状态合并时就是“后写覆盖先写”最后存下来的只有最后一个分支的结果。这个设计确实反直觉我刚接触时也是跑了一次发现数据只剩一条才回头去查文档。现在我的习惯是凡是会被并行写入的字段一律声明Reducer。4.3 结果质量不稳的兜底方案多代理系统最难缠的问题不是报错而是“跑通了但结果质量不稳定”。五个Agent串起来任何一个Prompt写得不够具体最终报告的质量就会被放大。我试过几种兜底方案组合起来效果比较明显。第一给每个Agent设定“输出护栏”。比如Researcher返回的资料必须包含“来源”Analyst必须返回“关键发现”Reviewer必须给出明确的通过/打回标记。宁可让模型多输出几个字段也不让它自由发挥。结构化输出配合Pydantic Schema能把不稳定率降一大截。第二审核阶段增加一个“关键数据抽查”环节。Reviewer不能只凭感觉看流畅度我会要求它针对报告里的数字和引用做核对。比如报告中出现“2025年市场规模达到XXX亿元”Reviewer就要检查这个数字是否在原始材料里有出处。如果回答“无法溯源”就得打回。实测这个方式能有效减少幻觉数据对外传递。第三在业务层面保留人工介入点。金融、医疗这类严肃场景建议用LangGraph的Checkpointer配合interrupt_before在最终报告生成前暂停图执行等待人工确认。多代理可以把80%的活干完最后20%的质量门槛还是需要人拍板。这就是可控性的价值也是LangGraph这类编排框架存在的意义。就我个人这几套实战下来还是那句老话多代理系统的上限由Prompt设计决定下限由流程控制兜住。LangGraph能把流程这个“下限”拉得很高剩下的就是在每个Agent的职责边界上多花心思打磨。后续我还会继续拆解更复杂的多代理编排思路比如分层级联、记忆共享、工具调用增强等场景欢迎持续关注。