LangChain 设计与实现

第4章 消息系统与多模态

作者 杨艺韬 · 11,575 字 · 发布于 · 更新于

本章基于 langchain-core 1.2.26 / langchain-classic 1.0.3 源码分析。核心代码位于 langchain_core/messages/ 目录。

消息（Message）是 LangChain 中最基础、最核心的数据结构之一。在与大语言模型交互的过程中，无论是用户向模型提问、模型生成回复、系统预设角色行为，还是工具返回执行结果，所有这些信息都以消息的形式在系统中流转。可以说，消息就是 LangChain 与大语言模型对话的通用"货币"。

从架构角度来看，LangChain 的消息系统承担着三重职责。首先是抽象统一：不同的模型提供商（OpenAI、Anthropic、Google、AWS Bedrock 等）各自定义了不同的消息格式，LangChain 需要用统一的数据模型屏蔽这些差异，让应用代码与具体提供商解耦。其次是多模态表达：随着大模型向多模态方向演进，消息不再仅仅是文本，还可能包含图片、音频、视频、文件等多种模态的数据，消息系统需要为此提供标准化的表示框架。最后是流式支持：现代大模型应用普遍采用流式输出来提升用户体验，消息系统需要支持片段（Chunk）级别的增量传输与合并。

LangChain 1.0 版本对消息系统进行了重大升级，引入了 Content Block 体系和标准化的多模态内容表示。本章将从 BaseMessage 基类出发，逐层剖析消息系统的类型层级、流式消息的合并机制、多模态 Content Block 体系以及丰富的消息工具函数，深入理解这套精心设计的数据基础设施。

本章要点

  1. BaseMessage 层级体系：BaseMessage 及其子类 HumanMessage、AIMessage、SystemMessage、ToolMessage 如何构成完整的消息类型系统
  2. MessageChunk 流式消息：BaseMessageChunk 通过 __add__ 运算符实现流式片段的增量合并
  3. Content Blocks 多模态体系：1.0 版本引入的标准化内容块（Text/Image/Audio/Video/File/Reasoning），以及 content_blocks 属性如何统一不同提供商的格式
  4. ToolCall 与工具调用：ToolCall、ToolCallChunk、InvalidToolCall 三级工具调用结构
  5. 消息工具函数：convert_to_messages、filter_messages、trim_messages 等实用函数的设计与实现

4.1 BaseMessage：消息体系的基石

LangChain 的整个消息系统构建在 BaseMessage 这个基类之上（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:93）。它继承自 Serializable，这意味着所有消息对象都天然具备序列化和反序列化能力。这一设计决策使得消息对象可以被持久化到数据库、通过网络传输到远程服务，或者存储到 LangSmith 等追踪平台，而无需额外的序列化适配代码。

从设计哲学角度看，BaseMessage 采用了"宽进严出"的原则：它允许 content 字段接受字符串或列表两种格式，给予上游代码最大的灵活性；但通过 content_blocks 属性提供了统一的标准化视图，让下游消费者可以用一致的方式处理各种格式的内容。这种双层设计在框架领域是一种经典的"适配器模式"应用。

4.1.1 核心字段设计

# langchain_core/messages/base.py

class BaseMessage(Serializable):
    content: str | list[str | dict]
    """消息内容：纯字符串或混合内容块列表"""

    additional_kwargs: dict = Field(default_factory=dict)
    """附加数据，例如模型提供商的原始工具调用信息"""

    response_metadata: dict = Field(default_factory=dict)
    """响应元数据：响应头、logprobs、token 计数、模型名称等"""

    type: str
    """消息类型标识符，用于序列化/反序列化时的类型辨别"""

    name: str | None = None
    """可选的消息名称，提供人类可读的标识"""

    id: str | None = Field(default=None, coerce_numbers_to_str=True)
    """可选的唯一标识符，通常由提供商或模型生成"""

content 字段的联合类型设计 str | list[str | dict] 是一个精妙的决策。当消息只包含纯文本时，它是一个简单的字符串，这是绝大多数场景下的使用方式。当消息包含多模态内容（如图片、音频等）时，它变成一个列表，列表中的每个元素可以是纯文本字符串或描述特定内容块的字典。这种设计既保证了简单场景的易用性（不需要将纯文本包装成列表），又为复杂的多模态场景提供了足够的表达力。值得注意的是，Pydantic 的 ConfigDict(extra="allow") 配置（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:142）允许消息对象携带任意额外字段，这为子类扩展和提供商特定属性留出了空间。

additional_kwargs 字段扮演着"后备箱"的角色，用来存放那些尚未被标准化的提供商特定数据。例如，在 LangChain 0.x 版本中，OpenAI 的工具调用信息是通过 additional_kwargs["tool_calls"] 传递的。随着框架的演进，这些数据逐渐被迁移到了专门的标准化字段（如 AIMessage.tool_calls），但 additional_kwargs 仍然为新出现的、尚未被标准化的数据提供了灵活的承载空间。这种"先兼容后标准化"的演进策略贯穿了 LangChain 消息系统的整个发展历程。

response_metadata 字段与 additional_kwargs 的定位不同。它专门存储与模型响应相关的元数据，例如 HTTP 响应头、logprobs（对数概率）、token 计数详情和模型名称等。将响应元数据与消息内容分离存储的好处是：序列化消息时可以选择性地包含或排除这些元数据，而且不会干扰消息内容的正常处理流程。

id 字段使用了 Pydantic 的 coerce_numbers_to_str=True 选项（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:135），这意味着即使传入数字类型的 ID，也会被自动转换为字符串。这种防御性设计处理了某些提供商可能返回数字 ID 的情况。

4.1.2 content_blocks 属性：多模态统一接口

1.0 版本引入的 content_blocks 属性是消息系统最重要的新增设计之一（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:200）。它将 content 字段中的原始数据统一转换为标准化的 ContentBlock 类型列表：

@property
def content_blocks(self) -> list[types.ContentBlock]:
    blocks: list[types.ContentBlock] = []
    content = (
        [self.content] if isinstance(self.content, str) and self.content
        else self.content
    )
    for item in content:
        if isinstance(item, str):
            blocks.append({"type": "text", "text": item})
        elif isinstance(item, dict):
            item_type = item.get("type")
            if item_type not in types.KNOWN_BLOCK_TYPES:
                blocks.append({"type": "non_standard", "value": item})
            else:
                blocks.append(cast("types.ContentBlock", item))

    # 多轮解析：尝试将 non_standard 块转为标准块
    for parsing_step in [
        _convert_v0_multimodal_input_to_v1,
        _convert_to_v1_from_chat_completions_input,
        _convert_to_v1_from_anthropic_input,
        _convert_to_v1_from_genai_input,
        _convert_to_v1_from_converse_input,
    ]:
        blocks = parsing_step(blocks)
    return blocks

这个设计的巧妙之处在于它的"渐进式解析"策略。整个转换过程分为两个阶段：第一阶段是快速分类，遍历所有内容项，将已知类型（通过 KNOWN_BLOCK_TYPES 集合判断）直接映射为标准块，将未知类型标记为 non_standard（分类逻辑见 langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:236）。第二阶段是多轮转换，依次尝试五个提供商格式转换器（v0 旧格式、OpenAI Chat Completions、Anthropic、Google GenAI、Bedrock Converse；这份有序列表见 langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:253），每个转换器只处理 non_standard 类型的块，跳过已经标准化的块。

这种"先分类后转换"的两阶段设计在性能和兼容性之间取得了很好的平衡：对于已经是标准格式的内容块，第一阶段就完成了处理，不需要经历任何转换器的尝试；对于提供商特定格式的内容块，它们会按照优先级依次被各个转换器尝试转换，最终要么被成功标准化，要么保持为 non_standard 交由应用层自行处理。

还有一个值得注意的细节：代码中对 source_type 字段进行了特殊检查（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:242；上面的简化代码为突出主干省略了这一步）。在 LangChain v0 时代，多模态块使用了 source_type 字段来区分数据来源（如 "base64"、"url" 等），而 v1 的标准块不使用这个字段。因此，即使一个 v0 块的 type 恰好与 v1 的已知类型名称相同（如 "image"），只要它包含 source_type 字段，就会被标记为 non_standard 交给专门的 v0 转换器处理，避免了格式混淆。

4.1.3 text 属性与向后兼容

@property
def text(self) -> TextAccessor:
    if isinstance(self.content, str):
        text_value = self.content
    else:
        blocks = [
            block for block in self.content
            if isinstance(block, str)
            or (block.get("type") == "text" and isinstance(block.get("text"), str))
        ]
        text_value = "".join(
            block if isinstance(block, str) else block["text"] for block in blocks
        )
    return TextAccessor(text_value)

text 属性定义在 langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:263。TextAccessor（同文件 :47）是一个继承自 str 的特殊类型，它既支持属性访问 message.text（推荐方式），又支持方法调用 message.text()（旧方式，会发出弃用警告）。这是一种典型的平滑迁移策略：在 LangChain 0.x 版本中，text 是一个方法，需要通过 message.text() 调用；在 1.0 版本中改为了更直观的属性访问方式 message.text。TextAccessor 通过继承 str 并实现 __call__ 方法，使得新旧两种访问方式都能正常工作，给了开发者充足的迁移窗口。

text 属性的提取逻辑也很精细：当 content 是字符串时直接返回；当 content 是列表时，它只提取纯字符串元素和 type 为 "text" 的字典块中的文本，然后将它们拼接在一起。这意味着图片块、音频块等非文本内容会被自动忽略，调用者始终能得到一个干净的纯文本表示。

classDiagram
    class Serializable {
        +is_lc_serializable() bool
        +get_lc_namespace() list
    }
    class BaseMessage {
        +content: str | list
        +additional_kwargs: dict
        +response_metadata: dict
        +type: str
        +name: str | None
        +id: str | None
        +content_blocks: list~ContentBlock~
        +text: TextAccessor
        +pretty_repr(html) str
        +pretty_print() void
    }
    class HumanMessage {
        +type: "human"
    }
    class AIMessage {
        +type: "ai"
        +tool_calls: list~ToolCall~
        +invalid_tool_calls: list~InvalidToolCall~
        +usage_metadata: UsageMetadata | None
    }
    class SystemMessage {
        +type: "system"
    }
    class ToolMessage {
        +type: "tool"
        +tool_call_id: str
        +artifact: Any
        +status: "success" | "error"
    }
    Serializable <|-- BaseMessage
    BaseMessage <|-- HumanMessage
    BaseMessage <|-- AIMessage
    BaseMessage <|-- SystemMessage
    BaseMessage <|-- ToolMessage

4.2 消息类型全景

4.2.1 HumanMessage：用户消息

了解了 BaseMessage 的核心设计之后，让我们逐一审视它的四个主要子类。每种子类都针对特定的交互角色进行了精确的字段定制。

HumanMessage（langchain-core-1.2.26/libs/core/langchain_core/messages/human.py:9）是所有消息类型中最简单的一个。它代表用户发送给模型的消息，是对话交互的起点。它的定义几乎就是对 BaseMessage 的最小扩展：

# langchain_core/messages/human.py

class HumanMessage(BaseMessage):
    type: Literal["human"] = "human"

它的 type 字段使用 Literal["human"] 类型注解硬编码为 "human"，默认值也是 "human"。这个看似简单的设计其实蕴含着 Pydantic 的区分联合类型（Discriminated Union）机制：当需要从 JSON 数据反序列化消息时，系统可以根据 type 字段的值精确地选择对应的消息类。

虽然 HumanMessage 的类定义很简洁，但它继承了 BaseMessage 的全部能力，包括通过 content_blocks 传入多模态内容的能力。以下是一些典型用法：

from langchain_core.messages import HumanMessage

# 纯文本
msg = HumanMessage(content="什么是 LangChain？")

# 多模态：文本 + 图片
msg = HumanMessage(content_blocks=[
    {"type": "text", "text": "请描述这张图片"},
    {"type": "image", "url": "https://example.com/photo.png"},
])

4.2.2 AIMessage：模型响应消息

AIMessage（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:160）是消息体系中最复杂的类型。作为模型的输出载体，它不仅承载文本响应，还需要处理工具调用请求、token 使用统计、推理思考过程等丰富的结构化元数据。可以说，AIMessage 的复杂度直接反映了现代大语言模型输出的丰富度。

# langchain_core/messages/ai.py

class AIMessage(BaseMessage):
    tool_calls: list[ToolCall] = Field(default_factory=list)
    """标准化的工具调用列表"""

    invalid_tool_calls: list[InvalidToolCall] = Field(default_factory=list)
    """解析失败的工具调用列表"""

    usage_metadata: UsageMetadata | None = None
    """token 使用统计"""

    type: Literal["ai"] = "ai"

AIMessage 的 _backwards_compat_tool_calls 模型验证器（使用 @model_validator(mode="before") 装饰器，见 langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:304）是消息系统中最重要的兼容性代码之一。它在 Pydantic 模型初始化之前运行，确保无论工具调用数据是以新格式（直接在 tool_calls 字段中）还是旧格式（在 additional_kwargs["tool_calls"] 中以 OpenAI 原始格式存储）存在，最终都能被正确解析到标准化的 tool_calls 和 invalid_tool_calls 字段中。这段代码的存在保证了从 LangChain 0.x 迁移到 1.0 的应用不会因为工具调用格式的变化而崩溃：

# 简化：省略了 try/except（解析出错只记 debug 日志）、AIMessageChunk 分支
# （改用 default_tool_chunk_parser 解析成 tool_call_chunks），
# 以及随后给各 tool_call 规整 type 字段的逻辑
@model_validator(mode="before")
@classmethod
def _backwards_compat_tool_calls(cls, values: dict) -> Any:
    check_additional_kwargs = not any(
        values.get(k) for k in ("tool_calls", "invalid_tool_calls", "tool_call_chunks")
    )
    if check_additional_kwargs and (
        raw_tool_calls := values.get("additional_kwargs", {}).get("tool_calls")
    ):
        parsed_tool_calls, parsed_invalid_tool_calls = default_tool_parser(
            raw_tool_calls
        )
        values["tool_calls"] = parsed_tool_calls
        values["invalid_tool_calls"] = parsed_invalid_tool_calls
    return values

UsageMetadata：token 使用统计

class UsageMetadata(TypedDict):
    input_tokens: int
    output_tokens: int
    total_tokens: int
    input_token_details: NotRequired[InputTokenDetails]
    output_token_details: NotRequired[OutputTokenDetails]

UsageMetadata（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:104）提供了跨模型的标准化 token 计数表示。InputTokenDetails（同文件 :38）进一步细分了缓存命中 (cache_read)、缓存创建 (cache_creation) 和音频 token (audio)；OutputTokenDetails（同文件 :74）则区分了推理 token (reasoning) 和音频输出 token。两者都声明为 TypedDict(total=False)，所以细分项是可选的。这种细粒度的分类为精确的成本核算提供了基础。

4.2.3 SystemMessage：系统指令

# langchain_core/messages/system.py

class SystemMessage(BaseMessage):
    type: Literal["system"] = "system"

SystemMessage（langchain-core-1.2.26/libs/core/langchain_core/messages/system.py:9）用于设定模型的行为规则和角色设定，通常作为消息序列的第一条消息出现。系统消息不参与对话的"来回"，而是作为一种"场外指令"影响模型在整个对话过程中的行为。例如，告诉模型"你是一个专业的法律顾问"或"请用中文回答所有问题"。

从 langchain-core 0.3.27 起（#28794，2024-12），convert_to_messages 函数同时支持 "system" 和 "developer" 两种角色名来创建 SystemMessage。"developer" 是 OpenAI 在其较新的 API 中引入的角色类型，其语义与系统消息基本相同，但在某些模型中会有微妙的行为差异。LangChain 将 "developer" 角色映射为 SystemMessage，同时在 additional_kwargs 中记录 __openai_role__: "developer" 以便在转换回 OpenAI 格式时保留原始角色信息（写入见 langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:650，回读见同文件 :2159）。这种处理方式体现了 LangChain 对"信息无损传递"原则的坚持。

4.2.4 ToolMessage：工具响应

# langchain_core/messages/tool.py

class ToolMessage(BaseMessage, ToolOutputMixin):
    tool_call_id: str
    """关联的工具调用 ID"""

    type: Literal["tool"] = "tool"

    artifact: Any = None
    """工具执行产物，不发送给模型"""

    status: Literal["success", "error"] = "success"
    """工具调用状态"""

ToolMessage（langchain-core-1.2.26/libs/core/langchain_core/messages/tool.py:26）是消息系统中连接模型与外部世界的桥梁。当模型发出工具调用请求（通过 AIMessage.tool_calls）后，应用代码执行实际的工具调用，然后将结果包装为 ToolMessage 反馈给模型。ToolMessage 同时继承了 BaseMessage 和 ToolOutputMixin，后者是一个标记性的空 mixin 类。

ToolMessage 的设计有几个值得深入探讨的要点：

  1. tool_call_id 关联机制：通过 tool_call_id 将工具响应与 AIMessage 中的 ToolCall 一一对应，这在并行工具调用场景下至关重要。

  2. artifact 字段：这是一个巧妙的"二通道"设计。content 存放发送给模型的工具结果摘要，而 artifact 存放完整的工具输出（如图表、原始数据等）。这样既控制了发送给模型的信息量，又保留了完整数据供其他处理环节使用。

  3. ToolOutputMixin：这是一个空的 mixin 类（langchain-core-1.2.26/libs/core/langchain_core/messages/tool.py:16），用作类型标记。当工具以 ToolCall（带 tool_call_id）方式调用时，如果输出是 ToolOutputMixin 的实例，LangChain 会直接返回它；否则会把输出包装为 ToolMessage，不是字符串或内容块列表的输出先被转成字符串（langchain-core-1.2.26/libs/core/langchain_core/tools/base.py:1251 的 _format_output）。

sequenceDiagram
    participant User as 用户
    participant Model as 语言模型
    participant Tool as 外部工具
    User->>Model: HumanMessage("今天北京天气如何？")
    Model->>User: AIMessage(tool_calls=[{name:"get_weather", args:{city:"北京"}, id:"call_123"}])
    User->>Tool: 调用 get_weather(city="北京")
    Tool->>User: {"temp": 25, "weather": "晴"}
    User->>Model: ToolMessage(content="25度，晴", tool_call_id="call_123")
    Model->>User: AIMessage("北京今天天气晴朗，气温25度。")

4.2.5 其他消息类型

除上面四种之外，messages/ 目录里还有三个较少直接使用的子类：ChatMessage（chat.py，type="chat"，多一个必填的 role: str，可承载任意角色名）、FunctionMessage（function.py，type="function"，name 为必填，对应 OpenAI 旧式 function calling 的函数返回，工具调用场景下新代码用 ToolMessage），以及 RemoveMessage（modifier.py，type="remove"）。RemoveMessage 只携带要删除的消息 id，构造时传入非空 content 会抛 ValueError: RemoveMessage does not support 'content' field.；它不是给模型看的，而是交给 LangGraph 的 add_messages 这类 reducer，按 id 从状态里删掉对应消息。除 RemoveMessage 外，每种消息都有对应的 *MessageChunk。

4.3 MessageChunk：流式消息的增量合并

流式输出是现代大语言模型应用的标配能力。用户不愿意等待模型生成完整回复后才看到结果，而是希望看到文字逐步"流淌"出来。LangChain 通过 BaseMessageChunk 及其子类来表示流式传输中的消息片段。Chunk 的核心设计理念是：每个片段都是一个"增量"，它可以与前面的片段通过 + 运算符合并，最终得到一个完整的消息。这种基于运算符重载的合并方式使得流式消息的累积处理变得极其自然。

4.3.1 BaseMessageChunk 的合并机制

# langchain_core/messages/base.py

class BaseMessageChunk(BaseMessage):
    def __add__(self, other: Any) -> BaseMessageChunk:
        if isinstance(other, BaseMessageChunk):
            return self.__class__(
                id=self.id,
                type=self.type,
                content=merge_content(self.content, other.content),
                additional_kwargs=merge_dicts(
                    self.additional_kwargs, other.additional_kwargs
                ),
                response_metadata=merge_dicts(
                    self.response_metadata, other.response_metadata
                ),
            )
        ...

BaseMessageChunk（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:409）通过重载 __add__ 运算符（同文件 :412）实现了消息片段的增量合并。两个 Chunk 相加时，它们的 content、additional_kwargs 和 response_metadata 都会被分别合并。内容合并的核心依赖 merge_content 函数（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:366），该函数根据参与合并的数据类型采取不同的策略：

def merge_content(first_content, *contents):
    merged = "" if first_content is None else first_content
    for content in contents:
        if isinstance(merged, str):
            if isinstance(content, str):
                merged += content              # str + str = str
            else:
                merged = [merged, *content]    # str + list = list
        elif isinstance(content, list):
            merged = merge_lists(merged, content)  # list + list = merge by index
        elif merged and isinstance(merged[-1], str):
            merged[-1] += content              # 追加到最后一个字符串元素
        ...
    return merged

merge_lists（langchain-core-1.2.26/libs/core/langchain_core/utils/_merge.py:89，与之配套的 merge_dicts 在同文件 :6）是一个基于 index 键的列表合并函数，它是整个流式合并机制中最精妙的组件。其工作原理是：右侧列表中带 index（整数，或以 lc_ 开头的字符串）的字典块，若左侧已有 index 相同且 id 不冲突（任一方 id 为空或两者相等）的块，就交给 merge_dicts 合并成一个（其中字符串字段拼接、嵌套字典递归合并）；没有匹配的块，以及不带 index 的元素（包括纯字符串），都直接追加到列表末尾。这种基于索引的合并语义对于流式工具调用场景至关重要：当模型并行调用多个工具时，每个工具调用的 JSON 参数会以多个 Chunk 的形式分批到达，index 字段确保属于同一个工具调用的片段被正确合并在一起。

4.3.2 AIMessageChunk 的特殊处理

AIMessageChunk（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:413）是所有 Chunk 类型中最复杂的，它通过多重继承同时获得了 AIMessage 的工具调用能力和 BaseMessageChunk 的合并能力。除了基本的内容合并之外，它还需要处理流式工具调用的逐步解析：

# langchain_core/messages/ai.py

class AIMessageChunk(AIMessage, BaseMessageChunk):
    type: Literal["AIMessageChunk"] = "AIMessageChunk"
    tool_call_chunks: list[ToolCallChunk] = Field(default_factory=list)
    chunk_position: Literal["last"] | None = None

tool_call_chunks 字段专门用于存储流式传输中的工具调用片段。与完整的 ToolCall（其 args 是已解析的 Python 字典）不同，ToolCallChunk 的 args 是 JSON 字符串的原始片段。例如，一个工具调用可能先收到 args='{"a":'，再收到 args='1}'。当多个 Chunk 通过 __add__ 合并时，这些 JSON 片段会按 index 分组拼接。拼接后的字符串会在 init_tool_calls 模型验证器（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:495）中通过 parse_partial_json（langchain-core-1.2.26/libs/core/langchain_core/utils/json.py）进行解析尝试。parse_partial_json 会自动补全未闭合的 JSON，所以尚不完整的片段通常也能解析成字典（实跑：单独的 args='{"a":' 得到 args={} 的 tool_call），这类工具调用会被放入 tool_calls 列表；只有补全后仍无法解析、或解析结果不是字典（如 'abc'、'[1,'）时，才会被放入 invalid_tool_calls 列表。

chunk_position 字段标记当前 Chunk 在流中的位置。当值为 "last" 时，表示这是流的最终片段。此时 init_tool_calls 验证器会执行一项额外操作：将 content 列表中的 tool_call_chunk 类型块替换为完整的 tool_call 类型块。注意这个替换还有额外的前提——tool_call_chunks 非空、response_metadata["output_version"] == "v1" 且 content 本身是列表，这些条件同时满足才会改写，而且只替换 id 能在 tool_calls 中找到的块（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:560）。目的是让消费者将所有 Chunk 合并后，得到的最终消息在结构上与非流式调用产生的消息保持一致。

graph LR
    subgraph 流式输入
        C1["Chunk 1<br/>content='Hello'"]
        C2["Chunk 2<br/>content=' World'"]
        C3["Chunk 3<br/>tool_call_chunks=..."]
        C4["Chunk 4<br/>tool_call_chunks=..."]
        C5["Chunk 5<br/>chunk_position='last'"]
    end
    subgraph 合并结果
        R["AIMessageChunk<br/>content='Hello World'<br/>tool_calls=[{name:'foo', args:{a:1}}]"]
    end
    C1 --> |"__add__"| C2
    C2 --> |"__add__"| C3
    C3 --> |"__add__"| C4
    C4 --> |"__add__"| C5
    C5 --> R

4.3.3 add_ai_message_chunks 函数

AIMessageChunk.__add__ 实际委托给 add_ai_message_chunks 函数（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:638，token 累加用的 add_usage 在同文件 :721），该函数执行一系列精细的合并操作：

# 简化：省略了 docstring、usage 的 else 分支（usage_metadata = None）与 ID 选择的循环体
def add_ai_message_chunks(left: AIMessageChunk, *others: AIMessageChunk) -> AIMessageChunk:
    content = merge_content(left.content, *(o.content for o in others))
    additional_kwargs = merge_dicts(left.additional_kwargs, *(o.additional_kwargs for o in others))
    response_metadata = merge_dicts(left.response_metadata, *(o.response_metadata for o in others))

    # 合并 tool_call_chunks
    if raw_tool_calls := merge_lists(
        left.tool_call_chunks, *(o.tool_call_chunks for o in others)
    ):
        tool_call_chunks = [create_tool_call_chunk(...) for rtc in raw_tool_calls]
    else:
        tool_call_chunks = []

    # Token 使用量累加
    if left.usage_metadata or any(o.usage_metadata is not None for o in others):
        usage_metadata = left.usage_metadata
        for other in others:
            usage_metadata = add_usage(usage_metadata, other.usage_metadata)

    # ID 选择策略：提供商 ID > lc_run ID > 其他 ID
    ...
    return left.__class__(content=content, ...)

合并过程中 ID 的选择策略值得注意（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:683）：优先使用提供商分配的 ID（既不以 LC_AUTO_PREFIX（即 lc_）也不以 LC_ID_PREFIX（即 lc_run-）开头，两个常量都定义在 langchain-core-1.2.26/libs/core/langchain_core/utils/utils.py:495），其次是 LangChain 生成的运行 ID（lc_run-* 前缀），最后是其他 lc_* ID；同级时先出现的胜出。lc_run-* ID 来自 BaseChatModel：提供商没给 id 时，它把 LC_ID_PREFIX 与本次运行的 run_id 用 - 连起来补到每个流式块上（langchain-core-1.2.26/libs/core/langchain_core/language_models/chat_models.py:588），所以同一次流的所有块 ID 相同，实跑形如 lc_run--01a0d6de-…（两个连字符）；非流式调用补的则是 lc_run--<run_id>-<序号>。这保证了在流式合并后，最终消息的 ID 尽可能有意义。

4.4 Content Blocks：多模态内容的标准化表示

4.4.1 设计理念

Content Block 体系是 LangChain 1.0 版本最具变革性的设计之一。在此之前，多模态内容的表示方式是碎片化的：OpenAI 使用 image_url 块，Anthropic 使用 source 块，Google 使用 inline_data 块。应用代码如果需要在不同提供商之间切换，就必须适配这些不同的格式。Content Block 体系引入了一个统一的、提供商无关的中间格式来应对这个问题。content.py 模块的文档开篇就阐明了这一设计理念：

不同的 LLM 提供商使用截然不同且互不兼容的 API schema。本模块提供了一个统一的、提供商无关的格式来促进这些交互。发送给或来自模型的消息就是一个内容块列表，允许文本、图片和其他内容在一个有序序列中自然交织。

4.4.2 内容块类型一览

graph TB
    CB["ContentBlock<br/>(联合类型)"]
    subgraph 文本类
        T["TextContentBlock<br/>type='text'"]
        R["ReasoningContentBlock<br/>type='reasoning'"]
    end
    subgraph 工具类
        TC["ToolCall<br/>type='tool_call'"]
        TCC["ToolCallChunk<br/>type='tool_call_chunk'"]
        ITC["InvalidToolCall<br/>type='invalid_tool_call'"]
        STC["ServerToolCall<br/>type='server_tool_call'"]
        STCC["ServerToolCallChunk<br/>type='server_tool_call_chunk'"]
        STR["ServerToolResult<br/>type='server_tool_result'"]
    end
    subgraph 多媒体类
        IMG["ImageContentBlock<br/>type='image'"]
        VID["VideoContentBlock<br/>type='video'"]
        AUD["AudioContentBlock<br/>type='audio'"]
        PT["PlainTextContentBlock<br/>type='text-plain'"]
        F["FileContentBlock<br/>type='file'"]
    end
    subgraph 扩展类
        NS["NonStandardContentBlock<br/>type='non_standard'"]
    end
    CB --> T
    CB --> R
    CB --> TC
    CB --> TCC
    CB --> ITC
    CB --> STC
    CB --> STCC
    CB --> STR
    CB --> IMG
    CB --> VID
    CB --> AUD
    CB --> PT
    CB --> F
    CB --> NS

4.4.3 多媒体内容块的统一结构

所有多媒体数据块（Image/Video/Audio/PlainText/File）共享一组标准字段：

字段 类型 说明
type Literal 块类型标识
id NotRequired[str] 唯一标识符
file_id NotRequired[str] 外部文件存储系统的引用
mime_type NotRequired[str] MIME 类型（PlainTextContentBlock 例外：必填的 Literal["text/plain"]）
index NotRequired[int | str] 流式合并时的位置索引
url NotRequired[str] 数据 URL
base64 NotRequired[str] Base64 编码数据
extras NotRequired[dict] 提供商特定元数据

这种统一的结构设计使得不同类型的多媒体数据可以用相同的方式处理。数据可以通过三种途径提供：url 指定远程资源地址、base64 内联编码数据、file_id 引用外部文件存储系统中的文件（如 OpenAI 或 Anthropic 的 Files API）。extras 字段为提供商特定的元数据提供了标准化的存放位置，例如 Google 的思维签名（thought signature）或 Anthropic 的缓存控制指令。

这种设计的一个重要优点是它允许渐进式扩展。当未来出现新的多媒体类型（如 3D 模型或表格数据）时，只需定义新的 TypedDict 类型并添加到 ContentBlock 联合类型中，现有代码完全不受影响。源码中有一段注释明确列出了未来可能添加的模态类型，包括 3D 模型和表格数据。

# libs/core/langchain_core/messages/content.py:498
# ImageContentBlock 示例
class ImageContentBlock(TypedDict):
    type: Literal["image"]
    id: NotRequired[str]
    file_id: NotRequired[str]
    mime_type: NotRequired[str]
    index: NotRequired[int | str]
    url: NotRequired[str]
    base64: NotRequired[str]
    extras: NotRequired[dict[str, Any]]

4.4.4 TextContentBlock 与 Annotation

文本内容块不仅包含文本本身，还支持引用注解（Citation），这对于 RAG（检索增强生成）应用尤为重要：

# libs/core/langchain_core/messages/content.py:207（TextContentBlock）
# + :126（Citation）—— 两处类定义拼在一起，各自的 docstring 已省略
class TextContentBlock(TypedDict):
    type: Literal["text"]
    id: NotRequired[str]
    text: str
    annotations: NotRequired[list[Annotation]]
    index: NotRequired[int | str]
    extras: NotRequired[dict[str, Any]]

class Citation(TypedDict):
    type: Literal["citation"]
    id: NotRequired[str]
    url: NotRequired[str]
    title: NotRequired[str]
    start_index: NotRequired[int]
    end_index: NotRequired[int]
    cited_text: NotRequired[str]
    extras: NotRequired[dict[str, Any]]

Citation 中的 start_index 和 end_index 指向的是响应文本中的位置（不是源文档），而 cited_text 则包含被引用的源文本片段。源码中有一段注释解释了为什么不包含源文档的文本偏移量：跨不同文件格式和编码方案可靠地提取源文档中的精确范围（span）是非常困难的，而 cited_text 字段对于大多数使用场景已经足够。这种务实的设计选择避免了过度工程化，同时保留了引用功能的核心价值。

4.4.5 ReasoningContentBlock：推理内容

随着推理模型（如 OpenAI o 系列、DeepSeek R1、Anthropic 的 extended thinking 等）的兴起，模型的输出不再仅仅是最终答案，还可能包含中间的推理过程。ReasoningContentBlock 应运而生，用于标准化表示模型的推理/思考内容：

# libs/core/langchain_core/messages/content.py:456
class ReasoningContentBlock(TypedDict):
    type: Literal["reasoning"]
    id: NotRequired[str]
    reasoning: NotRequired[str]
    index: NotRequired[int | str]
    extras: NotRequired[dict[str, Any]]

AIMessage 的 content_blocks 属性会执行最佳努力的推理内容提取。不同的模型提供商将推理内容放在不同的位置：Anthropic 直接放在内容块列表中；而 Ollama、DeepSeek、XAI、Groq 等则将其存储在 additional_kwargs["reasoning_content"] 中。LangChain 通过 _extract_reasoning_from_additional_kwargs 辅助函数统一处理后者的情况，将推理内容提取出来并作为 ReasoningContentBlock 插入到内容块列表的开头。选择插入到开头是因为推理过程在逻辑上先于最终回复发生，将其放在列表首位更符合时间顺序。同时，代码会检查是否已经存在推理块，避免重复添加。

4.4.6 提供商格式转换管线

content_blocks 属性实现了一个精巧的多阶段转换管线：

flowchart LR
    A["原始 content<br/>（提供商格式）"] --> B["第一阶段<br/>已知类型直接映射"]
    B --> C["标记 non_standard"]
    C --> D1["v0 旧格式转换"]
    D1 --> D2["OpenAI 格式转换"]
    D2 --> D3["Anthropic 格式转换"]
    D3 --> D4["Google GenAI 格式转换"]
    D4 --> D5["Bedrock Converse 格式转换"]
    D5 --> E["标准化<br/>ContentBlock 列表"]

每个转换步骤只处理 non_standard 类型的块，跳过已经标准化的块。这意味着一旦一个块被某个转换器成功处理，后续的转换器就不会再碰它，保证了转换效率和结果的确定性。

对于 AIMessage，content_blocks 覆写（langchain-core-1.2.26/libs/core/langchain_core/messages/ai.py:243）还有两条快速路径（fast path）：最先判断的是 response_metadata["output_version"] == "v1"——说明内容已经是标准化格式，直接 cast 返回 content 列表，零解析开销；其次才看 response_metadata 里有没有 model_provider，有就用 get_translator(model_provider) 取对应提供商的专用转换器精确转换，完全跳过通用的多轮猜测流程（转换器抛 NotImplementedError 时再退回通用路径）。

需要说清楚的是：这两条都是分支短路，不是缓存。content_blocks 是普通 @property，没有 functools.cached_property、也没有任何实例级记忆——每次访问都会重新走一遍解析。所以在循环里反复读 msg.content_blocks 是有真实代价的，热路径上应当自己取一次存进局部变量。

4.5 ToolCall 与工具调用体系

4.5.1 三级工具调用结构

LangChain 的工具调用体系包含三种密切相关但用途不同的类型：

# ToolCall / ToolCallChunk 取自 libs/core/langchain_core/messages/tool.py:206 / :261
# （AIMessage.tool_calls、AIMessageChunk.tool_call_chunks 用的就是这一组）
# 完整的工具调用（非流式）
class ToolCall(TypedDict):
    name: str                           # 工具名称
    args: dict[str, Any]                # 已解析的参数字典
    id: str | None                      # 关联 ID
    type: NotRequired[Literal["tool_call"]]

# 流式工具调用片段
class ToolCallChunk(TypedDict):
    name: str | None                    # 可能为 None（部分传输）
    args: str | None                    # JSON 字符串片段
    id: str | None
    index: int | None                   # 在序列中的索引，用于合并
    type: NotRequired[Literal["tool_call_chunk"]]

# 解析失败的工具调用（只定义在 content.py:336，tool.py 从那里 import）
class InvalidToolCall(TypedDict):
    type: Literal["invalid_tool_call"]
    id: str | None
    name: str | None
    args: str | None                    # 原始字符串，非字典
    error: str | None                   # 错误描述
    index: NotRequired[int | str]
    extras: NotRequired[dict[str, Any]]

ToolCallChunk 的 index 字段在流式合并中起到关键作用。考虑这样一个场景：模型同时调用了 get_weather 和 get_news 两个工具，流式传输中的 Chunk 可能交替到达。index=0 标识第一个工具调用的片段，index=1 标识第二个工具调用的片段。merge_lists 函数会按照 index 值将属于同一工具调用的 Chunk 合并在一起，确保每个工具调用的参数 JSON 被正确拼接。

还需要注意的是，ToolCall 和 ToolCallChunk 在 content.py 和 tool.py 中各有一份定义。tool.py 中的是旧版定义（用于 AIMessage.tool_calls 等字段），content.py 中的是新版定义（作为标准内容块类型）。两者的结构基本相同，差别在于：新版的 type 是必填字段，并多了可选的 extras；新版 ToolCall 还多了可选的 index，新版 ToolCallChunk 的 index 则由必填的 int | None 放宽为可选的 int | str。这种看似冗余的双重定义是版本演进过程中保持向后兼容的结果。

4.5.2 default_tool_parser：向后兼容的解析器

当工具调用数据以 OpenAI 格式存储在 additional_kwargs["tool_calls"] 中时，default_tool_parser 负责将其解析为标准格式：

# libs/core/langchain_core/messages/tool.py:349（签名在源码里是多行）
def default_tool_parser(raw_tool_calls: list[dict]) -> tuple[list[ToolCall], list[InvalidToolCall]]:
    tool_calls = []
    invalid_tool_calls = []
    for raw_tool_call in raw_tool_calls:
        if "function" not in raw_tool_call:
            continue
        function_name = raw_tool_call["function"]["name"]
        try:
            function_args = json.loads(raw_tool_call["function"]["arguments"])
            tool_calls.append(tool_call(
                name=function_name or "",
                args=function_args or {},
                id=raw_tool_call.get("id"),
            ))
        except json.JSONDecodeError:
            invalid_tool_calls.append(invalid_tool_call(
                name=function_name,
                args=raw_tool_call["function"]["arguments"],
                id=raw_tool_call.get("id"),
                error=None,
            ))
    return tool_calls, invalid_tool_calls

解析失败不会抛出异常，而是将失败的调用放入 invalid_tool_calls 列表。这种"宽容解析"的设计具有重要的实践价值：在生产环境中，模型有时会生成格式不完全正确的 JSON（例如多余的逗号或缺少引号），如果对此直接抛出异常，可能导致整个对话流程中断。通过将解析失败的调用收集而非丢弃，应用层可以自行决定处理策略——重试、修复、忽略或提示用户。同时，其他格式正确的工具调用不会受到影响，保证了系统的可用性。

4.5.3 ServerToolCall 与服务端工具调用

1.0 版本引入了 ServerToolCall 和 ServerToolResult 来表示在服务端执行的工具调用（如代码执行、网络搜索等）：

# libs/core/langchain_core/messages/content.py:372 / :425（节选：省略了两者的 index、extras，
# 以及 ServerToolResult 的可选 id）
class ServerToolCall(TypedDict):
    type: Literal["server_tool_call"]
    id: str
    name: str
    args: dict[str, Any]

class ServerToolResult(TypedDict):
    type: Literal["server_tool_result"]
    tool_call_id: str
    status: Literal["success", "error"]
    output: NotRequired[Any]

与客户端的 ToolCall 不同，ServerToolCall 的 id 字段是必需的 str（不允许 None），ServerToolResult 正是通过 tool_call_id 与它关联。

4.6 消息工具函数

4.6.1 convert_to_messages：万能消息转换器

# libs/core/langchain_core/messages/utils.py:735
def convert_to_messages(
    messages: Iterable[MessageLikeRepresentation] | PromptValue,
) -> list[BaseMessage]:
    if isinstance(messages, PromptValue):
        return messages.to_messages()
    return [_convert_to_message(m) for m in messages]

convert_to_messages（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:735）是消息系统中最常用的工具函数之一。它在 LangChain 的许多核心路径上被调用：BaseChatModel._convert_input 中用它将用户输入转换为消息列表（langchain-core-1.2.26/libs/core/langchain_core/language_models/chat_models.py:433），MessagesPlaceholder.format_messages 中用它把占位变量的值规整成消息（langchain-core-1.2.26/libs/core/langchain_core/prompts/chat.py:187），各种消息操作工具函数中也都以它作为入口。它能将多种格式的输入统一转换为 BaseMessage 列表。核心的 _convert_to_message 函数（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:675）内部支持以下输入格式：

  • BaseMessage 实例 -- 直接返回
  • 2-元组 (role, content) -- 例如 ("human", "Hello")
  • 字典 {"role": "...", "content": "..."} -- OpenAI 格式
  • 纯字符串 -- 视为 HumanMessage

角色映射支持多种别名："human" 和 "user" 都映射到 HumanMessage，"ai" 和 "assistant" 都映射到 AIMessage，"system" 和 "developer" 都映射到 SystemMessage，此外 "function"、"tool"、"remove" 分别对应 FunctionMessage、ToolMessage、RemoveMessage（分支见 langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:650）。

4.6.2 filter_messages：消息过滤器

filter_messages（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:806）提供了基于名称、类型和 ID 的多维度消息过滤能力：

@_runnable_support
def filter_messages(
    messages,
    *,
    include_names=None,     # 保留指定名称
    exclude_names=None,     # 排除指定名称
    include_types=None,     # 保留指定类型
    exclude_types=None,     # 排除指定类型
    include_ids=None,       # 保留指定 ID
    exclude_ids=None,       # 排除指定 ID
    exclude_tool_calls=None # 排除工具调用
) -> list[BaseMessage]:

@_runnable_support 装饰器（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:784）是消息工具函数的一项巧妙设计。它赋予了函数"双面身份"：当传入消息列表时，函数直接执行过滤操作并返回结果（即作为普通函数使用）；当不传入消息参数（messages=None）时，函数返回一个 RunnableLambda 对象，可以作为 LCEL 链中的一个环节参与管道组合。这种设计让同一个函数同时满足了直接调用和链式组合两种使用模式，大大提高了 API 的灵活性：

# 直接调用
filtered = filter_messages(messages, include_types=["human", "ai"])

# 作为 Runnable 使用
chain = filter_messages(include_types=["human", "ai"]) | model

exclude_tool_calls 参数支持两种模式：True 表示排除所有工具调用相关消息；传入具体的 tool_call_id 列表则只排除对应的工具调用。

4.6.3 trim_messages：消息裁剪器

trim_messages（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:1082）是对话历史管理中不可或缺的工具，也是目前 LangChain 消息工具函数中参数最丰富、功能最强大的一个。在长对话场景中，消息历史可能超出模型的上下文窗口限制，需要智能地裁剪。简单的截断会破坏消息序列的有效性（例如在 ToolMessage 之前截断，导致缺少对应的 AIMessage 工具调用），因此 trim_messages 提供了一系列参数来确保裁剪后的消息序列仍然是有效的模型输入：

@_runnable_support
def trim_messages(
    messages,
    *,
    max_tokens: int,
    token_counter: Callable | BaseLanguageModel | Literal["approximate"],
    strategy: Literal["first", "last"] = "last",
    allow_partial: bool = False,
    end_on: str | type | Sequence | None = None,
    start_on: str | type | Sequence | None = None,
    include_system: bool = False,
    text_splitter: Callable | None = None,
) -> list[BaseMessage]:

几个关键参数的设计思路：

  • strategy="last"：保留最新的消息，丢弃旧消息，这是对话场景最常见的需求
  • include_system=True：若原序列第 0 条是 SystemMessage，即使裁剪掉了大量旧消息，它仍然保留在开头（仅 strategy="last" 可用，配 "first" 抛 ValueError）
  • start_on="human"：确保裁剪后的消息序列（开头保留的系统消息之后）以 HumanMessage 开头，满足大多数模型的输入要求；同样只能配 strategy="last"
  • token_counter="approximate"：使用近似 token 计数器 count_tokens_approximately（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:2186），在热路径上避免精确 token 计数的开销
# 典型用法：保留系统消息，从 HumanMessage 开始，最多 4096 token
trimmed = trim_messages(
    messages,
    max_tokens=4096,
    strategy="last",
    token_counter="approximate",
    start_on="human",
    include_system=True,
)

4.6.4 其他消息工具

除了上述三个核心函数外，消息工具模块还提供了一系列实用的辅助函数：

  • get_buffer_string（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:287）：将消息列表转换为格式化字符串，由 format: Literal["prefix", "xml"] 参数选择，默认 "prefix"。"prefix" 格式使用传统的 "Role: content" 格式，简单但在内容包含类似前缀时可能产生歧义。"xml" 格式使用 <message type="role"> 标签包裹内容，并通过 xml.sax.saxutils.escape 和 quoteattr 自动转义特殊字符，能够可靠地处理任何文本内容。XML 格式还支持多模态内容块的格式化，但 base64 编码的数据会被自动跳过（因为它们对于字符串表示没有可读性价值）。

  • merge_message_runs（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:951，同样带 @_runnable_support）：把连续的同类消息合并成一条——内部先转成 Chunk 再用 + 合并，两边都是非空字符串时中间插入 chunk_separator（默认 "\n"）；ToolMessage 各有不同的 tool_call_id，不参与合并。实跑时，内容分别为 a、b 的两条相邻 HumanMessage 加一条 AIMessage，合并后只剩两条消息，第一条的内容是 a 与 b 以换行相连。

  • message_chunk_to_message（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:555）：将 MessageChunk 转换为对应的普通 Message。其实现巧妙地利用了 Python 的 MRO（方法解析顺序）-- Chunk 类的第一个父类就是对应的非 Chunk 类：

def message_chunk_to_message(chunk: BaseMessage) -> BaseMessage:
    if not isinstance(chunk, BaseMessageChunk):
        return chunk
    ignore_keys = ["type"]
    if isinstance(chunk, AIMessageChunk):
        ignore_keys.extend(["tool_call_chunks", "chunk_position"])
    return chunk.__class__.__mro__[1](
        **{k: v for k, v in chunk.__dict__.items() if k not in ignore_keys}
    )
  • AnyMessage（langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py:86）：使用 Pydantic 的 Discriminator 和 Tag 机制定义的联合类型，支持基于 type 字段的自动反序列化：
AnyMessage = Annotated[
    Annotated[AIMessage, Tag(tag="ai")]
    | Annotated[HumanMessage, Tag(tag="human")]
    | Annotated[ChatMessage, Tag(tag="chat")]
    | Annotated[SystemMessage, Tag(tag="system")]
    | Annotated[FunctionMessage, Tag(tag="function")]
    | Annotated[ToolMessage, Tag(tag="tool")]
    | ...,  # 省略：6 个对应的 *MessageChunk 类，tag 为各自类名
    Field(discriminator=Discriminator(_get_type)),
]

4.7 设计决策分析

4.7.1 TypedDict vs Pydantic Model

Content Block 使用 TypedDict 而非 Pydantic Model，这是一个经过深思熟虑的选择：

  1. 性能：TypedDict 是纯类型注解，运行时就是普通字典，没有 Pydantic 验证的开销。在高频流式场景下，内容块数量多，省掉逐块验证更有意义。

  2. 互操作性：普通字典与 JSON 序列化/反序列化天然兼容，不需要额外的转换步骤。

  3. 灵活性：TypedDict 允许 NotRequired 字段，使得内容块可以只包含必要的字段。同时 extras 字段为提供商特定数据提供了标准化的扩展点。

4.7.2 content 与 content_blocks 的双轨制

content 字段存储原始的提供商格式数据，而 content_blocks 属性提供按需计算的标准化视图。这种双轨设计的好处是：

  1. 不用则不付费：如果应用不需要标准化视图（例如只处理纯文本），content_blocks 的解析开销完全不会产生——但反过来，用一次就算一次，属性本身不做记忆。

  2. 无损存储：原始数据保留在 content 中，序列化/反序列化不会丢失信息。

  3. 渐进迁移：通过 output_version="v1" 配置，可以让模型输出直接存储标准化格式，为未来的全面迁移铺路。

4.7.3 宽容解析策略

整个消息系统贯穿了一种"宽容解析"的设计哲学：

  • 工具调用解析失败不抛异常，而是放入 invalid_tool_calls
  • 未知的内容块标记为 non_standard 而非报错
  • ToolMessage 的 coerce_args 验证器尝试把不合规的 content（既非字符串也非列表，或列表中既非字符串也非字典的元素）强制转换为字符串

这种设计极大地提高了系统的健壮性，使其能够优雅地处理各种预期外的输入，而不是在第一个异常处崩溃。

4.8 小结

LangChain 的消息系统是整个框架的数据基础设施。本章深入分析了以下核心内容：

  • BaseMessage 层级体系：以 Serializable 为根，通过 type 字段辨别的四种核心消息类型（Human/AI/System/Tool），每种类型都针对其使用场景进行了精确的字段设计（基类见 langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:93）。

  • 流式消息合并：BaseMessageChunk（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:409）通过 __add__ 运算符和 merge_content（同文件 :366）/ merge_lists（langchain-core-1.2.26/libs/core/langchain_core/utils/_merge.py:89）函数实现了增量合并，AIMessageChunk 进一步处理了流式工具调用的合并和解析。

  • Content Block 多模态体系：1.0 版本引入的标准化内容块系统，通过 content_blocks 属性（langchain-core-1.2.26/libs/core/langchain_core/messages/base.py:200）的惰性转换管线，将各种提供商格式统一为 Text/Image/Audio/Video/File/Reasoning 等标准类型。

  • 工具调用三级结构：ToolCall（完整调用）、ToolCallChunk（流式片段）、InvalidToolCall（解析失败）构成了完整的工具调用生命周期表示，三者都是 TypedDict；消息字段使用的 ToolCall、ToolCallChunk 定义在 langchain-core-1.2.26/libs/core/langchain_core/messages/tool.py:206、:261，InvalidToolCall 只在 content.py:336 定义（content.py 的 :247、:291 另有一组作为标准内容块的同名新版定义）。

  • 实用工具函数：convert_to_messages 的万能转换能力、filter_messages 的多维度过滤、trim_messages 的智能裁剪，以及它们通过 @_runnable_support 获得的 LCEL 集成能力——它们都在 langchain-core-1.2.26/libs/core/langchain_core/messages/utils.py 一个文件里（依次为 :735、:806、:1082、:784）。

消息系统的设计充分体现了"简单场景简单用，复杂场景不受限"的理念，它既是一个轻量的数据传输格式，又是一个可扩展的多模态内容表示框架。

跨专栏对照

  • 消息类型化：Message 角色体系（Human / AI / System / Tool）把"这句话谁说的"从文本约定提升成数据结构，与 MCP 协议的 role 字段、《OpenClaw 设计与实现》专栏讲的消息管线遵循同一种"显式类型优于隐式约定"的思路。
  • MessageChunk 的 Monoid 式语义：__add__ 在一次操作里合并 content、metadata、tool_call_chunks，可以看作幺半群（monoid）式的运算。LangGraph 的 Channel reducer（见《LangGraph 设计与实现》专栏第 4 章）把同一抽象推到整个图状态的并发合并。
  • Content Block 与多厂商兼容：本章介绍的 block 结构要覆盖的是各家互不兼容的 wire format——langchain-core-1.2.26/libs/core/langchain_core/messages/block_translators/ 下有 8 个翻译模块（anthropic / openai / google_genai / google_vertexai / bedrock / bedrock_converse / groq / langchain_v0），各负责一种格式（langchain_v0 对应 LangChain 自己的 v0 旧格式）；这些翻译器放在 langchain-core 里，partner 包一侧的消息格式转换见第 17 章 §17.8（以 ChatOpenAI 为例）。
  • 按需计算的 content_blocks：不访问就不解析，没用到多模态视图的 Message 零开销——这一半和 Vue 3 的 computed（见《Vue 3 设计与实现》专栏）、Tokio 的 Future（见《Tokio 源码深度解析》专栏）的惰性求值同源。但另一半不同：computed 求值后会记住结果，content_blocks 只是个普通 @property，每次读都重算一遍——只有惰性，没有记忆。