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让模型带着记忆作答② 捕获CaptureAgent 一轮跑完后把这一轮新增的 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( endpointhttps://your-memory-gateway, api_keyos.environ[MEMORY_API_KEY], service_idyour-instance-id, ) # 异步推荐 Agent 场景用 async with AsyncMemoryClient( endpointhttps://your-memory-gateway, api_keyos.environ[MEMORY_API_KEY], service_idyour-instance-id, ) as client: ...从源码看两个客户端都基于httpx构建MemoryClient内部持有httpx.ClientAsyncMemoryClient持有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请求头传给网关相当于记忆实例 IDAgent 场景几乎都用 async不要用同步版——否则会阻塞事件循环拖慢整个 Agent 主流程。从 v2 客户端的构造函数看service_id是必填项缺失直接抛ValueError同时可以传入timeout默认 30 秒与verify同步版默认False异步版默认False。同步客户端还支持stub参数注入自定义传输实现方便在测试中替换 HTTP 层。需要说明的是SDK 的包名distribution name是tencentdb-agent-memory-sdk-pythonPython 导入名是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唯一硬依赖是httpx0.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(queryuser_query, limit5), client.read_core(), # L3 用户画像 client.list_scenarios(), # L2 场景索引 return_exceptionsTrue, # 关键单路挂不影响其它 ) 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_exceptionsTrue)是关键——任何一路超时/失败其它两路结果照常用不影响主对话。三路请求分别对应调用记忆层返回结构从 v2 客户端与 README 确认client.search_atomic(queryuser_query, limit5)L1 结构化原子记忆items列表每项含type/content等字段client.read_core()L3 用户画像content尚未生成时为Noneclient.list_scenarios()L2 场景索引entries列表从 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(fuser-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_idsession_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每一轮都会重复写入之前所有轮次的内容——既浪费存储又会造成同一对话被反复处理。正确做法是召回阶段记一下消息数 Noriginal_user_message_count结束时messages[N:]就是本轮新增的消息。从 v2/client.py 可以确认add_conversation对应POST /v2/conversation/add接受session_id与messagesrole/content/timestamp列表返回accepted_ids与total_count。它支持可选的隔离四元组参数且可以同时传入多个会话消息一次性批量写入。3. 工具暴露让 LLM 自己再查只靠 prompt 注入的记忆是有限的大小受限、召回有损。再注册三个工具让 LLM 自己查工具何时用实现tdai_memory_search找结构化偏好/事实client.search_atomic(query..., limit...)tdai_conversation_search找原始对话片段client.search_conversation(query..., limit...)tdai_read_file读 persona / scene block 全文client.read_file(path)三个工具对应 SDK 的三个数据面接口search_atomicPOST /v2/atomic/search搜结构化原子记忆、search_conversationPOST /v2/conversation/search搜 L0 原始对话原文、read_fileSTS 凭证直读记忆管道产物文件。其中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. 错误降级记忆挂了不能挂主对话记忆服务是辅助能力挂了不能挂主对话。三条原则召回用asyncio.gather(..., return_exceptionsTrue)单路失败不影响其它——已在上文实现捕获包 try/except失败只记日志try: await capture(...) except Exception as e: logger.warning(fcapture failed: {e})工具返回错误字符串而不是抛异常让 LLM 自己看到 memory unavailable 然后继续聊——模型具备对工具错误的自然语言理解能力把错误翻译成提示语交给模型处理比直接中断对话体验好得多。这三条原则的本质是记忆是加分项不是必需品。任何一条链路失效Agent 都应退化为无记忆也能正常对话的基线状态。5. 错误处理TDAMError 与 request_idSDK 的 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(fmemory 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的元数据用MetadataClientv3 管理面不需要 isolation 四元组from tencentdb_agent_memory.v3 import MetadataClient meta MetadataClient( endpointhttp://127.0.0.1:8420, api_keyverify-token, service_idyour-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-idteam_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_knowledgeupsert幂等重复提交覆盖、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(rdata: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总结一条完整的接入链路把第 07 节串起来一条完整的 TencentDB Agent Memory 接入链路是初始化AsyncMemoryClient(endpoint, api_key, service_id)service_id 决定隔离空间每轮对话前recall()并行拉 L1/L2/L3返回prependappend两块 prompt 内容同时缓存原始用户文本与消息数每轮对话后capture()位置切片 替换被污染的用户消息 sanitize 清洗 add_conversation写 L0工具注册tdai_memory_search/tdai_conversation_search/tdai_read_file三个工具 次数上限全程兜底召回return_exceptionsTrue、捕获 try/except、工具返回错误字符串可选管理面需要管理 wiki / code-graph 知识源元数据时使用MetadataClientv3 管理面。这条链路既可以在 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考