给长跑智能体搭两层对话记忆：压实、衰减与强化

跑得久的智能体都有一个共同的毛病：前二十轮它还记着“合同编号是 KX-7731”，到第六十轮再问，它开始编一个编号出来。原因往往不是模型笨，而是上下文里塞满了最近几十轮寒暄，真正关键的那句话要么被截断丢掉，要么被一堆相似的历史片段淹没了。

这篇给出一套可以照抄的两层记忆结构：短期窗口负责最近几轮的原话，长期记忆库负责跨轮次的关键信息。窗口涨到阈值就自动压实成摘要落到长期库；长期库条目按“相似度 + 强化分 + 新近度”召回，被召回一次就加分；长期没被用到、强化分掉到底线的条目自动归档淘汰。最后用一个多轮任务脚本，把两层方案和纯 RAG 记忆方案放在一起，对比关键信息留存率。

这套东西能做出什么

照着做完，你会得到一个能离线跑起来的四文件小工程，具备下面四项行为：

1. 自动压实：短期窗口的估算 token 超过阈值时，最老的一段对话被压成一条摘要写进长期库，窗口水位回落到阈值的一半左右，智能体继续干活不报错。

2. 召回强化：每条长期记忆有强度分。被检索命中一次 +0.5；真正被引用进回答再加 1.0。用得越多越难沉底。

3. 衰减淘汰：强度分按半衰期随时间打折，低于底线的条目被归档，不再进入检索结果，库不会无限膨胀。

4. 量化对比：一个 60 轮的任务脚本，同一批查询分别走两层结构和纯 RAG，输出一张留存率对照表，让你看清差别在哪。

代码全程只用 Python 标准库，SQLite 存长期记忆，不需要AI 词典：向量数据库">向量数据库也能跑通。想换成 embedding 检索，只需要替换一个函数。

前置条件清单

动手前确认这几件事：

• Python 3（建议 3.8 以上，具体以官方文档当前版本为准），只用 sqlite3、math、re、time、dataclasses 这些标准库。
• 一个可写目录，长期库存成 memory.db 单文件即可。
• 一个能调 LLM 的接口（可选）。不接也能跑：本文提供一个规则版摘要器兜底，保证离线可复现。接真模型时，把调用函数传进 compact() 即可。
• 知道自己模型的上下文窗口大小。短期窗口阈值一般设为窗口的 1/4 到 1/3，具体以所用模型官方文档当前版本为准。
• 接受 token 是估算的。下面的 estimate_tokens() 按“中文 1 字约 1 token、英文 4 字符约 1 token”粗算，够用且不依赖分词库。

步骤一：先定数据模型

两层记忆的本质不是两个 prompt，而是两套生命周期不同的存储。短期窗口是易失的、按轮次滚动的；长期库是持久的、需要打分的。所以第一件事是把长期库的表结构定下来。

```sql

CREATE TABLE IF NOT EXISTS memories (

id INTEGER PRIMARY KEY AUTOINCREMENT,

kind TEXT NOT NULL, -- preference / decision / fact / todo / digest / chunk

content TEXT NOT NULL,

tokens TEXT NOT NULL, -- 分词结果，空格拼接，用于关键词召回

salience REAL NOT NULL DEFAULT 0.5, -- 初始重要度 0~1

strength REAL NOT NULL DEFAULT 1.0, -- 强化分，命中后累加

hits INTEGER NOT NULL DEFAULT 0,

created_at REAL NOT NULL, -- 写入时间戳

last_hit_at REAL, -- 最近一次被召回的时间戳

pinned INTEGER NOT NULL DEFAULT 0, -- 保护位，跳过淘汰

archived INTEGER NOT NULL DEFAULT 0 -- 归档位，不参与检索

);

CREATE INDEX IF NOT EXISTS idx_kind ON memories(kind);

```

kind 字段决定了衰减速度：用户偏好可以放很久，待办事项两三天就该退场。pinned 用于保护“这条不能删”，比如用户明确说过的硬性约束。

步骤二：短期窗口与自动压实

短期窗口要做两件事：算自己有多满，以及在溢出时把最老的一段交出去。

```python

window.py

import re

from dataclasses import dataclass

MARKERS = ("必须", "不要", "禁止", "截止", "决定", "约定", "编号",

"金额", "负责人", "上限", "格式", "偏好")

def estimate_tokens(text: str) -> int:

cjk = len(re.findall(r"[\u4e00-\u9fff]", text))

other = len(text) - cjk

return cjk + max(1, other // 4)

@dataclass

class Turn:

role: str

content: str

class ShortTermWindow:

def __init__(self, max_tokens=1200, keep_recent=6):

self.max_tokens = max_tokens

self.keep_recent = keep_recent

self.turns = []

def add(self, role, content):

self.turns.append(Turn(role, content))

def tokens(self):

return sum(estimate_tokens(t.content) for t in self.turns)

def overflow(self):

return self.tokens() > self.max_tokens

def take_for_compaction(self):

"""从最老的轮次开始取，直到剩余水位降到阈值一半以下。"""

if len(self.turns) <= self.keep_recent:

return []

cut = 0

while cut < len(self.turns) - self.keep_recent:

rest = sum(estimate_tokens(t.content) for t in self.turns[cut:])

if rest <= self.max_tokens * 0.5:

break

cut += 1

taken = self.turns[:cut]

self.turns = self.turns[cut:]

return taken

def render(self):

return "\n".join(f"{t.role}: {t.content}" for t in self.turns)

```

压实这一步最容易出事：摘要一写虚，关键信息就永久丢了。所以摘要提示词要把“逐条照抄关键值”写死。

```python

COMPACT_PROMPT = """你在做对话历史的压实。把下面的历史记录压成不超过 150 字的中文摘要。

必须保留（逐条照抄关键值，不要改写数字、编号、人名、时间）：

1. 已经定下的决定和结论

2. 硬性约束、禁止项、格式要求

3. 具体数字：金额、编号、日期、阈值、联系人

4. 尚未完成的待办和它的负责人

不要推测，不要评价，不要写"用户表示"这类空话。用短横线分条输出。

历史记录：

{history}

"""

def compact(turns, llm=None):

history = "\n".join(f"{t.role}: {t.content}" for t in turns)

keep_lines = [t.content for t in turns if any(m in t.content for m in MARKERS)]

if llm is not None:

content = llm(COMPACT_PROMPT.format(history=history)).strip()

else:

规则兜底：把带标记词的整句原样留下来，保证离线也能复现实验

content = "；".join(keep_lines)[:600] or history[:300]

return {"kind": "digest", "content": "【历史摘要】" + content, "salience": 0.6}

```

规则兜底版本故意“照抄整句”，听起来笨，但它保证了下文的留存率对比不被摘要质量干扰——你要测的是结构，不是摘要模型。

步骤三：长期库的写入与惰性衰减

长期库有两个设计选择值得说明。

其一是惰性衰减：不跑定时任务去改库，而是在读的时候按“距上次命中了多久”现算当前强度。好处是即使定时任务挂了、进程重启了，算出来的分数依然正确，不需要补算。

其二是半衰期按类型分档：偏好类的记忆衰减慢，待办类衰减快。

```python

memory.py

import math

import re

import sqlite3

import time

SCHEMA = open("schema.sql", encoding="utf-8").read()

HALF_LIFE_DAYS = {

"preference": 60,

"decision": 30,

"fact": 14,

"todo": 2,

"digest": 10,

"chunk": 10,

}

def tokenize(text: str):

"""英文按词、中文按单字 + 相邻二元组，不依赖第三方分词库。"""

text = text.lower()

words = re.findall(r"[a-z0-9_]+", text)

cjk = re.findall(r"[\u4e00-\u9fff]", text)

bigrams = [a + b for a, b in zip(cjk, cjk[1:])]

return words + cjk + bigrams

class MemoryStore:

def __init__(self, path="memory.db"):

self.conn = sqlite3.connect(path)

self.conn.row_factory = sqlite3.Row

self.conn.execute("PRAGMA journal_mode=WAL")

self.conn.execute("PRAGMA busy_timeout=3000")

self.conn.executescript(SCHEMA)

self.conn.commit()

def add(self, kind, content, salience=0.5, pinned=0, now=None):

now = now if now is not None else time.time()

cur = self.conn.execute(

"INSERT INTO memories(kind, content, tokens, salience, strength,"

" created_at, last_hit_at, pinned) VALUES (?,?,?,?,?,?,?,?)",

(kind, content, " ".join(tokenize(content)), salience, 1.0, now, None, pinned),

)

self.conn.commit()

return cur.lastrowid

def _decayed(self, row, now):

half_life = HALF_LIFE_DAYS.get(row["kind"], 14) * 86400

ref = row["last_hit_at"] or row["created_at"]

dt = max(0.0, now - ref)

return row["strength"] * math.exp(-math.log(2) * dt / half_life)

```

半衰期的具体天数没有标准答案，按你的业务节奏定：如果一个项目通常跑两周，decision 给 30 天就够宽松；如果是一次性排障，fact 给 7 天更合适。

步骤四：召回打分——相似度、强化、新近度

只按相似度排序的检索有个通病：库里同类表述越多，越容易互相“霸榜”，反复返回几条几乎一样的旧片段。加入强化分和新近度能缓解：真正被用过、被确认过的条目排在前面。

```python

def search(self, query, k=6, now=None, reinforce=True, bonus=0.5,

weights=(0.6, 0.3, 0.1)):

now = now if now is not None else time.time()

rows = self.conn.execute("SELECT * FROM memories WHERE archived = 0").fetchall()

q = set(tokenize(query))

if not q or not rows:

return []

df = {}

for r in rows:

toks = set(r["tokens"].split())

for t in q & toks:

df[t] = df.get(t, 0) + 1

n = len(rows)

idf = {t: math.log(1.0 + n / (1.0 + df.get(t, 0))) for t in q}

norm = sum(idf.values()) or 1.0

w_sim, w_reinf, w_rec = weights

scored = []

for r in rows:

toks = set(r["tokens"].split())

sim = sum(idf[t] for t in q & toks) / norm

if sim <= 0:

continue

s = self._decayed(r, now)

reinf = s / (s + 1.0) # 压到 0~1

ref = r["last_hit_at"] or r["created_at"]

recency = math.exp(-(now - ref) / (7 * 86400))

score = w_sim * sim + w_reinf * reinf + w_rec * recency

if r["pinned"]:

score += 0.2 # 保护位加权，但不至于压过强相关

scored.append((score, r))

scored.sort(key=lambda x: x[0], reverse=True)

top = scored[:k]

if reinforce and top:

for _, r in top:

self.reinforce(r["id"], now=now, bonus=bonus)

return top

```

三个权重是这套结构唯一需要调的东西。相似度占大头保证相关性，强化分占三成保证“验证过的信息优先”，新近度占一成只用来打破同分僵局。如果你的场景里用户偏好比当前话题更重要，把第二项提到 0.4 试试。

步骤五：命中即强化

强化分要有区分度：被检索到和被真正用上是两件事。前者加 0.5，后者再加 1.0。同时设个上限，防止某条记忆被反复命中后锁死前排。

```python

def reinforce(self, mid, now=None, bonus=0.5):

now = now if now is not None else time.time()

row = self.conn.execute("SELECT * FROM memories WHERE id = ?", (mid,)).fetchone()

if row is None:

return

s = min(self._decayed(row, now) + bonus, 10.0)

self.conn.execute(

"UPDATE memories SET strength = ?, hits = hits + 1, last_hit_at = ? WHERE id = ?",

(s, now, mid),

)

self.conn.commit()

```

在生成回答之后，判断哪些记忆被真正引用（文本里出现了它的内容片段，或模型在引用列表里点名），对这些 id 再调一次 reinforce(mid, bonus=1.0)。

步骤六：衰减与淘汰

衰减负责“变弱”，淘汰负责“清理”。清理只在扫库时做，检索路径上不做删除，避免把正在用的条目删掉。

```python

def sweep(self, now=None, floor=0.05, min_age_days=1.0,

keep_kinds=("preference",)):

now = now if now is not None else time.time()

rows = self.conn.execute("SELECT * FROM memories WHERE archived = 0").fetchall()

killed = 0

for r in rows:

if r["pinned"] or r["kind"] in keep_kinds:

continue

if (now - r["created_at"]) / 86400 < min_age_days:

continue

if self._decayed(r, now) < floor:

self.conn.execute("UPDATE memories SET archived = 1 WHERE id = ?", (r["id"],))

killed += 1

self.conn.commit()

return killed

def stats(self):

row = self.conn.execute(

"SELECT COUNT(*) AS total, SUM(archived = 0) AS alive,"

" SUM(archived = 1) AS archived FROM memories"

).fetchone()

return {"total": row["total"], "alive": row["alive"] or 0,

"archived": row["archived"] or 0}

```

三个保护条件缺一不可：pinned 挡住用户明确要求保留的内容；keep_kinds 让偏好类不进淘汰；min_age_days 防止刚写入、还没来得及被命中的新条目当天就被清掉。

步骤七：组装上下文

窗口和长期库不是拼在一起就完事，顺序影响模型注意力。稳妥的排法是：长期记忆在前（作为背景约束）、最近对话在中（作为即时语境）、当前输入在最后。

```python

def build_context(window, store, user_input, k=6):

hits = store.search(user_input, k=k)

mem_block = "\n".join(f"- [{r['kind']}] {r['content']}" for _, r in hits)

return (

"## 长期记忆（跨轮次，需遵守）\n" + (mem_block or "（无）") +

"\n\n## 最近对话\n" + window.render() +

"\n\n## 当前输入\n" + user_input

), hits

```

给每条长期记忆打上 [kind] 标签，模型对“这是历史约定还是当前指令”的区分会清楚很多。

步骤八：跑一个 60 轮的任务脚本

评测的关键是埋点：在脚本的固定位置放入带唯一标记的关键信息，跑完再看这些标记还在不在上下文里。

```python

eval_retention.py

import time

from window import ShortTermWindow, compact

from memory import MemoryStore

DAY = 86400

FILLER = [

("user", "先过一下今天的进度"),

("assistant", "三条线都在推进：数据侧已交付，接口侧联调中，文档侧待补。"),

("user", "预算这块下周再核一次"),

("assistant", "好，下周核对时把变更项单独列出来。"),

]

PLANT = {

2: "记一下，这个项目的合同编号是 KX-7731，后面所有对账都用它。",

8: "同步会议固定在每周三 10:00，不要改。",

18: "预算上限 4800 元，超了要重新走审批。",

26: "周报里不要用表格，用分条。",

40: "这块的负责人是林岚，找她对接口径。",

}

QUERIES = {

"KX-7731": "合同编号是多少",

"每周三 10:00": "同步会议安排在什么时间",

"上限 4800 元": "预算上限是多少",

"不要用表格": "周报的格式要求",

"负责人 林岚": "这块的负责人是谁",

}

def build_script(turns=60):

script = []

for i in range(turns):

script.append(("user", PLANT[i]) if i in PLANT else FILLER[i % len(FILLER)])

return script

def retention(context, marker):

return 1.0 if marker.lower() in context.lower() else 0.0

def run_two_layer(script, k=6):

store = MemoryStore("two_layer.db")

window = ShortTermWindow(max_tokens=800, keep_recent=6)

for role, content in script:

window.add(role, content)

while window.overflow():

turns = window.take_for_compaction()

if not turns:

break

mem = compact(turns, llm=None)

store.add(mem["kind"], mem["content"], salience=mem["salience"])

rows = []

for marker, q in QUERIES.items():

hits = store.search(q, k=k)

ctx = ("## 长期记忆\n" + "\n".join(f"- {r['content']}" for _, r in hits)

+ "\n## 最近对话\n" + window.render())

rows.append((marker, q, retention(ctx, marker), len(ctx)))

return rows, store

def run_pure_rag(script, k=6):

store = MemoryStore("pure_rag.db")

for role, content in script:

store.add("chunk", f"{role}: {content}")

rows = []

for marker, q in QUERIES.items():

hits = store.search(q, k=k, reinforce=False, weights=(1.0, 0.0, 0.0))

ctx = "\n".join(r["content"] for _, r in hits)

rows.append((marker, q, retention(ctx, marker), len(ctx)))

return rows, store

if __name__ == "__main__":

script = build_script()

two, store = run_two_layer(script)

rag, _ = run_pure_rag(script)

print(f"{'关键信息':<16}{'两层':<8}{'纯RAG':<8}")

for (m, q, a, _), (_, _, b, _) in zip(two, rag):

print(f"{m:<16}{a:<8.2f}{b:<8.2f}")

avg_two = sum(r[2] for r in two) / len(two)

avg_rag = sum(r[2] for r in rag) / len(rag)

print(f"\n平均留存率：两层 {avg_two:.2f} / 纯 RAG {avg_rag:.2f}")

print(f"平均上下文长度：两层 {sum(r[3] for r in two)//len(two)} 字符 / "

f"纯 RAG {sum(r[3] for r in rag)//len(rag)} 字符")

now = time.time()

store.add("todo", "待办：把返点明细补进对账表", salience=0.4)

print("\n30 天后扫库归档条数：", store.sweep(now=now + 30 * DAY))

print("库状态：", store.stats())

```

跑法就是先删掉旧的 *.db，再执行：

```bash

rm -f two_layer.db pure_rag.db memory.db

python eval_retention.py

```

步骤九：怎么看结果

输出会是类似下面这样的表（数值随脚本内容、阈值、top-k 变化，不构成通用结论）：

重点看三件事：

• 纯 RAG 的漏检集中在早期信息。第 2 轮埋的编号，到第 60 轮时被几十条格式相近的对话盖住，top-k 里挤不进去。
• 两层方案的漏检通常来自压实。某条信息如果没被摘要保留，后面再检索也找不回来——这正是 COMPACT_PROMPT 要写死“照抄数字和编号”的原因。
• 上下文长度。纯 RAG 为了捞回早期信息，往往要放大 k，token 成本跟着涨；两层方案用一条摘要换掉十几轮对话，长度更可控。

再往下看衰减的效果：脚本最后新建的待办半衰期是 2 天，模拟 30 天后扫库，它会被归档；而 digest 类半衰期 10 天、又经过多轮命中的条目会留下来。这说明“用得多”确实能换来更长的存活时间。

常见坑与排错

坑一：压实之后关键信息凭空消失。

症状是早期埋点全部漏检。原因通常是摘要提示词只说“总结一下”，模型就顺手把编号、金额这些“细节”抹平了。修法是把“逐条照抄数字、编号、人名、时间”写成硬约束，并让摘要保留待办的负责人。

坑二：反复压实导致摘要漂移。

摘要被再次压实、再被压实，几轮之后语义变形。修法是给 digest 类条目加代数标记，超过两代就不再参与压实，或把摘要只当作索引、原始片段仍留档。

坑三：衰减太快，把长期偏好删了。

症状是用户说了三次“不要用表格”，第七天智能体又开始输出表格。检查 HALF_LIFE_DAYS 里 preference 的取值，以及 sweep 的 keep_kinds 有没有把它排除在外。

坑四：强化分锁死前排。

某条被反复命中的记忆占据 top-1，新信息挤不进来。把 reinforce 的上限从 10.0 降到 5.0，或者提高相似度权重。

坑五：新写入的条目当天被清掉。

忘了设 min_age_days。刚写进库、还没被任何查询命中的条目强度是 1.0，但某些类型的半衰期只有 2 天，跨一个周末就掉到阈值下。

坑六：SQLite 多进程写入报 locked。

长跑智能体常有后台扫库任务和前台对话同时写。加上 PRAGMA journal_mode=WAL 和 PRAGMA busy_timeout=3000，把写操作收拢到单线程队列里。

坑七：token 估算偏差过大导致上下文超限。

粗估函数在长英文段落上会低估。保守做法是把 max_tokens 设在窗口的四分之一，并在组装上下文之前再算一次总长度，超了就临时缩小 k。

坑八：对照组不公平。

拿纯 RAG 的 k=3 去打两层的 k=8，结论没意义。两组用同一个 k、同一套分词、同一批查询。

坑九：把敏感信息写进了长期库。

长期库会跨会话留存。写入前做一次敏感字段过滤，或给 kind='sensitive' 的条目设 pinned=0 且短半衰期。

下一步建议

• 把关键词检索换成向量检索。tokenize 和 search 里的相似度部分换成 embedding 余弦相似度，其余打分、强化、衰减逻辑完全不动。embedding 模型选型以官方文档当前版本为准。
• 给压实加一次“事实抽取”。压实的同时让模型输出结构化的 {kind, content, salience} 列表，把一条摘要拆成若干原子记忆，检索粒度更细。
• 把强化信号接到真实反馈上。用户点了“对”、任务成功结束、人工确认，都可以作为 bonus 的来源，比“被检索到就加分”更准。
• 给长期库加版本与溯源。每条记忆记录来源轮次，出问题时能回溯到原始对话，排查“这条错误约定是谁什么时候说的”。
• 做一次 A/B 上线。同一个智能体，一半会话开强化衰减、一半关掉，比较任务完成率和人工返工率。留存率只是中间指标，最终还是要看任务有没有做完。