Nap-Reduce模式有点像分而治之的感觉将大任务化解为一个个小任务然后各个小任务并发处理最后再归总先贴代码 SendDemo.py LangGraph Map-Reduce 模式演示 通过使用 Send 对象LangGraph 提供了一种优雅的方式来实现这种动态图结构 使得我们可以根据运行时状态来决定执行路径。 解释 1首先执行 generate_subjects主题列表节点生成主题列表[猫, 狗, 程序员] 2然后通过条件边函数 map_subjects_to_jokes 为每个主题创建一个 Send 对象 3make_joke 节点被并行执行3次每次处理一个主题 4最终将所有生成的笑话合并到一个列表中 这种模式非常适合处理动态数量的任务 from typing import Annotated, List, Sequence from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Send import threading import time # 定义状态 class AtguiguState(TypedDict): subjects: List[str] jokes: Annotated[List[str], lambda x, y: x y] # 使用列表合并的方式 # 第一个节点生成需要处理的主题列表 def generate_subjects(state: AtguiguState) - dict: 生成需要处理的主题列表 print(执行节点(第一个节点生成需要处理的主题列表): generate_subjects) subjects [猫, 狗, 程序员] print(f生成主题列表: {subjects}) return {subjects: subjects} # Map节点为每个主题生成笑话 def make_joke(state: AtguiguState) - dict: 为单个主题生成笑话 subject state.get(subject, 未知) print( f执行节点: make_joke处理主题: {subject}, f线程: {threading.current_thread().name} ) time.sleep(2) # 根据主题生成相应笑话 jokes_map { 猫: 为什么猫不喜欢在线购物因为它们更喜欢实体店, 狗: 为什么狗不喜欢计算机因为它们害怕被鼠标咬, 程序员: 为什么程序员喜欢洗衣服因为他们在寻找bugs, 未知: 这是一个关于未知主题的神秘笑话。 } joke jokes_map.get(subject, f这是一个关于{subject}的即兴笑话。) print(f生成笑话: {joke}) print( f完成处理: {subject}, f线程: {threading.current_thread().name} ) return {jokes: [joke]} # 条件边函数根据主题列表生成Send对象列表 def map_subjects_to_jokes(state: AtguiguState) - List[Send]: 将主题列表映射到joke生成任务 print(执行条件边函数: map_subjects_to_jokes) subjects state[subjects] print(f映射主题到joke任务: {subjects}) # 为每个主题创建一个Send对象指向make_joke节点 # 每个Send对象包含节点名称和传递给该节点的状态 send_list [Send(make_joke, {subject: subject}) for subject in subjects] print(f生成Send对象列表: {send_list}) return send_list def main(): 演示Map-Reduce模式 print( Map-Reduce 模式演示 \n) # 创建图 builder StateGraph(AtguiguState) # 添加节点 builder.add_node(generate_subjects, generate_subjects) builder.add_node(make_joke, make_joke) # 添加边 builder.add_edge(START, generate_subjects) # 添加条件边使用Send对象实现map-reduce builder.add_conditional_edges( generate_subjects, # 源节点 map_subjects_to_jokes # 路由函数返回Send对象列表 ) # 从make_joke到结束 builder.add_edge(make_joke, END) # 编译图 graph builder.compile() print(graph.get_graph().print_ascii()) # 执行图 initial_state {subjects: [], jokes: []} print(初始状态:, initial_state) print(\n开始执行图...) start time.perf_counter() result graph.invoke(initial_state) elapsed time.perf_counter() - start print(f总耗时{elapsed:.2f}秒) print(f\n最终结果: {result}) print(\n 演示完成 ) if __name__ __main__: main()这里最关键的是Send关键字Send(目标节点名称, 发送给该节点的数据)代码讲解这个案例演示的是 LangGraph 中的 Map-Reduce 模式先生成一批任务 ↓ 把每个任务分别发送给同一个节点 ↓ 节点分别处理 ↓ 把多个处理结果合并起来本例中Map把“猫、狗、程序员”拆成三个笑话生成任务单任务处理make_joke分别生成笑话Reduce把三个笑话合并到jokes列表整体运行流程START ↓ generate_subjects ↓ map_subjects_to_jokes ├─ Send(make_joke, {subject: 猫}) ├─ Send(make_joke, {subject: 狗}) └─ Send(make_joke, {subject: 程序员}) ↓ 三次 make_joke ↓ 合并三个 jokes ↓ END1. 状态定义class AtguiguState(TypedDict): subjects: List[str] jokes: Annotated[List[str], lambda x, y: x y]整个图的主状态包含两个字段{ subjects: [], jokes: [] }subjectssubjects: List[str]它保存需要处理的所有主题[猫, 狗, 程序员]这个字段没有配置 reducer因此节点返回新的subjects时默认使用新值覆盖旧值。最初是subjects: []generate_subjects返回subjects: [猫, 狗, 程序员]更新后就是subjects: [猫, 狗, 程序员]jokesjokes: Annotated[List[str], lambda x, y: x y]它保存生成的笑话。这里的lambda x, y: x y是一个 reducer等价于def merge_jokes(current, update): return current update其中x当前已经保存的笑话列表y本次节点新返回的笑话列表它的作用是追加并合并结果而不是覆盖结果。例如current [猫的笑话] update [狗的笑话]合并结果[猫的笑话, 狗的笑话]如果没有这个 reducer三个make_joke分支会同时更新jokesLangGraph 无法确定应该保留哪个结果通常会出现并发状态更新冲突。2. 初始状态initial_state { subjects: [], jokes: [] }调用result graph.invoke(initial_state)表示以该字典作为图的初始状态。此时subjects [] jokes []3. 从 START 进入第一个节点builder.add_edge(START, generate_subjects)它规定图开始后首先运行generate_subjects节点代码def generate_subjects(state: AtguiguState) - dict: subjects [猫, 狗, 程序员] return {subjects: subjects}因此控制台打印执行节点(第一个节点生成需要处理的主题列表): generate_subjects 生成主题列表: [猫, 狗, 程序员]该节点返回{ subjects: [猫, 狗, 程序员] }它会更新图的主状态{ subjects: [猫, 狗, 程序员], jokes: [] }注意节点只需要返回自己想更新的字段不必返回完整状态。4. 条件边开始动态分发builder.add_conditional_edges( generate_subjects, map_subjects_to_jokes )意思是generate_subjects执行完成后调用map_subjects_to_jokes()由它决定下一步要运行哪些节点。之所以称为“条件边”是因为接下来走向哪里由运行时状态决定不是在构建图时完全固定。路由函数收到的是更新后的状态def map_subjects_to_jokes(state: AtguiguState) - List[Send]:此时state[subjects]为[猫, 狗, 程序员]所以控制台打印执行条件边函数: map_subjects_to_jokes 映射主题到joke任务: [猫, 狗, 程序员]5.Send是什么核心代码是send_list [ Send(make_joke, {subject: subject}) for subject in subjects ]可以把一个Send理解为安排一次节点调用并为这一次调用单独提供输入数据。它接收两个参数Send(目标节点名称, 发送给该节点的数据)循环展开后相当于send_list [ Send(make_joke, {subject: 猫}), Send(make_joke, {subject: 狗}), Send(make_joke, {subject: 程序员}) ]这不是把整个列表一次性交给make_joke而是安排三次独立调用make_joke({subject: 猫}) make_joke({subject: 狗}) make_joke({subject: 程序员})因此任务数量是在运行时动态决定的subjects 有3个元素 → 创建3个Send → 执行3次make_joke subjects 有100个元素 → 创建100个Send → 执行100次make_joke subjects 是空列表 → 不创建对应任务这就是Send的主要价值。6. 为什么make_joke能读取subjectmake_joke中写的是def make_joke(state: AtguiguState) - dict: subject state.get(subject, 未知)但是AtguiguState中只声明了subjects jokes并没有声明subject。这里的subject来自Send的第二个参数Send(make_joke, {subject: subject})所以三次调用分别收到{subject: 猫} {subject: 狗} {subject: 程序员}更准确地说这个节点接收的是 Map 任务的局部输入而不是完整的AtguiguState。因此当前类型标注不够严谨。建议单独定义输入类型class JokeInput(TypedDict): subject: str def make_joke(state: JokeInput) - dict: subject state[subject] ...这样代码含义更加准确class AtguiguState(TypedDict): subjects: List[str] jokes: Annotated[List[str], lambda x, y: x y] class JokeInput(TypedDict): subject: str其中AtguiguState图的公共状态JokeInput每个make_joke任务单独收到的输入7. 三个make_joke任务第一个任务收到{subject: 猫}返回{ jokes: [ 为什么猫不喜欢在线购物因为它们更喜欢实体店 ] }第二个任务收到{subject: 狗}返回{ jokes: [ 为什么狗不喜欢计算机因为它们害怕被鼠标咬 ] }第三个任务收到{subject: 程序员}返回{ jokes: [ 为什么程序员喜欢洗衣服因为他们在寻找bugs ] }这就是控制台为什么出现三次执行节点: make_joke因为不是一次调用处理三个主题而是三个独立任务分别调用同一个节点。8. Reduce合并三个结果每个节点返回的都是{jokes: [一条笑话]}jokes字段配置了 reducerlambda x, y: x y所以合并过程可以理解为[] [猫的笑话]得到[猫的笑话]然后[猫的笑话] [狗的笑话]得到[猫的笑话, 狗的笑话]最后[猫的笑话, 狗的笑话] [程序员的笑话]得到[猫的笑话, 狗的笑话, 程序员的笑话]因此最终状态是{ subjects: [猫, 狗, 程序员], jokes: [ 猫的笑话, 狗的笑话, 程序员的笑话 ] }这里的 Reduce 没有单独定义一个叫reduce的节点而是通过状态字段上的 reducer 自动完成的jokes: Annotated[List[str], lambda x, y: x y]9. 为什么图形里没有显示make_joke打印结果中只显示了START ↓ generate_subjects ↓ END这是因为make_joke是通过条件边函数返回的Send动态选择的而你的add_conditional_edges()没有提供静态路径提示builder.add_conditional_edges( generate_subjects, map_subjects_to_jokes )LangGraph 在绘制静态图时无法仅根据这个普通 Python 函数可靠确定它运行时会返回哪些目标节点。但是实际运行时函数返回Send(make_joke, ...)所以make_joke确实执行了三次。如果希望可视化图能够显示这个动态目标可以给路由函数添加返回类型提示from typing import Literal def map_subjects_to_jokes( state: AtguiguState ) - list[Send]: ...不过Send动态路由的图形展示会受 LangGraph 版本影响。也可以在条件边中显式提供路径映射或路径提示。打印出的None则来自print(graph.get_graph().print_ascii())print_ascii()自己已经把图打印出来了但它的返回值是None。外层的print()又把这个返回值打印了一遍。直接改成graph.get_graph().print_ascii()就不会额外出现None10.make_joke → ENDbuilder.add_edge(make_joke, END)表示每个make_joke任务完成后该分支到达END。LangGraph 会等待这一批已调度的任务完成并通过 reducer 合并它们对jokes的更新最后返回完整状态。本例可以记成generate_subjects生成任务列表 map_subjects_to_jokes将任务列表转换成多个Send Send指定每个任务调用哪个节点以及传什么数据 make_joke处理一个主题 jokes reducer合并多个并行任务的结果 graph.invoke运行整个图并返回最终状态最关键的一句是Send负责动态“拆任务”Annotated中的 reducer 负责“合结果”。ThreadPoolExecutor是 Python 的线程池执行器。LangGraph 将同一轮的三个Send任务分配给了三个工作线程。本次执行过程主线程启动 graph.invoke() ↓ generate_subjects 生成3个主题 ↓ map_subjects_to_jokes 创建3个Send ↓ LangGraph把任务放入线程池 ↓ ThreadPoolExecutor-1_0 处理猫 ThreadPoolExecutor-1_1 处理狗 ThreadPoolExecutor-1_2 处理程序员 ↓ 三个线程约2秒后分别完成 ↓ reducer合并三个 jokes 更新 ↓ graph.invoke()返回最终状态