摘要
AI Agent 正在从“能对话”走向“能记住”。但 LLM 本身是无状态的,每一次 API 调用都从零开始——用户昨天告诉 Agent 的偏好、上一次工单的上下文、反复沟通后才达成的结论,在响应结束后便全部消失。解决这个问题的关键不在模型层面,而在基础设施层面:把“过去”存在模型之外,每次对话时检索出相关片段注入 Prompt。本文从 Agent 记忆系统的四步核心循环出发,结合 TiDB 8.5 的原生向量搜索、混合检索和 HTAP 能力,给出一套可落地的工程实现方案。
阅读收益:理解 Agent 记忆系统的本质架构;掌握在 TiDB 中存储文本、向量和元数据的统一建模方式;了解 8.5 版本中与 AI 工作负载相关的重要优化;获取一个可直接运行的完整示例。
一、为什么 Agent 记忆是一个基础设施问题
LLM 的无状态性意味着,每一次 API 调用都是一次“失忆重启”。用户前天告诉 Agent 的偏好、上周提交的支持工单、经过多轮对话才对齐的需求,全部丢失在响应结束的那一刻。
记忆的本质是基础设施模式,不是模型功能。你在模型之外存储过去的信息,在每一轮对话中拉取相关片段并注入 Prompt。Agent 看起来“记住了”,是因为你替它记住了。
真实的工程问题有两个:记忆存在哪里?怎么快速找到对的那一条?
剥去框架的包装,每一个 Agent 记忆系统最终都收敛为同一个循环:存储一条记录 → 将其文本嵌入为向量 → 搜索与当前查询向量最近的记录 → 将匹配到的记忆注入下一轮 LLM Prompt。摘要、事实抽取、衰减评分都是这个循环上层的附加逻辑。
二、为什么选择 TiDB 作为记忆存储
在构建 Agent 记忆系统时,一个自然的思路是“拼装式架构”:MySQL 存元数据,向量库存 Embedding,ES 存全文索引。但这意味着三套系统之间的数据同步、一致性保证和运维成本。
TiDB 的思路不同。它将文本、向量和元数据放在同一张表中,过滤条件与向量(或混合)搜索在同一个查询中完成,不需要额外的向量存储来同步。其底层原理在于:TiDB 通过同一条 Raft 流同时驱动行存、列存和向量索引,保证三者之间的一致性。
这张表的 schema 设计是整套方案的基石:
sql
CREATE TABLE memories (
id BIGINT PRIMARY KEY AUTO_RANDOM,
user_id VARCHAR(64) NOT NULL,
content TEXT NOT NULL,
embedding VECTOR(1024),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user (user_id),
VECTOR INDEX idx_embedding ((VEC_COSINE_DISTANCE(embedding)))
);
几个设计细节值得说明:
VECTOR(1024) 是原生列类型。 不需要扩展,不需要在数据库旁边额外运行一个向量存储,也不需要在两者之间做同步。VECTOR INDEX ... VEC_COSINE_DISTANCE 行构建 HNSW 向量索引,随着表增长保持近邻查找的速度。
AUTO_RANDOM 而非 AUTO_INCREMENT。 这一点对从单机 MySQL 迁移过来的用户尤为重要。顺序整数主键在分布式系统中会造成写热点——每次插入都落在同一个节点上。随机主键将插入均匀分散到集群中。
从存储引擎层面看,TiDB 使用 TiKV 作为行存引擎处理 OLTP 工作负载,TiFlash 作为列存引擎服务实时分析场景,两者通过 Raft 协议自动同步,保持强一致性。在 100 TB 级别的数据规模下,推荐使用 TiFlash MPP 作为 HTAP 的主要方案。
三、混合检索:让记忆的召回更准
单纯的向量搜索在语义相似度上表现良好,但在精确匹配场景下可能遗漏。例如用户问“我的订单 #12345 状态如何”,向量搜索可能返回语义相近但订单号不同的记录。TiDB 的混合搜索将向量搜索与基于 BM25 的全文搜索结合,通过重排序器融合两组结果。
在 Agent 记忆场景中,混合检索的价值尤为明显:向量搜索负责捕捉语义层面的“相似记忆”,全文搜索负责精确匹配用户 ID、订单号、时间范围等结构化信息。
TiDB 提供了 pytidb Python SDK,内置了 Embedding 和重排序支持,简化了混合搜索的实现:
python
from pytidb import TiDBClient
db = TiDBClient.connect(
host="gateway01.ap-southeast-1.prod.aws.tidbcloud.com",
port=4000,
username="your_username",
password="your_password",
database="agent_memory",
)
# 混合搜索:向量 + 全文
results = db.search(
table="memories",
query="用户对航班的座位偏好",
search_type="hybrid",
filters={"user_id": "user_42"},
top_k=5,
)
插入一条记忆同样简洁。如果不想自己维护 Embedding 管道,TiDB Cloud 支持在插入时自动生成向量:
python
# 方式一:传入已有的 Embedding
db.insert("memories", {
"user_id": "user_42",
"content": "用户偏好4小时以上航班的靠窗座位",
"embedding": [0.0123, -0.0456, ..., 0.0789],
})
# 方式二:由 TiDB 自动生成 Embedding
db.insert("memories", {
"user_id": "user_42",
"content": "用户偏好4小时以上航班的靠窗座位",
"auto_embed": True,
})
四、TiDB 8.5 对 AI 工作负载的关键优化
Agent 记忆系统在实际运行中会产生大量写入(每次对话都可能新增记忆),同时伴随频繁的索引维护和 Schema 变更。TiDB 8.5 在几个关键维度上做了针对性优化。
DDL 性能大幅提升。 在 Agent 记忆系统的迭代过程中,调整表结构是常态。8.5.5 对特定有损 DDL 操作的优化效果显著:在无数据截断风险的场景下,BIGINT → INT 类型的列变更在索引列场景下从 6 小时 25 分钟降至 0.05 秒,性能提升达到 46 万倍。
分布式 ADD INDEX 支持动态调优。 当记忆表增长到数亿行时,索引构建本身可能成为瓶颈。8.5.5 支持通过 ADMIN ALTER DDL JOBS 动态调整线程数、批次和写入速度,允许在不中断服务的情况下优化索引构建过程。
资源管控支持后台任务限流(GA)。 在混合负载场景中,后台的 ANALYZE、备份等任务可能抢占前台查询的资源。8.5.6 将资源管控的后台任务资源上限设为了正式功能,确保 Agent 的实时检索不受后台任务影响。
PITR 日志压缩加速恢复。 对于生产环境,恢复时间目标(RTO)至关重要。从 8.5.5 开始,PITR 支持从压缩日志备份恢复,内部基准测试中日志恢复时间缩短了 52% 至 99%,使 206.6 TiB 集群的恢复变得可行。
五、完整实现:一个最小可运行的 Agent 记忆系统
下面是一个完整的 Python 实现,展示了记忆的存储、检索和注入循环。核心逻辑可以概括为四步:存储、嵌入、搜索、注入。
python
import openai
from pytidb import TiDBClient
# 初始化
client = openai.OpenAI(api_key="your_openai_key")
db = TiDBClient.connect(
host="your_tidb_host",
port=4000,
username="root",
password="your_password",
database="agent_memory",
)
def remember(user_id: str, content: str):
"""存储一条记忆"""
embedding = client.embeddings.create(
model="text-embedding-3-small",
input=content,
).data[0].embedding
db.insert("memories", {
"user_id": user_id,
"content": content,
"embedding": embedding,
})
def recall(user_id: str, query: str, top_k: int = 5) -> list[str]:
"""检索相关记忆"""
query_embedding = client.embeddings.create(
model="text-embedding-3-small",
input=query,
).data[0].embedding
results = db.search(
table="memories",
query_vector=query_embedding,
search_type="hybrid",
filters={"user_id": user_id},
top_k=top_k,
)
return [r["content"] for r in results]
def chat(user_id: str, user_message: str) -> str:
"""完整的对话循环:检索记忆 → 注入 Prompt → 生成回答"""
memories = recall(user_id, user_message)
system_prompt = "你是一个有记忆的助手。以下是用户的历史偏好和上下文:\n"
for m in memories:
system_prompt += f"- {m}\n"
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message},
],
)
answer = response.choices[0].message.content
# 将本轮对话存入记忆
remember(user_id, f"用户问:{user_message};助手答:{answer}")
return answer
这段代码展示了一个关键模式:记忆的存储和检索与对话生成在同一个循环中完成。每一轮对话结束后,新的记忆自动写入;下一轮对话开始前,相关记忆自动检索并注入。TiDB 的统一存储让这一切不需要在多个系统之间做数据搬运。
六、写在最后
回到最开始的那个判断:Agent 记忆是基础设施问题,不是模型功能。当你接受了这个前提,数据库选型就变成了一个工程决策。
TiDB 在这个场景下的核心优势不在于“支持向量搜索”这个功能本身——市面上支持向量搜索的数据库不少——而在于向量搜索、事务处理和分析查询运行在同一个引擎上。记忆的存储需要事务保证,记忆的检索需要向量索引,记忆的使用模式分析需要 OLAP 能力,这三件事在一个 TiDB 集群中完成,不需要在三个系统之间同步数据。
如果你正在构建 AI Agent 应用,或者在评估向量数据库的选型方案,TiDB 社区专栏中有更多来自一线工程师的实践分享,涵盖 RAG 架构解析、向量检索性能实测、混合搜索调优等主题。也欢迎参与社区的技术征文活动,把你自己的实践和踩坑经验写下来——社区的认知,正是在这样的分享中不断向前推进的。