核心架构 | LangChain 1.0 Python 知识文档

📚 什么是消息系统？

消息（Messages）是 LangChain 中模型与用户、工具之间交互的基本单位。 LangChain 1.0 提供了标准化的消息系统，让你能够：

统一接口：使用一致的消息格式与所有模型交互
类型安全：清晰的消息类型区分（系统、用户、AI、工具）
多模态支持：文本、图像、音频、视频、文件等多种内容
跨提供商标准化：自动将提供商特定格式转换为统一格式

💡 核心优势

通过 content_blocks 属性，LangChain 1.0 将不同提供商的响应格式（如 OpenAI 的 reasoning、Anthropic 的 thinking）统一转换为标准化的内容块，让你的代码无需修改即可切换不同模型。

🔤 四种核心消息类型

graph LR A[对话流程] --> B[SystemMessage] B --> C[HumanMessage] C --> D[AIMessage] D --> E{需要工具？} E -->|是| F[ToolMessage] F --> C E -->|否| G[完成] style B fill:#6366f1,color:#fff style C fill:#10b981,color:#fff style D fill:#f59e0b,color:#fff style F fill:#8b5cf6,color:#fff style G fill:#3b82f6,color:#fff

1. SystemMessage - 系统消息

设置模型行为的初始指令，定义 Agent 角色、响应风格和工作方式。

Python 🟢 基础

"""
SystemMessage 示例
功能：设置模型的基本行为和角色
"""
from langchain.messages import SystemMessage

# 创建系统消息
system_msg = SystemMessage("你是一个专业的 Python 编程助手。")

# 或者使用字典格式
system_msg = {"role": "system", "content": "你是一个专业的 Python 编程助手。"}

2. HumanMessage - 用户消息

代表用户输入，支持文本、图像、音频等多模态内容。

Python 🟢 基础

"""
HumanMessage 示例
"""
from langchain.messages import HumanMessage

# 纯文本消息
human_msg = HumanMessage("你好，请帮我解决一个问题。")

# 带元数据的消息
human_msg = HumanMessage(
    content="你好！",
    name="alice",      # 用户标识（可选）
    id="msg_123",      # 追踪用的唯一 ID（可选）
)

# 或使用字典格式
human_msg = {"role": "user", "content": "你好！"}

3. AIMessage - AI 消息

模型的输出对象，包含文本内容、工具调用和元数据。

Python 🟡 中级

"""
AIMessage 示例
功能：理解 AI 响应的结构
"""
from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o")

# 调用模型会返回 AIMessage 对象
response = model.invoke("什么是 LangChain？")

# AIMessage 的关键属性
print(f"文本内容: {response.content}")
print(f"纯文本: {response.text}")
print(f"工具调用: {response.tool_calls}")
print(f"Token 使用: {response.usage_metadata}")
print(f"响应元数据: {response.response_metadata}")

# 手动创建 AIMessage（用于模拟对话历史）
from langchain.messages import AIMessage

ai_msg = AIMessage("我很乐意帮助你！")
messages = [
    {"role": "user", "content": "你能帮我吗？"},
    ai_msg,  # 插入为来自模型的消息
    {"role": "user", "content": "太好了！2+2 等于多少？"}
]

response = model.invoke(messages)
print(response.content)  # 输出：2+2 等于 4。

4. ToolMessage - 工具消息

工具执行结果的消息，用于将工具输出传回模型。

Python 🔴 高级

"""
ToolMessage 完整流程示例
功能：展示工具调用 → 执行 → 返回结果的完整循环
"""
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage, ToolMessage, HumanMessage
from langchain.tools import tool

@tool
def get_weather(location: str) -> str:
    """获取天气信息"""
    return f"{location} 今天晴朗，22°C"

model = init_chat_model("gpt-4o")
model_with_tools = model.bind_tools([get_weather])

# 第 1 步：用户提问
messages = [HumanMessage("旧金山今天天气怎么样？")]

# 第 2 步：模型生成工具调用
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)

# 第 3 步：执行工具并创建 ToolMessage
for tool_call in ai_message.tool_calls:
    # 执行工具
    result = get_weather.invoke(tool_call["args"])

    # 创建 ToolMessage
    tool_message = ToolMessage(
        content=result,
        tool_call_id=tool_call["id"],  # 必须与工具调用 ID 匹配
        name="get_weather"
    )
    messages.append(tool_message)

# 第 4 步：模型整合结果生成最终答案
final_response = model_with_tools.invoke(messages)
print(final_response.content)

消息类型对比表

消息类型	作用	谁创建	典型用途
`SystemMessage`	设置模型行为	开发者	定义角色、风格、规则
`HumanMessage`	用户输入	用户/开发者	提问、指令、多模态输入
`AIMessage`	模型输出	模型	回答、工具调用
`ToolMessage`	工具执行结果	开发者/系统	返回工具调用结果

🧩 content_blocks 统一属性

content_blocks 是 LangChain 1.0 的核心创新，它将不同提供商的特定格式自动解析为标准化、类型安全的内容块。

💡 为什么需要 content_blocks？

不同 LLM 提供商使用不同的响应格式：

OpenAI：使用 reasoning 字段表示推理步骤
Anthropic：使用 thinking 字段
Google：可能使用其他格式

content_blocks 将这些提供商特定的格式统一转换为 LangChain 的标准块类型，让你的代码无需修改即可切换模型。

标准内容块类型

类别	块类型	说明
文本	`TextContentBlock`	标准文本输出
	`ReasoningContentBlock`	模型推理步骤（统一 OpenAI/Anthropic）
	`PlainTextContentBlock`	文档文本（.txt、.md）
多模态	`ImageContentBlock`	图像（URL、Base64、文件 ID）
	`AudioContentBlock`	音频数据
	`VideoContentBlock`	视频数据
	`FileContentBlock`	通用文件（PDF 等）
工具	`ToolCall`	函数调用
	`ToolCallChunk`	流式工具调用片段
	`InvalidToolCall`	格式错误的调用
服务端	`ServerToolCall`	服务端执行的工具调用
服务端	`ServerToolResult`	搜索结果

跨提供商标准化示例

Python 🟡 中级

"""
content_blocks 跨提供商标准化示例
功能：展示不同提供商的响应如何统一处理
"""
from langchain.messages import AIMessage

# Anthropic 的响应格式
anthropic_message = AIMessage(
    content=[
        {"type": "thinking", "thinking": "让我思考一下...", "signature": "xyz"},
        {"type": "text", "text": "这是我的答案"},
    ],
    response_metadata={"model_provider": "anthropic"}
)

# OpenAI 的响应格式
openai_message = AIMessage(
    content=[
        {
            "type": "reasoning",
            "id": "rs_123",
            "summary": [{"type": "summary_text", "text": "推理过程..."}],
        },
        {"type": "text", "text": "这是我的答案"},
    ],
    response_metadata={"model_provider": "openai"}
)

# 统一访问内容块（自动解析为标准格式）
for block in anthropic_message.content_blocks:
    if block["type"] == "reasoning":  # 自动标准化为 "reasoning"
        print(f"推理: {block['content']}")
    elif block["type"] == "text":
        print(f"文本: {block['text']}")

# 两种提供商的处理代码完全相同！
for block in openai_message.content_blocks:
    if block["type"] == "reasoning":
        print(f"推理: {block['content']}")
    elif block["type"] == "text":
        print(f"文本: {block['text']}")

🖼️ 多模态消息

LangChain 支持在消息中包含图像、音频、视频等多种内容类型。

图像输入

Python 🟡 中级

"""
多模态消息示例 - 图像输入
功能：让模型处理图像内容
"""
from langchain.messages import HumanMessage
from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o")  # 需要支持视觉的模型

# 方式 1：使用图像 URL
message = HumanMessage(content=[
    {"type": "text", "text": "描述这张图片的内容。"},
    {"type": "image", "url": "https://example.com/image.jpg"},
])

response = model.invoke([message])
print(response.content)

# 方式 2：使用 Base64 编码的图像
import base64

with open("local_image.jpg", "rb") as image_file:
    base64_image = base64.b64encode(image_file.read()).decode('utf-8')

message = HumanMessage(content=[
    {"type": "text", "text": "这张图片里有什么？"},
    {
        "type": "image",
        "base64": base64_image,
        "mime_type": "image/jpeg",
    },
])

response = model.invoke([message])
print(response.content)

⚠️ 重要提示

并非所有模型都支持所有文件类型。使用前请检查模型提供商的文档，了解支持的格式和大小限制。例如：

GPT-4o、Claude 3.5 Sonnet：支持图像
Gemini 2.0：支持图像、音频、视频
GPT-4o-mini：仅支持文本

🔄 消息流转流程

graph TB A[开始对话] --> B[SystemMessage:
设置角色和规则] B --> C[HumanMessage:
用户提问] C --> D[模型处理] D --> E[AIMessage:
模型响应] E --> F{包含工具调用？} F -->|是| G[执行工具函数] G --> H[ToolMessage:
工具结果] H --> I[继续对话] I --> D F -->|否| J{用户继续提问？} J -->|是| K[HumanMessage:
新问题] K --> D J -->|否| L[对话结束] style B fill:#6366f1,color:#fff style C fill:#10b981,color:#fff style E fill:#f59e0b,color:#fff style H fill:#8b5cf6,color:#fff style L fill:#3b82f6,color:#fff

完整对话示例

Python 🔴 高级

"""
完整多轮对话示例
功能：展示消息在完整对话中的流转
"""
from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage, AIMessage
from langchain.tools import tool

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 初始化模型
model = init_chat_model("gpt-4o")
model_with_tools = model.bind_tools([calculate])

# 构建对话历史
messages = [
    SystemMessage("你是一个有用的数学助手。"),
    HumanMessage("你能帮我计算吗？"),
]

# 第 1 轮：模型确认
response1 = model_with_tools.invoke(messages)
messages.append(response1)
print(f"助手: {response1.content}\n")

# 第 2 轮：用户提出计算问题
messages.append(HumanMessage("计算 (25 + 75) * 2"))
response2 = model_with_tools.invoke(messages)
messages.append(response2)

# 如果有工具调用，执行工具
if response2.tool_calls:
    for tool_call in response2.tool_calls:
        result = calculate.invoke(tool_call["args"])
        messages.append(result)

    # 第 3 轮：模型整合工具结果
    response3 = model_with_tools.invoke(messages)
    messages.append(response3)
    print(f"助手: {response3.content}\n")

# 第 4 轮：用户继续提问
messages.append(HumanMessage("如果再除以 4 是多少？"))
response4 = model_with_tools.invoke(messages)
messages.append(response4)

# 处理工具调用
if response4.tool_calls:
    for tool_call in response4.tool_calls:
        result = calculate.invoke(tool_call["args"])
        messages.append(result)

    response5 = model_with_tools.invoke(messages)
    print(f"助手: {response5.content}")

# 查看完整对话历史
print(f"\n总消息数: {len(messages)}")
for i, msg in enumerate(messages):
    print(f"{i+1}. {msg.__class__.__name__}: {msg.content[:50]}...")

🌊 流式消息处理

流式调用返回 AIMessageChunk 对象，可以逐块累积成完整消息。

Python 🟡 中级

"""
流式消息处理示例
功能：实时接收和累积消息块
"""
from langchain.chat_models import init_chat_model

model = init_chat_model("claude-sonnet-4-5-20250929")

# 流式调用
chunks = []
full_message = None

print("流式输出: ", end="")
for chunk in model.stream("用三句话解释什么是量子计算"):
    chunks.append(chunk)
    print(chunk.content, end="", flush=True)

    # 累积完整消息
    if full_message is None:
        full_message = chunk
    else:
        full_message = full_message + chunk  # AIMessageChunk 支持加法运算

print("\n")

# 查看完整消息属性
print(f"总块数: {len(chunks)}")
print(f"完整内容: {full_message.content}")
print(f"Token 使用: {full_message.usage_metadata}")

📎 带元数据的工具消息

在 RAG 等场景中，可以在 ToolMessage 中附加元数据（如文档 ID、页码），供应用逻辑访问，但不发送给模型。

Python 🔴 高级

"""
带元数据的 ToolMessage 示例
功能：在工具结果中附加元数据
"""
from langchain.messages import ToolMessage

# 发送给模型的内容
message_content = "这是检索到的文档内容：LangChain 是一个..."

# 应用可访问的元数据（不发送给模型）
artifact = {
    "document_id": "doc_12345",
    "page": 3,
    "relevance_score": 0.95,
    "source_url": "https://docs.langchain.com/..."
}

# 创建带元数据的工具消息
tool_message = ToolMessage(
    content=message_content,        # 给模型看的内容
    tool_call_id="call_123",
    name="search_documents",
    artifact=artifact               # 应用层元数据
)

# 应用可以访问元数据
print(f"文档 ID: {tool_message.artifact['document_id']}")
print(f"来源页码: {tool_message.artifact['page']}")
print(f"相关度: {tool_message.artifact['relevance_score']}")

# 但模型只看到 content 部分

✅ artifact 最佳实践

RAG 系统：存储文档来源、相关度评分
审计日志：记录工具调用的详细参数
调试信息：保存原始 API 响应
业务逻辑：附加订单 ID、用户 ID 等业务数据

❓ 常见问题

Q1: content 和 content_blocks 有什么区别？

content：原始消息内容，格式因提供商而异。
content_blocks：标准化的内容块列表，LangChain 自动解析。

content_blocks 不是 content 的替代品，而是一个新属性，用于以标准格式访问消息内容。两者都可以使用，但 content_blocks 提供更好的跨提供商兼容性。

Q2: 什么时候使用字典格式，什么时候使用消息类？

Python

# 字典格式 - 简单快捷，适合简单场景
messages = [
    {"role": "system", "content": "你是助手"},
    {"role": "user", "content": "你好"}
]

# 消息类 - 更强大，适合复杂场景
from langchain.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage("你是助手"),
    HumanMessage("你好", name="alice", id="msg_001")
]

# 建议：简单对话用字典，复杂应用用消息类

Q3: 如何处理消息历史过长的问题？

长对话会超出模型的上下文窗口限制。解决方案：

消息修剪：只保留最近 N 条消息
消息总结：定期总结历史，替换旧消息
短期记忆管理：使用 LangChain 的记忆管理功能

Python

# 简单的消息修剪
MAX_MESSAGES = 10

if len(messages) > MAX_MESSAGES:
    # 保留 SystemMessage 和最近的消息
    system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    recent_msgs = messages[-MAX_MESSAGES:]
    messages = system_msgs + recent_msgs

Q4: AIMessage 的 text 和 content 有什么区别？

text：仅提取文本内容块的文本部分（推荐使用）
content：原始内容，可能是字符串或列表

Python

response = model.invoke("你好")

# text：总是返回字符串（纯文本部分）
print(response.text)  # "你好！很高兴见到你。"

# content：可能是字符串或列表
print(response.content)  # 可能是字符串，也可能是 [{"type": "text", ...}]

# 建议：需要纯文本时使用 text 属性

Q5: 如何启用标准化输出格式？

Python

from langchain.chat_models import init_chat_model

# 启用 v1 标准化输出
model = init_chat_model("gpt-4o", output_version="v1")

# 消息内容将以标准内容块格式存储
response = model.invoke("解释 AI")
print(response.content_blocks)  # 标准化的块列表

💬 消息系统详解

📚 什么是消息系统？

🔤 四种核心消息类型

1. SystemMessage - 系统消息

2. HumanMessage - 用户消息

3. AIMessage - AI 消息

4. ToolMessage - 工具消息

消息类型对比表

🧩 content_blocks 统一属性

标准内容块类型

跨提供商标准化示例

🖼️ 多模态消息

图像输入

🔄 消息流转流程

完整对话示例

🌊 流式消息处理

📎 带元数据的工具消息

❓ 常见问题

Q1: content 和 content_blocks 有什么区别？

Q2: 什么时候使用字典格式，什么时候使用消息类？

Q3: 如何处理消息历史过长的问题？

Q4: AIMessage 的 text 和 content 有什么区别？

Q5: 如何启用标准化输出格式？

🔗 参考资源