agentmemory Python SDK 实战：通过 iii-sdk 在 WebSocket 上调用 `mem::*` 记忆函数

agentmemory Python SDK 实战：通过 iii-sdk 在 WebSocket 上调用mem::*记忆函数

【免费下载链接】agentmemory#1 Persistent memory for AI coding agents based on real-world benchmarks项目地址: https://gitcode.com/GitHub_Trending/age/agentmemory

本篇指南讲解如何用官方 Python SDK（iii-sdk）直接对接 agentmemory 守护进程：从安装依赖、启动 daemon，到调用mem::remember保存记忆、mem::smart-search做混合检索、再用mem::observe摄入观察并按 token 预算渲染上下文。读完你将掌握一套「零 REST 客户端」的跨语言调用范式，并能在 Python 脚本里复现 agentmemory 的核心记忆闭环，代码可直接照抄运行。

为什么是mem::*函数：agentmemory 的核心操作即 iii 函数

agentmemory 把它的核心记忆操作全部注册为 iii（SDK 运行时）函数：mem::remember、mem::observe、mem::context、mem::smart-search、mem::forget。这意味着任何拥有 iii SDK 的语言，都可以直接在 WebSocket 传输层ws://localhost:49134上调用它们，无需再维护一套独立的 REST 客户端。

从 src/index.ts 的注册逻辑可以看到，守护进程启动时会把这些函数逐一挂载到 iii 运行时上，例如 remember.ts 中的sdk.registerFunction("mem::remember", ...)、observe.ts 中的sdk.registerFunction("mem::observe", ...)。也就是说，Python 脚本里iii.trigger({...})发出去的每个调用，最终都会精确落到这些 TypeScript 实现上。

在 iii 生态内部，直接调用mem::*函数延迟更低；只有当你需要从一台没有 iii 运行时的宿主机访问 daemon 时，才需要走 HTTP 包装层（api::*，见下文）。

环境准备：安装 SDK 并启动守护进程

第一步：安装官方 Python SDK

pip install iii-sdk

第二步：启动 agentmemory 守护进程

npx -y @agentmemory/agentmemory

daemon 的默认监听地址为：

• WebSocket（iii 引擎）：ws://localhost:49134—— Python SDK 从这里调用mem::*函数；
• REST：:3111—— 供api::*包装函数使用，也即各编辑器/Agent 集成默认的http://localhost:3111。

从 cli.ts 的注释可见，3111 是 REST 端口基准，streams 端口为 N+1、viewer 端口依此类推，而 49134 是引擎 WebSocket 的默认锚点；--port <N>可覆盖 REST 端口，--instance <N>则以3111 + N*100的方式运行多个实例，方便多实例隔离测试。

第三步：Python 侧建立连接

from iii import register_worker iii = register_worker("ws://localhost:49134") iii.connect()

register_worker会绑定到一个 worker 上，connect()建立 WebSocket 连接；此后所有调用都通过iii.trigger({...})完成，调用体是一个包含function_id与payload的字典。

快速上手：保存一条记忆并做混合搜索

下面是 examples/python/quickstart.py 的完整代码，它演示了「先写入、再检索」的最小闭环：

"""Minimal agentmemory usage via iii-sdk. Prerequisites: pip install iii-sdk npx -y @agentmemory/agentmemory # daemon at ws://localhost:49134 Run: python examples/python/quickstart.py """ from iii import register_worker def main() -> None: iii = register_worker("ws://localhost:49134") iii.connect() iii.trigger( { "function_id": "mem::remember", "payload": { "project": "demo", "title": "auth-stack", "content": "Service uses HMAC bearer tokens; refresh every 24h.", "concepts": ["auth", "hmac", "refresh"], }, } ) hits = iii.trigger( { "function_id": "mem::smart-search", "payload": { "project": "demo", "query": "how do tokens refresh", "limit": 5, }, } ) for memory in hits.get("results", []): print(f"[{memory.get('score', 0):.3f}] {memory.get('title')}: {memory.get('content')}") if __name__ == "__main__": main()

运行方式：

python examples/python/quickstart.py

执行前请确保 daemon 已经在运行（两个脚本都假定 daemon 已启动）。这条示例里有三个值得注意的细节：

1. project字段用于项目级隔离。在 remember.ts 中，project会被trim()归一化后写入记忆记录，并在后续的 supersession 判断中用于「跨项目绝不覆盖」的保护。
2. concepts是可选的显式概念标签；即使不传，mem::smart-search也会通过查询扩展与概念召回补足语义信息。
3. limit指定返回条数。在 smart-search.ts 中，limit会被钳制在1..100之间（Math.max(1, Math.min(data.limit ?? 20, 100))），默认 20。

mem::*函数清单：用途与必填 payload

下面这张表来自 examples/python/README.md，完整列出 Python 侧可直接调用的五个核心函数：

其中mem::remember的实现在 remember.ts 中除了content为硬性必填（缺失时返回{ success: false, error: "content is required" }），还接受type、concepts、files、ttlDays、sourceObservationIds、agentId等可选字段：

• type只能是pattern、preference、architecture、bug、workflow、fact之一，非法值会回退为fact；
• ttlDays若为正数，会为记忆设置forgetAfter过期时间，由自动遗忘机制清理；
• agentId会把记忆标记到某个 Agent 名下，配合AGENTMEMORY_AGENT_SCOPE=isolated实现跨 Agent 隔离。

mem::smart-search并不只是表层的「关键词匹配」：它的底层是 hybrid-search.ts 中的HybridSearch类，采用BM25（权重 0.4）+ 向量（权重 0.6）+ 知识图谱（权重 0.3）三路召回，再用 RRF（Reciprocal Rank Fusion，RRF_K=60）融合排序，并对命中结果按会话去重（maxPerSession = 3）与按需重排（RERANK_ENABLED=true时启用）。同时，mem::smart-search还会附带召回 lessons（经验教训），默认返回前 10 条。

实战进阶：观察摄入 + 按 token 预算渲染上下文

单条记忆适合「显式记住」，而真实编码会话中更常见的模式是：把 hook 风格的事件持续喂给 daemon，再在需要时让 agentmemory 把最相关的上下文渲染回来。这正是 examples/python/observe_and_recall.py 演示的完整闭环：

"""Observation ingest + context rendering at a token budget. Pattern: send hook-style observations during a coding session, then ask agentmemory to render the most relevant context back at a fixed token budget. Prerequisites: pip install iii-sdk npx -y @agentmemory/agentmemory Run: python examples/python/observe_and_recall.py """ from datetime import datetime, timezone from iii import register_worker SESSION_ID = "py-example-session-001" PROJECT = "demo" def now_iso() -> str: return datetime.now(timezone.utc).isoformat() def main() -> None: iii = register_worker("ws://localhost:49134") iii.connect() observations = [ ("PreToolUse", {"tool": "Bash", "command": "cargo test"}), ("PostToolUse", {"tool": "Bash", "exit_code": 0}), ("UserPromptSubmit", {"prompt": "refactor auth middleware to use HMAC"}), ] for hook_type, data in observations: iii.trigger( { "function_id": "mem::observe", "payload": { "hookType": hook_type, "sessionId": SESSION_ID, "project": PROJECT, "cwd": "/home/user/service", "timestamp": now_iso(), "data": data, }, } ) context = iii.trigger( { "function_id": "mem::context", "payload": { "sessionId": SESSION_ID, "project": PROJECT, "budget": 2000, }, } ) print(f"Rendered context ({context.get('token_count', 0)} tokens):\n") print(context.get("text", "")) if __name__ == "__main__": main()

mem::observe的摄入语义

对照 observe.ts 的源码，mem::observe的 payload 校验要点是：sessionId、hookType、timestamp为硬性必填（缺失直接返回错误）；project与cwd虽是可选，但同时存在时会在 session 记录缺失的情况下隐式创建一个会话（这对跳过/session/start的插件场景至关重要）。

摄入过程中还有几层幕后处理：

1. 去重（dedup）：相同sessionId + toolName + tool_input的重复事件会被DedupMap命中并返回{ deduplicated: true }，避免 hook 重放造成记忆膨胀（见 observe.ts）。
2. 隐私清洗：原始data会先经过stripPrivateData脱敏再落库（见 observe.ts）。
3. 自动压缩：默认路径走零 LLM 的「合成压缩」（buildSyntheticCompression），保证 BM25/向量索引可用且不消耗 token；只有显式开启AGENTMEMORY_AUTO_COMPRESS=true且配置 LLM key 后，每条观察才会走 LLM 压缩（见 observe.ts）。

mem::context的 token 预算渲染

budget是可选参数，缺省时使用 daemon 侧配置的默认 token 预算。从 context.ts 看，mem::context会先收集一系列候选「块」，再在预算内择优拼接：

• 置顶的 Memory Slots（若AGENTMEMORY_SLOTS开启）；
• 项目画像（ProjectProfile：topConcepts、Key files、Conventions、Common errors）；
• 相关 Lessons（项目内 lesson 权重 ×1.5，按 confidence 排序，取前 10）；
• 同项目其他会话的总结（Summary），没有总结的会话则回退到importance >= 5的高价值观察，每会话取前 5 条。

候选块按时间倒序排序后，在header/footer开销之外逐块累加 token 数，装不下的块直接跳过；最终返回带<agentmemory-context project="...">包裹的渲染文本，以及blocks与tokens统计。示例中的budget: 2000即 2000 token 上限，适合注入到一次会话的上下文窗口。

api::*：没有 iii 运行时时的 REST 兜底

如果调用方所在宿主机没有 iii 运行时，可以改走 REST（:3111）上的 HTTP 包装函数。这些包装器在 src/triggers/api.ts 中定义，与mem::*一一对应：

• POST /agentmemory/observe→api::observe→mem::observe（要求hookType、sessionId、project、cwd、timestamp均为非空字符串，否则 400）；
• POST /agentmemory/context→api::context→mem::context（budget必须是正整数）；
• POST /agentmemory/search→api::search→mem::search（format只接受full、compact、narrative三种取值）；
• 以及 liveness（GET /agentmemory/livez）、health（GET /agentmemory/health）、sessions、observations 等管理端点。

如果设置了AGENTMEMORY_SECRET，这些 REST 端点会要求Authorization: Bearer <secret>请求头（api.ts 使用常量时间比较防止时序攻击）。正如原文档所强调的：在 iii 生态内部，直接调用mem::*函数延迟更低，REST 包装层只是跨宿主机场景的兼容手段。

小结

agentmemory 的 Python 接入路径非常干净：pip install iii-sdk→ 启动npx -y @agentmemory/agentmemory→register_worker("ws://localhost:49134")→ 用iii.trigger调用mem::remember / mem::observe / mem::context / mem::smart-search / mem::forget。你既可以像quickstart.py一样做「保存即检索」的显式记忆，也可以像observe_and_recall.py一样把整个编码会话的 hook 事件摄入进来、在任意时刻按 token 预算渲染回最相关的上下文。仓库中对应的源码（remember.ts、observe.ts、context.ts、smart-search.ts、hybrid-search.ts、api.ts）可以作为你深入理解每个函数内部行为的参考。两个示例脚本都假设 daemon 已在运行，请先启动再执行。

【免费下载链接】agentmemory#1 Persistent memory for AI coding agents based on real-world benchmarks项目地址: https://gitcode.com/GitHub_Trending/age/agentmemory