Bộ nhớ agent: chọn theo truy vấn và cách cập nhật

Câu hỏi: khi nào một file đủ dùng, và database giúp giảm phần công việc nào?

Cần biết trước: JSON, SQL cơ bản, revision và checkpoint. Lab dùng Python 3.14.4/SQLite 3.50.4 trên macOS arm64, filesystem local; flock là POSIX, chưa thử Windows hoặc network filesystem. Dataset và note đều giả; không gọi LLM, vector search hoặc đo chất lượng câu trả lời. Không dùng benchmark này để chọn SurrealDB hay chứng nhận memory production.

Chốt hợp đồng trước kho lưu

Mỗi record có ID, text, parent ID, revision, note và SHA-256 của bundle nguồn. Các thao tác cần là lookup ID, substring case-sensitive, truy ngược parent, kiểm provenance, và cập nhật note nếu revision chưa đổi. Text search này không phải semantic retrieval/BM25; quan hệ một bước không phải graph reasoning nhiều chặng.

File vẫn biểu diễn được ID và quan hệ, nhưng code phải đọc/kiểm và quản lý xung đột. Database cung cấp query/constraint/transaction; ứng dụng vẫn chịu trách nhiệm provenance, policy và nội dung nguồn. Một query nhanh không khiến sự thật cũ thành mới.

Hai cách lưu cùng record

Lưu các file sau vào thư mục lab trống. sources.json là bundle nguồn; JSON và SQLite là hai snapshot cùng text và note. Rebuild chạy khi đã dừng writer/source editor, giữ note, tăng revision khi nguồn đổi. Không đo concurrent rebuild.

from __future__ import annotations

import fcntl
import hashlib
import json
import os
import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import TypedDict, cast


class Record(TypedDict):
    id: int
    text: str
    parent: int | None
    rev: int
    note: str
    sha: str


SOURCE = Path("sources.json")


def digest() -> str:
    return hashlib.sha256(SOURCE.read_bytes()).hexdigest()


def read_json(path: Path) -> list[Record]:
    return cast(list[Record], json.loads(path.read_text()))


def atomic_json(path: Path, rows: list[Record]) -> None:
    temporary = path.with_suffix(".tmp")
    with temporary.open("w") as stream:
        json.dump(rows, stream, ensure_ascii=False)
        stream.flush()
        os.fsync(stream.fileno())
    temporary.replace(path)


def rebuilt(old: list[Record]) -> list[Record]:
    previous = {r["id"]: r for r in old}
    sha = digest()
    rows = read_json(SOURCE)
    for row in rows:
        if prior := previous.get(row["id"]):
            row["note"] = prior["note"]
            row["rev"] = prior["rev"] + int(prior["sha"] != sha)
        row["sha"] = sha
    return rows


class FileStore:
    path = Path("memory.json")

    @contextmanager
    def locked(self) -> Iterator[None]:
        with Path("memory.lock").open("a") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(lock, fcntl.LOCK_UN)

    def all(self) -> list[Record]:
        return read_json(self.path) if self.path.exists() else []

    def get(self, key: int) -> Record:
        return next(r for r in self.all() if r["id"] == key)

    def search(self, term: str) -> list[int]:
        return [r["id"] for r in self.all() if term in r["text"]]

    def children(self, parent: int) -> list[int]:
        return [r["id"] for r in self.all() if r["parent"] == parent]

    def cas(self, key: int, rev: int, note: str) -> bool:
        with self.locked():
            rows = self.all()
            row = next(r for r in rows if r["id"] == key)
            if row["rev"] != rev:
                return False
            row["rev"] += 1
            row["note"] = note
            atomic_json(self.path, rows)
            return True

    def rebuild(self) -> None:
        with self.locked():
            atomic_json(self.path, rebuilt(self.all()))

    def close(self) -> None:
        pass


class SqlStore:
    def __init__(self, path: str = "memory.sqlite") -> None:
        self.db = sqlite3.connect(path, timeout=10, autocommit=True)
        self.db.execute("PRAGMA foreign_keys=ON")
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.execute("""CREATE TABLE IF NOT EXISTS docs (
            id INTEGER PRIMARY KEY, text TEXT NOT NULL,
            parent INTEGER REFERENCES docs(id) DEFERRABLE INITIALLY DEFERRED,
            rev INTEGER NOT NULL CHECK(rev>=0), note TEXT NOT NULL, sha TEXT NOT NULL
        )""")
        self.db.execute("CREATE INDEX IF NOT EXISTS by_parent ON docs(parent)")

    def select(self, sql: str, params: tuple[int | str, ...] = ()) -> list[Record]:
        return [
            Record(id=r[0], text=r[1], parent=r[2], rev=r[3], note=r[4], sha=r[5])
            for r in self.db.execute(sql, params)
        ]

    def all(self) -> list[Record]:
        return self.select("SELECT * FROM docs ORDER BY id")

    def get(self, key: int) -> Record:
        return self.select("SELECT * FROM docs WHERE id=?", (key,))[0]

    def search(self, term: str) -> list[int]:
        return [
            r[0]
            for r in self.db.execute(
                "SELECT id FROM docs WHERE instr(text,?)>0 ORDER BY id", (term,)
            )
        ]

    def children(self, parent: int) -> list[int]:
        return [
            r[0]
            for r in self.db.execute(
                "SELECT id FROM docs WHERE parent=? ORDER BY id", (parent,)
            )
        ]

    def cas(self, key: int, rev: int, note: str) -> bool:
        changed = self.db.execute(
            "UPDATE docs SET note=?,rev=rev+1 WHERE id=? AND rev=?", (note, key, rev)
        )
        return changed.rowcount == 1

    def rebuild(self) -> None:
        self.db.execute("BEGIN IMMEDIATE")
        try:
            rows = rebuilt(self.all())
            self.db.execute("DELETE FROM docs")
            self.db.executemany(
                "INSERT INTO docs VALUES(?,?,?,?,?,?)",
                [
                    (r["id"], r["text"], r["parent"], r["rev"], r["note"], r["sha"])
                    for r in rows
                ],
            )
            self.db.execute("COMMIT")
        except Exception:
            self.db.execute("ROLLBACK")
            raise

    def close(self) -> None:
        self.db.close()


Store = FileStore | SqlStore


def fresh(store: Store, key: int) -> Record:
    row = store.get(key)
    if row["sha"] != digest():
        raise ValueError("source changed; rebuild required")
    return row

File dùng lock riêng không bị thay inode khi replace; mọi writer phải hợp tác với lock và CAS. Reader nhận một snapshot nguyên file nhờ replace cùng filesystem. fsync file chưa fsync directory nên chưa chứng minh power-loss durability. TypedDict cast chỉ dùng với JSON tin cậy do lab sinh; importer production cần validate schema. flock.

SQLite chỉ có một writer tại một thời điểm; conditional UPDATE kiểm revision trong cùng statement. autocommit=True làm statement CAS commit riêng; rebuild dùng BEGIN/COMMIT rõ ràng. Busy timeout không phải retry nghiệp vụ. FK/parent index không tạo semantic search. sqlite3.

Đo query và kiểm các lỗi làm memory sai

from __future__ import annotations

import json
import platform
import select
import sqlite3
import subprocess
import sys
from contextlib import closing
from datetime import UTC, datetime
from pathlib import Path
from time import perf_counter

from stores import (
    SOURCE,
    FileStore,
    Record,
    SqlStore,
    Store,
    atomic_json,
    digest,
    fresh,
)


def queries(store: Store) -> list[object]:
    return [
        *[store.get(i) for i in range(0, 1000, 10)],
        *[store.search(f"group{g} ") for g in range(5)],
        *[store.children(i) for i in range(0, 100, 10)],
    ]


def race(mode: str, store: Store) -> None:
    clients: list[subprocess.Popen[str]] = []
    try:
        for _ in range(2):
            clients.append(
                subprocess.Popen(
                    [sys.executable, "writer.py", mode],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                )
            )
        for client in clients:
            assert client.stdout is not None
            assert select.select([client.stdout], [], [], 10)[0], "writer not ready"
            assert client.stdout.readline().strip() == "ready rev=0"
        for client in clients:
            assert client.stdin is not None
            client.stdin.write("go\n")
            client.stdin.flush()
        outputs = []
        for client in clients:
            out, err = client.communicate(timeout=15)
            assert client.returncode == 0 and not err, (out, err)
            outputs.append(out.strip())
        assert sorted(outputs) == ["applied", "stale"]
        assert store.get(0)["rev"] == 1 and store.get(0)["note"] == "reviewed"
        assert not store.cas(0, 0, "lost update")
        print(mode, "race applied/stale; rev=1")
    finally:
        for client in clients:
            if client.poll() is None:
                client.kill()
            client.communicate(timeout=5)


def main() -> None:
    assert sys.version_info[:3] == (3, 14, 4)
    assert sqlite3.sqlite_version == "3.50.4"
    rows = [
        Record(
            id=i,
            text=f"group{i % 5} decision {i} " + "context " * 20,
            parent=i - 1 if i else None,
            rev=0,
            note="",
            sha="",
        )
        for i in range(1000)
    ]
    atomic_json(SOURCE, rows)
    source_bytes = SOURCE.stat().st_size
    source_sha = digest()
    file = FileStore()
    sql = SqlStore()
    try:
        for store in (file, sql):
            store.rebuild()
        assert file.all() == sql.all()
        reference = queries(file)
        assert queries(sql) == reference
        samples = []
        for repeat in range(1, 4):
            order: list[tuple[str, Store]] = [("json", file), ("sqlite", sql)]
            if repeat % 2 == 0:
                order.reverse()
            for mode, store in order:
                assert fresh(store, 0)["sha"] == digest()
                start = perf_counter()
                result = queries(store)
                elapsed = perf_counter() - start
                assert result == reference
                samples.append({"mode": mode, "repeat": repeat, "seconds": elapsed})
        for mode, store in order:
            race(mode, store)
        rows[0]["text"] = "updated source"
        atomic_json(SOURCE, rows)
        for store in (file, sql):
            try:
                fresh(store, 0)
                raise AssertionError("accepted stale source")
            except ValueError as error:
                assert "source changed" in str(error)
            store.rebuild()
            updated = fresh(store, 0)
            assert updated["text"] == "updated source" and updated["rev"] == 2
            assert updated["note"] == "reviewed"
            assert 0 not in store.search("group0 ")
            assert store.children(0) == [1]
        assert file.all() == sql.all()
        atomic_json(Path("export.json"), file.all())
        with closing(sqlite3.connect("backup.sqlite", autocommit=True)) as backup:
            sql.db.backup(backup)
        restored = SqlStore("backup.sqlite")
        try:
            assert restored.all() == file.all()
            file.path.unlink()
            Path("export.json").replace(file.path)
            assert fresh(file, 0) == fresh(restored, 0)
        finally:
            restored.close()
        report = {
            "checked_at": datetime.now(UTC).isoformat(),
            "python": platform.python_version(),
            "sqlite": sqlite3.sqlite_version,
            "platform": platform.platform(),
            "source_bytes": source_bytes,
            "source_sha": source_sha,
            "rows": 1000,
            "scope": "100ID/5substring/10parent; warm, connections/hash before timer",
            "samples": samples,
        }
        Path("memory-results.json").write_text(json.dumps(report, indent=2) + "\n")
        print("MEMORY_RESULT", json.dumps(report))
        print("same queries; stale source rejected; rebuild notes kept; restore equal")
    finally:
        file.close()
        sql.close()


if __name__ == "__main__":
    main()

import sys

from stores import FileStore, SqlStore, Store

store: Store = FileStore() if sys.argv[1] == "json" else SqlStore()
try:
    revision = store.get(0)["rev"]
    print(f"ready rev={revision}", flush=True)
    assert input() == "go"
    print("applied" if store.cas(0, revision, "reviewed") else "stale", flush=True)
finally:
    store.close()

python3.14 exercise.py

Thư mục lab thuộc bạn; xóa cả thư mục sau khi đóng process. Verifier tự tạo thư mục tạm và xóa sau chạy; script đóng connections và child process trong finally.

Đọc kết quả đúng boundary

Lần đo 2026-10-03 12:36:42 UTC, bundle nguồn 252671 byte/1000 record; giây cho toàn bộ batch query, không latency từng request:

SQLite thấp hơn ở ba lượt của thiết kế này. Chưa đo memory footprint, startup, concurrent write throughput hoặc giá server; không tính tỷ lệ phổ quát cho agent memory.

Mỗi lượt gồm 100 lookup ID, 5 substring search và 10 parent query, cùng kết quả. Connection/schema/source hash trước timer; JSON đọc/parse toàn file mỗi query, SQLite giữ connection và query theo index ID/parent; substring cả hai đều scan. Warm-up và cache OS không eviction, ba lượt xoay thứ tự; không có cold/startup/write throughput trong bảng. File-per-ID, cache JSON trong RAM hoặc FTS là thiết kế khác cần đo lại, không bị benchmark này loại.

Ma trận quyết định

Lab stale source dùng SHA toàn bundle: đổi một record làm cả snapshot cần refresh. CAS bảo vệ note khỏi lost update, không chứng minh nguồn đáng tin. Editor ngoài lock có thể phá file protocol; rebuild dừng writer trong lab, chưa xử lý race nguồn đổi giữa hash/read hoặc xóa record đang được tham chiếu. Power-loss/crash injection, multi-host, authz/tenant, vector relevance và chi phí vận hành lớn đều chưa kiểm.

Đo workload của bạn trước khi thêm server. Files có thể giữ tài liệu cần review, database giữ state/query cần transaction; nếu dùng cả hai, chỉ rõ bản nào là nguồn cho từng field và index nào có thể rebuild. Học tiếp: context có nguồn, idempotent batch, đọc benchmark.