腾讯云数据库 AI 服务 自研 Agent 接入指引

注意：说明：

注意：适用对象：自研 AI Agent，且希望直接调用 SDK 把长期记忆能力嵌入到对话主流程中的开发者。如果你使用的是 OpenClaw、Hermes 等已有 Agent 框架，我们即将提供对应的安装接入指引（无需手写 SDK 调用）。

一、一、整体接入思路

把腾讯云数据库 Agent Memory 接入到 Agent 主流程，本质上就是 “召回 + 写入” 两步：

注意：说明：

注意：用户输入 > ① 召回（从云端拉相关记忆，拼进 prompt） > LLM > ② 写入（把本轮对话写回云端，沉淀记忆）

·召回：分两条并行路径

·主动召回：用户消息发给 LLM 之前，并行从云端拉取相关的原子记忆、场景记忆、核心记忆，分别拼到用户消息之前与 system prompt 末尾；

·工具召回：把检索接口注册成 LLM 工具暴露给模型，让它在主动召回未命中或需要补充信息时自己再查。

·写入：Agent 一轮对话跑完后，把本轮新增的用户/助手消息写入服务，由服务端自动抽取并沉淀事实 / 偏好 / 指令等原子记忆。

二、二、前置准备

在控制台开通 TencentDB Agent Memory 服务并创建一个 Memory 实例，进入实例详情页记录下：

·实例 ID（instance_id，形如 memory-xxxxxxxx，用于唯一标识当前 Memory 实例）

·服务访问地址（endpoint）

·API Key：实例开通后在详情页直接获取，无需额外创建，妥善保存即可。

三、三、安装 SDK

SDK 的环境要求、安装命令与版本说明详见 Python SDK 安装文档，安装完成后回到这篇继续第四步。

四、多 Agent 隔离说明

SDK 在初始化时需要传入 team_id、agent_id、user_id 三个字段，用于标识记忆数据归属。记忆的抽取、沉淀与组织以 agent_id 为粒度——不同 Agent 之间的记忆相互隔离，同一 Agent 下的对话数据会统一做记忆抽取。team_id 和 user_id 虽为必传参数，但不参与记忆抽取逻辑，仅作为标识 Agent 来源的辅助信息。

如果业务暂时没有多团队 / 多用户区分需求，可以将这两个字段传默认值（如 "default"）：

五、初始化客户端

SDK 同时提供同步客户端 MemoryClient 与异步客户端 AsyncMemoryClient。Agent 场景推荐使用 AsyncMemoryClient，避免阻塞事件循环。

import asynciofrom tencentdb_agent_memory.v3 import AsyncMemoryClientasync def main(): # 请先将下方占位符替换为实际配置。 async with AsyncMemoryClient( endpoint="https://{服务访问地址}", api_key="Memory 实例的 API-KEY", service_id="Memory 实例的实例 ID", team_id="{团队标识}", # 必填，无需区分时传 "default" agent_id="{Agent 唯一标识}", # 必填 user_id="{用户标识}", # 必填，无需区分时传 "default" session_id="{会话标识}", # 初始化时可选；写入时必须在此或本次调用中提供 ) as client: # 在此调用后文的召回或写入函数。 passif __name__ == "__main__": asyncio.run(main())

六、初始化参数说明

七、5.1 召回什么：三类记忆速览

TencentDB Agent Memory 把长期记忆按变化频率分成如下层级，召回时按需取用：

按触发方式，召回又分两种，本节依次说明：

·主动召回（5.2）：用户消息发给 LLM 之前，由 Agent 主动并行拉取三类记忆并拼进 prompt，每轮固定执行。

·工具召回（5.3）：把检索接口注册成 LLM 工具暴露给模型，主动召回未命中或需要补充信息时，由 LLM 自己再查。

八、步骤 1：并行拉取

三类记忆分别由不同接口提供，彼此无依赖，用 asyncio.gather 并发拉取以压缩等待时间。以下示例需配合使用：将本节 recall、步骤 2 的 format_prompt 和 5.3 的 MEMORY_TOOLS_GUIDE 放在同一 Python 文件中，并在这些定义之后调用 recall。隔离字段（team_id / agent_id / user_id）已在初始化客户端时绑定，调用时无需再传：

·search_atomic：按用户当前提问做语义检索，返回相关的 L1 原子记忆。

·read_core：读取当前 Agent 下的 L3 核心记忆（用户画像）全文。

·list_scenarios：列出当前 Agent 下的 L2 场景文件路径，只取索引、不读正文，正文交给 5.3 的工具按需取。

import asynciofrom tencentdb_agent_memory.v3 import AsyncMemoryClient# 调用 recall 前，请先定义步骤 2 的 format_prompt 和 5.3 的 MEMORY_TOOLS_GUIDE。async def recall(client: AsyncMemoryClient, user_query: str) -> dict: atomic, core, scenarios = await asyncio.gather( client.search_atomic(query=user_query, limit=5), client.read_core(), client.list_scenarios(), return_exceptions=True, # 单路失败不影响其它两路 ) atomic_items = atomic["items"] if not isinstance(atomic, Exception) else [] core_text = core["content"] if not isinstance(core, Exception) else None scenario_list = scenarios["entries"] if not isinstance(scenarios, Exception) else [] return format_prompt(atomic_items, core_text, scenario_list)

三个返回值的字段结构：

·atomic["items"]：原子记忆列表，每条含 type（事实 / 偏好 / 指令）和 content（条目正文）。

·core["content"]：核心记忆正文（用户画像，字符串）。

·scenarios["entries"]：场景路径列表，每项含 path（如 "工作.md"），不含正文。

九、步骤 2：拆成动态/稳定两个区块拼入 prompt

参照 5.1 表里的“变化频率”，把三类记忆拆成两块塞进上下文：

·动态区块（每轮都变）：L1 原子记忆召回结果，放在用户消息之前。

·稳定区块（多轮共用）：L3 核心记忆 + L2 场景索引 + 记忆工具说明，放在 system prompt 末尾。

下面的示例函数 format_prompt 把三类记忆按上述规则组织好，返回 prepend（动态区块文本，用于追加在用户消息前）和 append（稳定区块文本，用于追加在 system prompt 末尾），由调用方在发起 LLM 请求时分别注入到对应位置。

def format_prompt(atomic_items, core_text, scenarios) -> dict: prepend = None if atomic_items: lines = [f"- [{m['type']}] {m['content']}" for m in atomic_items] prepend = "<relevant-memories>\\n" + "\\n".join(lines) + "\\n</relevant-memories>" parts = [] if core_text: parts.append(f"<user-profile>\\n{core_text}\\n</user-profile>") if scenarios: parts.append("## 可用场景\\n*以下场景可调用 read_scenario 读取详情*") parts.extend(f"- `{s['path']}`" for s in scenarios) parts.append(MEMORY_TOOLS_GUIDE) # 记忆工具说明，定义见 5.3 return {"prepend": prepend, "append": "\\n\\n".join(parts)}

注意：说明：

注意：之所以把稳定内容放在 system prompt 末尾，是因为多轮对话中这部分基本不变，部分 LLM 推理框架可以复用这段前缀的中间状态，从而降低延迟和 token 成本。

十、5.3 工具召回：让 LLM 自己按需检索

5.2 的主动召回每轮固定执行，但受 prompt 容量限制只能注入有限几条。剩下的场景把以下三个能力封装成 LLM 工具，由模型自主调用：

先定义以下 MEMORY_TOOLS_GUIDE 常量，再调用 5.2 的 recall 或 format_prompt。format_prompt 会将这段工具说明加入返回的 append 文本，由调用方追加到 system prompt 末尾：

MEMORY_TOOLS_GUIDE = """## 记忆工具- memory_search：搜索结构化记忆（用户偏好、规则、历史事件）。- conversation_search：搜索原始对话原文。- read_scenario：读取场景记忆全文（路径来自系统提示中的「可用场景」列表）。一轮对话内 memory_search 与 conversation_search 合计最多调用 3 次。"""

不加次数限制时，LLM 容易反复发起无效检索，既影响响应速度也浪费配额。

十一、六、写入：将本轮对话写入云端

Agent 一轮对话跑完后，调用 add_conversation 把本轮的用户输入与 AI 回复写入云端，服务端会异步抽取并沉淀原子记忆 / 场景记忆 / 核心记忆。

写入时只需要遵守三条规则：

1. 只传用户的原始输入与 AI 的最终回复：把第五节召回阶段注入到 prompt 里的 <relevant-memories> 等记忆内容剔除掉，避免把“被记忆污染过的提问”重新写回云端形成反馈环。

2. 做一次基础文本清洗：剔除图片（如 base64、Markdown 图片语法）、代码块，以及过短或纯符号内容，这类内容对后续 embedding 与记忆抽取没有价值。

import loggingfrom tencentdb_agent_memory.v3 import AsyncMemoryClientlogger = logging.getLogger(__name__)async def capture( client: AsyncMemoryClient, session_id: str, user_text: str, # 用户原始输入（未注入记忆的清洁版本） assistant_text: str, # AI 最终回复): try: await client.add_conversation( messages=[ {"role": "user", "content": user_text}, {"role": "assistant", "content": assistant_text}, ], session_id=session_id, ) except Exception as e: logger.warning(f"write conversation failed: {e}")

注意：说明：

注意：session_id 为会话标识，同一段长对话整段沿用同一个 session_id，不要每轮变化，否则原始对话会被切碎，影响场景记忆的沉淀。