07 数据库与 Schema 指南

erDiagram
    memories ||--|| memory_vectors : "1:1"
    memories ||--o{ entity_index : "entity refs"
    memories ||--o{ relations : "source / target"
    memories ||--o{ evolution_log : "events"
    memories ||--o{ audit_log : "audit target"
    memories ||..|| memories_fts : "FTS shadow"

CREATE TABLE IF NOT EXISTS memories (
    id TEXT PRIMARY KEY,
    content TEXT NOT NULL,
    memory_type TEXT NOT NULL,
    created_at TEXT NOT NULL,
    last_accessed TEXT NOT NULL,
    access_count INTEGER NOT NULL DEFAULT 0,
    valid_from TEXT,
    valid_until TEXT,
    trust_score REAL NOT NULL,
    importance REAL NOT NULL,
    layer TEXT NOT NULL,
    decay_rate REAL NOT NULL,
    source_id TEXT NOT NULL,
    causal_parent_id TEXT REFERENCES memories(id),
    supersedes_id TEXT REFERENCES memories(id),
    entity_refs_json TEXT NOT NULL,
    tags_json TEXT NOT NULL,
    deleted_at TEXT
);

CREATE TABLE IF NOT EXISTS memory_vectors (
    memory_id TEXT PRIMARY KEY REFERENCES memories(id) ON DELETE CASCADE,
    memory_rowid INTEGER UNIQUE,
    embedding_json TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS entity_index (
    entity TEXT NOT NULL,
    memory_id TEXT NOT NULL REFERENCES memories(id) ON DELETE CASCADE,
    PRIMARY KEY (entity, memory_id)
);

CREATE TABLE IF NOT EXISTS relations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source_id TEXT NOT NULL REFERENCES memories(id) ON DELETE CASCADE,
    target_id TEXT NOT NULL REFERENCES memories(id) ON DELETE CASCADE,
    relation_type TEXT NOT NULL,
    created_at TEXT NOT NULL,
    UNIQUE (source_id, target_id, relation_type)
);

CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts
USING fts5(
    content,
    tags,
    content = 'memories',
    content_rowid = 'rowid'
);

CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
    INSERT INTO memories_fts(rowid, content, tags)
    VALUES (
        new.rowid,
        new.content,
        trim(replace(replace(replace(replace(new.tags_json, '[', ' '), ']', ' '), '\"', ' '), ',', ' '))
    );
END;

CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
    INSERT INTO memories_fts(memories_fts, rowid, content, tags)
    VALUES (
        'delete',
        old.rowid,
        old.content,
        trim(replace(replace(replace(replace(old.tags_json, '[', ' '), ']', ' '), '\"', ' '), ',', ' '))
    );
END;

CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN
    INSERT INTO memories_fts(memories_fts, rowid, content, tags)
    VALUES (
        'delete',
        old.rowid,
        old.content,
        trim(replace(replace(replace(replace(old.tags_json, '[', ' '), ']', ' '), '\"', ' '), ',', ' '))
    );
    INSERT INTO memories_fts(rowid, content, tags)
    VALUES (
        new.rowid,
        new.content,
        trim(replace(replace(replace(replace(new.tags_json, '[', ' '), ']', ' '), '\"', ' '), ',', ' '))
    );
END;

def _build_fts_query(query: str) -> str:
    terms = re.findall(r"[\w\u4e00-\u9fff-]+", query.lower())
    if not terms:
        return ""
    return " OR ".join(f'"{term}"' for term in terms)

rows = self.connection.execute(
    f"""
    SELECT m.*, v.embedding_json, bm25(memories_fts) AS rank_score
    FROM memories_fts
    JOIN memories m ON m.rowid = memories_fts.rowid
    LEFT JOIN memory_vectors v ON v.memory_id = m.id
    WHERE memories_fts MATCH ?
      AND m.deleted_at IS NULL
      {memory_type_clause}
    ORDER BY rank_score
    LIMIT ?
    """,
    params,
).fetchall()
return [(self._row_to_memory(row), 1.0 / (1.0 + abs(row["rank_score"]))) for row in rows]

def search_by_vector(self, embedding: list[float], limit: int = 10, memory_type: str | None = None) -> list[tuple[MemoryItem, float]]:
    if self._sqlite_vec_enabled:
        results = self._search_by_vector_sqlite_vec(embedding=embedding, limit=limit, memory_type=memory_type)
        if results:
            return results
    return self._search_by_vector_fallback(embedding=embedding, limit=limit, memory_type=memory_type)

for row in rows:
    stored_embedding = json.loads(row["embedding_json"])
    score = _cosine_similarity(embedding, stored_embedding)
    scored.append((self._row_to_memory(row), score))
scored.sort(key=lambda item: item[1], reverse=True)
return scored[:limit]

def _ensure_vec_index_table(self, dimension: int) -> None:
    current_dimension = self._get_backend_meta("sqlite_vec_dimension")
    if current_dimension and int(current_dimension) != dimension:
        raise ValueError(
            f"sqlite-vec index dimension mismatch: existing={current_dimension}, requested={dimension}"
        )
    self.connection.execute(
        f"""
        CREATE VIRTUAL TABLE IF NOT EXISTS memory_vec_index
        USING vec0(
            memory_rowid INTEGER PRIMARY KEY,
            embedding FLOAT[{dimension}],
            memory_type TEXT,
            layer TEXT,
            source_id TEXT,
            trust_score FLOAT,
            created_at TEXT,
            last_accessed TEXT
        )
        """
    )
    self._set_backend_meta("sqlite_vec_dimension", str(dimension))

WITH RECURSIVE ancestors(id, depth) AS (
	SELECT causal_parent_id, 1
	FROM memories
	WHERE id = ? AND causal_parent_id IS NOT NULL
	UNION ALL
	SELECT m.causal_parent_id, a.depth + 1
	FROM ancestors a
	JOIN memories m ON m.id = a.id
	WHERE a.depth < ? AND m.causal_parent_id IS NOT NULL
)
SELECT m.*, v.embedding_json
FROM ancestors a
JOIN memories m ON m.id = a.id
LEFT JOIN memory_vectors v ON v.memory_id = m.id
WHERE m.deleted_at IS NULL
ORDER BY a.depth ASC

WITH RECURSIVE ancestors(id, depth) AS (
    SELECT causal_parent_id, 1
    FROM memories
    WHERE id = ? AND causal_parent_id IS NOT NULL
    UNION ALL
    SELECT m.causal_parent_id, a.depth + 1
    FROM ancestors a
    JOIN memories m ON m.id = a.id
    WHERE a.depth < ? AND m.causal_parent_id IS NOT NULL
)
SELECT m.*, v.embedding_json, a.depth
FROM ancestors a
JOIN memories m ON m.id = a.id
LEFT JOIN memory_vectors v ON v.memory_id = m.id
WHERE m.deleted_at IS NULL
ORDER BY a.depth ASC