1.6 实战

1.6.1 对话历史管理
关键规则：每次调用必须传递完整的对话历史！
也就是说：

第 1 轮：
[system, user] → AI回复 → 保存回复
第 2 轮：
[system, user, assistant, user] → AI回复 → 保存回复
第 3 轮：
[system, user, assistant, user, assistant, user] → AI回复

注意：每次对话都要在原有的消息列表中添加新消息，不可重新创建新的列表。
错误举例1❌：

# 第一次
response1 = model.invoke("我叫张三")
# 第二次（没传历史）
response2 = model.invoke("我叫什么？") # AI 不记得！

错误举例2❌：

conversation = [{"role": "user", "content": "问题1"}]
response1 = model.invoke(conversation)
conversation = [{"role": "user", "content": "问题2"}] # 重新创建！
response2 = model.invoke(conversation) # 丢失了历史

错误举例3❌：

conversation = []
conversation.append({"role": "user", "content": "问题1"})
response1 = model.invoke(conversation)
# 忘记保存 response1.content！
conversation.append({"role": "user", "content": "问题2"})
response2 = model.invoke(conversation) # AI 不知道之前的回答

正确做法✅：

conversation = []
# 第一次
conversation.append({"role": "user", "content": "我叫张三"})
response1 = model.invoke(conversation)
# 关键：保存 AI 回复
conversation.append({"role": "assistant", "content": response1.content})
# 第二次（传递完整历史）
conversation.append({"role": "user", "content": "我叫什么？"})
response2 = model.invoke(conversation) # AI 记得！

1.6.2 对话历史优化
问题：对话历史会越来越长，消耗大量 tokens 和成本。
解决方案：只保留最近 N 轮对话。具体的：

• 总是保留 system 消息（定义角色）
• 只保留最近 N 轮对话，丢弃更早的历史

举例：
定义保留最近对话轮数的函数：

def keep_recent_messages(messages, max_pairs=3):
"""
保留最近的 N 轮对话
max_pairs: 保留的对话轮数（每轮 = user + assistant）
"""
# 分离 system 和对话
system_msgs = [m for m in messages if m.get("role") == "system"]
conversation_msgs = [m for m in messages if m.get("role") != "system"]
# 只保留最近的
recent_msgs = conversation_msgs[-(max_pairs * 2):]
# 返回：system + 最近对话
return system_msgs + recent_msgs

测试：

# 初始化
long_conversation = [{"role": "system", "content": "你是 Python 导师"}
]
# 第 1 轮
long_conversation.append({"role": "user", "content": "什么是列表？用一句解释"})
r1 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r1.content})
# 第 2 轮
long_conversation.append({"role": "user", "content": "列表和元组有什么区别？用一
句解释"})
r2 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r2.content})
# 第 3 轮
long_conversation.append({"role": "user", "content": "什么是字典呢？用一句解
释"})
r3 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r3.content})
print(f"原始消息数: {len(long_conversation)}")
# 优化：只保留最近 2 轮
optimized = keep_recent_messages(long_conversation, max_pairs=2)
print(f"优化后消息数: {len(optimized)}")
print(f"保留的内容: system + 最近2轮对话")
# 添加新的用户问题
optimized.append({"role": "user", "content": "我第一个问题问的是什么？"})
# 使用优化后的历史
response = model.invoke(optimized)
print(f"\nAI 回复: {response.content}")

原始消息数: 7
优化后消息数: 5
保留的内容: system + 最近2轮对话
AI 回复: 你第一个问题问的是：**“列表和元组有什么区别？用一句解释”**

1.6.3 多轮对话聊天机器人
基于模型初始化、流式响应以及消息列表的拼接来创建多轮聊天机器人。

from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv
load_dotenv(override=True)
# 1. 基础配置
MODEL_NAME = "gpt-5.4-mini"
MAX_PAIRS_HISTORY = 10
EXIT_WORD = "quit"
# 2. 初始化模型
model = init_chat_model(
model=MODEL_NAME,
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)
# 3. 初始化消息列表
messages = [
{
"role":"system",
"content":"你是小谷姐姐，尚硅谷教育的数字员工，也是一名耐心、友好的智能助手。我
会用自然、清晰的方式回答用户问题。"
}
]
# 4. 启动提示
print(f"✨ 请输入问题，输入 {EXIT_WORD} 结束对话\n")
# 5. 多轮对话主循环
# 轮次记录
i = 1
while True:
print("\n", "=" * 10, f'-> 第 {i} 轮对话开始 <-', "=" * 10, "\n")
user_input = input("🙋 请输入：")
# 退出判断
if user_input.lower() == EXIT_WORD:
print("🌙 对话已结束，欢迎下次再来！")
break
# 追加用户消息
messages.append({"role":"user","content":user_input})
# 流式输出模型回复
print("🧚 小谷姐姐：", end="", flush=True)
reply_content = ""
# 优化历史记忆
memory_messages = keep_recent_messages(messages,max_pairs =
MAX_PAIRS_HISTORY)
# 控制发送给模型的消息长度
for chunk in model.stream(memory_messages):
if chunk.content:
print(chunk.content, end="", flush=True)
reply_content += chunk.content
print("\n", "=" * 10, f'-> 第 {i} 轮对话结束 <-', "=" * 10, "\n")
i += 1
# 追加 AI 回复
messages.append({"role":"assistant","content":reply_content})

其中，keep_recent_messages()定义，见1.6.2小节。

1.7 拓展-消息属性：content、content_blocks

1.7.1 content
消息的content 可以理解为数据内容，它是弱类型的，支持字符串和列表（列表元素通常为字典）。
举例1：存储字符串
如果只是纯文本内容，直接传递字符串就好。

from LangChain.messages import HumanMessage
msg1 = HumanMessage(content = "你好啊")
msg2 = HumanMessage("你好啊")
print(msg1)
print(msg2)

说明：当content内容只有字符串时，可以省略参数名称。
举例2：存储字典列表
如果需要发送的不只是文本，如多模态内容，则需要content的字典列表形式。
字典内容遵循模型供应商的API规范，以openai: gpt-4.1 为例。
参考官方文档：https://developers.openai.com/api/reference/python/resources/chat/subresource
s/completions/methods/create

将下图置于代码所在的目录下（比如chapter04_message_prompt），命名为image_test.png

测试代码如下

import base64
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from dotenv import load_dotenv
import os
load_dotenv(override=True)
model = init_chat_model(
model="gpt-5.4-mini",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)
def encode_image(img_path, img_type='jpeg'):
"""将一张本地图片转换成 Base64 编码的 Data URI 字符串,方便在文本中嵌入图片数据"""
with open(img_path, "rb") as img_file:
return f"data:image/{img_type};base64,
{base64.b64encode(img_file.read()).decode("utf-8")}"
# 图像路径
img_path = "image_test.png"
# 获取图像base64编码字符串
base64_image = encode_image(img_path)
response = model.invoke(
[
HumanMessage(
content=[
{'type': 'text', 'text': '这张图里有什么？'},
{
'type': 'image_url',
"image_url": base64_image,
}
]
)
]
)
print(response.content)

输出如下

图里是一瓶香水，放在浅米色/暖黄色背景上，整体风格很简洁高级。
这瓶香水是透明玻璃瓶身，金色瓶盖和装饰，瓶身里能看到淡金色液体。光线从侧面照过来，在桌面上投下了长长的阴影，营造出一种温暖、柔和的氛围。

1.7.2 content_blocks
在 LangChain 1.x 中， content_blocks 是消息对象（BaseMessage）的一项重大升级。它的核心目标是提供一种跨模型供应商、标准化的多模态数据结构。

过去，处理图片、音频、甚至是模型生成的“思维链（Reasoning）”内容时，不同供应商（OpenAI,Anthropic, Google 等）的 API 格式各异，导致开发者需要写大量的适配代码。content_blocks 的出现终结了这种混乱。
在 LangChain 1.2 版本中，消息对象的 content 属性依然存在（为了向前兼容），但新增了content_blocks 属性，可以将content 解析为标准、类型安全的表示。

• 数据结构：它是一个 list[TypedDict] 。
• 统一格式：每个 block 都有一个 type 字段，用于区分内容类型。
• 支持类型：包括 text （文本）、image （图片）、audio （音频）、video （视频）、tool_call （工具调用）以及 reasoning （推理/思维链）。

支持的字段类型详见https://docs.langchain.com/oss/python/langchain/messages#openai

① 输入格式化
对于复杂的对话（带图片或工具结果），建议使用 content_blocks 列表形式构建 HumanMessage或 AIMessage 。
借助content_blocks ，我们可以用一套标准代码，无缝地在不同厂商的模型之间切换。
举例1：OpenAI模型

from langchain.messages import HumanMessage
import os
from dotenv import load_dotenv
load_dotenv(override=True)
model = init_chat_model(
model="gpt-5.4-mini",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)
def encode_image(img_path):
"""将一张本地图片转换成 Base64 编码的 Data URI 字符串,方便在文本中嵌入图片数据"""
with open(img_path, "rb") as img_file:
return base64.b64encode(img_file.read()).decode("utf-8")
# 图像路径
img_path = "image_test.png"
# 获取图像base64编码字符串
base64_image = encode_image(img_path)
response = model.invoke(
[
# 此种格式可用
# HumanMessage(
# content=[
# {'type': 'text', 'text': '这张图里有什么？'},
# {
# 'type': 'image_url',
# "image_url": base64_image,
# }
# ]
# )
# 推荐的统一写法
HumanMessage(
content_blocks=[
{'type': 'text', 'text': '这张图里有什么？'},
{
'type': 'image',
'base64': base64_image,
'mime_type': 'image/png',
}
]
)
]
)
print(response.content)

输出

图里是一瓶香水。
它是透明玻璃瓶身、金色瓶盖和金色装饰，看起来很精致，放在暖色背景上，带有柔和的光影效果。

举例2：Anthropic模型

import base64
from langchain.messages import HumanMessage
from dotenv import load_dotenv
load_dotenv(override=True)
model = init_chat_model(
model="claude-haiku-4-5",
model_provider="openai",
api_key=os.getenv("CLOSEAI_API_KEY"),
base_url=os.getenv("CLOSEAI_BASE_URL")
)
def encode_image(img_path):
"""将一张本地图片转换成 Base64 编码的 Data URI 字符串,方便在文本中嵌入图片数据"""
with open(img_path, "rb") as img_file:
return base64.b64encode(img_file.read()).decode("utf-8")
# 图像路径
img_path = "image_test.png"
# 获取图像base64编码字符串
base64_image = encode_image(img_path)
response = model.invoke(
[
# 推荐的统一写法
HumanMessage(
content_blocks=[
{'type': 'text', 'text': '这张图里有什么？'},
{
'type': 'image',
'base64': base64_image,
'mime_type': 'image/png',
}
]
)
]
)
print(response.content)

输出

这张图里展示的是一瓶**化妆品或香水**，具体特征如下：
- **瓶子设计**：透明或半透明的玻璃瓶，里面装有浅色液体（可能是粉色、米色或淡金色）
- **瓶盖**：金色或香槟色的金属盖子，设计精致优雅
- **背景**：米色或米黄色的简洁背景
- **光影效果**：侧光照射，产生清晰的影子，凸显产品的质感和高级感
- **风格**：整体呈现出高端护肤品、粉底液、或香水的专业产品形象
这类产品通常属于**化妆品或美妆护肤品类**，包装设计显得简约而奢华。

② 输出格式化
content_blocks 还可用于输出格式化，以deepseek官网的deepseek-v4-flash 为例，其输出包含思考内容，后者位于additional_kwargs 的reasoning_content 字段下。比如：

不同的模型其输出格式可能不同，仅为提取思考内容，切换模型都可能需要更改代码，非常不方便。
content_blocks提供了统一的输出格式，可以将不同格式的响应统一为标准格式。
注意：content_blocks是懒加载的，即调用时才会解析。

from langchain.chat_models import 1 init_chat_model
from dotenv import load_dotenv
load_dotenv()
load_dotenv(override=True)
model = init_chat_model(
model="deepseek:deepseek-v4-flash",
extra_body={"thinking": {"type": "enabled"}},
)
response = model.invoke("你好，一句话回答")
print('=' * 20, '-> response <-', '=' * 20)
print(response)
print('=' * 20, '-> response.content <-', '=' * 20)
print(response.content)
print('=' * 20, '-> response.content_blocks <-', '=' * 20)
print(response.content_blocks)

输出

==================== -> response <- ====================
content='你好，请说出您的问题，我会用一句话回答。' additional_kwargs=
{'refusal': None, 'reasoning_content': '好的，用户说“一句话回答”，那说明他希望
我回答得简洁直接。没有具体问题，可能是测试或者玩笑。我需要确保回应既符合“一句话”的要
求，又有礼貌。可以用“你好”开头，确认收到指令，然后表明已经准备好回答任何问题。这样既简
洁又完整。'} response_metadata={'token_usage': {'completion_tokens': 76,
'prompt_tokens': 8, 'total_tokens': 84, 'completion_tokens_details':
{'accepted_prediction_tokens': None, 'audio_tokens': None,
'reasoning_tokens': 63, 'rejected_prediction_tokens': None},
'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0},
'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 8},
'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash',
'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402',
'id': 'a678ed24-3aff-429f-b95d-6877eeb21efd', 'finish_reason': 'stop',
'logprobs': None} id='lc_run--019e4b28-6b9b-7f11-bfe6-84aca4ecc4dc-0'
tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 8,
'output_tokens': 76, 'total_tokens': 84, 'input_token_details':
{'cache_read': 0}, 'output_token_details': {'reasoning': 63}}
==================== -> response.content <- ====================
你好，请说出您的问题，我会用一句话回答。
==================== -> response.content_blocks <- ====================
[{'type': 'reasoning', 'reasoning': '好的，用户说“一句话回答”，那说明他希望我回
答得简洁直接。没有具体问题，可能是测试或者玩笑。我需要确保回应既符合“一句话”的要求，又
有礼貌。可以用“你好”开头，确认收到指令，然后表明已经准备好回答任何问题。这样既简洁又完
整。'}, {'type': 'text', 'text': '你好，请说出您的问题，我会用一句话回答。'}]

说明：优先检查 response.content_blocks 而不是 response.content ，特别是当你需要获取“思维
链”或者“引用（Citations）”信息时。