LangGraph认知篇-Send机制

作者：找了一圈尾巴
Send 机制简介

        在LangGraph 中，节点（Nodes）和边（Edges）默认是预先定义的，且基于同一个共享状态（state)进行操作。但在某些场景下，例如实现映射-规约（map-reduce）等设计模式时，可能存在以下需求：

动态分支创建：运行时根据数据决定分支数量（如处理用户上传的多个文档）
状态隔离：不同分支需要独立状态（如并行处理不同主题的文本生成）
并行处理：支持map-reduce等并行模式（如批量API调用）
未知拓扑：处理无法预先定义图结构的场景（如动态工作流生成）

        为满足上述需求，LangGraph 提供了 Send 机制，其核心功能是：通过条件边（conditional edges）动态生成下游节点的调用指令，实现状态的按需分发和节点的动态触发。
Send机制与静态图结构的区别

特性	静态图	Send机制
分支数量	预先定义	运行时动态生成
状态传递	所有节点共享同一状态	各分支拥有独立状态
执行模式	顺序/固定路径执行	并行/动态路径执行
适用场景	固定拓扑结构	数据驱动的动态拓扑

Send 对象的核心属性

node（str）：目标节点的名称，即需要被调用的节点标识；
arg（Any）：传递给目标节点的状态数据，可以是任意类型，通常是与该节点任务相关的特定信息。

工作机制图解

官方代码示例

1. from typing import Annotated, TypedDict
2. import operator
3. from langgraph.types import Send
4. from langgraph.graph import StateGraph, END, START
6. # 1. 定义整体状态结构
7. class OverallState(TypedDict):
8.     subjects: list[str]  # 存储需要生成笑话的主题列表
9.     # 存储生成的笑话列表，使用operator.add指定合并方式（列表元素累加）
10.     jokes: Annotated[list[str], operator.add]
12. # 2. 定义条件边处理函数：动态生成Send对象
13. def continue_to_jokes(state: OverallState):
14.     # 为每个主题创建一个Send对象，指定调用"generate_joke"节点并传递该主题
15.     return [Send("generate_joke", {"subject": s}) for s in state['subjects']]
17. # 3. 构建状态图工作流
18. builder = StateGraph(OverallState)
20. # 4. 添加"generate_joke"节点：根据传入的主题生成笑话
21. # 该节点接收包含"subject"的状态，返回包含笑话的字典（会合并到整体状态的jokes列表中）
22. builder.add_node("generate_joke", lambda state: {"jokes": [f"Joke about {state['subject']}"]})
24. # 5. 配置起始节点的条件边：通过continue_to_jokes函数动态决定下一步
25. builder.add_conditional_edges(START, continue_to_jokes)
27. # 6. 配置"generate_joke"节点的后续节点：生成笑话后到达结束节点
28. builder.add_edge("generate_joke", END)
30. # 7. 编译工作流图
31. graph = builder.compile()
33. # 8. 执行工作流：输入包含两个主题的状态
34. result = graph.invoke({"subjects": ["cats", "dogs"]})
35. print(result)
36. # 输出：{'subjects': ['cats', 'dogs'], 'jokes': ['Joke about cats', 'Joke about dogs']}

复制代码

定义整体状态，收集笑话主题列表，并存储;
创建条件边处理函数，该函数根据主状态中的 subjects 列表，为每个主题生成一个 Send 对象;
构建工作流

添加 generate_joke 节点，其逻辑是基于传入的 subject 生成笑话；
将起始节点（START）通过条件边关联到 continue_to_jokes 函数；
配置 generate_joke 节点完成后指向结束节点（END）。

执行效果：当输入 {"subjects": ["cats", "dogs"]} 时，工作流会：

通过 Send 动态触发两次 generate_joke 节点调用；
第一次调用使用 {"subject": "cats"} 生成对应笑话；
第二次调用使用 {"subject": "dogs"} 生成对应笑话；
最终汇总结果为 {'subjects': ['cats', 'dogs'], 'jokes': ['Joke about cats', 'Joke about dogs']}。

Map-reduce 设计模式简介

        映射 - 归约（Map-Reduce）是一种用于大规模数据处理的分布式计算设计模式，由 Google 提出，旨在通过拆分任务、并行处理和结果聚合，高效处理海量数据。其核心思想是将复杂任务分解为两个主要阶段：映射（Map） 和归约（Reduce），从而实现数据的分布式处理和结果的高效汇总。
核心思想与流程

        Map-Reduce 的工作流程可概括为 “拆分 - 处理 - 合并”，具体分为三个关键步骤：

映射（Map）阶段

输入：原始数据集（通常是大规模、非结构化或半结构化数据，如日志、文本等）。
操作：由 “映射函数”（Map Function）将输入数据拆分为一系列键值对（Key-Value Pair）。例如，对文本数据进行分词时，可将每个单词作为 “键（Key）”，出现次数（初始为 1）作为 “值（Value）”，即 (单词, 1)。
目的：将原始数据转换为易于处理的中间格式，为后续聚合做准备。

洗牌（Shuffle）阶段（隐式步骤）

作用：在 Map 和 Reduce 之间自动执行，将 Map 阶段输出的键值对按 “键” 进行分组、排序和分发，确保相同键的所有值被发送到同一个 Reduce 节点。
例如：将所有 (“apple”, 1) 键值对汇总到一起，形成 (“apple”, [1, 1, 1])。

归约（Reduce）阶段

输入：Shuffle 阶段处理后的键值对（同一个键对应一组值）。
操作：由 “归约函数”（Reduce Function）对同一键的所有值进行聚合计算（如求和、计数、平均值等）。例如，对 (“apple”, [1, 1, 1]) 求和，得到 (“apple”, 3)，即 “apple” 出现 3 次。
目的：将分散的中间结果合并为最终输出（如统计结果、分析报告等）。

Map-reduce 与 LangGraph 中 Send 机制的结合

        在 LangGraph 工作流框架中，Map-Reduce 模式可通过 Send 机制实现：

Map 阶段：用 Send 为每个输入对象生成独立任务（如为列表中的每个主题生成文本）；
Reduce 阶段：收集所有任务的输出，通过归约节点汇总结果（如合并多个文本生成最终报告）。

        这种结合既保留了 Map-Reduce 的并行处理能力，又通过动态节点调用适配了更灵活的工作流场景。
LangGraph 实现Map-Reduce 示例

1. from typing import TypedDict
2. from langgraph.graph import StateGraph, Send, END
4. # 状态定义
5. class MapReduceState(TypedDict):
6.     documents: list[str]        # 初始文档
7.     chunks: list[str]           # 文档分块
8.     summaries: list[str]        # 分块摘要
9.     final_summary: str          # 最终摘要
11. # 1. 文档分块（初始节点）
12. def split_documents(state: MapReduceState):
13.     documents = state["documents"]
14.     chunks = [chunk for doc in documents for chunk in split_text(doc)]
15.     return {"chunks": chunks}
17. # 2. 动态路由（Send核心）
18. def map_router(state: MapReduceState):
19.     return [
20.         Send("summarize_chunk", {"chunk": c})
21.         for c in state["chunks"]
22.     ]
24. # 3. 摘要生成（并行执行）
25. def summarize_chunk(state: MapReduceState):
26.     return {"summary": llm(f"Summarize: {state['chunk']}")}
28. # 4. 结果聚合
29. def reduce_summaries(state: MapReduceState):
30.     all_summaries = [s for branch in state.values() if "summary" in s]
31.     return {"final_summary": combine_summaries(all_summaries)}
33. # 构建工作流
34. graph = StateGraph(MapReduceState)
35. graph.add_node("split_docs", split_documents)
36. graph.add_node("summarize_chunk", summarize_chunk)
37. graph.add_node("reduce", reduce_summaries)
39. # 动态分支配置
40. graph.add_conditional_edges("split_docs", map_router)
41. graph.add_conditional_edges("summarize_chunk", lambda _: "reduce")
42. graph.add_edge("reduce", END)
44. # 执行工作流
45. result = graph.invoke({
46.     "documents": ["long_text1", "long_text2"],
47.     "chunks": [],
48.     "summaries": [],
49.     "final_summary": ""
50. })
51. print(result["final_summary"])

复制代码

执行流程：

主节点完成 → 触发条件边函数
生成 Send 对象列表 → 创建并行分支
每个分支执行指定节点 → 独立状态输入
所有分支完成后 → 触发聚合节点

其核心机制包括：

动态路由：运行时根据chunks列表长度决定创建多少分支，生成的Send列表会被LangGraph自动并行处理；

1. 动态路由（Send核心）
3. def map_router(state: MapReduceState):
5.     return [
7.         Send("summarize_chunk", {"chunk": c})
9.         for c in state["chunks"]
11.     ]

复制代码

状态隔离原理：每个分支只接收必要数据（减少内存开销），LangGraph自动管理分支状态的创建和传递；

1. # 主状态（分割后）
2. {
3.     "documents": [doc1, doc2],
4.     "chunks": [chunk1, chunk2, chunk3, ...],
5.     "chunk_summaries": [],
6.     "final_summary": ""
7. }
9. # 分支独立状态示例
10. [
11.     {"chunk": "chunk1内容", "chunk_id": 0},  # 分支1
12.     {"chunk": "chunk2内容", "chunk_id": 1},  # 分支2
13.     {"chunk": "chunk3内容", "chunk_id": 2}   # 分支3
14. ]

复制代码

结果聚合机制：summarize_chunk 全部执行完毕后，会调用reduce节点，通过state.values() 获取所有分支的状态，汇总出最终的结果；

1. graph.add_conditional_edges("summarize_chunk", lambda _: "reduce")
3. # 4. 结果聚合
4. def reduce_summaries(state: MapReduceState):
5.     all_summaries = [s for branch in state.values() if "summary" in s]
6.     return {"final_summary": combine_summaries(all_summaries)}

复制代码

Send 机制典型应用场景

文档分块处理

1. Send("process_chunk", {"chunk": c}) for c in doc_chunks

复制代码

API 并行调用

1. Send("call_api", {"request": r}) for r in requests

复制代码

批量数据处理

1. Send("process_item", {"item": i}) for i in batch

复制代码

动态工作流生成

1. Send(next_step, config) for config in dynamic_configs

复制代码

注意事项

• 状态设计：分支节点接收的状态是独立的，不包含主状态的所有字段
  1. # 推荐
  2. Send("process", {"item_id": 123})
  4. # 不推荐
  5. Send("process", full_state)

  复制代码
• 聚合触发：所有分支必须完成才能触发下游节点
  1. graph.add_conditional_edges(
  2.     "worker_node",
  3.     lambda _: "aggregate_node"  # 所有分支完成后触发
  4. )

  复制代码
  错误处理：单个分支失败会影响整个工作流
  性能考虑：避免创建过多分支导致资源耗尽
  

参考文献

https://github.com/langchain-ai/langgraph
Overview

原文地址：https://blog.csdn.net/weixin_41645817/article/details/149813487