AI Agent 在执行复杂任务时真正限制它表现的往往不是单个模型的能力而是它能不能记住上下文、能不能在关键节点调出正确信息。oGMemory 记忆系统要解决的正是 Agent 跨会话、跨任务的记忆组织问题而数据分支又是决定记忆系统能不能“用起来”的关键设计。这一期分集把视线聚焦在记忆系统与数据分支上先讲清楚数据分支为什么存在再给出一套可以落地的最小实现。很多项目在最初接入记忆能力时通常的做法是“对话结束后把文本塞进向量库下次检索再查出来”。这个思路在 Demo 阶段可行但一旦进入真实业务就会出现一个典型问题所有记忆混在一起用户偏好、任务记录、知识沉淀、短期上下文全部堆在同一个集合里。写入越久检索噪声越大Agent 越来越难分辨哪条信息是当前任务该用的。数据分支的作用就是按业务维度把这些记忆拆开让写入有明确归属让检索有明确范围。1. 记忆系统到底在解决什么问题1.1 没有记忆的 Agent 为什么不够用没有记忆系统的 Agent本质上是一个“每次对话都从零开始”的状态机。模型本身拥有训练阶段沉淀的静态知识但它不知道用户上一次问过什么、当前任务做到哪一步、用户偏好哪种回答风格。对于一次性的问答场景这没有问题但对于需要连续执行多步任务、跨天维护用户关系、持续沉淀团队知识的场景缺记忆就意味着每次都要用户重新交代背景。oGMemory 这类记忆系统的核心目标是给 Agent 增加一个可持续读写的外部状态层。模型本身不可变但记忆数据可以随时间、任务、用户不断更新。Agent 在执行任务前先读取记忆执行过程中写入新记忆任务结束后再整理归档这样它就能形成“越用越懂当前场景”的能力。需要注意的是记忆不是简单地等于聊天记录。聊天记录是原始素材记忆系统要把素材加工成结构化的、可检索的、能按需过滤的数据。这个加工过程包括抽取主题、过滤噪声、判定重要性、设置生命周期、分配数据分支。1.2 “能存下来”和“会用”之间的差距只把文本存进数据库并不是记忆系统。真正的记忆系统要处理四个连续问题第一什么值得记。不是每一句对话都值得写入长期记忆。寒暄、临时计算过程、重复确认信息都应该被过滤掉。真正值得记的是用户偏好、事实结论、任务状态、关键约束。第二记到哪里去。这个问题就是数据分支要解决的。不同性质的记忆需要不同的存储策略比如短期会话上下文可以放在快速缓存中用户长期偏好需要进入稳定的关系型存储或向量库而任务执行记录可能需要按项目维度隔离。第三如何被找到。写入时可读不等于检索时可命中。检索环节要处理相似度计算、过滤条件、排序策略、时效衰减。如果没有分支约束检索时会把不相关的记忆也召回导致上下文被污染。第四什么时候被更新或遗忘。记忆不能只增不减。当用户明确改变偏好、任务状态发生流转、数据超过保留期限时系统要具备更新、合并、降级、删除能力。这四个问题里存储和检索最容易理解但也最容易被低估。很多时候项目跑不起来不是因为模型不行而是因为记忆数据没有组织好导致 Agent 读到了错误的历史信息。1.3 oGMemory 在其中的定位从命名和常见记忆系统设计来看oGMemory 可以理解为一条独立的记忆数据链路重点解决 Agent 记忆的组织、存储、检索和生命周期管理。和直接在业务代码里零散调用向量库不同oGMemory 这类设计会把记忆系统抽象成独立服务对外提供写入、查询、更新、遗忘等接口让上层 Agent 业务只关心记忆内容不关心底层存储细节。本文作为分集解读的第一篇聚焦在数据分支这一层不展开讨论底层模型、检索算法、遗忘机制的全部细节。读完这一篇你能完成三件事理解记忆系统的整体分层掌握数据分支的划分思路用一套最小代码跑通“写入分支、按分支查询、分支归档”的完整流程。2. 搭建最小记忆系统需要先确认的核心组件2.1 记忆系统的五层结构一个可工作的记忆系统通常包含五层感知层、编码层、存储层、检索层、遗忘与合并层。数据分支主要落在存储层和检索层之间但每一层都会影响分支设计。感知层负责识别什么信息值得写入记忆。它可能是对话中断言抽取、任务状态识别、用户行为事件接收。感知层的输出是“一条候选记忆”这部分通常要依赖 LLM 或规则引擎完成。编码层负责把文本转换成可以检索的形态。最常见的方式是文本向量化同时保留原文和结构化字段。编码层还要生成主题标签、时间戳、重要程度、来源标识等元数据。存储层负责持久化。向量数据进入向量库结构化字段进入关系型数据库原始内容可能需要对象存储。数据分支在这一层体现为不同的集合、表或分片。检索层负责把用户当前的问题转换成查询条件从正确的分支中召回记忆。检索不是只做语义相似度还需要结合时间范围、用户身份、分支类型、权限范围等条件。遗忘与合并层负责控制记忆生命周期。它解决记忆过期、冲突覆盖、重复合并、降级归档等问题。没有这一层记忆数据会无限膨胀检索质量会持续下降。2.2 组件选型和环境准备记忆系统的选型不存在一套通用标准答案但通常会涉及以下几类组件组件职责常见选型参考LLM文本理解、信息抽取、摘要生成业务现有的大模型服务Embedding 服务将文本转换为向量本地嵌入模型接口或平台嵌入服务向量库存放向量数据支持语义检索轻量可用 Chroma、生产环境可用 Milvus 等关系型数据库存放记忆记录、分支元数据、状态SQLite 适合本地验证生产可用 PostgreSQL 等缓存存储高频访问的短期记忆Redis 等定时任务执行归档、合并、遗忘清理APScheduler、Celery Beat 或云厂商定时任务这里要特别说明本文的示例代码为了保持最小可运行使用 SQLite 存储结构化数据向量部分用一个本地 Python 列表和一个模拟的 embedding 函数代替。这样做的目的是先把数据分支逻辑讲清楚。真实项目接入时把 embedding 函数替换成内部嵌入服务把向量列表迁移到向量库即可。2.3 环境准备清单在开始写代码之前先确认本机环境满足以下条件Python 3.10 或更高版本已安装 FastAPI 和 uvicorn已安装 SQLite3 驱动Python 自带的 sqlite3 就够用已安装 pandas 或直接使用 Python 标准库处理数据一个可用的 embedding 函数开发阶段可以用随机向量模拟但验证阶段建议接入真实嵌入服务如果原始项目没有明确版本落地前务必先确认依赖版本和 Python 版本兼容性。这里给出一个 requirements 示例fastapi0.110.0 uvicorn0.29.0 pydantic2.6.0 numpy1.26.4 apscheduler3.10.4注意版本号只是参考。实际项目如果使用已有依赖锁文件以仓库中的版本为准不要直接复制最新版本号到生产环境。3. 数据分支是什么为什么记忆系统离不开它3.1 数据分支的定义数据分支是指按照某种业务维度把记忆数据划分成不同的逻辑通道。每个通道拥有独立的写入规则、存储位置、检索范围和生命周期策略。可以把数据分支理解成“记忆的分类目录”。没有分类目录时所有记忆是一条无序的长河有了分类目录后写入时先判断这条记忆属于哪个分类检索时只从对应分类中查找效率和准确率都会明显提升。在 oGMemory 这类 Agent 记忆系统中分支不是一个可有可无的优化项而是决定记忆能否被正确使用的关键设计。原因是 Agent 的记忆数据天然带有强烈的“上下文依赖”特征。用户说“我喜欢简洁的回答”这是一条偏好记忆用户说“项目 A 的数据库连接串已经改好了”这是一条任务事件记忆用户说“本周五要上线”这是一条计划记忆。这三者的使用场景完全不同混在一起存储检索时很难一次性命中正确信息。3.2 常见分支维度实际项目中记忆系统通常不会只使用一个分支维度而是几个维度组合使用。常用维度包括分支维度说明示例按时间区分短期上下文、中期工作记忆、长期沉淀short_term、mid_term、long_term按来源区分用户提供、Agent 推导、系统事件、外部知识user、agent、system、knowledge按类型区分事实、偏好、事件、技能、约束fact、preference、event、skill、constraint按业务域区分不同项目、不同知识库、不同团队project_a、project_b、wiki_base按生命周期区分活跃、候选归档、已过期、废弃active、archived、expired、trashed分支维度的选择不是越多越好。每增加一个维度写入时的路由判断就更复杂检索时的过滤条件也更多运维成本会指数上升。多数中小型记忆系统从“来源 类型 时间”三个维度起步就足够之后根据真实检索效果再细化。3.3 数据分支与分库分表的区别数据分支容易被误解成“分库分表”但它们解决的问题不同。分库分表解决的是存储容量和写入性能问题。当单表数据量达到千万级索引失效、写入变慢这时需要把数据按照哈希或范围分散到多个物理存储中。分库分表是物理层的伸缩方案对业务透明上层 SQL 大多数情况下不应该感知到分片逻辑。数据分支解决的是语义隔离和检索范围问题。它强调“哪些记忆属于哪个业务范围”是逻辑层的分类设计。即使数据量很小也需要做数据分支否则 Agent 会在两三百条记忆里被噪声干扰。两者可以叠加使用。先按业务维度做数据分支再在数据量增长后按哈希规则分片是常见的生产架构。实现时要注意分支规则代码和分片路由代码不要混在同一个函数里否则排查问题时很难定位。4. 用 Python 实现一个带数据分支的最小记忆服务4.1 项目结构下面这个项目结构适用于本地验证和最小 Demoogmemory_demo/ ├── app.py # FastAPI 入口 ├── memory_core.py # 记忆写入、检索、路由核心逻辑 ├── storage.py # SQLite 初始化与数据访问 ├── branch_rules.py # 分支路由规则 ├── models.py # Pydantic 数据模型 ├── requirements.txt # 依赖清单 └── data/ └── memory.db # SQLite 数据库文件这个结构把路由规则、存储访问、核心逻辑分开目的是让每个模块职责单一。后续如果要替换向量库只需改动 storage.py 和 memory_core.py 中的向量相关部分分支规则不需要动。4.2 数据模型设计记忆系统的核心表是 memory_record 表。它既保存原始文本也保存结构化元数据同时通过 memory_embedding 表关联向量数据。CREATE TABLE IF NOT EXISTS memory_record ( memory_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, branch_type TEXT NOT NULL, memory_type TEXT NOT NULL, source TEXT NOT NULL, content TEXT NOT NULL, importance REAL DEFAULT 0.5, status TEXT DEFAULT active, merged_from TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, expire_at TEXT ); CREATE TABLE IF NOT EXISTS memory_embedding ( memory_id TEXT PRIMARY KEY, vector TEXT NOT NULL, model_name TEXT, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_memory_branch ON memory_record(user_id, branch_type, status);字段含义说明字段含义memory_id记忆唯一标识建议用 UUIDuser_id记忆归属用户多用户场景必须隔离branch_type数据分支例如 user_preference、task_eventmemory_type记忆类型例如 fact、preference、eventsource来源例如 user、agent、systemcontent记忆原始内容importance重要程度0 到 1影响后续遗忘优先级status状态active、archived、expired、trashedmerged_from如果该记录由多条记忆合并而来记录来源 IDcreated_at / updated_at时间字段统一使用 UTC ISO 格式expire_at过期时间过期后进入遗忘候选这里要注意时间字段统一使用 UTC不要使用本地时间。否则跨时区部署时分支按时间归档会出现错位问题。4.3 写入流程中的分支路由记忆写入时系统先判断这条记忆应该进入哪个分支再执行存储。分支路由不只是一个字段赋值它可能影响后续的索引、召回策略和过期策略。下面是一个分支路由规则示例# branch_rules.py def route_branch(source: str, memory_type: str, content: str) - str: if source user and memory_type preference: return user_preference if memory_type event: return task_event if memory_type fact and len(content) 50: return long_term_knowledge if memory_type fact: return short_term_fact return general_memory这个示例展示了最基础的关键词维度路由。更复杂的项目可以使用 LLM 抽取结果来决定分支。比如先让模型输出{ source: user, memory_type: preference, importance: 0.9 }再交给路由函数判断。写入核心逻辑import uuid from datetime import datetime, timedelta, timezone def write_memory(user_id, source, memory_type, content, importance0.5): branch route_branch(source, memory_type, content) memory_id str(uuid.uuid4()) now datetime.now(timezone.utc) expire_at now timedelta(days30) if branch in (user_preference, long_term_knowledge): expire_at now timedelta(days365) record { memory_id: memory_id, user_id: user_id, branch_type: branch, memory_type: memory_type, source: source, content: content, importance: importance, status: active, created_at: now.isoformat(), updated_at: now.isoformat(), expire_at: expire_at.isoformat() } save_memory_record(record) vector get_embedding(content) save_memory_embedding(memory_id, vector) return record值得关注的是route_branch这个函数。它是数据分支的唯一决策点所有记忆写入都必须经过它。这样设计的好处是当业务需要调整分支规则时只改一个文件就能生效不需要在多个业务调用点里寻找散落的分支判断。4.4 检索流程中的分支过滤检索时的核心原则是先限定分支范围再做语义召回。如果没有指定分支系统应该使用默认分支或“全分支限权检索”而不是直接不做过滤地全局搜索。def search_memory(user_id, query, branch_listNone, top_k5): if branch_list is None: branch_list [user_preference, task_event, short_term_fact] where_conditions [user_id ?, status active] params [user_id] if branch_list: placeholders ,.join(? for _ in branch_list) where_conditions.append(fbranch_type IN ({placeholders})) params.extend(branch_list) sql fSELECT memory_id, content, branch_type, importance FROM memory_record WHERE { AND .join(where_conditions)} ORDER BY importance DESC, updated_at DESC LIMIT ? # 先按 SQL 条件过滤候选集 candidates query_memory_record(sql, params [top_k * 10]) # 再对候选集做向量相似度排序 query_vec get_embedding(query) ranked [] for record in candidates: emb load_embedding(record[memory_id]) score cosine_similarity(query_vec, emb) ranked.append({**record, score: score}) ranked.sort(keylambda x: x[score], reverseTrue) return ranked[:top_k]这里的实现顺序是先通过结构化条件缩小候选集再做向量检索。不要反过来否则每次查询都会在全量向量库中做相似度计算数据量上来后延迟会明显增加。4.5 分支合并与归档策略记忆系统不能只有写入和检索还需要定时处理分支中的数据状态。合并解决的是重复记忆问题归档解决的是数据膨胀问题。合并策略通常按以下规则处理同一用户在同一个分支下短期重复出现相似内容新记忆的重要性高于旧记忆旧记忆已经连续多次未被检索命中归档策略按时间触发def archive_expired_memories(): now datetime.now(timezone.utc).isoformat() sql UPDATE memory_record SET status archived, updated_at ? WHERE status active AND expire_at ? execute_update(sql, [now, now])这段代码用于把所有过期且仍处于 active 状态的记忆改成 archived。实际生产环境中expired 状态和 archived 状态可以分开处理expired 表示不再参与检索archived 表示仍然保留但降级为低频访问。不要在同一个状态里混用否则统计时很难区分。5. 关键参数和设计取舍5.1 分支数量应该怎么控制数据分支数量是一个典型的取舍问题。分支太少记忆混在一起检索噪声大分支太多路由规则复杂调用方需要记住一堆分支名运维维护成本也高。分支数量优点缺点适用场景3 到 5 个路由简单、检索稳定粒度较粗某些场景仍有噪声个人助手、小型业务6 到 15 个语义隔离清晰需要维护路由规则和使用文档中型团队知识库、多项目 Agent15 个以上高度隔离路由复杂分支管理和监控成本高大型组织、强权限隔离场景推荐做法是先用少量分支跑通完整链路观察检索命中率。当出现“检索结果混杂无关记忆”时再根据失败样本拆分新分支。5.2 记忆系统关键参数速查表参数常见值影响错误表现top_k5 到 10返回记忆条数过大易引入噪声Agent 上下文被无关信息挤占score_threshold0.6 到 0.8低于阈值的结果不返回返回不相关内容检索质量下降importance0 到 1排序权重重要记忆优先低价值记忆长期占坑expire_after_days30 到 365控制记忆自然过期时间过期太短丢失有用信息太长数据膨胀merge_interval_seconds3600 到 86400控制去重合并频率过高增加计算成本过低导致重复记忆残留branch_list 默认值3 个核心分支未指定分支时的候选范围误查全局导致权限混杂或噪声大这里要强调score_threshold 不是越高越好。设置太高时真正有用的记忆可能因为表述差异被过滤掉设置太低时大量低相关记忆进入上下文模型反而更糊涂。建议在测试集上统计相似度分布后再确定阈值。5.3 语义分支与向量集合的关系数据分支落到存储层有两种主流实现方式。第一种是一个分支对应一个独立向量集合。优点是隔离彻底分支之间互不影响权限控制容易缺点是跨分支检索时需要逐个集合查询再合并结果实现相对复杂。第二种是统一向量集合通过 metadata 字段中的 branch_type 标签过滤。优点是实现简单一次查询可以同时支持单分支和多分支缺点是数据量增大后过滤条件对向量检索的效率影响需要评估。两种方式的代码差异主要体现在存储和检索两个环节。使用第二种方式时写入向量时需要附带 metadata{ id: memory_id_1, vector: [0.1, 0.2, 0.3], metadata: { user_id: user_001, branch_type: user_preference, memory_type: preference } }检索时在向量查询请求中带上 filter 条件{ query: [0.1, 0.2, 0.3], limit: 5, filter: { user_id: user_001, branch_type: user_preference } }选择哪种实现取决于数据规模和是否需要严格的物理隔离。个人项目和小型团队建议用第二种复杂度低最容易落地。6. 运行验证从写入到分支查询6.1 启动最小服务这里使用 FastAPI 提供 HTTP 接口便于验证完整流程。启动命令如下uvicorn app:app --host 0.0.0.0 --port 8000启动成功后终端会输出 uvicorn 的运行地址。此时可以访问http://127.0.0.1:8000/docs查看接口文档。这一步确认服务本身没有报错。6.2 写入不同分支的数据写入接口请求示例POST /memories { user_id: user_001, source: user, memory_type: preference, content: 用户喜欢简洁的技术方案不喜欢长篇大论, importance: 0.9 }返回结果中应包含 branch_type 字段值为user_preference。这就是路由函数生效的验证。再写入一条任务事件POST /memories { user_id: user_001, source: agent, memory_type: event, content: 用户在今天 14:00 确认了订单系统的数据库选型, importance: 0.7 }返回结果中 branch_type 应改为task_event。两条数据属于不同分支证明路由规则正常工作。6.3 验证分支过滤查询查询接口请求示例GET /memories?user_iduser_001branch_typeuser_preferencequery喜欢简洁方案正常结果应该只返回 user_preference 分支中的记忆不返回 task_event 分支中的内容。如果查询时不传 branch_type只传 user_id则确认默认分支逻辑是否生效GET /memories?user_iduser_001query订单系统这里要注意观察返回结果是否同时包含 user_preference 和 task_event 分支的数据。如果只返回一个分支需要回头检查search_memory中的默认 branch_list 配置。6.4 验证归档与遗忘手动执行归档函数验证python -c from memory_core import archive_expired_memories; archive_expired_memories()然后在数据库里检查状态变化SELECT memory_id, branch_type, status, expire_at FROM memory_record WHERE user_id user_001;正常情况下未过期的记录仍是 active已过期记录变成 archived。如果所有记录都变成 archived说明写入时 expire_at 计算用的时间基准有问题检查是否使用了错误的时区或者过期时间被设置成当前时间之前。7. 常见问题排查7.1 写入后检索不到可能原因有很多按顺序排查写入时 route_branch 返回的分支与查询时传入的 branch_type 不一致。向量数据没有保存成功memory_embedding 表缺少记录。查询时的 user_id 与写入时的 user_id 不一致。score_threshold 设置过高导致相似度低于阈值的结果被丢弃。数据状态不是 active而是 archived、expired 或 trashed。推荐检查方式先不看向量直接用 SQL 查询 memory_record 表确认记录存在且 status 为 active。然后再确认 query 中传入的过滤条件是否包含该记录的分支。7.2 新记忆覆盖了旧记忆现象是用户修改偏好后新记录写入旧记录仍然存在但 Agent 仍然会读到旧记录。原因是写入新记忆时没有处理旧记忆的冲突旧记录仍处于 active 状态。处理方式在写入相同分支、相同 type 的新记忆时先将旧记忆的状态改为 superseded。增加 version 字段记录同主题记忆的先后版本。需要支持历史回溯时不物理删除旧记录只改变状态。推荐做法是给 memory_record 增加一个 topic_key 字段用来标识“同一主题”。写入时先查询相同 topic_key 的记录如果存在新版本就把旧版本置为更新状态。7.3 表数据膨胀后查询变慢当 memory_record 表数据量持续增长即使加了索引查询也可能变慢。检查方向是否按 user_id 和 branch_type 建立了联合索引。是否定时执行了归档任务active 数据是否被控制在合理范围。向量检索是否在 SQL 过滤之后执行而不是先全量向量搜索再过滤。如果 SQLite 单表数据量已经超过百万行建议把结构化数据迁移到 PostgreSQL把向量数据迁移到专业向量库避免在单个文件数据库上硬扛大数据量。7.4 时间分支错乱现象是按天归档时部分数据被归入错误日期或过期时间判断错误。常见原因是时区不统一。有的服务使用本地时间写入另一个服务使用 UTC 读取导致expire_at和当前时间比较时出现偏差。排查时先确认所有时间字段是否统一使用 UTC ISO 格式再确认写入时datetime.now(timezone.utc)是否被误写成datetime.now()。问题现象常见原因检查方式处理建议写入后检索不到分支条件不一致或向量缺失先查 SQL 再查向量表统一分支命名增加写入日志新记忆覆盖旧记忆失败未处理主题冲突检查 topic_key 查询逻辑写入前先查同主题 active 记录查询变慢没有及时归档或索引缺失查看 SQL 执行计划增加组合索引定期运行归档任务时间分支错乱时区混用检查数据库时间值全部改为 UTC 存储8. 最佳实践与扩展方向8.1 学习环境和生产环境的差异本地 Demo 可以接受 SQLite 和模拟 embedding但生产环境不能这样简单处理。层面本地验证环境生产环境存储SQLitePostgreSQL 或其他关系数据库向量检索Python 列表遍历专用向量库支持索引和分片嵌入服务模拟向量内部嵌入服务需监控稳定性和限流定时任务手动执行脚本独立调度服务记录执行日志鉴权无接口鉴权、用户数据隔离、操作审计配置写死在代码配置中心或环境变量外置在生产环境上线记忆系统前至少要补齐日志、监控、权限、回滚和数据备份这几项。记忆数据是 Agent 行为的上下文依据丢失或写错都会直接影响业务结果。8.2 生产环境必须补齐的保障记忆写入接口需要增加调用方标识、操作审计和实施限流。不能让任意客户端无限写入否则恶意调用可能撑爆存储。批量任务要重试和告警。归档、合并、遗忘任务如果失败不能静默结束应该记录失败原因并触发告警否则数据会持续膨胀。数据备份策略要区分全量备份和增量备份。向量数据也要纳入备份范围不能只备份关系型数据库。更重要的是要建立记忆回滚能力。当一次批量记忆更新导致 Agent 行为异常时能按分支、按用户、按时间段回滚到历史状态。这个能力在设计阶段就要预留不能在事故发生后靠手工改库。8.3 下一步可以怎么扩展数据分支是记忆系统的基础能力下一步可以考虑以下方向。第一接入知识图谱。将 fact 类型记忆抽取为实体和关系用图结构支撑多跳推理。数据分支可以按实体所属领域划分例如产品域、技术域、用户域。第二实现分级遗忘机制。根据重要程度、访问频率、最后一次使用时间综合计算记忆热度热度低的记忆先进入候选归档列表由人工或规则确认后删除。第三支持多 Agent 共享记忆。在 user_id 之上增加 agent_id、team_id 维度让一个团队的多个 Agent 共享知识但用户私有记忆仍然保持隔离。这一步需要更细的权限模型。第四建立记忆反馈闭环。当用户对 Agent 的回答给出负面反馈时回查命中的记忆数据分析是否记忆本身不准确。这样可以把数据分支与模型评估系统连接起来。对新手来说最有价值的练习不是直接接入完整框架而是先手写一遍最小的记忆写入、分支路由、查询和归档流程把数据模型和状态流转理解清楚再去学习向量库和遗忘机制。数据分支看起来只是简单的字段拆分但它决定了记忆系统在真实业务中能不能稳定工作。 AI Agent记忆系统：数据分支设计原理与最小实现
AI Agent 在执行复杂任务时真正限制它表现的往往不是单个模型的能力而是它能不能记住上下文、能不能在关键节点调出正确信息。oGMemory 记忆系统要解决的正是 Agent 跨会话、跨任务的记忆组织问题而数据分支又是决定记忆系统能不能“用起来”的关键设计。这一期分集把视线聚焦在记忆系统与数据分支上先讲清楚数据分支为什么存在再给出一套可以落地的最小实现。很多项目在最初接入记忆能力时通常的做法是“对话结束后把文本塞进向量库下次检索再查出来”。这个思路在 Demo 阶段可行但一旦进入真实业务就会出现一个典型问题所有记忆混在一起用户偏好、任务记录、知识沉淀、短期上下文全部堆在同一个集合里。写入越久检索噪声越大Agent 越来越难分辨哪条信息是当前任务该用的。数据分支的作用就是按业务维度把这些记忆拆开让写入有明确归属让检索有明确范围。1. 记忆系统到底在解决什么问题1.1 没有记忆的 Agent 为什么不够用没有记忆系统的 Agent本质上是一个“每次对话都从零开始”的状态机。模型本身拥有训练阶段沉淀的静态知识但它不知道用户上一次问过什么、当前任务做到哪一步、用户偏好哪种回答风格。对于一次性的问答场景这没有问题但对于需要连续执行多步任务、跨天维护用户关系、持续沉淀团队知识的场景缺记忆就意味着每次都要用户重新交代背景。oGMemory 这类记忆系统的核心目标是给 Agent 增加一个可持续读写的外部状态层。模型本身不可变但记忆数据可以随时间、任务、用户不断更新。Agent 在执行任务前先读取记忆执行过程中写入新记忆任务结束后再整理归档这样它就能形成“越用越懂当前场景”的能力。需要注意的是记忆不是简单地等于聊天记录。聊天记录是原始素材记忆系统要把素材加工成结构化的、可检索的、能按需过滤的数据。这个加工过程包括抽取主题、过滤噪声、判定重要性、设置生命周期、分配数据分支。1.2 “能存下来”和“会用”之间的差距只把文本存进数据库并不是记忆系统。真正的记忆系统要处理四个连续问题第一什么值得记。不是每一句对话都值得写入长期记忆。寒暄、临时计算过程、重复确认信息都应该被过滤掉。真正值得记的是用户偏好、事实结论、任务状态、关键约束。第二记到哪里去。这个问题就是数据分支要解决的。不同性质的记忆需要不同的存储策略比如短期会话上下文可以放在快速缓存中用户长期偏好需要进入稳定的关系型存储或向量库而任务执行记录可能需要按项目维度隔离。第三如何被找到。写入时可读不等于检索时可命中。检索环节要处理相似度计算、过滤条件、排序策略、时效衰减。如果没有分支约束检索时会把不相关的记忆也召回导致上下文被污染。第四什么时候被更新或遗忘。记忆不能只增不减。当用户明确改变偏好、任务状态发生流转、数据超过保留期限时系统要具备更新、合并、降级、删除能力。这四个问题里存储和检索最容易理解但也最容易被低估。很多时候项目跑不起来不是因为模型不行而是因为记忆数据没有组织好导致 Agent 读到了错误的历史信息。1.3 oGMemory 在其中的定位从命名和常见记忆系统设计来看oGMemory 可以理解为一条独立的记忆数据链路重点解决 Agent 记忆的组织、存储、检索和生命周期管理。和直接在业务代码里零散调用向量库不同oGMemory 这类设计会把记忆系统抽象成独立服务对外提供写入、查询、更新、遗忘等接口让上层 Agent 业务只关心记忆内容不关心底层存储细节。本文作为分集解读的第一篇聚焦在数据分支这一层不展开讨论底层模型、检索算法、遗忘机制的全部细节。读完这一篇你能完成三件事理解记忆系统的整体分层掌握数据分支的划分思路用一套最小代码跑通“写入分支、按分支查询、分支归档”的完整流程。2. 搭建最小记忆系统需要先确认的核心组件2.1 记忆系统的五层结构一个可工作的记忆系统通常包含五层感知层、编码层、存储层、检索层、遗忘与合并层。数据分支主要落在存储层和检索层之间但每一层都会影响分支设计。感知层负责识别什么信息值得写入记忆。它可能是对话中断言抽取、任务状态识别、用户行为事件接收。感知层的输出是“一条候选记忆”这部分通常要依赖 LLM 或规则引擎完成。编码层负责把文本转换成可以检索的形态。最常见的方式是文本向量化同时保留原文和结构化字段。编码层还要生成主题标签、时间戳、重要程度、来源标识等元数据。存储层负责持久化。向量数据进入向量库结构化字段进入关系型数据库原始内容可能需要对象存储。数据分支在这一层体现为不同的集合、表或分片。检索层负责把用户当前的问题转换成查询条件从正确的分支中召回记忆。检索不是只做语义相似度还需要结合时间范围、用户身份、分支类型、权限范围等条件。遗忘与合并层负责控制记忆生命周期。它解决记忆过期、冲突覆盖、重复合并、降级归档等问题。没有这一层记忆数据会无限膨胀检索质量会持续下降。2.2 组件选型和环境准备记忆系统的选型不存在一套通用标准答案但通常会涉及以下几类组件组件职责常见选型参考LLM文本理解、信息抽取、摘要生成业务现有的大模型服务Embedding 服务将文本转换为向量本地嵌入模型接口或平台嵌入服务向量库存放向量数据支持语义检索轻量可用 Chroma、生产环境可用 Milvus 等关系型数据库存放记忆记录、分支元数据、状态SQLite 适合本地验证生产可用 PostgreSQL 等缓存存储高频访问的短期记忆Redis 等定时任务执行归档、合并、遗忘清理APScheduler、Celery Beat 或云厂商定时任务这里要特别说明本文的示例代码为了保持最小可运行使用 SQLite 存储结构化数据向量部分用一个本地 Python 列表和一个模拟的 embedding 函数代替。这样做的目的是先把数据分支逻辑讲清楚。真实项目接入时把 embedding 函数替换成内部嵌入服务把向量列表迁移到向量库即可。2.3 环境准备清单在开始写代码之前先确认本机环境满足以下条件Python 3.10 或更高版本已安装 FastAPI 和 uvicorn已安装 SQLite3 驱动Python 自带的 sqlite3 就够用已安装 pandas 或直接使用 Python 标准库处理数据一个可用的 embedding 函数开发阶段可以用随机向量模拟但验证阶段建议接入真实嵌入服务如果原始项目没有明确版本落地前务必先确认依赖版本和 Python 版本兼容性。这里给出一个 requirements 示例fastapi0.110.0 uvicorn0.29.0 pydantic2.6.0 numpy1.26.4 apscheduler3.10.4注意版本号只是参考。实际项目如果使用已有依赖锁文件以仓库中的版本为准不要直接复制最新版本号到生产环境。3. 数据分支是什么为什么记忆系统离不开它3.1 数据分支的定义数据分支是指按照某种业务维度把记忆数据划分成不同的逻辑通道。每个通道拥有独立的写入规则、存储位置、检索范围和生命周期策略。可以把数据分支理解成“记忆的分类目录”。没有分类目录时所有记忆是一条无序的长河有了分类目录后写入时先判断这条记忆属于哪个分类检索时只从对应分类中查找效率和准确率都会明显提升。在 oGMemory 这类 Agent 记忆系统中分支不是一个可有可无的优化项而是决定记忆能否被正确使用的关键设计。原因是 Agent 的记忆数据天然带有强烈的“上下文依赖”特征。用户说“我喜欢简洁的回答”这是一条偏好记忆用户说“项目 A 的数据库连接串已经改好了”这是一条任务事件记忆用户说“本周五要上线”这是一条计划记忆。这三者的使用场景完全不同混在一起存储检索时很难一次性命中正确信息。3.2 常见分支维度实际项目中记忆系统通常不会只使用一个分支维度而是几个维度组合使用。常用维度包括分支维度说明示例按时间区分短期上下文、中期工作记忆、长期沉淀short_term、mid_term、long_term按来源区分用户提供、Agent 推导、系统事件、外部知识user、agent、system、knowledge按类型区分事实、偏好、事件、技能、约束fact、preference、event、skill、constraint按业务域区分不同项目、不同知识库、不同团队project_a、project_b、wiki_base按生命周期区分活跃、候选归档、已过期、废弃active、archived、expired、trashed分支维度的选择不是越多越好。每增加一个维度写入时的路由判断就更复杂检索时的过滤条件也更多运维成本会指数上升。多数中小型记忆系统从“来源 类型 时间”三个维度起步就足够之后根据真实检索效果再细化。3.3 数据分支与分库分表的区别数据分支容易被误解成“分库分表”但它们解决的问题不同。分库分表解决的是存储容量和写入性能问题。当单表数据量达到千万级索引失效、写入变慢这时需要把数据按照哈希或范围分散到多个物理存储中。分库分表是物理层的伸缩方案对业务透明上层 SQL 大多数情况下不应该感知到分片逻辑。数据分支解决的是语义隔离和检索范围问题。它强调“哪些记忆属于哪个业务范围”是逻辑层的分类设计。即使数据量很小也需要做数据分支否则 Agent 会在两三百条记忆里被噪声干扰。两者可以叠加使用。先按业务维度做数据分支再在数据量增长后按哈希规则分片是常见的生产架构。实现时要注意分支规则代码和分片路由代码不要混在同一个函数里否则排查问题时很难定位。4. 用 Python 实现一个带数据分支的最小记忆服务4.1 项目结构下面这个项目结构适用于本地验证和最小 Demoogmemory_demo/ ├── app.py # FastAPI 入口 ├── memory_core.py # 记忆写入、检索、路由核心逻辑 ├── storage.py # SQLite 初始化与数据访问 ├── branch_rules.py # 分支路由规则 ├── models.py # Pydantic 数据模型 ├── requirements.txt # 依赖清单 └── data/ └── memory.db # SQLite 数据库文件这个结构把路由规则、存储访问、核心逻辑分开目的是让每个模块职责单一。后续如果要替换向量库只需改动 storage.py 和 memory_core.py 中的向量相关部分分支规则不需要动。4.2 数据模型设计记忆系统的核心表是 memory_record 表。它既保存原始文本也保存结构化元数据同时通过 memory_embedding 表关联向量数据。CREATE TABLE IF NOT EXISTS memory_record ( memory_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, branch_type TEXT NOT NULL, memory_type TEXT NOT NULL, source TEXT NOT NULL, content TEXT NOT NULL, importance REAL DEFAULT 0.5, status TEXT DEFAULT active, merged_from TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, expire_at TEXT ); CREATE TABLE IF NOT EXISTS memory_embedding ( memory_id TEXT PRIMARY KEY, vector TEXT NOT NULL, model_name TEXT, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_memory_branch ON memory_record(user_id, branch_type, status);字段含义说明字段含义memory_id记忆唯一标识建议用 UUIDuser_id记忆归属用户多用户场景必须隔离branch_type数据分支例如 user_preference、task_eventmemory_type记忆类型例如 fact、preference、eventsource来源例如 user、agent、systemcontent记忆原始内容importance重要程度0 到 1影响后续遗忘优先级status状态active、archived、expired、trashedmerged_from如果该记录由多条记忆合并而来记录来源 IDcreated_at / updated_at时间字段统一使用 UTC ISO 格式expire_at过期时间过期后进入遗忘候选这里要注意时间字段统一使用 UTC不要使用本地时间。否则跨时区部署时分支按时间归档会出现错位问题。4.3 写入流程中的分支路由记忆写入时系统先判断这条记忆应该进入哪个分支再执行存储。分支路由不只是一个字段赋值它可能影响后续的索引、召回策略和过期策略。下面是一个分支路由规则示例# branch_rules.py def route_branch(source: str, memory_type: str, content: str) - str: if source user and memory_type preference: return user_preference if memory_type event: return task_event if memory_type fact and len(content) 50: return long_term_knowledge if memory_type fact: return short_term_fact return general_memory这个示例展示了最基础的关键词维度路由。更复杂的项目可以使用 LLM 抽取结果来决定分支。比如先让模型输出{ source: user, memory_type: preference, importance: 0.9 }再交给路由函数判断。写入核心逻辑import uuid from datetime import datetime, timedelta, timezone def write_memory(user_id, source, memory_type, content, importance0.5): branch route_branch(source, memory_type, content) memory_id str(uuid.uuid4()) now datetime.now(timezone.utc) expire_at now timedelta(days30) if branch in (user_preference, long_term_knowledge): expire_at now timedelta(days365) record { memory_id: memory_id, user_id: user_id, branch_type: branch, memory_type: memory_type, source: source, content: content, importance: importance, status: active, created_at: now.isoformat(), updated_at: now.isoformat(), expire_at: expire_at.isoformat() } save_memory_record(record) vector get_embedding(content) save_memory_embedding(memory_id, vector) return record值得关注的是route_branch这个函数。它是数据分支的唯一决策点所有记忆写入都必须经过它。这样设计的好处是当业务需要调整分支规则时只改一个文件就能生效不需要在多个业务调用点里寻找散落的分支判断。4.4 检索流程中的分支过滤检索时的核心原则是先限定分支范围再做语义召回。如果没有指定分支系统应该使用默认分支或“全分支限权检索”而不是直接不做过滤地全局搜索。def search_memory(user_id, query, branch_listNone, top_k5): if branch_list is None: branch_list [user_preference, task_event, short_term_fact] where_conditions [user_id ?, status active] params [user_id] if branch_list: placeholders ,.join(? for _ in branch_list) where_conditions.append(fbranch_type IN ({placeholders})) params.extend(branch_list) sql fSELECT memory_id, content, branch_type, importance FROM memory_record WHERE { AND .join(where_conditions)} ORDER BY importance DESC, updated_at DESC LIMIT ? # 先按 SQL 条件过滤候选集 candidates query_memory_record(sql, params [top_k * 10]) # 再对候选集做向量相似度排序 query_vec get_embedding(query) ranked [] for record in candidates: emb load_embedding(record[memory_id]) score cosine_similarity(query_vec, emb) ranked.append({**record, score: score}) ranked.sort(keylambda x: x[score], reverseTrue) return ranked[:top_k]这里的实现顺序是先通过结构化条件缩小候选集再做向量检索。不要反过来否则每次查询都会在全量向量库中做相似度计算数据量上来后延迟会明显增加。4.5 分支合并与归档策略记忆系统不能只有写入和检索还需要定时处理分支中的数据状态。合并解决的是重复记忆问题归档解决的是数据膨胀问题。合并策略通常按以下规则处理同一用户在同一个分支下短期重复出现相似内容新记忆的重要性高于旧记忆旧记忆已经连续多次未被检索命中归档策略按时间触发def archive_expired_memories(): now datetime.now(timezone.utc).isoformat() sql UPDATE memory_record SET status archived, updated_at ? WHERE status active AND expire_at ? execute_update(sql, [now, now])这段代码用于把所有过期且仍处于 active 状态的记忆改成 archived。实际生产环境中expired 状态和 archived 状态可以分开处理expired 表示不再参与检索archived 表示仍然保留但降级为低频访问。不要在同一个状态里混用否则统计时很难区分。5. 关键参数和设计取舍5.1 分支数量应该怎么控制数据分支数量是一个典型的取舍问题。分支太少记忆混在一起检索噪声大分支太多路由规则复杂调用方需要记住一堆分支名运维维护成本也高。分支数量优点缺点适用场景3 到 5 个路由简单、检索稳定粒度较粗某些场景仍有噪声个人助手、小型业务6 到 15 个语义隔离清晰需要维护路由规则和使用文档中型团队知识库、多项目 Agent15 个以上高度隔离路由复杂分支管理和监控成本高大型组织、强权限隔离场景推荐做法是先用少量分支跑通完整链路观察检索命中率。当出现“检索结果混杂无关记忆”时再根据失败样本拆分新分支。5.2 记忆系统关键参数速查表参数常见值影响错误表现top_k5 到 10返回记忆条数过大易引入噪声Agent 上下文被无关信息挤占score_threshold0.6 到 0.8低于阈值的结果不返回返回不相关内容检索质量下降importance0 到 1排序权重重要记忆优先低价值记忆长期占坑expire_after_days30 到 365控制记忆自然过期时间过期太短丢失有用信息太长数据膨胀merge_interval_seconds3600 到 86400控制去重合并频率过高增加计算成本过低导致重复记忆残留branch_list 默认值3 个核心分支未指定分支时的候选范围误查全局导致权限混杂或噪声大这里要强调score_threshold 不是越高越好。设置太高时真正有用的记忆可能因为表述差异被过滤掉设置太低时大量低相关记忆进入上下文模型反而更糊涂。建议在测试集上统计相似度分布后再确定阈值。5.3 语义分支与向量集合的关系数据分支落到存储层有两种主流实现方式。第一种是一个分支对应一个独立向量集合。优点是隔离彻底分支之间互不影响权限控制容易缺点是跨分支检索时需要逐个集合查询再合并结果实现相对复杂。第二种是统一向量集合通过 metadata 字段中的 branch_type 标签过滤。优点是实现简单一次查询可以同时支持单分支和多分支缺点是数据量增大后过滤条件对向量检索的效率影响需要评估。两种方式的代码差异主要体现在存储和检索两个环节。使用第二种方式时写入向量时需要附带 metadata{ id: memory_id_1, vector: [0.1, 0.2, 0.3], metadata: { user_id: user_001, branch_type: user_preference, memory_type: preference } }检索时在向量查询请求中带上 filter 条件{ query: [0.1, 0.2, 0.3], limit: 5, filter: { user_id: user_001, branch_type: user_preference } }选择哪种实现取决于数据规模和是否需要严格的物理隔离。个人项目和小型团队建议用第二种复杂度低最容易落地。6. 运行验证从写入到分支查询6.1 启动最小服务这里使用 FastAPI 提供 HTTP 接口便于验证完整流程。启动命令如下uvicorn app:app --host 0.0.0.0 --port 8000启动成功后终端会输出 uvicorn 的运行地址。此时可以访问http://127.0.0.1:8000/docs查看接口文档。这一步确认服务本身没有报错。6.2 写入不同分支的数据写入接口请求示例POST /memories { user_id: user_001, source: user, memory_type: preference, content: 用户喜欢简洁的技术方案不喜欢长篇大论, importance: 0.9 }返回结果中应包含 branch_type 字段值为user_preference。这就是路由函数生效的验证。再写入一条任务事件POST /memories { user_id: user_001, source: agent, memory_type: event, content: 用户在今天 14:00 确认了订单系统的数据库选型, importance: 0.7 }返回结果中 branch_type 应改为task_event。两条数据属于不同分支证明路由规则正常工作。6.3 验证分支过滤查询查询接口请求示例GET /memories?user_iduser_001branch_typeuser_preferencequery喜欢简洁方案正常结果应该只返回 user_preference 分支中的记忆不返回 task_event 分支中的内容。如果查询时不传 branch_type只传 user_id则确认默认分支逻辑是否生效GET /memories?user_iduser_001query订单系统这里要注意观察返回结果是否同时包含 user_preference 和 task_event 分支的数据。如果只返回一个分支需要回头检查search_memory中的默认 branch_list 配置。6.4 验证归档与遗忘手动执行归档函数验证python -c from memory_core import archive_expired_memories; archive_expired_memories()然后在数据库里检查状态变化SELECT memory_id, branch_type, status, expire_at FROM memory_record WHERE user_id user_001;正常情况下未过期的记录仍是 active已过期记录变成 archived。如果所有记录都变成 archived说明写入时 expire_at 计算用的时间基准有问题检查是否使用了错误的时区或者过期时间被设置成当前时间之前。7. 常见问题排查7.1 写入后检索不到可能原因有很多按顺序排查写入时 route_branch 返回的分支与查询时传入的 branch_type 不一致。向量数据没有保存成功memory_embedding 表缺少记录。查询时的 user_id 与写入时的 user_id 不一致。score_threshold 设置过高导致相似度低于阈值的结果被丢弃。数据状态不是 active而是 archived、expired 或 trashed。推荐检查方式先不看向量直接用 SQL 查询 memory_record 表确认记录存在且 status 为 active。然后再确认 query 中传入的过滤条件是否包含该记录的分支。7.2 新记忆覆盖了旧记忆现象是用户修改偏好后新记录写入旧记录仍然存在但 Agent 仍然会读到旧记录。原因是写入新记忆时没有处理旧记忆的冲突旧记录仍处于 active 状态。处理方式在写入相同分支、相同 type 的新记忆时先将旧记忆的状态改为 superseded。增加 version 字段记录同主题记忆的先后版本。需要支持历史回溯时不物理删除旧记录只改变状态。推荐做法是给 memory_record 增加一个 topic_key 字段用来标识“同一主题”。写入时先查询相同 topic_key 的记录如果存在新版本就把旧版本置为更新状态。7.3 表数据膨胀后查询变慢当 memory_record 表数据量持续增长即使加了索引查询也可能变慢。检查方向是否按 user_id 和 branch_type 建立了联合索引。是否定时执行了归档任务active 数据是否被控制在合理范围。向量检索是否在 SQL 过滤之后执行而不是先全量向量搜索再过滤。如果 SQLite 单表数据量已经超过百万行建议把结构化数据迁移到 PostgreSQL把向量数据迁移到专业向量库避免在单个文件数据库上硬扛大数据量。7.4 时间分支错乱现象是按天归档时部分数据被归入错误日期或过期时间判断错误。常见原因是时区不统一。有的服务使用本地时间写入另一个服务使用 UTC 读取导致expire_at和当前时间比较时出现偏差。排查时先确认所有时间字段是否统一使用 UTC ISO 格式再确认写入时datetime.now(timezone.utc)是否被误写成datetime.now()。问题现象常见原因检查方式处理建议写入后检索不到分支条件不一致或向量缺失先查 SQL 再查向量表统一分支命名增加写入日志新记忆覆盖旧记忆失败未处理主题冲突检查 topic_key 查询逻辑写入前先查同主题 active 记录查询变慢没有及时归档或索引缺失查看 SQL 执行计划增加组合索引定期运行归档任务时间分支错乱时区混用检查数据库时间值全部改为 UTC 存储8. 最佳实践与扩展方向8.1 学习环境和生产环境的差异本地 Demo 可以接受 SQLite 和模拟 embedding但生产环境不能这样简单处理。层面本地验证环境生产环境存储SQLitePostgreSQL 或其他关系数据库向量检索Python 列表遍历专用向量库支持索引和分片嵌入服务模拟向量内部嵌入服务需监控稳定性和限流定时任务手动执行脚本独立调度服务记录执行日志鉴权无接口鉴权、用户数据隔离、操作审计配置写死在代码配置中心或环境变量外置在生产环境上线记忆系统前至少要补齐日志、监控、权限、回滚和数据备份这几项。记忆数据是 Agent 行为的上下文依据丢失或写错都会直接影响业务结果。8.2 生产环境必须补齐的保障记忆写入接口需要增加调用方标识、操作审计和实施限流。不能让任意客户端无限写入否则恶意调用可能撑爆存储。批量任务要重试和告警。归档、合并、遗忘任务如果失败不能静默结束应该记录失败原因并触发告警否则数据会持续膨胀。数据备份策略要区分全量备份和增量备份。向量数据也要纳入备份范围不能只备份关系型数据库。更重要的是要建立记忆回滚能力。当一次批量记忆更新导致 Agent 行为异常时能按分支、按用户、按时间段回滚到历史状态。这个能力在设计阶段就要预留不能在事故发生后靠手工改库。8.3 下一步可以怎么扩展数据分支是记忆系统的基础能力下一步可以考虑以下方向。第一接入知识图谱。将 fact 类型记忆抽取为实体和关系用图结构支撑多跳推理。数据分支可以按实体所属领域划分例如产品域、技术域、用户域。第二实现分级遗忘机制。根据重要程度、访问频率、最后一次使用时间综合计算记忆热度热度低的记忆先进入候选归档列表由人工或规则确认后删除。第三支持多 Agent 共享记忆。在 user_id 之上增加 agent_id、team_id 维度让一个团队的多个 Agent 共享知识但用户私有记忆仍然保持隔离。这一步需要更细的权限模型。第四建立记忆反馈闭环。当用户对 Agent 的回答给出负面反馈时回查命中的记忆数据分析是否记忆本身不准确。这样可以把数据分支与模型评估系统连接起来。对新手来说最有价值的练习不是直接接入完整框架而是先手写一遍最小的记忆写入、分支路由、查询和归档流程把数据模型和状态流转理解清楚再去学习向量库和遗忘机制。数据分支看起来只是简单的字段拆分但它决定了记忆系统在真实业务中能不能稳定工作。