如果把 AI Agent 比作一个“人”,那么今天大部分 Agent 缺的不是大脑,而是记忆。它们能推理、能调工具、能写代码,但转过身就忘了两轮对话之前用户说过什么。这个痛点,正是“oGMemory 记忆系统与数据分支”这一系列内容真正想解决的问题。
先说结论:记忆系统是 Agent 应用从“能用”走向“好用”的分水岭,而“数据分支”是理解记忆系统内部如何区分、路由和沉淀数据的关键线索。oGMemory 不是又一个向量数据库的包装,它更像是一套为 Agent 设计的记忆管理范式:把对话历史、用户偏好、事实知识、任务状态分门别类,用数据分支的方式决定什么该记、什么该忘、什么该进长期库、什么只留在会话里。
这篇文章会围绕三件事展开:第一,说清楚 Agent 记忆系统为什么如此重要,它到底在解决什么工程问题;第二,拆解“数据分支”这个概念,看看它在记忆系统中承担什么角色;第三,给出从零到一实现一个最小记忆系统的思路和代码,让你能真正动手验证,而不是停留在概念层。如果你正在做 Agent 应用、RAG 检索增强、或者任何需要“上下文理解”的 AI 产品,这篇文章值得读完并收藏。
1. 这篇文章真正要解决的问题
过去一年里,Agent 领域最明显的变化,不是模型能力的提升,而是工程架构的复杂化。早期做一个 ChatBot,只需要把用户消息发给大模型、拿到回复返回即可。但今天的 Agent 需要规划任务、调用外部工具、读取文档、操作数据库,甚至多个 Agent 协作完成一条业务链路。
在这个过程中,最容易被低估、又最让开发者头疼的,就是记忆。
举一个很实际的场景:用户对 Agent 说“帮我查一下上个月华东区的销售数据,然后做一份周报”,Agent 执行完后,用户又说“把里面的异常项标出来”。如果没有记忆系统,第二个请求到达大模型时,模型完全不知道“里面”指的是哪份报告、哪个数据范围、哪个表格。结果就是,用户必须把上一轮要求重新描述一遍,Agent 还得重新查询一次。体验极差,效率极低。
更麻烦的是成本问题。把全部历史记录塞进上下文,既不经济也不可靠。大模型的上下文窗口有限,即使支持长文本,塞入大量无关信息也会导致注意力分散、回答质量下降,Token 费用还水涨船高。
这就引出了一个核心矛盾:Agent 需要记忆,但记忆不是简单地把所有东西堆在一起。它需要一套机制来区分信息层级、决定保留周期、控制写入策略。这套机制,本质上就是记忆系统;而“数据分支”,就是这套机制里用来划分信息流向和存储位置的核心架构。
所以要读这篇文章的读者,大概率会满足下面几种情况之一:
- 你正在开发 Agent 应用,已经遇到了“多轮对话失忆”或“上下文混乱”的问题;
- 你了解了 RAG(检索增强生成),但发现单纯向量检索并不能解决 Agent 状态管理的问题;
- 你想理解 oGMemory 这类记忆系统在设计上的取舍,以及“数据分支”到底指什么;
- 你想知道如何在自己的项目里落地一套简单但可扩展的记忆系统,而不是依赖某个现成产品的黑盒。
无论哪种情况,这篇文章的落点都是一致的:先建立正确的概念框架,再用工程手段把它变成可运行的代码。
2. 什么是 oGMemory:从命名到定位
“oGMemory”这个词拆开看,是“oG”加“Memory”。在输入材料里,oGMemory 与“记忆系统”“数据分支”一起出现,又被归类到“agent记忆系统”这一热词语境中,说明它不是一个孤立的概念,而是 Agent 记忆系统的具体实现或研究项目。“第二期”和“分集1”的表述也暗示,这是一套持续更新的系列内容,每期围绕记忆系统的不同侧面展开。
从命名习惯看,“oGMemory”中的“Memory”很容易理解,就是记忆模块;而“oG”可能是项目代号或组织名,也可能是“Original Generation”之类含义的缩写。由于目前公开信息有限,这里不做武断猜测,只谈更重要的部分:它在 Agent 技术栈中处在什么位置。
如果在纸上画一个 Agent 的系统架构,从上到下大致是:
- 交互层:用户输入、前端界面、消息通道;
- 规划层:任务拆解、ReAct 循环、工具选择;
- 记忆层:会话缓存、向量存储、事实库、状态管理;
- 工具层:API 调用、数据库读写、代码执行;
- 模型层:大语言模型、多模态模型、推理模型。
oGMemory 这类记忆系统,处在“规划层”和“工具层”之间,充当 Agent 的信息中枢。它不负责生成回复,也不负责调用工具,它负责一件事:让 Agent 在任何时刻都能拿到“当前这一步最该知道的信息”。
这意味着记忆系统需要同时做三件事:
- 捕获信息:记录用户说了什么、Agent 做了什么、外部工具返回了什么;
- 组织信息:对这些原始信息进行分类、缓存、摘要、向量化;
- 分发信息:在 Agent 决策的每个节点,把最相关的记忆片段送进上下文。
而“数据分支”要解决的问题,正是在第 2 步的“组织”和第 3 步的“分发”过程中产生的:不同类型的数据应该走不同的通路,不能一股脑塞进同一个存储空间。
3. 当我们在讨论“记忆系统”时,到底在讨论什么
“记忆系统”这个说法在 AI 圈子里经常被混用。有人把 Redis 缓存叫记忆,有人把向量数据库叫记忆,有人把 RAG 检索叫记忆。严格来说,这些都不是完整的记忆系统,它们只是记忆系统的组件。
要理解记忆系统的全貌,可以先做一个类比。
人脑的记忆大致可以分成两类:短期记忆和长期记忆。短期记忆容量有限,比如你临时记一个电话号码,拨完就忘;长期记忆容量大、持久性强,比如你的母语、亲人名字、工作技能。人脑在记忆时还有一个特点:重要的事会反复强化,不重要的会自动淡化。
Agent 的记忆系统设计,几乎是对这个机制的直接模仿:
- 短期记忆(Short-term Memory):通常指当前会话中的上下文,保存在内存或 Redis 这类高速存储中,会话结束或超时后自动清理;
- 长期记忆(Long-term Memory):指跨会话保留的信息,包括用户偏好、历史事实、项目背景,通常持久化到数据库;
- 情景记忆(Episodic Memory):记录“发生过什么”,比如某次任务的执行过程、某次对话的决策链;
- 语义记忆(Semantic Memory):记录“事情是什么”,比如产品知识库、业务规则、领域术语。
RAG 工程里常说的“向量检索”,更多服务于语义记忆的提取。但 Agent 记忆系统必须同时管理这几种记忆,并且能区分它们的来源、时效和优先级。
这里就涉及 Agent 记忆系统的特殊难点:不只是“存”,还要“记时”和“记权”。
- 存:把信息以合适的格式保存下来;
- 记时:知道这条信息是什么时候产生的,它的有效期限是什么;
- 记权:知道这条信息对当前决策有多重要,应该优先被检索到还是压低优先级。
如果一个记忆系统只做到了“存”,那它就不够格叫“系统”,充其量是个存储工具。oGMemory 这类方案的价值,恰恰在于把“记时”和“记权”也纳入了设计。
4. 数据分支:理解记忆系统的核心线索
现在到了这篇文章的核心:数据分支。
什么是“数据分支”?从字面看,它描述的是数据在系统内部的流向被拆分成了多个方向。在记忆系统的语境里,数据分支可以理解为两层含义。
第一层是“入口分支”。Agent 执行一次任务,会产生大量数据:用户原始输入、模型中间推理、工具调用日志、外部 API 返回、代码执行输出。这些数据不能不加区分地写入记忆。用户输入意味着明确意图,优先级最高;模型推理中间过程代表思考路径,适合做复盘但不适合直接长期保存;工具返回结果则可能是临时数据,也可能是需要沉淀的事实。数据入口就要分叉。
第二层是“存储分支”。不同性质的数据要放到不同的存储介质或数据集中。高时效的会话上下文放缓存,需要跨会话复用的用户偏好放键值数据库,需要语义检索的文档片段转成向量放向量库,需要可靠留痕的任务日志放文档数据库。这些存储路径之间的差异,就是数据分支的体现。
用一个电商客服 Agent 的例子来说明。
假设用户说:“我上周买的键盘坏了,想退货。”
这一步会产生哪些数据分支?
- 意图分支:用户意图是“退货申请”,这个信号要立刻进入任务规划模块;
- 事实分支:“上周买键盘”“键盘坏了”是事实信息,适合写入短期记忆,并可能抽取出来存入长期记忆中的用户购买记录;
- 状态分支:退货申请需要订单号、售后单号,这些是任务状态,要单独管理;
- 策略分支:平台退货政策、商家退货规则,属于静态知识,应该从长期记忆中检索;
- 日志分支:整个对话过程、工具调用过程,需要留档,但不需要在下一步决策时全部加载。
如果这五类数据全部放进同一个变量或同一个向量库,会发生什么?要么上下文爆炸,要么检索结果混杂不相关的记录,要么状态信息被事实信息淹没。而通过数据分支,系统可以让 Agent 在“规划退货流程”时只关注意图和状态分支,在“确认用户购买信息”时只检索事实分支,在“查询退货政策”时只访问策略分支。分工明确,互不干扰。
这正是“数据分支”设计的重要性:它让记忆从“一个装东西的大袋子”变成了“一个分好隔间的收纳柜”。oGMemory 将数据分支作为解读主题,说明它的核心设计思想就是围绕信息分流来组织记忆系统的。
5. Agent 记忆系统中各分支的工程定位
理解了数据分支的概念后,还需要知道每个分支在工程落地时应该怎么选型和实现。下面用表格做一个系统性的整理。
| 数据分支 | 典型内容 | 存储形式 | 生命周期 | 主要用途 |
|---|---|---|---|---|
| 会话分支 | 用户消息、Agent 回复、当前对话轮次 | 内存 / Redis List | 会话结束即清 | 维持多轮对话流畅性 |
| 状态分支 | 任务状态、已完成步骤、待办步骤 | 数据库行记录 / Redis Hash | 任务完成即归档 | 支持任务中断续跑 |
| 事实分支 | 用户偏好、业务事实、历史决策点 | 键值存储 / 文档数据库 | 长期保留 | 跨会话复用用户画像 |
| 语义分支 | 文档、知识库、历史优秀回答 | 向量数据库 | 长期保留 | RAG 检索与相似召回 |
| 日志分支 | 工具调用记录、模型推理日志 | 日志系统 / 文件存储 | 按策略归档 | 调试、审计、复盘 |
| 摘要分支 | 长对话的压缩摘要 | 键值存储 | 按摘要有效期保留 | 降低长会话的 Token 成本 |
从这张表可以看出,数据分支并不只是一个“概念”,它对存储选型、生命周期管理、读写频率都有直接约束。实际开发中,开发者需要为每个分支定义:
- 写入时机:什么事件触发数据写入这个分支;
- 读取策略:Agent 在什么条件下读取这个分支;
- 清理策略:数据保留多久,何时删除或转储。
以“状态分支”为例,它的写入时机是 Agent 每一步执行完成之后,读取时机是 Agent 开始新一轮规划时,清理策略是任务完成或超时后归档。如果这三个策略没有定义清楚,状态分支很容易退化成日志分支,数据越积越多,检索越来越慢。
这也是为什么很多 Agent 框架自带记忆能力,但效果不佳:它们只实现了“存储”,没有实现“分支策略”。数据都堆在一起,谈不上精准分发。
6. 数据分支与 Agent 记忆的核心链路设计
理解了数据分支的定位,再看记忆系统的完整链路,就容易多了。
一个典型的 Agent 记忆读写链路,可以拆成六个环节:
- 事件捕获:Agent 与用户交互、调用工具时,产生原始事件;
- 分支判定:根据事件类型,决定数据进入哪个或哪几个分支;
- 结构化处理:对原始文本做抽取、摘要、清洗、向量化;
- 分支存储:写入对应的存储介质;
- 按需检索:在 Agent 决策节点,按当前意图检索对应分支的数据;
- 上下文组装:把检索到的记忆片段拼装成模型可用的上下文。
这个链路中,最容易出问题的环节是第 2 步“分支判定”和第 5 步“按需检索”。
第 2 步的难点在于,一个事件可能同时属于多个分支。用户说“帮我记住这个客户的邮箱是 xx@example.com”,这句话既是对话内容,又是事实知识,还可能影响后续客服任务的状态。判定规则如果写死“对话消息只进会话分支”,就会丢失关键信息;如果全部写入所有分支,又会造成冗余。合理的做法是维护一份“信息类型 -> 目标分支”的映射规则,同时引入小规模的模型抽取来判断哪些信息值得进入长期分支。
第 5 步的难点在于,检索时机和检索范围直接影响模型效果。如果每轮对话都把用户画像、订单记录、知识库文档全部检索一遍,效果不一定更好,成本却一定更高。更合理的策略是在每个环节只加载必要的分支。规划阶段检索状态分支,理解用户意图时检索会话分支,回答事实性问题时检索语义分支,推荐个性化内容时检索事实分支。
数据分支的设计,本质上就是在回答上述“何时读、读什么、从哪里读”的问题。它可以被视为记忆系统内部的“路由规则”,就像 API 网关将不同请求转发给不同后端服务一样。
7. 从概念到落地:最小记忆系统的工程实现
概念讲得再多,最终还是要落到代码。下面用一个最小实现,演示“数据分支”思维在工程上怎么落地。
这里不依赖任何具体的 Agent 框架,用 Python 实现一个简化版记忆管理器。它包含三个分支:会话分支、事实分支、语义分支。每个分支有独立的读写接口。
# 文件路径:memory_system/memory.py from typing import Any, Dict, List, Optional from datetime import datetime, timedelta class MemoryBranch: """记忆分支基类,所有分支共用统一的读写接口。""" def __init__(self, name: str, ttl: Optional[int] = None): self.name = name # ttl 单位:秒。None 表示永久保留 self.ttl = ttl self._store: Dict[str, Any] = {} self._timestamps: Dict[str, datetime] = {} def write(self, key: str, value: Any) -> None: self._store[key] = value if self.ttl: self._timestamps[key] = datetime.now() def read(self, key: str) -> Optional[Any]: if key in self._store: if self.ttl: ts = self._timestamps.get(key) if ts and datetime.now() - ts > timedelta(seconds=self.ttl): # 已过期,清理并返回 None self.delete(key) return None return self._store.get(key) return None def delete(self, key: str) -> None: self._store.pop(key, None) self._timestamps.pop(key, None) def keys(self) -> List[str]: # 返回当前未被清理的 key 列表 result = [] for k in self._store: if self.read(k) is not None: result.append(k) return result class SessionBranch(MemoryBranch): """会话分支:存当前会话上下文,短 TTL。""" def __init__(self): super().__init__(name="session", ttl=1800) self._turn_count = 0 def append_turn(self, user_msg: str, agent_msg: str) -> None: self._turn_count += 1 key = f"turn_{self._turn_count}" self.write(key, {"user": user_msg, "agent": agent_msg}) def get_recent_turns(self, n: int = 3) -> List[Dict[str, str]]: keys = self.keys() recent_keys = keys[-n:] return [self.read(k) for k in recent_keys if self.read(k)] class FactBranch(MemoryBranch): """事实分支:存用户偏好和业务事实,永久保留。""" def __init__(self): super().__init__(name="fact", ttl=None) def set_fact(self, key: str, value: str) -> None: self.write(key, value) def get_fact(self, key: str) -> Optional[str]: return self.read(key) class SemanticBranch(MemoryBranch): """语义分支:模拟向量检索,实际项目中会替换为向量数据库。""" def __init__(self): super().__init__(name="semantic", ttl=None) self._docs: List[Dict[str, str]] = [] def add_document(self, doc_id: str, text: str) -> None: self._docs.append({"doc_id": doc_id, "text": text}) def search(self, keyword: str, top_k: int = 2) -> List[str]: """极简关键词检索,仅示意。正式场景应使用向量相似度检索。""" scored = [] for doc in self._docs: if keyword in doc["text"]: scored.append(doc["text"]) return scored[:top_k]上面的基类统一了读写接口,三个分支各自有独立的存储语义。SessionBranch 带 TTL,过期自动清理;FactBranch 永久保留;SemanticBranch 简化为关键词检索,但接口设计上与向量存储对齐。
接下来,实现一个 MemoryManager,用来调度不同分支的写入和读取。
# 文件路径:memory_system/manager.py from typing import Any, Dict, Optional from .memory import SessionBranch, FactBranch, SemanticBranch class MemoryManager: """记忆管理器:对外统一入口,对内按数据分支路由。""" def __init__(self): self.session = SessionBranch() self.fact = FactBranch() self.semantic = SemanticBranch() def record_turn(self, user_msg: str, agent_msg: str) -> None: """对话事件 -> 会话分支""" self.session.append_turn(user_msg, agent_msg) def remember_fact(self, key: str, value: str) -> None: """明确告诉 Agent 记住的事实 -> 事实分支""" self.fact.set_fact(key, value) def add_knowledge(self, doc_id: str, text: str) -> None: """领域知识文档 -> 语义分支""" self.semantic.add_document(doc_id, text) def build_context(self, keyword: Optional[str] = None) -> Dict[str, Any]: """组装上下文:按需从不同分支取值。""" context = { "recent_turns": self.session.get_recent_turns(3), "facts": {}, "semantic_hits": [], } # 事实分支全量读取(实际项目中可按用户维度读取) for key in self.fact.keys(): context["facts"][key] = self.fact.get_fact(key) # 语义分支按关键词检索 if keyword: context["semantic_hits"] = self.semantic.search(keyword) return context最后,写一个调用示例,演示分支读写在真实 Agent 循环中的使用方式。
# 文件路径:demo.py from memory_system.manager import MemoryManager # 初始化记忆管理器 mm = MemoryManager() # 第 1 轮对话:用户咨询键盘退货 user_msg_1 = "我上周买的键盘坏了,想退货" agent_msg_1 = "好的,请问您的订单号是多少?" mm.record_turn(user_msg_1, agent_msg_1) # 用户提供了订单号,同时 Agent 识别到这是一个重要事实 user_msg_2 = "订单号是 JD123456" agent_msg_2 = "已记录,正在为您查询退货政策。" mm.record_turn(user_msg_2, agent_msg_2) mm.remember_fact("order_id", "JD123456") mm.remember_fact("user_intent", "return_keyboard") # 向知识库添加退货政策文档 mm.add_knowledge( doc_id="policy_001", text="键盘类商品支持七天无理由退货,需保持包装完整。", ) # 下一轮用户追问时,组装上下文 context = mm.build_context(keyword="退货政策") print("最近对话:", context["recent_turns"]) print("记忆事实:", context["facts"]) print("知识命中:", context["semantic_hits"])运行这段代码,输出大致如下:
最近对话: [{'user': '我上周买的键盘坏了,想退货', 'agent': '好的,请问您的订单号是多少?'}, {'user': '订单号是 JD123456', 'agent': '已记录,正在为您查询退货政策。'}] 记忆事实: {'order_id': 'JD123456', 'user_intent': 'return_keyboard'} 知识命中: ['键盘类商品支持七天无理由退货,需保持包装完整。']从输出可以看到,三个分支各司其职:会话分支保存“最近发生了什么”,事实分支保存“当前用户的关键信息”,语义分支提供“退货政策这类领域知识”。Agent 在下一轮决策时,可以只读取 context,拿到当前最需要的信息,而不是把 100 条历史日志全部塞给模型。
8. 完整示例:在模拟 Agent 循环中使用记忆系统
上面的最小实现只是基础。为了让“数据分支”的工程价值更明显,这里再演示一个完整的模拟 Agent 循环:一个客服 Agent,根据记忆内容判断用户请求是否需要新建售后单。
# 文件路径:demo_agent_loop.py from memory_system.manager import MemoryManager class CustomerServiceAgent: """模拟客服 Agent,使用记忆管理器支撑多轮对话。""" def __init__(self): self.memory = MemoryManager() self.after_sale_created = False def handle(self, user_msg: str) -> str: # 1. 将用户输入写入会话分支 self.memory.record_turn(user_msg, "") # 2. 从事实分支读取关键信息 order_id = self.memory.fact.get_fact("order_id") user_intent = self.memory.fact.get_fact("user_intent") # 3. 判断是否已经有售后单状态 if "售后" in user_msg or "退货" in user_msg: if order_id and not self.after_sale_created: self.after_sale_created = True reply = f"已为您创建售后单,订单号 {order_id},稍后会有客服专员联系您。" elif self.after_sale_created: reply = "您的售后单正在处理中,请留意短信通知。" else: reply = "需要您的订单号才能创建售后单,请问订单号是多少?" elif "订单号" in user_msg: # 简单抽取订单号信息,写入事实分支 extracted = user_msg.split("是")[-1].strip() self.memory.remember_fact("order_id", extracted) self.memory.remember_fact("user_intent", "return_keyboard") reply = f"好的,订单号 {extracted} 已记录。需要帮您申请退货吗?" else: reply = "请问您遇到了什么问题?" # 4. 把回复写入会话分支 self.memory.session.append_turn("", reply) # 简化:只更新回复 return reply # 模拟对话 agent = CustomerServiceAgent() print("用户:键盘坏了想退货") print("Agent:", agent.handle("键盘坏了想退货")) print() print("用户:订单号是 JD123456") print("Agent:", agent.handle("订单号是 JD123456")) print() print("用户:确认退货") print("Agent:", agent.handle("确认退货"))这个 Agent 循环演示了数据分支的关键价值:用户说“确认退货”时,模型并不需要重新理解整段历史。它只需要从事实分支读取 order_id 和 user_intent,判断售后单是否已创建,就能给出正确响应。这就是记忆系统对 Agent 的“状态支撑”。
如果不用这个方案,第二个和第三个请求就得重复携带“我上周买了键盘”“订单号是 JD123456”这些背景信息。一旦会话拉长,上下文长度无法控制,模型还容易忘记早期信息。
9. 记忆系统的常见问题与排查方法
实际开发记忆系统时,会遇到不少问题。下面整理了几个高频场景,方便大家按表排查。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| Agent 回答时忘记早期对话内容 | 会话分支未写入或 TTL 过短 | 检查会话分支的写入方法是否被调用;查看存储中是否有历史轮次 | 延长会话 TTL,或改用“摘要分支”保存长会话压缩结果 |
| 上下文 Token 不断膨胀,单次请求超限 | 没有数据分支,所有信息都进上下文 | 在组装上下文前打印各部分长度,确认是否加载了不必要分支 | 按需读取分支,增加上下文裁剪策略 |
| 用户偏好更新后,Agent 仍使用旧值 | 事实分支写入时机不对,或读取优先级错误 | 检查事实更新是否覆盖了旧 key;检查查询时是否命中了缓存 | 统一事实写入接口,版本化更新 |
| 向量检索返回大量无关内容 | 语义分支没有做知识库范围过滤 | 检查检索时是否传入了领域过滤条件 | 为文档添加元数据字段,检索时配合过滤 |
| 任务中断续跑时,Agent 不知道执行到哪一步 | 状态分支缺失 | 查看任务执行日志有无步骤记录 | 在每一步执行后写入状态分支,任务恢复时先读取 |
| 记忆内容涉及用户隐私,未做隔离 | 数据分支没有区分用户维度 | 检查存储 key 是否包含 userId | 所有分支 key 增加用户维度,落地最小权限原则 |
这里特别提醒两个工程细节。
第一,会话分支的 TTL 不宜无限长,否则每轮对话都会把全部历史加载进上下文,这会让“短时记忆”失去意义。更合理的做法是:超过一定轮次后,对早期对话做摘要,摘要存到摘要分支,原始内容归档到日志分支。
第二,事实分支的写入要谨慎。用户说“我想退货”,并不代表“用户偏好退货”。事实分支应该保存的是可以复用、不容易变化的信息,比如订单号、收货地址、会员等级。对意图类信息,用完之后应该清理或降级到日志。
10. 记忆系统的最佳实践与工程建议
把概念和代码都过了一遍,最后给出几条可以放进实际项目的建议。
10.1 先定分支,再写功能模块
很多开发者做记忆功能时,习惯先写“存储工具类”,再在需要时调用。这个顺序容易导致分支边界模糊。更推荐的做法是先画出数据分支图,明确每种信息属于哪个分支,再为每个分支实现独立的读写接口。分支一旦定好,后续加功能只是往对应分支里加数据,不会污染其他模块。
10.2 所有分支的 key 都要带命名空间
如果系统服务多个用户,记忆的隔离是第一优先级。所有 key 都应该带上用户 ID 或会话 ID 前缀,例如:
user:12345:fact:order_id user:12345:session:turn_12 tenant:abc:semantic:policy_001这样即使底层存储是同一个 Redis 实例,数据之间也不会串。同时,后续做权限控制、数据导出、数据清理都会更容易。
10.3 上下文组装要有专门的调度器
不建议在业务代码里到处调用 memory 读取。更优雅的方式是做一个 ContextBuilder,统一负责从各个分支取数、裁剪、拼装。这样当检索策略改变时,只需要改一处,不需要改动所有 Agent 调用点。
10.4 写操作要有审计日志
记忆系统是 Agent 的信息底座,写坏了影响面极大。建议对关键分支的写入记录审计日志:谁在什么时间写了什么数据。一旦线上出现 Agent 行为异常,可以通过审计日志快速定位是不是记忆数据被污染。
10.5 记忆内容脱敏与权限控制要前置
存入记忆系统的用户数据可能包含地址、订单号、身份信息等敏感内容。在写入阶段就应该做脱敏或加密,而不是等到检索阶段再处理。如果使用向量数据库,也要确认向量库自带访问控制,避免通过检索接口越权读取他人数据。
10.6 定期做记忆数据治理
记忆系统里的数据质量会随时间下降。用户可能更换订单、修改偏好、注销账号。需要建立数据治理任务,定期清理过期事实、归档旧会话、删除用户主动要求删除的记忆数据。否则,记忆系统会从“助手”退化成“噪音源”。
11. 总结与后续学习方向
回顾这篇文章,核心想讲清楚三件事。
第一,Agent 记忆系统解决的不是“存数据”的问题,而是“信息如何分类、分流、按需分发”的问题。oGMemory 将记忆系统与数据分支并提,说明它关注的就是这一层工程架构,而不是简单的存储选型。
第二,数据分支是记忆系统的信息路由骨架。会话、状态、事实、语义、日志、摘要这六类分支各有各的存储形式、生命周期和读写策略。理解了分支划分,就理解了记忆系统的设计核心。使用数据分支时,需要明确每个分支的写入时机、读取策略和清理策略。
第三,通过最小 Python 实现,可以看到数据分支在代码层面并不复杂,难的是业务规则的梳理。先把分支边界定义清楚,再谈具体技术栈,才有意义。
如果你对这个方向感兴趣,下一步可以按照这几个路径深入:学习向量数据库的真实语义检索实现,替换掉文中极简的 keyword search;研究 Agent 框架中记忆模块的接入方式;尝试为你的现有 Agent 项目加入状态分支,让任务支持中断续跑;关注 oGMemory 后续系列内容,特别是它如何进一步拆解“数据分支”与记忆周期的关系。
记忆系统是 Agent 应用从 demo 走向产品的必经之路。现在动手设计好数据分支,未来扩展就不会被历史包袱拖住。文中的代码可以直接复制到本地运行,建议按自己的业务场景改一改,跑通一遍分支路由的完整链路。