0
0
1
0
博客/.../

用 TiDB 构建 AI Agent 记忆系统:从向量搜索到混合检索的完整实践

 TiDBer_HR53nQBE  发表于  2026-09-23
原创

摘要

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 架构解析、向量检索性能实测、混合搜索调优等主题。也欢迎参与社区的技术征文活动,把你自己的实践和踩坑经验写下来——社区的认知,正是在这样的分享中不断向前推进的。

0
0
1
0

版权声明:本文为 TiDB 社区用户原创文章,遵循 CC BY-NC-SA 4.0 版权协议,转载请附上原文出处链接和本声明。

评论
暂无评论