04 · RAG 与记忆系统

架构剖析第四篇：记忆分层（scope × category）、RAG 打分公式、对话树、MemoryWriter/MemoryReader/Consolidator、DeciderService 驱逐策略。这是 Familiars 的"长期记忆"核心。对应 ARCHITECTURE.md §5。

§5.1 为什么不用向量数据库

方案	不选原因
Pinecone / Weaviate / Qdrant	数据量级（单用户，几千~上万 embedding）内存 + SQLite 就够；引入额外服务和运维
纯文件 + 无索引	检索太慢；无法做稠密召回
Familiars 方案	内存 + SQLite + 文件树混合；向量存在 `workspace/` 与 `familiars.db`，隐私不出本机；抽象在 `memory_store` 层，后续换外部向量库也直接

§5.2 记忆作用域 × 类别矩阵

5.2.1 作用域（Scope）

定义在 src/utils/memory_store.py:SCOPES：

python

SCOPES: tuple[str, ...] = ("user", "identity", "dialogue", "ephemeral")

Scope	持久化路径	含义	TTL 语义
`user`	`workspace/user/<category>.md`	用户画像（跨 agent 共享）	长期，跨会话
`identity`	`workspace/agents/<agent>/identity/<category>.md`	Agent 自我认知	长期，跟随 agent
`dialogue`	`workspace/agents/<agent>/<dialogue>/dialogue/<category>.md`	对话上下文	跟随对话
`ephemeral`	`workspace/agents/<agent>/<dialogue>/ephemeral/<category>.md`	每轮临时	回合结束即清除

5.2.2 类别（Category）

定义在 src/utils/memory_store.py:CATEGORIES：

python

CATEGORIES: tuple[str, ...] = (
    "profile",      # 用户/Agent 画像
    "preference",   # 偏好
    "relation",     # 关系
    "knowledge",    # 知识
    "event",        # 事件
    "state",        # 状态
    "procedure",    # 流程
    "directive",    # 指令
)

5.2.3 合法（Scope, Category）组合

不是完整笛卡尔积——仅以下组合合法（VALID_CELLS）：

python

VALID_CELLS: frozenset[tuple[str, str]] = frozenset({
    ("user", "profile"),       # 用户画像
    ("user", "preference"),    # 用户偏好
    ("user", "relation"),      # 用户关系
    ("user", "knowledge"),     # 用户知识
    ("user", "event"),         # 用户事件
    
    ("identity", "preference"),    # Agent 偏好
    ("identity", "procedure"),     # Agent 行为流程
    ("identity", "directive"),     # Agent 指令
    
    ("dialogue", "preference"),    # 对话内偏好
    ("dialogue", "relation"),      # 对话内关系
    ("dialogue", "knowledge"),     # 对话内知识
    ("dialogue", "event"),         # 对话内事件
    ("dialogue", "state"),         # 对话内状态
    ("dialogue", "directive"),     # 对话内指令
    
    ("ephemeral", "state"),        # 临时状态
    ("ephemeral", "directive"),    # 临时指令
})

5.2.4 路径解析

python

def resolve_path(*, workspace, scope, category, agent_id="agent_0",
                 dialogue_id=None) -> Path:
    fname = f"{category}.md"
    if scope == "user":
        return ws / "user" / fname
    if scope == "identity":
        return ws / "agents" / agent_id / "identity" / fname
    if scope == "dialogue":
        return ws / "agents" / agent_id / dialogue_id / "dialogue" / fname
    if scope == "ephemeral":
        return ws / "agents" / agent_id / dialogue_id / "ephemeral" / fname

§5.3 记忆条目格式

5.3.1 磁盘格式

每条记忆是一个逻辑条目，格式为 HTML 注释包裹 JSON 元数据 + markdown 正文：

<!-- {"id":"m_a1b2c3d4","scope":"user","category":"profile","ts":"2026-08-29T12:00:00+00:00","tags":["name"],"salience":0.8,"ttl":null,"supersedes":null,"deprecated":false,"schema_version":2} -->
用户名字是 Alice，喜欢猫。

5.3.2 MemoryRecord 数据结构

python

@dataclass
class MemoryRecord:
    id: str                                    # 稳定 str，applier 生成 "m_" + token_hex(4)
    scope: str                                 # "user" | "identity" | "dialogue" | "ephemeral"
    category: str = "knowledge"                # 8 类别之一
    ts: str = ""                               # ISO 8601 时间戳
    content: str = ""                          # markdown 正文
    tags: list[str] = []                       # 标签
    salience: float = 0.5                      # 显著度 [0, 1]
    ttl: Optional[dict] = None                 # TTL 策略
    supersedes: Optional[str] = None           # 被替代的条目 id
    deprecated: bool = False                   # 已弃用 → reader 过滤
    schema_version: int = 2                    # 格式版本
    evidence: Optional[dict] = None            # 来源证明 {"dialogue_id", "turn_id"}
    epistemic_status: str = "asserted"         # "asserted" | "inferred" | "speculative"
    last_accessed_ts: Optional[str] = None     # 最近访问时间（reader 批量 flush）
    access_count: int = 0                      # 访问频次信号
    pinned: bool = False                       # 固定 → 永不自动 consolidate
    derived_from: list[str] = []               # 来源 id 列表（consolidator 产出）
    source_quote: list[str] = []               # writer 来源引用

5.3.3 TTL 策略

python

def is_expired(rec, *, now=None, turn_counter=None,
               session_active=True, dialogue_active=True) -> bool:
    if rec.ttl is None:
        return False
    kind = rec.ttl.get("kind")
    if kind == "absolute":
        # {"kind": "absolute", "expires_at": "2026-09-01T00:00:00Z"}
        return now >= datetime.fromisoformat(rec.ttl["expires_at"])
    if kind == "turns":
        # {"kind": "turns", "turns": 5, "created_turn": 10}
        return (turn_counter - created_turn) >= n
    if kind == "scope":
        # {"kind": "scope", "scope": "session" | "dialogue"}
        scope = rec.ttl.get("scope")
        if scope == "session":
            return not session_active
        if scope == "dialogue":
            return not dialogue_active

§5.4 RAG 打分公式

5.4.1 对话树附着（Construction）

新 Q 节点选择父节点时的评分：

score = α·cos_sim + β·recency + δ·is_immediate_predecessor

默认参数（DialogTree.__init__）：

参数	默认值	含义
`alpha`	0.3	语义相似度权重
`beta`	0.3	时间衰减权重
`delta`	0.35	直接前驱权重
`tau_attach`	0.35	附着阈值（低于此分 → 新根）
`recency_tau`	600.0	时间衰减速率（秒），`exp(-Δt / 600)`
`tree_tau`	3.0	树距离衰减速率（跳数）

python

def _choose_parent(self, node: Node) -> Optional[int]:
    best_score = -math.inf
    for cand in self.node_list:
        sim = self._cosine(node.embedding, cand.embedding)
        dt = max(0.0, node.timestamp - cand.timestamp)
        recency = math.exp(-dt / self.recency_tau)
        is_pred = 1.0 if cand.id == self._last_id else 0.0
        score = self.alpha * sim + self.beta * recency + self.delta * is_pred
        if score > best_score:
            best_score = score
            best_cand = cand
    if best_score < self.tau_attach:
        return None  # 新根（话题转移 → 森林）
    # 如果最佳是 Q 节点且有父 A，爬到 A（子话题挂在共同回答下）
    if best_cand.node_type == "Q" and best_cand.parent is not None:
        return best_cand.parent
    return best_cand.id

5.4.2 查询（Query）

DialogTree.query(text, k) 复用 embedding + 拓扑：

final = α·cos_sim + β·recency + γ·exp(-tree_distance / τ_d)

查询参数：

参数	默认值	含义
`alpha`	0.3	语义相似度
`beta`	0.3	时间衰减
`gamma`	0.15	树距离权重
`tree_tau`	3.0	跳数衰减尺度

5.4.3 Embedding 模型

模型：BAAI/bge-m3（默认 EMBEDDING_MODEL 环境变量）
所有组件共用：DialogTree、AssistantRouter、RAG 稠密召回、rewriter 数据管线 → 全仓可比向量
不缓存模型实例在 self 上（property 路由到 singleton getter），允许 VRAM 驱逐释放 ~700MB

§5.5 Node 数据结构

python

@dataclass
class Node:
    title: str                          # 标题（≤40 字符 + "…"）
    content: str = ""                   # 正文
    id: int = -1                        # 列表索引（create_node 时分配）
    level: int = 0                      # 树深度
    parent: Optional[int] = None        # 父节点 id
    node_type: str = "Q"                # "Q" (用户) / "A" (助手)
    embedding: list[float] = []         # bge-m3 向量
    children: list[int] = []            # 子节点 id 列表
    timestamp: float = time.time()      # Unix 时间戳
    origin: str = "response"            # "user" / "response" / "proactive"
    metadata: dict = {}                 # 元数据（redactions, decision_id, ...）

5.5.1 origin 语义

origin	含义
`"user"`	用户输入的 Q 节点
`"response"`	模型回复的 A 节点（响应用户 Q）
`"proactive"`	Agent 主动发起的 A 节点（decider/reminder 触发，无前置 Q）

5.5.2 隐私脱敏回放

python

def node_content_for_llm(node: Node) -> str:
    """Return node.content with stored privacy decisions replayed.
    
    Q 节点的 metadata['redactions'] 携带每发现的决策：
    - span=[start, end]
    - decision: "ignore" | "mask" | "record"
    - token: 替换字符串
    
    按 span 倒序替换。redactions_content_hash 不匹配 → 回退到原始内容。
    """

§5.6 DialogTree 存储格式

json

{
  "nodes": [
    {"id": 0, "title": "...", "content": "...", "node_type": "Q", 
     "parent": null, "children": [1], "embedding": [0.1, ...],
     "timestamp": 1724860800.0, "origin": "user", "metadata": {}},
    {"id": 1, "title": "...", "content": "...", "node_type": "A",
     "parent": 0, "children": [], "embedding": [0.2, ...],
     "timestamp": 1724860801.0, "origin": "response", "metadata": {}}
  ],
  "roots": [0],
  "_last_id": 1
}

存储路径：workspace/agents/agent_<id>/dialogue_<n>/dialogue.json

§5.7 MemoryWriter（写入链）

src/models/memorizer/writer.py 决定什么值得记。

5.7.1 写入流程

用户消息 + 对话历史 → MemorizerWriter（本地小模型）
    → 判断是否值得记
    → 分类 (scope, category)
    → apply_writes(writer_output_json, ...)
    → MemoryStore.append(scope, category, content, ...)
    → 写入 workspace/user/<category>.md（或 identity/dialogue/ephemeral 对应路径）

5.7.2 WriterOutput JSON

json

{
  "writes": [
    {
      "op": "add",           // "add" | "update" | "skip"
      "scope": "user",
      "category": "profile",
      "entry": {
        "content": "用户名字是 Alice",
        "tags": ["name"],
        "salience": 0.8,
        "ttl": null,
        "supersedes_id": null,
        "epistemic_status": "asserted",
        "pinned": false,
        "source_quote": ["用户消息: '我叫Alice'"]
      }
    }
  ]
}

5.7.3 apply_writes 预算

python

def apply_writes(writer_output_json, *, workspace, agent_id="agent_0",
                 dialogue_id=None, turn_counter=None, turn_id=None,
                 max_chars_total=1500) -> list[MemoryRecord]:
    """Apply WriterOutput JSON to disk. Returns new records committed.
    
    贪心预算：按顺序接受写入，直到累计内容超过 max_chars_total。
    第一条写入始终允许。超预算的 turn 保留最高优先级事实。
    """

§5.8 MemoryReader（读取链）

src/models/memorizer/reader.py 组装 RAG 上下文。

5.8.1 读取流程

每次 NL 质问 → DialogueManager.build_rag_context(query)
    → 稠密召回（bge-m3 cosine）
    → 时间衰减
    → 树距离
    → 组合分排序
    → top_k 注入上下文

5.8.2 is_live 过滤

python

def is_live(rec, **ctx) -> bool:
    return not rec.deprecated and not is_expired(rec, **ctx)

§5.9 Consolidator（整合器）

src/models/memorizer/consolidator.py 定期整合重复/相似的记忆。

python

# consolidator 流程：
# 1. 按 (scope, category) 分组
# 2. 计算 embedding 相似度
# 3. 高相似度集群 → 生成综合条目
# 4. 标记原始条目 deprecated=True
# 5. 新条目 derived_from=[...source_ids]

§5.10 DeciderService 驱逐策略

5.10.1 VRAM 门控集成

Decider 模型被 VRAMGate 纳入可驱逐列表（优先级 3 — NICE）：

优先级	模型	类别	说明
1 — CRITICAL	TTS/whisper/embedder（用户活跃请求）	sync	永不驱逐
2 — IMPORTANT	选中 agent 的 decider, bge embedding, reranker	sync	用户交互时
3 — NICE	memorizer writer, mini-model, EasyOCR, emotion tagger	queue	可被中断 + checkpoint
4 — BACKGROUND	索引 / 训练	queue	永不用户可见

5.10.2 驱逐类别

python

@dataclass
class _Entry:
    model_id: str
    priority: int
    estimated_mb: float
    on_evict: Optional[Callable[[], None]] = None
    category: Literal["sync", "queue"] = "sync"
    is_working: bool = False
    on_checkpoint: Optional[Callable[[], Any]] = None
    last_active_at: float = field(default_factory=time.monotonic)
    idle_ttl_sec: Optional[float] = None

sync：用户阻塞。新 sync acquire 可驱逐任何 idle resident 或 queue resident（带 checkpoint）。
queue：可中断。eviction 时运行 on_checkpoint() 快照状态供后续恢复。

5.10.3 空闲回收

python

# gate 比较 time.monotonic() - last_active_at 与 VRAM_GATE_IDLE_EVICT_SEC
# 超过阈值的 resident 被回收，VRAM 归还池中

5.10.4 线程优先级提升

python

@contextmanager
def boost_thread_priority(priority: int) -> Iterator[None]:
    """为用户阻塞线程临时提升优先级。
    
    嵌套取 min —— 内层 CRITICAL + 外层 NICE = 内层 CRITICAL。
    """

§5.11 可观测性

接口	用途
`memory writer-status`（CLI）	查看 writer 状态
`GET /api/memory_status`	记忆规模/命中统计
`POST /api/agents/{id}/release-memory`	手动释放常驻内存
`GET /api/memory`	记忆操作

下一篇：05-持久化层 →

04 · RAG 与记忆系统 ​

§5.1 为什么不用向量数据库 ​

§5.2 记忆作用域 × 类别矩阵 ​

5.2.1 作用域（Scope） ​

5.2.2 类别（Category） ​

5.2.3 合法（Scope, Category）组合 ​

5.2.4 路径解析 ​

§5.3 记忆条目格式 ​

5.3.1 磁盘格式 ​

5.3.2 MemoryRecord 数据结构 ​

5.3.3 TTL 策略 ​

§5.4 RAG 打分公式 ​

5.4.1 对话树附着（Construction） ​

5.4.2 查询（Query） ​

5.4.3 Embedding 模型 ​

§5.5 Node 数据结构 ​

5.5.1 origin 语义 ​

5.5.2 隐私脱敏回放 ​

§5.6 DialogTree 存储格式 ​

§5.7 MemoryWriter（写入链） ​

5.7.1 写入流程 ​

5.7.2 WriterOutput JSON ​

5.7.3 apply_writes 预算 ​

§5.8 MemoryReader（读取链） ​

5.8.1 读取流程 ​

5.8.2 is_live 过滤 ​

§5.9 Consolidator（整合器） ​

§5.10 DeciderService 驱逐策略 ​

5.10.1 VRAM 门控集成 ​

5.10.2 驱逐类别 ​

5.10.3 空闲回收 ​

5.10.4 线程优先级提升 ​

§5.11 可观测性 ​

04 · RAG 与记忆系统

§5.1 为什么不用向量数据库

§5.2 记忆作用域 × 类别矩阵

5.2.1 作用域（Scope）

5.2.2 类别（Category）

5.2.3 合法（Scope, Category）组合

5.2.4 路径解析

§5.3 记忆条目格式

5.3.1 磁盘格式

5.3.2 MemoryRecord 数据结构

5.3.3 TTL 策略

§5.4 RAG 打分公式

5.4.1 对话树附着（Construction）

5.4.2 查询（Query）

5.4.3 Embedding 模型

§5.5 Node 数据结构

5.5.1 origin 语义

5.5.2 隐私脱敏回放

§5.6 DialogTree 存储格式

§5.7 MemoryWriter（写入链）

5.7.1 写入流程

5.7.2 WriterOutput JSON

5.7.3 apply_writes 预算

§5.8 MemoryReader（读取链）

5.8.1 读取流程

5.8.2 is_live 过滤

§5.9 Consolidator（整合器）

§5.10 DeciderService 驱逐策略

5.10.1 VRAM 门控集成

5.10.2 驱逐类别

5.10.3 空闲回收

5.10.4 线程优先级提升

§5.11 可观测性