Attestari

Attestari

面向 AI Agent 的可审计记忆层。 为你的 Agent 提供长期记忆—— 和任何记忆层一样，只不过每条事实都附带凭证（来源和时间），运行在纯 Postgres 上，审计轨迹具有防篡改能力，而且任何用户的数据都可以得到可验证的删除， 并附有签名证书。

from attestari import Memory

mem = Memory()
mem.add("Hi, I'm Alice. I live in Toronto and I work at Acme.",
        subject_id="alice", valid_from="2021-06-01")
mem.add("Update: I moved to Berlin and I now work at Globex.",
        subject_id="alice", valid_from="2026-01-01")          # supersedes Acme + Toronto

mem.answer("where does the user work", subject_id="alice")    # -> "Globex"  (latest)
mem.answer("where did the user live",  subject_id="alice",
           as_of="2022-01-01")                                # -> "Toronto" (time-travel)

cert = mem.forget("alice")   # right-to-be-forgotten -> a signed deletion certificate

运行这些内容无需数据库、API 密钥或模型下载——核心引擎零依赖。

---

目录

• 为何与众不同
• 你将获得什么
• 快速开始（30 秒）
• Python API
• 实际运行 — Postgres · 服务 + 控制台 · MCP · TypeScript
• REST API
• 配置
• 在一个演示中展示可验证删除 + 防篡改审计
• 已经在使用 Mem0 或 Zep？为它套上封装
• 面向审计人员和 DPO
• 工作原理 — 以及 Atlas，用一页插图呈现的引擎
• 项目结构
• 文档与贡献

为何与众不同

托管式记忆是一个黑箱：你无法查看某条“记忆”的来源，无法彻底删除某位用户的数据，也无法证明历史记录未被篡改。 对于银行、医院或保险公司而言——以及在 GDPR / EU AI Act 的要求下——这完全不可接受。Attestari 是一个中立、可自托管的层，专门解决这些问题。

	Attestari	典型记忆层
运行在纯 Postgres 上（无需图数据库）	✅	✗（需要 Neo4j / 向量服务）
每条事实都带有溯源信息	✅	部分支持
双时态（“日期 D 当天什么是真实的？”）	✅	✗
可验证删除 + 证书（GDPR）	✅	✗
防篡改审计轨迹（哈希链）	✅	✗
兼容不同模型供应商	✅	通常被锁定于单一供应商

最后两项就是护城河。**可证明的删除：**每位用户的数据都用其专属密钥加密；forget() 会销毁该密钥，使内容无法恢复——同时保留不可变日志和签名证书作为证明。 **可验证的历史记录：**每个事件都通过哈希链接，因此 verify_audit() 能检测任何编辑、插入或删除——即使删除后，证明仍然有效。

你将获得

• 每条事实均可追溯来源 — 每条记忆都能追溯到确切的源消息， 并附有字符跨度、置信度和时间戳。
• 双时态时间旅行 — 可查询任意过去时刻 as_of 的记忆；更正会取代旧事实，但不会将其删除，因此始终可以还原历史。 as_of 会筛选有效时间（事实为真时的时间）。系统时间（Attestari 记录事实的时间）会存储在每条事实中，但目前还不能据此查询。
• 可证明的删除 — forget(subject_id) 会对用户数据进行加密销毁，并返回签名的 DeletionCertificate；内容备份和副本也包含在内（密钥存储需要单独的备份策略 — 参见 威胁模型）。
• 防篡改审计 — 哈希链接事件链；verify_audit() 可检测任何编辑、插入或删除，且证明在加密销毁后仍然有效。
• 冲突解决 — 区分单值与多值谓词；冲突会通过 conflicts() 显示，而不会被悄悄丢弃。
• 实体解析 — 可逆地合并“同一实体的多种表面形式”。 合并会被记录，但读取时目前不会跟随合并关系：search() 和 timeline() 仍会将每种表面形式分别处理。
• 混合检索 — 语义检索（pgvector）⊕ 关键词检索（全文），内置双时态筛选。
• 随处运行 — 零依赖内存引擎，或在一个 Postgres
  • pgvector 容器中持久化运行。无需图数据库。兼容不同模型厂商。

快速入门（30 秒）

git clone https://github.com/attestari/attestari && cd attestari
python examples/spike.py              # zero-dep end-to-end loop
python examples/agent_with_memory.py  # the "give an agent memory" pattern

你会看到事实随时间变化、双时态查询在不同日期的结果有所不同、可追溯至源内容的来源记录，以及会签发证书的 forget()。无需安装、无需 API 密钥、无需数据库。

Python API

一个门面 Memory，涵盖所有功能：

from attestari import Memory

mem = Memory()                       # zero-dep, in-memory (tests, demos)
# mem = Memory.local()               # durable in one local SQLite file — zero infrastructure
# mem = Memory.postgres()            # durable on Postgres + pgvector (production service)

# --- write -------------------------------------------------------------
fact_ids = mem.add(
    "I moved to Berlin and I now work at Globex.",
    subject_id="alice",              # whose memory this is
    valid_from="2026-01-01",         # when it became true (defaults to now)
    source_ref="chat:msg-42",        # where it came from (for provenance)
)

# --- read --------------------------------------------------------------
mem.search("where does the user work", subject_id="alice")          # ranked SearchResults
mem.answer("where does the user work", subject_id="alice")          # -> "Globex"
mem.answer("where did the user live",  subject_id="alice",
           as_of="2022-01-01")                                      # time travel: recall as-of a past date
mem.timeline(subject_id="alice")                                    # full bi-temporal history
mem.get_provenance(fact_ids[0])                                     # source episode + span
mem.conflicts(subject_id="alice")                                   # surfaced conflicts

# --- govern ------------------------------------------------------------
report = mem.verify_audit()          # AuditReport — is the hash chain intact?
cert   = mem.forget("alice")         # DeletionCertificate — provable erasure

方法	返回值	功能
add(text, *, subject_id, valid_from=…, source_ref=…)	list[str]	摄取消息；提取、去重并取代事实。
search(query, *, subject_id, as_of=…, limit=5)	list[SearchResult]	混合检索，可选时间筛选。
answer(query, **kwargs)	str | None	查询对应的首要对象。
timeline(*, subject_id)	list[Edge]	某个主体的所有事实，包括当前有效和已取代的事实。
get_provenance(fact_id)	Provenance | None	追溯事实的源事件和字符跨度。
conflicts(*, subject_id=None)	list[dict]	根据谓词基数解决冲突。
resolve_entities(names=None, *, auto=True)	ResolutionResult	合并重复实体（可逆）。
forget(subject_id)	DeletionCertificate	对主体进行加密销毁；返回证明。
verify_audit(deep=False)	AuditReport	验证防篡改哈希链；deep=True 还可检测对事件内容的静默修改。

实际运行

三种存储层级，一个引擎 — 所有保障（审计链、加密销毁、 深度验证、时间旅行）在这三种层级上均有效：

层级	存储	适用场景	设置
Memory()	内存中	测试、演示、确定性	无
Memory.local()	一个 SQLite 文件（~/.attestari/attestari.db）	个人 Agent、MCP、原型 — 持久化、单进程	无（标准库）
Memory.postgres()	Postgres + pgvector	生产环境：并发访问、带索引的混合搜索	一个容器

零基础设施的持久化运行（重启后仍保留；无需安装或运行任何内容）：

from attestari import Memory
mem = Memory.local()      # or Memory.local("path/to/agent.db")

在 Postgres + pgvector 上持久化运行（一个容器，无需图数据库）：

ATTESTARI_PG_PORT=5433 docker compose up -d       # applies the schema on first boot
pip install -e ".[postgres,embeddings]"
export ATTESTARI_DATABASE_URL=postgresql://attestari:attestari@localhost:5433/attestari

已有 Postgres（托管或本地）？Schema 已包含在 pip 包中 — 无需克隆：

python -m attestari.initdb postgresql://user:pass@host:5432/db   # idempotent

from attestari import Memory
mem = Memory.postgres()   # durable; materialized projections + pgvector + full-text search

作为 REST API + 可视化控制台：

pip install -e ".[server]"
uvicorn attestari.server:app   # API at /v1/*, the memory-graph console at /

作为 MCP 服务（任何 Agent——Claude、框架等——都可以使用）——通过 stdio 暴露 add_memory / search_memory / get_provenance，如果选择启用（见下文），还会暴露 forget_subject。 在 MCP 客户端的配置中注册它（例如 Claude Desktop 的 claude_desktop_config.json）；客户端会为你启动进程：

{
  "mcpServers": {
    "attestari": {
      "command": "attestari-mcp",
      "args": [],
      "env": {
        "ATTESTARI_SQLITE_PATH": "~/.attestari/attestari.db",
        "ANTHROPIC_API_KEY": "sk-ant-...",
        "ATTESTARI_KEK": "base64-kek-here"
      }
    }
  }
}

只需设置 command/args。默认情况下数据会持久保存（记忆写入本地 SQLite 文件，因此应用重启后仍会保留）；env 块用于配置此服务的选项——添加 ATTESTARI_DATABASE_URL，即可改用 Postgres 而非 SQLite； 添加 ANTHROPIC_API_KEY，即可将提取升级为 Claude；添加 ATTESTARI_KEK，即可启用 加密粉碎。若要独立运行（例如用于调试）：attestari-mcp（或 python -m attestari.mcp）。如果没有本地安装，MCP 客户端可以直接从 PyPI 启动它：uvx --from "attestari[server]" attestari-mcp。

通过 MCP 擦除数据需选择启用。 Agent 会受到所读文本的影响，而 擦除操作无法撤销，因此只有在 env 块设置 "ATTESTARI_MCP_ALLOW_FORGET": "1" 时， 才会提供 forget_subject。启用后，该工具会标记为破坏性操作（支持工具注释的客户端会在运行前请求确认），并且需要两次调用：第一次预览将要擦除的内容并返回 manifest_hash； 第二次将该哈希作为 confirm_manifest_hash 传回，只有当该主体的记录仍与预览一致时才会执行擦除。除非 Agent 需要擦除人员数据，否则请勿启用；Memory.forget() 和 POST /v1/forget 不受影响。

从 TypeScript 使用——TS 客户端通过 REST API 通信，因此请**先启动服务 **（见上文；默认地址为 http://localhost:8000）。然后参阅 clients/ts（@attestari/client），这是一个精简的类型化客户端，映射了 Memory 的接口。

**与 LangChain 搭配使用：**参阅 clients/langchain（attestari-langchain） ——其中包含 AttestariRetriever（通过来源信息回忆事实）和 AttestariChatMessageHistory （可直接用于 RunnableWithMessageHistory 的记忆），适用于任意链或 Agent。

使用真实的 Claude 提取（而非零依赖的确定性提取器）：

pip install -e ".[anthropic]"
export ANTHROPIC_API_KEY=sk-ant-...
python examples/spike.py --llm anthropic

启用加密粉碎删除（将 forget() 从逻辑删除变为 加密擦除）。加密需选择启用，并使用根密钥加密密钥 （KEK）；如果未设置，forget() 仍可使用，但只会让数据不再出现在读取结果中。 生成 KEK 并将其设置到环境变量中：

pip install -e ".[crypto]"
export ATTESTARI_KEK=$(python -c "from attestari.crypto import generate_kek; print(generate_kek())")

现在，每个主体的 PII 都会使用专属密钥进行静态加密，而 forget() 会销毁该密钥——密文将无法恢复，但审计证明仍会保留。设置 KEK 后， DeletionCertificate 也会经过签名 （使用从 KEK 派生的密钥进行 HMAC-SHA256 签名）；持有 KEK 的任何人都可以离线验证证书—— verify_certificate(cert, kek)——证书的任何字段一旦被修改，验证都会失败。未设置 KEK 时，forget() 执行的是逻辑删除，生成的证书也不会签名。请勿将 KEK 与数据库及其备份放在一起（应使用环境变量 或 KMS）——将其与数据存储在一起会使粉碎失去意义。请参阅 docs/the-moat.md 中的备份边界。 forget() 的覆盖范围及其未覆盖的内容：它会擦除在该 subject_id 下记录的事件，以及从这些事件中提取的事实。从他人消息中提取、与此人相关的事实，属于另一主体的范围，因此会保留。加密仅覆盖消息文本和事实值；擦除后，每个事件的其他字段仍可读取——subject_id、事实的主语和谓语（使用 Claude 提取时，主语可以是人名）、范围 ID、时间戳、字符跨度（可泄露值的长度）以及 source_ref。因此，请使用不透明的假名 subject_id，并避免在 source_ref 中存放个人数据。已擦除的 subject_id 会保持封闭：对其调用 add() 会引发 ForgottenSubjectError，且不会存储任何内容（POST /v1/add 会返回 409，MCP 工具也会拒绝），因此，若某人再次回来，请用新的 ID 记录。Postgres 会保留证书登记表；Memory() 和 Memory.local() 会在 forget() 时将证书返回一次，因此请自行保存。

正式部署。 生产环境检查清单：

• 存储： 使用 Memory.postgres()（支持并发访问）；通过 python -m attestari.initdb "$ATTESTARI_DATABASE_URL" 应用架构，并在每次升级后重新运行（该操作具有幂等性；在运行之前，写入会停止并提示相应操作，读取则继续正常工作）。Memory.local()（SQLite）是单进程的——适合单个 Agent 或 MCP 服务，但不适用于共享服务。
• 服务端： 在进程管理器下运行，例如通过反向代理运行 uvicorn attestari.server:app --host 0.0.0.0 --port 8000 --workers 4； 在前面加上你自己的身份验证（此 API 未内置身份验证）。
• 单个主体的写入： 在 0.0.7 及更早版本中，请在应用中串行处理。针对同一主体同时调用两次 add()，可能都会覆盖同一个旧值，导致单值谓词存在两个有效值。后续版本会在存储的写锁下决定覆盖关系，因此并发写入是安全的。
• 擦除后的写入： 在 0.0.7 及更早版本中，请在调用 add() 前检查 is_forgotten(subject_id)。如果已设置 KEK，向已遗忘的主体写入会导致之后对所有主体读取日志时都失败，并报错 InvalidTag。要恢复，请从 keyring 表中删除该主体对应的行。后续版本会直接拒绝此类写入。
• 提取与嵌入： 设置 ANTHROPIC_API_KEY（提取会自动升级为使用 Claude），并安装 [embeddings] 以获取真正的语义向量。
• 密钥： 从 KMS／密钥管理器将 ATTESTARI_KEK 注入为环境变量——切勿将其写入镜像或数据库。按照单独的短期保留策略备份 keyring 表（或轮换 KEK），以免恢复的数据备份复原已销毁的主体——请参阅 docs/the-moat.md。
• 备份： 也要排除派生的投影表（edge、entity）——它们存有用于检索的明文事实文本，并且可以从（密文）事件日志完全重建，因此备份它们只会削弱销毁效果。不包含这些表的备份恢复后，首次连接时会重建它们。
• 密钥： 不会自动读取 .env 文件——启动进程前，请导出这些变量（或使用编排器注入密钥）。

REST API

uvicorn attestari.server:app 会提供：

方法与路径	用途
GET /healthz	存活检查。
POST /v1/add	导入一条消息。
GET /v1/search	混合检索（q、subject_id、as_of、limit）。
GET /v1/timeline	主体的完整双时态历史。
GET /v1/provenance/{fact_id}	追溯事实的来源。
POST /v1/forget/{subject_id}	可证明的删除 → 证书。
GET /v1/conflicts	已发现的冲突。
GET /v1/audit/verify	验证审计哈希链。
GET /v1/graph	记忆图（供控制台使用）。
GET /	可视化图控制台。

配置

环境变量（均为可选——引擎无需设置任何变量即可运行）：

变量	作用
ATTESTARI_DATABASE_URL	Postgres DSN；服务/MCP 使用 Postgres，而非本地 SQLite。
ATTESTARI_SQLITE_PATH	Memory.local() 支持的服务/MCP 存放 SQLite 文件的位置（默认值为 ~/.attestari/attestari.db）。
ATTESTARI_KEK	根密钥加密密钥；启用加密销毁删除。
ATTESTARI_MCP_ALLOW_FORGET	1 会在 MCP 服务中提供 forget_subject 工具（默认关闭；启用后需调用两次）。
ATTESTARI_PG_PORT	随附的 docker compose Postgres 的主机端口（默认值为 5432）。
ATTESTARI_WRAP_UPSTREAM	要治理的内存服务基础 URL；挂载 /v1/wrap/* 端点（未设置时不提供 wrap 路由）。
ATTESTARI_WRAP_UPSTREAM_TOKEN	作为 Authorization: Bearer … 发送给上游。
ATTESTARI_WRAP_*_PATH	覆盖上游路径——ADD、SEARCH、DELETE、GET_ALL（默认值为 /add、/search、/delete、/get_all）。将 GET_ALL 设为空值可禁用删除后的回读。
ANTHROPIC_API_KEY	启用 Claude 事实提取——服务/MCP 会自动从正则提取器升级。
ATTESTARI_EXTRACTOR_MODEL	覆盖提取模型（默认值为 claude-opus-4-8）。

安装额外依赖（pip install -e ".[extra]"）：

额外依赖	添加内容
postgres	psycopg + pgvector——持久化事件存储。
embeddings	sentence-transformers——真实的语义嵌入。
crypto	cryptography——加密销毁删除。
server	FastAPI + uvicorn + MCP——REST 服务和 MCP 服务。
anthropic	Anthropic SDK——Claude 事实提取。
dev	pytest + ruff——测试和代码检查。

一场演示：可证明的删除 + 防篡改审计

别只相信要点——破坏这些属性，看看它们如何被检测出来。 此演示无需数据库、API 密钥或模型下载即可运行，并会自行验证 （每项声明最后都有一个 assert；如果任何属性不成立，程序就会崩溃）：

python examples/prove_the_moat.py   # tamper -> caught · crypto-shred -> unrecoverable · time-travel

它通过对抗性测试证明了全部三个差异化优势：被悄然改写的事实会被 verify_audit(deep=True) 在准确的 seq 处捕获；经过加密销毁的 主体，其密文会被证明无法恢复，而审计证明依然有效； 更正后的事实可以按过去的时间点查询，而不必抹去历史。（可在 裸克隆仓库上运行；pip install "attestari[crypto]" 会将声明 2 从逻辑删除升级为 加密销毁。）威胁模型和明确说明的边界，请参阅 docs/the-moat.md。

如需针对真实加密的 Postgres 行执行完整的加密销毁：

ATTESTARI_DATABASE_URL=postgresql://attestari:attestari@localhost:5433/attestari \
    python examples/audit_and_forget_demo.py   # audit -> trace -> forget -> PROVE

它会展示：追溯事实的来源、遗忘一个主体、确认原始行变为 无法读取的密文、检索不到任何内容——并且删除后审计链 仍然有效。

已在使用 Mem0 或 Zep？为其添加封装

无需替换现有记忆层，也能获得审计追踪。wrap() 会在你已使用的客户端前接入 Attestari：

from attestari.wrap import wrap

governed = wrap(mem0_client)                              # or zep, or your own

governed.add("I live in Berlin.", subject_id="u1")        # recorded, then stored
governed.search("where do I live", subject_id="u1")       # straight through
receipt = governed.forget("u1")                           # both stores + proof

receipt.complete             # True only if BOTH halves succeeded
receipt.certificate          # Attestari's signed deletion certificate
receipt.downstream_verified  # True = we read the store back and it was empty
receipt.downstream_error     # what the wrapped store said, if it refused
governed.verify_audit(deep=True)                          # the chain still holds

不是用 Python？ 将应用指向 Attestari 服务，而不是记忆服务，即可通过 REST 获得相同的保障：

ATTESTARI_WRAP_UPSTREAM=https://memory.internal uvicorn attestari.server:app

POST /v1/wrap/add          {"text": …, "subject_id": …}   # recorded, then forwarded
POST /v1/wrap/search       {"query": …, "subject_id": …}  # passthrough
POST /v1/wrap/forget/{id}                                 # both stores + evidence

部分擦除会返回 409，而不是 200——只检查状态码的调用方绝不能将未完成的删除 误判为成功。只有配置了上游服务时，这些路由才会出现。请参阅 attestari.wrap_http， 了解上游服务预期使用的简易 JSON 契约（仅使用标准库，无需新增依赖）。

它不会只听信删除调用的说法。 删除后，forget() 会从封装的存储中重新读取该主体。 如果存储返回 {"deleted": true}，却仍保留数据行，就会当场被发现—— downstream_verified 为 False，complete 为 False，这一差异也会写入审计链。 如果适配器没有回读操作，downstream_verified 将为 None，而不是 True：“无人核查” 和“核查后确认无误”是两种不同的声明，只有后者才是证据。

写入会先记录到防篡改链中，再传递给下游；读取则原样通过（检索正是你保留原有存储的 原因）；forget() 会删除下游数据、对 Attestari 的副本进行密码学销毁，并记录下游存储 的实际操作结果，包括失败情况。可通过 Adapter 配置方法名称和主体关键字，因此也能 用于纯向量存储。

封装能证明什么，不能证明什么。 Attestari 无法对他人服务中的数据进行密码学销毁—— 它不持有对方的密钥。封装式部署能证明删除请求已发出、已调用下游删除操作及其返回结果、 Attestari 自身的副本已无法恢复，以及此后这些记录均未被篡改。这是跨两个系统的可审计 删除记录，而不是处处都已进行密码学销毁：封装存储能否彻底擦除，取决于其删除端点是否 诚实；封装的作用是将该端点的行为转化为证据，而不是承诺。要实现完整的密码学擦除，数据 必须存储在 Attestari 自身。

面向审计员和 DPO

需要为 AI 系统记忆负责的人，可以在 auditor/ 找到专属文档：一份 通俗易懂的单页说明（保证了什么、没有保证什么，以及如何自行核查）、 一份 EU AI Act 第 12 条映射，以及一份关于密码学擦除和 相关部署策略的 GDPR 第 17 条说明。

任何部署都可以生成一份带日期的快照，展示其自身可验证的状态：

attestari evidence --deep --out ./evidence   # EVIDENCE.md + evidence.json

该套件包含审计链校验结果和头部哈希、逐一根据当前账本重新核查的擦除登记簿，以及保留的 删除证书。生成套件时，所有声明都会根据实时账本重新推导——它之所以是证据，是因为 你可以在拥有读取权限、无需系统运营方配合的情况下重新生成它。

attestari verify --deep          # re-check the chain, re-hashing content
attestari verify --user u_123    # confirm one subject's erasure; non-zero exit if not

工作原理

事实来源是一个仅追加事件日志；所有可供查询的内容（知识图谱、向量索引、关键词索引） 都是可以从中重建的投影。因此，审计、时间回溯、溯源和可证明的删除都是设计本身 带来的能力，无需后期拼凑。

• Attestari Atlas — 用一页配有插图的页面展示整个引擎：每次写入、读取和擦除，可在浏览器中篡改的审计链、适用场景，以及容易出错之处。面向 0.0.7 编写。
• ARCHITECTURE.md — 从头到尾介绍设计。
• LEARN.md — 从基础开始为新手讲解相同的理念。

项目结构

src/attestari/         the engine — events, store, projection, retrieve, memory,
                    crypto (shred), audit (hash chain), predicates, resolver
src/attestari/server.py, console.py   FastAPI REST API + the graph console
src/attestari/mcp.py   the MCP server
src/attestari/cli.py, evidence.py     `attestari verify` + the evidence bundle
src/attestari/wrap.py, wrap_http.py   govern an existing memory layer (Mem0,
                    Zep, or any HTTP service) instead of replacing it
auditor/            the auditor pack — DPO one-pager, AI Act + GDPR mappings
examples/           runnable demos — start with spike.py
eval/               quality + retrieval-latency harness
clients/ts/         the TypeScript SDK (@attestari/client)
clients/langchain/  the LangChain integration (attestari-langchain)
src/attestari/db/schema.sql       the Postgres bi-temporal schema
docker-compose.yml  Postgres + pgvector

文档与贡献

• ARCHITECTURE.md — 设计
• auditor/ — 面向 DPO、合规与内部审计
• LEARN.md — 从零开始了解 Attestari 的工作原理
• CONTRIBUTING.md — 开发环境搭建、适合新手的议题
• SECURITY.md — 漏洞报告

状态

引擎及其差异化功能——可验证删除、防篡改审计、双时态溯源、Postgres 原生检索——均已构建并通过测试（122 项测试；Postgres p95 ≈ 1 ms）。

许可证

Apache-2.0。核心部分保持宽松许可；托管云服务和企业／治理功能属于商业层。