LangGraph中的Map与Reduce

在LangGraph构建的智能体中，信息传递的“State（状态）”就像智能体的随身档案袋——所有和交互相关的信息，从用户刚发的提问、工具返回的天气数据，到智能体自己的思考记录，都会汇总存放在这里。但这些信息刚产生时既杂又散，直接堆在档案袋里会一团乱，智能体想调用时根本无从下手。这时就需要Map和Reduce这两位高效的“信息管家”登场，它们一个负责分类收纳，一个负责整理汇总，把杂乱的信息打理得井

新知图书

518人浏览 · 2026-03-24 11:27:08

新知图书 · 2026-03-24 11:27:08 发布

新书预告《LangGraph智能体设计模式与多智能体开发》-CSDN博客

以WeatherState中的消息列表来说，那个用特殊方式标记的messages字段就是它们工作的最佳例子。

先说说Map——它像个带很多小格子的收纳盒。用户问的话、工具查询出来的结果、智能体自己说的话，这些零零碎碎的信息，都会被当成单独的“小卡片”，一张一张放进格子里。就像拼图还没拼时，每块碎片都乖乖待在自己的位置，模样和刚产生时一模一样，谁也不打扰谁。

而add_messages就是Reduce的具体干活儿的——它像个自动拼图高手，每当有新消息来时（比如工具查完天气返回结果），它会立刻跑过来，把新的“小卡片”和盒子里已有的旧卡片摆在一起，去掉重复的，再按时间顺序排好。最后你会发现，原本零散的碎片已经变成一整块完整的对话记录了。

这种机制让开发者无须手动处理消息拼接逻辑，既保证了状态更新的简洁性，又确保了智能体始终能基于完整的上下文决策，让工具调用、对话流转更流畅自然。

3.3.1 LangGraph中的Map

在LangGraph中，Map是状态（State）中用于承载离散信息的“容器系统”，其核心价值在于对多样化数据的“原生性存储”。它并非单一的变量，而是一组能独立存储不同类型、不同来源信息的结构集合，像一个分类明确的档案柜，每个抽屉（字段）都能存放一类信息，且每个信息单元（元素）都保持其原始形态与属性。

以智能体交互场景为例，除了messages列表外，Map还可包含tool_calls（记录模型发起的工具调用指令）、user_context（用户的基础信息，如地理位置偏好）、intermediate_results（中间计算结果）等字段。每个字段内的信息都是离散的：tool_calls中可能包含“调用get_weather时的参数loc=北京”“调用get_traffic时的参数time=18:00”等独立记录；user_context中可能存储“用户历史查询过上海天气”“用户习惯使用口语化表达”等零散信息。

这些离散信息之所以能稳定存在，得益于Map的“无侵入性存储”特性——它不强制信息格式化或合并，而是允许每个元素保留其原生属性（如消息的类型、工具调用的时间戳、数据的结构类型等）。这种设计让智能体在流程中能随时“提取单条信息”进行处理（例如，模型仅需从messages中读取最新的用户提问，无须解析整个上下文），同时为后续的信息整合（即Reduce操作）保留完整的“原始素材”。

from typing import Annotated, List, Dict

from pydantic import BaseModel

from langchain_core.messages import HumanMessage, AIMessage

from langgraph.prebuilt import add_messages

class ChatState(BaseModel):

# 存储对话消息（离散的消息单元）

messages: Annotated[List, add_messages] = []

# 存储用户基础信息（离散的用户属性）

user_info: List[Dict] = []

# 存储工具调用记录（离散的调用行为）

tool_records: List[Dict] = []

假设用户与智能体的交互流程如下：

1.用户发送第一条消息："你好，我叫小明，在深圳"

messages会添加 [HumanMessage(content="你好，我叫小明，在深圳")]（保留原始消息类型和内容）；

user_info会添加 [{"name": "小明"}, {"location": "深圳"}]（拆分用户信息为离散单元）

2.智能体回复："你好小明，需要帮你查深圳的天气吗？"。

messages会添加 [AIMessage(content="你好小明，需要帮你查深圳的天气吗？")]（新增一条独立消息）。

3.用户说："查一下今天的天气"，智能体调用工具get_weather(loc="深圳", date="2025-11-12")

tool_records会添加 [{"tool": "get_weather", "params": {"loc": "深圳", "date": "2025-11-12"}, "status": "success"}]（记录单次工具调用的完整信息）。

此时Map字段的状态：

# messages 存储了3条离散消息（用户1→智能体1→用户2）

messages = [

HumanMessage(content="你好，我叫小明，在深圳"),

AIMessage(content="你好小明，需要帮你查深圳的天气吗？"),

HumanMessage(content="查一下今天的天气")

]

# user_info 存储了两条离散的用户属性

user_info = [{"name": "小明"}, {"location": "深圳"}]

# tool_records 存储了1条离散的工具调用记录

tool_records = [{"tool": "get_weather", "params": {...}, "status": "success"}]

这些字段都是Map机制的载体，每个字段内的信息都是离散的独立单元（如单条消息、单个用户属性、单次工具调用），且保留了原始形态（消息类型、参数结构等）。后续如需整合（比如用add_messages合并消息生成上下文，或从user_info中提取位置参数），这些“原始颗粒”就是可靠的数据源。

3.3.2 LangGraph中的Reduce

在LangGraph中，Reduce是对Map存储的离散信息进行“结构化整合”的核心机制，其核心价值在于将零散的“信息颗粒”转换为可直接用于决策的“有效知识”。如果说Map是分类收纳的档案柜，那么Reduce就是负责档案汇编的整理者——它会基于预设规则（或自定义逻辑）对Map中分散的信息单元进行合并、过滤、提炼，最终输出结构化、无冗余且符合场景需求的整合结果。

Reduce的运作逻辑具有三个关键特征：动态触发、规则驱动和目标导向。

动态触发：每当Map中的信息新增时（如用户发送新消息、工具返回新结果），Reduce会自动启动，确保整合结果始终与最新状态同步。
规则驱动：整合逻辑由明确规则定义（如去重、时序排序、字段提取等），避免信息处理的随机性。
目标导向：所有操作都围绕“让智能体更高效决策”展开，例如将零散对话整合成上下文、将多次工具调用结果提炼为关键结论。

以我们在3.4.1节中定义的ChatState中的字段为例，Reduce会针对不同类型的离散信息进行精准整合。

1）对messages的Reduce：构建完整对话上下文

前面提到的add_messages就是典型的Reduce工具——它针对Map中messages字段的离散消息（HumanMessage、AIMessage等）执行“时序合并”规则：

保留每条消息的原始类型（确保区分用户与智能体发言）。
按时间顺序拼接成连续列表（最新消息自动追加至末尾）。
过滤重复或无效消息（如空内容消息）。

最终输出的messages列表不再是孤立的消息碎片，而是能直接被大语言模型读取的完整对话历史。例如，当工具返回天气结果后，messages会新增一条ToolMessage，add_messages会自动将其合并，形成以下格式：

messages = [

HumanMessage(content="你好，我叫小明，在深圳"),

AIMessage(content="你好小明，需要帮你查深圳的天气吗？"),

HumanMessage(content="查一下今天的天气"),

ToolMessage(content="深圳2025-11-12天气：晴，22-28℃", tool_call_id="xxx")

]

这个整合结果能直接作为模型生成回复的上下文，无须模型手动筛选或排序。

2）对user_info的Reduce：生成统一用户档案

Map中user_info存储的是离散的用户属性（如{"name": "小明"}、{"location": "深圳"}），而Reduce会通过自定义逻辑将其整合为结构化的用户档案。例如，定义一个merge_user_info函数作为Reduce工具：

def merge_user_info(prev_info: List[Dict], new_info: Dict) -> Dict:

# 合并新旧用户信息，新信息覆盖旧信息（如用户更新位置时）

merged = {**dict((k, v) for item in prev_info for k, v in item.items()),** new_info}

return merged

当用户后续补充“我现在在广州”时，user_info先通过Map新增{"location": "广州"}，再经merge_user_info整合为：

merged_user_info = {"name": "小明", "location": "广州"} # 新位置覆盖旧位置

这个结果可直接用于工具调用（如后续查询天气时自动填充最新位置），避免智能体重复询问或使用过时信息。

3）对tool_records的Reduce：提炼工具调用关键结论

Map中tool_records存储的是单次工具调用的完整细节（如参数、状态、原始返回值），而Reduce会从中提取核心结果，形成“工具调用摘要”。例如，针对天气查询的tool_records：

tool_records = [

{"tool": "get_weather", "params": {"loc": "深圳", "date": "2025-11-12"},

"status": "success", "result": {"temp": 25, "condition": "晴", "wind": "3级"}}]

Reduce可通过规则提炼为：

tool_summary = {"date": "2025-11-12", "location": "深圳", "weather": "晴", "temperature": "25℃"}

这个摘要无须智能体解析原始工具返回的复杂结构，可直接用于生成自然语言回复（如“深圳今天晴，气温25℃”）。

因此，Reduce的核心价值可以概括为让状态“可用且高效”。

3.3.3 各司其职的Map与Reduce

Map确保了信息的完整性与原始性，但零散的信息无法直接支撑智能体决策（例如，模型无法基于孤立的用户消息生成连贯回复，工具调用也无法基于分散的用户属性自动填充参数）。而Reduce通过结构化整合。解决了以下两个关键问题。

信息冗余：过滤重复或无关内容，减少智能体处理的信息量。
结构混乱：将非结构化的离散单元转换为符合场景需求的格式（如对话上下文、用户档案、工具结果摘要）。

这种“Map存原始，Reduce出结果”的协作模式，让LangGraph智能体的状态管理既灵活（支持多样化信息输入）又高效（整合结果可直接复用），最终实现对话流转、工具调用的流畅性与准确性。

LangGraph中Map与Reduce的协作机制构成了智能体状态管理的核心逻辑——Map以“原生性存储”为核心，像分类明确的档案柜，将用户消息、工具调用记录、用户属性等信息以离散单元的形式留存，既保证了信息的完整性，又为后续处理保留了原始素材；Reduce则以“结构化整合”为目标，像专业的整理者，通过动态触发、规则驱动的方式，将零散的信息颗粒转换为完整对话上下文、统一用户档案、工具结果摘要等可用知识。

两者一“存”一“整”，既免去了开发者手动处理信息拼接、筛选的烦琐，又确保了智能体始终能基于完整、有序、无冗余的状态决策，最终实现对话流转与工具调用的流畅性、准确性，让LangGraph智能体的状态管理既灵活又高效。

3.3.4 State中的辅助参数BaseModel、Annotated与Literal详解

在LangGraph的状态（State）定义中，BaseModel、Annotated、Literal是三个核心辅助工具——它们共同作用于状态的“结构规范”“功能增强”和“取值约束”，让状态数据既安全可控，又能适配智能体的复杂流转需求。本小节将结合实战场景拆解三者的核心用法与配合逻辑，帮你彻底掌握状态定义的进阶技巧。

1. BaseModel：状态的“结构化骨架”

BaseModel是LangGraph状态定义的基础载体，来自Python的Pydantic库，其核心作用是给状态数据定结构、做校验、设默认值。没有它，状态会变成无规则的字典，容易出现数据类型错误、字段缺失等问题；有了它，状态就像一张“结构化表单”，所有字段的类型、约束都清晰可查。

核心用法与实战价值如下。

定义字段类型：明确每个状态字段的数据类型（如List、Dict、str），智能体流转时会自动校验，避免传入无效数据。
设置默认值与描述：通过Field指定字段默认值、说明文档，提升代码的可读性和易用性。
自动数据校验：当外部数据传入状态时，BaseModel会自动校验字段类型、取值范围，若不符合规则，则抛出明确异常，方便调试。

2. Annotated：状态字段的“功能增强器”

Annotated是Python的内置类型，其核心作用是给字段添加“元数据注解”——它不改变字段本身的类型，而是给字段绑定额外的功能（如消息合并逻辑、校验规则）。在LangGraph中，它最常用的场景是给messages字段绑定add_messages工具，实现消息的自动合并（即之前讲的Reduce逻辑）。

核心用法与实战价值如下。

绑定LangGraph工具：将LangGraph的内置工具（如add_messages）与字段绑定，让字段具备自动处理能力。
添加额外描述与校验：结合Field或自定义注解，给字段补充更丰富的元数据（如校验规则、文档说明）。
不破坏原有类型：注解仅附加功能，字段的原始类型（如List [BaseMessage]）依然保留，不影响类型校验。

下面给出一个典型的代码示例：

from langgraph.prebuilt import add_messages

from langchain_core.messages import BaseMessage

# Annotated的格式：Annotated[原始类型, 注解1, 注解2, ...]

class AssistantState(BaseModel):

# 给messages字段绑定add_messages工具，实现消息自动合并

messages: Annotated[List[BaseMessage], add_messages, Field(description="自动合并的对话消息列表")] = []

这里的add_messages是LangGraph内置的Reduce工具，通过Annotated绑定后，每当有新消息传入messages字段，会自动触发add_messages的逻辑：合并新旧消息、按时间排序、过滤重复内容。

除了add_messages外，还可绑定自定义注解（如校验消息长度、过滤敏感词），示例如下：

# 自定义注解：限制消息列表最大长度为10

def max_messages(length: int):

def validator(messages: List[BaseMessage]):

if len(messages) > length:

raise ValueError(f"消息列表长度不能超过{length}条")

return messages

return validator

# 绑定自定义注解+add_messages

messages: Annotated[List[BaseMessage], add_messages, max_messages(10)] = []

3. Literal：状态字段的“取值限定器”

Literal的核心作用是限定字段的“可选值范围”，让字段只能取预设的几个固定值，无法传入其他内容。它是实现多分支Route的关键，比如之前的intent字段，通过Literal限定可选值，能确保分支判断函数返回的结果一定是有效分支名。

核心用法与实战价值如下。

限定固定取值：明确字段只能取指定的几个值（如"info_query"、"tool_call"），避免无效值导致的分支流转错误。
提升代码可读性：字段的可选值直接写在类型注解中，无须额外注释，读者能快速了解字段用途。
配合分支判断：与Route的条件判断函数联动，确保意图判断结果与分支节点名完全匹配，避免流转异常。

from typing import Literal

class AssistantState(BaseModel):

messages: Annotated[List[BaseMessage], add_messages] = []

# 限定intent只能取以下4个值，其他值会触发校验异常

intent: Literal["info_query", "tool_call", "chat", "default"] = Field(

default="default", description="用户意图类型，仅支持预设的4个值"

)

tool_result: str = ""

核心用法与实战价值：

当条件判断函数judge_intent返回值时，Literal会自动校验该值是否在["info_query", "tool_call", "chat", "default"]中。
若误返回其他值（如"query_info"），则直接抛出ValueError，避免因无效分支名导致智能体流转失败。
配合IDE的类型提示，编写分支判断函数时会自动联想可选值，减少拼写错误。

可以看到，BaseModel、Annotated、Literal并非孤立使用，而是形成“骨架+增强+约束”的三层架构相互配合。