這一篇把 藍圖 轉成可以直接照抄的定義。先給語言中立的介面契約（可以照著在任何語言實作），再給一份 TypeScript 的具體版本與對應的 SQL schema。設計目標只有一個：讓底層框架成為 adapter，換掉它不需要改上層程式碼。
這是設計草案，不是產品程式碼。 它刻意省略了錯誤處理、重試、批次最佳化與觀測埋點，那些應該按專案的既有慣例補上。凡是我的設計主張而非取自某個框架的部分，都會標明。
一、核心資料型別（語言中立）
五個型別。這是最小完整集合——少任何一個都會導致某類記憶無處可放。
MemoryUnit（記憶單元）——所有可召回記憶的基本單位。
MemoryUnit:
id : 唯一識別碼
kind : "semantic" | "episodic" | "procedural"
scope : Scope // 主體歸屬，見下
content : 文字 // 注入 context 的內容
summary : 文字（可選） // 用於粗篩，通常較短
content_hash : 文字 // 內容雜湊，去重用
source : SourceRef // provenance，指回原始事件
time : TemporalMeta // 四個時間戳，見下
confidence : 0.0–1.0 // 來源可信度
labels : 文字列表 // 自由標記，用於過濾
embedding : 向量（可選，衍生物）
Scope（範圍）——決定誰能讀、誰能寫、刪除時刪哪些。
Scope:
subject_type : "user" | "agent" | "shared" | "session"
subject_id : 文字 // user_id / agent_id / session_id
project_id : 文字（可選） // 多專案隔離
取自 Mem0 的 scope keys 與 LangMem 的 namespace，並補上 subject_type 這個明確的型別欄位。加這個欄位的理由是它讓「刪除某使用者的全部資料」與「共享知識不被誤刪」變成一個 WHERE 條件，而不是一段需要小心維護的邏輯。
TemporalMeta（時間中介資料）——雙時間軸，取自 Graphiti。
TemporalMeta:
created_at : 時間戳 // 系統何時知道（必填）
expired_at : 時間戳（可選） // 系統何時判定不再成立
valid_at : 時間戳（可選） // 現實中何時開始為真
invalid_at : 時間戳（可選） // 現實中何時停止為真
last_recalled : 時間戳（可選） // 最後一次被召回，歸檔判斷用
recall_count : 整數 // 召回次數，重要性訊號
後兩個欄位不在 Graphiti 的設計裡，是我加的：召回統計是決定歸檔優先序最便宜的訊號，而且它不需要任何 LLM 判斷。
MemoryBlock（記憶分區）——常駐層的單位，取自 Letta。
MemoryBlock:
id : 唯一識別碼
label : 文字 // "rules" | "user_profile" | "persona" ...
value : 文字
token_budget : 整數 // 這個分區的 context 配額
writable : 布林 // false = 只有開發者能改
description : 文字（可選） // 告訴 agent 這裡該放什麼
Episode（情節）——不可變的原始事件。
Episode:
id : 唯一識別碼
scope : Scope
occurred_at : 時間戳
raw : 文字或結構化內容 // 原始內容，不做任何處理
channel : 文字 // "chat" | "tool_result" | "document" ...
derived_ids : 識別碼列表 // 從這個 episode 抽出的 MemoryUnit
Episode 只 append、永不修改。 這是整份 schema 裡最重要的約束——它讓 provenance、重跑抽取、以及「區分刪除事實與刪除歷史」三件事同時成立。
二、服務介面（語言中立）
三個介面，對應藍圖的三條路徑。上層程式碼只依賴這三個介面，框架實作藏在它們後面。
MemoryStore: // 真源的讀寫
put_episode(Episode) -> id
put_unit(MemoryUnit) -> id
update_unit(id, patch) -> void
invalidate_unit(id, at: 時間戳) -> void // 標記失效，不刪除
delete_units(scope, filter) -> 數量 // 物理刪除，僅供刪除請求使用
get_unit(id) -> MemoryUnit
list_units(scope, filter, 分頁) -> MemoryUnit 列表
MemoryIndex: // 衍生索引，可從真源重建
upsert(MemoryUnit) -> void
remove(id) -> void
search_semantic(query, scope, k) -> (id, 分數) 列表
search_lexical(query, scope, k) -> (id, 分數) 列表
rebuild(scope) -> void // 換 embedding 模型時用
MemoryPolicy: // 所有「該不該」的決策集中在這裡
should_write(candidate, context) -> 布林 ＋ 理由
reconcile(candidate, neighbors) -> Decision
should_archive(unit, now) -> 布林
budget_for(block_label) -> 整數
把 MemoryPolicy 抽成獨立介面是這份草案最重要的一個設計主張。 六個框架的差異幾乎全部落在這個介面的實作裡：Letta 把它交給 agent 推理、Mem0 交給一條 LLM 管線、LangMem 交給開發者。把它獨立出來，你就可以在不改儲存與檢索的情況下換掉整套記憶治理策略——包括從「LLM 判斷」換成「規則判斷」來降低成本，或反過來。
Decision:
op : "ADD" | "UPDATE" | "DELETE" | "NOOP"
target_id : 識別碼（UPDATE/DELETE 時必填）
reason : 文字 // 必填，不可省略
confidence : 0.0–1.0
reason 必填不是形式主義。 記憶系統的錯誤是靜默的，而在沒有理由紀錄的情況下，「為什麼這條記憶被覆寫了」在事後幾乎無法回答。這一欄的成本是幾十個 token，價值是整個系統的可稽核性。
三、TypeScript 具體實作
// ---------- 型別 ----------
export type MemoryKind = "semantic" | "episodic" | "procedural"
export type SubjectType = "user" | "agent" | "shared" | "session"
export interface Scope {
subjectType: SubjectType
subjectId: string
projectId?: string
}
export interface TemporalMeta {
createdAt: string // ISO 8601
expiredAt?: string
validAt?: string
invalidAt?: string
lastRecalledAt?: string
recallCount: number
}
export interface SourceRef {
episodeId: string
channel: string
extractorVersion: string // 抽取邏輯的版本，重跑時用來識別
}
export interface MemoryUnit {
id: string
kind: MemoryKind
scope: Scope
content: string
summary?: string
contentHash: string
source: SourceRef
time: TemporalMeta
confidence: number
labels: string[]
}
export interface MemoryBlock {
id: string
label: string
value: string
tokenBudget: number
writable: boolean
description?: string
}
export interface Episode {
id: string
scope: Scope
occurredAt: string
raw: string
channel: string
derivedIds: string[]
}
export type Decision =
| { op: "NOOP"; reason: string; confidence: number }
| { op: "ADD"; reason: string; confidence: number }
| { op: "UPDATE"; targetId: string; reason: string; confidence: number }
| { op: "DELETE"; targetId: string; reason: string; confidence: number }
// ---------- 介面 ----------
export interface MemoryStore {
putEpisode(e: Episode): Promise<string>
putUnit(u: MemoryUnit): Promise<string>
updateUnit(id: string, patch: Partial<MemoryUnit>): Promise<void>
invalidateUnit(id: string, at: string): Promise<void>
deleteUnits(scope: Scope, filter?: UnitFilter): Promise<number>
getUnit(id: string): Promise<MemoryUnit | null>
listUnits(scope: Scope, filter?: UnitFilter): Promise<MemoryUnit[]>
}
export interface UnitFilter {
kind?: MemoryKind
labels?: string[]
validAt?: string // 只要在這個時點有效的
includeExpired?: boolean // 預設 false
}
export interface MemoryIndex {
upsert(u: MemoryUnit): Promise<void>
remove(id: string): Promise<void>
searchSemantic(q: string, scope: Scope, k: number): Promise<Hit[]>
searchLexical(q: string, scope: Scope, k: number): Promise<Hit[]>
rebuild(scope: Scope): Promise<void>
}
export interface Hit { id: string; score: number }
export interface MemoryPolicy {
shouldWrite(c: Candidate, ctx: WriteContext): Promise<{ ok: boolean; reason: string }>
reconcile(c: Candidate, neighbors: MemoryUnit[]): Promise<Decision>
shouldArchive(u: MemoryUnit, now: string): boolean
budgetFor(blockLabel: string): number
}
export interface Candidate {
kind: MemoryKind
scope: Scope
content: string
validAt?: string
labels: string[]
confidence: number
}
export interface WriteContext {
episode: Episode
isResidentTarget: boolean // 目標是常駐層嗎？門檻要更嚴
}寫入管線的骨架——藍圖的五個階段一一對應，可以直接當成實作起點：
export async function writePipeline(
episode: Episode,
deps: { store: MemoryStore; index: MemoryIndex; policy: MemoryPolicy;
extract: (e: Episode) => Promise<Candidate[]>;
hash: (s: string) => string },
): Promise<Decision[]> {
// 階段 0：真源先落地。抽取失敗也不能丟掉原始事件。
await deps.store.putEpisode(episode)
// 階段 1：抽取候選
const candidates = await deps.extract(episode)
const decisions: Decision[] = []
for (const c of candidates) {
// 階段 0'：把關。常駐層目標用更嚴的門檻。
const gate = await deps.policy.shouldWrite(c, {
episode,
isResidentTarget: c.kind === "procedural",
})
if (!gate.ok) {
decisions.push({ op: "NOOP", reason: gate.reason, confidence: 1 })
continue
}
// 階段 2：雜湊去重。零成本，放在 LLM 對帳之前。
const contentHash = deps.hash(c.content)
const existing = await deps.store.listUnits(c.scope, { kind: c.kind })
if (existing.some((u) => u.contentHash === contentHash)) {
decisions.push({ op: "NOOP", reason: "exact duplicate", confidence: 1 })
continue
}
// 階段 3：對帳。只在同 scope、同 kind 的鄰域內比對。
const hits = await deps.index.searchSemantic(c.content, c.scope, 5)
const neighbors = (
await Promise.all(hits.map((h) => deps.store.getUnit(h.id)))
).filter((u): u is MemoryUnit => u !== null && u.kind === c.kind)
const decision = await deps.policy.reconcile(c, neighbors)
decisions.push(decision)
// 階段 4：寫入真源，再更新索引。順序不可顛倒。
const now = new Date().toISOString()
switch (decision.op) {
case "ADD": {
const unit: MemoryUnit = {
id: crypto.randomUUID(),
kind: c.kind,
scope: c.scope,
content: c.content,
contentHash,
source: { episodeId: episode.id, channel: episode.channel,
extractorVersion: "v1" },
time: { createdAt: now, validAt: c.validAt, recallCount: 0 },
confidence: c.confidence,
labels: c.labels,
}
await deps.store.putUnit(unit)
await deps.index.upsert(unit)
break
}
case "UPDATE": {
// 舊值標記失效、寫入新值：保留歷史而非覆寫。
await deps.store.invalidateUnit(decision.targetId, now)
await deps.index.remove(decision.targetId)
// 接著同 ADD 的流程寫入新 unit（略）
break
}
case "DELETE":
// 預設是失效而非物理刪除。物理刪除只走 deleteUnits()。
await deps.store.invalidateUnit(decision.targetId, now)
await deps.index.remove(decision.targetId)
break
case "NOOP":
break
}
}
return decisions
}召回與組裝——注意回傳值裡的 trace，它是可觀測性的資料來源：
export async function assembleContext(
query: string,
scope: Scope,
deps: { store: MemoryStore; index: MemoryIndex; policy: MemoryPolicy;
blocks: MemoryBlock[]; countTokens: (s: string) => number },
): Promise<{ text: string; trace: AssembleTrace }> {
const parts: string[] = []
const trace: AssembleTrace = { residentIds: [], recalledIds: [], tokens: 0 }
// 階段 1：常駐層。無條件載入，按各自配額截斷。不經過檢索。
for (const b of deps.blocks) {
const budget = deps.policy.budgetFor(b.label)
const value = truncateToTokens(b.value, budget, deps.countTokens)
parts.push(`## ${b.label}\n${value}`)
trace.residentIds.push(b.id)
}
// 階段 2：候選生成。語意＋詞彙兩路，之後合併。
const [sem, lex] = await Promise.all([
deps.index.searchSemantic(query, scope, 20),
deps.index.searchLexical(query, scope, 20),
])
const merged = reciprocalRankFusion([sem, lex])
// 階段 3：時效過濾。只要目前有效的事實。
const now = new Date().toISOString()
const units = (
await Promise.all(merged.map((h) => deps.store.getUnit(h.id)))
).filter((u): u is MemoryUnit =>
u !== null && !u.time.expiredAt && (!u.time.invalidAt || u.time.invalidAt > now),
)
// 階段 4：按召回層自己的配額截斷（與常駐層的配額互不侵蝕）。
let used = 0
const recallBudget = deps.policy.budgetFor("__recall__")
for (const u of units) {
const t = deps.countTokens(u.summary ?? u.content)
if (used + t > recallBudget) break
parts.push(u.summary ?? u.content)
trace.recalledIds.push(u.id)
used += t
}
const text = parts.join("\n\n")
trace.tokens = deps.countTokens(text)
return { text, trace } // trace 必須被記錄下來
}
export interface AssembleTrace {
residentIds: string[]
recalledIds: string[]
tokens: number
}四、SQL schema
單一資料庫就能實作藍圖的全部結構，不需要圖資料庫（除了 Q6 的多跳需求）。
CREATE TABLE episodes (
id TEXT PRIMARY KEY,
subject_type TEXT NOT NULL,
subject_id TEXT NOT NULL,
project_id TEXT,
occurred_at TIMESTAMPTZ NOT NULL,
channel TEXT NOT NULL,
raw TEXT NOT NULL
);
-- episodes 只 INSERT，不 UPDATE、不 DELETE（刪除請求除外）
CREATE TABLE memory_units (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL CHECK (kind IN ('semantic','episodic','procedural')),
subject_type TEXT NOT NULL CHECK (subject_type IN ('user','agent','shared','session')),
subject_id TEXT NOT NULL,
project_id TEXT,
content TEXT NOT NULL,
summary TEXT,
content_hash TEXT NOT NULL,
episode_id TEXT NOT NULL REFERENCES episodes(id),
extractor_version TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL, -- 系統何時知道
expired_at TIMESTAMPTZ, -- 系統何時判定失效
valid_at TIMESTAMPTZ, -- 現實中何時開始為真
invalid_at TIMESTAMPTZ, -- 現實中何時停止為真
last_recalled_at TIMESTAMPTZ,
recall_count INTEGER NOT NULL DEFAULT 0,
confidence REAL NOT NULL DEFAULT 1.0,
labels TEXT[] NOT NULL DEFAULT '{}'
);
-- 主查詢路徑：範圍 ＋ 型別 ＋ 只要有效的
CREATE INDEX idx_units_scope ON memory_units
(subject_type, subject_id, kind) WHERE expired_at IS NULL;
-- 去重路徑
CREATE UNIQUE INDEX idx_units_hash ON memory_units
(subject_type, subject_id, kind, content_hash) WHERE expired_at IS NULL;
-- 歸檔候選：失效已久、或久未召回
CREATE INDEX idx_units_archive ON memory_units (expired_at, last_recalled_at);
CREATE TABLE memory_blocks (
id TEXT PRIMARY KEY,
subject_type TEXT NOT NULL,
subject_id TEXT NOT NULL,
label TEXT NOT NULL,
value TEXT NOT NULL,
token_budget INTEGER NOT NULL,
writable BOOLEAN NOT NULL DEFAULT TRUE,
description TEXT,
UNIQUE (subject_type, subject_id, label)
);
-- 決策紀錄：稽核與 debug 的唯一依據
CREATE TABLE memory_decisions (
id BIGSERIAL PRIMARY KEY,
episode_id TEXT NOT NULL REFERENCES episodes(id),
op TEXT NOT NULL,
target_id TEXT,
reason TEXT NOT NULL,
confidence REAL NOT NULL,
decided_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 組裝紀錄：可觀測性與 token 效率量測的來源
CREATE TABLE assemble_traces (
id BIGSERIAL PRIMARY KEY,
turn_id TEXT NOT NULL,
resident_ids TEXT[] NOT NULL,
recalled_ids TEXT[] NOT NULL,
tokens INTEGER NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);向量索引可以放在同一個資料庫（pgvector）或外部向量庫；詞彙檢索用 Postgres 的全文檢索或 SQLite 的 FTS5 都可以。關鍵是這兩者都是衍生物——MemoryIndex.rebuild() 應該能從 memory_units 完整重建它們。
最後兩張表（memory_decisions 與 assemble_traces）在多數實作裡不存在，而它們的缺席正是「不知道 agent 為什麼記住／召回了什麼」的直接原因。加上它們的成本是兩張 append-only 表，收益是整個系統從黑盒變成可觀測。
五、Adapter 範例：把框架接到這組介面
// 用 Mem0 實作寫入管線的抽取與對帳，其餘自己控制
export class Mem0Policy implements MemoryPolicy {
constructor(private client: Mem0Client) {}
async shouldWrite(c: Candidate, ctx: WriteContext) {
// 常駐層（規則）不交給 LLM 判斷，走人工審核佇列
if (ctx.isResidentTarget) {
return { ok: false, reason: "resident writes require human review" }
}
return { ok: true, reason: "non-resident, delegate to reconcile" }
}
async reconcile(c: Candidate, neighbors: MemoryUnit[]): Promise<Decision> {
const r = await this.client.decide(c.content, neighbors.map((n) => n.content))
return { op: r.op, targetId: r.targetId, reason: r.reason ?? "mem0",
confidence: r.score ?? 0.5 } as Decision
}
shouldArchive(u: MemoryUnit, now: string) {
if (u.time.expiredAt && daysBetween(u.time.expiredAt, now) > 90) return true
if (u.time.recallCount === 0 && daysBetween(u.time.createdAt, now) > 180) return true
return false
}
budgetFor(label: string) {
return { rules: 1200, user_profile: 600, persona: 300, __recall__: 1500 }[label] ?? 400
}
}這個 adapter 示範了藍圖的實際價值：shouldWrite 對常駐層加了一道框架本身沒有的護欄（規則變更需人工審核），shouldArchive 與 budgetFor 補上了 Mem0 沒有的容量控制——而這些都不需要修改 Mem0，也不需要在換掉 Mem0 時重寫。