工作流常见模式上提示链、并行化、路由从本篇开始我们进入「工作流常见模式」章节。这些模式是预先定义好的执行路径就像工厂流水线每个步骤都有明确的输入输出和顺序。本篇讲最基础、最常用的三种提示链 Prompt Chaining、并行化 Parallelization、路由 Routing。目录模式一提示链 Prompt Chaining模式二并行化 Parallelization模式三路由 Routing三种模式对比本篇小结1. 模式一提示链 Prompt Chaining1.1 概念提示链就像流水线前一个步骤的输出作为下一个步骤的输入。内容创作场景最典型大纲 → 初稿 → 润色 → 最终稿每一步的结果传给下一步内容质量逐步提升。START → 大纲生成 → 初稿生成 → 润色文章 → 最终稿生成 → END1.2 状态设计用输入输出模式fromtyping_extensionsimportTypedDict# 输入只含用户主题classInputState(TypedDict):topic:str# 输出只含最终内容classOutputState(TypedDict):final_content:str# 内部完整状态保存每一步的产物classOverallState(InputState,OutputState):outline:str# 大纲draft:str# 初稿polished_draft:str# 润色后的稿件1.3 节点实现fromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessage modelinit_chat_model(gpt-4o-mini)defgenerate_outline(state:InputState):outlinemodel.invoke([HumanMessage(contentf根据主题生成文章大纲只要两个核心标题{state[topic]})]).contentreturn{outline:outline,topic:state[topic]}defgenerate_draft(state:OverallState):draftmodel.invoke([HumanMessage(contentf根据大纲写初稿每个标题下最多三句话\n{state[outline]})]).contentreturn{draft:draft}defpolish_content(state:OverallState):polishedmodel.invoke([HumanMessage(contentf润色下面的初稿文章不要太长\n{state[draft]})]).contentreturn{polished_draft:polished}deffinalize_content(state:OverallState):finalmodel.invoke([HumanMessage(contentf根据润色稿生成终稿\n{state[polished_draft]})]).contentreturn{final_content:final}1.4 组装fromlanggraph.graphimportStateGraph,START,END builderStateGraph(OverallState,input_schemaInputState,output_schemaOutputState)builder.add_node(generate_outline,generate_outline)builder.add_node(generate_draft,generate_draft)builder.add_node(polish_content,polish_content)builder.add_node(finalize_content,finalize_content)builder.add_edge(START,generate_outline)builder.add_edge(generate_outline,generate_draft)builder.add_edge(generate_draft,polish_content)builder.add_edge(polish_content,finalize_content)builder.add_edge(finalize_content,END)chainbuilder.compile()resultchain.invoke({topic:人工智能的未来发展})print(result[final_content])适用场景任务能清晰拆成几个顺序固定、逐步加工的步骤写作、数据处理流水线。不适用步骤之间没有先后依赖、或需要根据中间结果改变路径。2. 模式二并行化 Parallelization2.1 概念并行化指多个任务同时进行最后汇总结果。多角度处理同一个问题时常用既省时又能提升决策质量。案例研发一款主打城市通勤的智能电动自行车导航、社交、防盗研发前要做多维度分析市场分析用户关注续航、车重、防盗对「骑行社交」感兴趣竞品分析传统品牌智能化不足互联网品牌续航和售后是短板技术分析轻量化电池、车身设计、GPS 防盗与社交 App 集成。┌→ 市场分析 ─┐ START ──┼→ 竞品分析 ─┼→ 汇总报告 → END └→ 技术分析 ─┘2.2 完整代码fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassAnalysisState(TypedDict):concept:strmarket:strcompetitor:strtech:strreport:str# 三个并行分析任务defmarket_task(state:AnalysisState):return{market:用户关注续航、重量、防盗对骑行社交有兴趣...}defcompetitor_task(state:AnalysisState):return{competitor:传统品牌智能化不足互联网品牌续航和售后差...}deftech_task(state:AnalysisState):return{tech:轻量化电池车身、GPS防盗、社交App集成...}# 汇总defcombine_results(state:AnalysisState):reportf产品分析报告\n\nreportf市场分析\n{state[market]}\n\nreportf竞品分析\n{state[competitor]}\n\nreportf技术分析\n{state[tech]}\n\nreport建议聚焦续航、防盗、社交功能的平衡发展return{report:report}builderStateGraph(AnalysisState)builder.add_node(market,market_task)builder.add_node(competitor,competitor_task)builder.add_node(tech,tech_task)builder.add_node(combine,combine_results)# 关键从 START 分别连到三个节点 → 它们并行执行builder.add_edge(START,market)builder.add_edge(START,competitor)builder.add_edge(START,tech)# 三个节点都连到汇总builder.add_edge(market,combine)builder.add_edge(competitor,combine)builder.add_edge(tech,combine)builder.add_edge(combine,END)workflowbuilder.compile()resultworkflow.invoke({concept:城市通勤智能电动自行车})print(result[report])关键点从同一个节点这里是 START连出多条边到不同节点这些节点就会并行执行它们都连到汇总节点等全部完成后才汇总。适用场景多个子任务互相独立、可同时进行最后要合并多视角分析、批量处理。不适用子任务之间有先后依赖必须等上一个结果。3. 模式三路由 Routing3.1 概念路由模式也叫「智能分流」根据输入内容决定执行哪个分支。最典型的是智能客服按问题自动分类处理START → 分析问题类型 ┬─ 售前问题 → 售前处理 → END ├─ 售后问题 → 售后处理 → END └─ 技术问题 → 技术处理 → END核心设计动态路径选择LLM 结构化输出决定走哪条分支隔离各类问题由专用处理器处理。3.2 完整代码fromtyping_extensionsimportLiteral,TypedDictfrompydanticimportBaseModel,Fieldfromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.graphimportStateGraph,START,ENDclassState(TypedDict):input:strdecision:stroutput:str# 路由决策的数据结构结构化输出classRoute(BaseModel):step:Literal[pre_sale,after_sale,technical]Field(description根据用户问题类型决定路由到售前、售后还是技术处理)modelinit_chat_model(gpt-4o-mini)# 路由决策节点defmodel_call_router(state:State):decisionmodel.with_structured_output(Route).invoke(state[input])return{decision:decision.step}# 三个处理节点defpre_sale_handler(state:State):return{output:售前咨询已处理处理内容.....}defafter_sale_handler(state:State):return{output:售后问题已处理处理内容.....}deftechnical_handler(state:State):return{output:技术问题已处理处理内容.....}# 路由函数根据决策返回下一个节点defroute_decision(state:State)-str:ifstate[decision]pre_sale:returnpre_sale_handlerelifstate[decision]after_sale:returnafter_sale_handlerreturntechnical_handler# 构建图builderStateGraph(State)builder.add_node(pre_sale_handler)builder.add_node(after_sale_handler)builder.add_node(technical_handler)builder.add_node(model_call_router)builder.add_edge(START,model_call_router)# 条件边根据路由结果选分支builder.add_conditional_edges(model_call_router,route_decision,[pre_sale_handler,after_sale_handler,technical_handler],)builder.add_edge(pre_sale_handler,END)builder.add_edge(after_sale_handler,END)builder.add_edge(technical_handler,END)router_workflowbuilder.compile()# 测试test_cases[我想了解一下你们产品的价格和功能,# 售前我购买的产品有质量问题需要退货,# 售后这个软件安装后无法正常运行报错0x80070005,# 技术]forcaseintest_cases:resultrouter_workflow.invoke({input:case})print(f用户问题{case}\n{result[output]}\n)适用场景输入有明显类别之分、不同类别处理逻辑完全不同客服分流、意图分类。不适用类别边界模糊、或所有输入都走同一套处理。4. 三种模式对比模式核心特征一句话识别提示链步骤顺序固定上一步输出喂下一步「一步步加工」并行化多个独立任务同时跑、最后汇总「分头干、一起收」路由按输入内容分流到不同分支「看类型、走不同门」5. 本篇小结提示链上一步输出作为下一步输入适合顺序固定、逐步加工的任务常用输入输出模式定义状态。并行化从同一节点连多条边到不同节点即并行全部连到汇总节点等齐后合并适合互相独立的子任务。路由用 LLM结构化输出做分类再用条件边分流到专用处理器适合意图分类、客服分流。三者可以组合使用比如路由后某个分支内部再并行。下一篇讲另外两种更高级的模式协调者-工作者运行时动态派活和评估器-优化器生成-打分-改进循环。