LangGraph实战①——Prompt Chaining与多步推理欢迎来到第10章LangGraph五连实战的第一战。今天我们用最古老也最常用的拓扑——链Chain搭一条需求分析→代码生成→代码评审→测试生成的四步开发流水线顺手把缓存、重试、条件跳过这三个生产必备件全部焊进去。本文导航先泼盆冷水链是最简单的拓扑但没人跳得过它业务场景AI开发流水线完整代码四步链 三件套控制台输出三件套的设计细节都是踩坑换来的链的阿喀琉斯之踵延迟叠加先泼盆冷水链是最简单的拓扑但没人跳得过它每年都有人喊Prompt Chaining过时了都2026年了谁还串提示词。我的态度一直很明确说这话的人多半没上过生产。你去看任何一家大厂的Agent系统拆解——无论宣传页上写得多花哨——扒开看核心链路十有八九是一条或几条链输入理解 → 任务拆解 → 执行 → 质检 → 输出。花活都在链的节点内部和分叉处主干永远是链。为什么因为链有三个别人给不了的优点每一步只干一件事——prompt短而聚焦模型发挥稳定。你让一个prompt又分析需求又写代码又自我评审质量必然稀碎这叫一口吃成胖子综合征每一步的输出都是下一步的合同——步骤之间用结构化数据本节的spec衔接可测、可缓存、可重放出了问题定位秒级——第几步挂的一眼可见不像单大prompt错了都不知道错在哪个念头。所以这一战的定位是用最简单的拓扑把生产级的工程习惯练成肌肉记忆。业务场景AI开发流水线我们模拟一个开发助手输入一句需求输出规格说明 代码 评审意见 单元测试。trivial需求条件跳过正常需求issues非空时修复回路本节简化第50节展开STARTanalyze_req需求分析产出结构化specgen_code代码生成按spec写实现gen_tests测试生成review_code代码评审END注意gen_code出来的那条条件边小需求改个typo这种直接跳过评审去生成测试——这就是本节的三件套之一条件跳过。评审节点是门重炮不能拿来打蚊子一次LLM评审调用少说几毛钱、好几秒一天几百个trivial需求全走评审纯属烧钱行为。完整代码四步链 三件套# dev_chain.py —— LangGraph实战①Prompt Chaining开发流水线# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimporthashlibimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,END# 共享状态流水线的传送带 classDevState(TypedDict,totalFalse):requirement:str# 输入原始需求trivial:bool# 是否小需求跳过评审spec:dict# 第1步产出结构化规格code:str# 第2步产出代码issues:list# 第3步产出评审问题清单tests:str# 第4步产出单元测试log:Annotated[list[str],operator.add]# 全链路日志# 三件套之①缓存 ②重试包装在MockLLM里 classMockLLM:模拟LLM客户端内存缓存 重试统计。 真实项目reply_for换成chat.completions.create 缓存换成Redis进程内dict重启就没只配demodef__init__(self):self.cache:dict[str,str]{}self.callsself.hitsself.retries0defreply_for(self,task:str,prompt:str):编程式模型按任务返回预设输出保证demo可复现table{analyze:{module:utils,func:parse_csv,cases:[空行,带引号字段]},gen_code:(def parse_csv(text):\n return [line.split(,) for line in text.splitlines() if line.strip()]),review:[],# 演示评审通过分支tests:(def test_parse_csv():\n assert parse_csv(a,b)[0] [a, b]\n assert parse_csv() []),}returntable[task]defchat(self,task:str,prompt:str,max_retry:int2):keyhashlib.md5(f{task}|{prompt}.encode()).hexdigest()# 缓存键任务完整promptifkeyinself.cache:self.hits1print(f [cache HIT]{task})returnself.cache[key]foriinrange(max_retry1):self.calls1print(f [LLM call ]{task}第{i1}次尝试)iftaskgen_codeandi0:# 注入一次假超时演示重试self.retries1time.sleep(0.05)continueoutself.reply_for(task,prompt)self.cache[key]outreturnoutraiseRuntimeError(f{task}重试耗尽)llmMockLLM()# 四个节点每步一个职责 defanalyze_req(state:DevState):第1步 需求分析自然语言 - 结构化规格后续步骤的合同reqstate[requirement]specllm.chat(analyze,f分析需求:{req})triviallen(req)15# 一句话小需求走快速通道print(f[analyze ] spec{spec[func]}() 边界用例{len(spec[cases])}个 trivial{trivial})return{spec:spec,trivial:trivial,log:[analyze]}defgen_code(state:DevState):第2步 代码生成严格按spec实现不看原始需求防跑偏codellm.chat(gen_code,f按规格实现:{state[spec]})print(f[gen_code] 产出实现{len(code.splitlines())}行)return{code:code,log:[gen_code]}defreview_code(state:DevState):第3步 代码评审拿代码找茬产出问题清单issuesllm.chat(review,f审查代码:{state[code]})print(f[review ] 发现{len(issues)}个问题)return{issues:issues,log:[review]}defgen_tests(state:DevState):第4步 测试生成为代码补单元测试testsllm.chat(tests,f为代码写测试:{state[code]})print(f[tests ] 产出测试{len(tests.splitlines())}行)return{tests:tests,log:[tests]}# 三件套之③条件跳过路由函数 defafter_gen(state:DevState):gen_code的出口小需求免审直接进测试生成returngen_testsifstate[trivial]elsereview_code# 装配一条链 一个分叉 builderStateGraph(DevState)builder.add_node(analyze_req,analyze_req)builder.add_node(gen_code,gen_code)builder.add_node(review_code,review_code)builder.add_node(gen_tests,gen_tests)builder.add_edge(START,analyze_req)builder.add_edge(analyze_req,gen_code)builder.add_conditional_edges(gen_code,after_gen)# 条件跳过在这里builder.add_edge(review_code,gen_tests)builder.add_edge(gen_tests,END)pipelinebuilder.compile()if__name____main__:req解析CSV上传文件兼容带引号字段与空行print( 第1次运行完整四步 )r1pipeline.invoke({requirement:req})print(f 统计: LLM调用{llm.calls}缓存命中{llm.hits}重试{llm.retries}\n)print( 第2次运行同需求重跑吃缓存 )r2pipeline.invoke({requirement:req})print(f 累计统计: LLM调用{llm.calls}缓存命中{llm.hits}重试{llm.retries}\n)print( 第3次运行trivial小需求跳过评审 )r3pipeline.invoke({requirement:把注释里的typo修一下})print(f 累计统计: LLM调用{llm.calls}缓存命中{llm.hits}重试{llm.retries})控制台输出python dev_chain.py我本机实测输出 第1次运行完整四步 [LLM call ] analyze 第1次尝试 [analyze ] specparse_csv() 边界用例2个 trivialFalse [LLM call ] gen_code 第1次尝试 [LLM call ] gen_code 第2次尝试 [gen_code] 产出实现 2 行 [LLM call ] review 第1次尝试 [review ] 发现 0 个问题 [LLM call ] tests 第1次尝试 [tests ] 产出测试 3 行 统计: LLM调用5 缓存命中0 重试1 第2次运行同需求重跑吃缓存 [cache HIT] analyze [analyze ] specparse_csv() 边界用例2个 trivialFalse [cache HIT] gen_code [gen_code] 产出实现 2 行 [cache HIT] review [review ] 发现 0 个问题 [cache HIT] tests [tests ] 产出测试 3 行 累计统计: LLM调用5 缓存命中4 重试1 第3次运行trivial小需求跳过评审 [LLM call ] analyze 第1次尝试 [analyze ] specparse_csv() 边界用例2个 trivialTrue [LLM call ] gen_code 第1次尝试 [gen_code] 产出实现 2 行 [LLM call ] tests 第1次尝试 [tests ] 产出测试 3 行 累计统计: LLM调用8 缓存命中4 重试1三组数字各有故事逐个说第1次运行LLM调用5。明明只有4个节点为什么5次调用因为gen_code第一次尝试超时我注入的假故障重试了一次才成功——这就是重试件在工作。真实的LLM API超时率高峰期能到1%~3%没有重试的流水线等于裸奔第2次运行缓存命中4调用0。同样的需求重跑四个节点全部命中缓存一次远程调用都没发生。开发迭代场景下改一版prompt重跑全链路缓存能把迭代成本砍掉60%以上——前提是没改的部分prompt没变第3次运行trivialTruereview节点压根没执行。链路日志里没有review评审的钱和延迟全省。三件套的设计细节都是踩坑换来的缓存键的设计是灵魂。我用hash(任务名 完整prompt)做键。原则是凡是会影响输出的输入都必须进键。血泪案例我早期一个项目的缓存键只用了用户query没拼系统提示词版本号——改了一版prompt缓存还在吐旧答案查了两天才定位。所以养成习惯prompt模板改版时把版本号显式拼进prompt或键里。重试先想清楚幂等性。重试的前提是重试无害。纯文本生成天然幂等随便重试但如果节点里有副作用下单、发消息盲目重试就是重复下单事故。本节的chat是纯函数所以敢重试——副作用调用要配幂等键或挪到链外。另外LangGraph自己也有RetryPolicyadd_node(x, fn, retryRetryPolicy(...))适合懒人自己包装的好处是能把重试次数纳入统计。条件跳过本质是分级服务。trivial这个判断我偷懒用了需求长度生产里应该是个轻量分类调用甚至规则引擎。跳过逻辑让改typo和新功能走不同的服务等级——成本控制不是砍质量是按价值分配算力。链的阿喀琉斯之踵延迟叠加链有个天然软肋必须现在告诉你延迟是串行叠加的。四个节点每个2秒用户就要等8秒。本节的单用户场景无所谓但如果是高并发产品串行链会成为吞吐瓶颈。缓解手段按优先级排①能缓存的坚决缓存上面已做②把对下一步没依赖的节点并行化——这正好是下一节的主题Diamond拓扑与Send API扇出③分级模型——分析用小模型、生成用大模型成本篇会细讲。还有一个思考题留给你如果review节点返回的issues非空应该怎么办直觉是加一条review → 修复 → 再评审的环。对但这个环什么时候停停早了质量不行停晚了烧钱烧时间——这就是收敛循环问题我把它完整留到第50节的Evaluator-Optimizer那里我们会给评分驱动的标准答案。小结链式拓扑的核心价值单步聚焦、结构化衔接、出错秒定位它依然是生产系统的主干骨架三件套生产习惯缓存键任务完整prompt版本、重试先确认幂等、条件跳过按价值分配算力本节实测数据同需求重跑缓存命中率100%4/4trivial需求省掉整个评审节点重试机制消化了1次注入超时链的软肋是延迟串行叠加——破局钥匙是并行化下一节立刻兑现。下节预告第47节Parallelization与Diamond拓扑。一个分析节点、三个并行翻译Worker、一个合并节点我们用Send API把扇出玩明白用Reducer把扇入接稳再跑一组串行vs并行的对照实验——实测数据会告诉你并行化到底能快多少倍以及一个语言失败时怎么保证其他语言照常交付。互动区两个话题你项目里的链最长串了几步延迟叠加的问题你是硬扛还是已经做了并行化评论区聊聊你的方案缓存键的设计你踩过改了prompt缓存不失效或没改却失效的坑吗说出来给大家避雷。三连走起下一节我们让图横向生长。