ai对话 上下文管理
一、Memory 管理与对话系统概述
1.1 什么是 Memory 管理
定义：Memory 管理是对话系统（如聊天机器人、AI 助手）中的核心技术，研究如何让系统拥有并有效管理记忆，从而能够进行连贯、有深度、个性化的多轮对话。
1.2 为什么需要 Memory 管理
| 对比项 | 无 Memory | 有 Memory |
|---|---|---|
| 上下文理解 | 每次对话独立，无法理解上文 | 记住历史对话，理解指代关系 |
| 用户体验 | 需要重复说明背景信息 | 无需重复，对话自然流畅 |
| 个性化 | 每次都是"新用户" | 记住用户偏好和习惯 |
| 任务完成 | 无法处理多步骤任务 | 可以分步引导完成复杂任务 |
示例对比：
无 Memory 的对话：
用户: 请帮我生成论文摘要
AI: 已生成摘要
用户: 太长了，缩短一点
AI: 你想缩短什么？ ← ❌ 不记得上一轮的摘要
有 Memory 的对话：
用户: 请帮我生成论文摘要
AI: 已生成摘要
用户: 太长了，缩短一点
AI: 已将摘要从 500 字缩短到 300 字 ← ✅ 记得上一轮的摘要
二、长短期记忆分离（Long-term/Short-term Memory Separation）
2.1 什么是长短期记忆分离
定义：一种核心的记忆管理策略，将对话中的信息区分为两种类型：
| 记忆类型 | 内容 | 生命周期 | 存储位置 | 示例 |
|---|---|---|---|---|
| 短期记忆 | 当前对话上下文 | 会话期间 | 内存（上下文窗口） | "刚才提到的摘要"、"上一轮的问题" |
| 长期记忆 | 用户偏好、核心事实 | 持久化 | 数据库 | "用户是计算机专业"、"论文标题" |
2.2 工作原理
┌─────────────────────────────────────────────────────────┐
│ 短期记忆（Short-term Memory） │
│ - 存储：内存中的对话历史 │
│ - 容量：受 LLM 上下文窗口限制（如 128K tokens） │
│ - 策略：FIFO 或动态截断 │
│ │
│ 对话 1: "请生成摘要" │
│ 对话 2: "再生成关键词" │
│ 对话 3: "摘要太长了" ← 记得对话 1 │
└─────────────────────────────────────────────────────────┘
↓
定期提取关键信息
↓
┌─────────────────────────────────────────────────────────┐
│ 长期记忆（Long-term Memory） │
│ - 存储：数据库（如 MySQL、Redis） │
│ - 容量：几乎无限 │
│ - 内容：用户画像、论文信息、历史评审结果 │
│ │
│ - 用户ID: 12345 │
│ - 专业: 计算机科学 │
│ - 论文标题: "深度学习图像分割研究" │
│ - 历史评分: [7.2, 8.1, 8.6] ← 跨会话的持久化数据 │
└─────────────────────────────────────────────────────────┘
2.3 项目中的实现
长期记忆的效果：
第 1 天：用户提交论文 V1
→ AI 评审：摘要评分 6.5/10
→ 保存到数据库（长期记忆）
第 2 天：用户提交论文 V2
→ AI 加载 V1 的评审结果（长期记忆）
→ AI 评审：摘要评分 7.8/10，改进了 V1 的问题 ✓
→ 保存到数据库
第 3 天：用户提交论文 V3
→ AI 加载 V1、V2 的评审结果（长期记忆）
→ AI 评审：摘要评分 8.6/10，持续改进 ✓
→ AI 判断：已收敛，可以提交
三、对话上下文保持（Maintaining Conversational Context）
3.1 什么是对话上下文保持
定义：对话上下文保持是实现多轮对话的基础，要求系统能够在连续的交互中理解并"记住"之前聊过的内容，确保新的回应与历史对话是连贯和相关的。
3.2 核心挑战
| 挑战 | 说明 | 解决方案 |
|---|---|---|
| 上下文窗口限制 | LLM 上下文有限（如 128K tokens） | 动态截断，优先保留最近对话 |
| Token 成本 | 历史对话越多，Token 消耗越大 | 压缩历史，提取关键信息 |
| 相关性衰减 | 早期对话可能不再相关 | FIFO 策略，丢弃最早的对话 |
| 指代消解 | "它"、"这个"等指代关系 | 保留必要的上下文 |
3.3 项目中的实现：智能截断策略
五大策略详解：
策略 1：倒序添加（Reverse Order）
┌──────────────────────────────────────┐
│ 历史对话（升序存储） │
│ [对话1, 对话2, 对话3, 对话4, 对话5] │
└──────────────────────────────────────┘
↓ 倒序遍历
┌──────────────────────────────────────┐
│ 优先添加最近的对话 │
│ 对话5 → 对话4 → 对话3 → 对话2 → 对话1 │
└──────────────────────────────────────┘
策略 2：动态 Token 计算
┌──────────────────────────────────────┐
│ 每轮对话的 Token 消耗： │
│ - System Prompt: 500 tokens │
│ - 当前问题: 200 tokens │
│ - 对话5: 300 tokens ← 最近的 │
│ - 对话4: 400 tokens │
│ - 对话3: 500 tokens │
│ Total: 1900 tokens ✓ │
│ │
│ - 对话2: 600 tokens │
│ Total: 2500 tokens > 2000 ✗ 超限 │
└──────────────────────────────────────┘
策略 3：超限截断
if total_tokens + tokens > max_prompt_tokens:
break # 停止添加，保留对话5、4、3
策略 4：倒序插入
遍历顺序：对话5 → 对话4 → 对话3
插入顺序：对话3 → 对话4 → 对话5
最终顺序：[对话3, 对话4, 对话5] ← 时间正序
策略 5：轮数限制
最多保留 5 轮对话（agent_options.max_histories_turns = 5）
即使 Token 未超限，也只保留最近的 5 轮
四、多轮交互管理（Multi-turn Interaction Management）
4.1 什么是多轮交互管理
定义：指对话系统控制和引导整个对话流程的能力。这不仅包括记住上下文，还涉及理解用户的意图转变、处理澄清和纠正、并在适当的时候主动引导话题。
4.2 多轮交互的核心要素
| 要素 | 说明 | 示例 |
|---|---|---|
| 意图识别 | 理解用户当前的目标 | "缩短"指的是缩短摘要 |
| 指代消解 | 理解"它"、"这个"等指代 | "它太长了" → "摘要太长了" |
| 澄清请求 | 当不确定时主动询问 | "你是想缩短摘要还是关键词？" |
| 主动引导 | 引导用户完成任务 | "摘要已生成，是否需要生成关键词？" |
| 错误纠正 | 理解并纠正用户的错误表达 | "帮我把摘要缩短到 300 个字符" → "已缩短到 300 字" |
4.3 项目中的实现：Conversation ID 会话管理
① 会话 ID 追踪
class BaseRewriteParam(BaseModel):
"""
🔥 核心：使用 conversation_id 追踪多轮对话
"""
conversation_id: Optional[str] = Field(None, title="会话ID")
class BaseRewriteExecutor(ABC):
async def execute_stream(self, params: BaseRewriteParam):
"""
🔥 流式执行：管理整个对话会话
"""
# 生成或使用已有的 conversation_id
conversation_id = params.conversation_id or str(uuid.uuid4())
# 🔥 将 conversation_id 注入 Langfuse 追踪
langfuse_context.update_current_trace(
name="Rewrite[Stream Mode]",
session_id=f"{conversation_id}", # 🔥 会话级追踪
metadata={
"conversation_id": conversation_id,
"request_id": request_id
}
)
# 发送开始消息（包含 conversation_id）
msg = SSEMsg(
event=SSEMsgEvent.Start,
data=SSEMsgBody(
conversation_id=conversation_id,
request_id=request_id
)
)
yield msg
# ... 执行 Agent
② 数据库存储：会话历史
class RewriteTask(Base):
"""
🔥 数据库表：存储每轮对话的输入和输出
"""
__tablename__ = 'rewrite_task'
id = Column(Integer, primary_key=True)
# 🔥 核心：conversation_id 用于关联同一会话的多轮对话
conversation_id = Column(String, nullable=False, index=True)
request_id = Column(String, nullable=False, index=True)
# 🔥 存储每轮对话的输入和输出
user_input = Column(JSON, nullable=True)
output = Column(JSON, nullable=True)
status = Column(Integer, nullable=False, default=0)
start_time = Column(TIMESTAMP, nullable=True)
end_time = Column(TIMESTAMP, nullable=True)
create_time = Column(TIMESTAMP, server_default=func.now())
③ 加载历史对话
class RewriteExecutor(BaseRewriteExecutor):
async def _build_agent_input(self, params: RewriteExecutorParam):
"""
🔥 核心方法：加载同一 conversation_id 的历史对话
"""
rewrite_input = RewriteAgentInput(
selected_text=params.selected_text,
user_instruction=params.user_instruction,
# ... 其他参数
)
# 🔥 如果有 conversation_id，加载历史对话
if params.conversation_id:
# 从数据库查询同一 conversation_id 的所有任务
tasks = RewriteTaskService.get_by_conversation(params.conversation_id)
# 🔥 按时间排序
existing_tasks = sorted(tasks, key=lambda x: x.start_time)
# 🔥 遍历历史任务，追加到当前输入
for existing_task in existing_tasks:
if existing_task.status == TaskStatus.Finished:
# 解析历史输出
history_output = json.loads(existing_task.output)
history_input = RewriteExecutorParam.model_validate_json(
existing_task.user_input
)
# 构造历史 Agent 输入
history_agent_input = RewriteAgentInput(
selected_text=history_input.selected_text,
user_instruction=history_input.user_instruction,
# ... 其他参数
)
# 构造历史 Agent 输出
history_agent_output = RewriteAgentResponse.model_validate(
history_output["rewrite"]
)
# 🔥 追加到当前输入的历史列表
rewrite_input.append_histories(
history_agent_input,
history_agent_output
)
return rewrite_input
完整的多轮交互流程：
┌─────────────────────────────────────────────────────────┐
│ 第 1 轮对话 │
│ - conversation_id: "abc123"（首次生成） │
│ - 用户: "请改写这段文字" │
│ - AI: "已改写" │
│ - 存储到数据库（conversation_id="abc123"） │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 第 2 轮对话 │
│ - conversation_id: "abc123"（客户端传入） │
│ - 加载历史：[第1轮] │
│ - 用户: "语气正式一点" │
│ - AI: （理解"它"指的是第1轮的改写结果）"已正式化" │
│ - 存储到数据库（conversation_id="abc123"） │
└─────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 第 3 轮对话 │
│ - conversation_id: "abc123" │
│ - 加载历史：[第1轮, 第2轮] │
│ - 用户: "缩短到 100 字" │
│ - AI: （理解上下文，缩短第2轮的结果）"已缩短" │
│ - 存储到数据库 │
└─────────────────────────────────────────────────────────┘
五、面试话术建议
问题 1："你们如何管理对话历史和上下文？"
回答模板：
"我们采用了长短期记忆分离的策略：
短期记忆存储在内存中的对话历史（histories_in_asc 字段），实现了智能截断机制：倒序添加历史消息，优先保留最近的对话；动态计算 Token，超限则停止添加；最多保留 5 轮对话。这让 10 轮对话的 Token 消耗从 45,000 降至 15,000，节省了 67%。
长期记忆持久化到数据库（RewriteTask 表），使用 conversation_id 关联同一会话的多轮对话。每次请求时，从数据库加载历史对话，追加到当前输入的 histories 字段。
效果：用户可以在多天后继续之前的对话，系统依然记得之前的上下文。比如论文评审场景，系统会记住每个版本的评审结果，给出针对性的改进建议。"
问题 2："长短期记忆分离的优势是什么？"
回答模板：
"长短期记忆分离有三个核心优势：
第一是 Token 优化。短期记忆只保留最近 5 轮对话，避免历史对话无限累积。如果全部加载到上下文，10 轮对话会消耗 45,000 tokens；分离后只需 15,000 tokens，节省 67%。
第二是跨会话持久化。长期记忆存储在数据库，用户可以在不同设备、不同时间继续同一个对话。比如论文评审场景，用户今天提交 V1，明天提交 V2，系统依然记得 V1 的评审结果。
第三是灵活的内存管理。短期记忆可以快速清理（重新开始对话），长期记忆可以选择性加载（只加载相关的历史）。这让系统在处理长对话时依然保持高效。"
问题 3："如何处理上下文窗口限制？"
回答模板：
"我们实现了五层策略来处理上下文窗口限制：
策略1：倒序添加。从最近的对话开始添加，确保最相关的上下文被保留。
策略2：动态 Token 计算。每添加一轮对话都计算 Token 消耗，实时检查是否超限。
策略3：超限截断。一旦 Token 超过 max_prompt_tokens，立即停止添加更早的对话。
策略4：倒序插入。虽然倒序遍历，但插入时保证最终顺序是时间正序，保持对话的逻辑连贯性。
策略5：轮数限制。即使 Token 未超限，也最多保留 5 轮对话，避免上下文过长影响生成质量。
实际效果：在 10 轮对话的场景下，Token 消耗从 45,000 降至 15,000，同时保证了最相关的上下文被保留，用户满意度达到 95%。"
浙公网安备 33010602011771号