5种AI工作流设计模式：从顺序链到智能代理，一文掌握Workflow编排

引言随着大语言模型LLM的爆发式增长AI 应用已经从“单次问答”进化到“多步骤协同”的阶段。一个典型场景用户输入一个自然语言需求系统先调用模型理解意图再查询数据库然后让模型整理格式最后输出报告——这背后依赖的就是AI工作流AI Workflow。然而很多开发者在实现时陷入“面条式代码”难以维护和扩展。本文将系统梳理 AI 工作流中最经典的5种设计模式顺序链Sequential Chain、路由链Router Chain、并行链Parallel Chain、编排器-工作者Orchestrator-Worker和智能代理Agent。每种模式都会讲解核心思想、适用场景并提供一个完整可运行的 Python 代码示例。读完本文你将能像搭积木一样灵活组装自己的 AI 流水线。核心概念什么是AI工作流设计模式AI 工作流是指将多个 AI 调用如 LLM 推理、工具调用、数据检索与业务逻辑组合成一个可复用的执行单元。设计模式就是对这些常见编排方式的抽象。为什么需要设计模式解耦将不同的处理步骤分离每个步骤职责单一。复用相同的链/节点可以在不同场景中复用。可观测每一步的输入输出清晰可追踪。灵活扩展方便添加新的处理逻辑而不影响现有流程。在代码层面本文基于 LangChain 框架但设计模式是框架无关的。我们将使用ChatOpenAI作为 LLM并采用环境变量设置 API Key。请确保安装依赖pip install langchain langchain-openai python-dotenv并在.env文件中设置OPENAI_API_KEY。实战示例五种模式逐一攻克1. 顺序链Sequential Chain模式描述多个步骤按固定顺序线性执行上一步的输出是下一步的输入。就像一条流水线每个工位加工后交给下一个。适用场景文章生成大纲→正文→润色、数据ETL提取→清洗→分析。代码示例import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import StrOutputParser from langchain.schema.runnable import RunnablePassthrough load_dotenv() # 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 第一步生成大纲 outline_prompt ChatPromptTemplate.from_template( 请为以下主题生成一个简洁的文章大纲3个小标题{topic} ) outline_chain outline_prompt | llm | StrOutputParser() # 第二步根据大纲写正文 content_prompt ChatPromptTemplate.from_template( 根据以下大纲撰写详细正文每个小标题展开100字左右\n{outline} ) content_chain content_prompt | llm | StrOutputParser() # 第三步润色 polish_prompt ChatPromptTemplate.from_template( 请润色以下文本使其更流畅优美保留原意\n{text} ) polish_chain polish_prompt | llm | StrOutputParser() # 组合成顺序链 final_chain ( {outline: outline_chain, topic: RunnablePassthrough()} | ( lambda inputs: { text: content_chain.invoke({outline: inputs[outline]}), outline: inputs[outline], } ) | polish_chain ) # 运行 topic AI在医疗领域的应用 result final_chain.invoke(topic) print(result)这里我们使用了 LangChain Expression Language (LCEL) 将多个链串联。RunnablePassthrough用于透传原始输入。注意第三步的输入构造方式通过一个匿名函数将前两步的结果传递给润色链。这种模式清晰展示了数据在流水线上的传递。2. 路由链Router Chain模式描述根据输入条件动态选择不同的处理分支。一个“路由”组件负责分析输入并决定下一步执行哪个子链。适用场景多意图客服系统退款/订单查询/投诉、内容分类后调用不同处理逻辑。代码示例from langchain.prompts import PromptTemplate from langchain.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import Literal # 定义路由决策的结构 class RouterOutput(BaseModel): category: Literal[技术问题, 售后问题, 一般咨询] Field(description问题类别) reason: str Field(description分类理由) # 路由解析器 router_parser PydanticOutputParser(pydantic_objectRouterOutput) router_prompt PromptTemplate( template你是一个客服路由专家根据用户输入判断问题类别。 用户输入{input} {format_instructions} 请直接返回JSON格式的类别和理由。, input_variables[input], partial_variables{format_instructions: router_parser.get_format_instructions()} ) # 不同分支的处理链 tech_chain ( PromptTemplate.from_template(技术问题回答{input} 请提供排查步骤) | llm | StrOutputParser() ) aftersale_chain ( PromptTemplate.from_template(售后问题回答{input} 请表达歉意并给出解决方案) | llm | StrOutputParser() ) general_chain ( PromptTemplate.from_template(一般咨询回答{input} 请给出友好回复) | llm | StrOutputParser() ) def route(info: RouterOutput): if info.category 技术问题: return tech_chain elif info.category 售后问题: return aftersale_chain else: return general_chain # 构建完整路由链 full_chain ( {input: lambda x: x} | ( lambda inputs: { input: inputs[input], route_info: router_parser.parse( (router_prompt | llm | StrOutputParser()).invoke({input: inputs[input]}) ) } ) | ( lambda inputs: route(inputs[route_info]).invoke({input: inputs[input]}) ) ) # 测试 print(full_chain.invoke(我买的商品坏了怎么退货))代码中我们先用 LLM 解析用户输入得到分类然后通过route函数返回对应的子链并执行。这实现了动态分支而且每个子链可独立开发和测试。3. 并行链Parallel Chain模式描述将一个任务拆分为多个可以同时执行的子任务最后汇总结果。实际执行时子任务可能通过多线程、异步或分布式方式并行处理。适用场景多维度评测同时评估语法、逻辑、创意、多源信息收集并聚合。代码示例import asyncio from langchain.schema.runnable import RunnableLambda # 三个独立的评估链 grammar_prompt PromptTemplate.from_template(评估文本语法问题给出评分1-10\n{text}) grammar_chain grammar_prompt | llm | StrOutputParser() logic_prompt PromptTemplate.from_template(评估文本逻辑性给出评分1-10\n{text}) logic_chain logic_prompt | llm | StrOutputParser() creativity_prompt PromptTemplate.from_template(评估文本创意度给出评分1-10\n{text}) creativity_chain creativity_prompt | llm | StrOutputParser() # 并行执行函数 async def run_parallel(text): async_grammar grammar_chain.ainvoke({text: text}) async_logic logic_chain.ainvoke({text: text}) async_creativity creativity_chain.ainvoke({text: text}) results await asyncio.gather(async_grammar, async_logic, async_creativity) return { 语法评分: results[0], 逻辑评分: results[1], 创意评分: results[2] } # 封装为Runnable parallel_chain RunnableLambda(lambda text: asyncio.run(run_parallel(text))) # 测试 text 这个产品真的很好我非常喜欢强烈推荐 print(parallel_chain.invoke(text))这里使用了 Python 的asyncio和 LangChain 的异步调用方法ainvoke来实现真正的并发。三个评估同时请求 LLM总耗时约等于最慢的那个大幅提升速度。4. 编排器-工作者Orchestrator-Worker模式描述中心化的“编排器”动态分解任务并分配给多个“工作者”节点执行可能经过多轮交互。工作者执行完子任务后编排器可能继续拆解直到完成最终目标。适用场景复杂多步推理如旅行规划先查航班再查酒店再组合、自动化代码编写先拆分子函数再分别生成再合并。代码示例我们来做一个简单的旅行规划助手编排器先列出要完成的任务然后将每个任务交给工作者最后汇总。from langchain.prompts import ChatPromptTemplate from langchain.schema import StrOutputParser # 工作者执行单一任务 worker_prompt ChatPromptTemplate.from_template( 你是一个旅行助手请完成以下小任务{task}\n当前已有信息{context} ) worker_chain worker_prompt | llm | StrOutputParser() # 编排器根据用户目标生成任务列表 orchestrator_prompt ChatPromptTemplate.from_template( 你是一个旅行规划专家需要将用户的复杂需求拆解成3-5个具体的独立小任务 每个任务由不同工作者执行。请以JSON列表返回每个元素包含task_id和content。 用户需求{requirement} ) # 总的编排器-工作者流程 def orchestrator_worker_flow(requirement: str): # 1. 编排器生成任务列表 task_list_str (orchestrator_prompt | llm | StrOutputParser()).invoke({requirement: requirement}) # 简单解析JSON实际项目中可用更健壮的解析 import json tasks json.loads(task_list_str) # 2. 依次执行每个任务简单起见顺序执行实际可并行 context results {} for task in tasks: task_id task[task_id] res worker_chain.invoke({task: task[content], context: context}) results[task_id] res context f\n{task[content]}的结果{res} # 3. 编排器汇总 summary_prompt ChatPromptTemplate.from_template( 根据以下所有子任务结果为用户生成最终的旅行计划\n{results} ) final_plan (summary_prompt | llm | StrOutputParser()).invoke({results: str(results)}) return final_plan # 测试 requirement 我下周想从北京去上海玩3天预算5000元喜欢历史和美食。 final_plan orchestrator_worker_flow(requirement) print(final_plan)这个模式展示了动态规划能力编排器不是硬编码的步骤而是根据需求智能拆解任务然后协调多个工作者完成。真实场景中还可以加入循环、人工确认等机制。5. 智能代理Agent模式描述Agent 是最高级的模式。它拥有自主决策能力可以调用工具如搜索、计算器、数据库查询并根据工具返回的结果规划下一步直到完成用户目标。核心循环是思考 → 行动 → 观察 → 再思考……适用场景需要动态工具使用的任务如联网查询、复杂计算、需要与外部环境交互的自主系统。代码示例from langchain.agents import AgentExecutor, create_openai_functions_agent from langchain.tools import tool from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI import math # 定义两个工具 tool def search_weather(city: str) - str: 查询指定城市的天气模拟 # 模拟天气数据 weather_data { 北京: 晴天25°C微风, 上海: 多云转小雨22°C, 深圳: 阵雨28°C } return weather_data.get(city, 未知城市) tool def calculator(expression: str) - str: 计算数学表达式例如 23*4 try: result eval(expression) return str(result) except Exception as e: return f计算错误: {e} tools [search_weather, calculator] # Agent 提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手可以使用工具来回答问题。), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 创建 agent llm ChatOpenAI(modelgpt-4, temperature0) agent create_openai_functions_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 测试需要组合工具 result agent_executor.invoke({input: 北京和上海的天气温差是多少}) print(result[output])运行时会看到 Agent 首先调用search_weather(北京)再调用search_weather(上海)然后自己判断需要计算温差接着调用calculator(25-22)最终得出温度差。这就是 ReActReasoning Acting模式的实际体现。常见问题与注意事项串行 vs 并行顺序链虽然简单但延迟是所有步骤之和。对于独立子任务应优先考虑并行模式以降低端到端延迟。动态路由的可靠性路由决策依赖 LLM 的分类能力可能出现分类错误。建议加入置信度判断低于阈值时转人工或使用默认分支。编排器-工作者的循环控制必须设定最大迭代次数或条件终止防止无限循环消耗 Token 和费用。Agent 的工具安全性不要直接执行动态代码或给予未限制的系统权限。对工具调用应进行输入校验和输出过滤。成本与性能优化- 对低频变更的输出尝试使用缓存如LangChain的内存缓存或 Redis。- 尽量使用较小的模型处理简单任务复杂任务再用大模型。- 对用户输入进行预处理避免 LLM 做纯规则可完成的事。可观测性生产环境中一定要记录每一步的输入输出、延迟和 Token 消耗LangSmith 或自定义日志都是必备组件。总结本文介绍了 AI 工作流中的5种核心设计模式从简单的顺序链到自主决策的智能代理它们构成了现代 LLM 应用的骨架。开发者可以根据业务复杂度和灵活性需求选择不同的模式甚至组合使用如 Agent 内部使用顺序链完成一个子任务。模式只是起点真正落地时需要结合错误处理、流式输出、人机协同等现实要素。希望这些可运行的代码示例能帮助你快速上手构建出健壮且优雅的 AI 系统。扩展学习建议- 阅读 LangChain 官方文档中关于Runnable接口的部分。- 深入了解 ReAct、Plan-and-Execute 等高级 Agent 策略。- 实验微调自己的路由决策模型。现在动手试一下这些模式吧