多轮对话的消息历史管理

LLM API 是无状态的（即每次调用都是全新的，服务器不保存上一次的对话内容）。每次调用都是独立的，模型不记得上一次你问了什么。要实现多轮对话，你需要自己管理消息历史，把完整的对话记录发给模型。

这是理解 LLM 应用架构的关键认知。

常见错误：新手最常犯的错误是认为"对话历史是自动保存的"。当用户说"你还记得我刚才说的吗"，如果你没有传入历史消息，模型真的不知道。

1.1 无状态 API 的含义

python

# 这样做不能实现多轮对话
client.chat.completions.create(messages=[{"role": "user", "content": "我叫张三"}])
client.chat.completions.create(messages=[{"role": "user", "content": "你知道我叫什么吗？"}])
# 第二次调用，模型不知道第一次的内容，会回答"我不知道"

要让模型"记住"之前的对话，必须把历史消息一起发过去：

python

messages = [
    {"role": "user", "content": "我叫张三"},
    {"role": "assistant", "content": "你好，张三！有什么我可以帮你的吗？"},
    {"role": "user", "content": "你知道我叫什么吗？"}
]

response = client.chat.completions.create(
    model="deepseek-chat",
    max_tokens=1024,
    messages=messages
)
# 现在模型会回答"你叫张三"

多轮对话消息历史结构图
多轮对话的 messages 数组结构及上下文窗口限制示意

1.2 消息格式规则

messages 列表有几个规则：

必须以 user 消息开头（不能以 assistant 开头）
role 必须交替出现：user → assistant → user → assistant...
不能有连续的同一 role

python

# 正确的消息格式
messages = [
    {"role": "user", "content": "第一个问题"},
    {"role": "assistant", "content": "第一个回答"},
    {"role": "user", "content": "第二个问题"},
    {"role": "assistant", "content": "第二个回答"},
    {"role": "user", "content": "第三个问题"},
]

# 错误：以 assistant 开头
messages = [
    {"role": "assistant", "content": "..."},  # ❌
]

# 错误：连续的 user 消息
messages = [
    {"role": "user", "content": "第一条"},
    {"role": "user", "content": "第二条"},  # ❌
]

1.3 实现一个简单的对话循环

python

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.deepseek.com",
    api_key=os.environ.get("DEEPSEEK_API_KEY")
)

def chat():
    """简单的命令行对话程序"""
    messages = [
        {"role": "system", "content": "你是一个友好的 AI 助手，用简洁的中文回答问题。"}
    ]

    print("开始对话（输入 'quit' 退出）\n")

    while True:
        user_input = input("你: ").strip()
        if user_input.lower() == "quit":
            break
        if not user_input:
            continue

        # 添加用户消息到历史
        messages.append({"role": "user", "content": user_input})

        # 调用 API
        response = client.chat.completions.create(
            model="deepseek-chat",
            max_tokens=1024,
            messages=messages
        )

        assistant_message = response.choices[0].message.content

        # 把助手回复也加入历史
        messages.append({"role": "assistant", "content": assistant_message})

        print(f"\nAI: {assistant_message}\n")

chat()

关键点：每次调用后，把助手的回复也追加到 messages 列表。这样下次调用时，模型能看到完整的对话历史。

1.3.1 生产环境警告：消息历史无限增长

上面的代码在演示中没问题，但在生产环境有一个隐患：消息历史会无限增长。

每次对话都追加消息，长对话会导致：

Token 消耗线性增长（成本失控）
超过模型上下文窗口限制（报错）
响应速度越来越慢

生产环境的三种解决方案：

方案1：滑动窗口（最简单）

python

MAX_HISTORY = 20  # 保留最近20条消息

def trim_messages(messages: list, max_count: int = MAX_HISTORY) -> list:
    """保留最近 N 条消息（保持user/assistant配对完整）"""
    # 保留 system 消息
    system_msgs = [m for m in messages if m.get("role") == "system"]
    other_msgs = [m for m in messages if m.get("role") != "system"]

    if len(other_msgs) <= max_count:
        return messages
    # 保留最近的消息，确保从user消息开始
    trimmed = other_msgs[-max_count:]
    if trimmed and trimmed[0].get("role") == "assistant":
        trimmed = trimmed[1:]
    return system_msgs + trimmed

# 使用
messages = trim_messages(messages)

方案2：Token 预算控制（更精确）

python

import tiktoken  # pip install tiktoken

def count_tokens(messages: list, model: str = "gpt-4o") -> int:
    enc = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += len(enc.encode(msg.get("content", ""))) + 4  # 每条消息的固定开销
    return total

MAX_TOKENS = 8000  # 预留足够的输出空间

def trim_by_tokens(messages: list) -> list:
    while count_tokens(messages) > MAX_TOKENS and len(messages) > 2:
        # 删除最旧的非system消息
        for i, msg in enumerate(messages):
            if msg["role"] != "system":
                messages.pop(i)
                break
    return messages

方案3：摘要压缩（保留更多语义信息）

当历史太长时，用 LLM 把旧对话压缩成摘要，再继续对话：

python

async def summarize_history(messages: list, client) -> str:
    """把历史对话压缩成摘要"""
    history_text = "\n".join([
        f"{m['role']}: {m['content']}" for m in messages
        if m.get("role") != "system"
    ])
    response = await client.chat.completions.create(
        model="deepseek-chat",  # 用同一模型做摘要
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"请用200字总结以下对话的关键信息：\n\n{history_text}"
        }]
    )
    return response.choices[0].message.content

# 当消息超过阈值时触发摘要
if len(messages) > 30:
    summary = await summarize_history(messages[:-10], client)
    system_msgs = [m for m in messages if m.get("role") == "system"]
    messages = system_msgs + [
        {"role": "user", "content": f"[之前对话摘要]\n{summary}"},
        {"role": "assistant", "content": "好的，我已了解之前的对话内容。"},
    ] + messages[-10:]  # 保留最近10条

选择建议：

简单聊天机器人 → 方案1（滑动窗口）
精确成本控制 → 方案2（Token预算）
需要保留长期上下文 → 方案3（摘要压缩）

1.4 消息历史的 Token 消耗

多轮对话的 Token 消耗会随轮次增加：

code

第1轮：输入 = system + 用户消息1
第2轮：输入 = system + 用户消息1 + 助手回复1 + 用户消息2
第3轮：输入 = system + 用户消息1 + 助手回复1 + 用户消息2 + 助手回复2 + 用户消息3

对话越长，每次调用的输入 Token 越多，费用越高。这是多轮对话应用的核心成本问题。

python

# 追踪累计 Token 消耗
total_input_tokens = 0
total_output_tokens = 0

response = client.chat.completions.create(...)
total_input_tokens += response.usage.prompt_tokens
total_output_tokens += response.usage.completion_tokens

print(f"累计消耗：输入 {total_input_tokens} tokens，输出 {total_output_tokens} tokens")

1.5 对话历史截断策略

当对话很长时，需要截断历史来控制 Token 消耗。常见策略：

策略一：固定保留最近 N 轮

python

def trim_messages(messages: list, max_turns: int = 10) -> list:
    """保留最近 max_turns 轮对话"""
    system_msgs = [m for m in messages if m.get("role") == "system"]
    other_msgs = [m for m in messages if m.get("role") != "system"]
    # 每轮 = 1条user + 1条assistant = 2条消息
    max_messages = max_turns * 2
    if len(other_msgs) > max_messages:
        other_msgs = other_msgs[-max_messages:]
    return system_msgs + other_msgs

策略二：按 Token 数截断

python

def trim_by_tokens(messages: list, max_tokens: int = 4000) -> list:
    """粗略估算，保留不超过 max_tokens 的消息"""
    # 粗略估算：1个中文字符 ≈ 1.5 tokens，1个英文单词 ≈ 1.3 tokens
    def estimate_tokens(text: str) -> int:
        return len(text) * 2  # 粗略估算，偏保守

    total = 0
    result = []
    for msg in reversed(messages):
        tokens = estimate_tokens(str(msg["content"]))
        if total + tokens > max_tokens:
            break
        result.insert(0, msg)
        total += tokens

    return result

策略三：摘要压缩（适合长对话）

把早期对话压缩成摘要，保留完整的近期对话：

python

def summarize_old_messages(messages: list, keep_recent: int = 6) -> list:
    """把旧消息压缩成摘要，保留最近 keep_recent 条"""
    system_msgs = [m for m in messages if m.get("role") == "system"]
    other_msgs = [m for m in messages if m.get("role") != "system"]

    if len(other_msgs) <= keep_recent:
        return messages

    old_messages = other_msgs[:-keep_recent]
    recent_messages = other_msgs[-keep_recent:]

    # 用 LLM 生成摘要
    history_text = "\n".join([
        f"{m['role']}: {m['content']}" for m in old_messages
    ])
    summary_response = client.chat.completions.create(
        model="deepseek-chat",
        max_tokens=512,
        messages=[
            {
                "role": "user",
                "content": f"请用简短的中文总结以下对话的关键信息：\n\n{history_text}"
            }
        ]
    )
    summary = summary_response.choices[0].message.content

    # 把摘要作为第一条用户消息
    return system_msgs + [
        {"role": "user", "content": f"[之前对话摘要]: {summary}"},
        {"role": "assistant", "content": "好的，我已了解之前的对话背景。"},
        *recent_messages
    ]

1.6 content 字段的多种格式

content 不只是字符串，还可以是列表（用于多模态输入，即同时包含文字和图片等不同类型内容）：

python

# 文本内容（简写）
{"role": "user", "content": "你好"}

# 文本内容（完整格式）
{"role": "user", "content": [{"type": "text", "text": "你好"}]}

# 图片 + 文本（多模态，DeepSeek-VL 等多模态模型支持）
{"role": "user", "content": [
    {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}},
    {"type": "text", "text": "这张图片里有什么？"}
]}

对于纯文本对话，用字符串格式就够了。多模态输入会在 Agent 相关章节涉及。

1.7 完整示例：带历史管理的对话类

python

import os
from openai import OpenAI
from dataclasses import dataclass, field

@dataclass
class Conversation:
    """带历史管理的对话类"""
    system: str = ""
    max_turns: int = 20
    model: str = "deepseek-chat"
    messages: list = field(default_factory=list)

    def __post_init__(self):
        self._client = OpenAI(
            base_url="https://api.deepseek.com",
            api_key=os.environ.get("DEEPSEEK_API_KEY")
        )
        # 如果有 system 提示，作为第一条消息
        if self.system:
            self.messages = [{"role": "system", "content": self.system}]

    def chat(self, user_message: str) -> str:
        self.messages.append({"role": "user", "content": user_message})

        # 截断历史（保留 system 消息）
        system_msgs = [m for m in self.messages if m.get("role") == "system"]
        other_msgs = [m for m in self.messages if m.get("role") != "system"]
        if len(other_msgs) > self.max_turns * 2:
            other_msgs = other_msgs[-(self.max_turns * 2):]
            self.messages = system_msgs + other_msgs

        response = self._client.chat.completions.create(
            model=self.model,
            max_tokens=1024,
            messages=self.messages
        )
        assistant_reply = response.choices[0].message.content

        self.messages.append({"role": "assistant", "content": assistant_reply})
        return assistant_reply

    def reset(self):
        if self.system:
            self.messages = [{"role": "system", "content": self.system}]
        else:
            self.messages = []


# 使用
conv = Conversation(system="你是一个 Python 专家，用简洁的中文回答。")
print(conv.chat("Python 的 GIL（Global Interpreter Lock，全局解释器锁，限制同一时刻只能有一个线程执行 Python 代码的机制）是什么？"))
print(conv.chat("它对多线程有什么影响？"))
print(conv.chat("有什么解决方案？"))

掌握消息历史管理，是构建任何多轮对话应用的基础。下一篇讲流式输出——让模型的回答逐字出现，而不是等待完整响应。

1.8 常见错误和解决方法

错误现象	原因	解决方法
模型"忘记"之前说的话	没有传入历史消息	把历史消息一起传入 messages 列表
对话越来越慢，成本越来越高	messages 无限增长	使用滑动窗口或摘要压缩
API 返回 400 错误	消息格式不对（如连续 user 消息）	确保 user/assistant 交替出现
超出上下文窗口报错	messages 太长	使用 token 预算控制或摘要压缩

1.9 小结

策略	适合场景	信息保留度	实现难度
滑动窗口	简单聊天，最近内容为主	低（丢失早期信息）	简单
Token 预算控制	需要精确成本控制	中	中等
摘要压缩	需要保留长期上下文	高	较复杂