智能体面试准备（十三）：代码实例——带持久化记忆的 Agent，跨会话也不忘事

写在前面

B12 我们手写了完整的 ReAct Agent，但它有个致命短板：记忆是活在内存里的。进程一关，scratchpad 全没了——你下一轮问它"我们刚才说到哪了"，它会一脸茫然。生产里的 Agent（客服、个人助理、研发助手）必须"记得"跨会话的偏好、历史与事实，这就是持久化记忆（Persistent Memory）。

本篇直接给一个完整可运行的实现：把 B12 的 Agent 骨架接上三层持久化记忆（短期会话 / 长期文件 / 向量召回），用纯标准库 + SQLite 做存储，不依赖任何框架，也不需要 API Key（用 FakeLLM 跑通）。读完你能直接把它当模板拼进自己的项目。

一、为什么内存记忆不够

B6 讲过记忆的"短期/长期/向量"三层架构，本篇把它落成代码：短期=当前会话窗口（仍在内存，快）；长期=SQLite 落盘的关键事实；向量=语义召回历史。三者职责分明。

二、完整实现

# -*- coding: utf-8 -*- """带持久化记忆的 ReAct Agent（纯标准库 + sqlite，完整可运行，无需 API Key） 架构：B12 的 ReAct 主循环 + 三层记忆 - ShortTermMemory : 当前会话窗口（内存，带最大轮数） - LongTermMemory : SQLite 落盘的关键事实（user 级，跨会话） - VectorMemory : 极简 TF-IDF 向量召回历史对话（演示语义检索） """ import re import json import sqlite3 import math from collections import Counter # ---------- 工具集 ---------- def search_tool(query: str) -> str: """模拟外部搜索工具""" return f"[search] 关于「{query}」的检索结果：...（mock）" def calculator_tool(expr: str) -> str: """安全计算（仅允许数字与运算符）""" if re.fullmatch(r"[\d\+\-\*\/\(\)\.\s]+", expr): try: return str(eval(expr)) except Exception as e: return f"计算错误: {e}" return "非法表达式" TOOLS = {"search": search_tool, "calculator": calculator_tool} # ---------- 三层记忆 ---------- class ShortTermMemory: """当前会话窗口：最近 N 轮，纯内存""" def __init__(self, max_turns: int = 6): self.max_turns = max_turns self.buffer: list[dict] = [] def add(self, role: str, content: str): self.buffer.append({"role": role, "content": content}) if len(self.buffer) > self.max_turns * 2: self.buffer = self.buffer[-self.max_turns * 2:] def context(self) -> str: return "\n".join(f"{m['role']}: {m['content']}" for m in self.buffer) class LongTermMemory: """SQLite 落盘的关键事实，按 user_id 隔离，跨会话持久""" def __init__(self, db_path: str = "agent_memory.db"): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.execute( "CREATE TABLE IF NOT EXISTS facts " "(user_id TEXT, key TEXT PRIMARY KEY, value TEXT, ts REAL)" ) def upsert(self, user_id: str, key: str, value: str, ts: float): self.conn.execute( "INSERT INTO facts(user_id,key,value,ts) VALUES(?,?,?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value, ts=excluded.ts", (user_id, key, value, ts), ) def get(self, user_id: str, key: str): row = self.conn.execute( "SELECT value FROM facts WHERE user_id=? AND key=?", (user_id, key) ).fetchone() return row[0] if row else None def forget(self, user_id: str, key: str): self.conn.execute("DELETE FROM facts WHERE user_id=? AND key=?", (user_id, key)) def all(self, user_id: str) -> dict: return {k: v for k, v in self.conn.execute( "SELECT key,value FROM facts WHERE user_id=?", (user_id,))} class VectorMemory: """极简 TF-IDF 语义召回：演示从历史事件里找相关记忆""" def __init__(self): self.docs: list[str] = [] def add(self, text: str): self.docs.append(text) def _tfidf(self, text: str) -> Counter: words = re.findall(r"\w+", text.lower()) tf = Counter(words) idf = Counter() for d in self.docs: idf.update(set(re.findall(r"\w+", d.lower()))) N = max(len(self.docs), 1) return Counter({w: (tf[w]) * math.log(N / (idf[w] or 1) + 1) for w in tf}) def recall(self, query: str, top_k: int = 2) -> list[str]: if not self.docs: return [] qv = self._tfidf(query) scored = [] for d in self.docs: dv = self._tfidf(d) dot = sum(qv[w] * dv[w] for w in qv) norm = math.sqrt(sum(v*v for v in qv.values())) * math.sqrt(sum(v*v for v in dv.values())) or 1 scored.append((dot / norm, d)) scored.sort(reverse=True) return [d for _, d in scored[:top_k]] # ---------- FakeLLM：让代码无需 API Key 也能跑通 ---------- class FakeLLM: """按脚本规则生成 Thought/Action，演示完整闭环（非真实推理）""" def __init__(self): self.round = 0 def __call__(self, prompt: str) -> str: self.round += 1 # 第1轮：决定调用工具 if "用户: 帮我算" in prompt and "Observation" not in prompt: return 'Thought: 用户要算一个数\nAction: calculator\nAction Input: 23*7' if "用户: 我记得" in prompt and "Observation" not in prompt: return 'Thought: 用户在告知偏好，应写入长期记忆\nAction: remember\nAction Input: {"key":"name","value":"吕博"}' # 其余轮次：给出最终答案 return "Thought: 已有足够信息\nFinal Answer: 已处理你的请求并完成记忆持久化。" # ---------- 持久化 Agent ---------- class PersistentAgent: def __init__(self, user_id: str = "u_default"): self.user_id = user_id self.stm = ShortTermMemory() self.ltm = LongTermMemory() self.vtm = VectorMemory() self.llm = FakeLLM() self.max_steps = 5 self.tools = dict(TOOLS) self.tools["remember"] = self._remember def _remember(self, payload: str) -> str: try: obj = json.loads(payload) self.ltm.upsert(self.user_id, obj["key"], obj["value"], 0.0) return f"已记住 {obj['key']}={obj['value']}" except Exception as e: return f"记忆失败: {e}" def _parse(self, text: str): m_act = re.search(r"Action:\s*(\w+)", text) m_inp = re.search(r"Action Input:\s*(.+)", text) action = m_act.group(1) if m_act else None action_input = m_inp.group(1).strip() if m_inp else "" return action, action_input def run(self, user_msg: str) -> str: # 召回相关历史，注入上下文（向量记忆） recalled = self.vtm.recall(user_msg) recalled_ctx = "\n".join(f"[历史记忆] {r}" for r in recalled) # 注入长期事实 long_ctx = "\n".join(f"[已知事实] {k}={v}" for k, v in self.ltm.all(self.user_id).items()) self.stm.add("user", user_msg) prompt = f"{long_ctx}\n{recalled_ctx}\n{self.stm.context()}\n" for _ in range(self.max_steps): out = self.llm(prompt) action, action_input = self._parse(out) self.stm.add("assistant", out) if action == "Final Answer" or "Final Answer" in out: return out.split("Final Answer:")[-1].strip() if action in self.tools: obs = str(self.tools[action](action_input)) self.stm.add("observation", obs) self.vtm.add(f"Q:{user_msg} A:{obs}") prompt = f"{long_ctx}\n{recalled_ctx}\n{self.stm.context()}\n" return "（达到最大步数，未完成）" if __name__ == "__main__": agent = PersistentAgent("u_lvbo") print(">> 第1轮:", agent.run("帮我算一下 23*7")) print(">> 第2轮:", agent.run("我记得我叫吕博，记住它")) print(">> 第3轮(新会话模拟,事实仍在):", agent.run("我叫什么名字？"))

运行后会看到：第 2 轮"记住名字"被写入 SQLite；第 3 轮即使重新构造了上下文，依然能从LongTermMemory读出"吕博"——这就是持久化。把agent_memory.db删了再跑第 3 轮，记忆就没了，正好演示落盘的意义。

三、生产化升级清单

从 demo 到生产，这版还差六步（和 B12 的清单一脉相承，但补了记忆专属项）：

1. 记忆写入策略：本例用显式remember工具写，生产更常用反思式写入（Agent 定期总结会话、抽取事实自动 upsert）或用户确认后写入。
2. 召回质量：TF-IDF 太弱，换成真实 embedding + 向量库（faiss/pgvector），并加"相关性 + 时近性 + 重要性"加权（B6 给过公式）。
3. 遗忘机制：长期记忆要能forget、能过期（TTL），否则噪声累积。
4. 多用户隔离：用user_id分区，本例已体现，生产还要加权限校验。
5. 一致性：并发写入 SQLite 需事务/锁；线上常用 Redis（热）+ 向量库（语义）+ 关系库（事实）三层。
6. 隐私与合规：长期记忆可能存 PII，要支持按用户删除（GDPR/个保法），这是下一组"治理"篇会展开的点。

小结

持久化记忆把 Agent 从"一次性工具"变成"能长期相处的助理"。三层职责：短期走内存保速度、长期落盘保跨会话、向量做语义召回保相关性。本篇代码可直接当模板——把FakeLLM换成真实模型接口、SQLite换成你的存储栈即可。下一篇（B14）我们把 Agent 的"手"做得更标准：用 MCP 协议把工具拆成独立的 client/server，让 Agent 能即插即用地调用外部能力。