企业官网建设流程全解析

做Agent编排这几年，我最大的感受是：单Agent能干活是运气，多Agent能稳妥地把活干完，才是系统工程。前阵子我重构了一个多角色协作的调研分析系统，核心流程从“一个模型从开场白说到结尾”改成了“多个Agent各管一段、并行推进任务、最后汇总评审”的结构，踩了不少调度上的坑。这里把整个从设计到实现的思路原样梳理一遍，重点放在LangGraph这套图执行框架上：状态怎么设计、节点怎么拆分、并行分支怎么调度、出了问题怎么排查。

这篇内容适合两类人：一类是已经用LangChain或其他编排框架写过单链，但对多Agent架构没有头绪；另一类是准备把原型推向生产，正在纠结并行执行、恢复和人工介入这些硬骨头。我尽量不写官方文档已经讲透的Hello World，而是把“为什么要这么切”“为什么卡在这里”“为什么最后这么改”一并交代清楚。

1. 多Agent系统的核心需求拆解

1.1 多角色拆解不是炫技，是给复杂度定位

先说我那个场景：需要根据一段模糊的客户需求，生成一份带市场分析、竞品对比、风险提示和行动建议的综合报告。这种任务如果只用一个Agent来做，会出现两个典型问题。第一，上下文会膨胀得很快，尤其是“竞品对比”和“风险提示”这两块需要的检索和分析逻辑完全不同，硬塞进同一个prompt里很容易互相干扰；第二，返工效率低，每一步都要把整个链路重跑一遍，改一个数据点也要等全流程走完。

所以我把Agent拆成了五个角色：规划Agent负责把用户目标拆解成多个子任务；每个子任务对应一个执行Agent，真正去跑检索、计算、写结论；审阅Agent负责检查执行Agent的输出并给出修改意见；聚合Agent负责把所有通过检查的内容拼成终稿。这个结构本质上就是把任务按专业方向和验收标准切分，让每个Agent的职责域足够窄，输出质量反而更好控制。

这里有个值得强调的设计原则：角色不是越多越好，而是要让每个角色具备“独立可验收”的产出。如果两个Agent干的事情必须混合在一起才算完成，那就说明边界没有切对。我在做拆分时用的一个实用判断标准是：问一遍“如果这个Agent的输出直接丢给用户，用户能不能用”。能，则边界清晰；不能，要么是它还需要补充检索，要么是它跟旁边的Agent职责重叠了。

1.2 为什么选LangGraph而不是自己写编排

多Agent系统最难的不是调用模型，而是管理状态流转。自己写一套管理队列和回调的代码也不是不行，但很快就会发现你要重复造的是“状态存储、分支路由、失败重试、断点恢复”这些基础设施。LangGraph的价值在于它把“图”作为一等公民，把Agent之间的连线当作显式逻辑，而不是藏在Python函数调用里的隐式依赖。

我用LangGraph和手写编排做过一次对比，感受很直接：手写方式在单路径时很爽，但一旦出现“一个节点后面挂着三个并行节点，其中一个失败后如何重试、另外两个的结果如何处理”这类不确定性，代码很快就变成了“如何判断状态”的插件墙。LangGraph在这种情况下只需要定义清楚节点函数、状态更新规则和条件边，执行引擎会接管并发和恢复。

2. 状态设计：多Agent协作的“公共记忆”

2.1 状态结构怎么定义才不会后期返工

多Agent之所以容易乱，绝大多数情况是“公共状态”没设计好。每个Agent在自己的prompt里说话的时候，背后需要的共享记忆是哪些字段、谁负责写哪个字段、谁负责读哪个字段，这些在一开始就要定下来。我在LangGraph里用TypeDict方式定义状态，每个字段就是一个通道，理解成“Agent群聊时贴在墙上的一张便签纸”即可。

from typing import TypedDict, List, Dict class AgentState(TypedDict): goal: str # 用户原始目标 task_list: List[Dict] # 规划产出：子任务清单 task_results: Dict[str, str] # 各执行节点的产出，按任务ID存放 review_feedback: Dict[str, str] # 审阅意见 final_report: str # 聚合后的最终报告 history: List[str] # 事件记录，方便追踪

这个结构看起来简单，但有一个细节很容易被忽略：task_results用的是字典而不是列表列表。原因是为了支持并行写入时不互相覆盖。并行调度时多个执行Agent都会往状态里写东西，如果用整个列表字段作为更新单位，后完成的Agent会把先完成的Agent的写入整个覆盖掉；而用字典字段时，LangGraph能把键级别合并，不同任务ID写自己的键位，天然避免了并发写冲突。如果你确实想用列表，需要额外定义Reducer，告诉框架新元素怎么跟旧元素合并，否则就老老实实按我的方式来。

2.2 节点函数的签名和更新规则

LangGraph里的每个节点函数接收当前状态，返回一个“部分状态字典”。这个返回字典会被合并进全局状态。记住这个模型的逻辑：节点不是直接改全局状态，而是通过返回值声明“我想更新这些字段”。这意味着节点函数最好是纯函数式的：同一个输入状态，输出结果应该是确定的，至少不应依赖节点内部维护的随机全局变量。这个习惯会在并行调度和断点恢复时救你很多次。

from langgraph.graph import StateGraph, START, END def planner_node(state: AgentState) -> dict: task_list = build_tasks(state["goal"]) return {"task_list": task_list, "history": state["history"] + ["planning_done"]}

这里需要注意history字段的更新方式。我上面直接用了state["history"] + [...]，这意味着我在用整个列表覆盖旧值。如果两个节点同时更新history，就会发生覆盖。所以如果history只是流水日志，建议单独给它配一个Reducer，或者干脆不放在共享状态里，放到外部日志系统。共享状态不是越大越好，真正常被读取和依赖的字段才应该放进去。

2.3 并行调度背后的两种图模式

LangGraph里的并行并不玄乎，本质上就是图里的“一个节点后面挂多个节点”或者“一个节点多次唤起同一个节点”。前者适合静态并行：分支数量在图编译时就能确定；后者适合动态并行：分支数量要到运行时由数据决定。这两个场景我后来的所有项目里都会遇到。

静态并行用普通的边就能表达。比如规划完成后同时启动市场调研Agent和竞品分析Agent，就直接给规划节点挂两条输出边。动态并行更麻烦一点，因为你不知道有10个任务还是100个任务，所以不能让工程师手动画100条边，而是需要通过Send机制来触发。我把这个叫作“地图-归约”模式：把任务列表发给多个执行节点实例，各自处理一份，最后汇入同一个聚合节点。

from langgraph.types import Send def continue_to_executors(state: AgentState): tasks = [] for task in state["task_list"]: tasks.append(Send("task_executor", {"task": task, "goal": state["goal"]})) return tasks

这段代码的含义是：规划节点结束后，continue_to_executors会为每个子任务创建一个执行分支。每个分支拿到的输入是{"task": ..., "goal": ...}，经过执行节点后返回自己的结果，最终所有分支汇聚到后面的聚合节点。这个写法是LangGraph多Agent并行调度里最核心的一块，它解决的是“代码不用写死并发数”的问题。

3. 实操落地：一个可运行的多Agent并行调研系统

3.1 一个完整的小Demo场景

光讲概念大家容易飘，我直接放一个可以本地跑通的最小实现。场景是：输入一句“帮我调研一下某类产品的市场空间和风险”，系统会规划出数据收集、竞品对比、风险分析三个子任务，三个执行Agent并行动手，审阅Agent逐个把关，最终聚合Agent输出报告。这个Demo没有接真实检索工具，执行Agent内部只做了简单的文本拼接和模拟计算，但代码结构换到真实工具上完全通用。

import os import httpx from typing import TypedDict, List, Dict from langgraph.graph import StateGraph, START, END from langgraph.types import Send class AgentState(TypedDict): goal: str task_list: List[Dict] task_results: Dict[str, str] final_report: str def call_model(prompt: str) -> str: # 这里替换成你自己的模型服务调用即可 # 我方生产环境用的是内部统一的模型网关 api_key = os.environ["LLM_API_KEY"] resp = httpx.post( "https://your-model-gateway/v1/chat", headers={"Authorization": f"Bearer {api_key}"}, json={"prompt": prompt, "max_tokens": 800}, timeout=120, ) return resp.json()["choices"][0]["text"] def planner_node(state: AgentState) -> dict: tasks = [ {"id": "market", "desc": "市场规模与增长趋势分析"}, {"id": "competitor", "desc": "主要竞品对比分析"}, {"id": "risk", "desc": "政策与市场风险提示"}, ] return {"task_list": tasks} def task_executor(state: dict) -> dict: task_id = state["task"]["id"] prompt = f"你是{task_id}方向的分析师。任务：{state['task']['desc']}。用户目标：{state['goal']}" result = call_model(prompt) return {"task_results": {task_id: result}} def review_node(state: AgentState) -> dict: bad_tasks = [] for task_id, content in state["task_results"].items(): prompt = f"请评估以下分析是否有明显遗漏，回答OK或者需要修改：\n{content}" judge = call_model(prompt).strip().upper() if judge != "OK": bad_tasks.append(task_id) return {"bad_tasks": bad_tasks}

注意上面这段代码里我为了简化，把bad_tasks直接放进了返回值，但它没有在AgentState里定义过。LangGraph对未定义字段的处理规则比较宽松，有些版本会忽略，有些版本会告警。更稳重的做法是在AgentState里显式加上bad_tasks: List[str]和needs_revision: bool这样的字段。状态字段宁多勿缺，少一个字段，后面写条件路由的时候会非常别扭。

3.2 聚合节点与条件路由的衔接

假设执行结果不需要审阅返工，聚合节点就是最后一步。但如果审阅发现某些任务质量不达标，我希望系统能把任务重新塞回执行节点。这事如果放在手写代码里，你需要一个队列缓存和循环控制；在LangGraph里，这就是一条条件边的问题。

def should_rewrite(state: AgentState) -> str: if state.get("bad_tasks") and state.get("rewrite_count", 0) < 2: return "rewrite" return "aggregate" def aggregate_node(state: AgentState) -> dict: parts = [] for task_id, content in state["task_results"].items(): parts.append(f"## {task_id}\n{content}") final_text = "\n\n".join(parts) return {"final_report": final_text}

图构建部分，把节点和边串起来：

def spawn_executors(state: AgentState): return [Send("task_executor", {"task": t, "goal": state["goal"]}) for t in state["task_list"]] builder = StateGraph(AgentState) builder.add_node("planner", planner_node) builder.add_node("task_executor", task_executor) builder.add_node("review", review_node) builder.add_node("aggregate", aggregate_node) builder.add_edge(START, "planner") builder.add_conditional_edges("planner", spawn_executors, ["task_executor"]) builder.add_edge("task_executor", "review") builder.add_conditional_edges("review", should_rewrite, {"rewrite": "task_executor", "aggregate": "aggregate"}) builder.add_edge("aggregate", END) graph = builder.compile()

这段代码有些细节要特别提示：task_executor节点被Send多次触发，意味着它会以多个并发副本运行。每个副本收到的状态不是完整的AgentState，而是我在Send里指定的那部分数据。所以在task_executor函数中，不能假设state["task_results"]已经存在，因为分支其实还没跑过其他任务。如果需要读取全局面，就得像我一样把goal显式传进Send的payload里。这个“分支状态与全局状态分离”的认知，是我见过新手出错率最高的点。

3.3 运行图并观察并行效果

图编译完成后，直接调用invoke启动一次完整执行。传入的初始状态只需要包含goal，其他字段会让框架自己填充。

result = graph.invoke({"goal": "某新式茶饮品牌的市场空间与主要风险"}) print(result["final_report"])

在这个Demo里，三个并列的执行任务理论上会同时启动。如果你用的模型网关支持并发请求，你会发现整轮执行的整体耗时约等于“最慢的那个任务+规划+审阅+聚合”，而不是所有任务耗时的总和。这就是并行调度的收益。我在本地验证时用了计时接口去包住每个节点，三个任务如果串行大约需要9秒，并行后总耗时压缩到了4秒左右，效果非常直观。

4. 调度过程的调试：看状态流而不是看日志

4.1 用Stream模式观察每一步的真实状态变化

多Agent系统调试最痛苦的是因果链太长。执行Agent写的结论不对，究竟是规划任务没理解需求，还是执行Agent本身漏了关键条件，抑或审阅Agent误判了质量？只看最终输出很难定位。LangGraph的stream方法可以让我按节点逐个看到状态的变化。

for event in graph.stream({"goal": "..."}, stream_mode="updates"): for node_name, state_update in event.items(): print(f"[{node_name}] {state_update.keys()}")

我喜欢把stream_mode="updates"当作默认调试模式，因为它是按节点更新来输出，能非常清楚看到每个节点当时往状态里塞了什么。比如当执行节点的输出为空，你立刻能看到是哪类任务的结果缺失，再回去查那个子任务的prompt设计是否合理。如果想看更细的LLM调用输入输出，就结合模型网关的日志一起看。总之一句话：调试多Agent的原则是“先看状态流，后看模型质量”，因为模型质量问题往往是被状态设计问题放大的。

4.2 并行不等于零成本：并发和限流的博弈

并行调度的第一个问题就是触发模型服务的限流。如果你有20个执行Agent同时挤向同一个推理网关，网关通常会返回限流错误或者显著增加响应延迟。我实测下来的处理方式是两层控制：首先要给LangGraph的每个节点调用加上重试机制，一般对限流错误做1-2次指数退避重试就能扛过去；其次要在更上层控制真正的并发度，不要让一个图并行分支数超过服务承诺的并发上限。

LangGraph本身不限制同一节点的并行副本数量，自由度很高，但业务上必须有限额。我有一次把任务切成了将近50个细粒度子任务，结果第一轮就触发了网关限流，整条工作流失败重跑，白白浪费了成本和等待时间。后来我在规划节点里加了一个简单控制逻辑：如果子任务数量超过设定阈值，就自动把小任务合并成几个大任务包。这个措施比任何高级调参都管用。并行调度应该服务于“最迟完成时间“，而不是为了并行而并行。

5. 常见坑与生产化的经验总结

5.1 一张表说清楚高频问题

我把反复踩过的坑整理成速查表，供直接参考：

5.2 从Demo走向生产，必须补齐的三个能力

Demo能跑通只是第一步，生产环境下多Agent系统还需要补三件事。第一是持久化。给编译后的图配上内存版以外的Checkpointer，比如支持SQLite或Redis的实现，然后让每次任务的执行都带上唯一的配置ID。这样执行中途挂了，可以用graph.get_state(config)拿到当时的完整现场，然后从断点恢复，而不是整条链路再花一遍钱。

第二是人工介入窗口。不是所有审阅都应该交给模型来完成，高风险的最终发布前，我建议加一档HumanInTheLoop节点，用LangGraph的interrupt机制把执行挂起，等人点击“通过”按钮之后再继续。这既是为了安全，也是为了让业务方对AI产出有掌控感。

第三是安全和脱敏。多Agent系统的工具调用权分散在多个节点，权限面被扩大了，Prompt注入的风险也随之增加。必须要做的底线措施是：不让执行Agent直接无限制调用高风险工具；外部检索结果在进入Agent上下文前做内容过滤；对Agent输出的URL、文件路径等做白名单校验。安全设计不能等上线之后再补，尤其当Agent内部开始并行调用多个工具的时候，一条恶意输入就可能被放大成多次违规操作。

我自己在实际项目里还有一个习惯：每个Agent节点的prompt里都要求输出结构化JSON，并由scheme层做强校验。这比让模型输出自由文本再靠下一个Agent猜要可靠一个数量级。因为并行调度一旦铺开，一个节点的文本格式漂移就可能让后续所有相关节点跟着返工，格式化契约是多Agent稳定的地基。

做多Agent系统，归根结底是状态流、控制流和资源流的统一设计。LangGraph给了我一个非常顺手的表达工具，但架构的眼光还是要靠自己在真实任务里磨。希望这篇基于实践梳理的内容，能让你在从单Agent走向多Agent协作时少踩几个坑。