消息体系: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)的共同地基。
type:human/ai/system/tool)+ 发言内容(content)+ 旁注(additional_kwargs、response_metadata)。模型每次"读完整份纪要再发言",所以纪要格式必须统一。类比二:tool_calls 像 AI 在会上开的派工单——"帮我查下北京天气,工单号 call_abc123";工具干完活,用 ToolMessage(tool_call_id="call_abc123") 拿着同一个工单号回来交差,谁的活对应谁的结果,绝不会张冠李戴。痛点:对话不是一根字符串
BaseMessage 定义"一条发言记录长什么样",HumanMessage/AIMessage/SystemMessage/ToolMessage 是四种发言人。所有 chat 模型(OpenAI/Anthropic/千问…)的输入输出都翻译成这套格式——纪要格式统一了,换哪家模型开会都照读不误。这是 langchain_core 里被依赖最广的一层。| 类 | 位置 | 角色 |
|---|---|---|
BaseMessage | messages/base.py:93 | 抽象底座:content/type/additional_kwargs… |
HumanMessage | messages/human.py:9 | 用户发言(type="human") |
SystemMessage | messages/system.py:9 | 系统指令/人设(type="system") |
AIMessage | messages/ai.py:160 | 模型回复,带 tool_calls/usage_metadata |
ToolMessage | messages/tool.py:26 | 工具执行结果,带 tool_call_id |
BaseMessage:所有消息的底座
先看底座 BaseMessage(libs/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 等。写监控/计费逻辑就从这读。Human / System / AI:三种发言人
HumanMessage(libs/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("你好")
SystemMessage(messages/system.py:9)结构完全一样,只是 type="system"。真正"重"的是 AIMessage(libs/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)。写计费、限额逻辑不用再适配每家厂商的字段名。tool_calls:AI 的"派工单"长什么样
工单本体是个 TypedDict:ToolCall(libs/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 个结果各自凭号入座。additional_kwargs 里,AIMessage 有个校验器 _backwards_compat_tool_calls(messages/ai.py:309)会自动把它们解析成标准 tool_calls / invalid_tool_calls——这就是"翻译统一"发生的现场。"北京今天多少度?",模型绑定了 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 循环的判断条件)。ToolMessage:拿着工单号回来交差
工具执行完,结果装进 ToolMessage(libs/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"。AIMessage(带 tool_calls) 之后紧跟对应的 ToolMessage。你要是漏掉带工单的那条 AIMessage、或者 tool_call_id 对不上号,OpenAI 直接报 400。手写 Agent 循环时这是最常见的翻车点——D14 会看 LangChain 怎么帮你自动维护这个闭环。Chunk 与 merge_content:流式碎片怎么拼回整句
stream() 会一小段一小段吐 AIMessageChunk。碎片好显示,但最后写进对话历史时需要一条完整消息。谁来拼?怎么拼 content 是字符串、又是列表的情况?答案是 Chunk 类支持 + 运算:BaseMessageChunk.__add__(libs/core/langchain_core/messages/base.py:412),核心内容拼接靠 merge_content(libs/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")。+ 运算符),于是框架任何地方收集流式输出都只需 final = None; for chunk in stream: final = chunk if final is None else final + chunk,不用关心 content 是文本还是多模态块。顺带一提,工单碎片也一样:ToolCallChunk(messages/tool.py:261)里 args 还是没拼完的 JSON 字符串片段,拼完才解析成 dict。convert_to_messages(messages/utils.py:786)把 ("human", "你好") 这类简写元组批量转成消息对象(D07 的 ChatPromptTemplate 就靠它);merge_message_runs(messages/utils.py:1002)把连续同类型消息合并——有些模型不允许连续两条 human 消息。串起来 + 今日小结
①
[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 # 厂商格式翻译现场
[SystemMessage(...), HumanMessage(...)] 太原始了——明天看 prompts/:PromptTemplate 填空、ChatPromptTemplate.from_messages 批量产消息、MessagesPlaceholder 给聊天历史留座位,以及 few-shot 范例注入。