你跟AI说"我习惯用vim，不要用nano"，下次对话它照样给你nano。每次都要重复说，烦不烦？
先说结论
核心思路就三行：
1
| session结束 → 调免费LLM提取记忆 → 向量去重后存入ChromaDB
|
整篇文章读完，你的Agent就能在每次对话结束时自动把用户偏好、事件、工作流提取出来，下次对话自动加载。成本：¥0。
架构
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| ┌──────────┐ session end ┌──────────────┐
│ 对话记录 │ ──────────────→ │ 已有记忆注入P2 │
└──────────┘ └──────┬───────┘
│
┌──────▼───────┐
│ 免费LLM提取P0 │
│ 分类+结构化 │
└──────┬───────┘
│
┌──────▼───────┐
│ 向量去重 P1 │
│ cosine<0.15 │
└──────┬───────┘
│
┌──────▼───────┐
│ ChromaDB │
│ 本地存储 │
└──────────────┘
|
三层防护确保不存垃圾、不存重复：
| 层级 | 机制 | 作用 |
|---|
| P2 提取前 | 已有记忆注入prompt | 模型直接跳过已知内容 |
| P0 提取中 | 免费LLM分类提取 | persona/episodic/instruction |
| P1 存入时 | ChromaDB向量去重 | 兜底防重复存储 |
前置条件
- ChromaDB：本地向量数据库，
pip install chromadb - 免费LLM API Key：在 智谱开放平台 注册账号，自动获得API Key。GLM-4-Flash模型永久免费，无需绑定支付方式
- Python 3.9+
Step 1: 提取Prompt模板
这个prompt是整个系统的灵魂。把它加到你的MemoryProvider插件里：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| EXTRACTION_SYSTEM_PROMPT = (
'你是一个记忆提取助手。从对话中提取值得长期记住的信息。\n'
'严格按以下JSON格式输出，不要输出其他任何内容：\n'
'{"persona":["字符串1","字符串2"],"episodic":["字符串1"],"instruction":["字符串1"]}\n\n'
'分类规则：\n'
'- persona: 用户偏好、习惯、身份、性格特征、沟通风格\n'
'- episodic: 具体事件和行动\n'
'- instruction: 工作流程、规则、纠正、技术方案\n\n'
'关键规则：\n'
'1. 只提取新的、具体的、有价值的信息\n'
'2. 数组中每个元素必须是纯字符串，不要用对象\n'
'3. 没有有价值的信息就全部输出空数组\n'
'4. 每条不超过80字，用简洁陈述句\n'
'5. 不要重复同一条信息\n'
'6. 偏好类信息归persona，不要归episodic'
)
|
为什么这么设计：
temperature: 0.1 + 严格格式约束 → 输出稳定可解析- 三分类覆盖了90%以上的记忆类型
- “没有有价值的信息就输出空数组” → 避免硬造记忆
Step 2: on_session_end钩子
Session结束时触发提取。关键设计：后台线程异步执行，不阻塞用户。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| import threading
import json
def on_session_end(self, messages):
# 过滤出user/assistant消息
conversation_lines = []
for msg in messages:
role = msg.get('role', '')
content = msg.get('content', '')
if role in ('user', 'assistant') and content:
prefix = '用户' if role == 'user' else '助手'
conversation_lines.append(f'{prefix}：{content[:500]}')
if len(conversation_lines) < 2:
return # 太短不提取
conversation_text = '\n'.join(conversation_lines)
if len(conversation_text) > 4000:
conversation_text = conversation_text[-4000:] # 截取最近4000字符
# 预取已有记忆（P2缓存）
existing_context = self._build_existing_context()
# 后台线程执行提取
def _extract():
self._run_extraction(conversation_text, existing_context)
threading.Thread(target=_extract, daemon=True).start()
|
注意几个细节：
- 每条消息截取500字符 → 控制总量
- 总文本超过4000字符只取末尾 → 最近的内容最有价值
- 对话少于2条直接跳过 → 避免无效提取
daemon=True → 主进程退出时不会卡住
Step 3: API调用实现
直接用标准库，不依赖额外包：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| def _run_extraction(self, conversation_text, existing_context=''):
import urllib.request, urllib.error
system_prompt = self.EXTRACTION_SYSTEM_PROMPT
if existing_context:
system_prompt += f'\n\n已知记忆（不要重复提取这些内容）：\n{existing_context}'
payload = json.dumps({
'model': 'glm-4-flash', # 免费模型
'messages': [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': conversation_text},
],
'max_tokens': 800,
'temperature': 0.1,
}, ensure_ascii=False).encode('utf-8')
req = urllib.request.Request(
'https://open.bigmodel.cn/api/paas/v4/chat/completions',
data=payload, method='POST'
)
req.add_header('Authorization', f'Bearer {self._extract_api_key}')
req.add_header('Content-Type', 'application/json')
try:
with urllib.request.urlopen(req, timeout=30) as resp:
result = json.loads(resp.read().decode('utf-8'))
except (urllib.error.URLError, json.JSONDecodeError, KeyError) as e:
return # 静默失败，不影响主流程
content = result['choices'][0]['message']['content']
extracted = self._parse_extraction_json(content)
self._store_extracted_memories(extracted)
|
JSON解析加个容错：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import re
def _parse_extraction_json(self, content):
"""容错解析，处理LLM输出前后可能的多余文本"""
try:
return json.loads(content)
except json.JSONDecodeError:
# 尝试提取花括号内容
match = re.search(r'\{[^{}]*\}', content, re.DOTALL)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
return {'persona': [], 'episodic': [], 'instruction': []}
|
Step 4: 向量去重（P1）
这是防重复的最后防线。即使LLM没看到已有记忆，向量相似度也能拦住：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
| _DEDUP_MAX_DISTANCE = 0.15 # ~92%相似度
def _is_duplicate(self, text, wing):
results = self._col.query(
query_texts=[text],
n_results=3,
where={'wing': wing},
include=['distances'],
)
distances = results.get('distances', [[]])[0]
return distances and distances[0] < self._DEDUP_MAX_DISTANCE
def _store_extracted_memories(self, extracted):
import time
stored = 0
for category, items in extracted.items():
wing_map = {
'persona': 'user',
'episodic': 'episodic',
'instruction': 'instruction'
}
wing = wing_map.get(category, 'user')
for item in items:
if not item or len(item.strip()) < 5:
continue # 跳过太短的
if self._is_duplicate(item, wing):
continue # 跳过重复
self._col.add(
documents=[item],
metadatas=[{
'wing': wing,
'room': 'auto-extract',
'added_by': 'auto-extract',
'category': category,
'timestamp': str(int(time.time())),
}],
ids=[f'extract_{category}_{int(time.time()*1000)}_{stored}']
)
stored += 1
|
为什么阈值是0.15：
实测数据——
| 距离阈值 | 相似度 | 去重效果 | 误杀率 |
|---|
| 0.10 | ~95% | 去重80% | 极低 |
| 0.15 | ~92% | 去重90%+ | <3% |
| 0.20 | ~88% | 去重95% | ~8% |
0.15是甜点。
Step 5: 已知记忆注入（P2）
在调LLM之前，把已有记忆塞进prompt。这样模型直接跳过已知内容，省token又防重复：
1
2
3
4
5
6
7
8
9
10
11
12
| def _build_existing_context(self):
parts = []
for wing in ('user', 'instruction'):
results = self._col.get(
where={'wing': wing},
include=['documents', 'metadatas'],
limit=20
)
for doc, meta in zip(results['documents'], results['metadatas']):
if meta.get('added_by') in ('auto-extract', 'hermes', 'builtin-mirror'):
parts.append(doc[:100])
return '；'.join(parts[:15])[:500] if parts else ''
|
只取最近15条、截断到500字符。控制prompt长度，别把提取请求搞得太贵。
Step 6: 配置和测试
在 mempalace.json 中启用自动提取：
1
2
3
4
| {
"auto_extract": true,
"extract_model": "glm-4-flash"
}
|
验证方法
测试1：基本提取
跟Agent说几句话：
1
2
3
| 用户：我习惯用vim编辑器，不要给我推荐nano
用户：我的项目统一用Python 3.11
用户：代码风格遵循PEP8，docstring用Google风格
|
结束session，然后检查ChromaDB：
1
2
3
4
5
6
| import chromadb
client = chromadb.PersistentClient(path='./mempalace_db')
col = client.get_collection('memories')
results = col.get(where={'added_by': 'auto-extract'}, include=['documents', 'metadatas'])
for doc, meta in zip(results['documents'], results['metadatas']):
print(f"[{meta['category']}] {doc}")
|
预期输出类似：
1
2
3
| [persona] 用户习惯使用vim编辑器
[instruction] 项目统一使用Python 3.11
[instruction] 代码风格遵循PEP8，docstring用Google风格
|
测试2：去重验证
同一段对话连续触发两次提取，第二次应该0条新增。
测试3：新session加载
开启新对话，问Agent：“你知道我用什么编辑器吗？"——它应该能从记忆中回答"vim”。
实测效果数据
| 指标 | 数据 |
|---|
| 提取延迟 | 3-7秒（后台异步，不阻塞） |
| Token消耗 | 300-500/次 |
| 成本 | ¥0（GLM-4-Flash免费） |
| 分类准确率 | >90%（persona/episodic/instruction） |
| 去重拦截率 | 90%+（重复对话） |
| 误存率 | <2%（无效信息被存入） |
三层防护机制总结
再说一遍这个设计，因为它是整个系统可靠性的关键：
1
2
3
| P2 提取前：已知记忆注入 → LLM直接跳过 → 省token
P0 提取中：结构化分类 → 只保留有价值信息 → 控质量
P1 存入时：向量相似度去重 → 兜底拦截 → 防重复
|
三层不是冗余，是互补：
- P2漏了（记忆太多超过500字符截断了）→ P1兜底
- P1漏了（措辞差异太大向量没拦住）→ P2已经拦了大部分
- P0是核心 → 决定提取质量
一句话总结
自动记忆不是奢侈品，是Agent的基础设施。当AI能记住你说过的每一句偏好，你就不需要每次对话都从头开始。