一、定义
短期会话记忆（Short-term Conversation Memory）
指在当前单次对话会话内有效、会话结束后即可丢弃的信息。
它承载的是当前对话上下文中的临时状态，包括：
用户刚刚提出的问题
当前任务的进行步骤（如填表单到第几步）
指代消解所需的临时绑定（如“它多少钱”中的“它”指哪个商品）
本轮对话中产生的临时修正或澄清
核心特征：会话结束即清空，不跨会话保留。
长期会话记忆（Long-term Conversation Memory）
指跨会话持久化存储、在未来对话中仍可复用的信息。
它承载的是用户稳定的、可复用的信息，包括：
用户的身份属性（姓名、年龄、职业、地址等）
用户的长期偏好（喜欢简洁回答、不喜欢价格敏感内容等）
跨会话的累积事实（学习进度、项目讨论要点等）
用户明确要求记住的信息
核心特征：会话结束后仍然保留，可在后续对话中被检索和注入。
ISSUE：
1、长短期会话记忆是由谁定义和判断的？
本质上是产品需求，不是模型自己决定的。
结论：定义是人为设定的规则，不是模型推理出来的。
2、谁判断"这条信息属于哪种记忆"？
判断者：可以有三种模式，各有优劣
实际生产中最常用的是：模式A + 模式B 的混合
模式B的具体实现：用Prompt调用模型判断
记忆分类Prompt模板
MEMORY_CLASSIFIER_PROMPT = """
你是一个对话记忆分类器。你的任务是根据用户消息，判断其中的信息应该存入【长期记忆】还是【短期记忆】。
## 定义
- 长期记忆：跨会话有效，未来对话仍需复用的信息（身份、偏好、习惯、重要事实）
- 短期记忆：仅当前会话有效，会话结束即可丢弃的信息（临时状态、时效信息、指代）
## 判断规则
1. 身份信息（姓名、年龄、职业、住址）→ 长期
2. 偏好态度（喜欢/不喜欢/习惯）→ 长期
3. 用户明确要求记住（"记住"、"以后别问了"）→ 长期
4. 时效信息（今天/现在的天气、新闻、价格）→ 短期
5. 临时状态（心情、当前操作）→ 短期
6. 指代消解信息（"它"指什么）→ 短期
7. 有疑问时，默认短期（谨慎原则）
## 用户消息
{user_message}
## 对话历史（可选）
{conversation_history}
## 输出格式（严格JSON）
{
"memory_type": "LONG" 或 "SHORT",
"confidence": 0.0-1.0,
"extracted_info": {"key": "value"}, // 可提取的结构化信息
"reason": "判断理由"
}
"""调用示例代码
import json
from typing import Dict, Any
from openai import OpenAI # 或其他LLM客户端
class LLMConsciousnessDecider:
"""
使用LLM进行记忆类型判断
"""
def __init__(self, model_name: str = "gpt-3.5-turbo"):
self.client = OpenAI()
self.model_name = model_name
def decide(self, user_message: str, history: str = "") -> Dict[str, Any]:
"""调用LLM判断记忆类型"""
prompt = MEMORY_CLASSIFIER_PROMPT.format(
user_message=user_message,
conversation_history=history or "无"
)
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "你是一个精确的记忆分类器，只输出JSON。"},
{"role": "user", "content": prompt}
],
temperature=0.1, # 低温度，保证稳定性
response_format={"type": "json_object"} # 强制JSON输出
)
result = json.loads(response.choices[0].message.content)
# 置信度过滤：低于0.7的按默认短期处理
if result.get("confidence", 0) < 0.7:
result["memory_type"] = "SHORT"
result["reason"] += " (置信度不足，降级为短期)"
return result
# 使用示例
decider = LLMConsciousnessDecider()
# 测试
test_messages = [
"我叫张三，今年28岁",
"今天天气真好",
"我不喜欢推荐太贵的东西，记住这个",
"那个多少钱？" # 指代
]
for msg in test_messages:
result = decider.decide(msg)
print(f"消息: {msg}")
print(f"判断: {result}")
print("---")输出示例
// 输入："我叫张三，今年28岁"
{
"memory_type": "LONG",
"confidence": 0.95,
"extracted_info": {"name": "张三", "age": 28},
"reason": "包含身份信息（姓名、年龄），属于长期记忆"
}
// 输入："今天天气真好"
{
"memory_type": "SHORT",
"confidence": 0.92,
"extracted_info": {},
"reason": "时效性信息（今天），会话结束后无意义，属于短期记忆"
}三种模式的详细对比与选型建议
混合模式架构图
用户消息
↓
┌─────────────────────────────────────┐
│ 第一层：规则引擎（毫秒级） │
│ - 关键词匹配 │
│ - 正则表达式 │
│ - 黑名单/白名单 │
└─────────────┬───────────────────────┘
│
▼
命中规则？
├── 是 → 直接返回结果（95%流量）
│
└── 否 ↓
│
▼
┌─────────────────────────────────────┐
│ 第二层：LLM分类器（百毫秒级） │
│ - 小模型（GPT-3.5/TinyBERT） │
│ - 仅处理边界模糊的情况 │
└─────────────┬───────────────────────┘
│
▼
返回结果
二、什么时候用？——决策框架
短期记忆的使用条件（满足任一即用短期）
长期记忆的使用条件（需同时满足）
快速判断流程
用户信息
↓
下次对话还会用到吗？
├── 否 → 短期记忆
└── 是 → 进一步判断
├── 是临时状态/情绪？ → 短期记忆
├── 是稳定属性/偏好？ → 长期记忆
└── 用户要求记住？ → 长期记忆三、两种记忆对比总结
四、一个重要的经验法则
大多数对话系统的问题是“存了太多不该存的信息到长期记忆”，而不是记性不好。
写入原则
典型错误案例
实战决策流程（可落地为规则引擎）
当系统收到一条用户消息时，按以下顺序判断：
用户消息 → 信息提取
↓
【规则1】是否包含时间限定词？
- "今天/刚才/现在" → 倾向短期
- "总是/从来/我一直" → 倾向长期
↓
【规则2】信息类型分类？
- 身份属性（姓名/年龄/职业/地址） → 长期
- 临时状态（心情/位置/当前操作） → 短期
- 偏好态度（喜欢/不喜欢/习惯） → 长期
- 任务进度（填表第几步/临时选择） → 短期
↓
【规则3】复用概率评估？
- 下次会话用到的概率 > 50% → 长期
- 概率低或仅当前会话有用 → 短期
↓
【规则4】是否有明确信号？
- 用户说"记住"、"以后" → 强制长期
- 用户说"只是这次"、"暂时" → 强制短期
↓
【规则5】冲突处理（用户修正）？
- "我之前说的那个不对，其实是..." → 更新/覆盖长期记忆一个简单实用的分层策略
如果你不想做复杂的判断，可以按这个经验法则：
实际操作中的"用与不用"速查表
大多数对话系统的问题是"存了太多不该存的短期信息到长期记忆里"，而不是记性不好。
比如：
用户随口说"今天好冷" → 被存成"用户怕冷"（错误）
用户问"北京天气" → 被存成"用户住在北京"（错误）
用户抱怨"这个答案太长了" → 被存成"用户讨厌长回答"（可能正确，需要多次确认）
好的记忆策略是"谨慎写入，积极读取"：
写入长期记忆前，需要多次确认或高置信度
短期记忆全部写入，但会话结束即清空
长期记忆写入门槛高，但读取时可以宽松召回
五、工程架构总结
┌─────────────────────────────────────────────────────────┐
│ 用户对话 │
└─────────────────────────┬───────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────┐
│ 【判断层】信息分类 │
│ - 是否跨会话复用？ │
│ - 是临时状态还是稳定属性？ │
└─────────────────────────┬───────────────────────────────┘
↓ ↓
┌───────────────┐ ┌───────────────┐
│ 短期记忆 │ │ 长期记忆 │
│ （Redis） │ │ （向量DB+SQL） │
│ │ │ │
│ 会话结束清空 │ │ 持久化保留 │
│ 滑动窗口管理 │ │ 语义检索注入 │
└───────────────┘ └───────────────┘
↓ ↓
┌─────────────────────────────────────────────────────┐
│ 【注入】合并后送入大模型上下文 │
└─────────────────────────────────────────────────────┘六、总结
短期记忆存“当前会话的临时状态”，会话结束即清空；长期记忆存“可跨会话复用的稳定信息”，谨慎写入、积极检索。判断标准只有一个核心问题：下次对话用户还会希望我记得这个吗？记忆判断不应该是主LLM的任务，正确做法：记忆分类由独立模块完成（规则引擎 或 专用小模型），主LLM只负责生成回复，接收已经分类好的记忆上下文。
完整 Demo：智能客服记忆系统
"""
会话记忆系统 Demo
- 短期记忆：当前会话内的临时信息
- 长期记忆：跨会话持久化的用户画像和偏好
"""
import json
import sqlite3
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
# ==================== 数据结构定义 ====================
class MemoryType(Enum):
"""记忆类型"""
SHORT_TERM = "short_term"
LONG_TERM = "long_term"
@dataclass
class MemoryItem:
"""单条记忆"""
key: str
value: Any
memory_type: MemoryType
confidence: float = 1.0 # 置信度（0-1）
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
@dataclass
class ConversationTurn:
"""单轮对话"""
role: str # "user" 或 "assistant"
content: str
timestamp: datetime = field(default_factory=datetime.now)
# ==================== 记忆判断器 ====================
class MemoryDecider:
"""
判断信息应该存入短期还是长期记忆
"""
# 长期记忆的关键词信号
LONG_TERM_SIGNALS = [
"我叫", "我是", "我的名字", # 身份信息
"我喜欢", "我讨厌", "我不喜欢", # 偏好
"我总是", "我从来不", # 习惯
"记住", "别忘了", "以后", # 明确要求
"我是做", "我的职业", # 职业信息
"我住在", "我的地址", # 地址信息
]
# 短期记忆的关键词信号
SHORT_TERM_SIGNALS = [
"今天", "刚才", "现在", # 时效性
"这次", "暂时", # 临时性
]
@classmethod
def decide(cls, user_message: str, context: Dict = None) -> MemoryType:
"""
判断记忆类型
返回：MemoryType.LONG_TERM 或 MemoryType.SHORT_TERM
"""
msg_lower = user_message.lower()
# 规则1：明确要求长期记忆
for signal in cls.LONG_TERM_SIGNALS:
if signal in msg_lower:
return MemoryType.LONG_TERM
# 规则2：明确短期信号
for signal in cls.SHORT_TERM_SIGNALS:
if signal in msg_lower:
return MemoryType.SHORT_TERM
# 规则3：信息类型推断
if cls._is_personal_info(user_message):
return MemoryType.LONG_TERM
if cls._is_temporal_info(user_message):
return MemoryType.SHORT_TERM
# 默认：短期记忆（谨慎原则）
return MemoryType.SHORT_TERM
@classmethod
def _is_personal_info(cls, message: str) -> bool:
"""判断是否包含个人信息"""
patterns = ["岁", "年", "工作", "学习", "住在", "来自"]
return any(p in message for p in patterns)
@classmethod
def _is_temporal_info(cls, message: str) -> bool:
"""判断是否包含时效信息"""
patterns = ["天气", "温度", "时间", "几点"]
return any(p in message for p in patterns)
# ==================== 信息提取器 ====================
class InfoExtractor:
"""
从用户消息中提取结构化信息
实际项目中可以用 LLM 做更智能的提取
"""
@classmethod
def extract(cls, message: str) -> Dict[str, Any]:
"""提取信息键值对"""
extracted = {}
msg = message
# 提取姓名
if "我叫" in msg:
name = msg.split("我叫")[-1].split("，")[0].split("。")[0].strip()
extracted["name"] = name
# 提取年龄
if "岁" in msg:
import re
match = re.search(r'(\d+)岁', msg)
if match:
extracted["age"] = int(match.group(1))
# 提取偏好
if "我喜欢" in msg:
like = msg.split("我喜欢")[-1].split("，")[0].split("。")[0].strip()
extracted["preference"] = like
if "我不喜欢" in msg or "我讨厌" in msg:
dislike = msg.replace("我不喜欢", "").replace("我讨厌", "").split("，")[0].split("。")[0].strip()
extracted["dislike"] = dislike
# 提取位置
if "住在" in msg:
location = msg.split("住在")[-1].split("，")[0].split("。")[0].strip()
extracted["location"] = location
return extracted
# ==================== 长期记忆存储（SQLite + 向量检索简化版）====================
class LongTermMemory:
"""长期记忆存储"""
def __init__(self, db_path: str = "memory.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""初始化数据库"""
self.conn = sqlite3.connect(self.db_path)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS user_memory (
user_id TEXT,
memory_key TEXT,
memory_value TEXT,
confidence REAL,
created_at TEXT,
updated_at TEXT,
PRIMARY KEY (user_id, memory_key)
)
""")
self.conn.commit()
def save(self, user_id: str, memory_key: str, memory_value: Any, confidence: float = 1.0):
"""保存长期记忆"""
now = datetime.now().isoformat()
self.conn.execute("""
INSERT OR REPLACE INTO user_memory
(user_id, memory_key, memory_value, confidence, created_at, updated_at)
VALUES (?, ?, ?, ?, COALESCE((SELECT created_at FROM user_memory
WHERE user_id = ? AND memory_key = ?), ?), ?)
""", (user_id, memory_key, json.dumps(memory_value), confidence,
user_id, memory_key, now, now))
self.conn.commit()
print(f"💾 [长期记忆] 已保存: {memory_key} = {memory_value}")
def load_all(self, user_id: str) -> Dict[str, Any]:
"""加载用户的所有长期记忆"""
cursor = self.conn.execute("""
SELECT memory_key, memory_value FROM user_memory
WHERE user_id = ?
""", (user_id,))
memory = {}
for key, value in cursor:
memory[key] = json.loads(value)
return memory
def load(self, user_id: str, memory_key: str) -> Optional[Any]:
"""加载单个记忆"""
cursor = self.conn.execute("""
SELECT memory_value FROM user_memory
WHERE user_id = ? AND memory_key = ?
""", (user_id, memory_key))
row = cursor.fetchone()
return json.loads(row[0]) if row else None
def close(self):
self.conn.close()
# ==================== 短期记忆（内存存储）====================
class ShortTermMemory:
"""短期记忆存储（会话级）"""
def __init__(self):
self.memory: Dict[str, Any] = {}
self.conversation_history: List[ConversationTurn] = []
def save(self, key: str, value: Any):
"""保存短期记忆"""
self.memory[key] = value
print(f"📝 [短期记忆] 已保存: {key} = {value}")
def get(self, key: str) -> Optional[Any]:
"""获取短期记忆"""
return self.memory.get(key)
def add_conversation_turn(self, role: str, content: str):
"""添加对话轮次"""
self.conversation_history.append(ConversationTurn(role=role, content=content))
# 只保留最近10轮
if len(self.conversation_history) > 10:
self.conversation_history = self.conversation_history[-10:]
def get_recent_context(self, n: int = 5) -> str:
"""获取最近n轮对话上下文"""
recent = self.conversation_history[-n:]
context = ""
for turn in recent:
context += f"{turn.role}: {turn.content}\n"
return context
def clear(self):
"""清空短期记忆（会话结束调用）"""
self.memory.clear()
self.conversation_history.clear()
print("🗑️ [短期记忆] 已清空")
# ==================== 记忆管理器 ====================
class MemoryManager:
"""统一的记忆管理器"""
def __init__(self, user_id: str):
self.user_id = user_id
self.short_term = ShortTermMemory()
self.long_term = LongTermMemory()
self.decider = MemoryDecider()
self.extractor = InfoExtractor()
# 加载用户的长期记忆到缓存
self.user_profile = self.long_term.load_all(user_id)
def process_user_message(self, message: str) -> Dict[str, Any]:
"""
处理用户消息
1. 判断记忆类型
2. 提取信息
3. 存储到对应记忆
4. 返回需要注入到 LLM 的上下文
"""
print(f"\n{'='*50}")
print(f"👤 用户: {message}")
# 1. 判断记忆类型
memory_type = self.decider.decide(message)
print(f"🔍 判断结果: {memory_type.value}")
# 2. 提取信息
extracted = self.extractor.extract(message)
# 3. 存储
for key, value in extracted.items():
if memory_type == MemoryType.LONG_TERM:
# 长期记忆：需要较高置信度，可能需要多次确认
self.long_term.save(self.user_id, key, value, confidence=0.8)
# 更新缓存
self.user_profile[key] = value
else:
# 短期记忆：直接存储
self.short_term.save(key, value)
# 4. 特殊处理：指代消解
resolved = self._resolve_reference(message)
if resolved != message:
print(f"🔗 [指代消解] {message} → {resolved}")
# 5. 添加对话历史
self.short_term.add_conversation_turn("user", message)
# 6. 构建注入上下文
context = self._build_context(resolved)
return {
"resolved_message": resolved,
"memory_type": memory_type.value,
"extracted": extracted,
"context_for_llm": context
}
def _resolve_reference(self, message: str) -> str:
"""简单的指代消解"""
# 处理"它"
if "它" in message and self.short_term.get("last_mentioned"):
message = message.replace("它", self.short_term.get("last_mentioned"))
# 处理"那个"
if "那个" in message and self.short_term.get("last_item"):
message = message.replace("那个", self.short_term.get("last_item"))
return message
def _build_context(self, resolved_message: str) -> str:
"""构建注入到 LLM 的上下文"""
context = ""
# 注入长期记忆（用户画像）
if self.user_profile:
context += "## 用户画像（长期记忆）\n"
for key, value in self.user_profile.items():
context += f"- {key}: {value}\n"
context += "\n"
# 注入短期记忆（最近对话）
recent = self.short_term.get_recent_context(3)
if recent:
context += "## 最近对话（短期记忆）\n"
context += recent
context += "\n"
# 注入当前消息
context += f"## 当前用户消息\n{resolved_message}\n"
return context
def assistant_response(self, response: str):
"""记录助手的回复"""
self.short_term.add_conversation_turn("assistant", response)
# 提取可能的关键信息用于指代消解
# 简单示例：提取商品名等
if "商品" in response or "产品" in response:
# 实际应用中可以用更智能的提取
pass
def end_session(self):
"""结束会话"""
self.short_term.clear()
print(f"👋 用户 {self.user_id} 会话已结束")
def close(self):
self.long_term.close()
# ==================== 模拟 LLM 响应 ====================
def mock_llm_response(context: str, user_message: str) -> str:
"""
模拟 LLM 响应
实际应用中这里调用真实的 LLM API
"""
# 简单的规则响应（演示用）
msg_lower = user_message.lower()
if "你好" in msg_lower or "嗨" in msg_lower:
return "你好！有什么我可以帮你的吗？"
if "天气" in msg_lower:
return "今天深圳天气晴朗，气温22-28°C。需要我帮你查其他城市的天气吗？"
if "价格" in msg_lower or "多少钱" in msg_lower:
# 检查长期记忆中是否有价格敏感偏好
if "价格敏感" in context or "dislike" in context and "价格" in context:
return "关于价格，我建议我们私下讨论。或者您可以先看看我们的免费试用版。"
return "我们的产品有多种套餐，基础版每月99元，专业版每月299元。您想了解哪个？"
if "推荐" in msg_lower:
# 根据偏好推荐
if "偏好" in context and "电影" in context:
return f"根据您喜欢{context.split('偏好')[-1]}的偏好，我推荐您看《奥本海默》和《流浪地球3》。"
return "您想了解哪方面的推荐？我可以为您推荐商品、电影、书籍等。"
if "我叫" in user_message:
return f"好的，{user_message.split('我叫')[-1].strip()}，很高兴认识你！"
if "谢谢" in msg_lower:
return "不客气！如果还有其他问题，随时问我。"
return "我理解了。请问还需要我帮你做什么？"
# ==================== 演示运行 ====================
def run_demo():
"""运行完整演示"""
print("="*60)
print("🎯 智能会话记忆系统 Demo")
print("="*60)
# 创建用户会话
manager = MemoryManager(user_id="user_001")
# 模拟对话场景
conversations = [
# 场景1：用户介绍自己（应存入长期记忆）
"你好，我叫张三",
# 场景2：简单问候
"你好！今天深圳天气怎么样？",
# 场景3：用户表达偏好（应存入长期记忆）
"我不喜欢推荐太贵的东西，我是个学生",
# 场景4：询问产品（应使用短期记忆中的偏好）
"那你们的产品有什么推荐吗？",
# 场景5：指代消解测试
"它多少钱？", # "它"指上一轮推荐的产品
# 场景6：结束对话
"谢谢，再见！"
]
for i, msg in enumerate(conversations, 1):
print(f"\n{'─'*40}")
print(f"📌 第 {i} 轮对话")
# 处理用户消息
result = manager.process_user_message(msg)
# 模拟 LLM 响应
response = mock_llm_response(result["context_for_llm"], result["resolved_message"])
# 记录助手响应
manager.assistant_response(response)
print(f"🤖 助手: {response}")
print(f"📋 注入上下文:\n{result['context_for_llm'][:200]}...")
# 显示最终存储的长期记忆
print("\n" + "="*60)
print("📚 最终长期记忆（用户画像）:")
for key, value in manager.user_profile.items():
print(f" - {key}: {value}")
# 结束会话
manager.end_session()
manager.close()
print("\n" + "="*60)
print("✅ Demo 运行完成")
print("="*60)
# ==================== 可选：查看长期记忆数据库 ====================
def show_stored_memory():
"""查看数据库中存储的长期记忆"""
conn = sqlite3.connect("memory.db")
cursor = conn.execute("SELECT user_id, memory_key, memory_value, confidence FROM user_memory")
print("\n💾 数据库中存储的长期记忆:")
for row in cursor:
print(f" user={row[0]}, key={row[1]}, value={row[2]}, confidence={row[3]}")
conn.close()
# ==================== 运行 ====================
if __name__ == "__main__":
run_demo()
show_stored_memory()运行结果示例
============================================================
🎯 智能会话记忆系统 Demo
============================================================
────────────────────────────────────────
📌 第 1 轮对话
👤 用户: 你好，我叫张三
🔍 判断结果: long_term
💾 [长期记忆] 已保存: name = "张三"
🤖 助手: 好的，张三，很高兴认识你！
────────────────────────────────────────
📌 第 2 轮对话
👤 用户: 你好！今天深圳天气怎么样？
🔍 判断结果: short_term
📝 [短期记忆] 已保存: location = "深圳"
🤖 助手: 今天深圳天气晴朗，气温22-28°C...
────────────────────────────────────────
📌 第 3 轮对话
👤 用户: 我不喜欢推荐太贵的东西，我是个学生
🔍 判断结果: long_term
💾 [长期记忆] 已保存: dislike = "太贵的东西"
💾 [长期记忆] 已保存: occupation = "学生"
🤖 助手: 我理解了...
────────────────────────────────────────
📌 第 4 轮对话
👤 用户: 那你们的产品有什么推荐吗？
🔍 判断结果: short_term
📋 注入上下文包含用户画像中的 dislike
🤖 助手: 关于价格，我建议我们私下讨论...
============================================================
📚 最终长期记忆（用户画像）:
- name: 张三
- dislike: 太贵的东西
- occupation: 学生快速上手
bash
# 保存代码为 memory_demo.py
# 直接运行
python memory_demo.py这个 Demo 演示了：
✅ 短期/长期记忆的判断逻辑
✅ 信息提取和存储
✅ 指代消解
✅ 上下文注入 LLM
✅ 会话结束清空短期记忆
