TencentDB Agent Memory Python SDK 接入指南：从召回、捕获到工具化的 Agent 长期记忆工程实践

TencentDB Agent Memory Python SDK 接入指南：从召回、捕获到工具化的 Agent 长期记忆工程实践

【免费下载链接】TencentDB-Agent-MemoryTencentDB Agent Memory is a team-level memory hub for AI Agents — turning conversations, docs, and code into four reusable memory assets (Chat Memory, Skill, LLM-Wiki, Code-Graph) that are governed, shared, and equipped across agents and frameworks.项目地址: https://gitcode.com/GitHub_Trending/te/TencentDB-Agent-Memory

本篇技术指南以sdk/memory-core/python/AGENT_GUIDE.python.zh-CN.md为核心，讲解如何将tencentdb-agent-memory-sdk-python完整接入一个 AI Agent：包括初始化同步/异步客户端、在用户消息发往 LLM 前并行召回 L1/L2/L3 记忆并注入 prompt、在每一轮结束后捕获新增对话写回 L0、为 LLM 暴露三个记忆检索工具，以及一整套错误降级与性能预算策略。读完本文，你将掌握一条可复制、可运行、具备源码级原理支撑的 Agent 长期记忆接入流水线。

本文围绕 TencentDB Agent Memory 的 L0–L3 分层记忆模型展开：L0 原始对话（Raw Log）全量保留作为证据兜底，L1 结构化原子记忆（Atomic Memory）自动抽取事实与偏好，L2 场景文件（Scene Block）按主题聚类、带上下文召回，L3 用户画像（Persona）沉淀稳定的协作方式。下面四件事就是把这套体系接进 Agent 的全部关键。

接入要做的四件事

把tencentdb-agent-memory-sdk-python接进一个 Agent，本质上只需要完成四件事，它们共同构成一条闭环流水线：

用户输入 → ① 召回（注入 prompt） → LLM → ② 捕获（写 L0） ↑ ③ 工具：让 LLM 自己再查 ↑ ④ 错误降级：失败不挂主流程

• ① 召回（Recall）：在用户消息发送给 LLM 之前，并行拉取三类记忆（L1 结构化记忆、L3 用户画像、L2 场景索引），拼进 system prompt，让模型"带着记忆"作答；
• ② 捕获（Capture）：Agent 一轮跑完后，把这一轮新增的 user/assistant 消息清洗后写回 L0，作为后续记忆加工的原始证据；
• ③ 工具暴露：prompt 注入的记忆有限，再给 LLM 注册tdai_memory_search、tdai_conversation_search、tdai_read_file三个工具，让模型在信息不足时主动再查；
• ④ 错误降级：记忆服务任何一路失败都不能挂掉主对话——这是接入的底线原则。

SDK 的 14 个数据面 API 速查表见 sdk/memory-core/python/README.md，本文则重点讲解如何把它们组装成一套长期记忆。

0. 初始化：同步还是异步？

SDK 同时提供同步与异步两个客户端，导入路径统一为tencentdb_agent_memory包顶层：

from tencentdb_agent_memory import MemoryClient, AsyncMemoryClient # 同步 client = MemoryClient( endpoint="https://your-memory-gateway", api_key=os.environ["MEMORY_API_KEY"], service_id="your-instance-id", ) # 异步（推荐 Agent 场景用） async with AsyncMemoryClient( endpoint="https://your-memory-gateway", api_key=os.environ["MEMORY_API_KEY"], service_id="your-instance-id", ) as client: ...

从源码看，两个客户端都基于httpx构建，MemoryClient内部持有httpx.Client，AsyncMemoryClient持有httpx.AsyncClient（见 v2/client.py）。底层 HTTP 传输层会固定带上三个头：Authorization: Bearer {api_key}、x-tdai-service-id: {service_id}与Content-Type: application/json（见 _http.py），因此：

• service_id决定 memory space 隔离粒度：同 id 数据共享、不同 id 完全隔离。它通过x-tdai-service-id请求头传给网关，相当于"记忆实例 ID"；
• Agent 场景几乎都用 async，不要用同步版——否则会阻塞事件循环，拖慢整个 Agent 主流程。

从 v2 客户端的构造函数看，service_id是必填项（缺失直接抛ValueError），同时可以传入timeout（默认 30 秒）与verify（同步版默认False，异步版默认False）。同步客户端还支持stub参数注入自定义传输实现，方便在测试中替换 HTTP 层。

需要说明的是：SDK 的包名（distribution name）是tencentdb-agent-memory-sdk-python，Python 导入名是tencentdb_agent_memory。安装方式见 sdk/memory-core/python/pyproject.toml：

# 从 PyPI 安装 pip install tencentdb-agent-memory-sdk-python # 或从本地 wheel 安装 pip install ./tencentdb_agent_memory_sdk_python-0.1.0-py3-none-any.whl

项目要求 Python >= 3.9，唯一硬依赖是httpx>=0.24.0。

版本布局说明：包顶层默认导出的MemoryClient/AsyncMemoryClient指向 v2 数据面 API，老代码升级 SDK 后零修改即可继续工作；如需 v3 严格 isolation 版本（构造时team_id/agent_id/user_id全部必填，路径走/v3），可显式from tencentdb_agent_memory.v3 import MemoryClient（见 tencentdb_agent_memory/init.py）。

1. 召回（Recall）：并行拉三类记忆注入 prompt

在用户消息发给 LLM之前，并行拉三类记忆，拼到 system prompt 里。这是记忆发挥作用的第一入口：

import asyncio async def recall(client: AsyncMemoryClient, user_query: str) -> dict: l1, persona, scenes = await asyncio.gather( client.search_atomic(query=user_query, limit=5), client.read_core(), # L3 用户画像 client.list_scenarios(), # L2 场景索引 return_exceptions=True, # 关键：单路挂不影响其它 ) l1_items = l1["items"] if not isinstance(l1, Exception) else [] persona_text = persona["content"] if not isinstance(persona, Exception) else None scene_list = scenes["entries"] if not isinstance(scenes, Exception) else [] return format_prompt(l1_items, persona_text, scene_list)

asyncio.gather(..., return_exceptions=True)是关键——任何一路超时/失败，其它两路结果照常用，不影响主对话。三路请求分别对应：

从 v2/client.py 的源码可以确认，这三个方法分别对应POST /v2/atomic/search、POST /v2/core/read、POST /v2/scenario/ls，且都支持可选的team_id/agent_id/user_id/task_id隔离四元组参数（v2 中全部可选，缺失时服务端resolveIsolation会回退到x-tdai-*header）。search_atomic还可以额外传type过滤（如只搜偏好、只搜规则）和time_start/time_end时间窗口。

拼 prompt 的两个区块

召回结果拼进 prompt 时，分成两个区块，各自承担不同职责：

• prepend_context（动态）：L1 召回结果，每轮都变，放在用户消息前。让模型在当前问题前"看到"相关历史事实；
• append_system_context（稳定）：Persona + Scene 索引 + 工具调用指南，放在 system prompt 末尾，KV cache 友好。（原文档标注为待确定项：放到 system prompt 末尾仍可能造成 KV cache miss，需要继续讨论。）

def format_prompt(l1_items, persona, scenes) -> dict: prepend = None if l1_items: lines = [f"- [{m['type']}] {m['content']}" for m in l1_items] prepend = "<relevant-memories>\n" + "\n".join(lines) + "\n</relevant-memories>" parts = [] if persona: parts.append(f"<user-persona>\n{persona}\n</user-persona>") if scenes: parts.append("## Scene Navigation\n*以下场景可用 tdai_read_file 读取详情*") parts.extend(f"- `{s['path']}`" for s in scenes) parts.append(MEMORY_TOOLS_GUIDE) # 见下文 return {"prepend": prepend, "append": "\n\n".join(parts)}

实现要点：在召回阶段缓存原始用户文本（清洁版，未注入 recall），后面 capture 阶段要用——见第 2 节。这是防止记忆被"污染"的关键一环。

注意场景索引只列 path 不列内容：L2 场景文件可能很长，prompt 里只放路径清单，真正需要时由模型通过tdai_read_file工具按需拉全文（对应 SDK 的client.read_file(path)方法，源码实现在 cos.py：先从平台POST /v2/cos/secret获取 STS 临时凭证并缓存自动刷新，再用 COS V5 签名直接读取persona.md、scene_blocks/*.md等记忆管道产物，返回文件内容字符串）。

2. 捕获（Capture）：把本轮对话清洗后写回 L0

在 agent 一轮跑完后，把这一轮新增的 user/assistant 消息清洗后写回 L0。捕获的质量直接决定后续所有记忆加工（L1 抽取、L2 场景聚类、L3 画像生成）的输入质量：

async def capture( client: AsyncMemoryClient, session_key: str, raw_messages: list, # 框架给的完整消息历史 original_user_text: str, # 召回阶段缓存的清洁版用户文本 original_user_message_count: int, # 召回阶段缓存的消息数 ): # ① 位置切片：只保留这一轮新增的消息 new_messages = raw_messages[original_user_message_count:] # ② 提取 user/assistant，去掉 tool calls / system / 多模态噪声 extracted = extract_user_assistant(new_messages) # ③ 把被 recall 污染的用户消息换回原始版 for m in extracted: if m["role"] == "user" and m["timestamp"] == new_messages[0].get("timestamp"): m["content"] = original_user_text break # ④ 文本清洗：去图片 base64、去代码块、过滤太短/纯符号 cleaned = [ {**m, "content": sanitize(m["content"])} for m in extracted if len(sanitize(m["content"]).strip()) > 5 ] if not cleaned: return # ⑤ 提交 await client.add_conversation( session_id=session_key, messages=[ { "role": m["role"], "content": m["content"], "timestamp": datetime.fromtimestamp(m["timestamp"] / 1000).isoformat(), } for m in cleaned ], )

这段代码蕴含两个非常重要的工程决策，值得单独展开：

为什么要替换"被污染的用户消息"

召回阶段会往用户消息前 prepend 一段<relevant-memories>...</relevant-memories>。如果不还原成原始文本就写 L0，下一轮召回就会基于这段被污染的文本去 search/embedding——形成反馈环：记忆里混入记忆，再基于混入的记忆召回，记忆会越来越乱、越来越膨胀。所以捕获阶段必须用第 1 节缓存的original_user_text把用户消息换回清洁版，从源头切断反馈环。

为什么要位置切片

agent 一轮结束时框架给的是完整历史，不是本轮新增。如果直接把整个历史写进 L0，每一轮都会重复写入之前所有轮次的内容——既浪费存储，又会造成同一对话被反复处理。正确做法是：召回阶段记一下消息数 N（original_user_message_count），结束时messages[N:]就是本轮新增的消息。

从 v2/client.py 可以确认add_conversation对应POST /v2/conversation/add，接受session_id与messages（role/content/timestamp列表），返回accepted_ids与total_count。它支持可选的隔离四元组参数，且可以同时传入多个会话消息一次性批量写入。

3. 工具暴露：让 LLM 自己再查

只靠 prompt 注入的记忆是有限的（大小受限、召回有损）。再注册三个工具让 LLM 自己查：

三个工具对应 SDK 的三个数据面接口：search_atomic（POST /v2/atomic/search，搜结构化原子记忆）、search_conversation（POST /v2/conversation/search，搜 L0 原始对话原文）、read_file（STS 凭证直读记忆管道产物文件）。其中search_conversation还支持session_id限定会话范围、time_start/time_end时间窗口过滤。

在 system prompt 里说清楚什么时候调，加上次数上限：

## 记忆工具 - tdai_memory_search：搜结构化记忆（用户偏好、规则、历史事件） - tdai_conversation_search：搜原始对话原文 - tdai_read_file：读取场景文件（用 Scene Navigation 列出的路径） ⚠️ memory_search + conversation_search 一轮总共最多调 3 次。

不限次数 LLM 会反复瞎搜——既是 token 开销，也会让对话节奏失控。给足使用指引和次数上限，是工具化记忆的工程必修课。

4. 错误降级：记忆挂了不能挂主对话

记忆服务是辅助能力，挂了不能挂主对话。三条原则：

1. 召回用asyncio.gather(..., return_exceptions=True)，单路失败不影响其它——已在上文实现；

2. 捕获包 try/except，失败只记日志：

   try: await capture(...) except Exception as e: logger.warning(f"capture failed: {e}")

3. 工具返回错误字符串而不是抛异常，让 LLM 自己看到 "memory unavailable" 然后继续聊——模型具备对工具错误的自然语言理解能力，把错误"翻译"成提示语交给模型处理，比直接中断对话体验好得多。

这三条原则的本质是：记忆是加分项，不是必需品。任何一条链路失效，Agent 都应退化为"无记忆也能正常对话"的基线状态。

5. 错误处理：TDAMError 与 request_id

SDK 的 HTTP 传输层统一处理响应信封：code == 0时返回data字段；非零 code 一律抛TDAMError（见 _http.py 与 errors.py）。

from tencentdb_agent_memory import TDAMError try: content = await client.read_file("scene_blocks/x.md") except TDAMError as e: if e.code == 404: pass # 文件不存在，正常情况 else: logger.warning(f"memory error code={e.code} request_id={e.request_id}")

TDAMError携带code、message、request_id三个核心字段，还支持可选的details（服务端返回的额外数据，例如/v3/skill/*接口在版本冲突时会通过它回传current_version/latest_version，便于调用方干净地重试或升级）。request_id在 server 端也有日志，排障时把 request_id 发给后端即可快速定位——同时 SDK 会把响应头的x-trace-id透传进返回结果里，方便链路追踪。

需要特别注意的是read_file的 404：文件不存在是正常情况（比如该用户还没有 persona 或某个场景块），应当静默处理而不是当作故障告警。

6. 性能建议

记忆链路是每轮对话的固定开销，必须控制在预算内：

• 召回总预算 < 200ms：三路并行后取最快可用结果，超时的丢掉。asyncio.gather并行 + 每路独立的超时控制是实现手段；
• prompt 注入控制大小：L1 ≤ 5 条、Scene 列表只列 path 不列内容、Persona 一份就够。让 LLM 不够用时再用工具拉详情——"prompt 里放摘要、工具里放全文"是核心原则；
• session 粒度：session_key是 L0 partition key，长期对话用稳定 id（用户 id + 会话 id），不要每轮换——否则同一对话被切碎成多个 session，检索与场景聚类都会失效；
• 不要在主线程同步调：用AsyncMemoryClient，别用MemoryClient，否则会阻塞事件循环。

7. 管理面：Knowledge / 元数据

上面讲的都是MemoryClient（数据面：读写记忆）。如果你还需要管理Knowledge 知识源（wiki / code-graph）的元数据，用MetadataClient（v3 管理面，不需要 isolation 四元组）：

from tencentdb_agent_memory.v3 import MetadataClient meta = MetadataClient( endpoint="http://127.0.0.1:8420", api_key="verify-token", service_id="your-instance-id", ) # 登记 / 列出 / 改名 / 删除 Knowledge 实体（管理面 CRUD，详见 README.md） meta.create_knowledge({ "knowledge_id": "wiki-1", "type": "wiki", "service_url": "http://ks:8421/v3", "name": "Wiki", "team_id": "team-1", }) meta.list_knowledge({"team_id": "team-1", "type": "wiki"})

从 v3/metadata_client.py 源码看，MetadataClient与数据面客户端的区别很明确：

• 鉴权方式不同：数据面客户端构造时必须提供team_id/agent_id/user_id四元组（严格 isolation）；管理面客户端不需要isolation 四元组，鉴权用 Bearer +x-tdai-service-id，team_id等业务字段放在请求 body 里。可选user_key会走x-tdai-user-key头（user/create、user/delete等 system_admin 接口需要）；
• 封装范围更广：覆盖/v3/meta/*公开接口 54 条（与 Panel ControlMETA_ACTIONS对齐，含user-key/*），外加/v3/knowledge/*Knowledge 实体 CRUD 5 条。除 Knowledge 外，还包含 user、team、team-member、agent、task、task-agent、participation-log、asset、agent-fixed-asset、ACL、auth、config-param 等一整套管理面操作；
• Knowledge CRUD 明细：create_knowledge（upsert，幂等，重复提交覆盖）、get_knowledge、update_knowledge（局部更新 name/summary/service_url/repo_url/branch）、delete_knowledge（批量删除，≤100 个）、list_knowledge（按 team_id 列表，支持 type 过滤）。

注意：MetadataClient只管元数据 CRUD；真正搜 wiki 内容、读页面、同步仓库要调 Knowledge Service 数据面（service_url指向的:8421），那是另一组接口，不在本 SDK 范围内。

附：sanitize 实现参考

捕获阶段的文本清洗函数处理这几类噪声——base64 图片、代码块、过短内容。这段实现来自原文档附录，可直接复制使用：

import re import time _IMAGE_DATA_URI = re.compile(r"data:image/[a-z+]+;base64,[A-Za-z0-9+/=]+", re.IGNORECASE) _CODE_BLOCK = re.compile(r"```[\s\S]*?```") def sanitize(text: str) -> str: # 去 base64 图片 text = _IMAGE_DATA_URI.sub("[image]", text) # 去代码块（assistant 输出常见，对 embedding 是噪声） text = _CODE_BLOCK.sub("[code]", text) return text.strip() def extract_user_assistant(messages: list) -> list: """从原始消息列表里提取 user/assistant 文本，丢掉 tool / system / 空内容。""" out = [] for m in messages: role = m.get("role") if role not in ("user", "assistant"): continue content = m.get("content") if isinstance(content, list): # 多模态消息：拼接 text 部分 content = "\n".join(p.get("text", "") for p in content if p.get("type") == "text") if not isinstance(content, str) or not content.strip(): continue out.append({ "role": role, "content": content.strip(), "timestamp": m.get("timestamp", int(time.time() * 1000)), }) return out

总结：一条完整的接入链路

把第 0～7 节串起来，一条完整的 TencentDB Agent Memory 接入链路是：

1. 初始化：AsyncMemoryClient(endpoint, api_key, service_id)，service_id 决定隔离空间；
2. 每轮对话前：recall()并行拉 L1/L2/L3，返回prepend+append两块 prompt 内容，同时缓存原始用户文本与消息数；
3. 每轮对话后：capture()位置切片 + 替换被污染的用户消息 + sanitize 清洗 +add_conversation写 L0；
4. 工具注册：tdai_memory_search/tdai_conversation_search/tdai_read_file三个工具 + 次数上限；
5. 全程兜底：召回return_exceptions=True、捕获 try/except、工具返回错误字符串；
6. 可选管理面：需要管理 wiki / code-graph 知识源元数据时，使用MetadataClient（v3 管理面）。

这条链路既可以在 MemoryCore 网关之上运行，也可以对接仓库内的其他插件形态（如 MemoryCore/hermes-plugin 与 MemoryCore/openclaw-plugin 提供了不同 Agent 框架的现成接入示例）。核心方法论是一致的：召回注入、增量捕获、工具补查、失败降级，四件事缺一不可。

【免费下载链接】TencentDB-Agent-MemoryTencentDB Agent Memory is a team-level memory hub for AI Agents — turning conversations, docs, and code into four reusable memory assets (Chat Memory, Skill, LLM-Wiki, Code-Graph) that are governed, shared, and equipped across agents and frameworks.项目地址: https://gitcode.com/GitHub_Trending/te/TencentDB-Agent-Memory