TencentDB Agent Memory Python SDK 实战:用 v2/v3 双客户端接入团队级 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
本文是tencentdb-agent-memory-sdk-python(Python 模块名tencentdb_agent_memory)的完整使用指南。该 SDK 是TencentDB Agent Memory(一个面向 AI Agent 的团队级记忆中心)的官方 Python 客户端,提供同步(MemoryClient)与异步(AsyncMemoryClient)两套接口,覆盖 L0 对话、L1 原子记忆、L2 场景文件、L3 核心记忆、Offload 上下文压缩,以及 v3 严格隔离数据面、管理面(Metadata / Knowledge)与 Skill 资产管理。读完本文,你将掌握从安装、分层调用到隔离模型选型与错误处理的全部实战技能。
一、SDK 概览与包布局
先明确几个关键标识,避免在安装与导入时混淆:
| 维度 | 值 |
|---|---|
| 发行名(PyPI) | tencentdb-agent-memory-sdk-python |
| 导入路径 | tencentdb_agent_memory |
| 当前版本 | 0.2.0(见 pyproject.toml) |
| Python 版本要求 | >=3.9 |
| 运行时依赖 | httpx>=0.24.0(自带异步支持) |
| License | MIT |
包内版本布局(与 tencentcloud-sdk-python 子模块拆版本风格一致,见 tencentdb_agent_memory/init.py):
- 默认导出指向 v2:
from tencentdb_agent_memory import MemoryClient拿到的是 v2 客户端,老代码升级 SDK 后零修改即可继续工作; - v3 需显式导入:
from tencentdb_agent_memory.v3 import MemoryClient切换到 v3 严格 isolation 版本(构造时team_id/agent_id/user_id全部必填,路径走/v3); - 管理面客户端:
from tencentdb_agent_memory.v3 import MetadataClient/AsyncMetadataClient,封装 v3 管理面接口; - Skill 客户端:
from tencentdb_agent_memory.v3 import SkillClient/AsyncSkillClient,封装/v3/skill/*。
SDK 源码结构清晰,核心文件均在sdk/memory-core/python/tencentdb_agent_memory/下:v2/client.py(v2 数据面)、v3/client.py(v3 数据面)、v3/metadata_client.py(管理面)、v3/skill_client.py(技能面)、_http.py/_v3_http.py(HTTP 传输层)、cos.py(对象存储工件读取)、errors.py(错误类型)。中文版文档与 Agent 接入指南可参阅 README_CN.md 与 AGENT_GUIDE.python.zh-CN.md。
二、安装与构建打包
2.1 安装
从 PyPI 安装(发布后):
pip install tencentdb-agent-memory-sdk-python或安装本地构建出的 wheel:
pip install ./tencentdb_agent_memory_sdk_python-0.1.0-py3-none-any.whl2.2 自行构建
项目使用hatchling作为构建后端(见 pyproject.toml)。构建 wheel:
python -m build # → dist/tencentdb_agent_memory_sdk_python-0.1.0-py3-none-any.whl或只构建 wheel(不拉取构建依赖):
pip wheel . --no-deps -w dist/pyproject.toml还声明了开发依赖组dev(pytest、pytest-asyncio、respx、build、python-dotenv),需要时用pip install -e ".[dev]"安装。值得注意,[tool.hatch.build.targets.wheel]只打包tencentdb_agent_memory一个包,SDK 唯一的运行时依赖是httpx,因此整体体积与依赖面都很轻。
三、快速开始:同步 MemoryClient(v2 数据面)
3.1 构造参数
MemoryClient的构造签名(见 v2/client.py):
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
endpoint | str | "" | 记忆服务 Base URL,如http://127.0.0.1:8420 |
api_key | str | "" | Bearer Token,通过Authorization头发送 |
service_id | str | None(必填) | 记忆实例 ID,通过x-tdai-service-id头发送 |
timeout | float | 30 | 请求超时(秒) |
verify | bool | False | 是否校验 TLS 证书(v2 默认关闭) |
stub | Stub | None | 注入自定义传输层(便于测试) |
注意:未注入
stub时必须提供service_id,否则抛出ValueError。
3.2 最小可用示例
from tencentdb_agent_memory import MemoryClient client = MemoryClient( endpoint="http://127.0.0.1:8420", api_key="your-api-key", service_id="your-memory-space-id", ) # L0: append a conversation result = client.add_conversation( session_id="sess-1", messages=[ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi!"}, ], ) print(result["accepted_ids"]) # L1: search structured memories hits = client.search_atomic(query="user preferences", limit=5) print(hits["items"]) # L1: update a memory note client.update_atomic(id="note-xxx", content="updated content", background="context") # L2: list scenario files scenarios = client.list_scenarios(path_prefix="") print(scenarios["entries"]) # L2: read a scenario file file = client.read_scenario("工作.md") print(file["content"]) # L2: update a scenario file (must already exist) client.write_scenario("工作.md", "# Updated content", summary="new summary") # L3: read core memory (persona) core = client.read_core() print(core["content"]) # L3: write core memory client.write_core("# User Profile\n...") # Offload v2: send tool pairs for server-side L1 async processing (fire-and-forget) client.offload_ingest( session_id="agent_sess_123", tool_pairs=[ {"tool_name": "search", "tool_call_id": "call_1", "params": {"q": "..."}, "result": "...", "timestamp": "..."}, ], ) # Offload v2: server-side context compaction (sync wait for result) compacted = client.offload_compact( session_id="agent_sess_123", messages=[...], ratio=0.7, context_window=128000, ) print(compacted["messages"], compacted["report"]) # Read memory pipeline artifacts (e.g. persona.md, scene_blocks/*.md) raw = client.read_file("scene_blocks/工作.md")3.3 底层传输原理
所有方法最终都经由 _http.py 中的HttpStub发出POST请求,关键机制有三点:
- 鉴权头:请求固定携带
Authorization: Bearer {api_key}、x-tdai-service-id: {service_id}与Content-Type: application/json; - 响应 envelope 解包:服务端返回
{code, message, data}包裹结构,code == 0时只把data返回给调用方;code != 0时抛出TDAMError; - 链路追踪:若响应头带有
x-trace-id,会被合并进返回 dict(键名trace_id),便于与网关侧日志串接排查。
另一个值得注意的细节是团队记忆 4 ID 隔离字段:v2 的每个方法都接受可选的team_id/agent_id/user_id/task_id(源码中_id_fields辅助函数负责剔除None后注入请求体)。服务端resolveIsolation优先取 body 字段,缺失时回退到x-tdai-*header——这为后续 v3 的严格隔离语义做了铺垫。
四、分层 API 全解(L0–L3 与 Offload)
SDK 完整暴露了 v2 数据面的 14 条分层 API 与 3 条 Offload API。分层语义对应 Agent 记忆的加工深度:
- L0 Conversation(对话原始层):追加、查询、搜索、删除会话消息,是记忆体系的原始输入;
- L1 Atomic(结构化原子记忆):把对话蒸馏成一条条可检索的"原子记忆"(偏好、事实、决策等),支持
type过滤与全文检索; - L2 Scenario(场景文件):以 Markdown 文件组织场景化记忆(如
工作.md),读写需显式指定路径; - L3 Core(核心记忆 / Persona):面向 agent 人格与长期画像的核心记忆,通常由管线自动生成,也可手动覆写;
- Offload(服务端卸载):把 L1 异步抽取、上下文压缩、任务流程图查询下沉到服务端执行。
4.1 API 方法总表
下表完整列出各方法与其对应端点(同步/异步客户端 API 面一致):
| Layer | Method | Endpoint |
|---|---|---|
| L0 | add_conversation() | POST /v2/conversation/add |
| L0 | query_conversation() | POST /v2/conversation/query |
| L0 | search_conversation() | POST /v2/conversation/search |
| L0 | delete_conversation() | POST /v2/conversation/delete |
| L1 | update_atomic() | POST /v2/atomic/update |
| L1 | query_atomic() | POST /v2/atomic/query |
| L1 | search_atomic() | POST /v2/atomic/search |
| L1 | delete_atomic() | POST /v2/atomic/delete |
| L2 | list_scenarios() | POST /v2/scenario/ls |
| L2 | read_scenario() | POST /v2/scenario/read |
| L2 | write_scenario() | POST /v2/scenario/write |
| L2 | rm_scenario() | POST /v2/scenario/rm |
| L3 | read_core() | POST /v2/core/read |
| L3 | write_core() | POST /v2/core/write |
| Offload | offload_ingest() | POST /v2/offload/ingest |
| Offload | offload_compact() | POST /v2/offload/compact |
| Offload | offload_query_mmd() | POST /v2/offload/query-mmd |
4.2 各层方法参数细节
L0 对话:query_conversation支持session_id、limit/offset分页与time_start/time_end时间窗过滤;search_conversation额外接受query关键词;delete_conversation的message_ids与session_id二选一(源码中二者均为可选,但至少要提供一个才会产生有效删除动作)。
L1 原子记忆:search_atomic(query, limit, type, ...)按语义/关键词检索;query_atomic(type, limit, offset, time_start, time_end)按类型与时间窗枚举;update_atomic(id, content, background)的background用于携带更新时的上下文信息,供服务端判断改写策略;delete_atomic(ids)接受 ID 列表批量删除。
L2 场景文件:read_scenario返回{content, created_at, updated_at},文件不存在时content为None;write_scenario要求目标文件已存在(必须先创建再更新),可附带summary供场景索引使用;rm_scenario删除指定路径文件。
L3 核心记忆:read_core返回{content, created_at, updated_at},若核心记忆尚未生成则content为None;write_core(content)直接覆写。
Offload(服务端卸载):
offload_ingest(session_id, tool_pairs, prompt=None, recent_messages=None):上报工具调用对以触发服务端 L1 异步处理,可 fire-and-forget(忽略返回值)。tool_pairs每项含tool_name、tool_call_id、params、result、timestamp,可选duration_ms;prompt携带最新 user message 用于 L1.5 任务判断;recent_messages(role+content)辅助 L1 提取上下文;offload_compact(session_id, messages, ratio, total_tokens, context_window=None, message_tokens=None):对完整对话执行服务端压缩,同步等待结果,返回{messages, report}。ratio为当前 token 使用比例(已用 / context_window),total_tokens需包含 system prompt、tool schemas 等不在 messages 中的隐性开销,服务端据此计算 fixed overhead 并校准估算;若提供message_tokens列表则可跳过服务端逐条估算,提升性能;offload_query_mmd(session_id, limit=None):查询会话的任务流程图(MMD 文件),返回{mmds, current_mmd},mmds每项含filename、content、version;limit=1时走快速路径只返回当前活跃 MMD。
五、read_file:直接读取记忆管线工件
除分层 API 外,SDK 还提供client.read_file(path)直接读取记忆管线产物,例如根目录的persona.md或scene_blocks/*.md(scene_blocks/工作.md这类相对路径)。它是存储无关的公开接口:当前底层使用 COS 对象存储,但对调用方透明。
其实现(见 cos.py)分四步:
- 向平台
POST /v2/cos/secret获取 STS 临时凭证(含CosUrl、TmpSecretId、TmpSecretKey、TmpToken、ExpirationTime、PathPrefix); - 凭证按过期时间缓存,
StsCredentialManager线程安全地自动刷新,并保留 120 秒缓冲(提前过期),同时合并并发刷新请求; - 使用 STS 凭证对 COS V5 GET 请求做签名(hmac 签名);
- 以字符串返回文件内容。
read_file首次调用时惰性初始化StsCredentialManager与MemoryFileReader(复用同一个传输层的 endpoint / api_key / service_id),因此不会给纯分层调用带来额外开销。读取失败(404、鉴权失败等)统一抛TDAMError。
六、异步客户端 AsyncMemoryClient
在 asyncio 应用(FastAPI、异步 Agent 框架等)中,使用AsyncMemoryClient获得同样的 API 面,所有方法均为协程,并支持异步上下文管理器:
import asyncio from tencentdb_agent_memory import AsyncMemoryClient async def main(): async with AsyncMemoryClient( endpoint="http://127.0.0.1:8420", api_key="your-api-key", service_id="your-memory-space-id", ) as client: result = await client.search_atomic(query="preferences") print(result["items"]) asyncio.run(main())异步客户端底层使用httpx.AsyncClient(见 _http.py 的AsyncHttpStub),envelope 解包、错误抛升与x-trace-id传播逻辑与同步版完全一致;close()/__aexit__会同时关闭传输层与 COS 读取器。
七、v3 严格隔离客户端:团队级数据治理的正确姿势
当记忆需要按 团队 → Agent → 用户 严格隔离时,应切换到 v3 客户端(v3/client.py)。
7.1 与 v2 的核心差异
| 维度 | v2 | v3 |
|---|---|---|
| 构造要求 | 仅service_id必填 | team_id/agent_id/user_id全部必填,缺一立刻抛ParamError |
session_id(写入) | 可选 | add_conversation写入必填(构造或调用二选一),缺失抛ValueError |
session_id(读取) | 可选 | 可选;缺省时按(team, agent, user)跨 session 聚合(agent 维度全量视图) |
| HTTP 路径 | /v2/... | /v3/... |
TLS 校验verify | 默认False | 默认True,且构造时严格校验 endpoint / api_key / service_id / timeout |
| 附加能力 | — | 新增count_*统计接口(conversation / atomic / scenario / core) |
v3 强制team+agent+user的原因(源码注释明确):避免服务端把无 session 的写入静默合并到默认 bucket,导致不同调用方数据串扰。add_conversation缺 session_id 时的ValueError提示也直接给出了规避方案——写入必须显式带session_id,而读取可以省略以做跨 session 聚合(治理面板的 layer-counts、跨会话 L0/L1 列表等场景正是这种语义)。
7.2 典型用法
from tencentdb_agent_memory.v3 import MemoryClient client = MemoryClient( endpoint="https://memory.tencentyun.com", api_key="sk-...", service_id="mem-...", team_id="t1", agent_id="a1", user_id="u1", session_id="s1", # 可选;不传时 L0/L1 查询走跨 session 聚合 ) client.add_conversation(messages=[{"role": "user", "content": "hi"}]) client.read_scenario("notes/2026Q2.md") # L2 不消费 session_id # 跨 session 拉某 agent 的全部 L0 对话总数 client.with_isolation(session_id=None).query_conversation(limit=1)7.3 with_isolation:按需切换隔离上下文
with_isolation(team_id=None, agent_id=None, user_id=None, session_id=..., task_id=...)返回一个共享同一传输层的克隆客户端,用于在不重建连接的前提下切换隔离字段:传session_id=None或task_id=None显式清除已绑定的值,省略参数则保留当前值。这非常适合多会话 Agent 在同一进程内复用连接、逐会话处理记忆的场景。
v3 的 L2/L3 是team+agent级 profile 聚合,天然不消费session_id;此外 v3 客户端未暴露offload/read_file等非 L0–L3 接口,需要时应继续使用 v2 客户端。
八、MetadataClient:v3 管理面(元数据治理与 Knowledge 注册)
8.1 管理面 vs 数据面
MetadataClient/AsyncMetadataClient(v3/metadata_client.py)封装的是网关 v3管理面接口。与数据面MemoryClient最大的不同:不需要 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 Control 的META_ACTIONS对齐,含user-key/*),涵盖 User、UserKey、Team、TeamMember、Agent、Task、TaskAgent、ParticipationLog、Asset、AgentFixedAsset、ACL、Auth(verify_auth)、ConfigParam(get_instance_quota/get/set_user_config)等域;/v3/knowledge/*Knowledge 实体 CRUD5 条(非 meta 前缀,保留兼容)。
方法命名直接反映语义:create_user/get_user/delete_users/list_users、create_team/get_team/update_team/delete_teams/list_teams、add_team_member/remove_team_member、create_agent/archive_agent、link_task_agent/unlink_task_agent、grant_acl/revoke_acl/check_acl、verify_auth(user_key)、get_instance_quota()等,并统一支持pagination参数。
8.2 Knowledge 注册与管理
from tencentdb_agent_memory.v3 import MetadataClient meta = MetadataClient( endpoint="http://127.0.0.1:8420", api_key="verify-token", # gateway Bearer (KERNEL_AUTH_TOKEN) service_id="knowledge-debug", # x-tdai-service-id # user_key="...", # optional; only for system_admin endpoints ) # Register a wiki knowledge source k = meta.create_knowledge({ "knowledge_id": "wiki-docs", "type": "wiki", "service_url": "http://127.0.0.1:8421/v3", # Knowledge Service>from tencentdb_agent_memory.v3 import SkillClient skills = SkillClient( endpoint="https://memory.tencentyun.com", api_key="sk-...", service_id="mem-abc", team_id="t1", agent_id="agent-coder", user_id="u1", ) created = skills.create(name="py-tips", content="---\nname: py-tips\n---\n# tips\n") skills.list()文件类操作使用encode_utf8(path, content, mime_type=None, is_executable=None)/encode_base64(...)静态辅助函数构造SkillResourcePayload。SDK 同时导出一份SKILL_ERROR_CODE错误码映射,便于精确处理业务错误:
| 错误码 | 常量 | 含义 |
|---|---|---|
| 40001 | BAD_REQUEST | 请求参数非法 |
| 40301 | NOT_OWNER | 非技能所有者 |
| 40302 | TEAM_MISMATCH | 团队不匹配 |
| 40401 | NOT_FOUND | 技能/资源不存在 |
| 40901 | VERSION_STALE | 版本过期(可依据details.current_version重试) |
| 41002 | VERSION_EXPIRED | 版本已失效(可依据details.latest_version升级) |
| 41301 | RESOURCE_TOO_LARGE | 资源过大 |
| 42201 | NAME_DUPLICATE | 名称重复 |
| 42202 | PATCH_NOT_UNIQUE | patch 不唯一 |
| 42203 | FRONTMATTER_INVALID | frontmatter 非法 |
| 50301 | QUEUE_UNAVAILABLE/STORAGE_NOT_FOUND | 队列/存储不可用 |
| 50302 | LLM_UNAVAILABLE | LLM 不可用 |
| 50303 | COS_REQUIRED | 需要 COS 支持 |
十、错误处理:TDAMError 与 ParamError
10.1 TDAMError
所有返回非零code的 API 响应都会抛出TDAMError(定义见 errors.py),其字段包括:
code:服务端业务错误码;message:错误描述;request_id:请求 ID(优先取响应头x-qcloud-transaction-id,其次 envelope 内request_id),便于与服务端日志串联;details:envelope 中携带的data负载(dict 类型)——/v3/skill/*的版本类错误(40901 / 41002)会通过它返回current_version/latest_version,方便调用方干净地重试或升级。
from tencentdb_agent_memory import TDAMError try: client.read_core() except TDAMError as e: print(f"code={e.code} message={e.message} request_id={e.request_id}")底层逻辑位于 _http.py:响应体code != 0时抛错;code == 0时解包data并合并x-trace-id。v3 专用传输层 _v3_http.py 更进一步:构造时对endpoint(必须是合法 http/https URL)、api_key、service_id、timeout(正数)做严格校验,缺一即抛ParamError;响应解析时兼容 HTTP 错误状态码与非 JSON 响应。
10.2 ParamError
ParamError用于调用方参数非法(本地即抛出,不发起请求),例如:v3MemoryClient构造缺team_id/agent_id/user_id、service_id缺失、delete_conversation既无message_ids也无session_id、extract缺隔离字段、管理面请求体非 dict 等。顶层导出TDAMError与ParamError两个错误类型(见 tencentdb_agent_memory/init.py),业务代码按需捕获。
十一、总结
tencentdb-agent-memory-sdk-python用一套轻薄(仅依赖httpx)的封装,把 TencentDB Agent Memory 的团队级记忆能力完整暴露给 Python 开发者:
- v2 数据面(默认导出):L0 对话 → L1 原子记忆 → L2 场景文件 → L3 核心记忆 → Offload 服务端卸载,同步/异步双客户端 API 面一致,老代码零迁移;
- v3 数据面(显式导入):构造即强约束
team/agent/user隔离四元组,写入强制session_id、读取支持跨 session 聚合,并新增count_*统计接口,适合治理严格的多租户/多 Agent 场景; - 管理面 MetadataClient:54 条
/v3/meta/*+ 5 条/v3/knowledge/*,覆盖用户、团队、成员、Agent、任务、资产、ACL、配额与 Knowledge 元数据注册; - Skill 客户端:版本化的技能资产 CRUD 与错误码语义化;
- 工程细节:Bearer +
x-tdai-service-id鉴权、响应 envelope 解包、x-trace-id链路透传、COS 工件直读(STS 凭证自动刷新)、TDAMError/ParamError双错误体系。
接入路径建议:单 Agent 快速试用走 v2 默认导出;多团队/多 Agent 治理优先 v3 +with_isolation;知识库元数据与技能资产管理则直接使用MetadataClient与SkillClient。相关源码均可在本仓库 sdk/memory-core/python 目录下继续深挖,中英文文档与 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),仅供参考