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

智能体面试准备十三代码实例——带持久化记忆的 Agent跨会话也不忘事写在前面B12 我们手写了完整的 ReAct Agent但它有个致命短板记忆是活在内存里的。进程一关scratchpad 全没了——你下一轮问它我们刚才说到哪了它会一脸茫然。生产里的 Agent客服、个人助理、研发助手必须记得跨会话的偏好、历史与事实这就是持久化记忆Persistent Memory。本篇直接给一个完整可运行的实现把 B12 的 Agent 骨架接上三层持久化记忆短期会话 / 长期文件 / 向量召回用纯标准库 SQLite 做存储不依赖任何框架也不需要 API Key用 FakeLLM 跑通。读完你能直接把它当模板拼进自己的项目。一、为什么内存记忆不够维度内存记忆B12 版持久化记忆本篇生命周期进程存活期间跨进程、跨会话、跨天容量受上下文窗口限制可无限增长按需召回故障恢复进程崩即丢重启可恢复多用户一团混在一起按 user_id 隔离适用Demo / 单次任务生产 / 长期助理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_threadFalse) 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 valueexcluded.value, tsexcluded.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(reverseTrue) 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(rAction:\s*(\w), text) m_inp re.search(rAction 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(fQ:{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 的清单一脉相承但补了记忆专属项记忆写入策略本例用显式remember工具写生产更常用反思式写入Agent 定期总结会话、抽取事实自动 upsert或用户确认后写入。召回质量TF-IDF 太弱换成真实 embedding 向量库faiss/pgvector并加相关性 时近性 重要性加权B6 给过公式。遗忘机制长期记忆要能forget、能过期TTL否则噪声累积。多用户隔离用user_id分区本例已体现生产还要加权限校验。一致性并发写入 SQLite 需事务/锁线上常用 Redis热 向量库语义 关系库事实三层。隐私与合规长期记忆可能存 PII要支持按用户删除GDPR/个保法这是下一组治理篇会展开的点。小结持久化记忆把 Agent 从一次性工具变成能长期相处的助理。三层职责短期走内存保速度、长期落盘保跨会话、向量做语义召回保相关性。本篇代码可直接当模板——把FakeLLM换成真实模型接口、SQLite换成你的存储栈即可。下一篇B14我们把 Agent 的手做得更标准用 MCP 协议把工具拆成独立的 client/server让 Agent 能即插即用地调用外部能力。