算销量增长率为什么要用图把任务拆成检索、计算、校验三个节点用条件边让校验失败的流程自动回头重查。附已实跑验证的完整代码与真实日志。§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_rateNone 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 和 attemptscompute 读 sales_data 写 growth_ratevalidate 读 growth_rate 写 validation。字段交接就是数据流。compute 拿到脏数据不自己回头只把 growth_rate 写成 None把判断权交给校验。节点不越权才能被独立替换、独立重试。§3 State节点间的共享状态设计State 是所有节点共享的状态容器一个 TypedDict所有跨节点的数据都从这里过。官方给了两条设计原则要跨步骤保留的才进 State能从别处推导的不存。存原始数据不存格式化文本格式化在节点内做。本例的读写规则如下字段类型写入节点读取节点questionstr输入数据检索sales_datadict数据检索计算、校验growth_ratefloat计算校验validationstr校验条件边(router)attemptsint数据检索(自增)校验(限流)final_answerstrEND 前汇总—代码里的 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不引入额外 reducertrace 用 Annotated[list[str], add] 累加每步日志追加进列表历史不丢为什么不存格式化文本增长率永远从 sales_data 现算不单独存格式化字符串。存了格式化文本数据一变就得同步改两处。State 只放真相不放展示层。§4 条件边用校验结果驱动重试回环普通边是 A 永远去 B。条件边不同一个路由函数读当前 State返回目标节点名可以配一张映射字典。条件边挂在校验节点之后因为 validation 由校验产出。为什么不挂计算节点因为 validation 还没写出来路由函数没东西可读。route_after_validate 读 state[“validation”]valid 去 ENDinvalid 回 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 步超出抛 GraphRecursionErrorrecursion_limit 是全局护栏attempts 是局部护栏两层都设脏数据拖不垮流程。生产环境可以把 recursion_limit 调小让失败来得快一点。共享状态里的 validation 字段加上条件边就是 agent 的纠错能力。§5 代码实现完整闭环与真实运行日志把节点、State、条件边三块拼起来就是下面的完整闭环。核心盯住 validation它是整条闭环的方向盘写在 State 里条件边只读它决定去向。如何跑安装依赖pip install langgraph。mock RAG 是本地销量库不依赖外部 API、密钥。换真实 RAG 时只替换 mock_rag 一个函数把上面代码原样存成 closed_loop.py运行python closed_loop.py预期输出首轮脏数据 → 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本地销量库。第一次(attempt1)返回脏数据缺上季基数 # 重试(attempt2)返回正确数据以演示闭环自愈。 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_rateNone 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_asciiFalse)) print( 运行stream_modevalues 观察每一步 State 快照) for step in graph.stream(init, {recursion_limit: 10}, stream_modevalues): print(--- State 快照 ---) print(json.dumps(step, ensure_asciiFalse)) print( 运行结束 ) final graph.invoke(init, {recursion_limit: 10}) print(最终增长率:, final.get(growth_rate))真实运行日志主编实跑留底中间快照省略重复字段。stream_modevalues 会把每一步 State 快照打出来这是观察图流转的标准姿势照着自己跑一遍应该逐行一致[检索] 第1次调用RAG拿到销量数据: {product: 手机X, current: 1500, previous: None} [计算] 上季基数缺失无法计算增长率growth_rateNone [校验] 增长率 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读日志看闭环怎么自愈第一次循环脏数据缺 previouscompute 写 growth_rateNone校验判 invalid条件边指回 retrieve第二次循环数据到位算出 0.25判 valid流向 ENDattempts 从 1 变 2validation 从 invalid 变 valid就是一次自愈的全过程。mock_rag 的 attempts1 分支是故意的故障注入先喂一次脏数据验证闭环真会自愈。这种注入测试比读文档管用。§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 PromptingarXiv: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 回 retrievevalid 到 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%免费】