从理论到代码用LangGraph 30分钟实现Part 2全部概念Part 2我们花25节讲了节点、边、状态、失败路由、七种拓扑——全在纸上。这一节我把这些概念用一个200行的Python文件全部变成能跑的代码。这是Part 2通往Part 3的独木桥过完这座桥后面的实战章节你就都是温故知新了。本文导航为什么需要这一节Part 2概念 → LangGraph API 对照表业务场景工单处理系统完整代码一个文件实现全部概念跑起来控制台里看图流动三个最值得记住的设计决策30分钟学习法怎么消化这个文件为什么需要这一节说个大实话Part 2讲理论的时候评论区就有同学扛不住了——“概念我都懂了然后呢”这个然后呢问得特别好。理论不落到代码上就像背了一肚子游泳教程却没下过水。但直接跳到第10章的实战又太陡那里一上来就是完整业务系统节点里混着业务逻辑、图里藏着工程技巧你会分不清哪行代码对应哪个概念。所以这一节的存在意义就一个把Part 2的每个概念逐一映射成LangGraph的API用一个最小但完整的例子——智能客服工单处理系统——让理论落地有声。环境要求先说清楚Python 3.10pip install langgraph0.4就这一个依赖。我们全程用模拟数据不调真模型保证你复制粘贴就能跑跑起来就能在控制台看到图是怎么流动的。Part 2概念 → LangGraph API 对照表先上全课最硬的一张表建议收藏Part 2概念LangGraph对应物一句话解释处理节点普通函数节点拿状态、加工数据、返回增量决策节点节点 条件边组合判断写进状态流向交给边工具节点调外部系统的函数节点查库、调API失败要转状态聚合节点汇聚多路输入的节点典型如Diamond的收口直连边add_edgeA完了必去B条件边add_conditional_edges路由函数说了算动态分发边Send运行时才知道发给谁、发几份共享状态TypedDictAnnotatedReducer多分支并行写的合并规则失败路由try/except 条件边失败转状态不炸整张图链式拓扑节点串行连边一条道走到黑分支拓扑条件边多出口一分为多环形拓扑条件边指回上游节点重试、反思、收敛循环菱形拓扑Send分发 共同后继并行展开再收口如果你对左边这列已经模糊了别慌代码看完回来再看这张表保证每行都能对上号。业务场景工单处理系统我们做一个智能客服工单处理麻雀虽小五脏俱全处理节点intake接单、识别VIP决策节点classify意图分类技术/账单/未知工具节点search_kb查知识库、check_service查服务状态两个并行演示菱形聚合节点aggregate汇总并行结果出结论兜底节点fallback升级人工失败路由的终点。账单问题技术问题Send并行分发技术问题Send并行分发未知意图失败且重试预算未耗尽环形回流失败且重试耗尽成功STARTintake处理节点接单识别VIPclassify决策节点意图分类billing_desk处理节点退款登记search_kb工具节点查知识库check_service工具节点查服务状态fallback兜底节点转人工aggregate聚合节点合并并行结果END看图说话classify到两个工具节点的那两条边就是菱形拓扑的上半截Send动态分发aggregate收口是下半截aggregate指回intake的那条边就是环形拓扑失败重试整张图里链、分支、环、菱形全齐了。完整代码一个文件实现全部概念直接上代码。我按①状态 → ②节点 → ③路由 → ④装配的顺序写这也是你以后写任何LangGraph应用的固定四步# part2_all_in_one.py —— Part 2全部概念的最小落地LangGraph版工单系统# 依赖pip install langgraph0.4 Python 3.10fromtypingimportTypedDict,Annotatedimportoperatorimporttimefromlanggraph.graphimportStateGraph,START,ENDtry:# Send的导入路径新老版本不同做个兜底fromlanggraph.typesimportSendexceptImportError:fromlanggraph.constantsimportSenddefnow()-str:returntime.strftime(%H:%M:%S)# ① 共享状态TypedDict Reducer classTicketState(TypedDict,totalFalse):ticket:str# 原始工单文本vip:bool# 是否VIPcategory:str# billing / tech / unknownattempt:int# 第几轮尝试环形拓扑的燃料error:str# 失败标记失败路由的开关findings:Annotated[list[str],operator.add]# 并行结果追加式合并log:Annotated[list[str],operator.add]# 全链路日志追加式合并resolution:str# 最终处理结论# ② 节点Part 2讲的4种节点类型 defintake(state:TicketState):处理节点纯加工数据不决定流向vipVIPinstate[ticket]nstate.get(attempt,0)1print(f[{now()}] intake 处理节点 | VIP{vip}第{n}轮)return{vip:vip,attempt:n,log:[fintake:第{n}轮]}defclassify(state:TicketState):决策节点只做判断写状态流向交给条件边——职责分离的关键textstate[ticket]cat(billingif(退款intextor扣费intext)elsetechif(报错intextor登录intext)elseunknown)print(f[{now()}] classify 决策节点 | 意图{cat})return{category:cat,log:[fclassify:{cat}]}KB{登录报错:重置SSO缓存后重试,扣费异常:核对订阅周期与发票}defsearch_kb(state:TicketState):工具节点查外部知识库。注意失败不raise转成error状态try:keynext(kforkinKBifkinstate[ticket])print(f[{now()}] search_kb 工具节点 | 命中知识库[{key}])return{findings:[fKB方案:{KB[key]}],error:,log:[kb:命中]}exceptStopIteration:print(f[{now()}] search_kb 工具节点 | 未命中 → 写入error标记)return{error:kb_miss,log:[kb:未命中]}defcheck_service(state:TicketState):工具节点查服务状态和search_kb并行跑构成菱形print(f[{now()}] check_svc 工具节点 | 全部服务正常)return{findings:[服务状态:正常],log:[svc:正常]}defaggregate(state:TicketState):聚合节点合并并行分支的结果产出结论fsstate.get(findings,[])print(f[{now()}] aggregate 聚合节点 | 合并{len(fs)}条线索)return{resolution:综合方案[.join(fs)],log:[agg:完成]}defbilling_desk(state:TicketState):处理节点账单问题的专属通道分支拓扑的一支print(f[{now()}] billing 处理节点 | 登记退款单)return{resolution:已登记退款预计3个工作日原路退回,log:[billing:完成]}deffallback(state:TicketState):兜底节点失败路由的终点人工接盘print(f[{now()}] fallback 兜底节点 | 升级人工客服)return{resolution:已转人工30分钟内回电,log:[fallback:升级]}# ③ 路由函数条件边 Send动态分发 defroute_category(state:TicketState):classify的出口账单走分支技术走Send并行分发未知走兜底ifstate[category]tech:return[Send(search_kb,state),Send(check_service,state)]# 动态分发边ifstate[category]billing:returnbilling_desk# 条件边returnfallbackdeffinal_check(state:TicketState):aggregate的出口失败路由 环形拓扑的重试预算ifstate.get(error):returnfallbackifstate.get(attempt,0)2elseintake# 环回流重试returnEND# ④ 装配把节点和边钉成一张图 builderStateGraph(TicketState)forname,fnin[(intake,intake),(classify,classify),(search_kb,search_kb),(check_service,check_service),(aggregate,aggregate),(billing_desk,billing_desk),(fallback,fallback)]:builder.add_node(name,fn)builder.add_edge(START,intake)# 直连边链式拓扑开头builder.add_edge(intake,classify)builder.add_conditional_edges(classify,route_category)# 分支 菱形入口builder.add_edge(search_kb,aggregate)# 菱形收口builder.add_edge(check_service,aggregate)builder.add_conditional_edges(aggregate,final_check)# 失败路由 环形拓扑builder.add_edge(billing_desk,END)builder.add_edge(fallback,END)graphbuilder.compile()if__name____main__:tickets[VIP用户App登录报错401快帮忙,# 走并行菱形一次成功恶意扣费要求退款,# 走账单分支登录一直失败页面打不开,# 知识库未命中失败路由环形重试]fortintickets:print(f\n 工单:{t})resultgraph.invoke({ticket:t},config{recursion_limit:50})print(f 结论:{result[resolution]})print(f 链路日志({len(result[log])}步):{ - .join(result[log])})跑起来控制台里看图流动python part2_all_in_one.py输出如下我本机实测时间戳每次不同 工单: VIP用户App登录报错401快帮忙 [20:31:02] intake 处理节点 | VIPTrue 第1轮 [20:31:02] classify 决策节点 | 意图tech [20:31:02] search_kb 工具节点 | 命中知识库[登录报错] [20:31:02] check_svc 工具节点 | 全部服务正常 [20:31:02] aggregate 聚合节点 | 合并2条线索 结论: 综合方案[KB方案:重置SSO缓存后重试服务状态:正常] 链路日志(7步): intake:第1轮 - classify:tech - kb:命中 - svc:正常 - agg:完成 工单: 恶意扣费要求退款 [20:31:02] intake 处理节点 | VIPFalse 第1轮 [20:31:02] classify 决策节点 | 意图billing [20:31:02] billing 处理节点 | 登记退款单 结论: 已登记退款预计3个工作日原路退回 链路日志(3步): intake:第1轮 - classify:billing - billing:完成 工单: 登录一直失败页面打不开 [20:31:02] intake 处理节点 | VIPFalse 第1轮 [20:31:02] classify 决策节点 | 意图tech [20:31:02] search_kb 工具节点 | 未命中 → 写入error标记 [20:31:02] check_svc 工具节点 | 全部服务正常 [20:31:02] aggregate 聚合节点 | 合并1条线索 [20:31:02] intake 处理节点 | VIPFalse 第2轮 ← 环形拓扑回流重试 [20:31:02] classify 决策节点 | 意图tech [20:31:02] search_kb 工具节点 | 未命中 → 写入error标记 [20:31:02] check_svc 工具节点 | 全部服务正常 [20:31:02] aggregate 聚合节点 | 合并2条线索 ← 注意findings被追加了两次 [20:31:02] fallback 兜底节点 | 升级人工客服 ← 重试预算耗尽失败路由兜底 结论: 已转人工30分钟内回电 链路日志(11步): intake:第1轮 - classify:tech - kb:未命中 - svc:正常 - agg:完成 - intake:第2轮 - classify:tech - kb:未命中 - svc:正常 - agg:完成 - fallback:升级第三张工单的信息量最大逐帧拆给你看失败路由search_kb没查到但它没有raise炸掉全图而是写入errorkb_miss把怎么办的决策权交给下游条件边——这就是Part 2讲的失败也是一种正常状态环形拓扑final_check看到error且attempt2把执行流指回intake图转了一圈Reducer的坑注意第二圈的aggregate合并了2条线索——findings用了operator.add追加式合并重试导致服务状态:正常出现两次。这是我把Reducer比作双刃剑的原因并行合并方便重试去重要自己管。生产代码里我会给findings加去重或在intake里重置。三个最值得记住的设计决策1. 决策节点只判断不路由。classify明明可以顺手把路由也做了为什么拆开因为判断业务逻辑和路由图结构变化频率不同意图规则天天改图拓扑很少动。拆开后改意图规则不用碰图。这是Part 2关注点分离原则在代码里的样子。2. Send的分发是运行时决定的。route_category返回的是[Send(...), Send(...)]这个列表——如果业务需要查3个系统改这个列表就行图结构一个字不用动。静态add_edge做不到这一点这就是动态分发边存在的意义第47、49节会把它玩出花。3. 重试预算防死循环。环形拓扑最怕转不停所以attempt计数 final_check里的上限判断是标配。LangGraph有recursion_limit兜底默认25步但那是保险丝不是业务逻辑——用业务的 attempt 上限优雅退出让框架的 recursion_limit 只管真正的Bug。30分钟学习法怎么消化这个文件标题说30分钟我给你拆好了0-10分钟跑代码只改tickets列表里的文本观察路由变化试试把退款改成别的词10-20分钟对着概念对照表逐行标注代码亲手在classify里加一个complaint投诉意图和一个专属节点20-30分钟把search_kb里的KB字典换成你业务里真实的三条知识感受这玩意儿真能用。做完这三步你已经完成了从看懂拓扑图到改得出拓扑图的跃迁。第10章的全部实战本质都是这个文件的加厚版把模拟函数换成真LLM调用、把三个节点换成十五个、把重试预算换成更精细的策略——骨架一模一样。小结Part 2全部核心概念在LangGraph里的落点节点4类函数节点四种职责、边3类add_edge/条件边/Send、共享状态TypedDictReducer、失败路由错误转状态条件边出口四步固定套路定义状态 → 写节点 → 写路由 → 装配图以后每个LangGraph应用都是这个骨架三个工程铁律决策与路由分离、Send分发运行时决定、环形拓扑必配重试预算Reducer追加式合并在重试场景会翻车findings重复去重逻辑自己管。下节预告概念全部落地了接下来的两节我们做开工准备第44节把开发环境从零搭起来Python版本、依赖清单、工程骨架第45节深入StateGraph的五脏六腑——状态定义、传递机制与Reducer的深度拆解。弹药备齐后第10章五连实战正式开打——第一战就是最经典的Prompt Chaining多步推理流水线。互动区留个作业检验你是不是真消化了给这张图加一个投诉意图classify识别投诉关键词 → 新节点complaint_desk送优惠券安抚→ END。改完在评论区贴你的节点函数和边装配代码我来批改第三张工单的findings重复问题你会怎么修在intake里重置还是在aggregate里去重说说你的理由。跑通的同学扣个1卡住的同学贴报错下节见。