企业官网建设流程全解析

1. 从一句话到可靠交付，中间隔着什么

先说个真实场景。我接到的需求是做一个 AI 工作流系统，用户输入一句话，比如"帮我把这份销售数据按区域汇总，找出增长最快的三条线，再生成周报"，系统要能自己理解意图，规划步骤，调用工具，执行计算，最后交付一份完整报告。

听起来像是一个 ChatBot 加几个 Prompt 的事。但真正动手之后才发现，从一句话到可靠交付，中间隔着的不是模型能力，而是工程能力。模型负责天马行空，工程负责踩刹车、修轨道、确保结果稳定复现。

这篇文章就围绕这个项目展开，涉及意图识别、多 Agent 协作、任务规划、质量评审、LangGraph 状态恢复和接口幂等设计。核心关键词是：意图识别、多 Agent、任务规划、LangGraph、幂等。

适合谁来参考？如果你在用 LangChain 或 LangGraph 搭 Agent 应用，如果你被"Agent 跑飞""结果不稳定""任务执行一半崩了要重新来"这些问题折磨过，这篇文章应该能帮上忙。我不会堆砌概念，只讲我实测下来的方案、参数选择理由，以及踩过的坑。

2. 意图识别与任务规划：别让模型自由发挥

2.1 意图识别不是分类，而是结构化理解

很多人把意图识别做成一个分类任务——定义十几个意图类型，让模型选一个。但实际业务里，一句话往往包含多个意图，而且意图之间还有依赖关系。

比如"对比本月和上月各区域销售，找出下滑最严重的，写个改进建议"。这里至少有四个意图：对比分析、筛选排序、归因分析、内容生成。如果只做一个单选题，后面四个 Agent 的执行逻辑就完全没法展开。

我的方案是：第一层用大模型做结构化抽取，输出一个包含意图标签、关键实体、约束条件的 JSON，而不是单纯分类。具体实现上用 Pydantic 定义输出结构，让模型严格按 schema 返回。

from pydantic import BaseModel, Field from typing import List, Optional class IntentResult(BaseModel): primary_intent: str = Field(description="主意图，必须是意图表中的一个") secondary_intents: List[str] = Field(description="从属意图列表，可多个") entities: dict = Field(description="抽取的关键实体，如时间范围、对象、指标") constraints: dict = Field(description="约束条件，如排序方式、数量限制、输出格式") risk_flags: List[str] = Field(description="风险标记，如无明确时间范围、指标缺失")

模型输出后先过一层校验，再进入下一步。实测下来这个设计有两点好处：

第一，意图识别结果变成了结构化数据，后续任务规划节点可以直接读取实体和约束，不用重新解析原文，减少误差传递。第二，当模型识别不确定时，它会主动暴露风险，而不是猜一个答案蒙混过关。

2.2 从意图到任务的拆解：规划器是跑在轨道上的

有了意图和实体，下一步是把它们映射成具体任务序列。这一步我交给一个专门的规划 Agent，但它的自由度是受限的。

规划 Agent 的系统提示里有一个任务库，每个任务有编号、输入参数、输出参数、前置依赖。它只能从任务库里选任务、确定顺序、填入参数，不能自己发明任务。比如主意图是"对比分析"，它就选load_data->aggregate_data->compare_periods->generate_insight这条链。

这样做的原因是：自由规划看起来很智能，但在真实项目里，你根本不知道模型会吐出什么步骤组合，下游工具也没法保证被正确调用。把规划空间约束在一个封闭集合里，既保留了灵活性，又让后续的恢复、重试、幂等变得可管理。

任务拆解的结果也是一个结构化对象，包含任务列表、每个任务的输入来源（前一个任务的哪个输出字段）、执行模式（同步还是并行）。这一步的本质是把"自然语言需求"翻译成"机器可执行的 DAG"。

2.3 规划校验：宁可拒绝，不要瞎跑

任务规划完成后，我会加一个校验节点，做三件事：

1. 依赖完整性检查：每个任务的输入字段是否都有来源，缺了直接拒绝。
2. 死循环检查：虽然任务库不允许循环，但规划器偶尔会输出重复任务，要能识别。
3. 资源可行性检查：比如并行任务数量是否超过限流阈值，超了就改为串行。

这个校验节点一开始并没有，是后来加上的。原因是发生过一次事故：规划器输出了一个任务依赖自己输出的配置，Agent 在那个节点上反复重试了十多次，直接打爆了上游数据服务的接口。

所以我的建议是：规划器输出之后，必须有一个确定性代码来兜底，不能完全信任模型输出。模型负责给出方案，代码负责裁决方案是否合法。

3. 多 Agent 协作：各司其职，但别各说各话

3.1 角色划分：Planner、Executor、Reviewer，三分天下

多 Agent 不是把一堆模型丢在一起聊天。我这里的 Agent 分三类，职责边界非常清楚：

Executor 按角色还可以再拆成查询型、计算型、生成型，但核心区别只在工具权限上。查询型只能读，计算型只能跑代码，生成型只能写文本，权限隔离防止 Agent 越权操作数据。

3.2 上下文传递：只传必要信息，不传聊天记录

很多多 Agent 项目翻车，翻在上下文管理上。每个 Agent 都拿到全部对话历史，导致三个后果：Token 消耗暴涨、无关信息干扰判断、敏感数据暴露面变大。

我的做法是建立一个全局状态对象，每个节点只读写自己需要的字段。比如数据加载节点只负责往状态里写data_summary字段，评审节点只读final_report字段。节点之间不直接传递大段文本，遇到大对象只传引用或摘要。

这里补充说明一下，实际项目里我用 LangGraph 来管理这个状态流。LangGraph 的状态图结构天然适合这种节点化、状态化的设计，后面第 5 节我会重点讲它的恢复机制。

3.3 Agent 跑飞了怎么办：工具调用的"硬约束"

多 Agent 最让人头疼的问题就是跑飞——模型不按既定步骤走，自己编造工具参数，或者拒绝调用工具直接编答案。

我的方案是在 Executor 的工具调用层做硬校验。每个工具的参数都有 JSON Schema，Agent 在调用工具前，代码先对参数做校验，不符合 Schema 就拦截并返回错误信息。模型输出的 if 判断写错了，工具层直接报参数类型错误，Agent 就必须重新生成参数，而不是将错就错。

这一步是纯代码逻辑，不依赖模型自律。实测下来，工具调用失败率从最初的 16% 降到了 2% 左右。剩下的 2% 基本都是模型对参数理解确实有歧义，需要规划器重新给出澄清参数。

4. 质量评审：让结果从"能看"变"能用"

4.1 评审维度怎么定：结果要用起来才算数

评审节点最容易犯的错是把标准定成"模型自评"，让生成结果的同一个模型给自己打分。这样评出来的结果基本都是 9 分以上，一点参考价值都没有。

我的评审 Agent 用的是独立模型，而且是配置不同温度、不同提示词的独立模型。评审从五个维度打分：

• 结构完整性：是否包含标题、摘要、正文、数据附录，缺一项就不能通过
• 数据一致性：正文引用的数字是否和原始数据表一致，抽查三五处
• 逻辑连贯性：结论是否由前面的分析推导而来，有没有跳跃
• 格式合规性：是否符合交付模板，比如 Markdown 结构、表格字段
• 可执行性：给出的建议是否具体到能直接落地，还是正确的废话

每个维度都有评分规则和失败示例。评审输出不是简单一个"通过/不通过"，而是一份结构化报告，包含每个维度的分数、问题列表、建议修改位置。

4.2 评审失败后的迭代策略：有限重试，不能死循环

刚开始时，我把评审不通过的节点直接送回生成节点重新生成，结果出现了一个经典问题：生成节点重新生成的内容和上次几乎一样，评审又不通过，循环往复，白白消耗大量 Token。

后来我加了两条限制：

第一，评审不通过时，必须把"哪里不通过、为什么、期望改成什么样"这三条信息结构化回传给生成节点，而不是只回传一句"请改进"。生成节点拿到具体的修改要求后，才能有针对性地改。

第二，设置最大重试次数。我目前设为 2 次，也就是说同一个生成节点最多被评审打回 2 次。第 3 次评审还不通过，直接走人工兜底流程——把结果标记为"需人工审核"，而不是继续烧钱让模型反复试。

核心原则是：AI 系统要承认自己能力的边界，评审的重试可以暴露问题，但不能因为反复重试把线上流程堵死。

4.3 评审环节的成本控制

评审 Agent 挂的是更强的模型，成本比生成节点高。为了控成本，我做了两个优化：

一是在评审之前先跑一个规则过滤器。用正则和代码检查格式合规性、数字一致性，这些不需要模型判断的东西先拦下来。规则过滤器能发现的问题，不让大模型重复判断，能省下不少 Token。

二是评审时只输入"结果 + 评分标准 + 原始数据摘要"，不输入完整对话历史。评审需要的上下文其实很短，完整对话历史里 99% 的信息对评审没有帮助。

5. LangGraph 恢复机制与幂等设计：可靠交付的底线

5.1 为什么运行时恢复是刚需

任何一个真实系统都会遇到进程崩溃、网络超时、数据库连接断开。Agent 应用更脆弱，因为一次任务要跑十几个节点，每个节点还可能调用外部服务，任何一环断了，整个任务就前功尽弃。

没有恢复机制时，任务失败只能从头开始。从头开始意味着又要重新调用数据接口、重新跑计算、重新生成内容。我遇到过一个真实案例：上游数据服务在任务跑到第 8 个节点时超时了，重试只能从节点 1 开始，而节点 1 要扫描百万行数据，整整跑了 40 分钟。这种体验没人能接受。

LangGraph 的恢复机制解决的就是这个问题：把每一步执行结果持久化，崩溃后从最近的成功节点继续，而不是从零开始。

5.2 LangGraph 检查点：持久化与恢复的实现

LangGraph 自带检查点机制，核心是checkpointer。我把检查点存储接入了 Redis，原因是任务状态不仅要让应用进程自己在崩溃后能读到，还需要让多个无状态服务实例共享状态。

使用 LangGraph 的方式很简单，在编译图的时候传入一个 Checkpointer 实例。每次节点执行完，图的状态会自动被持久化。LangGraph 官方支持内存、SQLite、PostgreSQL 等多种存储，我这里用 Redis 主要是考虑到部署环境的统一性。

from langgraph.graph import StateGraph from langgraph.checkpoint.redis import RedisSaver checkpointer = RedisSaver(host="localhost", port=6379, db=0) graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "task-20240512-001"}}

关键点在于thread_id。这个 ID 是任务的唯一标识，LangGraph 靠它来关联同一个任务的历史状态。恢复执行时传入同样的thread_id，LangGraph 会从持久化的状态里恢复中断的节点，而不是重新构建一个新的状态。

恢复后从哪里开始执行？LangGraph 的机制是：能恢复的节点直接返回缓存结果，跳过不执行；中断的节点重新执行；下游节点继续往下走。这要求每个节点的执行结果必须是确定性的、可缓存的，这正好引出了幂等设计这个话题。

5.3 幂等设计：让重复执行没有副作用

幂等是什么？简单说，同一个操作执行一次和执行一百次，对外部系统产生的结果是一致的。幂等是恢复机制的前置条件。如果节点在恢复后被重复执行，而重复执行会产生副作用，那系统就完了。

5.3.1 接口幂等：任务 ID + 去重表

外部接口调用是副作用的主要来源。我在所有 Agent 调用的外部接口上都加了幂等控制。

实现方式很简单：客户端在请求头里带一个Idempotency-Key，服务端根据这个 key 判断请求是否已经处理过。处理过的直接返回缓存结果，没处理过的执行并缓存。

POST /api/v1/query-data Idempotency-Key: task-20240512-001-node-03-retry-02 X-Task-ID: task-20240512-001 X-Request-Hash: d41d8cd98f00b204e9800998ecf8427e

服务端逻辑分三步：

1. 先查去重表，key 存在则直接返回上一次的响应体。
2. 不存在则执行正常逻辑，结果写入去重表，状态标记为 completed。
3. 若执行过程中出现异常，写入去重表的记录标记为 failed，允许下次重试。

去重表我用 Redis 实现，key 用 MD5 校验请求内容，防止同一个 key 带上不同参数导致缓存错乱。过期时间设为 24 小时，长任务也能覆盖。

5.3.2 任务级幂等：节点重复执行不产生脏数据

接口幂等解决的是外部系统的重复调用问题。但任务内部还有一个坑：Agent 节点执行时，如果输入数据没变，输出也应该保持一致。

我之前遇到过的场景是：生成报告节点被恢复机制重新执行了一次，但这一次模型生成的报告内容跟第一次不一样——两篇报告都有价值，但和后续的评审、结论都不一致，导致整个任务的输出状态混乱。

解决办法是在节点入口做状态级幂等判断：每个节点的输出都带一个output_hash，保存进状态。节点执行前先检查输入是否变化，如果输入没变且已有输出，直接返回缓存的输出结果。

node_output_key = f"{task_id}:{node_id}:{input_hash}" cached = redis.get(node_output_key) if cached: return deserialize(cached)

这个"输入哈希 -> 输出缓存"的模式，是保证恢复后整个工作流状态一致性的关键。

5.4 幂等和恢复的配合：它们是一套组合拳

单独设计幂等或单独设计检查点恢复，都解决不了问题。幂等的目的是保证"重复执行没有副作用"，恢复的目标是"崩溃后能继续执行"。两者结合起来，才能做到 "at-most-once 语义" 变成 "effectively-once"。

实际执行时，恢复机制从检查点恢复状态，发现节点 3 已经成功执行过，就跳过节点 3 直接执行节点 4。但中途可能遇到网络波动，节点 6 其实已经执行了，只是响应超时没有把结果写回状态，此时恢复机制会让节点 6 再执行一次。如果节点 6 没有幂等保护，数据就被写了两遍，后面的数据汇总就全错了。

用代码来表达,我的工作流节点统一封装了一层"安全执行器"，逻辑是：

def safe_execute(node_name, execute_fn, task_id, input_data): # 1. 检查状态：如果该节点已经有成功输出，且输入数据未变化，直接返回缓存 input_hash = compute_hash(input_data) output_cache_key = f"output:{task_id}:{node_name}:{input_hash}" cached_output = redis.get(output_cache_key) if cached_output: return deserialize(cached_output) # 2. 执行真实逻辑 result = execute_fn(input_data) # 3. 写入输出缓存，这一步保证重试时能拿到一致性结果 output_hash = compute_hash(result) redis.set(output_cache_key, serialize(result), ex=86400) # 4. 记录执行痕迹，供评审和排障使用 redis.rpush(f"audit:{task_id}", f"{node_name}:{output_hash}") return result

有了这个统一封装，任何节点在恢复后都不会造成双重副作用，整个工作流的状态就是可追溯、可验证的。

5.5 关于 LangChain 与 LangGraph 的关系

很多人会问 LangChain 和 LangGraph 到底什么关系。个人理解是这样的：LangChain 是一套工具集合，提供了大量 Tool 和链式调用的封装，适合快速搭建串行流水线；LangGraph 则是在其基础上提供了有向图的工作流编排，核心差异就是状态管理、条件跳转和持久化检查点。

标准 LangChain 的 LCEL（LangChain Expression Language）适合固定链路的场景，一旦链路分支复杂、需要人工干预、需要可暂停可恢复，LCEL 就力不从心了。LangGraph 可以理解为"带状态的图执行引擎"，它把节点、边、状态、检查点这四个概念作为一等公民，天然适合我在这个项目里需要的多 Agent 编排和恢复机制。

6. 常见问题与排查技巧实录

6.1 意图识别被"自由发挥"改写

现象：用户输入"帮我算一下每个月的增长率"，意图识别返回的主意图是"generate_report"，实体里却出现了"环比"这类原文不存在的概念。这种自由发挥会把后续的执行带到完全错误的轨道上。

排查思路：打开意图识别节点的完整输出，对比实体列表和原文。如果模型在抽取时补充了原文没有的信息，多半是提示词里没有明确"只能基于原文抽取，不得推理补充"的约束。

解决办法：在意图识别提示词里加了两条规则：一是"所有实体必须能在原文中找到对应词，找不到就用原文原词"，二是"如果原文没有时间范围，不允许猜测本月或上月，必须标记为 missing"。另外我在解析阶段还会做一个简单的词汇对齐校验，模型抽出的实体如果不在原文分词集合里，就触发补充提醒。

6.2 评审反复不通过的死循环

现象：生成节点生成报告，评审节点不通过，生成节点修改后再提交，再被拒绝，反复消耗大量 Token。

排查思路：查看评审节点的输出详情，确认评审不通过的原因是否相同。如果两次都是同一个原因，说明生成节点根本没有理解修改意见，或者根本没有能力改到位。

解决办法：把评审意见从自然语言改为结构化问题列表，每个问题带位置信息和建议修改方向。同时设置重试上限为 2 次，超过直接转人工。另外还加了一个改良：第一次评审不通过时，生成节点可以用"补丁模式"只修改问题位置，而不是重新生成全文，这样既节省 Token，又不容易破坏原本写得好的内容。

6.3 恢复之后出现重复副作用

现象：原来已经执行成功的数据写入节点，在恢复后又被执行了一次，导致数据重复写入。

排查思路：先看检查点状态里该节点的执行标记，再看输出缓存是否存在。如果状态里标记成功但输出缓存是空的，说明该节点执行成功但没有持久化结果，这是缓存写入时机的问题。

解决办法：把步骤拆成"先缓存后返回"：节点执行成功的第一时间就写输出缓存，然后再返回给调用方。这个顺序反过来就会造成状态和缓存不一致，也是我踩过的最深的坑之一。另外所有写操作的工具调用必须走幂等接口，用任务 ID 外加节点标识做去重键。

6.4 任务一直重试，恢复了还是失败

现象：任务执行到某个节点时崩溃，恢复后还是在这个节点崩溃，往复循环，任务卡死。

排查思路：这类问题通常是死节点问题——某个外部服务不稳定或代码逻辑有 bug，导致该节点每次都会失败。检查该节点在 Redis 里的执行记录，如果失败次数反复增长且失败原因一致，基本可以断定是死节点。

解决办法：加一个断路器逻辑。每个节点连续失败 3 次后自动熔断，不再重试，直接转入人工处理流程。这个策略看起来很基础，但能避免系统在故障状态下继续浪费资源。同时为每个任务节点加上前置条件检查，如果上游数据为空或格式不对，提前终止，而不是带着问题继续往下跑。

6.5 幂等键应该选什么

这个问题是实践中最高频的疑问。幂等键不是随便一个随机字符串，它的选择直接决定了幂等的正确性。

幂等键的三个层级：

参数级幂等键有一个特别注意点：请求中的时间戳字段会导致同样的业务请求生成不同的幂等键。所以我计算 MD5 时会剔除时间戳一类的非业务字段，否则幂等形同虚设。

7. 从一句话到可靠交付的完整链路复盘

走完这一整套方案之后，我最大的体会是：AI 项目的大部分工作量不在模型怎么调，而在模型外面那层工程骨架。

意图识别是入口，它的质量决定了后续所有环节的上限。多 Agent 是分工，但分工的前提是明确的职责边界和硬约束。任务规划是调度，需要把模型自由度约束在封闭集合内。质量评审是底线，用独立视角给模型的输出踩一脚刹车。LangGraph 检查点是心脏，让执行过程在故障时不至于断崖式归零。幂等设计是免疫系统，让重复和重试不再造成二次伤害。

这套链路跑通之后，一个任务的交付时间从原来平均 30 分钟（含人工介入）压缩到 8 分钟，失败率从 21% 降到 4% 左右。4% 的失败任务里，绝大多数也已经能自动定位到具体的失败节点和原因，真正需要人工从头干预的情况很少了。

从一句话到可靠交付，模型是很容易迭代的，难的是把"想说清楚要做的事"和"确实做到了的事"这两者之间的差距，用工程手段填平。

如果你也在搭类似的 Agent 工作流，我建议的顺序是：先做幂等，再做恢复，最后才优化意图识别。前两者决定了系统的下限，意图识别决定了上限。下限不稳，上限再高都没意义。