目录1. 模式一提示链模式Prompt Chaining2. 模式二并行化模式Parallelization3. 模式三路由模式Routing4. 模式四协调者-工作者模式Orchestrator-WorkersSend5. 模式五评估器-优化器模式Evaluator-Optimizer6. 总结1. 模式一提示链模式Prompt Chaining核心概念这是最基础的线性模式前一个节点的输出严格作为后一个节点的输入就像写文章必须经历“大纲 → 初稿 → 润色 → 终稿”一样每一步都不可跳跃代码实现要点通过定义 InputState用户输入和 OutputState最终结果来隔离内部状态利用 OverallState 传递中间产物大纲、初稿2. 模式二并行化模式Parallelization核心概念多个独立任务同时执行最后汇总结果。常用于多维度分析如市场、竞品、技术同步调研代码实现要点从 START 节点引出多条边指向不同的分析节点最后汇聚到汇总节点3. 模式三路由模式Routing核心概念根据输入内容动态选择分支。经典的智能客服场景问题被分类为售前、售后或技术交由不同的 Handler 处理代码实现要点利用 with_structured_output 强制 LLM 输出枚举值Literal配合 add_conditional_edges 进行分发4. 模式四协调者-工作者模式Orchestrator-Workers核心概念协调者Orchestrator在运行时分析任务动态生成 n 个子任务并通过 Send API 将不同数据分发到 n 个工作者节点并行执行代码实现要点assign_workers 函数返回一个 Send 列表动态决定并发的数量协调者-工作者模式和并行化模式都涉及同时执行多个任务但它们的核心区别在于任务分配方式并行化任务在设计时就确定所有任务同时开始协调者-工作者任务在运行时由协调者动态分配SendSend 的核心作用是让一个节点在运行时能动态地、并行地调用其他节点并为每个调用传递定制化的状态参数说明参数类型是否必填描述nodestr是目标节点的名称必须与使用add_node添加时的名称完全一致argAny是传递给目标节点的状态数据可以是一个字典或自定义对象timeoutfloat/timedelta/TimeoutPolicy否为该特定任务单独设置的超时策略会覆盖目标节点自身的超时配置行为与机制特性说明执行模式从条件边返回多个Send对象目标节点会被并行执行状态更新每个并行任务拥有独立的状态副本。任务完成后可通过主 State 中定义的 Reducer如operator.add将结果合并回主状态错误处理可为每个任务单独设置超时时间。某个任务失败默认会影响整个图执行需结合Command等进行精细控制import operator from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Send # 1. 定义主图状态 class OverallState(TypedDict): subjects: list[str] # 输入的主题列表 # jokes 字段使用 operator.add 作为 reducer用于合并所有并发生成的笑话 jokes: Annotated[list[str], operator.add] # 2. 定义条件边函数 def continue_to_jokes(state: OverallState): 该函数从 START 节点被调用它根据状态动态生成一个 Send 对象列表。 # 为 subjects 列表中的每一个主题创建一个 Send 对象 # 每个 Send 对象都指向 generate_joke 节点并传递一个包含该主题的字典 return [Send(generate_joke, {subject: s}) for s in state[subjects]] # 3. 定义工作节点 def generate_joke(state: dict): # 此节点接收由 Send 传来的定制状态 {subject: s} subject state[subject] # 生成一个笑话并返回一个字典该字典会通过 reducer 合并到主状态 return {jokes: [fJoke about {subject}]} # 4. 构建图 builder StateGraph(OverallState) builder.add_node(generate_joke, generate_joke) # 关键从 START 节点添加一个条件边指向 continue_to_jokes 函数 builder.add_conditional_edges(START, continue_to_jokes) # 所有由 Send 触发的 generate_joke 节点执行完后进入 END builder.add_edge(generate_joke, END) graph builder.compile() # 5. 执行 result graph.invoke({subjects: [cats, dogs]}) print(result) # 输出{subjects: [cats, dogs], jokes: [Joke about cats, Joke about dogs]}并行局限性并行任务不能直接修改主图的核心状态。必须通过返回一个字典并依赖主 State 中定义的 Reducer 函数如 operator.add来合并结果目标节点要求被 Send 调用的节点如示例中的 generate_joke其函数签名接收的状态必须是 Send 传递的定制状态而不是主图的完整状态5. 模式五评估器-优化器模式Evaluator-Optimizer核心概念先执行任务再评估质量若不达标带着反馈重新执行代码实现要点循环结构生成 → 评估 → 不合格回到生成6. 总结业务场景推荐模式核心工程准则内容生成、数据 ETL提示链加入interrupt人工断点做好 checkpoint 备份独立多源数据同时检索并行化必须处理单个任务的异常降级避免“一颗老鼠屎坏了一锅粥”客服、指令分类路由必须预留Fallback兜底节点别信 LLM 会 100% 听话长文档拆解、多章节写作协调者-工作者必须在汇总时按Index重排序并控制并发数代码修复、高质量翻译评估器-优化器必须设置循环上限max_trials防止死循环烧钱