企业官网建设流程全解析

算销量增长率为什么要用图？把任务拆成检索、计算、校验三个节点，用条件边让校验失败的流程自动回头重查。附已实跑验证的完整代码与真实日志。

§1 问题背景：为什么一个函数写不完销量增长率？

分析「手机X 的 2024Q2 销量环比增长率」，第一版我用一个函数就写完了。跑第一次就翻车：数据查回来缺了上季基数，增长率算成 None。然后呢？函数没有回头路。这种活至少会以三种方式翻车：

• 数据查错或缺失：RAG 返回的销量数据缺字段，算不出数
• 算出来离谱：上季基数缺失导致除零，增长率越界
• 结果不可信：没有校验，错误答案直接交付

单函数是直通管道，数据错一次，答案就错到底。

更麻烦的是流程问题：错误答案看起来有模有样，交付就是事故。

要救回来，流程得能回头。查错了重查，算错了重算，校验不过不交付。这就是本文要搭的闭环：任务拆解，分步执行，结果校验。三个环节各司其职，缺一个都不闭环。

§2 节点：把任务拆成可独立重试的三个单元

LangGraph 把流程拆成离散节点。节点就是做一件具体事的函数，里面可以放 LLM，也可以只是普通代码。节点之间不直接调用，只通过共享 State 协作，官方一句话概括：nodes do the work, edges tell what to do next。整个图只有两种元素：框和箭头，框是节点，箭头是边。

先把闭环拆成三个节点：检索查数据，计算算增长率，校验判合理性。这层隔离是能独立重试的前提，retrieve 重跑一次，不会把 compute 的脏中间结果带进来。三个节点函数如下：

def retrieve(state: SalesState) -> dict: attempts = state["attempts"] + 1 # 节点内读改写，自增计数 data = mock_rag(state["product"], attempts) msg = f"[检索] 第{attempts}次调用RAG，拿到销量数据: {data}" print(msg) return {"sales_data": data, "attempts": attempts, "trace": [msg]} def compute(state: SalesState) -> dict: d = state["sales_data"] # 只读别人写好的字段 if d.get("previous") in (None, 0): msg = "[计算] 上季基数缺失，无法计算增长率，growth_rate=None" print(msg) return {"growth_rate": None, "trace": [msg]} rate = (d["current"] - d["previous"]) / d["previous"] msg = f"[计算] 增长率 = ({d['current']} - {d['previous']}) / {d['previous']} = {rate:.2%}" print(msg) return {"growth_rate": rate, "trace": [msg]} def validate(state: SalesState) -> dict: rate = state["growth_rate"] attempts = state["attempts"] if rate is None or rate < -1.0 or rate > 5.0: if attempts >= 2 and rate is not None: # 重试上限兜底，防死循环 msg = f"[校验] 第{attempts}次仍不通过({rate:.2%})，已达重试上限，强制放行并告警" print(msg) return {"validation": "valid", "trace": [msg]} msg = f"[校验] 增长率 {rate} 不合理(缺失或越界)，判 invalid → 回检索重试" print(msg) return {"validation": "invalid", "trace": [msg]} msg = f"[校验] 增长率 {rate:.2%} 合理，判 valid → 输出结果" print(msg) return {"validation": "valid", "trace": [msg]}

图上的每个框，对应代码里的一个函数。retrieve 写 sales_data 和 attempts，compute 读 sales_data 写 growth_rate，validate 读 growth_rate 写 validation。字段交接就是数据流。

compute 拿到脏数据，不自己回头，只把 growth_rate 写成 None，把判断权交给校验。节点不越权，才能被独立替换、独立重试。

§3 State：节点间的共享状态设计

State 是所有节点共享的状态容器，一个 TypedDict，所有跨节点的数据都从这里过。官方给了两条设计原则：要跨步骤保留的才进 State，能从别处推导的不存。存原始数据，不存格式化文本，格式化在节点内做。本例的读写规则如下：

代码里的 TypedDict 长这样：

class SalesState(TypedDict): question: str # 原始问题（输入），数据检索节点读 product: str # 目标产品 sales_data: dict # 销量数据：数据检索写，计算/校验读 growth_rate: float # 增长率：计算写，校验读 validation: str # valid/invalid：校验写，条件边读 attempts: int # 检索/重试计数：检索自增，限流防死循环 trace: Annotated[list[str], add] # 状态流转日志（累加 reducer）

表里的读写规则与 TypedDict 一一对应；真实代码比表多一个 product 字段（检索节点实际用产品名查库），少一个 final_answer 汇总字段（最小实现里 END 直接收尾，不单独汇总）。字段交接即数据流。

State 的合并规则，三条就能说清：

• 默认 reducer 是覆盖：节点返回什么，字段就变成什么
• attempts 在节点内读改写，返回 attempts + 1，不引入额外 reducer
• trace 用 Annotated[list[str], add] 累加：每步日志追加进列表，历史不丢

为什么不存格式化文本？增长率永远从 sales_data 现算，不单独存格式化字符串。存了格式化文本，数据一变就得同步改两处。State 只放真相，不放展示层。

§4 条件边：用校验结果驱动重试回环

普通边是 A 永远去 B。条件边不同：一个路由函数读当前 State，返回目标节点名，可以配一张映射字典。

条件边挂在校验节点之后，因为 validation 由校验产出。为什么不挂计算节点？因为 validation 还没写出来，路由函数没东西可读。

route_after_validate 读 state[“validation”]：valid 去 END，invalid 回 retrieve 重查。灵魂代码只有三行：

def route_after_validate(state: SalesState) -> str: # 路由函数：读当前 State，返回目标节点名 return state["validation"] builder.add_conditional_edges( "validate", # 条件边挂在校验节点之后 route_after_validate, # 路由函数读 validation 字段 {"valid": END, "invalid": "retrieve"}, # valid 结束，invalid 回检索重试 )

add_conditional_edges 的第三个参数是映射表。路由函数的返回值是键，映射表告诉引擎该去哪。不传映射表，返回值直接当节点名。返回值与节点名解耦，想改去向只动映射表。

防死循环有双保险：

• 节点内 attempts 上限，超过 2 次强制放行并告警
• graph 级 recursion_limit，默认 1000 步，超出抛 GraphRecursionError

recursion_limit 是全局护栏，attempts 是局部护栏，两层都设，脏数据拖不垮流程。生产环境可以把 recursion_limit 调小，让失败来得快一点。

共享状态里的 validation 字段，加上条件边，就是 agent 的纠错能力。

§5 代码实现：完整闭环与真实运行日志

把节点、State、条件边三块拼起来，就是下面的完整闭环。核心盯住 validation：它是整条闭环的方向盘，写在 State 里，条件边只读它决定去向。

如何跑：

1. 安装依赖：pip install langgraph。mock RAG 是本地销量库，不依赖外部 API、密钥。换真实 RAG 时只替换 mock_rag 一个函数
2. 把上面代码原样存成 closed_loop.py，运行：python closed_loop.py
3. 预期输出：首轮脏数据 → invalid → 回检索 → 二轮正确 → valid → END，退出码 0

完整代码（langgraph 1.2.10 下实跑验证，退出码 0）：

""" 实战闭环：用 LangGraph 实现 "任务拆解 → 分步执行 → 结果校验" 的最小闭环 场景：分析某产品季度销量增长率 数据流：检索(RAG) → 计算增长率 → 校验合理性 → 不合法则回检索重试 依赖：pip install langgraph 说明：本例用 mock 本地销量库模拟 RAG，无需外部 API / 密钥，可直接复现。 第一次检索故意返回"脏数据"(缺上季基数)以演示闭环自愈；重试后返回正确数据。 """ from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END import json # ---------- 1. State 设计：共享内存（各节点只通过它协作）---------- class SalesState(TypedDict): question: str # 原始问题（输入），数据检索节点读 product: str # 目标产品 sales_data: dict # 销量数据：数据检索节点写，计算/校验读 growth_rate: float # 增长率：计算节点写，校验节点读 validation: str # 校验结果 valid/invalid：校验节点写，条件边读 attempts: int # 检索/重试计数：数据检索节点自增（限流防死循环） trace: Annotated[list[str], add] # 状态流转日志（累加 reducer，观察每一步） # mock RAG：本地销量库。第一次(attempt<=1)返回脏数据（缺上季基数）， # 重试(attempt>=2)返回正确数据，以演示闭环自愈。 SALES_DB = {"手机X": {"2024Q1": 1200, "2024Q2": 1500}} # 本季1500, 上季1200 → 增长率25% def mock_rag(product: str, attempts: int) -> dict: if attempts <= 1: # 第一次查：脏数据（只拿到本季，上季缺失） return {"product": product, "current": SALES_DB[product]["2024Q2"], "previous": None} # 重试后：正确数据 return { "product": product, "current": SALES_DB[product]["2024Q2"], "previous": SALES_DB[product]["2024Q1"], } # ---------- 2. 三个节点 ---------- def retrieve(state: SalesState) -> dict: attempts = state["attempts"] + 1 data = mock_rag(state["product"], attempts) msg = f"[检索] 第{attempts}次调用RAG，拿到销量数据: {data}" print(msg) return {"sales_data": data, "attempts": attempts, "trace": [msg]} def compute(state: SalesState) -> dict: d = state["sales_data"] if d.get("previous") in (None, 0): # 上季基数缺失 → 无法计算，增长率记 None（交给校验节点判无效） msg = "[计算] 上季基数缺失，无法计算增长率，growth_rate=None" print(msg) return {"growth_rate": None, "trace": [msg]} rate = (d["current"] - d["previous"]) / d["previous"] msg = f"[计算] 增长率 = ({d['current']} - {d['previous']}) / {d['previous']} = {rate:.2%}" print(msg) return {"growth_rate": rate, "trace": [msg]} def validate(state: SalesState) -> dict: rate = state["growth_rate"] attempts = state["attempts"] # 校验规则：增长率必须可计算，且在合理区间 [-100%, +500%] if rate is None or rate < -1.0 or rate > 5.0: # 超过 2 次仍不通过则强制放行并告警（防死循环兜底） if attempts >= 2 and rate is not None: msg = f"[校验] 第{attempts}次仍不通过({rate:.2%})，已达重试上限，强制放行并告警" print(msg) return {"validation": "valid", "trace": [msg]} msg = f"[校验] 增长率 {rate} 不合理(缺失或越界)，判 invalid → 回检索重试" print(msg) return {"validation": "invalid", "trace": [msg]} msg = f"[校验] 增长率 {rate:.2%} 合理，判 valid → 输出结果" print(msg) return {"validation": "valid", "trace": [msg]} # ---------- 3. 条件边：用 validation 驱动闭环 ---------- def route_after_validate(state: SalesState) -> str: # 返回 "valid" / "invalid"，由下方映射决定去向 return state["validation"] # ---------- 4. 组装图 ---------- builder = StateGraph(SalesState) builder.add_node("retrieve", retrieve) builder.add_node("compute", compute) builder.add_node("validate", validate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "compute") builder.add_edge("compute", "validate") # 校验节点之后挂条件边：valid→END(输出)，invalid→回 retrieve(重试) builder.add_conditional_edges("validate", route_after_validate, {"valid": END, "invalid": "retrieve"}) graph = builder.compile() # ---------- 5. 运行 ---------- if __name__ == "__main__": init = { "question": "分析手机X的2024Q2销量环比增长率", "product": "手机X", "attempts": 0, } print("=== 初始状态 ===") print(json.dumps({k: v for k, v in init.items()}, ensure_ascii=False)) print("=== 运行（stream_mode=values 观察每一步 State 快照）===") for step in graph.stream(init, {"recursion_limit": 10}, stream_mode="values"): print("--- State 快照 ---") print(json.dumps(step, ensure_ascii=False)) print("=== 运行结束 ===") final = graph.invoke(init, {"recursion_limit": 10}) print("最终增长率:", final.get("growth_rate"))

真实运行日志（主编实跑留底，中间快照省略重复字段）。stream_mode=values 会把每一步 State 快照打出来，这是观察图流转的标准姿势，照着自己跑一遍应该逐行一致：

[检索] 第1次调用RAG，拿到销量数据: {'product': '手机X', 'current': 1500, 'previous': None} [计算] 上季基数缺失，无法计算增长率，growth_rate=None [校验] 增长率 None 不合理(缺失或越界)，判 invalid → 回检索重试 --- State 快照 --- {..., "growth_rate": null, "validation": "invalid", "attempts": 1, "trace": [...]} [检索] 第2次调用RAG，拿到销量数据: {'product': '手机X', 'current': 1500, 'previous': 1200} [计算] 增长率 = (1500 - 1200) / 1200 = 25.00% [校验] 增长率 25.00% 合理，判 valid → 输出结果 --- State 快照 --- {..., "growth_rate": 0.25, "validation": "valid", "attempts": 2, "trace": [...]} 最终增长率: 0.25

读日志，看闭环怎么自愈：

• 第一次循环：脏数据缺 previous，compute 写 growth_rate=None，校验判 invalid，条件边指回 retrieve
• 第二次循环：数据到位，算出 0.25，判 valid，流向 END

attempts 从 1 变 2，validation 从 invalid 变 valid，就是一次自愈的全过程。

mock_rag 的 attempts<=1 分支是故意的故障注入，先喂一次脏数据，验证闭环真会自愈。这种注入测试，比读文档管用。

§6 原理与边界：三类错误来源与适用场景

6.1 三类错误从哪来

这套拆解思路有出处。Plan-and-Solve 论文指出，Zero-shot-CoT 的推理结果有三类错误：计算错误、缺失步骤、语义误解。解法是先把大任务 plan 成子任务，再按序 solve。

论文在 GPT-3 上评测十个数据集，Plan-and-Solve 全面超过 Zero-shot-CoT，在数学推理任务上与 8-shot CoT 相当。

Despite the success of Zero-shot-CoT, it still suffers from three pitfalls: calculation errors, missing-step errors, semantic misunderstanding errors.
—— Plan-and-Solve Prompting（arXiv:2305.04091）

论文提的三类错误，对应到本文是三处拦截：

• 计算错误 → 计算节点先判基数，缺失就不硬算
• 缺失步骤 → 校验判 invalid，强制回检索补齐
• 语义误解 → 校验的合理区间拦下越界值

LangChain 官方博客把同一思想做成 plan-and-execute 架构：planner 拆计划，executor 按计划执行。

• ReAct 每走一步都要调一次大模型，plan-and-execute 只在拆计划和收尾时调
• ReAct 一次只规划一个子问题，plan-and-execute 先想清楚整条路径

executor 执行完，planner 再被叫回来，决定收尾还是改计划。

校验节点 = 三类错误的兜底拦截器。数据不对就重查，算错就重算，不让错误答案直接交付。

6.2 什么时候别用图

反过来也要知道边界：

• 一次函数能算完的事，别上图。图的收益来自「要回头重试」的场景
• 自检有代价。官方明说 reflection 耗时长，低延迟应用不合适，适合质量优先的知识密集任务
• 更重的兜底：interrupt 人机回环、checkpointer 持久化、多智能体，留待下篇

判断标准就一条：流程里有没有「要回头」的决策点。没有，就别上图的复杂度。数据质量越差、越需要自检的任务，图的收益越大。

§7 小结：拆解、执行、校验的三段式

条件边加共享状态，等于让 agent 拥有纠错能力。全文可归结为一个三段式：

• 任务拆解 → 三个节点：retrieve / compute / validate
• 分步执行 → State 字段交接：sales_data → growth_rate → validation
• 结果校验 → 条件边回环：invalid 回 retrieve，valid 到 END

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】