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

本文介绍如何在自研 Agent 中通过 TencentDB Agent Memory SDK 接入云端长期记忆能力，让你的 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 安装文档，安装完成后回到本文继续第四步。
四、初始化客户端
SDK 同时提供同步客户端 MemoryClient 与异步客户端 AsyncMemoryClient。Agent 场景推荐使用 AsyncMemoryClient，避免阻塞事件循环。
import os
from tencentdb_agent_memory import AsyncMemoryClient
﻿
async with AsyncMemoryClient(
    endpoint="https://{服务访问地址}",
    api_key="Memory 实例的 API-KEY",
    service_id="Memory 实例的实例 ID",
) as client:
    ...
初始化参数说明
参数
是否必选
说明
endpoint
是
TencentDB Agent Memory 服务访问地址，从控制台实例详情页获取。
api_key
是
API Key，用于接口鉴权。
service_id
是
Memory 实例 ID
五、召回：在用户消息发给 LLM 前注入记忆
5.1 召回什么：三类记忆速览
TencentDB Agent Memory 把长期记忆按变化频率分成如下层级，召回时按需取用：
层级
名称
内容
变化频率
单轮召回量
L1
原子记忆
结构化的事实 / 偏好 / 指令条目
每轮都变
≤ 5 条
L2
场景记忆
按主题归档的长文（如「工作.md」「家庭.md」）
多轮不变
只拉路径列表，正文按需取
L3
核心记忆
用户画像总结
很少变
一份全文
说明：
数量约束是为了控制注入 prompt 的体积——召回阶段只塞"必读"内容，其余让 LLM 按需自取，避免上下文被记忆撑爆。
按触发方式，召回又分两种，本节依次说明：
主动召回（5.2）：用户消息发给 LLM 之前，由 Agent 主动并行拉取三类记忆并拼进 prompt，每轮固定执行。
工具召回（5.3）：把检索接口注册成 LLM 工具暴露给模型，主动召回未命中或需要补充信息时，由 LLM 自己再查。
5.2 主动召回：获取三类记忆并拼入 prompt
步骤 1：并行拉取
三类记忆分别由不同接口提供，彼此无依赖，用 asyncio.gather 并发拉取以压缩等待时间：
search_atomic：按用户当前提问做语义检索，返回相关的 L1 原子记忆。
read_core：读取当前用户的 L3 核心记忆（用户画像）全文。
list_scenarios：列出当前用户的 L2 场景文件路径，只取索引、不读正文，正文交给 5.3 的工具按需取。
import asyncio
﻿
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)}
Running Environment
Operating System: Ubuntu 24.04.3 LTS / x86_64
Runtime Version: Python 3.11.1
说明：
之所以把稳定内容放在 system prompt 末尾，是因为多轮对话中这部分基本不变，部分 LLM 推理框架可以复用这段前缀的中间状态，从而降低延迟和 token 成本。
5.3 工具召回：让 LLM 自己按需检索
﻿5.2 的主动召回每轮固定执行，但受 prompt 容量限制只能注入有限几条。剩下的场景把以下三个能力封装成 LLM 工具，由模型自主调用：
工具名
何时调用
对应 SDK 方法
memory_search
检索结构化记忆（事实 / 偏好 / 指令）
client.search_atomic(query=..., limit=...)
conversation_search
检索原始对话片段
client.search_conversation(query=..., limit=...)
read_scenario
按路径读取场景记忆全文
client.read_scenario(path=...)
在 system prompt 末尾（即 5.2 中 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 与记忆抽取没有价值。
async def capture(
    client: AsyncMemoryClient,
    session_id: str,
    user_text: str,        # 用户原始输入（未注入记忆的清洁版本）
    assistant_text: str,   # AI 最终回复
):
    try:
        await client.add_conversation(
            session_id=session_id,
            messages=[
                {"role": "user", "content": user_text},
                {"role": "assistant", "content": assistant_text},
            ],
        )
    except Exception as e:
        logger.warning(f"write conversation failed: {e}")
说明：
session_id 建议使用「用户 ID + 会话 ID」拼接，同一段长对话整段沿用同一个 session_id，不要每轮变化，否则原始对话会被切碎，影响场景记忆的沉淀。

参数	是否必选	说明
`endpoint`	是	TencentDB Agent Memory 服务访问地址，从控制台实例详情页获取。
`api_key`	是	API Key，用于接口鉴权。
`service_id`	是	Memory 实例 ID

层级	名称	内容	变化频率	单轮召回量
L1	原子记忆	结构化的事实 / 偏好 / 指令条目	每轮都变	≤ 5 条
L2	场景记忆	按主题归档的长文（如「工作.md」「家庭.md」）	多轮不变	只拉路径列表，正文按需取
L3	核心记忆	用户画像总结	很少变	一份全文

工具名	何时调用	对应 SDK 方法
`memory_search`	检索结构化记忆（事实 / 偏好 / 指令）	`client.search_atomic(query=..., limit=...)`
`conversation_search`	检索原始对话片段	`client.search_conversation(query=..., limit=...)`
`read_scenario`	按路径读取场景记忆全文	`client.read_scenario(path=...)`

自研 Agent 接入指引

本页目录：

一、整体接入思路

二、前置准备

三、安装 SDK

四、初始化客户端

初始化参数说明

五、召回：在用户消息发给 LLM 前注入记忆

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

5.2 主动召回：获取三类记忆并拼入 prompt

步骤 1：并行拉取

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

Running Environment

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

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