第五章：核心组件详解 - Messages

在现代 LangChain 体系中，消息（Messages）不只是聊天记录，而是驱动模型、工具、状态和中间件的核心数据结构。很多初学者会把一切输入都写成一个超长字符串，这在简单任务中还能工作，但一旦系统开始出现多轮对话、工具调用、检索上下文、流式输出和状态恢复，字符串拼接就会迅速失控。消息模型的价值在于：把上下文拆成有角色、有顺序、有来源的结构化单元。

本章会从消息的概念、消息类型、模板化组织、历史裁剪、工具消息、流式消息、多模态消息到工程实践，系统说明为什么现代 LangChain 项目几乎都应该围绕 Messages 来建模。

传统 LLM 的输入形式往往像这样：

prompt = "你是一个助手，请总结下面这篇文章，并给出 3 条建议：..."
result = llm.invoke(prompt)

这种方式的问题是：

• 系统规则、用户问题、历史信息、检索结果全部混在一起；
• 不方便增量更新；
• 不能自然表示多轮对话；
• 工具调用返回值很难插入；
• 很难做历史裁剪、日志落盘和安全审计。

现代聊天模型更推荐这样组织输入：

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
messages = [
    SystemMessage(content="你是资深项目顾问，回答要专业、谨慎、结构化。"),
    HumanMessage(content="请总结这份项目周报的风险点，并给出缓解建议。")
]
 
response = model.invoke(messages)
print(response.content)

消息列表的好处在于：

• System 与 Human 的职责天然分层；
• AI 的历史回答可以显式回放；
• 工具结果可以作为 ToolMessage 插入上下文；
• 历史可以按条目裁剪，而不是暴力截断整个字符串；
• 更贴近 OpenAI、Anthropic 等主流模型的原生交互形式。

一个真实的企业助手在回答前，往往要装配很多上下文：

• 系统角色说明；
• 用户本轮问题；
• 最近若干轮历史；
• 用户档案摘要；
• 检索到的制度条款；
• 上一步工具返回；
• 输出格式要求；
• 风险控制指令。

如果没有消息抽象，这些内容会混成一个难以维护的超级 prompt；如果用消息来表达，则每一段信息都可以被单独查看、替换、裁剪和审计。

根据 LangChain 官方文档，消息对象通常由三部分组成：

• Role：消息角色，例如 `system`、`user`、`assistant`；
• Content：消息内容，可以是文本、图片、文件信息等；
• Metadata：附加信息，例如消息 ID、token 用量、响应元数据。

SystemMessage 用于定义模型的高层行为边界，例如：

• 你是谁；
• 你应该优先遵守哪些规则；
• 回答应该采用什么风格；
• 什么时候必须保守回答；
• 什么时候应拒绝执行高风险动作。

from langchain_core.messages import SystemMessage
 
system_msg = SystemMessage(
    content="""
你是企业法务助手。
1. 只能基于提供的合同条款与制度解释。
2. 若证据不足，必须明确说明“不确定”。
3. 不要生成最终法律意见书，只给出审阅建议。
"""
)

SystemMessage 的设计原则：

• 放稳定规则，不放本轮临时变量；
• 长度适中，避免把几十条相互冲突的规则塞进去；
• 优先写清楚“必须遵守”的边界，而不是泛泛描述角色；
• 尽量使用项目级统一模板，避免不同模块各写一套系统提示。

HumanMessage 表示用户或外部调用方输入的本轮请求。它通常承载：

• 用户问题；
• 临时背景；
• 业务参数；
• 本轮输出要求。

from langchain_core.messages import HumanMessage
 
human_msg = HumanMessage(
    content="请根据以下会议纪要，提取待办事项，并按优先级排序。"
)

一个常见经验是：

• 长期稳定规则放在 SystemMessage；
• 本轮任务信息放在 HumanMessage；
• 不要把项目级规则和用户级输入反复混写。

AIMessage 表示模型历史输出。它不仅能承载普通文本，也能承载：

• 工具调用信息；
• provider 返回的 reasoning / thinking 内容；
• usage metadata；
• 完整消息 ID。

response = model.invoke(messages)
print(type(response))
print(response.content)
print(response.usage_metadata)

在工程上，AIMessage 有两个重要用途：

• 作为多轮对话历史继续传给模型；
• 作为日志和评估的关键原始记录。

当模型发起工具调用后，工具执行结果通常通过 ToolMessage 回传给模型。它是 Agent 与外部世界交互的关键桥梁。

from langchain_core.messages import AIMessage, ToolMessage
 
ai_message = AIMessage(
    content="",
    tool_calls=[{
        "name": "get_weather",
        "args": {"city": "上海"},
        "id": "call_001",
        "type": "tool_call"
    }]
)
 
tool_message = ToolMessage(
    content="上海今日多云，26 摄氏度，空气湿度 72%。",
    tool_call_id="call_001",
    name="get_weather"
)

ToolMessage 的关键要求：

• `tool_call_id` 必须与 AIMessage 中的工具调用 ID 对上；
• 内容必须清晰、准确、尽量可推理；
• 若工具原始返回很长，应先清洗，再回传给模型；
• 原始数据可放在 `artifact` 或其他日志字段中，不一定要全量送给模型。

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
messages = [
    SystemMessage(content="你是一名面向初学者的 Python 讲师。"),
    HumanMessage(content="请解释什么是装饰器，并给一个最简单示例。")
]
 
response = model.invoke(messages)
print(response.content)

LangChain 也支持直接用字典形式表示消息，便于与前端或第三方接口打通。

result = model.invoke([
    {"role": "system", "content": "你是简历优化助手。"},
    {"role": "user", "content": "帮我把这段项目经历改得更专业。"}
])
print(result.content)

这种格式适合：

• 和 Web 前端交互时复用 OpenAI 风格的数据结构；
• 需要把消息存进 JSON 数据库；
• 做接口回放和调试。

如果直接手写消息列表，随着变量变多会越来越难维护。所以 LangChain 常配合 `ChatPromptTemplate` 来生成消息。

from langchain_core.prompts import ChatPromptTemplate
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是公司的培训讲师，擅长把复杂概念讲简单。"),
    ("human", "请用类比法解释 {topic}，并给出一个工作场景案例。")
])
 
messages = prompt.invoke({"topic": "工具调用"})
for msg in messages.messages:
    print(type(msg).__name__, msg.content)

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
chain = prompt | model
result = chain.invoke({"topic": "RAG"})
print(result.content)

这种写法的价值在于：

• 模板与模型解耦；
• 变量替换逻辑清晰；
• 后续容易接 `StrOutputParser`、结构化输出、日志中间件等组件。

对话系统中最常见的需求就是把已有消息历史插入模板。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是企业知识助手。资料不足时必须明确说明。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}")
])
 
history = [
    HumanMessage(content="差旅报销多久到账？"),
    AIMessage(content="通常在审批完成后的 3 个工作日内到账。")
]
 
messages = prompt.invoke({
    "history": history,
    "question": "那发票抬头写错了怎么办？"
})

这样做的好处是：

• 历史管理逻辑独立于模板；
• 历史来自数据库、缓存还是摘要模块都无所谓；
• 便于做 token 预算控制。

保留所有历史会带来：

• token 成本膨胀；
• 早期无关内容污染当前问题；
• 延迟上升；
• 隐私与合规风险加大。

因此，一个好的会话系统通常会把历史拆成三层：

• 近期原文层：最近几轮消息；
• 摘要层：更早历史的压缩摘要；
• 档案层：用户稳定资料，如部门、角色、偏好。

from langchain_core.messages import SystemMessage
 
def trim_messages(messages, keep_last=6):
    system_messages = [m for m in messages if isinstance(m, SystemMessage)]
    others = [m for m in messages if not isinstance(m, SystemMessage)]
    return system_messages[:1] + others[-keep_last:]

这个版本很简单，但已经体现出一个重要原则：

• SystemMessage 往往应该保留；
• 其他消息按窗口裁剪；
• 真实项目还应结合 token 数量而不是只按条数。

在企业应用中，更推荐“按 token 预算”而不是“按条数”裁剪：

def pack_context(system_msg, summary, recent_messages, budget_tokens=6000):
    """示意函数：按预算拼接上下文。"""
    packed = [system_msg]
    if summary:
        packed.append(summary)
 
    current_tokens = estimate_tokens(packed)
    for msg in reversed(recent_messages):
        msg_tokens = estimate_tokens([msg])
        if current_tokens + msg_tokens > budget_tokens:
            break
        packed.insert(1, msg)
        current_tokens += msg_tokens
    return packed

这里的 `estimate_tokens()` 可以由你使用模型官方 tokenizer 或统一估算逻辑实现。

当你把工具绑定到模型后，模型在需要时会返回带 `tool_calls` 的 AIMessage，而不是直接文本答案。

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """查询指定城市天气。"""
    return f"{city} 今日晴，27 摄氏度。"
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("北京今天的天气怎么样？")
 
print(response.tool_calls)

你会看到类似这样的工具调用结构：

• 工具名；
• 参数；
• tool call id；
• 类型信息。

理解 ToolMessage 的最佳方式，是手动跑一遍完整闭环。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """查询指定城市天气。"""
    return f"{city} 今日小雨，建议带伞。"
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([get_weather])
messages = [HumanMessage(content="杭州今天适合骑车上班吗？请先查天气。")]
 
ai_msg = model.invoke(messages)
messages.append(ai_msg)
 
for tool_call in ai_msg.tool_calls:
    if tool_call["name"] == "get_weather":
        result = get_weather.invoke(tool_call["args"])
        messages.append(ToolMessage(
            content=result,
            tool_call_id=tool_call["id"],
            name=tool_call["name"]
        ))
 
final_answer = model.invoke(messages)
print(final_answer.content)

这段代码能帮助你理解：

• AIMessage 可能不是最终答案，而是“下一步动作”；
• ToolMessage 的本质是“把真实世界结果喂回模型”；
• Agent 框架只是在替你管理这一套循环。

除了 `content`，消息对象还可能包含很多工程上非常重要的信息：

• `id`：消息唯一标识；
• `usage_metadata`：token 计数；
• `response_metadata`：模型名、finish_reason、provider 元数据；
• `name`：可选角色名；
• `artifact`：一些不直接发给模型的补充信息。

response = model.invoke([HumanMessage(content="请用一句话解释什么是消息对象")])
print("id:", response.id)
print("usage:", response.usage_metadata)
print("response_metadata:", response.response_metadata)

这些字段很适合用于：

• 成本分析；
• trace 追踪；
• 请求回放；
• 异常定位；
• A/B 测试记录。

LangChain 官方消息体系支持文本、图片、文件等更复杂的内容块。即便你的项目当前只用文本，也建议建立一个认知：消息不一定只是字符串。

from langchain_core.messages import HumanMessage
 
message = HumanMessage(content=[
    {"type": "text", "text": "请识别这张发票中的金额和日期。"},
    {"type": "image_url", "image_url": {"url": "https://example.com/invoice.png"}}
])

• 原图尽量不要反复塞入长对话历史；
• 应把 OCR、结构化字段和原图引用区分开；
• 原始附件建议走对象存储，消息中只保留必要内容；
• 对隐私图片应做脱敏和权限控制。

在流式输出时，你收到的往往不是完整 AIMessage，而是一个个 `AIMessageChunk`。

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
full = None
for chunk in model.stream("用三句话解释为什么要做消息分层"):
    print(chunk.text, end="")
    full = chunk if full is None else full + chunk
 
print("
---")
print(full.content)

这意味着：

• 前端可以边接收边渲染；
• 后端可以边流式输出边记录日志；
• 如果你想在最后得到完整消息，需要把 chunk 聚合起来。

下面给出一个小型示例：它支持系统规则、历史消息插入和本轮问答。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
 
faq_prompt = ChatPromptTemplate.from_messages([
    ("system", """
你是公司内部 FAQ 助手。
1. 只回答行政制度、报销流程、IT 支持类问题。
2. 若历史中已经给出答案，应保持前后一致。
3. 不确定时请明确说明，并建议用户联系对应部门。
"""),
    MessagesPlaceholder("history"),
    ("human", "{question}")
])
 
history = [
    HumanMessage(content="电脑密码过期后怎么办？"),
    AIMessage(content="请通过统一身份平台重置密码，若失败联系 IT 服务台。")
]
 
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = faq_prompt | model
 
result = chain.invoke({
    "history": history,
    "question": "如果统一身份平台打不开呢？"
})
 
print(result.content)

这个示例虽然简单，但已经体现了消息体系的 4 个优点：

• 历史和模板解耦；
• 不同角色边界清晰；
• 可以继续加入检索消息或工具消息；
• 便于扩展成更完整的客服或办公助手。

• 把所有内容都塞进一个 HumanMessage；
• 把系统规则反复写在每一轮用户消息中；
• 工具结果未经清洗直接全量回灌；
• 历史越来越长却没有裁剪；
• 没有记录最终发给模型的完整消息序列；
• SystemMessage 与 HumanMessage 要求互相矛盾。

当回答质量不稳定时，优先排查消息层：

• 最终发给模型的消息列表是什么？
• System 规则是否清晰、是否过长？
• 有没有重复注入历史？
• ToolMessage 是否过长、过脏、过多？
• 关键历史是否在裁剪时丢了？
• 是否存在提示注入风险内容被当作普通消息放入？

• 消息是现代 LangChain 应用的基础上下文单元；
• System、Human、AI、Tool 四类消息各司其职；
• `ChatPromptTemplate` 与 `MessagesPlaceholder` 是组织消息的高频工具；
• 历史管理不是无脑保留，而是裁剪、摘要和分层；
• ToolMessage 是 Agent 与外部世界交互的关键桥梁；
• 流式输出、多模态输入、用量统计都建立在消息对象之上。

1. 使用 `ChatPromptTemplate` 和 `MessagesPlaceholder` 实现一个支持多轮追问的 FAQ 助手。

2. 编写一个消息裁剪器，要求始终保留第一条 SystemMessage 和最近 8 条消息。

3. 手动模拟一次工具调用闭环：创建 AIMessage、执行工具、构造 ToolMessage，再继续调用模型。

4. 设计一个多模态报销助手的消息结构，说明哪些信息应放在图片内容、OCR 文本、结构化字段和 ToolMessage 中。

5. 为你的项目设计一份“消息日志格式”，至少包含 message_id、role、token、trace_id 和 created_at 字段。

• LangChain 官方文档：Messages
• LangChain 官方文档：Short-term memory
• LangChain 官方文档：Streaming