n1mem · N1Mem 记忆 API Python SDK

powered by T1Mem engine · BYOK（自带 Key）· 同步零硬依赖

给 Agent 一个长期记忆后端：写入即可召回，召回基于命中记忆接地； 并且能把你已有的 Agent 记忆（身份文件 / 技能库 / 关键文档）一次性迁进来。

安装

pip install n1mem            # 同步版，零硬依赖（仅标准库）
pip install n1mem[async]     # 需要异步客户端时（依赖 httpx）

10 行代码跑通

from n1mem import N1Mem

m = N1Mem(api_key="tk_xxx")        # 或设环境变量 N1MEM_API_KEY 后不传参

print(m.health())                  # 服务健康 + provider 可用性
m.ingest("我今天换了新工位，在 3 楼靠窗")
print(m.recall("我坐哪儿"))
print(m.list_memories(limit=5))    # 看看存进去了什么

迁移：把已有 Agent 记忆导进来

n1mem.importers 在你本机把记忆资产枚举成条目（不上传原始目录结构）， 再由 ingest_batch() 分批提交。全程离线可复现，不依赖服务端读你的磁盘。

from n1mem import N1Mem
from n1mem.importers import WorkBuddyAdapter

m = N1Mem(api_key="tk_xxx")
ad = WorkBuddyAdapter()                    # 默认导入身份层 + 技能层

plan = ad.plan(ad.enumerate())             # 先看要导什么，再决定导不导
r = m.ingest_batch(plan, source_id=ad.source_id)
print(r["inserted"], r["source_id"])

# 反悔：按批次整体撤销（被删内容会进 tombstone，防止同内容被后续导入"复活"）
m.import_rollback(source_id=r["source_id"])

已支持的适配器：

⚠️ 导入必须走 ingest_batch()，不要循环调用 ingest()。 ingest() 只发送 text，会丢掉 tier / mtype / source_id / blob： 结果是身份层从 resident 静默降级成普通事实（常驻身份失效且没有任何报错）， 并且因为没有批次 id 而无法回滚。

安全（2026-09-17 起口径变更）：导入时服务端会扫一遍疑似凭据（阿里云 AK / PyPI token / PEM 私钥 / 带密码的 DSN / 高熵串）与个人信息，命中条目照常入库并在回执里给出 detected_secrets（只给序号与类型，不回显值）。

N1Mem 是中间件，不做内容脱敏 —— 你决定存什么，我们尊重并严格执行。 服务端的责任是：系统安全、不主动泄密、主动保护隐私（租户隔离 / 最小权限 / 静态加密 / 访问留痕）。若你不想让某内容进库，请在入参里就不要发它。 blocked_secrets 为兼容字段，此后恒为空数组。

对话提炼：n1mem.distill（0.3.0 起随包提供）

把本机的会话记录提炼成记忆条目，而不是把原始对话整段灌进去 —— 单条对话最长可达数十万字符（实测 47 万），远超任何上下文窗口， 不分块直接发必然失败。

它的核心设计是把「算钱」和「花钱」分成两步，让费用在发生前就可见：

from n1mem import distill

st  = distill.scan_dialogues()                   # 本地扫描，不花钱
est = distill.estimate(st, price_in_per_m=1.0,   # 本地估算，仍不花钱
                       price_out_per_m=4.0)
print(est["turns"], est["est_cost_total_cny"])   # 先看清要花多少
# 确认之后，才逐块调用 distill_chunk(complete=…, chunk=…, model=…)

estimate() 的返回值刻意分成两组，别混着看：

• certain_fields（turns / chars / skipped_* / merged_user）—— 本地数出来的确定值；
• estimated_fields（token 数、金额）—— 按 assumptions（字符/token 比、块大小、单价） 换算的估算值；输出长度由模型决定，实际费用可能高于或低于此数。

超长条目跳过并计数（skipped_oversize），不会悄悄截断。

API

异步版 AsyncN1Mem 接口完全一致：

from n1mem import AsyncN1Mem

async with AsyncN1Mem(api_key="tk_xxx") as m:
    await m.ingest("…")
    await m.ingest_batch(items)

环境变量

显式传入的构造参数优先于环境变量。

主体声明：N1MEM_SUBJECT（0.3.0 新增）

Key 回答「哪个组织」，N1MEM_SUBJECT 回答「谁」。

同一个 Key（同一组织）下可能有多台设备 / 多个人在调用。声明主体后，服务端按 (org, subject) 反查成员表，据此判定该主体是否已被接入。

from n1mem import N1Mem

m = N1Mem(api_key="tk_xxx", subject="workbuddy-pc")   # 也可设环境变量 N1MEM_SUBJECT

SDK 会把主体作为 x-n1mem-agent 请求头发给服务端。

• 须与组织里已登记的主体名逐字一致。拼错不会报错 —— 服务端只会认为「该主体未登记」。
• 不设 ⇒ 不发这个头，行为与本能力引入前完全一致 ⇒ 随时可回退（去掉该键即可）。
• 🔴 别拿 Agent 的类型名当主体名（如 workbuddy）：那是「哪个 Agent」，不是「谁」。 一台机器同时跑多个 Agent 时，它们应当共享同一个主体；否则该共享的 private 记忆反而互相看不见。
• 想确认服务端认到什么：调 /v1/member/status —— 未声明主体时它明确回报 missing_subject，不含糊。

配套 MCP Server 有同样的开关：见 n1mem-mcp。

错误处理

上游 4xx/5xx 统一抛 N1MemError，带 status / message / endpoint：

from n1mem import N1MemError
try:
    m.ingest("x")
except N1MemError as e:
    print(e.status, e.endpoint, e.message)   # 401 /v1/ingest {"detail":"invalid api key"}

参数用错（如 forget_all() 未确认、import_rollback() 两个参数都没给）会抛 ValueError / TypeError —— 在本地就失败，不浪费一次注定被拒的网络请求。

当前能力边界（诚实清单）

• ingest() 持久化到 N1Mem 存储层并生成向量，返回 {stored, embedded, memory_id}；写入后即可被召回。
• recall() 走关键词 + 向量混合检索（hybrid）；answer 基于命中记忆接地生成，未命中会诚实说明未命中，不编造。
• 租户隔离由服务端按 API Key 反查 org 保证，不存在传参越权读取的路径。
• ask / update 尚未提供，SDK 会明确抛 NotImplementedError，而不是静默返回空。

历史勘误

与 t1mem_sdk 的区别

同目录下有两个包，不要混用：

相关：MCP Server

想让 Claude / Cursor / OpenClaw 以工具方式调用记忆（而不是写代码）， 装配套的 n1mem-mcp。