企业官网建设流程全解析

最近一年多，我发现一个现象：凡是聊到复杂LLM应用，开口闭口都是Multi-Agent，但真正能把多个Agent有效编排跑起来、稳定输出结果的案例并不多。LangGraph作为基于图结构的Agent编排框架，是目前我在实际项目里用得最多的一套方案。这篇文章不打算重复官方文档里的入门Demo，而是用一个可落地的完整案例——一个三Agent协作的研究报告生成系统，把LangGraph Multi-Agent从设计、编码到排错的全过程拆开讲清楚。适合已经写过单Agent调用、想升级到多Agent协作的开发者，也适合正在做技术选型的读者作个参考。

1. 为什么需要Multi-Agent：单Agent的瓶颈从哪来

1.1 单Agent在高复杂度任务下的典型困境

单Agent本质上是一个"大而全"的执行者：给它一个目标，它自己去调工具、读上下文、生成结果。在简单场景下这种模式很舒服，比如只做翻译、只做摘要、只做单轮问答，效果都不差。但一旦任务复杂起来，问题就变得很明显：

• 上下文互相污染：检索、分析、写作全压在一个上下文窗口里，工具返回结果多了之后，Agent容易"忘记"最初目标，生成的结论偏离用户需求。
• 提示词互相干扰：同一个Agent既要做检索规划，又要做数据解读，还要控制输出格式，提示词越长，模型越容易顾此失彼。我见过不少项目把单个Prompt堆到三四千字之后，Agent反而开始频繁出错。
• 无法并行：单Agent天然串行，如果任务里有多个可以同时处理的子任务（比如同时查多个数据源），它也只能一个一个来，整体耗时成倍上涨。
• 错误难以定位：一个环节出错整个会话就失败，你很难判断是检索环节的问题、还是推理环节的问题，排错成本很高。

1.2 拆成多Agent为什么能缓解这些困境

Multi-Agent的思路是"拆"。把一个大任务按职责拆成若干个子Agent，每个Agent只负责一个相对聚焦的环节，通过图结构把这些环节组织起来，共享一份状态、有序或并行地推进。这样每个Agent的上下文窗口里只装自己关心的内容，Prompt短了，模型负担小了，出错率自然降下来；并行分支让耗时也能压下去。

这里要先泼一盆冷水：多Agent不是银弹，拆得不好反而比单Agent更慢更乱。第6节我会专门说什么时候不该用。但当你确实遇到前面这些情况时，LangGraph这套编排框架就是我目前最顺手的选择。

1.3 为什么选LangGraph而不是手写编排

我选择LangGraph而不是纯LangChain顺序链或者手写循环，原因有三点：

1. 它有显式的图结构，节点和边是声明式定义的，整个执行流程一眼能看懂，排错比黑盒循环清晰得多。
2. 状态管理是内置的，所有Agent共享一份state，不需要自己维护一堆全局变量或者反复传参。
3. 支持条件分支、并行发送、人机交互中断恢复等高级控制流，扩展性比固定链式调用强很多。

我实际用下来，LangGraph最大的价值在于：它把多Agent系统从"思路"变成了"图纸"。你先把图设计出来，再往节点里填Agent逻辑，整个系统的行为是可预期的。

2. LangGraph的核心概念：StateGraph、节点与共享状态

在进入案例之前，先把LangGraph里最关键的几个概念用"人话"说一遍。我见过太多人卡在概念上，其实它背后的思想很朴素，就是一个有向图在跑一个状态机。

2.1 State：所有Agent共享的"工作台"

LangGraph里，State是贯穿整个图执行过程的数据对象。你可以把它理解成所有Agent共用的"工作台"：每个Agent执行完后把产出写到工作台上，下一个Agent从工作台读取自己需要的东西。

State一般用一个TypedDict或dataclass定义，比如：

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): user_query: str search_results: list[dict] analysis_result: dict final_report: str messages: Annotated[list, add_messages]

里面有两个值得注意的细节。第一，Annotated[list, add_messages]是LangGraph里很常用的reducer写法，表示当多个节点往messages字段写入内容时不是覆盖，而是累加合并。这个机制对多Agent场景至关重要，后面讲并行节点时会再碰到。第二，除了messages之外，其余字段默认是"后写覆盖"，也就是同一轮里多节点都往search_results写的话，最后一次写入会覆盖前面的，这也是新手最容易踩的坑之一。

2.2 Node与Edge：谁干活、谁传给谁

Node就是图里的节点，一个节点通常就是一个函数，输入整个State，输出State的局部更新。Edge是节点之间的连接关系，决定执行顺序。

from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("search", search_agent) builder.add_node("analyze", analyze_agent) builder.add_node("write", write_agent) builder.add_edge(START, "search") builder.add_edge("search", "analyze") builder.add_edge("analyze", "write") builder.add_edge("write", END) graph = builder.compile()

这段代码就是一个最基础的三节点流水线。LangGraph执行时，会按图的拓扑顺序把整个图跑完：当前节点返回后，框架根据State的更新结果和边定义自动决定下一个节点。编译出的Graph对象可以直接invoke调用，也可以stream逐步观察。

2.3 条件边与Send API：图不是死板的流水线

纯顺序执行其实用普通函数链也能做，LangGraph真正的优势在条件边（conditional_edges）和Send API。

条件边解决"结果不同走不同支路"的问题，比如分析后认为数据不足，就返回"search"节点补搜一轮：

def route_after_analysis(state): if state.get("needs_more_search"): return "search" return "write" builder.add_conditional_edges( "analyze", route_after_analysis, {"search": "search", "write": "write"} )

Send API解决"并行下发"的问题：一份任务列表可以并发派发给多个同类型的节点，整体图结构是动态展开的。这一块结合下面的案例讲会更直观。

3. 案例拆解：一个三Agent研究助手是如何跑通的

下面进入正题。我以一个实际做过的项目为蓝本：做一个"行业研究报告生成助手"，用户输入一个研究主题，系统自动产出结构化的研究报告。这个场景非常适合讲Multi-Agent，因为它的子任务边界天然清晰。

3.1 需求定义与Agent角色划分

原始需求听起来很简单：输入"2025年AI Agent落地现状"这样的主题，输出一份报告。但真正拆开之后会发现，里面其实包含三个性质完全不同的子任务：

1. 资料搜集：需要基于主题拆出多个子问题，然后分别调用搜索工具获取资料。这个环节的关键是"广"和"快"，越多独立检索结果越好。
2. 数据分析：把搜集到的原始资料做清洗、提炼、归纳，形成论点清单和关键数据。这个环节的关键是"准"，需要过滤掉无关信息。
3. 报告撰写：基于分析结果组织成段落结构，生成最终报告。这个环节的关键是"稳"，输出格式必须统一。

所以我把系统拆成三个Agent：search_agent（资料搜集）、analyze_agent（数据分析）、write_agent（报告撰写）。每个Agent的Prompt都刻意保持短小聚焦，只描述自己的职责和输出格式约束，不掺杂其他环节的内容。

一个重要经验：角色划分的粒度不是越细越好，而是按"上下文需求差异"来划分。如果两个环节需要关注的上下文完全不同，才应该拆成两个Agent；如果只是顺序不同但上下文相似，拆开反而增加通信成本。

3.2 代码实现：从State定义到Graph编排

先定义三个Agent。这里我用的是LangGraph提供的create_agent辅助函数（实际项目里用这个版本比手写bind_tools的方式少很多样板代码，旧版本里这个函数叫create_react_agent，用法基本一致）：

from langgraph.prebuilt import create_agent from search_tools import web_search, fetch_content search_agent = create_agent( llm, tools=[web_search, fetch_content], system_prompt="你负责资料搜集。请根据研究主题拆解出3-5个子问题，" "逐一调用搜索工具获取资料，返回结构化参考资料列表。" "注意：只负责搜集，不要做深度分析。", )

分析Agent不接外部工具，纯做推理：

analyze_agent = create_agent( llm, tools=[], system_prompt="你是数据分析员。基于参考资料提炼3-5条核心结论，" "每条结论需要注明来源索引，输出格式为JSON。", )

写作Agent同样不接工具，负责最终润色：

write_agent = create_agent( llm, tools=[], system_prompt="你是资深报告撰写人。基于分析结论生成结构完整的行业研究报告，" "包含摘要、正文各节、关键数据表、结论与风险提示。", )

接下来定义State和节点。比较关键的是搜集环节需要并行处理多个子问题，这里我用Send API实现：

import operator from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.types import Send class ReportState(TypedDict): topic: str sub_questions: list[str] all_results: Annotated[list, operator.add] # 并行结果累加，不互相覆盖 conclusions: list[dict] report: str def plan_questions(state: ReportState) -> dict: questions = split_topic_to_questions(state["topic"]) return {"sub_questions": questions} def search_one(state: dict) -> dict: # Send分发后，这里的state是"全局State + 本次Send传入payload"合并后的结果 question = state["question"] results = run_search(question) return {"all_results": [{"question": question, "results": results}]} def analyze_one(state: ReportState) -> dict: conclusions = analyze_all(state["all_results"]) return {"conclusions": conclusions} def write_report(state: ReportState) -> dict: report = generate_report(state["topic"], state["conclusions"]) return {"report": report}

在Graph里完成编排：

builder = StateGraph(ReportState) builder.add_node("plan", plan_questions) builder.add_node("search_task", search_one) builder.add_node("analyze", analyze_one) builder.add_node("write", write_report) builder.add_edge(START, "plan") def dispatch_search(state: ReportState): return [ Send("search_task", { "topic": state["topic"], "question": q }) for q in state["sub_questions"] ] # 动态并行：每个子问题都会创建一份独立的search_task执行 builder.add_conditional_edges("plan", dispatch_search, ["search_task"]) builder.add_edge("search_task", "analyze") builder.add_edge("analyze", "write") builder.add_edge("write", END) graph = builder.compile()

这里Send("search_task", payload)的含义是：为payload里的每个元素创建独立的search_task节点实例并行执行。每个实例的返回值通过operator.add这个reducer累加到all_results上，不会互相覆盖。这就是LangGraph里并行多Agent的基本写法，也是这套框架对比手写并发代码最舒服的地方。

3.3 运行结果与状态流转解读

拿真实输入"2025年AI Agent落地现状"跑一遍，整个流程是这样的：

1. plan节点把主题拆成"企业采用率数据""主要应用场景""技术栈分布""落地瓶颈""未来趋势"五个子问题。
2. 五个search_task并行启动，每个调用搜索工具并抓取页面内容，耗时基本同时完成。
3. 所有结果汇总到all_results后，analyze节点把所有资料压缩成5条核心结论，并标注来源索引。
4. write节点基于结论生成一篇约3000字的完整报告。

我把streaming打开做可视化观测，能看到典型的"先发散再收敛再发散"过程：plan阶段拆问题是在发散，search阶段并行分支达到最大发散度，analyze阶段收敛成结论，write阶段又为了组织语言而发散。这种动态过程的掌控感，正是图编排框架相对传统链式调用的核心优势。

LangGraph的streaming能力在调试多Agent时非常有用。graph.stream(initial_state, stream_mode="updates")可以逐步看到每个节点返回的增量状态；stream_mode="messages"则能看到模型逐字生成的过程。我在调试阶段几乎全程开着这两个模式。

4. 三种主流协作模式：Supervisor、分层与对等

上面的案例是"流水线式"协作，属于最基础的一种。实际项目中做技术选型时，LangGraph里常见的有三种组织模式，我分别说下适用场景和坑。

4.1 Supervisor模式：一个老板管一堆员工

Supervisor模式里有一个"主管Agent"负责决策，把任务分发给多个"执行Agent"，并根据执行结果决定下一步派给谁：

def supervisor_agent(state): decision = llm.invoke(supervisor_prompt.format(state=state)) return {"next": decision.action} builder.add_conditional_edges("supervisor", route_based_on_next, { "researcher": "researcher", "coder": "coder", "writer": "writer", "FINISH": END })

这种模式适合任务类型不固定、执行路径依赖中间结果的场景。坑在于：主管Agent本身可能做出错误决策，尤其是Agent数量超过四个之后，选错分支的概率会明显上升。我的应对方式是给主管Agent提供"当前各Agent能力说明"和"上次执行摘要"，并在关键路径上做硬编码校验。

4.2 分层模式：Agent之间还有子Agent

分层结构是多层嵌套，顶层Agent把任务拆给中层，中层再拆给底层。好处是每个层级的上下文窗口更可控，适合非常复杂的任务；代价是延迟翻倍，调试困难，而且任何一层出错都会被放大。

我在实际中很少直接写多层嵌套，除非有明确的上下文隔离需求。一个相对好用的替代方案是：先让顶层Agent做"任务规划"，输出一个带依赖序的任务列表，然后交给扁平流水线去执行。这样逻辑上是分层的，执行上还是扁平图，兼顾了清晰与可控。

4.3 对等模式：一群Agent互相协作

对等模式里每个Agent地位相同，可以自由地互相发送消息。在手写循环里这种模式经常失控，因为没有收敛机制。LangGraph里我会用"固定轮次上限+统一汇总节点"来约束，比如最多三轮对话后强制进入汇总节点。

三种模式的选择逻辑，我用一个表格总结：

5. 实测中的坑与优化：并发、超时与状态污染

这部分是我在真实项目里踩过坑之后沉淀下来的，常规教程很少会写，但直接影响系统能不能稳定上线。

5.1 并行节点的状态覆盖问题

这是我在第一次用Send API时踩的坑。当时我在State里定义all_results用的不是Annotated[list, operator.add]，而是一个普通list字段。结果发现五个并行search_task节点都返回后，all_results里只剩最后一个节点写的结果。

原因就是LangGraph默认的"后写覆盖"语义：并行节点往同一个普通字段写数据时，后执行的覆盖先执行的。解决方案就是给字段加reducer，用operator.add做累加合并。这个坑很隐蔽，因为串行流程下你怎么写都正常，一旦切到并行立刻翻车。

5.2 超时与重试

多Agent系统最怕一件事：某个Agent调外部工具时卡住了。不设置超时，整个Graph会一直挂着，用户看到一个转圈圈但是没有任何反馈。

我通常给每个外部工具调用加上超时控制，并在节点层做降级处理：

import asyncio async def safe_search(question: str, timeout: float = 20.0) -> dict: try: return await asyncio.wait_for(run_search(question), timeout=timeout) except asyncio.TimeoutError: return {"error": "timeout", "question": question}

在节点函数里遇到timeout时返回降级数据，而不是让异常直接冒泡到Graph执行器。这样即使某个搜索源挂了，顶上来的也是一条"该子问题未获取到数据"的占位结果，后续分析环节仍然可以继续，只是结论里会多一条"数据有限"说明。多Agent系统的容错设计，本质上就是"任何单点失败都不能拖垮整条链路"。

5.3 Token消耗的隐性放大

多Agent的Token消耗比单Agent高很多，这点很多人启动前没算过账。比如3个Agent各维护一段上下文，光系统提示词就是三份；并行search时每个分支都带一份完整历史，如果任务切成5个子问题，状态里的消息数据会被重复传递多次。

我习惯在State里区分"局部上下文"和"全局上下文"两种字段。局部字段只存在于某个Agent执行期间，完成后就丢弃；只有必要的结论性内容才写入全局State。这个设计能明显降低Token消耗，尤其是并行分支多的时候。

5.4 调试手段：可视化与逐步执行

LangGraph编译后的图可以输出为结构表示，配合graph.stream的逐步输出，基本能替代肉眼Debug。我个人调试多Agent的固定流程是：

1. 先用stream_mode="updates"看节点级的返回，确认每个节点产出的State字段是否符合预期。
2. 再用stream_mode="messages"盯具体某一步的模型输出，确认Prompt没有跑偏。
3. 最后在开发环境打开debug=True，看每一步的函数调用顺序和耗时。

这套流程的核心思路是"从粗到细"：先看结构对不对，再看内容对不对，最后看性能对不对。直接跳到最后一步往往会被大量日志淹没。

6. 什么时候不该用Multi-Agent

这个章节本来不想写，但无论做技术分享还是自己复盘，我觉得"什么时候不该用"比"怎么用"更重要。这些年看过不少项目，多Agent用得飞起，最后效果反而不如单Agent，原因基本都落在下面几点。

6.1 任务边界本身不清晰

如果任务本身没有明确可拆分的子目标，硬拆多个Agent只会让每个Agent都"看着全局却只做局部"，上下文里堆满了自己用不到的信息。这种情况最好的方案是强化单个Agent的Prompt和工具集，让它自己决定何时调用什么工具。

6.2 延迟和成本敏感的场景要慎重

多Agent必然带来更多的模型调用次数和更大的上下文开销。如果是一个线上低延迟接口（比如要求几百毫秒内响应的场景），多Agent的链路往往跑不完，尤其是多轮Supervisor决策会带来数倍延迟。这种场景我宁愿单Agent加精调Prompt，或者用更小的模型做第一道应答。

6.3 团队对Agent机制不熟的时候先别急

这是很现实的约束。多Agent系统是"伴生复杂度"很高的系统，对状态管理、并发安全、可观测性的要求都高于普通应用。如果团队刚接触LLM应用，我建议先从单Agent加工作流编排入手，把状态、工具、上下文管理搞扎实，再逐步引入多Agent。毕竟拆Agent容易，把拆出来的Agent稳定协调起来才是真正的考验。

6.4 能用确定性代码解决的就别用Agent

很多所谓的"多步决策"其实用确定性代码写死更稳。比如"先查数据库，再根据结果走A/B分支"这类逻辑，本质上是状态机而不是智能决策，直接写代码比让Agent绕一圈更快、更便宜、更可控。Agent应该用在"决策空间大、规则难以穷举"的地方，比如开放式研究、复杂任务规划、多源信息综合。

我的经验判断法：如果一个任务能在五分钟内用普通代码写出确定性的执行流程，就不要引入Agent；如果这个流程会随着输入变化而变化，才值得让Agent做路由和编排。

最后再分享一个小技巧：在LangGraph里给每个节点都起一个简洁且语义明确的name，不要偷懒用默认函数名。因为后面做条件路由、做Supervisor决策、看可视化图的时候，这些名字是唯一能快速定位问题的线索。我在项目里吃过亏——初期不重视命名，一上Supervisor就懵了，最后花了大量时间在图上重新梳理节点关系。

多Agent的编排本质上是把"一个聪明但容易迷路的执行者"变成"一群各司其职的执行者加上一个清晰的指挥体系"。LangGraph给的是这套体系的骨架，真正让它跑得稳的，是你怎么切分职责、怎么设计状态、怎么控制并发。希望这篇案例拆解能帮你少走一些我走过的弯路。