客户端介绍
AsyncMemoryClient 是 MemoryClient 的异步版本，提供与同步客户端完全一致的数据面方法（18 个），所有方法均为异步方法，使用 await 调用。适用于异步编程框架（如 FastAPI、LangChain 异步回调），避免阻塞等待网络响应。导入
from tencentdb_agent_memory.v3 import AsyncMemoryClient
构造参数
参数名 | 类型 | 必填 | 描述说明 |
endpoint | str | 是 | Memory 服务接入地址 |
api_key | str | 是 | API Key，格式 sk-... |
service_id | str | 否 | 实例 ID，如 tdai-mem-xxxxxxxx。使用自定义传输通道构造时可省略 |
team_id | str | 是 | 团队 ID |
agent_id | str | 是 | Agent ID（全局唯一） |
user_id | str | 是 | 用户 ID |
session_id | str | 否 | 默认会话 ID |
task_id | str | 否 | 默认 Task ID |
timeout | float | 否 | 请求超时时间（秒），默认 30 |
verify | bool | 否 | 是否验证 SSL 证书，默认 False |
stub | Stub | 否 | 自定义网络传输通道，用于测试场景（如注入 mock） |
使用示例
import asynciofrom tencentdb_agent_memory.v3 import AsyncMemoryClientasync def main():async with AsyncMemoryClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",team_id="team-abc123",agent_id="agt-xyz789",user_id="usr-456",session_id="agent-main:sess-001",) as client:# 写入对话result = await client.add_conversation(messages=[{"role": "user", "content": "帮我查一下上周的会议纪要"},],)print(f"受理 {result['total_count']} 条消息")# 检索记忆memories = await client.search_conversation(query="会议纪要",limit=5,)for msg in memories["messages"]:print(f"[score={msg['score']:.2f}] {msg['content']}")asyncio.run(main())
上下文管理器（推荐）
使用
async with 语句自动管理连接的创建和销毁：async with AsyncMemoryClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",team_id="team-abc123",agent_id="agt-xyz789",user_id="usr-456",) as client:result = await client.query_conversation(limit=20)# 退出 async with 块时自动调用 await close()
with_isolation 动态切会话
with_isolation() 方法返回共享同一网络连接的新异步客户端副本，可在运行时动态调整隔离参数：async with AsyncMemoryClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",team_id="team-abc123",agent_id="agt-xyz789",user_id="usr-456",) as client:# 切到另一个 sessionalt = client.with_isolation(session_id="agent-main:sess-002")result = await alt.query_conversation(limit=20)# 跨全部 session 聚合查询global_client = client.with_isolation(session_id=None)all_messages = await global_client.query_conversation(limit=1)print(f"全部会话共 {all_messages['total']} 条消息")
close 销毁
关闭客户端连接，释放底层
httpx.AsyncClient 资源。建议在程序退出前调用，或使用 async with 上下文管理器自动管理。await client.close()
异常处理
与同步版完全一致，SDK 提供两种异常类型：
异常类 | 说明 |
TDAMError | 服务端返回的业务错误（ code != 0），包含 code、message、request_id、details 属性 |
ParamError | 客户端参数校验失败 |
from tencentdb_agent_memory.v3 import AsyncMemoryClientfrom tencentdb_agent_memory import TDAMError, ParamErrorasync def safe_query():async with AsyncMemoryClient(endpoint="https://memory.tdai.tencentyun.com",api_key="sk-xxxxxxxxxxxxxxxx",service_id="tdai-mem-xxxxxxxx",team_id="team-abc123",agent_id="agt-xyz789",user_id="usr-456",) as client:try:result = await client.query_conversation(limit=9999)except TDAMError as e:print(f"业务错误: code={e.code}, message={e.message}, request_id={e.request_id}")except ParamError as e:print(f"参数错误: {e}")
响应格式
所有方法均返回
Dict[str, Any]（通过 await 获取），为 ApiResponseEnvelope 的 data 字段内容。当
code == 0 时直接返回 data 字典；当 code != 0 时抛出 TDAMError 异常。额外字段
trace_id：如果服务端响应头中包含 x-trace-id，会追加到返回字典中，方便问题排查。适用场景
FastAPI / Starlette：在异步路由函数中直接调用，不阻塞事件循环
LangChain 异步回调：在
acall / ainvoke 链路中使用高并发批量处理：结合
asyncio.gather 或 asyncio.TaskGroup 并发请求多个接口流式 Agent 架构：在 async generator 中边生成边写入记忆
