企业官网建设流程全解析

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个阶段我分别命名为：任务规划、情报采集、数据分析、内容生成、质检修订。整体流程是流水线式的，但中间可以出现并行分支。下面这张表把每个阶段的核心职责说清楚：

阶段一的关键是“拆得开”。比如用户说“帮我调研一下新能源汽车充电桩的市场情况”，Planner不能直接开写，而是要把这个大问题拆成“市场规模、竞争格局、技术路线、政策环境、用户痛点”五个子问题。阶段二拿到子任务清单后，可以并行地去搜索资料（这里就是Send函数发挥作用的场景），每个子问题对应一个Worker任务。阶段三把采集回来的资料做去重、归纳、提炼，形成结构化结论。阶段四基于分析结果写报告，阶段五做质检，如果发现事实错误或逻辑不通，就打回给阶段四重写，形成循环。

这个流水线的好处是：每一级的输入和输出都是结构化的，Agent之间不需要“自由聊天”，而是通过消息缓冲传递状态。出了错也知道是哪个阶段的问题，排查起来非常快。

1.3 为什么选LangGraph而不是纯LangChain

LangChain和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(model="gpt-4o-mini", temperature=0.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(model="gpt-4o-mini", temperature=0.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(model="gpt-4o", temperature=0.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(model="gpt-4o", temperature=0.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_revision=True。

reviewer_llm = ChatOpenAI(model="gpt-4o", temperature=0.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 问题速查表

多代理工作流第一次跑通之后，紧接着就是各种问题。我整理了这段时间里最常遇到的几个，按症状、原因、解决方案列了个表。

这些坑几乎每个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的职责边界上多花心思打磨。后续我还会继续拆解更复杂的多代理编排思路，比如分层级联、记忆共享、工具调用增强等场景，欢迎持续关注。