11.1 写入流水线：抽取、去重、冲突消解
第一卷《AI Agent 开发入门》8.3 节给 minagent 加过一套写入策略：should_write 过滤寒暄、dedup 规范化去重、write 按槽位覆盖。那套策略跑在一个内存列表上，够讲清原理，但还不是一个服务——它没有并发、没有持久化、没有幂等，进程一退全没了。ResearchCopilot 是多人多租户的服务，写入必须经得起重放、并发和失败重试。
本节把 ResearchCopilot 推进到「记忆写入服务」：把写入拆成抽取 → 去重 → 冲突消解三段流水线，真连本机 Postgres 落库，并用同一条对话重放两次来验证幂等性。
11.1.1 为什么写入要做成一条流水线
最省事的写法是「模型说一句值得记的话，就 INSERT 一行」。这在演示里没问题，上线后立刻出事：同一条消息因为队列重试被消费两次，库里就有两行；用户换了说法讲同一件事，检索时召回两条几乎一样的记忆；用户改口说「我不用 Python 了」，新行和旧行同时 active，模型读到互相矛盾的两条。
所以写入必须是一条有阶段、有判定、可重放的流水线：
| 阶段 | 解决的问题 | 关键动作 |
|---|---|---|
| 抽取 | 从哪来 | 从对话轮次里挑出候选记忆 |
| 去重 | 别存两遍 | 文本归一 + 向量相似度两档判定 |
| 冲突消解 | 谁赢 | 时间戳定胜负，旧行保留历史并标记被取代 |
| 落库 | 可重放 | 幂等键 + 唯一约束，重放不产生新行 |
三段里，「抽取」是产生候选，「去重 / 冲突消解」是裁决候选，落库是固化裁决结果。之所以强调「异步」——真实系统里这条流水线通常挂在 run 结束之后由 worker 消费，而不是卡在用户请求的同步路径上；但它的判定逻辑与是否异步无关，本节就在同一个进程里把它跑通。
11.1.2 第一段：抽取候选记忆
抽取的输入是一段对话轮次，输出是零到多条候选记忆。生产系统常用一次轻量模型调用来做抽取（「这段对话里有没有关于用户的长期信息？抽成一句」），本节用规则抽取，因为它确定、可断言，且足以演示后两段。
# memory/pipeline.py
import re
WRITE_MARKERS = ("我喜欢", "我偏好", "我习惯", "我平时", "我在做", "我负责", "记住")
def extract(turn: dict) -> list[dict]:
"""规则抽取：只从 user 轮次里挑出带偏好/约束标记的句子。"""
if turn["role"] != "user":
return []
out = []
for sent in re.split(r"[。！？\n]", turn["content"]):
s = sent.strip()
if s and any(m in s for m in WRITE_MARKERS):
out.append({"slot": guess_slot(s), "content": s, "ts": turn["ts"],
"source_turn_id": turn["turn_id"]})
return out
两条纪律：只从 user 轮次抽取（模型自己的回复里混着客套与幻觉，不是一手事实），按标点切句（一句话里可能同时有寒暄和偏好，整条丢弃会误伤）。slot（槽位）是这条记忆的主题标签，guess_slot 用一张关键词表把句子归到「语言 / 回答风格 / 项目」等槽位——它正是冲突消解阶段判断「这两条说的是不是同一件事」的依据。
SLOT_RULES = [("语言", ("python", "typescript", "go ", "golang", "java")),
("回答风格", ("简洁", "简短", "详细", "示例")),
("项目", ("在做", "负责", "知识库", "检索", "报告"))]
def guess_slot(text: str) -> str:
low = text.lower()
for slot, kws in SLOT_RULES:
if any(k in low for k in kws):
return slot
return "其它"
11.1.3 第二段：两档去重
去重分两档，先便宜后昂贵。
第一档：文本归一化去重。 把大小写、空白、标点统一抹掉再比字符串。用户把同一句话说了三遍（带句号、不带句号、前后带空格），归一后是同一个键：
_PUNCT = re.compile(r"[\s，。！？、,.!?;；:：\"'（）()【】\[\]]+")
def normalize(text: str) -> str:
return _PUNCT.sub("", text.lower())
第二档：向量相似度去重。 归一化只抓字面重复，抓不到「我偏好简洁示例」和「回答请短一点」是同义。这一档给每条记忆算一个向量，同槽位内余弦相似度超过阈值就判为重复：
import hashlib
import numpy as np
def embed(text: str, dim: int = 128) -> np.ndarray:
"""确定性本地向量：字符 bigram 哈希到固定维度，无需任何模型。"""
t = normalize(text)
v = np.zeros(dim, dtype=np.float32)
for g in (t[i:i + 2] for i in range(max(len(t) - 1, 1))):
v[int(hashlib.md5(g.encode()).hexdigest(), 16) % dim] += 1.0
n = np.linalg.norm(v)
return v / n if n else v
def cosine(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b))
DUP_COS = 0.92 # 向量相似度去重阈值
def load_active(conn, ws: str) -> list[dict]:
"""读出该租户当前生效的记忆，供去重/冲突比对。"""
return conn.execute(
"select id, slot, content, content_norm, embedding, updated_at "
"from mem11_memories where workspace_id=%s and status='active'",
(ws,)).fetchall()
这里刻意用了哈希 bigram 这种离线向量，而不是调用嵌入模型：本节的目的是演示流水线的判定结构，判定用的向量换成任何真嵌入都不改变结构，但离线方案让整节可以零密钥、可复现地跑。第 6.2 节讲语义缓存时会换成真正的嵌入向量。两档的取舍：
| 档位 | 抓什么 | 成本 | 会漏什么 |
|---|---|---|---|
| 文本归一 | 标点/大小写/空白差异 | 极低 | 同义不同字 |
| 向量相似 | 语义近似 | 中（要算向量） | 阈值附近的边界情况 |
11.1.4 第三段：冲突消解
去重处理的是「同一件事被说两遍」，冲突处理的是「同一件事被说成了两个样子」：用户上周说用 Python，这周改用 TypeScript。两条不能都 active，否则检索会召回矛盾信息。策略是三条一起上：
- 时间戳定胜负——
updated_at更大的赢，防止乱序写入把新信息盖成旧的。 - 保留历史——旧行不删，只把
status从active改成superseded，并记下superseded_by指向新行。审计、回溯、模型解释都需要这条链。 - 标记取代——同一槽位同一时刻只允许一条
active，取代动作在一个事务里完成。
def write_one(conn, ws: str, cand: dict) -> str:
vec = embed(cand["content"])
norm = normalize(cand["content"])
for old in load_active(conn, ws):
old_vec = np.array(old["embedding"], dtype=np.float32)
same_text = old["content_norm"] == norm
similar = old["slot"] == cand["slot"] and cosine(vec, old_vec) >= DUP_COS
if same_text or similar:
return f"去重({('文本归一' if same_text else '向量%.2f' % cosine(vec, old_vec))})，忽略"
row = conn.execute(
"insert into mem11_memories "
"(workspace_id, slot, content, content_norm, embedding, source_turn_id, updated_at) "
"values (%s,%s,%s,%s,%s,%s,%s) on conflict do nothing returning id",
(ws, cand["slot"], cand["content"], norm, vec.tolist(),
cand["source_turn_id"], cand["ts"])).fetchone()
if row is None: # 幂等：同轮同文本重放，命中唯一约束
return "幂等命中（同轮同文本已存在），忽略"
new_id = row["id"]
old = conn.execute(
"select id, content from mem11_memories "
"where workspace_id=%s and slot=%s and status='active' and id<>%s "
"and updated_at < %s order by updated_at desc limit 1",
(ws, cand["slot"], new_id, cand["ts"])).fetchone()
if old:
conn.execute("update mem11_memories set status='superseded', superseded_by=%s "
"where id=%s", (new_id, old["id"]))
return f"冲突消解：新事实取代 slot={cand['slot']}（旧:{old['content']} -> 新:{cand['content']}）"
return f"新增 slot={cand['slot']}: {cand['content']}"
注意「去重」判定排在「插入」之前：先去重能省掉一次插入尝试，也避免让同义句互相触发冲突消解。
11.1.5 幂等性：同一条对话重放两次
流水线必须幂等：同一条对话被消费两次，结果和一次完全相同。这靠幂等键实现——每条记忆带上它来自哪个对话轮次 source_turn_id，表上建唯一约束 (workspace_id, source_turn_id, content_norm)：
create table if not exists mem11_memories (
id bigserial primary key,
workspace_id text not null,
slot text not null,
content text not null,
content_norm text not null,
embedding real[] not null,
status text not null default 'active', -- active | superseded
superseded_by bigint,
source_turn_id text not null, -- 幂等键
updated_at bigint not null,
unique (workspace_id, source_turn_id, content_norm)
);
为什么要专门为「重放」设计？因为异步流水线天然会重放：worker 消费完消息、准备 ack 时进程挂了，消息会被重新投递；网络抖动让发布方重发；人工补录也可能把同一段对话再喂一遍。只要流水线在任何一次重放下都收敛到同一状态，这些都不需要特判。
两条防线缺一不可。第一道是去重逻辑：重放时，load_active 看到已存在的行，文本归一直接判重复。但第一道有个漏洞——如果旧行已经被 superseded，load_active 就看不到它，去重会漏判，这时第二道 ON CONFLICT DO NOTHING 兜底，靠数据库的唯一约束把重放挡在门外。所以插入用的是 on conflict do nothing returning id，row is None 就意味着「这条已经落过库了」。
11.1.6 真跑：5 条候选走完整条流水线
真连本机 Postgres 跑一遍。若本机没有该容器，先起一个：
docker run -d --name ab-pg -e POSTGRES_PASSWORD=postgres \
-e POSTGRES_DB=research -p 15435:5432 postgres:17-alpine
连不上先 docker ps | grep ab-pg 确认容器在跑。5 个对话轮次（其中 t3 是 t2 的重复、t5 与 t1 冲突）走完整条流水线：
# demo_11_1.py（节选）
TURNS = [
{"turn_id": "t1", "role": "user", "ts": 100, "content": "我喜欢用 Python 写 Agent"},
{"turn_id": "t2", "role": "user", "ts": 110, "content": "我偏好简洁的代码示例"},
{"turn_id": "t3", "role": "user", "ts": 120, "content": "我偏好简洁的代码示例。"}, # 重复
{"turn_id": "t4", "role": "user", "ts": 130, "content": "我在做本地知识库问答项目"},
{"turn_id": "t5", "role": "user", "ts": 140, "content": "我偏好改用 TypeScript 写 Agent"}, # 冲突
]
本机实测（Python 3.14.6 / psycopg 3.3.6 / PostgreSQL 17.11）输出：
== 第一遍写入 ==
[t1] 我喜欢用 Python 写 Agent -> 新增 slot=语言: 我喜欢用 Python 写 Agent
[t2] 我偏好简洁的代码示例 -> 新增 slot=回答风格: 我偏好简洁的代码示例
[t3] 我偏好简洁的代码示例 -> 去重(文本归一)，忽略
[t4] 我在做本地知识库问答项目 -> 新增 slot=项目: 我在做本地知识库问答项目
[t5] 我偏好改用 TypeScript 写 Age -> 冲突消解：新事实取代 slot=语言（旧:我喜欢用 Python 写 Agent -> 新:我偏好改用 TypeScript 写 Agent）
== 重放同一段对话（幂等验证）==
[t1] 我喜欢用 Python 写 Agent -> 幂等命中（同轮同文本已存在），忽略
[t2] 我偏好简洁的代码示例 -> 去重(文本归一)，忽略
[t3] 我偏好简洁的代码示例 -> 去重(文本归一)，忽略
[t4] 我在做本地知识库问答项目 -> 去重(文本归一)，忽略
[t5] 我偏好改用 TypeScript 写 Age -> 去重(文本归一)，忽略
== 最终 active 记忆 ==
slot=语言 status=superseded sup_by=4 我喜欢用 Python 写 Agent
slot=回答风格 status=active sup_by=None 我偏好简洁的代码示例
slot=项目 status=active sup_by=None 我在做本地知识库问答项目
slot=语言 status=active sup_by=None 我偏好改用 TypeScript 写 Agent
总行数=4 active=3
逐行读：第一遍里 t3 被判文本重复，t5 触发冲突消解，把 语言 槽的 Python 行标记为 superseded、superseded_by=4 指向新的 TypeScript 行。重放时 t1 命中的是数据库幂等那条路径（幂等命中）——因为 Python 行已被 superseded，load_active 看不到它，去重逻辑漏判，是唯一约束兜住了这次重放。最终库里 4 行、3 条 active：被取代的历史还在（审计需要），但检索只会看到 3 条当前事实。
11.1.7 真实环境与验收清单
真实系统与本节演示的差距集中在三处：抽取换成模型调用（更灵活，但结果不再确定，需要采样评测）；向量换成真嵌入（同义判定准得多，代价是要维护嵌入服务与索引）；流水线从同步搬到 worker 异步消费（写盘不阻塞用户请求，但要处理乱序与重试——这正是幂等键的价值）。
验收清单：
- 抽取只取 user 轮次，且能识别同句中的寒暄与偏好。
- 去重有文本归一与向量相似两档，阈值可配。
- 冲突消解保留历史（
superseded+superseded_by），不物理删除。 - 每条记忆带幂等键，同一条对话重放两次不产生新行。
- 同一槽位任意时刻只有一条
active。
记忆能可靠地写进来只是第一步。写进来的记忆要在每一次 run 里真正被用上才算数。下一节把长期记忆接进 run，并按 token 预算截断后注入，然后用一个评测小集量化「注入记忆」到底带来多少分。
阅读导航：上一节：10.3 工具沙箱：子进程/容器隔离高危工具 · 下一节：11.2 检索注入：长期记忆接入 run 并量化收益 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。