LangGraph 工作流模式实践从线性链到动态协作当应用从“一问一答”走向长流程任务真正棘手的通常不是让模型多生成几句话而是让整个过程可控、可追踪、可恢复。一个成熟的智能应用需要知道当前处于哪一步、下一步该走哪条路径、失败后是否可以重试以及多个任务的结果如何合并。LangGraph 的价值就在于把这些工程问题显式地建模为图状态State保存共享数据节点Node负责一个具体动作边Edge决定执行顺序或分支。围绕这个模型可以组合出几种非常实用的工作流模式。先建立统一的心智模型可以把一次工作流想象成一张物流网络包裹信息是状态分拣站和配送站是节点运输路线是边。状态记录输入、处理中间结果、错误信息和最终输出。它是节点之间交换信息的唯一媒介。节点通常就是一个函数接收当前状态并返回需要更新的字段。节点最好保持单一职责。边连接节点既可以是固定边也可以根据状态选择下一步的条件边。汇聚多个分支完成后需要一个汇总节点把结果合成后续步骤可以使用的结构。这种拆分让流程从“模型自己决定一切”变成“模型负责判断程序负责组织”。下面的五种模式实际上就是在状态、节点和边之上形成的不同拓扑结构。1. 提示链让每一步都建立在上一步之上提示链Prompt Chaining适合有明显先后依赖的任务。前一个节点的输出会成为后一个节点的输入整个过程像内容生产流水线先列提纲再写初稿然后润色最后整合。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassContentState(TypedDict):topic:stroutline:strdraft:strpolished:strfinal:strdefmake_outline(state:ContentState):return{outline:llm(为主题生成提纲state[topic])}defmake_draft(state:ContentState):return{draft:llm(f根据提纲写初稿{state[outline]})}defpolish(state:ContentState):return{polished:llm(f润色以下初稿{state[draft]})}deffinalize(state:ContentState):return{final:llm(f整合成最终稿{state[polished]})}builderStateGraph(ContentState)builder.add_node(outline,make_outline)builder.add_node(draft,make_draft)builder.add_node(polish,polish)builder.add_node(finalize,finalize)builder.add_edge(START,outline)builder.add_edge(outline,draft)builder.add_edge(draft,polish)builder.add_edge(polish,finalize)builder.add_edge(finalize,END)graphbuilder.compile()提示链的优点是逻辑清楚、容易调试每个阶段都可以单独评估。代价是总延迟会叠加任意一步的错误也可能传递到后续。因此节点之间最好传递结构化字段而不是一段难以解析的大文本同时可以通过输入模式和输出模式隐藏内部中间结果只向调用方暴露最终内容。2. 并行化把互不依赖的工作同时完成当一个问题需要从多个独立角度分析时顺序执行没有必要。并行化Parallelization会让多个节点同时运行最后由汇总节点合并结果。例如评估一款城市通勤电动自行车可以把市场、竞品和技术分析拆成三个独立任务fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassAnalysisState(TypedDict):concept:strmarket:strcompetitor:strtechnology:strreport:strdefmarket_task(state):return{market:llm(分析用户需求和市场趋势state[concept])}defcompetitor_task(state):return{competitor:llm(分析竞品优劣state[concept])}deftechnology_task(state):return{technology:llm(分析实现所需的技术方案state[concept])}defcombine(state):return{report:\n\n.join([state[market],state[competitor],state[technology]])}builderStateGraph(AnalysisState)builder.add_node(market,market_task)builder.add_node(competitor,competitor_task)builder.add_node(technology,technology_task)builder.add_node(combine,combine)builder.add_edge(START,market)builder.add_edge(START,competitor)builder.add_edge(START,technology)builder.add_edge(market,combine)builder.add_edge(competitor,combine)builder.add_edge(technology,combine)builder.add_edge(combine,END)graphbuilder.compile()并行化通常能降低总耗时也能让最终决策拥有更丰富的证据。但并行节点必须尽量独立写入共享资源时要考虑幂等性和冲突汇总节点还应处理某个分支超时或失败的情况例如保留成功结果、记录失败原因再决定是否重试。3. 路由先判断类型再进入专用分支路由Routing模式也可以理解为智能分流。系统先分析输入属于哪一类再把请求交给对应的处理器。智能客服是典型例子售前咨询、售后问题和技术故障应该由不同的提示词、工具和权限处理。关键是让路由结果结构化避免后续代码依赖模型生成的自然语言fromtypingimportLiteral,TypedDictfrompydanticimportBaseModel,Fieldfromlanggraph.graphimportStateGraph,START,ENDclassRoute(BaseModel):step:Literal[pre_sale,after_sale,technical]Field(description选择最匹配的处理分支)classSupportState(TypedDict):user_input:strdecision:stroutput:strdefclassify(state):routemodel.with_structured_output(Route).invoke(state[user_input])return{decision:route.step}defchoose_next(state):return{pre_sale:pre_sale,after_sale:after_sale,technical:technical,}.get(state[decision],fallback)builderStateGraph(SupportState)builder.add_node(classify,classify)builder.add_node(pre_sale,lambdas:{output:进入售前处理})builder.add_node(after_sale,lambdas:{output:进入售后处理})builder.add_node(technical,lambdas:{output:进入技术处理})builder.add_node(fallback,lambdas:{output:转人工确认问题类型})builder.add_edge(START,classify)builder.add_conditional_edges(classify,choose_next,[pre_sale,after_sale,technical,fallback])fornodein(pre_sale,after_sale,technical,fallback):builder.add_edge(node,END)graphbuilder.compile()路由节点的质量决定了整个系统的上限。生产环境应为低置信度、无法分类和多意图输入准备兜底分支并在不同分支中限制可用工具避免客服模型越权执行操作。4. 协调者-工作者运行时拆解未知数量的任务并行化适合任务数量在设计时就确定的场景如果任务数量和内容需要先分析才能知道就需要协调者-工作者Orchestrator-Workers模式。它通常包含三个角色协调者根据主题或数据特征制定计划。工作节点分别完成计划中的子任务。合成器收集所有结果并生成最终输出。例如处理一份结构未知的长文档协调者先识别出需要处理的章节再为每个章节动态创建工作任务。LangGraph 可以用Send在运行时生成数量不固定的边fromtypingimportAnnotated,TypedDictimportoperatorfromlanggraph.typesimportSendclassReportState(TypedDict):topic:strsections:listcompleted:Annotated[list,operator.add]final_report:strdeforchestrator(state):planplanner.invoke(f为主题生成报告结构{state[topic]})return{sections:plan.sections}defassign_workers(state):return[Send(worker,{section:section})forsectioninstate[sections]]defworker(state):sectionstate[section]contentllm(f完成章节{section.name}{section.description})return{completed:[content]}defsynthesizer(state):return{final_report:\n\n---\n\n.join(state[completed])}这里的难点不在于“开几个并发任务”而在于任务边界和结果管理。每个工作任务应带有可追踪的标识汇总时要考虑顺序是否重要同时应设置并发上限、超时和失败重试避免一次计划生成过多任务而耗尽资源。5. 评估器-优化器用反馈驱动循环改进评估器-优化器Evaluator-Optimizer适合结果质量无法一次确定的任务。流程先生成内容再由评估器按照明确标准打分或给出反馈不合格就带着反馈重新生成直到通过或达到最大迭代次数。fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassReviewState(TypedDict):request:stranswer:strfeedback:strpassed:boolattempts:intdefgenerate(state):promptf需求{state[request]}\n反馈{state.get(feedback,)}return{answer:llm(prompt),attempts:state.get(attempts,0)1}defevaluate(state):reviewreviewer.invoke(state[answer])return{passed:review.passed,feedback:review.feedback}defnext_step(state):ifstate[passed]orstate[attempts]3:returnfinishreturngeneratebuilderStateGraph(ReviewState)builder.add_node(generate,generate)builder.add_node(evaluate,evaluate)builder.add_node(finish,lambdas:{})builder.add_edge(START,generate)builder.add_edge(generate,evaluate)builder.add_conditional_edges(evaluate,next_step,[generate,finish])builder.add_edge(finish,END)graphbuilder.compile()检索增强问答中的“相关性检查”就是这种思路如果检索结果与问题不相关先重写问题再检索如果质量合格才生成答案。评估器必须使用可操作的标准例如事实一致性、引用完整性、格式约束而不是只让模型给一个模糊的“看起来不错”。此外一定要设置最大循环次数并保留每轮的答案和反馈方便定位质量波动。五种模式如何选择模式适合的问题决策发生在哪里主要优势主要风险提示链步骤有明确依赖设计时简单、可预测延迟叠加、错误传播并行化多个固定且独立的任务设计时降低耗时、信息全面资源竞争、汇总失败路由输入类型不同、处理方式不同运行时分类专业分支、权限清晰分类错误、缺少兜底协调者-工作者子任务数量和内容未知运行时规划灵活扩展、天然并行任务爆炸、顺序难保证评估器-优化器需要反复改进质量运行时评估结果可迭代提升循环失控、成本增加真实应用通常会组合这些模式而不是只使用一种。例如一个知识库助手可以先用路由判断“直接回答”还是“需要检索”再并行查询多个数据源最后用评估器检查证据相关性通过检查后再用提示链完成答案生成、格式化和摘要。落地时的工程检查清单先画出状态流转图再写节点函数。每个节点只负责一类动作输入和输出字段尽量明确。对模型输出使用结构化模式尤其是路由决策、评估结果和协调者计划。为并行分支和动态工作者设置超时、重试和并发上限涉及外部写操作时保证幂等。为所有循环设置退出条件并把迭代次数、反馈和失败原因写入状态。保留中间状态和节点日志。出现错误时能够回答“在哪个节点、基于什么输入、做出了什么决策”。将用户可见字段与内部中间字段分开避免把敏感数据或调试信息意外返回给调用方。工作流模式的核心不是把图画得复杂而是把复杂性放在正确的位置用固定边保证确定性用条件边表达选择用状态承载上下文用循环和反馈提升质量。掌握这几种基本结构后面对新的智能应用需求时就能先做流程设计再决定模型、工具和提示词系统也更容易测试、扩展和长期维护。