Day 06 / 共 20 天 · 阶段2 模型·消息·提示·输出

消息体系:HumanMessage / AIMessage / ToolMessage 与 tool_calls

Day05 我们看到 BaseChatModel.invoke() 吃进去、吐出来的都是"消息"。今天正式拆开这套消息体系——源码全在 libs/core/langchain_core/messages/。弄清三件事:①一条消息由哪几块组成(content / type / additional_kwargs…);②AI 要调工具时 tool_calls 长什么样、ToolMessage 怎么把结果"对号入座"送回去;③流式输出的碎片(Chunk)是怎么用 + 拼回完整消息的。这是后面 Prompt(D07)、工具调用(D14)、Agent(D15)的共同地基。

📍 你在 20 天里的位置(阶段2:模型·消息·提示·输出 · D05-08)
D05 ChatModel D06 消息体系 D07 Prompt 模板 D08 输出解析 S3 数据与RAG S4 工具/Agent S5 进阶收官
💡 先用两个类比兜住今天 类比一:一次对话就是一份会议纪要。每条消息 = 纪要里的一条发言记录:发言人标签type:human/ai/system/tool)+ 发言内容content)+ 旁注additional_kwargsresponse_metadata)。模型每次"读完整份纪要再发言",所以纪要格式必须统一。类比二:tool_calls 像 AI 在会上开的派工单——"帮我查下北京天气,工单号 call_abc123";工具干完活,用 ToolMessage(tool_call_id="call_abc123") 拿着同一个工单号回来交差,谁的活对应谁的结果,绝不会张冠李戴。
L01

痛点:对话不是一根字符串

🤔 痛点早期玩 LLM 就是"塞一个字符串、拿一个字符串"。但真实对话立刻碰到四个问题:①多轮对话里谁说的哪句要分清(用户?AI?系统指令?);②AI 说"我要调工具"时,这不是普通文本,得有结构化的调用请求;③工具执行完,结果要准确回填给对应的那次调用(AI 可能一口气要求并行调 3 个工具);④流式输出一小段一小段来,最后要能拼回一条完整消息。字符串统统扛不住。
💡 本质:给对话定一套"标准会议纪要格式"LangChain 的答案是一棵消息类型树:BaseMessage 定义"一条发言记录长什么样",HumanMessage/AIMessage/SystemMessage/ToolMessage 是四种发言人。所有 chat 模型(OpenAI/Anthropic/千问…)的输入输出都翻译成这套格式——纪要格式统一了,换哪家模型开会都照读不误。这是 langchain_core 里被依赖最广的一层。
位置角色
BaseMessagemessages/base.py:93抽象底座:content/type/additional_kwargs…
HumanMessagemessages/human.py:9用户发言(type="human")
SystemMessagemessages/system.py:9系统指令/人设(type="system")
AIMessagemessages/ai.py:160模型回复,带 tool_calls/usage_metadata
ToolMessagemessages/tool.py:26工具执行结果,带 tool_call_id
L02

BaseMessage:所有消息的底座

先看底座 BaseMessagelibs/core/langchain_core/messages/base.py:93),一条"发言记录"的全部字段:

# libs/core/langchain_core/messages/base.py:93
class BaseMessage(Serializable):
    """Base abstract message class.
    Messages are the inputs and outputs of a chat model."""

    content: str | list[str | dict[Any, Any]]          # 发言内容:字符串 或 多模态块列表
    additional_kwargs: dict[Any, Any] = Field(default_factory=dict)
    # 厂商私有的附加数据(例如 OpenAI 原始格式的 tool calls)
    response_metadata: dict[Any, Any] = Field(default_factory=dict)
    # 响应元数据:response headers、logprobs、token 数、模型名…
    type: str            # 消息类型标签,每个子类唯一("human"/"ai"/"tool"…),用于序列化识别
    name: str | None = None    # 可选的人类可读名字
    id: str | None = Field(default=None, coerce_numbers_to_str=True)  # 可选唯一 ID,最好由厂商提供
content发言正文。注意类型是 str | list——纯文本就是字符串;多模态(文字+图片)就是一个块列表,如 [{"type":"text","text":"这是什么"},{"type":"image_url",...}]
type★发言人标签。子类各自写死:human/ai/system/tool。序列化存盘再读回来时,靠它认出"这条该还原成哪个类"。
additional_kwargs厂商私货区。比如 OpenAI 返回的原始 tool_calls JSON 就先放这里,LangChain 再解析成标准字段(L04 会看到这个"向后兼容"逻辑)。
response_metadata和内容无关的"回执信息":用了多少 token、什么模型、logprobs 等。写监控/计费逻辑就从这读。
大白话一条消息 = 谁说的(type)+ 说了啥(content)+ 两个口袋(additional_kwargs 装厂商私货,response_metadata 装回执)。后面所有消息类都只是在这个底座上"多加几个字段"。
L03

Human / System / AI:三种发言人

HumanMessagelibs/core/langchain_core/messages/human.py:9)薄得惊人——它几乎只是"把 type 写死成 human":

# libs/core/langchain_core/messages/human.py:9
class HumanMessage(BaseMessage):
    """Message from the user."""

    type: Literal["human"] = "human"        # ★唯一的实质区别:发言人标签写死

    def __init__(self, content=None, content_blocks=None, **kwargs):
        """Specify `content` as positional arg or `content_blocks` for typing."""
        if content_blocks is not None:      # 新式多模态写法:显式传内容块列表
            super().__init__(content=cast("list[str | dict[Any, Any]]", content_blocks), **kwargs)
        else:
            super().__init__(content=content, **kwargs)   # 常规写法:HumanMessage("你好")

SystemMessagemessages/system.py:9)结构完全一样,只是 type="system"。真正"重"的是 AIMessagelibs/core/langchain_core/messages/ai.py:160)——模型的回复除了文本还带三样标准化好的东西:

# libs/core/langchain_core/messages/ai.py:160
class AIMessage(BaseMessage):
    """Message from an AI.
    ...consists of both the raw output as returned by the model and
    standardized fields (e.g., tool calls, usage metadata) added by the
    LangChain framework."""

    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"
tool_calls★AI 开出的"派工单"列表(L04 细看)。不管你用 OpenAI 还是 Anthropic,这里都是同一种标准格式——各厂商的原始格式被 LangChain 翻译统一了。
invalid_tool_calls模型有时生成的调用参数不是合法 JSON。LangChain 不直接丢掉,而是放进"废单堆",让你有机会自己补救(比如让模型重试)。
usage_metadata统一格式的 token 用量(input_tokens/output_tokens/total_tokens)。写计费、限额逻辑不用再适配每家厂商的字段名。
💡 本质:AIMessage = 原始回复 + 标准化附加层源码注释说得很直白:AIMessage 同时装着"模型原样返回的输出"和"LangChain 框架加工出的标准化字段"。这就像会议纪要里,除了记下 AI 的原话,书记员还把它口头开的工单誊抄成标准工单格式——后续流程(工具执行、计费)都只认标准格式,不用管 AI 是用哪家"方言"说的。
L04

tool_calls:AI 的"派工单"长什么样

工单本体是个 TypedDict:ToolCalllibs/core/langchain_core/messages/tool.py:206):

# libs/core/langchain_core/messages/tool.py:206
class ToolCall(TypedDict):
    """Represents an AI's request to call a tool.
    Example:
        {"name": "foo", "args": {"a": 1}, "id": "123"}
    """
    name: str                 # 要调哪个工具
    args: dict[str, Any]      # 调用参数(已经解析成 dict,不是 JSON 字符串!)
    id: str | None            # ★工单号:多个并行调用时靠它对号入座
    type: NotRequired[Literal["tool_call"]]   # 判别标签
name + args"调 get_weather,参数 {"city": "北京"}"。注意 args 已是 Python dict——OpenAI 原始返回里是 JSON 字符串,LangChain 帮你解析好了。
id★工单号。源码注释:"An identifier is needed to associate a tool call request with a tool call result in events when multiple concurrent tool calls are made"——AI 一口气派 3 个工单时,3 个结果各自凭号入座。
厂商兼容小细节:老代码把 OpenAI 原始 tool_calls 塞在 additional_kwargs 里,AIMessage 有个校验器 _backwards_compat_tool_callsmessages/ai.py:309)会自动把它们解析成标准 tool_calls / invalid_tool_calls——这就是"翻译统一"发生的现场
📝 真实值:一条带工单的 AIMessage 你问 "北京今天多少度?",模型绑定了 get_weather 工具后返回:
AIMessage(content="", tool_calls=[{"name": "get_weather", "args": {"city": "北京"}, "id": "call_Jja7J89XsjrOLA5r", "type": "tool_call"}])
注意 content 是空字符串——这轮 AI 没说话,只开了工单。你的代码看到 msg.tool_calls 非空,就知道该去执行工具了(这正是 D14/D15 Agent 循环的判断条件)。
L05

ToolMessage:拿着工单号回来交差

工具执行完,结果装进 ToolMessagelibs/core/langchain_core/messages/tool.py:26)送回给模型:

# libs/core/langchain_core/messages/tool.py:26
class ToolMessage(BaseMessage, ToolOutputMixin):
    """Message for passing the result of executing a tool back to a model.

    `tool_call_id` is used to associate the tool call request with the tool
    call response. Useful in situations where a chat model is able to
    request multiple tool calls in parallel."""

    tool_call_id: str                    # ★回填的工单号,必填!
    type: Literal["tool"] = "tool"
    artifact: Any = None                 # 不发给模型的"完整原始产物"
tool_call_id: str★必填字段(没有默认值)。必须等于 L04 那张工单的 id。少了它,模型(尤其 OpenAI API)直接报错——没有工单号的交差不收
content发给模型看的结果摘要。源码例子:ToolMessage(content="42", tool_call_id="call_Jja7J89XsjrOLA5r!MEOW!SL")
artifact★贴心设计:工具的完整产物(比如生成的一整张 base64 图、完整 stdout/stderr)放这里,不发给模型省 token;模型只看 content 里的摘要。源码例子就是"只把 stdout 摘要给模型,完整输出带图放 artifact"。
tool_calls → 执行 → ToolMessage:凭工单号闭环 AIMessage tool_calls: [{id: "call_abc", …}] 派工单 工具执行 get_weather(city="北京") 交差 ToolMessage tool_call_id: "call_abc" 消息列表追加后再次 invoke 模型 [Human, AI(带工单), Tool(带同号)] → AI 用结果作答 id 相同 → 对号入座
图注:工单号(tool_call id)贯穿"请求→执行→回填"全程。并行开 3 张工单,就回 3 条各带其号的 ToolMessage。
⚠️ 坑:ToolMessage 不能"裸发"消息序列必须是 AIMessage(带 tool_calls) 之后紧跟对应的 ToolMessage。你要是漏掉带工单的那条 AIMessage、或者 tool_call_id 对不上号,OpenAI 直接报 400。手写 Agent 循环时这是最常见的翻车点——D14 会看 LangChain 怎么帮你自动维护这个闭环。
L06

Chunk 与 merge_content:流式碎片怎么拼回整句

🤔 痛点Day05 说过 stream() 会一小段一小段吐 AIMessageChunk。碎片好显示,但最后写进对话历史时需要一条完整消息。谁来拼?怎么拼 content 是字符串、又是列表的情况?

答案是 Chunk 类支持 + 运算:BaseMessageChunk.__add__libs/core/langchain_core/messages/base.py:412),核心内容拼接靠 merge_contentlibs/core/langchain_core/messages/base.py:366):

# libs/core/langchain_core/messages/base.py:366
def merge_content(first_content, *contents):
    """Merge multiple message contents."""
    merged = "" if first_content is None else first_content
    for content in contents:
        if isinstance(merged, str):
            if isinstance(content, str):
                merged += content                # 字符串 + 字符串:直接拼
            else:
                merged = [merged, *content]      # 字符串 + 列表:升格成列表
        elif isinstance(content, list):
            merged = merge_lists(merged, content)   # 列表 + 列表:按块合并
        elif merged and isinstance(merged[-1], str):
            merged[-1] += content                # 列表末尾是字符串:接着写
        ...
    return merged
str + str最常见:"你好" + ",世界""你好,世界"。纯文本流式就是不断走这条分支。
str + list前面是文本、后面来了多模态块 → 把文本升格为列表的第一个元素,整体变成块列表。类型能"就高不就低"。
__add__ 的源码示例messages/base.py:412 的 docstring 直接写着:AIMessageChunk(content="Hello") + AIMessageChunk(content=" World") = AIMessageChunk(content="Hello World")
💡 本质:碎片自己知道怎么变完整这像看直播时的弹幕逐字上屏——每个字都是独立事件,但播放器负责把它们拼成完整句子。LangChain 把"怎么拼"内聚在 Chunk 类型自己身上(+ 运算符),于是框架任何地方收集流式输出都只需 final = None; for chunk in stream: final = chunk if final is None else final + chunk,不用关心 content 是文本还是多模态块。顺带一提,工单碎片也一样:ToolCallChunkmessages/tool.py:261)里 args 还是没拼完的 JSON 字符串片段,拼完才解析成 dict。
另外两个常用工具:convert_to_messagesmessages/utils.py:786)把 ("human", "你好") 这类简写元组批量转成消息对象(D07 的 ChatPromptTemplate 就靠它);merge_message_runsmessages/utils.py:1002)把连续同类型消息合并——有些模型不允许连续两条 human 消息。
L07

串起来 + 今日小结

📝 真实值:一轮完整的工具调用对话 消息列表的演变:
[HumanMessage("北京今天多少度?")] → invoke →
② 模型回 AIMessage(content="", tool_calls=[{"name":"get_weather","args":{"city":"北京"},"id":"call_abc","type":"tool_call"}]) → 追加进列表 →
③ 执行工具得 "31°C,晴",追加 ToolMessage(content="31°C,晴", tool_call_id="call_abc") → 再 invoke →
④ 模型回 AIMessage(content="北京今天 31°C,晴天。", usage_metadata={"input_tokens": 89, "output_tokens": 14, "total_tokens": 103})
四条消息、一个工单号,就是所有 Agent 的最小循环单元。

👶 小白:tool_calls 在 AIMessage 上,additional_kwargs 里好像也有一份 tool_calls,到底看哪个?

👨‍🏫 老师:永远看 msg.tool_calls 这个标准字段。additional_kwargs 里那份是厂商原始格式(比如 OpenAI 的 args 还是 JSON 字符串),是"翻译前的原文",仅为兼容保留;_backwards_compat_tool_calls(ai.py:309)校验器会自动把原文翻译成标准 tool_calls(args 已解析成 dict)。写业务永远用标准字段,换模型厂商代码零改动——这正是消息体系存在的意义。

🧠 今天你应该能回答

  • 一条消息由什么组成?(content + type + additional_kwargs + response_metadata,base.py:93)
  • HumanMessage 和 BaseMessage 差在哪?(几乎只差 type="human" 写死)
  • AIMessage 比别的消息多什么?(tool_calls / invalid_tool_calls / usage_metadata 三个标准化字段)
  • 一张 ToolCall 工单有哪三要素?(name / args(dict) / id)
  • ToolMessage 凭什么"对号入座"?(tool_call_id 必填,等于工单的 id)
  • 流式碎片怎么拼回完整消息?(Chunk 支持 +,content 由 merge_content 按 str/list 分情况合并)

✋ 10 分钟动手

cd /Users/bitmart/work/codes/github/AI_WORK/langchain/libs/core/langchain_core

# 1. 底座与三种发言人
sed -n '93,140p'  messages/base.py     # BaseMessage 全部字段
sed -n '9,30p'    messages/human.py    # HumanMessage:只是 type 写死
sed -n '160,185p' messages/ai.py       # AIMessage:tool_calls/usage_metadata

# 2. 工单闭环
sed -n '206,240p' messages/tool.py     # ToolCall 工单三要素
sed -n '26,75p'   messages/tool.py     # ToolMessage:tool_call_id 必填

# 3. 流式拼接
sed -n '366,408p' messages/base.py     # merge_content 分情况合并
grep -n "_backwards_compat_tool_calls" messages/ai.py   # 厂商格式翻译现场
明日预告 · Day 07:消息得有人"生产"。手拼 [SystemMessage(...), HumanMessage(...)] 太原始了——明天看 prompts/PromptTemplate 填空、ChatPromptTemplate.from_messages 批量产消息、MessagesPlaceholder 给聊天历史留座位,以及 few-shot 范例注入。
← Day 05 ChatModel 抽象 Day 07 · Prompt 模板 →