：數(shù)據(jù)分支設(shè)計原理與最小實現(xiàn))
AI Agent 在執(zhí)行復雜任務(wù)時真正限制它表現(xiàn)的往往不是單個模型的能力而是它能不能記住上下文、能不能在關(guān)鍵節(jié)點調(diào)出正確信息。oGMemory 記憶系統(tǒng)要解決的正是 Agent 跨會話、跨任務(wù)的記憶組織問題而數(shù)據(jù)分支又是決定記憶系統(tǒng)能不能“用起來”的關(guān)鍵設(shè)計。這一期分集把視線聚焦在記憶系統(tǒng)與數(shù)據(jù)分支上先講清楚數(shù)據(jù)分支為什么存在再給出一套可以落地的最小實現(xiàn)。很多項目在最初接入記憶能力時通常的做法是“對話結(jié)束后把文本塞進向量庫下次檢索再查出來”。這個思路在 Demo 階段可行但一旦進入真實業(yè)務(wù)就會出現(xiàn)一個典型問題所有記憶混在一起用戶偏好、任務(wù)記錄、知識沉淀、短期上下文全部堆在同一個集合里。寫入越久檢索噪聲越大Agent 越來越難分辨哪條信息是當前任務(wù)該用的。數(shù)據(jù)分支的作用就是按業(yè)務(wù)維度把這些記憶拆開讓寫入有明確歸屬讓檢索有明確范圍。1. 記憶系統(tǒng)到底在解決什么問題1.1 沒有記憶的 Agent 為什么不夠用沒有記憶系統(tǒng)的 Agent本質(zhì)上是一個“每次對話都從零開始”的狀態(tài)機。模型本身擁有訓練階段沉淀的靜態(tài)知識但它不知道用戶上一次問過什么、當前任務(wù)做到哪一步、用戶偏好哪種回答風格。對于一次性的問答場景這沒有問題但對于需要連續(xù)執(zhí)行多步任務(wù)、跨天維護用戶關(guān)系、持續(xù)沉淀團隊知識的場景缺記憶就意味著每次都要用戶重新交代背景。oGMemory 這類記憶系統(tǒng)的核心目標是給 Agent 增加一個可持續(xù)讀寫的外部狀態(tài)層。模型本身不可變但記憶數(shù)據(jù)可以隨時間、任務(wù)、用戶不斷更新。Agent 在執(zhí)行任務(wù)前先讀取記憶執(zhí)行過程中寫入新記憶任務(wù)結(jié)束后再整理歸檔這樣它就能形成“越用越懂當前場景”的能力。需要注意的是記憶不是簡單地等于聊天記錄。聊天記錄是原始素材記憶系統(tǒng)要把素材加工成結(jié)構(gòu)化的、可檢索的、能按需過濾的數(shù)據(jù)。這個加工過程包括抽取主題、過濾噪聲、判定重要性、設(shè)置生命周期、分配數(shù)據(jù)分支。1.2 “能存下來”和“會用”之間的差距只把文本存進數(shù)據(jù)庫并不是記憶系統(tǒng)。真正的記憶系統(tǒng)要處理四個連續(xù)問題第一什么值得記。不是每一句對話都值得寫入長期記憶。寒暄、臨時計算過程、重復確認信息都應(yīng)該被過濾掉。真正值得記的是用戶偏好、事實結(jié)論、任務(wù)狀態(tài)、關(guān)鍵約束。第二記到哪里去。這個問題就是數(shù)據(jù)分支要解決的。不同性質(zhì)的記憶需要不同的存儲策略比如短期會話上下文可以放在快速緩存中用戶長期偏好需要進入穩(wěn)定的關(guān)系型存儲或向量庫而任務(wù)執(zhí)行記錄可能需要按項目維度隔離。第三如何被找到。寫入時可讀不等于檢索時可命中。檢索環(huán)節(jié)要處理相似度計算、過濾條件、排序策略、時效衰減。如果沒有分支約束檢索時會把不相關(guān)的記憶也召回導致上下文被污染。第四什么時候被更新或遺忘。記憶不能只增不減。當用戶明確改變偏好、任務(wù)狀態(tài)發(fā)生流轉(zhuǎn)、數(shù)據(jù)超過保留期限時系統(tǒng)要具備更新、合并、降級、刪除能力。這四個問題里存儲和檢索最容易理解但也最容易被低估。很多時候項目跑不起來不是因為模型不行而是因為記憶數(shù)據(jù)沒有組織好導致 Agent 讀到了錯誤的歷史信息。1.3 oGMemory 在其中的定位從命名和常見記憶系統(tǒng)設(shè)計來看oGMemory 可以理解為一條獨立的記憶數(shù)據(jù)鏈路重點解決 Agent 記憶的組織、存儲、檢索和生命周期管理。和直接在業(yè)務(wù)代碼里零散調(diào)用向量庫不同oGMemory 這類設(shè)計會把記憶系統(tǒng)抽象成獨立服務(wù)對外提供寫入、查詢、更新、遺忘等接口讓上層 Agent 業(yè)務(wù)只關(guān)心記憶內(nèi)容不關(guān)心底層存儲細節(jié)。本文作為分集解讀的第一篇聚焦在數(shù)據(jù)分支這一層不展開討論底層模型、檢索算法、遺忘機制的全部細節(jié)。讀完這一篇你能完成三件事理解記憶系統(tǒng)的整體分層掌握數(shù)據(jù)分支的劃分思路用一套最小代碼跑通“寫入分支、按分支查詢、分支歸檔”的完整流程。2. 搭建最小記憶系統(tǒng)需要先確認的核心組件2.1 記憶系統(tǒng)的五層結(jié)構(gòu)一個可工作的記憶系統(tǒng)通常包含五層感知層、編碼層、存儲層、檢索層、遺忘與合并層。數(shù)據(jù)分支主要落在存儲層和檢索層之間但每一層都會影響分支設(shè)計。感知層負責識別什么信息值得寫入記憶。它可能是對話中斷言抽取、任務(wù)狀態(tài)識別、用戶行為事件接收。感知層的輸出是“一條候選記憶”這部分通常要依賴 LLM 或規(guī)則引擎完成。編碼層負責把文本轉(zhuǎn)換成可以檢索的形態(tài)。最常見的方式是文本向量化同時保留原文和結(jié)構(gòu)化字段。編碼層還要生成主題標簽、時間戳、重要程度、來源標識等元數(shù)據(jù)。存儲層負責持久化。向量數(shù)據(jù)進入向量庫結(jié)構(gòu)化字段進入關(guān)系型數(shù)據(jù)庫原始內(nèi)容可能需要對象存儲。數(shù)據(jù)分支在這一層體現(xiàn)為不同的集合、表或分片。檢索層負責把用戶當前的問題轉(zhuǎn)換成查詢條件從正確的分支中召回記憶。檢索不是只做語義相似度還需要結(jié)合時間范圍、用戶身份、分支類型、權(quán)限范圍等條件。遺忘與合并層負責控制記憶生命周期。它解決記憶過期、沖突覆蓋、重復合并、降級歸檔等問題。沒有這一層記憶數(shù)據(jù)會無限膨脹檢索質(zhì)量會持續(xù)下降。2.2 組件選型和環(huán)境準備記憶系統(tǒng)的選型不存在一套通用標準答案但通常會涉及以下幾類組件組件職責常見選型參考LLM文本理解、信息抽取、摘要生成業(yè)務(wù)現(xiàn)有的大模型服務(wù)Embedding 服務(wù)將文本轉(zhuǎn)換為向量本地嵌入模型接口或平臺嵌入服務(wù)向量庫存放向量數(shù)據(jù)支持語義檢索輕量可用 Chroma、生產(chǎn)環(huán)境可用 Milvus 等關(guān)系型數(shù)據(jù)庫存放記憶記錄、分支元數(shù)據(jù)、狀態(tài)SQLite 適合本地驗證生產(chǎn)可用 PostgreSQL 等緩存存儲高頻訪問的短期記憶Redis 等定時任務(wù)執(zhí)行歸檔、合并、遺忘清理APScheduler、Celery Beat 或云廠商定時任務(wù)這里要特別說明本文的示例代碼為了保持最小可運行使用 SQLite 存儲結(jié)構(gòu)化數(shù)據(jù)向量部分用一個本地 Python 列表和一個模擬的 embedding 函數(shù)代替。這樣做的目的是先把數(shù)據(jù)分支邏輯講清楚。真實項目接入時把 embedding 函數(shù)替換成內(nèi)部嵌入服務(wù)把向量列表遷移到向量庫即可。2.3 環(huán)境準備清單在開始寫代碼之前先確認本機環(huán)境滿足以下條件Python 3.10 或更高版本已安裝 FastAPI 和 uvicorn已安裝 SQLite3 驅(qū)動Python 自帶的 sqlite3 就夠用已安裝 pandas 或直接使用 Python 標準庫處理數(shù)據(jù)一個可用的 embedding 函數(shù)開發(fā)階段可以用隨機向量模擬但驗證階段建議接入真實嵌入服務(wù)如果原始項目沒有明確版本落地前務(wù)必先確認依賴版本和 Python 版本兼容性。這里給出一個 requirements 示例fastapi0.110.0 uvicorn0.29.0 pydantic2.6.0 numpy1.26.4 apscheduler3.10.4注意版本號只是參考。實際項目如果使用已有依賴鎖文件以倉庫中的版本為準不要直接復制最新版本號到生產(chǎn)環(huán)境。3. 數(shù)據(jù)分支是什么為什么記憶系統(tǒng)離不開它3.1 數(shù)據(jù)分支的定義數(shù)據(jù)分支是指按照某種業(yè)務(wù)維度把記憶數(shù)據(jù)劃分成不同的邏輯通道。每個通道擁有獨立的寫入規(guī)則、存儲位置、檢索范圍和生命周期策略?？梢园褦?shù)據(jù)分支理解成“記憶的分類目錄”。沒有分類目錄時所有記憶是一條無序的長河有了分類目錄后寫入時先判斷這條記憶屬于哪個分類檢索時只從對應(yīng)分類中查找效率和準確率都會明顯提升。在 oGMemory 這類 Agent 記憶系統(tǒng)中分支不是一個可有可無的優(yōu)化項而是決定記憶能否被正確使用的關(guān)鍵設(shè)計。原因是 Agent 的記憶數(shù)據(jù)天然帶有強烈的“上下文依賴”特征。用戶說“我喜歡簡潔的回答”這是一條偏好記憶用戶說“項目 A 的數(shù)據(jù)庫連接串已經(jīng)改好了”這是一條任務(wù)事件記憶用戶說“本周五要上線”這是一條計劃記憶。這三者的使用場景完全不同混在一起存儲檢索時很難一次性命中正確信息。3.2 常見分支維度實際項目中記憶系統(tǒng)通常不會只使用一個分支維度而是幾個維度組合使用。常用維度包括分支維度說明示例按時間區(qū)分短期上下文、中期工作記憶、長期沉淀short_term、mid_term、long_term按來源區(qū)分用戶提供、Agent 推導、系統(tǒng)事件、外部知識user、agent、system、knowledge按類型區(qū)分事實、偏好、事件、技能、約束fact、preference、event、skill、constraint按業(yè)務(wù)域區(qū)分不同項目、不同知識庫、不同團隊project_a、project_b、wiki_base按生命周期區(qū)分活躍、候選歸檔、已過期、廢棄active、archived、expired、trashed分支維度的選擇不是越多越好。每增加一個維度寫入時的路由判斷就更復雜檢索時的過濾條件也更多運維成本會指數(shù)上升。多數(shù)中小型記憶系統(tǒng)從“來源 類型 時間”三個維度起步就足夠之后根據(jù)真實檢索效果再細化。3.3 數(shù)據(jù)分支與分庫分表的區(qū)別數(shù)據(jù)分支容易被誤解成“分庫分表”但它們解決的問題不同。分庫分表解決的是存儲容量和寫入性能問題。當單表數(shù)據(jù)量達到千萬級索引失效、寫入變慢這時需要把數(shù)據(jù)按照哈?；蚍秶稚⒌蕉鄠€物理存儲中。分庫分表是物理層的伸縮方案對業(yè)務(wù)透明上層 SQL 大多數(shù)情況下不應(yīng)該感知到分片邏輯。數(shù)據(jù)分支解決的是語義隔離和檢索范圍問題。它強調(diào)“哪些記憶屬于哪個業(yè)務(wù)范圍”是邏輯層的分類設(shè)計。即使數(shù)據(jù)量很小也需要做數(shù)據(jù)分支否則 Agent 會在兩三百條記憶里被噪聲干擾。兩者可以疊加使用。先按業(yè)務(wù)維度做數(shù)據(jù)分支再在數(shù)據(jù)量增長后按哈希規(guī)則分片是常見的生產(chǎn)架構(gòu)。實現(xiàn)時要注意分支規(guī)則代碼和分片路由代碼不要混在同一個函數(shù)里否則排查問題時很難定位。4. 用 Python 實現(xiàn)一個帶數(shù)據(jù)分支的最小記憶服務(wù)4.1 項目結(jié)構(gòu)下面這個項目結(jié)構(gòu)適用于本地驗證和最小 Demoogmemory_demo/ ├── app.py # FastAPI 入口 ├── memory_core.py # 記憶寫入、檢索、路由核心邏輯 ├── storage.py # SQLite 初始化與數(shù)據(jù)訪問 ├── branch_rules.py # 分支路由規(guī)則 ├── models.py # Pydantic 數(shù)據(jù)模型 ├── requirements.txt # 依賴清單 └── data/ └── memory.db # SQLite 數(shù)據(jù)庫文件這個結(jié)構(gòu)把路由規(guī)則、存儲訪問、核心邏輯分開目的是讓每個模塊職責單一。后續(xù)如果要替換向量庫只需改動 storage.py 和 memory_core.py 中的向量相關(guān)部分分支規(guī)則不需要動。4.2 數(shù)據(jù)模型設(shè)計記憶系統(tǒng)的核心表是 memory_record 表。它既保存原始文本也保存結(jié)構(gòu)化元數(shù)據(jù)同時通過 memory_embedding 表關(guān)聯(lián)向量數(shù)據(jù)。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數(shù)據(jù)分支例如 user_preference、task_eventmemory_type記憶類型例如 fact、preference、eventsource來源例如 user、agent、systemcontent記憶原始內(nèi)容importance重要程度0 到 1影響后續(xù)遺忘優(yōu)先級status狀態(tài)active、archived、expired、trashedmerged_from如果該記錄由多條記憶合并而來記錄來源 IDcreated_at / updated_at時間字段統(tǒng)一使用 UTC ISO 格式expire_at過期時間過期后進入遺忘候選這里要注意時間字段統(tǒng)一使用 UTC不要使用本地時間。否則跨時區(qū)部署時分支按時間歸檔會出現(xiàn)錯位問題。4.3 寫入流程中的分支路由記憶寫入時系統(tǒng)先判斷這條記憶應(yīng)該進入哪個分支再執(zhí)行存儲。分支路由不只是一個字段賦值它可能影響后續(xù)的索引、召回策略和過期策略。下面是一個分支路由規(guī)則示例# 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這個示例展示了最基礎(chǔ)的關(guān)鍵詞維度路由。更復雜的項目可以使用 LLM 抽取結(jié)果來決定分支。比如先讓模型輸出{ source: user, memory_type: preference, importance: 0.9 }再交給路由函數(shù)判斷。寫入核心邏輯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值得關(guān)注的是route_branch這個函數(shù)。它是數(shù)據(jù)分支的唯一決策點所有記憶寫入都必須經(jīng)過它。這樣設(shè)計的好處是當業(yè)務(wù)需要調(diào)整分支規(guī)則時只改一個文件就能生效不需要在多個業(yè)務(wù)調(diào)用點里尋找散落的分支判斷。4.4 檢索流程中的分支過濾檢索時的核心原則是先限定分支范圍再做語義召回。如果沒有指定分支系統(tǒng)應(yīng)該使用默認分支或“全分支限權(quán)檢索”而不是直接不做過濾地全局搜索。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]這里的實現(xiàn)順序是先通過結(jié)構(gòu)化條件縮小候選集再做向量檢索。不要反過來否則每次查詢都會在全量向量庫中做相似度計算數(shù)據(jù)量上來后延遲會明顯增加。4.5 分支合并與歸檔策略記憶系統(tǒng)不能只有寫入和檢索還需要定時處理分支中的數(shù)據(jù)狀態(tài)。合并解決的是重復記憶問題歸檔解決的是數(shù)據(jù)膨脹問題。合并策略通常按以下規(guī)則處理同一用戶在同一個分支下短期重復出現(xiàn)相似內(nèi)容新記憶的重要性高于舊記憶舊記憶已經(jīng)連續(xù)多次未被檢索命中歸檔策略按時間觸發(fā)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 狀態(tài)的記憶改成 archived。實際生產(chǎn)環(huán)境中expired 狀態(tài)和 archived 狀態(tài)可以分開處理expired 表示不再參與檢索archived 表示仍然保留但降級為低頻訪問。不要在同一個狀態(tài)里混用否則統(tǒng)計時很難區(qū)分。5. 關(guān)鍵參數(shù)和設(shè)計取舍5.1 分支數(shù)量應(yīng)該怎么控制數(shù)據(jù)分支數(shù)量是一個典型的取舍問題。分支太少記憶混在一起檢索噪聲大分支太多路由規(guī)則復雜調(diào)用方需要記住一堆分支名運維維護成本也高。分支數(shù)量優(yōu)點缺點適用場景3 到 5 個路由簡單、檢索穩(wěn)定粒度較粗某些場景仍有噪聲個人助手、小型業(yè)務(wù)6 到 15 個語義隔離清晰需要維護路由規(guī)則和使用文檔中型團隊知識庫、多項目 Agent15 個以上高度隔離路由復雜分支管理和監(jiān)控成本高大型組織、強權(quán)限隔離場景推薦做法是先用少量分支跑通完整鏈路觀察檢索命中率。當出現(xiàn)“檢索結(jié)果混雜無關(guān)記憶”時再根據(jù)失敗樣本拆分新分支。5.2 記憶系統(tǒng)關(guān)鍵參數(shù)速查表參數(shù)常見值影響錯誤表現(xiàn)top_k5 到 10返回記憶條數(shù)過大易引入噪聲Agent 上下文被無關(guān)信息擠占score_threshold0.6 到 0.8低于閾值的結(jié)果不返回返回不相關(guān)內(nèi)容檢索質(zhì)量下降importance0 到 1排序權(quán)重重要記憶優(yōu)先低價值記憶長期占坑expire_after_days30 到 365控制記憶自然過期時間過期太短丟失有用信息太長數(shù)據(jù)膨脹merge_interval_seconds3600 到 86400控制去重合并頻率過高增加計算成本過低導致重復記憶殘留branch_list 默認值3 個核心分支未指定分支時的候選范圍誤查全局導致權(quán)限混雜或噪聲大這里要強調(diào)score_threshold 不是越高越好。設(shè)置太高時真正有用的記憶可能因為表述差異被過濾掉設(shè)置太低時大量低相關(guān)記憶進入上下文模型反而更糊涂。建議在測試集上統(tǒng)計相似度分布后再確定閾值。5.3 語義分支與向量集合的關(guān)系數(shù)據(jù)分支落到存儲層有兩種主流實現(xiàn)方式。第一種是一個分支對應(yīng)一個獨立向量集合。優(yōu)點是隔離徹底分支之間互不影響權(quán)限控制容易缺點是跨分支檢索時需要逐個集合查詢再合并結(jié)果實現(xiàn)相對復雜。第二種是統(tǒng)一向量集合通過 metadata 字段中的 branch_type 標簽過濾。優(yōu)點是實現(xiàn)簡單一次查詢可以同時支持單分支和多分支缺點是數(shù)據(jù)量增大后過濾條件對向量檢索的效率影響需要評估。兩種方式的代碼差異主要體現(xiàn)在存儲和檢索兩個環(huán)節(jié)。使用第二種方式時寫入向量時需要附帶 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 } }選擇哪種實現(xiàn)取決于數(shù)據(jù)規(guī)模和是否需要嚴格的物理隔離。個人項目和小型團隊建議用第二種復雜度低最容易落地。6. 運行驗證從寫入到分支查詢6.1 啟動最小服務(wù)這里使用 FastAPI 提供 HTTP 接口便于驗證完整流程。啟動命令如下uvicorn app:app --host 0.0.0.0 --port 8000啟動成功后終端會輸出 uvicorn 的運行地址。此時可以訪問http://127.0.0.1:8000/docs查看接口文檔。這一步確認服務(wù)本身沒有報錯。6.2 寫入不同分支的數(shù)據(jù)寫入接口請求示例POST /memories { user_id: user_001, source: user, memory_type: preference, content: 用戶喜歡簡潔的技術(shù)方案不喜歡長篇大論, importance: 0.9 }返回結(jié)果中應(yīng)包含 branch_type 字段值為user_preference。這就是路由函數(shù)生效的驗證。再寫入一條任務(wù)事件POST /memories { user_id: user_001, source: agent, memory_type: event, content: 用戶在今天 14:00 確認了訂單系統(tǒng)的數(shù)據(jù)庫選型, importance: 0.7 }返回結(jié)果中 branch_type 應(yīng)改為task_event。兩條數(shù)據(jù)屬于不同分支證明路由規(guī)則正常工作。6.3 驗證分支過濾查詢查詢接口請求示例GET /memories?user_iduser_001branch_typeuser_preferencequery喜歡簡潔方案正常結(jié)果應(yīng)該只返回 user_preference 分支中的記憶不返回 task_event 分支中的內(nèi)容。如果查詢時不傳 branch_type只傳 user_id則確認默認分支邏輯是否生效GET /memories?user_iduser_001query訂單系統(tǒng)這里要注意觀察返回結(jié)果是否同時包含 user_preference 和 task_event 分支的數(shù)據(jù)。如果只返回一個分支需要回頭檢查search_memory中的默認 branch_list 配置。6.4 驗證歸檔與遺忘手動執(zhí)行歸檔函數(shù)驗證python -c from memory_core import archive_expired_memories; archive_expired_memories()然后在數(shù)據(jù)庫里檢查狀態(tài)變化SELECT memory_id, branch_type, status, expire_at FROM memory_record WHERE user_id user_001;正常情況下未過期的記錄仍是 active已過期記錄變成 archived。如果所有記錄都變成 archived說明寫入時 expire_at 計算用的時間基準有問題檢查是否使用了錯誤的時區(qū)或者過期時間被設(shè)置成當前時間之前。7. 常見問題排查7.1 寫入后檢索不到可能原因有很多按順序排查寫入時 route_branch 返回的分支與查詢時傳入的 branch_type 不一致。向量數(shù)據(jù)沒有保存成功memory_embedding 表缺少記錄。查詢時的 user_id 與寫入時的 user_id 不一致。score_threshold 設(shè)置過高導致相似度低于閾值的結(jié)果被丟棄。數(shù)據(jù)狀態(tài)不是 active而是 archived、expired 或 trashed。推薦檢查方式先不看向量直接用 SQL 查詢 memory_record 表確認記錄存在且 status 為 active。然后再確認 query 中傳入的過濾條件是否包含該記錄的分支。7.2 新記憶覆蓋了舊記憶現(xiàn)象是用戶修改偏好后新記錄寫入舊記錄仍然存在但 Agent 仍然會讀到舊記錄。原因是寫入新記憶時沒有處理舊記憶的沖突舊記錄仍處于 active 狀態(tài)。處理方式在寫入相同分支、相同 type 的新記憶時先將舊記憶的狀態(tài)改為 superseded。增加 version 字段記錄同主題記憶的先后版本。需要支持歷史回溯時不物理刪除舊記錄只改變狀態(tài)。推薦做法是給 memory_record 增加一個 topic_key 字段用來標識“同一主題”。寫入時先查詢相同 topic_key 的記錄如果存在新版本就把舊版本置為更新狀態(tài)。7.3 表數(shù)據(jù)膨脹后查詢變慢當 memory_record 表數(shù)據(jù)量持續(xù)增長即使加了索引查詢也可能變慢。檢查方向是否按 user_id 和 branch_type 建立了聯(lián)合索引。是否定時執(zhí)行了歸檔任務(wù)active 數(shù)據(jù)是否被控制在合理范圍。向量檢索是否在 SQL 過濾之后執(zhí)行而不是先全量向量搜索再過濾。如果 SQLite 單表數(shù)據(jù)量已經(jīng)超過百萬行建議把結(jié)構(gòu)化數(shù)據(jù)遷移到 PostgreSQL把向量數(shù)據(jù)遷移到專業(yè)向量庫避免在單個文件數(shù)據(jù)庫上硬扛大數(shù)據(jù)量。7.4 時間分支錯亂現(xiàn)象是按天歸檔時部分數(shù)據(jù)被歸入錯誤日期或過期時間判斷錯誤。常見原因是時區(qū)不統(tǒng)一。有的服務(wù)使用本地時間寫入另一個服務(wù)使用 UTC 讀取導致expire_at和當前時間比較時出現(xiàn)偏差。排查時先確認所有時間字段是否統(tǒng)一使用 UTC ISO 格式再確認寫入時datetime.now(timezone.utc)是否被誤寫成datetime.now()。問題現(xiàn)象常見原因檢查方式處理建議寫入后檢索不到分支條件不一致或向量缺失先查 SQL 再查向量表統(tǒng)一分支命名增加寫入日志新記憶覆蓋舊記憶失敗未處理主題沖突檢查 topic_key 查詢邏輯寫入前先查同主題 active 記錄查詢變慢沒有及時歸檔或索引缺失查看 SQL 執(zhí)行計劃增加組合索引定期運行歸檔任務(wù)時間分支錯亂時區(qū)混用檢查數(shù)據(jù)庫時間值全部改為 UTC 存儲8. 最佳實踐與擴展方向8.1 學習環(huán)境和生產(chǎn)環(huán)境的差異本地 Demo 可以接受 SQLite 和模擬 embedding但生產(chǎn)環(huán)境不能這樣簡單處理。層面本地驗證環(huán)境生產(chǎn)環(huán)境存儲SQLitePostgreSQL 或其他關(guān)系數(shù)據(jù)庫向量檢索Python 列表遍歷專用向量庫支持索引和分片嵌入服務(wù)模擬向量內(nèi)部嵌入服務(wù)需監(jiān)控穩(wěn)定性和限流定時任務(wù)手動執(zhí)行腳本獨立調(diào)度服務(wù)記錄執(zhí)行日志鑒權(quán)無接口鑒權(quán)、用戶數(shù)據(jù)隔離、操作審計配置寫死在代碼配置中心或環(huán)境變量外置在生產(chǎn)環(huán)境上線記憶系統(tǒng)前至少要補齊日志、監(jiān)控、權(quán)限、回滾和數(shù)據(jù)備份這幾項。記憶數(shù)據(jù)是 Agent 行為的上下文依據(jù)丟失或?qū)戝e都會直接影響業(yè)務(wù)結(jié)果。8.2 生產(chǎn)環(huán)境必須補齊的保障記憶寫入接口需要增加調(diào)用方標識、操作審計和實施限流。不能讓任意客戶端無限寫入否則惡意調(diào)用可能撐爆存儲。批量任務(wù)要重試和告警。歸檔、合并、遺忘任務(wù)如果失敗不能靜默結(jié)束應(yīng)該記錄失敗原因并觸發(fā)告警否則數(shù)據(jù)會持續(xù)膨脹。數(shù)據(jù)備份策略要區(qū)分全量備份和增量備份。向量數(shù)據(jù)也要納入備份范圍不能只備份關(guān)系型數(shù)據(jù)庫。更重要的是要建立記憶回滾能力。當一次批量記憶更新導致 Agent 行為異常時能按分支、按用戶、按時間段回滾到歷史狀態(tài)。這個能力在設(shè)計階段就要預留不能在事故發(fā)生后靠手工改庫。8.3 下一步可以怎么擴展數(shù)據(jù)分支是記憶系統(tǒng)的基礎(chǔ)能力下一步可以考慮以下方向。第一接入知識圖譜。將 fact 類型記憶抽取為實體和關(guān)系用圖結(jié)構(gòu)支撐多跳推理。數(shù)據(jù)分支可以按實體所屬領(lǐng)域劃分例如產(chǎn)品域、技術(shù)域、用戶域。第二實現(xiàn)分級遺忘機制。根據(jù)重要程度、訪問頻率、最后一次使用時間綜合計算記憶熱度熱度低的記憶先進入候選歸檔列表由人工或規(guī)則確認后刪除。第三支持多 Agent 共享記憶。在 user_id 之上增加 agent_id、team_id 維度讓一個團隊的多個 Agent 共享知識但用戶私有記憶仍然保持隔離。這一步需要更細的權(quán)限模型。第四建立記憶反饋閉環(huán)。當用戶對 Agent 的回答給出負面反饋時回查命中的記憶數(shù)據(jù)分析是否記憶本身不準確。這樣可以把數(shù)據(jù)分支與模型評估系統(tǒng)連接起來。對新手來說最有價值的練習不是直接接入完整框架而是先手寫一遍最小的記憶寫入、分支路由、查詢和歸檔流程把數(shù)據(jù)模型和狀態(tài)流轉(zhuǎn)理解清楚再去學習向量庫和遺忘機制。數(shù)據(jù)分支看起來只是簡單的字段拆分但它決定了記憶系統(tǒng)在真實業(yè)務(wù)中能不能穩(wěn)定工作。 AI Agent記憶系統(tǒng)：數(shù)據(jù)分支設(shè)計原理與最小實現(xiàn)
AI Agent 在執(zhí)行復雜任務(wù)時真正限制它表現(xiàn)的往往不是單個模型的能力而是它能不能記住上下文、能不能在關(guān)鍵節(jié)點調(diào)出正確信息。oGMemory 記憶系統(tǒng)要解決的正是 Agent 跨會話、跨任務(wù)的記憶組織問題而數(shù)據(jù)分支又是決定記憶系統(tǒng)能不能“用起來”的關(guān)鍵設(shè)計。這一期分集把視線聚焦在記憶系統(tǒng)與數(shù)據(jù)分支上先講清楚數(shù)據(jù)分支為什么存在再給出一套可以落地的最小實現(xiàn)。很多項目在最初接入記憶能力時通常的做法是“對話結(jié)束后把文本塞進向量庫下次檢索再查出來”。這個思路在 Demo 階段可行但一旦進入真實業(yè)務(wù)就會出現(xiàn)一個典型問題所有記憶混在一起用戶偏好、任務(wù)記錄、知識沉淀、短期上下文全部堆在同一個集合里。寫入越久檢索噪聲越大Agent 越來越難分辨哪條信息是當前任務(wù)該用的。數(shù)據(jù)分支的作用就是按業(yè)務(wù)維度把這些記憶拆開讓寫入有明確歸屬讓檢索有明確范圍。1. 記憶系統(tǒng)到底在解決什么問題1.1 沒有記憶的 Agent 為什么不夠用沒有記憶系統(tǒng)的 Agent本質(zhì)上是一個“每次對話都從零開始”的狀態(tài)機。模型本身擁有訓練階段沉淀的靜態(tài)知識但它不知道用戶上一次問過什么、當前任務(wù)做到哪一步、用戶偏好哪種回答風格。對于一次性的問答場景這沒有問題但對于需要連續(xù)執(zhí)行多步任務(wù)、跨天維護用戶關(guān)系、持續(xù)沉淀團隊知識的場景缺記憶就意味著每次都要用戶重新交代背景。oGMemory 這類記憶系統(tǒng)的核心目標是給 Agent 增加一個可持續(xù)讀寫的外部狀態(tài)層。模型本身不可變但記憶數(shù)據(jù)可以隨時間、任務(wù)、用戶不斷更新。Agent 在執(zhí)行任務(wù)前先讀取記憶執(zhí)行過程中寫入新記憶任務(wù)結(jié)束后再整理歸檔這樣它就能形成“越用越懂當前場景”的能力。需要注意的是記憶不是簡單地等于聊天記錄。聊天記錄是原始素材記憶系統(tǒng)要把素材加工成結(jié)構(gòu)化的、可檢索的、能按需過濾的數(shù)據(jù)。這個加工過程包括抽取主題、過濾噪聲、判定重要性、設(shè)置生命周期、分配數(shù)據(jù)分支。1.2 “能存下來”和“會用”之間的差距只把文本存進數(shù)據(jù)庫并不是記憶系統(tǒng)。真正的記憶系統(tǒng)要處理四個連續(xù)問題第一什么值得記。不是每一句對話都值得寫入長期記憶。寒暄、臨時計算過程、重復確認信息都應(yīng)該被過濾掉。真正值得記的是用戶偏好、事實結(jié)論、任務(wù)狀態(tài)、關(guān)鍵約束。第二記到哪里去。這個問題就是數(shù)據(jù)分支要解決的。不同性質(zhì)的記憶需要不同的存儲策略比如短期會話上下文可以放在快速緩存中用戶長期偏好需要進入穩(wěn)定的關(guān)系型存儲或向量庫而任務(wù)執(zhí)行記錄可能需要按項目維度隔離。第三如何被找到。寫入時可讀不等于檢索時可命中。檢索環(huán)節(jié)要處理相似度計算、過濾條件、排序策略、時效衰減。如果沒有分支約束檢索時會把不相關(guān)的記憶也召回導致上下文被污染。第四什么時候被更新或遺忘。記憶不能只增不減。當用戶明確改變偏好、任務(wù)狀態(tài)發(fā)生流轉(zhuǎn)、數(shù)據(jù)超過保留期限時系統(tǒng)要具備更新、合并、降級、刪除能力。這四個問題里存儲和檢索最容易理解但也最容易被低估。很多時候項目跑不起來不是因為模型不行而是因為記憶數(shù)據(jù)沒有組織好導致 Agent 讀到了錯誤的歷史信息。1.3 oGMemory 在其中的定位從命名和常見記憶系統(tǒng)設(shè)計來看oGMemory 可以理解為一條獨立的記憶數(shù)據(jù)鏈路重點解決 Agent 記憶的組織、存儲、檢索和生命周期管理。和直接在業(yè)務(wù)代碼里零散調(diào)用向量庫不同oGMemory 這類設(shè)計會把記憶系統(tǒng)抽象成獨立服務(wù)對外提供寫入、查詢、更新、遺忘等接口讓上層 Agent 業(yè)務(wù)只關(guān)心記憶內(nèi)容不關(guān)心底層存儲細節(jié)。本文作為分集解讀的第一篇聚焦在數(shù)據(jù)分支這一層不展開討論底層模型、檢索算法、遺忘機制的全部細節(jié)。讀完這一篇你能完成三件事理解記憶系統(tǒng)的整體分層掌握數(shù)據(jù)分支的劃分思路用一套最小代碼跑通“寫入分支、按分支查詢、分支歸檔”的完整流程。2. 搭建最小記憶系統(tǒng)需要先確認的核心組件2.1 記憶系統(tǒng)的五層結(jié)構(gòu)一個可工作的記憶系統(tǒng)通常包含五層感知層、編碼層、存儲層、檢索層、遺忘與合并層。數(shù)據(jù)分支主要落在存儲層和檢索層之間但每一層都會影響分支設(shè)計。感知層負責識別什么信息值得寫入記憶。它可能是對話中斷言抽取、任務(wù)狀態(tài)識別、用戶行為事件接收。感知層的輸出是“一條候選記憶”這部分通常要依賴 LLM 或規(guī)則引擎完成。編碼層負責把文本轉(zhuǎn)換成可以檢索的形態(tài)。最常見的方式是文本向量化同時保留原文和結(jié)構(gòu)化字段。編碼層還要生成主題標簽、時間戳、重要程度、來源標識等元數(shù)據(jù)。存儲層負責持久化。向量數(shù)據(jù)進入向量庫結(jié)構(gòu)化字段進入關(guān)系型數(shù)據(jù)庫原始內(nèi)容可能需要對象存儲。數(shù)據(jù)分支在這一層體現(xiàn)為不同的集合、表或分片。檢索層負責把用戶當前的問題轉(zhuǎn)換成查詢條件從正確的分支中召回記憶。檢索不是只做語義相似度還需要結(jié)合時間范圍、用戶身份、分支類型、權(quán)限范圍等條件。遺忘與合并層負責控制記憶生命周期。它解決記憶過期、沖突覆蓋、重復合并、降級歸檔等問題。沒有這一層記憶數(shù)據(jù)會無限膨脹檢索質(zhì)量會持續(xù)下降。2.2 組件選型和環(huán)境準備記憶系統(tǒng)的選型不存在一套通用標準答案但通常會涉及以下幾類組件組件職責常見選型參考LLM文本理解、信息抽取、摘要生成業(yè)務(wù)現(xiàn)有的大模型服務(wù)Embedding 服務(wù)將文本轉(zhuǎn)換為向量本地嵌入模型接口或平臺嵌入服務(wù)向量庫存放向量數(shù)據(jù)支持語義檢索輕量可用 Chroma、生產(chǎn)環(huán)境可用 Milvus 等關(guān)系型數(shù)據(jù)庫存放記憶記錄、分支元數(shù)據(jù)、狀態(tài)SQLite 適合本地驗證生產(chǎn)可用 PostgreSQL 等緩存存儲高頻訪問的短期記憶Redis 等定時任務(wù)執(zhí)行歸檔、合并、遺忘清理APScheduler、Celery Beat 或云廠商定時任務(wù)這里要特別說明本文的示例代碼為了保持最小可運行使用 SQLite 存儲結(jié)構(gòu)化數(shù)據(jù)向量部分用一個本地 Python 列表和一個模擬的 embedding 函數(shù)代替。這樣做的目的是先把數(shù)據(jù)分支邏輯講清楚。真實項目接入時把 embedding 函數(shù)替換成內(nèi)部嵌入服務(wù)把向量列表遷移到向量庫即可。2.3 環(huán)境準備清單在開始寫代碼之前先確認本機環(huán)境滿足以下條件Python 3.10 或更高版本已安裝 FastAPI 和 uvicorn已安裝 SQLite3 驅(qū)動Python 自帶的 sqlite3 就夠用已安裝 pandas 或直接使用 Python 標準庫處理數(shù)據(jù)一個可用的 embedding 函數(shù)開發(fā)階段可以用隨機向量模擬但驗證階段建議接入真實嵌入服務(wù)如果原始項目沒有明確版本落地前務(wù)必先確認依賴版本和 Python 版本兼容性。這里給出一個 requirements 示例fastapi0.110.0 uvicorn0.29.0 pydantic2.6.0 numpy1.26.4 apscheduler3.10.4注意版本號只是參考。實際項目如果使用已有依賴鎖文件以倉庫中的版本為準不要直接復制最新版本號到生產(chǎn)環(huán)境。3. 數(shù)據(jù)分支是什么為什么記憶系統(tǒng)離不開它3.1 數(shù)據(jù)分支的定義數(shù)據(jù)分支是指按照某種業(yè)務(wù)維度把記憶數(shù)據(jù)劃分成不同的邏輯通道。每個通道擁有獨立的寫入規(guī)則、存儲位置、檢索范圍和生命周期策略?？梢园褦?shù)據(jù)分支理解成“記憶的分類目錄”。沒有分類目錄時所有記憶是一條無序的長河有了分類目錄后寫入時先判斷這條記憶屬于哪個分類檢索時只從對應(yīng)分類中查找效率和準確率都會明顯提升。在 oGMemory 這類 Agent 記憶系統(tǒng)中分支不是一個可有可無的優(yōu)化項而是決定記憶能否被正確使用的關(guān)鍵設(shè)計。原因是 Agent 的記憶數(shù)據(jù)天然帶有強烈的“上下文依賴”特征。用戶說“我喜歡簡潔的回答”這是一條偏好記憶用戶說“項目 A 的數(shù)據(jù)庫連接串已經(jīng)改好了”這是一條任務(wù)事件記憶用戶說“本周五要上線”這是一條計劃記憶。這三者的使用場景完全不同混在一起存儲檢索時很難一次性命中正確信息。3.2 常見分支維度實際項目中記憶系統(tǒng)通常不會只使用一個分支維度而是幾個維度組合使用。常用維度包括分支維度說明示例按時間區(qū)分短期上下文、中期工作記憶、長期沉淀short_term、mid_term、long_term按來源區(qū)分用戶提供、Agent 推導、系統(tǒng)事件、外部知識user、agent、system、knowledge按類型區(qū)分事實、偏好、事件、技能、約束fact、preference、event、skill、constraint按業(yè)務(wù)域區(qū)分不同項目、不同知識庫、不同團隊project_a、project_b、wiki_base按生命周期區(qū)分活躍、候選歸檔、已過期、廢棄active、archived、expired、trashed分支維度的選擇不是越多越好。每增加一個維度寫入時的路由判斷就更復雜檢索時的過濾條件也更多運維成本會指數(shù)上升。多數(shù)中小型記憶系統(tǒng)從“來源 類型 時間”三個維度起步就足夠之后根據(jù)真實檢索效果再細化。3.3 數(shù)據(jù)分支與分庫分表的區(qū)別數(shù)據(jù)分支容易被誤解成“分庫分表”但它們解決的問題不同。分庫分表解決的是存儲容量和寫入性能問題。當單表數(shù)據(jù)量達到千萬級索引失效、寫入變慢這時需要把數(shù)據(jù)按照哈?；蚍秶稚⒌蕉鄠€物理存儲中。分庫分表是物理層的伸縮方案對業(yè)務(wù)透明上層 SQL 大多數(shù)情況下不應(yīng)該感知到分片邏輯。數(shù)據(jù)分支解決的是語義隔離和檢索范圍問題。它強調(diào)“哪些記憶屬于哪個業(yè)務(wù)范圍”是邏輯層的分類設(shè)計。即使數(shù)據(jù)量很小也需要做數(shù)據(jù)分支否則 Agent 會在兩三百條記憶里被噪聲干擾。兩者可以疊加使用。先按業(yè)務(wù)維度做數(shù)據(jù)分支再在數(shù)據(jù)量增長后按哈希規(guī)則分片是常見的生產(chǎn)架構(gòu)。實現(xiàn)時要注意分支規(guī)則代碼和分片路由代碼不要混在同一個函數(shù)里否則排查問題時很難定位。4. 用 Python 實現(xiàn)一個帶數(shù)據(jù)分支的最小記憶服務(wù)4.1 項目結(jié)構(gòu)下面這個項目結(jié)構(gòu)適用于本地驗證和最小 Demoogmemory_demo/ ├── app.py # FastAPI 入口 ├── memory_core.py # 記憶寫入、檢索、路由核心邏輯 ├── storage.py # SQLite 初始化與數(shù)據(jù)訪問 ├── branch_rules.py # 分支路由規(guī)則 ├── models.py # Pydantic 數(shù)據(jù)模型 ├── requirements.txt # 依賴清單 └── data/ └── memory.db # SQLite 數(shù)據(jù)庫文件這個結(jié)構(gòu)把路由規(guī)則、存儲訪問、核心邏輯分開目的是讓每個模塊職責單一。后續(xù)如果要替換向量庫只需改動 storage.py 和 memory_core.py 中的向量相關(guān)部分分支規(guī)則不需要動。4.2 數(shù)據(jù)模型設(shè)計記憶系統(tǒng)的核心表是 memory_record 表。它既保存原始文本也保存結(jié)構(gòu)化元數(shù)據(jù)同時通過 memory_embedding 表關(guān)聯(lián)向量數(shù)據(jù)。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數(shù)據(jù)分支例如 user_preference、task_eventmemory_type記憶類型例如 fact、preference、eventsource來源例如 user、agent、systemcontent記憶原始內(nèi)容importance重要程度0 到 1影響后續(xù)遺忘優(yōu)先級status狀態(tài)active、archived、expired、trashedmerged_from如果該記錄由多條記憶合并而來記錄來源 IDcreated_at / updated_at時間字段統(tǒng)一使用 UTC ISO 格式expire_at過期時間過期后進入遺忘候選這里要注意時間字段統(tǒng)一使用 UTC不要使用本地時間。否則跨時區(qū)部署時分支按時間歸檔會出現(xiàn)錯位問題。4.3 寫入流程中的分支路由記憶寫入時系統(tǒng)先判斷這條記憶應(yīng)該進入哪個分支再執(zhí)行存儲。分支路由不只是一個字段賦值它可能影響后續(xù)的索引、召回策略和過期策略。下面是一個分支路由規(guī)則示例# 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這個示例展示了最基礎(chǔ)的關(guān)鍵詞維度路由。更復雜的項目可以使用 LLM 抽取結(jié)果來決定分支。比如先讓模型輸出{ source: user, memory_type: preference, importance: 0.9 }再交給路由函數(shù)判斷。寫入核心邏輯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值得關(guān)注的是route_branch這個函數(shù)。它是數(shù)據(jù)分支的唯一決策點所有記憶寫入都必須經(jīng)過它。這樣設(shè)計的好處是當業(yè)務(wù)需要調(diào)整分支規(guī)則時只改一個文件就能生效不需要在多個業(yè)務(wù)調(diào)用點里尋找散落的分支判斷。4.4 檢索流程中的分支過濾檢索時的核心原則是先限定分支范圍再做語義召回。如果沒有指定分支系統(tǒng)應(yīng)該使用默認分支或“全分支限權(quán)檢索”而不是直接不做過濾地全局搜索。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]這里的實現(xiàn)順序是先通過結(jié)構(gòu)化條件縮小候選集再做向量檢索。不要反過來否則每次查詢都會在全量向量庫中做相似度計算數(shù)據(jù)量上來后延遲會明顯增加。4.5 分支合并與歸檔策略記憶系統(tǒng)不能只有寫入和檢索還需要定時處理分支中的數(shù)據(jù)狀態(tài)。合并解決的是重復記憶問題歸檔解決的是數(shù)據(jù)膨脹問題。合并策略通常按以下規(guī)則處理同一用戶在同一個分支下短期重復出現(xiàn)相似內(nèi)容新記憶的重要性高于舊記憶舊記憶已經(jīng)連續(xù)多次未被檢索命中歸檔策略按時間觸發(fā)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 狀態(tài)的記憶改成 archived。實際生產(chǎn)環(huán)境中expired 狀態(tài)和 archived 狀態(tài)可以分開處理expired 表示不再參與檢索archived 表示仍然保留但降級為低頻訪問。不要在同一個狀態(tài)里混用否則統(tǒng)計時很難區(qū)分。5. 關(guān)鍵參數(shù)和設(shè)計取舍5.1 分支數(shù)量應(yīng)該怎么控制數(shù)據(jù)分支數(shù)量是一個典型的取舍問題。分支太少記憶混在一起檢索噪聲大分支太多路由規(guī)則復雜調(diào)用方需要記住一堆分支名運維維護成本也高。分支數(shù)量優(yōu)點缺點適用場景3 到 5 個路由簡單、檢索穩(wěn)定粒度較粗某些場景仍有噪聲個人助手、小型業(yè)務(wù)6 到 15 個語義隔離清晰需要維護路由規(guī)則和使用文檔中型團隊知識庫、多項目 Agent15 個以上高度隔離路由復雜分支管理和監(jiān)控成本高大型組織、強權(quán)限隔離場景推薦做法是先用少量分支跑通完整鏈路觀察檢索命中率。當出現(xiàn)“檢索結(jié)果混雜無關(guān)記憶”時再根據(jù)失敗樣本拆分新分支。5.2 記憶系統(tǒng)關(guān)鍵參數(shù)速查表參數(shù)常見值影響錯誤表現(xiàn)top_k5 到 10返回記憶條數(shù)過大易引入噪聲Agent 上下文被無關(guān)信息擠占score_threshold0.6 到 0.8低于閾值的結(jié)果不返回返回不相關(guān)內(nèi)容檢索質(zhì)量下降importance0 到 1排序權(quán)重重要記憶優(yōu)先低價值記憶長期占坑expire_after_days30 到 365控制記憶自然過期時間過期太短丟失有用信息太長數(shù)據(jù)膨脹merge_interval_seconds3600 到 86400控制去重合并頻率過高增加計算成本過低導致重復記憶殘留branch_list 默認值3 個核心分支未指定分支時的候選范圍誤查全局導致權(quán)限混雜或噪聲大這里要強調(diào)score_threshold 不是越高越好。設(shè)置太高時真正有用的記憶可能因為表述差異被過濾掉設(shè)置太低時大量低相關(guān)記憶進入上下文模型反而更糊涂。建議在測試集上統(tǒng)計相似度分布后再確定閾值。5.3 語義分支與向量集合的關(guān)系數(shù)據(jù)分支落到存儲層有兩種主流實現(xiàn)方式。第一種是一個分支對應(yīng)一個獨立向量集合。優(yōu)點是隔離徹底分支之間互不影響權(quán)限控制容易缺點是跨分支檢索時需要逐個集合查詢再合并結(jié)果實現(xiàn)相對復雜。第二種是統(tǒng)一向量集合通過 metadata 字段中的 branch_type 標簽過濾。優(yōu)點是實現(xiàn)簡單一次查詢可以同時支持單分支和多分支缺點是數(shù)據(jù)量增大后過濾條件對向量檢索的效率影響需要評估。兩種方式的代碼差異主要體現(xiàn)在存儲和檢索兩個環(huán)節(jié)。使用第二種方式時寫入向量時需要附帶 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 } }選擇哪種實現(xiàn)取決于數(shù)據(jù)規(guī)模和是否需要嚴格的物理隔離。個人項目和小型團隊建議用第二種復雜度低最容易落地。6. 運行驗證從寫入到分支查詢6.1 啟動最小服務(wù)這里使用 FastAPI 提供 HTTP 接口便于驗證完整流程。啟動命令如下uvicorn app:app --host 0.0.0.0 --port 8000啟動成功后終端會輸出 uvicorn 的運行地址。此時可以訪問http://127.0.0.1:8000/docs查看接口文檔。這一步確認服務(wù)本身沒有報錯。6.2 寫入不同分支的數(shù)據(jù)寫入接口請求示例POST /memories { user_id: user_001, source: user, memory_type: preference, content: 用戶喜歡簡潔的技術(shù)方案不喜歡長篇大論, importance: 0.9 }返回結(jié)果中應(yīng)包含 branch_type 字段值為user_preference。這就是路由函數(shù)生效的驗證。再寫入一條任務(wù)事件POST /memories { user_id: user_001, source: agent, memory_type: event, content: 用戶在今天 14:00 確認了訂單系統(tǒng)的數(shù)據(jù)庫選型, importance: 0.7 }返回結(jié)果中 branch_type 應(yīng)改為task_event。兩條數(shù)據(jù)屬于不同分支證明路由規(guī)則正常工作。6.3 驗證分支過濾查詢查詢接口請求示例GET /memories?user_iduser_001branch_typeuser_preferencequery喜歡簡潔方案正常結(jié)果應(yīng)該只返回 user_preference 分支中的記憶不返回 task_event 分支中的內(nèi)容。如果查詢時不傳 branch_type只傳 user_id則確認默認分支邏輯是否生效GET /memories?user_iduser_001query訂單系統(tǒng)這里要注意觀察返回結(jié)果是否同時包含 user_preference 和 task_event 分支的數(shù)據(jù)。如果只返回一個分支需要回頭檢查search_memory中的默認 branch_list 配置。6.4 驗證歸檔與遺忘手動執(zhí)行歸檔函數(shù)驗證python -c from memory_core import archive_expired_memories; archive_expired_memories()然后在數(shù)據(jù)庫里檢查狀態(tài)變化SELECT memory_id, branch_type, status, expire_at FROM memory_record WHERE user_id user_001;正常情況下未過期的記錄仍是 active已過期記錄變成 archived。如果所有記錄都變成 archived說明寫入時 expire_at 計算用的時間基準有問題檢查是否使用了錯誤的時區(qū)或者過期時間被設(shè)置成當前時間之前。7. 常見問題排查7.1 寫入后檢索不到可能原因有很多按順序排查寫入時 route_branch 返回的分支與查詢時傳入的 branch_type 不一致。向量數(shù)據(jù)沒有保存成功memory_embedding 表缺少記錄。查詢時的 user_id 與寫入時的 user_id 不一致。score_threshold 設(shè)置過高導致相似度低于閾值的結(jié)果被丟棄。數(shù)據(jù)狀態(tài)不是 active而是 archived、expired 或 trashed。推薦檢查方式先不看向量直接用 SQL 查詢 memory_record 表確認記錄存在且 status 為 active。然后再確認 query 中傳入的過濾條件是否包含該記錄的分支。7.2 新記憶覆蓋了舊記憶現(xiàn)象是用戶修改偏好后新記錄寫入舊記錄仍然存在但 Agent 仍然會讀到舊記錄。原因是寫入新記憶時沒有處理舊記憶的沖突舊記錄仍處于 active 狀態(tài)。處理方式在寫入相同分支、相同 type 的新記憶時先將舊記憶的狀態(tài)改為 superseded。增加 version 字段記錄同主題記憶的先后版本。需要支持歷史回溯時不物理刪除舊記錄只改變狀態(tài)。推薦做法是給 memory_record 增加一個 topic_key 字段用來標識“同一主題”。寫入時先查詢相同 topic_key 的記錄如果存在新版本就把舊版本置為更新狀態(tài)。7.3 表數(shù)據(jù)膨脹后查詢變慢當 memory_record 表數(shù)據(jù)量持續(xù)增長即使加了索引查詢也可能變慢。檢查方向是否按 user_id 和 branch_type 建立了聯(lián)合索引。是否定時執(zhí)行了歸檔任務(wù)active 數(shù)據(jù)是否被控制在合理范圍。向量檢索是否在 SQL 過濾之后執(zhí)行而不是先全量向量搜索再過濾。如果 SQLite 單表數(shù)據(jù)量已經(jīng)超過百萬行建議把結(jié)構(gòu)化數(shù)據(jù)遷移到 PostgreSQL把向量數(shù)據(jù)遷移到專業(yè)向量庫避免在單個文件數(shù)據(jù)庫上硬扛大數(shù)據(jù)量。7.4 時間分支錯亂現(xiàn)象是按天歸檔時部分數(shù)據(jù)被歸入錯誤日期或過期時間判斷錯誤。常見原因是時區(qū)不統(tǒng)一。有的服務(wù)使用本地時間寫入另一個服務(wù)使用 UTC 讀取導致expire_at和當前時間比較時出現(xiàn)偏差。排查時先確認所有時間字段是否統(tǒng)一使用 UTC ISO 格式再確認寫入時datetime.now(timezone.utc)是否被誤寫成datetime.now()。問題現(xiàn)象常見原因檢查方式處理建議寫入后檢索不到分支條件不一致或向量缺失先查 SQL 再查向量表統(tǒng)一分支命名增加寫入日志新記憶覆蓋舊記憶失敗未處理主題沖突檢查 topic_key 查詢邏輯寫入前先查同主題 active 記錄查詢變慢沒有及時歸檔或索引缺失查看 SQL 執(zhí)行計劃增加組合索引定期運行歸檔任務(wù)時間分支錯亂時區(qū)混用檢查數(shù)據(jù)庫時間值全部改為 UTC 存儲8. 最佳實踐與擴展方向8.1 學習環(huán)境和生產(chǎn)環(huán)境的差異本地 Demo 可以接受 SQLite 和模擬 embedding但生產(chǎn)環(huán)境不能這樣簡單處理。層面本地驗證環(huán)境生產(chǎn)環(huán)境存儲SQLitePostgreSQL 或其他關(guān)系數(shù)據(jù)庫向量檢索Python 列表遍歷專用向量庫支持索引和分片嵌入服務(wù)模擬向量內(nèi)部嵌入服務(wù)需監(jiān)控穩(wěn)定性和限流定時任務(wù)手動執(zhí)行腳本獨立調(diào)度服務(wù)記錄執(zhí)行日志鑒權(quán)無接口鑒權(quán)、用戶數(shù)據(jù)隔離、操作審計配置寫死在代碼配置中心或環(huán)境變量外置在生產(chǎn)環(huán)境上線記憶系統(tǒng)前至少要補齊日志、監(jiān)控、權(quán)限、回滾和數(shù)據(jù)備份這幾項。記憶數(shù)據(jù)是 Agent 行為的上下文依據(jù)丟失或?qū)戝e都會直接影響業(yè)務(wù)結(jié)果。8.2 生產(chǎn)環(huán)境必須補齊的保障記憶寫入接口需要增加調(diào)用方標識、操作審計和實施限流。不能讓任意客戶端無限寫入否則惡意調(diào)用可能撐爆存儲。批量任務(wù)要重試和告警。歸檔、合并、遺忘任務(wù)如果失敗不能靜默結(jié)束應(yīng)該記錄失敗原因并觸發(fā)告警否則數(shù)據(jù)會持續(xù)膨脹。數(shù)據(jù)備份策略要區(qū)分全量備份和增量備份。向量數(shù)據(jù)也要納入備份范圍不能只備份關(guān)系型數(shù)據(jù)庫。更重要的是要建立記憶回滾能力。當一次批量記憶更新導致 Agent 行為異常時能按分支、按用戶、按時間段回滾到歷史狀態(tài)。這個能力在設(shè)計階段就要預留不能在事故發(fā)生后靠手工改庫。8.3 下一步可以怎么擴展數(shù)據(jù)分支是記憶系統(tǒng)的基礎(chǔ)能力下一步可以考慮以下方向。第一接入知識圖譜。將 fact 類型記憶抽取為實體和關(guān)系用圖結(jié)構(gòu)支撐多跳推理。數(shù)據(jù)分支可以按實體所屬領(lǐng)域劃分例如產(chǎn)品域、技術(shù)域、用戶域。第二實現(xiàn)分級遺忘機制。根據(jù)重要程度、訪問頻率、最后一次使用時間綜合計算記憶熱度熱度低的記憶先進入候選歸檔列表由人工或規(guī)則確認后刪除。第三支持多 Agent 共享記憶。在 user_id 之上增加 agent_id、team_id 維度讓一個團隊的多個 Agent 共享知識但用戶私有記憶仍然保持隔離。這一步需要更細的權(quán)限模型。第四建立記憶反饋閉環(huán)。當用戶對 Agent 的回答給出負面反饋時回查命中的記憶數(shù)據(jù)分析是否記憶本身不準確。這樣可以把數(shù)據(jù)分支與模型評估系統(tǒng)連接起來。對新手來說最有價值的練習不是直接接入完整框架而是先手寫一遍最小的記憶寫入、分支路由、查詢和歸檔流程把數(shù)據(jù)模型和狀態(tài)流轉(zhuǎn)理解清楚再去學習向量庫和遺忘機制。數(shù)據(jù)分支看起來只是簡單的字段拆分但它決定了記憶系統(tǒng)在真實業(yè)務(wù)中能不能穩(wěn)定工作。