Python SDK

Python - 全部方法，三大类。

写入、读取、删除。以下每个示例都于 2026-08-01 针对线上 API 实际运行；响应为原样展示。

pip install wontopos
from wontopos import Client

mem = Client(api_key="wos-live-...")  # or read from an env var

选择模型

API 密钥决定用哪份记忆（您的账户）；模型决定用哪个引擎来读取。所有模型共享同一份记忆，因此可以用一个模型存储、用另一个召回。在客户端上设置默认值；给单个调用传 model= 即可覆盖。

mem = Client(api_key="wos-live-...", model="tablet-1")  # default engine
mem.recall("...", user_id="alice")                  # tablet-1
mem.recall("...", user_id="alice", model="scroll-1")  # or pick a model per call

list_models

模型目录 - 可传给 model 的 id 以及各自是否已上线。memory: "shared" 的模型读取同一存储；"isolated" 则各自独立。无需 API 密钥。

mem.list_models()
真实响应
[{"id": "tablet-1", "name": "Tablet 1", "available": true, "memory": "shared"},
 {"id": "tablet-2", "name": "Tablet 2", "available": true, "memory": "shared"},
 {"id": "scroll-1", "name": "Scroll 1", "available": true, "memory": "shared"},
 {"id": "scroll-1.2", "name": "Scroll 1.2", "available": true, "memory": "shared"}]

ping

用一行确认连接以及 API 密钥是否有效。

mem.ping()   # True, or raises AuthenticationError / PaymentRequiredError

上面的目录始终反映当前可用的模型 - 传入其他任何 id 都会得到明确的错误。新模型发布后会自动出现在其中。

写入

add

存储一条记忆。写入时不调用 LLM，您只需支付写入费用。

mem.add("she prefers tea over coffee", user_id="alice")
mem.add("I promised the summary by Friday", user_id="alice", speaker="me")  # its own words - no registration needed
真实响应
{"id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "stored (1 chunks)"}

add_turn

把一轮对话（用户 + 助手）同时写入短期与长期记忆。

mem.add_turn("hi", "hello!", user_id="alice")
真实响应
{"status": "ok"}

speaker

每条记忆都可以记录说话者。先注册一个人，之后把名字作为 speaker 传入；"me"（助手自己的话）无需注册。搜索也接受 speaker，可以只取某个人说过的话。

mem.add_speaker("Bob", user_id="alice")  # once per person; "me" needs no registration
mem.add("I promised to send the report on Friday", user_id="alice", speaker="me")
mem.add("Bob said the deadline moved to Tuesday", user_id="alice", speaker="Bob")
mem.search("what did Bob say about deadlines?", user_id="alice", speaker="Bob")
response
[{"content": "Bob said the deadline moved to Tuesday", "speaker": "Bob", ...}]
说话者和存储库一样是显式的。先注册，再以其名字保存。拼写错误绝不会悄悄变成一个新人。每个存储库起步可注册 50 人（会逐步提高），"me" 永远无需注册也不计数。

add_bulk

一次性回填一大段文本。在服务端切分并建立索引 - 适合导入既有历史。

mem.add_bulk("Alice moved to Brooklyn in March. She works at a design studio downtown.", user_id="alice")
真实响应
{"elapsed_secs": 0.154154944, "status": "ok", "stored": 1, "total_chunks": 1}

update

某个事实变了。旧记忆被标记为已取代（保留供追溯）；新记忆在召回中取而代之。

mem.update("576700aa-...", "she switched to coffee this year", user_id="alice")
真实响应
{"new_memory_id": "07e94433-b7cc-4e49-8d8f-f37fc1a392b7",
 "old_memory_id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "superseded"}

读取

search

语义搜索，最相关的排在最前。无论用哪种语言提问，都能找到用任何语言写下的记忆。 SDK 直接返回 memories 数组；下方展示的是原始 HTTP 响应体。部分模型会返回不止一组结果，SDK 会将它们合并返回，因此数组可能多于 max_results。请以实际收到的数组、而非请求的数量来规划提示词长度。

r = mem.search("what does she drink?", user_id="alice", limit=1)
真实响应（HTTP 响应体）
[{
   "id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624",
   "content": "she prefers tea over coffee",
   "category": "general",
   "time_bucket": "2026-06",
   "importance": 0.3,
   "similarity": 0.6316057443618774,
   "is_superseded": false,
   "superseded_by": null,
   "created_at": "2026-07-10T04:20:39.688276876Z"
 }]
字段含义
similarity这条记忆与您的查询有多接近 (0–1)。
is_superseded若该事实已被 update() 取代则为 true。
search_ms服务端检索耗时。

recall

一次往返返回您的 LLM 所需的一切 - 结果可直接粘贴进提示词：无论已存储多少内容，上下文大小固定且有界。

ctx = mem.recall("what does she drink?", user_id="alice")
真实响应（结构 - 列表已截短）
{"short_term":  {"count": 2, "turns": [{"role": "user", "content": "hi", ...}]},
 "long_term":   {"count": 4, "memories": [{"content": "she prefers tea over coffee",
                                           "similarity": 0.63, ...}]},
 "context":     {"count": 4, "around_top_memory": [
                  "[match] she prefers tea over coffee",
                  "[after] Alice moved to Brooklyn in March. ..."]},
 "instruction": "Use short_term for recent context, long_term for relevant
                 past memories, context for surrounding conversation of the
                 most relevant memory."}

history

最近的对话轮次（短期记忆），最早的在前。

turns = mem.history("alice")
真实响应（HTTP 响应体）
{"count": 2, "turns": [
   {"role": "user",      "content": "hi",     "timestamp": "2026-07-10T04:20:40.989011337Z"},
   {"role": "assistant", "content": "hello!", "timestamp": "2026-07-10T04:20:40.989013416Z"}
 ], "user_id": "alice"}

stats

单个用户的记忆条数统计。

mem.stats("alice")
真实响应
{"short_term_turns": 2, "total_memories": 4, "user_id": "alice"}

get

按 id 获取单条记忆 - 即 add 或 list_memories 返回的 id。只返回存储的原文和元数据。其他存储空间的 id，或已删除/失效的记忆，返回 404。

m = mem.get("alice", memory_id="576700aa-...")
response
{"id": "576700aa-...", "content": "she prefers tea over coffee",
 "category": "general", "created_at": "2026-07-10T04:20:39Z", "event_date": null,
 "is_superseded": false, "superseded_by": null}
真实响应
{"memory": {"id": "8bd090de-...", "content": "the office moved to the seventh floor in June",
  "category": "general", "created_at": "2026-07-31T18:20:30.531518060+00:00", "event_date": null,
  "is_superseded": false, "superseded_by": null}, "user_id": "docs_livetest"}

list_memories

列出存储中的记忆——只返回你保存的原文与元数据。按游标翻页：把返回的 next_cursor 传回以获取下一页。

page = mem.list_memories("alice", limit=100)
response
{"count": 2, "next_cursor": null, "memories": [
   {"id": "576700aa-...", "content": "she prefers tea over coffee",
    "category": "general", "created_at": "2026-07-10T04:20:39Z", "event_date": null,
 "is_superseded": false, "superseded_by": null}
 ]}

iter_memories · export_memories

无需管理游标即可遍历全部记忆，或一次性取回整个存储。

for m in mem.iter_memories("alice"):   # every page, no cursor bookkeeping
    print(m["id"], m["content"])
everything = mem.export_memories("alice")   # the whole store as a list

删除

delete

按 id 删除单条记忆。

mem.delete("alice", memory_id="576700aa-...")
真实响应
{"memory_id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "deleted"}

delete_all

抹除单个用户的全部数据 - 一次调用，符合 GDPR。

mem.delete_all("alice")
真实响应
{"memories_deleted": 4, "status": "deleted", "user_id": "alice"}

错误与可靠性

每个失败都是带类型的错误——可按具体情况(限流、认证、付费)分别捕获，或用基类 WosError 统一捕获。

from wontopos import PaymentRequiredError, NotFoundError
try:
    mem.add("...", user_id="alice")
except NotFoundError:
    mem.create_store("alice")   # store didn't exist yet
except PaymentRequiredError:
    top_up()                       # out of credit - don't retry

rate_limit

在任意调用后读取剩余额度，在触达上限前主动放慢。

mem.search("...", user_id="alice")
rl = mem.rate_limit   # {"limit": 150, "remaining": 3, "reset": ...}

search_self

在自我记忆模型（Scroll 1.2+）上一次调用返回两条通道：别人说的话与智能体自己说的话分开返回，读取方不会弄混说话人。

r = mem.search_self("what did I promise?", user_id="alice")
r["memories"]       # what others said / general memories
r["self_memories"]  # the agent's OWN words (speaker "me")

list_engrams

向服务询问当前模型可运行的 engram 与投递形式，而不是硬编码名称 —— 一旦有新 engram 上线，硬编码的代码就再也看不到它。

cat = mem.list_engrams()
[e["name"] for e in cat["engrams"]]   # ask, never hard-code

filters

把搜索收窄到存储的一部分。在排序之前应用，因此得到的是过滤范围内最相关的结果 - 而不是对 top-N 再做过滤。

mem.search("what did we decide", user_id="alice", filters={
    "categories": ["work"],
    "event_from": "2026-01-01",   # when it HAPPENED
})
键：categories · event_from / event_to（内容发生的时间 - metadata.event_date）· time_from / time_to（写入的时间）· min_importance。未列出的键会被丢弃而不是报错，所以拼错会悄悄扩大搜索范围。

idempotency_key

让同一次写入可以安全重复。当重试来自你这边时使用 - 中断后重跑的任务、会重投的队列。

mem.add("she prefers tea", "alice", idempotency_key=f"import:{row.id}")
密钥要从被存储的对象派生（import:row-42），不要用常量：两次不同的写入复用同一密钥会重放第一次的响应，第二次会被悄悄丢弃。格式：1-128 个 [A-Za-z0-9._:-] 字符。

with_timeout / with_retries / with_deadline

不改动已经建好的客户端，只调整单个调用点：大批量回填用超时更长的克隆，自己写重试循环时用关闭重试的克隆。

timeout 限定的是单次尝试，因此会重试的调用可能比它活得更久 — 默认情况下一次调用可以占用连接 30 秒、退避、再试、再试。deadline 限定的是整次调用：每次尝试都被压缩到剩余的时间内，退避也绝不会睡过预算。当调用方有真实上限时设置它 — 比如只有五秒的请求处理器。

mem.with_timeout(120).add_bulk(big_blob, "alice")  # this slow call only
mem.with_retries(0).add("...", "alice")              # you retry, not the SDK
mem.with_deadline(5).recall("...", "alice")             # 5s for the whole call