引言当你发现AI工作流需要循环和人工审核时LangChain的链式思维就开始捉襟见肘了。一、那个让我朋友放弃Chain的项目今年年初我朋友接了一个他们公司内部AI审批系统的活儿。需求听起来不复杂员工提交报销单AI先初审一遍判断票据是否合规、金额是否合理。如果AI觉得没问题直接自动通过如果AI拿不准就转给财务人工复核。我朋友一开始用LangChain的Chain来实现。第一步把报销单传给模型做初审第二步根据模型的输出决定是通过还是转人工。代码大概长这样# 第一步AI初审review_resultreview_chain.invoke({expense:expense_data})# 第二步判断if通过inreview_result:auto_approve(expense_data)else:send_to_human(review_result)上线第一周财务主管在群里我朋友“这个AI初审太严了十张单子有八张被转人工我们根本审不过来。能不能让AI先打标签标出具体问题我们再决定是打回修改还是直接通过”我朋友想了想这需求不难加一步就行AI初审→打标签→人工判断→如果打回修改员工改完再提交→AI再审→……等等这里出现了一个循环。员工改完重新提交整个流程要再走一遍。而且如果第二次AI还是拿不准还得再转人工。他试着用Chain实现这个循环。Chain是什么是流水线。原料从A工位流到B工位再到C工位一路向前不能回头。要让Chain支持循环他得在外面包一层while循环手动维护状态判断什么时候退出。代码很快变成了一团乱麻whileTrue:resultchain.invoke(state)ifresult[status]approved:breakelifresult[status]rejected:stateget_user_fix(state)elifresult[status]human_review:statewait_for_human(state)ifiteration5:break更头疼的是状态管理。每一轮循环报销单的内容、AI的初审意见、人工的复核意见、员工的修改记录这些信息散落在不同的变量里没有统一的地方存放。一旦流程中断比如服务重启所有中间状态全丢了用户得从头再来。他折腾了三天代码越来越复杂bug越来越多。最后我不得不承认Chain的线性思维根本承载不了这种有循环、有分支、需要持久化状态的复杂工作流。那用Agent呢Agent不是能自主决策吗他试了一下LangChain的ReAct Agent。确实Agent可以自己决定调用什么工具看起来挺智能的。但问题是——Agent是个黑盒。你只知道它输入了一个问题输出了一个答案中间经历了多少轮思考、调用了哪些工具、为什么走这条路而不是那条路你完全看不到也控制不了。财务主管提了一个很合理的要求AI初审金额超过5000元的单子必须人工复核不能自动通过。“这个规则在Agent里怎么实现Agent的决策逻辑藏在模型的内心独白里你没法强行插入一条硬规则”。要么在Prompt里反复强调但模型可能听也可能不听要么在Agent外层再包一层判断又回到Chain的老路。那一刻他明白了Chain太死板Agent太黑盒。我们需要一种既能精确控制流程、又能灵活处理分支和循环、还能持久化状态的编排方式。这就是LangGraph诞生的原因。二、从流水线到交通网LangGraph到底是什么如果把LangChain的Chain比作工厂里的流水线那LangGraph就是城市里的交通网。流水线的特点是起点固定、终点固定、中间每个工位的顺序固定。原料从入口进去按A→B→C→D的顺序依次加工最后从出口出来。它适合简单、线性、步骤明确的任务比如翻译→润色→输出。但交通网不一样。交通网有十字路口、有立交桥、有单行道、有环岛。车辆从A点出发可以根据路况选择走B路还是C路如果B路堵了可以掉头回A重新选甚至可以在D点停下来等人等完再继续走。LangGraph的核心思想就是把AI工作流建模成一个有向图节点Node是路口是执行具体任务的单元。比如AI初审是一个节点人工复核是一个节点员工修改也是一个节点。边Edge是路是节点之间的连接。它定义了任务执行完以后下一步该往哪儿走。条件边Conditional Edge是红绿灯是根据实时情况动态决定方向的开关。比如AI初审节点执行完后如果金额小于5000且票据合规走自动通过这条路否则走人工复核这条路。状态State是车里拉的货是在整个交通网中流转的数据。它从起点装载经过每个节点时被加工、被补充最终带着完整的信息到达终点。这种图结构的表达能力远超线性链。理论上图编排模型等价于图灵机而链式编排模型只等价于有限状态自动机。换句话说图能表达任何可计算的流程而链不能。三、LangGraph的四大核心概念理解了定位我们来看LangGraph的四大核心概念。掌握这四个概念你就掌握了LangGraph的骨架。概念一StateGraph——状态图整个交通网的蓝图StateGraph是LangGraph中定义工作流的起点。它相当于一张空白图纸你在上面画节点、连边最终拼成一张完整的交通网。fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict# 第一步定义状态结构classApprovalState(TypedDict):expense_data:dict# 报销单原始数据ai_review:str# AI初审意见risk_tags:list# AI打的风险标签human_decision:str# 人工复核结果iteration:int# 循环次数防死循环# 第二步创建状态图workflowStateGraph(ApprovalState)ApprovalState这个TypedDict定义了整辆货车能拉什么货。所有节点共享同一个State每个节点读取自己需要的字段加工后写回新的字段。这种设计让数据流变得异常清晰——你知道每个阶段产生了什么数据数据在节点之间怎么传递。概念二Node——节点路口上的执行单元节点是图中最基础的执行单元本质上就是一个Python函数。它接收当前State执行业务逻辑返回一个字典来更新State。defai_review_node(state:ApprovalState)-dict:AI初审节点读取报销单输出初审意见和风险标签expensestate[expense_data]# 调用LLM进行初审reviewllm.invoke(f请审核以下报销单{expense})# 解析出风险标签tagsextract_risk_tags(review)return{ai_review:review,risk_tags:tags,iteration:state.get(iteration,0)1}节点的设计原则是单一职责。一个节点只做一件事要么读取数据要么调用模型要么调用工具要么做判断。不要把初审判断通知三件事塞在一个节点里那样图结构就失去了意义。概念三Edge——边连接节点的路边定义了节点之间的执行顺序。最简单的边是无条件边A节点执行完固定去B节点。# 普通边AI初审完后固定进入人工复核节点workflow.add_edge(ai_review,human_review)但真实世界中大部分流程都需要根据状态动态决定下一步。这就需要条件边。概念四Conditional Edge——条件边交通网上的红绿灯条件边是LangGraph的灵魂。它允许你根据当前State的内容动态选择下一个节点。defroute_after_ai_review(state:ApprovalState)-str:根据AI初审结果决定下一步走向tagsstate.get(risk_tags,[])amountstate[expense_data].get(amount,0)iterationstate.get(iteration,0)# 硬规则金额超过5000必须人工复核ifamount5000:returnhuman_review# 如果AI打了高风险标签也转人工if高风险intagsor票据异常intags:returnhuman_review# 如果已经循环超过3次还没过强制转人工ifiteration3:returnhuman_review# 其他情况自动通过returnauto_approve# 添加条件边从ai_review节点出发根据route_after_ai_review的返回值决定去向workflow.add_conditional_edges(ai_review,route_after_ai_review,{human_review:human_review,# 返回值human_review跳转到human_review节点auto_approve:auto_approve# 返回值auto_approve跳转到auto_approve节点})条件边的强大之处在于路由逻辑完全由你控制。不是让模型在黑盒里猜而是你用代码写死规则。金额阈值、风险标签、循环次数这些业务规则清清楚楚地写在route_after_ai_review函数里可审计、可测试、可修改。四、实战搭建一个报销审批流把上面的概念串起来我们搭建一个完整的报销审批工作流。fromlanggraph.graphimportStateGraph,END,STARTfromlanggraph.checkpoint.memoryimportMemorySaverfromtypingimportTypedDictfromlangchain_openaiimportChatOpenAI# 1. 定义状态 classApprovalState(TypedDict):expense_data:dictai_review:strrisk_tags:listhuman_decision:striteration:intfinal_status:str# 2. 定义节点 llmChatOpenAI(modelgpt-4o,temperature0)defai_review_node(state:ApprovalState)-dict:AI初审审核报销单输出意见和标签expensestate[expense_data]promptf你是一位财务初审AI。请审核以下报销单判断票据合规性和金额合理性。 输出格式 意见通过/需复核 标签合规/票据异常/金额超限/类别不符 报销单{expense}responsellm.invoke(prompt)contentresponse.content tags[]if票据异常incontent:tags.append(票据异常)if金额超限incontent:tags.append(金额超限)if类别不符incontent:tags.append(类别不符)return{ai_review:content,risk_tags:tags,iteration:state.get(iteration,0)1}defhuman_review_node(state:ApprovalState)-dict:人工复核模拟财务人员的判断实际项目中这里是人工审批接口tagsstate.get(risk_tags,[])# 模拟人工判断逻辑if票据异常intags:return{human_decision:打回修改,final_status:rejected}elif金额超限intags:return{human_decision:通过但备注,final_status:approved_with_note}else:return{human_decision:通过,final_status:approved}defauto_approve_node(state:ApprovalState)-dict:自动通过AI初审无风险直接审批return{final_status:approved,human_decision:AI自动通过}defnotify_node(state:ApprovalState)-dict:通知节点发送审批结果通知statusstate.get(final_status,unknown)print(f【通知】报销单审批结果{status})return{}# 3. 定义条件路由 defroute_after_ai(state:ApprovalState)-str:AI初审后的路由逻辑tagsstate.get(risk_tags,[])amountstate[expense_data].get(amount,0)iterationstate.get(iteration,0)# 硬规则金额5000必须人工复核ifamount5000:returnhuman# 有风险标签转人工iftags:returnhuman# 循环超限强制人工ifiteration3:returnhumanreturnauto# 4. 组装图 workflowStateGraph(ApprovalState)# 注册节点workflow.add_node(ai_review,ai_review_node)workflow.add_node(human_review,human_review_node)workflow.add_node(auto_approve,auto_approve_node)workflow.add_node(notify,notify_node)# 设置入口workflow.add_edge(START,ai_review)# 条件边AI初审后分流workflow.add_conditional_edges(ai_review,route_after_ai,{human:human_review,auto:auto_approve})# 人工复核后统一进入通知节点workflow.add_edge(human_review,notify)# 自动通过后也进入通知节点workflow.add_edge(auto_approve,notify)# 通知后结束workflow.add_edge(notify,END)# 5. 编译并运行 # 加上MemorySaver状态持久化到内存生产环境可换成Redis/SQLitecheckpointerMemorySaver()appworkflow.compile(checkpointercheckpointer)# 测试一张3000元的合规报销单resultapp.invoke({expense_data:{amount:3000,category:差旅,receipt:合规},iteration:0},config{configurable:{thread_id:expense_001}})print(result[final_status])# 输出approved# 测试一张8000元的大额报销单resultapp.invoke({expense_data:{amount:8000,category:招待,receipt:合规},iteration:0},config{configurable:{thread_id:expense_002}})print(result[final_status])# 输出approved_with_note 或 approved这段代码的精妙之处在于流程可视化你把节点和边画出来就是一张清晰的流程图。AI初审→条件判断→人工/自动→通知→结束。任何人看一眼代码就能理解业务逻辑。状态全链路追踪thread_id让每次审批都有独立的会话状态。如果审批流程中断比如服务重启只要用相同的thread_id重新调用LangGraph会从断点继续执行而不是从头开始。规则硬编码金额超过5000转人工这不是让模型看着办而是代码里写死的if amount 5000。业务规则可控、可审计。循环天然支持如果未来需求变成人工打回修改→员工重新提交→AI再审只需要在human_review_node返回时增加一个修改分支连一条边回到ai_review节点就是一个完整的循环。不需要在外层包while不需要手动维护状态。五、LangGraph不是银弹什么时候该用什么时候不该用写到这里有必要再冷静复盘一下。LangGraph很强大但不是所有场景都需要上交通网。适合用LangGraph的场景流程有循环或回溯。比如生成初稿→审核→不达标→修改→再审这种循环用Chain几乎无法实现用Agent又太黑盒。需要人工介入Human-in-the-Loop。比如审批、审核、确认环节流程必须停下来等人人操作完再继续。LangGraph的interrupt机制专门支持这种场景。状态需要持久化和恢复。比如一个审批流程可能持续三天中间服务重启了无数次状态必须能存能取。LangGraph的Checkpointer支持SQLite、Redis、PostgreSQL等多种后端。需要并行执行。比如同时查天气、查汇率、查股价三个节点互不依赖可以并行跑最后汇总结果。图结构天然支持这种并行编排。不适合用LangGraph的场景纯线性的简单任务。比如翻译→润色→输出用LCEL的prompt | llm | parser一行搞定上LangGraph是杀鸡用牛刀。对延迟极度敏感的单轮问答。图编排有额外的调度和状态管理开销简单场景下不如裸调API快。团队还没掌握LangChain基础。LangGraph的学习曲线比Chain陡如果团队连PromptTemplate和LCEL都没搞熟直接上LangGraph会事倍功半。六、写在最后从搭积木到画电路图LangChain 1.0之后整个框架的定位发生了本质变化LangChain退居幕后成为提供积木的零件库LangGraph走到台前成为搭建乐高的电路图。这个变化反映了一个深层趋势AI应用开发正在从单点调用走向系统工程。早期的开发者关心的是怎么调模型现在的开发者关心的是怎么编排一个可靠的、可审计的、可恢复的工作流。Chain和Agent解决了让AI动起来的问题LangGraph解决的是让AI工作流可控、可观测、可维护的问题。如果你今天还在用Chain硬凑循环、用Agent黑盒赌运气不妨花一个周末学学LangGraph。当你第一次画出一张StateGraph看到节点和边清清楚楚地拼成一张流程图看到状态在节点之间有序流转看到服务重启后流程从断点无缝恢复——你会明白这不是又一个框架这是AI工作流编排的成人礼。