AI 应用开发实战（3）：多轮对话进阶 —— 清除历史、注入 System Prompt、上下文截断策略

前言

前两篇我们搭建了 AI 应用的基础骨架和 Prompt 工程体系。这一篇解决一个实际生产中最常见的问题：多轮对话的上下文管理。

用户和 AI 聊天时，对话一直在累积。不管理，Token 会超限、费用会飙升、模型会"忘记"前面的内容。但管理方式不对，对话体验又会变差。

本篇覆盖三个核心问题：
1. 上下文截断——对话太长时怎么裁剪
2. System Prompt 注入——如何动态控制模型行为
3. 记忆机制——跨对话记住用户信息

1. 对话上下文管理基础

1.1 问题定义

每次对话，我们向 LLM 发送的 messages 结构如下：

messages = [
    {"role": "system", "content": "你是一个 AI 助手。"},
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好！有什么可以帮助你的？"},
    {"role": "user", "content": "帮我写一个 Python 函数"},
    {"role": "assistant", "content": "好的，以下是一个示例函数..."},
    # ... 不断累加
]

问题：当对话进行到第 50 轮时，messages 长度可能超过 10 万 Token，远超模型上下文窗口。必须截断。

1.2 三种基础截断策略

策略 1：固定窗口（Fixed Window）

只保留最近 N 轮对话，之前的全部丢弃。

def truncate_fixed_window(messages: list, max_rounds: int = 10) -> list:
    # 保留最近的 max_rounds 轮对话（每轮 = user + assistant 两条）
    system_msgs = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]
    keep = history[-(max_rounds * 2):]
    return system_msgs + keep

优点：实现简单、速度快
缺点：不考虑 Token 数，长消息的轮次会占更多空间；固定轮次可能浪费窗口或溢出

策略 2：Token 窗口（Token Window）

以 Token 数量为截断标准，精确控制上下文窗口。

import tiktoken

def truncate_token_window(
    messages: list,
    max_tokens: int = 4000,
    model: str = "gpt-4",
    reserve_for_output: int = 1000,
) -> list:
    # 以 Token 数为标准截断对话历史。
    # 策略：从后往前保留，直到接近 max_tokens。
    # 保留 system prompt，去掉最早的历史。
    try:
        enc = tiktoken.encoding_for_model(model)
    except Exception:
        enc = tiktoken.get_encoding("cl100k_base")

    def count_tokens(text: str) -> int:
        return len(enc.encode(text))

    system_msgs = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]

    system_tokens = sum(count_tokens(m["content"]) for m in system_msgs)
    available = max_tokens - reserve_for_output - system_tokens

    kept = []
    total = 0
    for m in reversed(history):
        tokens = count_tokens(m["content"])
        if total + tokens > available:
            break
        total += tokens
        kept.append(m)

    kept.reverse()
    return system_msgs + kept

优点：精确控制 Token 使用量、不会溢出
缺点：每次都要计算 Token 数，有性能开销；可能截断在对话中间，导致上下文不连贯

策略 3：滑动窗口（Sliding Window）

基于 Token 窗口，但保证窗口内的对话是完整的整轮。

def truncate_sliding_window(
    messages: list,
    max_tokens: int = 4000,
    model: str = "gpt-4",
    reserve_for_output: int = 1000,
) -> list:
    # 滑动窗口截断：以完整的对话轮次为单位，从后往前保留。
    # 每轮 = 一条 user msg + 一条 assistant msg（可能有多条）
    try:
        enc = tiktoken.encoding_for_model(model)
    except Exception:
        enc = tiktoken.get_encoding("cl100k_base")

    def count_tokens(text: str) -> int:
        return len(enc.encode(text))

    system_msgs = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]
    system_tokens = sum(count_tokens(m["content"]) for m in system_msgs)
    available = max_tokens - reserve_for_output - system_tokens

    # 按轮次分组
    rounds = []
    current_round = []
    for m in history:
        current_round.append(m)
        if m["role"] == "assistant":
            rounds.append(current_round)
            current_round = []

    if current_round:
        rounds.append(current_round)

    # 从后往前保留完整轮次
    kept = []
    total = 0
    for rnd in reversed(rounds):
        rnd_tokens = sum(count_tokens(m["content"]) for m in rnd)
        if total + rnd_tokens > available:
            break
        total += rnd_tokens
        kept = rnd + kept

    return system_msgs + kept

三种策略对比：

策略	控制粒度	上下文连贯性	实现复杂度	适用场景
固定窗口	轮次	好（完整轮次）	低	对话稳定、每轮长度均匀
Token 窗口	Token	差（可能截断到一半）	中	Token 敏感场景
滑动窗口	轮次+Token	好	中	✅ 生产推荐

1.3 在系列第一篇的应用中集成

在上篇的 ai-chat-app 中，原来的 trim_history 用的是简单的 Token 窗口。现在升级为滑动窗口：

def trim_history(messages: list, max_tokens: int = 4000) -> list:
    # 滑动窗口截断：保留完整轮次，精确控制 Token
    system_msgs = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]

    system_tokens = sum(len(m["content"]) / 1.5 for m in system_msgs)
    available = max_tokens - system_tokens

    # 按轮次分组
    rounds = []
    current = []
    for m in history:
        current.append(m)
        if m["role"] == "assistant":
            rounds.append(current)
            current = []
    if current:
        rounds.append(current)

    # 从后往前保留
    kept = []
    total = 0
    for rnd in reversed(rounds):
        tokens = sum(len(m["content"]) / 1.5 for m in rnd)
        if total + tokens > available:
            break
        total += tokens
        kept = rnd + kept

    return system_msgs + kept

2. System Prompt 动态注入

2.1 为什么要动态注入？

System Prompt 不应是静态的。不同场景下你需要：

用户登录后注入用户信息（名字、偏好）
对话中切换模式（翻译模式→编程模式）
注入当前时间、天气、上下文信息
A/B 测试不同的 System Prompt 版本

2.2 动态注入实现

class SystemPromptManager:
    # System Prompt 动态注入管理器

    def __init__(self):
        self._templates = {}
        self._dynamic_vars = {}

    def register(self, name: str, template: str):
        # 注册 System Prompt 模板，支持 {variable} 占位符
        self._templates[name] = template

    def set_variable(self, key: str, value: str):
        # 设置动态变量
        self._dynamic_vars[key] = value

    def build(self, name: str = "default", extra_vars: dict = None) -> str:
        # 构建最终的 System Prompt
        template = self._templates.get(name, self._templates.get("default", ""))
        vars_dict = dict(self._dynamic_vars)
        if extra_vars:
            vars_dict.update(extra_vars)
        return template.format(**vars_dict)


# 使用示例
sp_manager = SystemPromptManager()

# 注册默认模板
sp_manager.register("default", "你是一个 AI 助手。")

# 注册带用户信息的模板
sp_manager.register("with_user", (
    "你是一个 AI 助手。\n"
    "当前用户信息：\n"
    "- 姓名：{user_name}\n"
    "- 角色：{user_role}\n"
    "- 偏好语言：{preferred_language}\n"
    "回答要求：\n"
    "- 用 {preferred_language} 回答\n"
    "- 根据用户角色调整回答的深度\n"
    "- 称呼用户的名字"
))

# 注册翻译模式
sp_manager.register("translator", (
    "你是一个专业翻译。\n"
    "当前模式：{source_lang} -> {target_lang}\n"
    "领域：{domain}\n"
    "要求：\n"
    "- 术语准确\n"
    "- 保持原文格式\n"
    "- 不要添加原文没有的内容"
))

# 在请求中动态注入
sp_manager.set_variable("user_name", "张三")
sp_manager.set_variable("user_role", "后端开发工程师")
sp_manager.set_variable("preferred_language", "中文")

system_prompt = sp_manager.build("with_user")

2.3 与第一篇应用的集成

在 main.py 的对话接口中集成：

from system_prompt import sp_manager

@app.post("/api/chat")
async def chat(request: Request):
    body = await request.json()
    session_id = body.get("session_id", "default")
    user_message = body.get("message", "")
    mode = body.get("mode", "chat")

    # 根据模式和用户信息动态构建 System Prompt
    if mode == "translate":
        target_lang = body.get("target_lang", "中文")
        sp = sp_manager.build("translator", {
            "source_lang": "英文",
            "target_lang": target_lang,
            "domain": "technology",
        })
    else:
        sp = sp_manager.build("with_user", {
            "user_name": get_user_name(session_id),
            "user_role": get_user_role(session_id),
        })

    # 设置 System Prompt
    history = get_session(session_id)
    if history and history[0]["role"] == "system":
        history[0]["content"] = sp  # 更新已有的
    else:
        history.insert(0, {"role": "system", "content": sp})

    history.append({"role": "user", "content": user_message})
    messages = trim_history(history)

    reply = chat_sync(messages)
    history.append({"role": "assistant", "content": reply})
    return {"reply": reply}

3. 记忆机制

多轮对话不只是"管理上下文"，还要让模型"记住"用户。

3.1 三种记忆层次

记忆级别	说明	存储位置
短期记忆	当前窗口内的对话消息	messages 列表
中期记忆	对话关键信息摘要	Redis / 内存
长期记忆	跨会话的用户画像	数据库

短期记忆：对话窗口内的消息

就是上一节讲的 messages 列表。按滑动窗口策略截断即可。

中期记忆：关键信息提取

把对话中的关键信息提取出来，压缩成结构化的"记忆摘要"：

class ConversationMemory:
    # 会话级记忆管理

    def __init__(self, redis_client=None):
        self._summary_prompt = (
            "请从以下对话中提取用户的关键信息，以 JSON 格式输出：\n"
            "{\n"
            '  "user_preferences": ["偏好1", "偏好2"],\n'
            '  "mentioned_topics": ["话题1", "话题2"],\n'
            '  "user_facts": {"key": "value"},\n'
            '  "pending_tasks": ["待办事项"],\n'
            '  "summary": "对话要点的简短摘要"\n'
            "}\n"
            "如果没有相关信息，对应字段设为空数组或空对象。\n"
            "对话历史：\n{conversation}"
        )
        self.redis = redis_client
        self._local_store = {}

    def extract_summary(self, session_id: str, messages: list) -> dict:
        # 从对话中提取结构化记忆
        recent = messages[-6:] if len(messages) > 6 else messages
        conv_text = "\n".join(
            f"{m['role']}: {m['content'][:500]}" for m in recent
        )
        from llm import chat_sync
        summary = chat_sync([
            {"role": "system", "content": self._summary_prompt.format(
                conversation=conv_text
            )}
        ])
        import json
        try:
            return json.loads(summary)
        except json.JSONDecodeError:
            return {"summary": summary[:200]}

    def save_memory(self, session_id: str, memory: dict):
        # 保存记忆
        if self.redis:
            import json
            self.redis.setex(
                f"memory:{session_id}", 86400 * 7,
                json.dumps(memory, ensure_ascii=False)
            )
        else:
            self._local_store[session_id] = memory

    def load_memory(self, session_id: str) -> dict:
        # 加载记忆
        if self.redis:
            data = self.redis.get(f"memory:{session_id}")
            import json
            return json.loads(data) if data else {}
        return self._local_store.get(session_id, {})

    def inject_memory_to_prompt(self, session_id: str, messages: list) -> list:
        # 将记忆注入到 System Prompt 中
        memory = self.load_memory(session_id)
        if not memory:
            return messages

        memory_text = "\n".join([
            f"- 用户偏好：{', '.join(memory.get('user_preferences', []))}",
            f"- 用户信息：{str(memory.get('user_facts', {}))}",
        ])

        if messages and messages[0]["role"] == "system":
            messages[0]["content"] += f"\n\n## 关于当前用户\n{memory_text}\n"
        return messages


# 使用方式
memory = ConversationMemory()

# 每 5 轮对话提取一次记忆
if len(history) % 10 == 0 and len(history) > 0:
    summary = memory.extract_summary(session_id, history)
    memory.save_memory(session_id, summary)

# 注入记忆到 prompt
history = memory.inject_memory_to_prompt(session_id, history)

长期记忆：跨会话持久化

使用数据库存储用户的长期信息：

import sqlite3, json, time

class LongTermMemory:
    # 长期记忆：跨对话周期的用户信息存储

    def __init__(self, db_path="memory.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS user_memory (
                user_id TEXT PRIMARY KEY,
                memory_data TEXT,
                updated_at INTEGER
            )
        """)
        self.conn.commit()

    def save(self, user_id: str, memory: dict):
        now = int(time.time())
        self.conn.execute(
            "INSERT OR REPLACE INTO user_memory (user_id, memory_data, updated_at) VALUES (?, ?, ?)",
            (user_id, json.dumps(memory, ensure_ascii=False), now),
        )
        self.conn.commit()

    def load(self, user_id: str) -> dict:
        row = self.conn.execute(
            "SELECT memory_data FROM user_memory WHERE user_id=?", (user_id,)
        ).fetchone()
        return json.loads(row[0]) if row else {}

    def update_fact(self, user_id: str, key: str, value: str):
        # 更新单个事实
        memory = self.load(user_id)
        if "user_facts" not in memory:
            memory["user_facts"] = {}
        memory["user_facts"][key] = value
        self.save(user_id, memory)

3.2 三种记忆的协作流程

用户发送消息
    |
    v
加载长期记忆 -> 注入用户画像到 System Prompt
    |
    v
加载中期记忆 -> 注入会话摘要到 System Prompt
    |
    v
滑动窗口截断 messages
    |
    v
调用 LLM
    |
    v
保存本轮对话到历史
    |
    v
判断是否需要提取记忆？
    +-- 是 -> 提取摘要 -> 保存中期记忆 -> 更新长期记忆
    +-- 否 -> 等待下一条消息

4. 高级话题

4.1 Tool Call 的上下文管理

当 Agent 使用工具时，上下文中不仅有对话文本，还有工具调用记录。管理策略有所不同：

def truncate_with_tool_calls(messages: list, max_tokens: int = 8000) -> list:
    # 包含工具调用的上下文截断。
    # 工具调用记录（tool call + tool result）必须成对出现。
    system_msgs = [m for m in messages if m["role"] == "system"]
    history = [m for m in messages if m["role"] != "system"]

    kept = []
    total = sum(len(m["content"]) for m in system_msgs) / 1.5
    available = max_tokens - total

    i = len(history) - 1
    while i >= 0 and total = 0 and history[j]["role"] not in ("assistant", "user"):
                j -= 1
            block = history[j+1:i+1]
            block_tokens = sum(len(m["content"]) / 1.5 for m in block)
            if total + block_tokens > available:
                break
            total += block_tokens
            kept = block + kept
            i = j
        else:
            tokens = len(history[i]["content"]) / 1.5
            if total + tokens > available:
                break
            total += tokens
            kept = [history[i]] + kept
            i -= 1

    return system_msgs + kept

4.2 多 Session 管理

生产环境中，你需要同时管理成千上万个对话 Session：

class SessionManager:
    # 多 Session 管理器

    def __init__(self, redis_client, ttl: int = 86400):
        self.redis = redis_client
        self.ttl = ttl

    def get_history(self, session_id: str) -> list:
        import json
        data = self.redis.get(f"chat:{session_id}")
        return json.loads(data) if data else []

    def append_message(self, session_id: str, message: dict):
        import json
        key = f"chat:{session_id}"
        history = self.get_history(session_id)
        history.append(message)
        if len(history) > 100:
            history = history[-100:]
        self.redis.setex(key, self.ttl, json.dumps(history, ensure_ascii=False))

    def clear(self, session_id: str):
        self.redis.delete(f"chat:{session_id}")

5. 完整集成示例

将所有技术整合到上篇的应用中：

# conversation.py - 对话管理模块
import json, time
from llm import chat_sync, count_tokens

class ConversationManager:
    # 完整的对话管理器

    def __init__(self, redis_client=None, db_path="memory.db"):
        self.sessions = {}
        self.memory = LongTermMemory(db_path)
        self.conv_memory = ConversationMemory(redis_client)
        self.sp_manager = SystemPromptManager()

        self.sp_manager.register("default", (
            "你是一个有用的 AI 助手。\n"
            "当前时间：{current_time}\n"
            "用户信息：{user_info}\n\n"
            "回答要求：\n"
            "- 简洁准确\n"
            "- 不知道就说不知道\n"
            "- 根据之前的对话上下文回答"
        ))

    def process_message(self, session_id: str, user_message: str, user_id: str = None) -> str:
        # 1. 获取或创建会话
        if session_id not in self.sessions:
            self.sessions[session_id] = [{"role": "system", "content": ""}]
        history = self.sessions[session_id]

        # 2. 注入 System Prompt（含用户信息）
        user_info = ""
        if user_id:
            long_memory = self.memory.load(user_id)
            facts = long_memory.get("user_facts", {})
            user_info = "; ".join(f"{k}={v}" for k, v in facts.items())

        session_memory = self.conv_memory.load_memory(session_id)
        memory_summary = session_memory.get("summary", "")

        new_sp = self.sp_manager.build("default", {
            "current_time": time.strftime("%Y-%m-%d %H:%M"),
            "user_info": user_info,
        })
        if memory_summary:
            new_sp += f"\n\n## 对话摘要\n{memory_summary}"
        history[0]["content"] = new_sp

        # 3. 添加用户消息
        history.append({"role": "user", "content": user_message})

        # 4. 滑动窗口截断
        trimmed = truncate_sliding_window(history, max_tokens=4000)

        # 5. 调用 LLM
        reply = chat_sync(trimmed)

        # 6. 保存回复
        history.append({"role": "assistant", "content": reply})

        # 7. 定期提取记忆（每 5 轮提取一次）
        non_system = [m for m in history if m["role"] != "system"]
        if len(non_system) >= 10 and len(non_system) % 10 == 0:
            summary = self.conv_memory.extract_summary(session_id, history[-10:])
            self.conv_memory.save_memory(session_id, summary)

        return reply

    def clear_session(self, session_id: str):
        self.sessions[session_id] = [{"role": "system", "content": ""}]

总结

多轮对话上下文管理的核心原则：

原则	说明
保留完整轮次	不要截断到一半，破坏上下文连贯性
动态注入	System Prompt 应根据用户和场景动态构建
分层次记忆	短期（窗口）、中期（摘要）、长期（数据库）
定期提炼	不要等超限才处理，定期提取记忆摘要
工具调用成对	Tool call + result 必须一起保留或一起丢弃

下一篇我们将进入 RAG 系统的实现——如何从零构建一个文档问答系统，包括文档解析、Chunk 策略、Embedding、向量检索和生成。

本文是 《AI 应用开发实战》系列 的第 3 篇。
系列目录：
1. ✅ 从零搭建你的第一个 AI 应用
2. ✅ Prompt 工程实战
3. ✅ 多轮对话进阶 ← 你在这里
4. 📝 从零实现 RAG 系统
5. 📝 AI Agent——工具调用与自主决策

本文由 Zyentor（智元界）原创发布