DynamoDB 原生向量搜索正式 GA：告别「业务库 + 向量库」双写时代，AI 应用数据库架构的终局之战

2026 年 8 月 11 日，亚马逊云科技宣布 Amazon DynamoDB 向量搜索功能正式可用（GA）。这条消息在技术圈引起的涟漪，远比表面看起来深远——一个 2012 年问世、服务全球超过 100 万客户、每秒处理超过 10 亿次请求的「数据库老兵」，终于把向量索引做进了核心引擎。

这意味着什么？意味着从今天起，你不再需要为了一个 RAG 功能，在业务数据库旁边再养一套 Pinecone、Milvus 或者 Weaviate；不再需要写定时任务把业务表的数据同步成向量库里的 chunk；不再需要在「业务数据」和「向量数据」之间做二选一。业务仓库和 AI 检索仓库，第一次真正意义上合二为一。

这篇文章不打算做新闻复读。我会从工程视角拆解这件事：向量检索到底难在哪、DynamoDB 为什么敢把 ANN 索引塞进 NoSQL 引擎、它的架构设计有什么取舍、你该怎么用、以及——更重要的——这套「数据库原生向量」的路线图，对整个 AI 应用架构意味着什么。

一、背景：AI 应用正在被「数据库撕裂」

先回到一个真实到不能再真实的场景。

你做一个电商客服 Agent。用户问：「我上周买的那个蓝色机械键盘，现在能退吗？」

这个问题的回答需要三样东西：

1. 业务事实：订单状态、商品信息、售后政策——这些是结构化数据，躺在 MySQL/PostgreSQL/DynamoDB 里；
2. 语义理解：「蓝色机械键盘」要匹配到商品库里的「Keychron K8 Pro 蓝色背光版」——这是向量检索的活；
3. 上下文记忆：用户上周的聊天记录、这个 Agent 跟用户的会话历史——这又是另一套存储。

于是你的架构变成了这样：

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  业务数据库   │    │  向量数据库   │    │  记忆/缓存   │
│  DynamoDB   │    │  Milvus/    │    │  Redis      │
│             │    │  Pinecone   │    │             │
└──────┬──────┘    └──────┬──────┘    └──────┬──────┘
       │                   │                   │
       └─────── 同步管道（CDC/定时任务）────────┘
                    双写、延迟、不一致

这套「三库架构」有四个挥之不去的痛点：

痛点一：数据同步是永远的坑。 业务表新增一条记录，要等 CDC 管道把它捞出来，调 embedding 模型生成向量，再写进向量库。管道挂了怎么办？数据不一致了怎么办？向量库里的数据比业务库慢 5 分钟，用户刚下的单搜不到，客服 Agent 回答「查无此单」——这是生产事故级别的体验。

痛点二：两套系统的运维成本翻倍。 向量数据库是独立的分布式系统，有自己的分片策略、容量规划、备份恢复、监控告警。为了一个搜索功能，你要养两套基础设施团队的知识。小团队根本扛不住。

痛点三：事务性被打破。 业务库删了一条记录，向量库里的 embedding 可能还在。你得自己实现「删除传播」。多租户场景下，租户 A 的数据删除还要扫一遍向量库——性能灾难。

痛点四：多一跳就是多一层延迟。 Agent 应用是延迟敏感型：先查业务库拿订单，再查向量库拿语义候选，两次网络往返。每多一跳，用户体验就多一分卡顿。

DynamoDB 向量搜索要解决的，就是把这四层痛苦全部抹平：向量数据就是业务数据，业务数据就是向量数据，一个表、一套 API、一份账单。

二、核心概念：向量检索的「不可能三角」

在拆 DynamoDB 的架构之前，先把地基打好。很多人对向量检索的理解停留在「算个余弦相似度」，但生产级的向量检索，难点全在工程上。

2.1 从暴力搜索到 ANN

最朴素的做法：把查询向量和库里的每一个向量算一遍相似度，取 Top-K。这叫暴力搜索（Brute-force / KNN）。准确率 100%，但复杂度 O(N)，1000 万条向量就要算 1000 万次点积。一次查询几十毫秒起步，规模一大直接爆炸。

生产系统用的是 ANN（Approximate Nearest Neighbor，近似最近邻）——牺牲一点点召回率，换取数量级的性能提升。核心思想是不搜全部，只搜「可能相关」的一部分。

主流的 ANN 算法分几大家族：

其中 HNSW（Hierarchical Navigable Small World） 是当下最主流的选择：它构建一张多层图，上层是「高速公路」，下层是「街区小路」，搜索时从顶层粗定位、逐层细化。你用的很多向量数据库，底层都是 HNSW 或者它的变体。

DynamoDB 官方的说法是支持 ANN 搜索、99%+ 召回率。这个数字很关键——它暗示了底层索引的调优目标：不是追求 100% 精确，而是在工程上把召回率压到 99% 这个「够用」的线上。

2.2 召回率不是玄学

召回率（Recall@K）的定义很朴素：ANN 返回的 Top-K 里，有多少是真正的 Top-K。比如库里有 100 万条向量，暴力搜索的精确 Top-10 是 {A,B,C,...,J}，ANN 返回了 {A,B,D,...,J}（9 个命中），召回率就是 90%。

工程上，召回率、延迟、内存三者构成「不可能三角」：

• 想要召回率更高 → 搜索时访问更多节点 → 延迟上升；
• 想要延迟更低 → 减少搜索范围 → 召回率下降；
• 想同时保两头 → 加内存/加机器 → 成本上升。

DynamoDB 说 99% 召回率 + 毫秒级延迟，意味着它在索引参数上做了面向「生产默认值」的调优。你在使用时也可以调整候选数等参数，在召回率和延迟之间做自己的权衡——后面代码实战部分会演示。

2.3 维度、度量和量化

向量检索还有三个基础参数：

维度（dimension）：embedding 模型的输出长度。小模型如 all-MiniLM-L6-v2 是 384 维，OpenAI 的 text-embedding-3-small 是 1536 维，大模型可以到 3072 甚至更高。维度越高，信息越丰富，但存储和计算成本越高，而且高维空间里「维度灾难」会让距离度量逐渐失效。

距离度量（metric）：常见三种——余弦相似度（cosine）、欧氏距离（L2）、内积（dot product）。注意：语义检索通常用余弦相似度，但向量数据库内部经常把向量归一化后转成内积计算，因为内积可以用高度优化的矩阵乘法（SIMD/GPU）加速。

量化（quantization）：把 float32 的向量压成 int8 甚至二进制，大幅降低内存占用。IVF-PQ 就是「聚类 + 量化」的组合拳：先粗聚类缩小搜索范围，再量化压缩每个向量的存储。代价是精度损失，需要重排（re-rank）来补偿。

三、架构分析：DynamoDB 是怎么把 ANN 塞进 NoSQL 的

这才是这篇文章的重头戏。把向量索引做进一个已有 14 年历史、每秒处理 10 亿请求的 NoSQL 引擎，不是「加个功能」那么简单，它牵动的是存储引擎、分区模型、一致性语义、计费模型四层设计。

3.1 DynamoDB 的原生架构回顾

先回顾 DynamoDB 的基本模型，后面所有设计决策都建立在这上面：

• 表（Table）：由分区键（Partition Key） 和可选的排序键（Sort Key） 组成主键；
• 分区（Partition）：数据按分区键的哈希分布到多个分区，每个分区是底层的存储单元，单分区有 10GB 上限（近似）；
• 二级索引：LSI（本地二级索引，必须与表同分区键）和 GSI（全局二级索引，可跨分区）；
• 一致性：默认最终一致（Eventual Consistency），可选的强一致读；
• 计费：按读写容量单位（RCU/WCU）或按请求次数（On-Demand）付费。

这套模型的关键约束是：单次查询只能基于一个分区键（或者一个 GSI 的分区键）。查询的「路由」是确定性的——给定分区键，就知道去哪个分区找。

3.2 向量索引的设计：把「全局搜索」降维成「分区内搜索」

向量检索和 DynamoDB 的原生模型有个根本冲突：向量检索天然是全局的——你不知道目标向量在哪个分区，你得在全库范围内找最近邻。但 DynamoDB 的查询必须基于分区键路由。怎么解？

DynamoDB 的做法是引入向量索引分区键（vector index partition key）：你显式指定一个属性作为向量索引的「分区维度」，向量索引按这个属性把向量划分成多个索引分区（index partitions）。查询时，你指定查询的向量分区键值，系统只在对应的索引分区内做 ANN 搜索。

这个设计非常聪明，它把「全局最近邻」问题拆成了「可控范围内的最近邻」：

向量索引分区键 = tenant_id（租户ID）

租户A的向量 → 索引分区 A
租户B的向量 → 索引分区 B
查询时指定 tenant_id=A → 只在分区 A 内做 ANN

多租户场景的查询隔离、数据隔离、成本隔离，一个分区键全解决了。 这在多租户 SaaS 里是刚需：你不能让租户 A 的查询去扫描租户 B 的向量，那既是性能灾难也是数据泄露。

换一个角度理解：DynamoDB 向量索引分区键 ≈ HNSW 里的「图分区」+「图导航头」的结合——每个索引分区维护自己的一组 HNSW 图，查询时先路由到正确的图，再在图中导航。

3.3 写入路径：业务写入和向量索引的原子更新

这是 DynamoDB 向量搜索最被低估的设计点。

传统的「业务库 + 向量库」架构，写入是两步：写业务库，再异步同步到向量库。DynamoDB 的做法是：一次 PutItem，业务属性和向量属性一起写入，向量索引在存储引擎内部同步维护。

这意味着：

1. 没有同步管道。没有 CDC、没有定时任务、没有消息队列中间层。少了一个会挂的组件，少了一大类故障。
2. 数据一致性由存储引擎保证。业务属性和向量永远在同一个 item 里，要么都更新，要么都不更新。不存在「业务库删了、向量库还留着」的孤儿数据。
3. 原子性。一个事务里既改业务字段又改向量，这是独立的向量数据库做不到的——两套系统之间没有事务边界。

当然，索引更新和读路径之间，仍然是 DynamoDB 经典的最终一致语义：写入成功后，向量索引的更新是异步落盘的，极短时间内新向量可能搜不到。这对绝大多数 AI 应用（语义检索、推荐、RAG）完全够用——没有人要求「刚写入的 embedding 必须毫秒内可被语义搜索到」，但所有人都要求「写入不能因为向量索引而变慢」。这个取舍是对的。

3.4 与独立向量数据库的正面交锋

把 DynamoDB 向量搜索和主流独立向量数据库放在一张表里对比：

有一个点必须说公道话：独立向量数据库在「纯向量检索」的极致性能上仍然领先。如果你的场景是「1 亿条向量、纯检索、不在乎业务数据」——比如以图搜图、大规模相似图片去重——独立向量数据库依然是更专业的工具。DynamoDB 向量搜索的定位是「AI 应用的数据底座」，不是「向量检索性能冠军」。选型看场景，不是看参数。

3.5 与 pgvector / MongoDB 的路线之争

DynamoDB 不是第一个吃螃蟹的。PostgreSQL 有 pgvector，MongoDB 有 Atlas Vector Search，OpenSearch 有 k-NN，Redis 有 RediSearch 向量能力，连 SQLite 都有向量扩展。

这场「数据库原生向量」的军备竞赛，本质是数据库厂商对 AI 应用数据主权的争夺：谁能让 AI 应用把数据留在自己这里，谁就赢得了下一代应用的基础设施入口。

DynamoDB 的差异化打法是无服务器 + 超大规模 + 多租户友好。pgvector 适合中小规模、PostgreSQL 生态的团队；MongoDB Atlas Vector Search 适合文档模型 + 混合搜索；DynamoDB 适合的是：高并发、多租户、无服务器、业务数据已经（或准备）在 AWS 的团队。它不是要打败所有向量数据库，而是要成为「AI 应用默认的在线数据层」。

四、代码实战：从建表到生产可用的完整流程

理论讲完，上手。我用 AWS CLI 和 Python（boto3）各演示一遍完整流程：建表 → 写数据 → 向量查询 → 带过滤的向量查询 → 混合检索。

说明：以下代码基于 DynamoDB 向量搜索的公开 API 结构编写，参数名以 AWS 官方文档为准。核心流程（创建向量索引 → 写入 → QueryVector）是稳定的。

4.1 环境准备

# 安装 boto3（要求版本支持向量搜索 API）
pip install -U boto3

# 配置凭证（用已有的 AWS CLI 配置即可）
export AWS_DEFAULT_REGION=us-east-1

4.2 创建带向量索引的表

我们做一个「商品语义搜索」的场景：商品表里既有业务属性（价格、库存、分类），又有向量属性（商品描述的 embedding）。

aws dynamodb create-table \
    --table-name products \
    --attribute-definitions \
        AttributeName=product_id,AttributeType=S \
        AttributeName=category,AttributeType=S \
        AttributeName=name,AttributeType=S \
        AttributeName=description_embedding,AttributeType=B \
    --key-schema \
        AttributeName=product_id,KeyType=HASH \
    --vector-index-specifications \
        '[
            {
                "IndexName": "category-vector-index",
                "PartitionKey": {"AttributeName": "category", "KeyType": "HASH"},
                "VectorKey": {"AttributeName": "description_embedding"},
                "Dimension": 1536,
                "MetricType": "COSINE",
                "Projection": {"ProjectionType": "ALL"}
            }
        ]' \
    --billing-mode PAY_PER_REQUEST

逐行解读这个命令里的关键设计：

• PartitionKey: category：这就是前面说的向量索引分区键。按商品分类分区，查询时指定 category=electronics 就只搜电子产品的向量。如果业务上需要按租户隔离，这里就填 tenant_id。
• VectorKey: description_embedding：指明哪个属性存向量。注意类型是 B（Binary），向量以紧凑二进制存储，比 JSON 数组省内存、省带宽。
• Dimension: 1536：与你的 embedding 模型输出维度一致。维度在建索引时固定，之后不能改——所以选 embedding 模型要慎重，换模型 = 重建索引。
• MetricType: COSINE：距离度量。语义检索选 COSINE；个性化推荐（要保留向量模长信息）选 DOT_PRODUCT；坐标类数据选 L2。
• Projection: ALL：索引里投影所有属性。如果只投影必要属性，查询更快更省，但注意索引里没有的属性要在查回主表。

4.3 写入数据：业务属性和向量一起写

import boto3
from botocore.config import Config

dynamodb = boto3.client("dynamodb", config=Config(retries={"max_attempts": 5}))
TABLE = "products"

def embed(text: str, dim: int = 1536) -> bytes:
    """调用 embedding 模型生成向量（示例用 Bedrock Titan Embeddings）"""
    bedrock = boto3.client("bedrock-runtime")
    resp = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=f'{{"inputText": "{text}"}}',
    )
    import json
    vec = json.loads(resp["body"].read())["embedding"]
    # 转成 float32 二进制
    import struct
    return struct.pack(f"{dim}f", *vec[:dim])

items = [
    {
        "product_id": "p-1001",
        "category": "electronics",
        "name": "Keychron K8 Pro 机械键盘",
        "price": 89.99,
        "stock": 120,
        "tags": {"mechanical", "bluetooth", "brown-switch"},
        "description_embedding": embed("Keychron K8 Pro wireless mechanical keyboard with brown switches, bluetooth 5.1, hot-swappable"),
    },
    {
        "product_id": "p-1002",
        "category": "electronics",
        "name": "Logitech MX Master 3S",
        "price": 99.99,
        "stock": 80,
        "tags": {"mouse", "wireless", "ergonomic"},
        "description_embedding": embed("Logitech MX Master 3S wireless ergonomic mouse, 8K DPI sensor, silent clicks"),
    },
    {
        "product_id": "p-2001",
        "category": "home",
        "name": "Nespresso Vertuo 咖啡机",
        "price": 149.00,
        "stock": 45,
        "tags": {"coffee", "kitchen"},
        "description_embedding": embed("Nespresso Vertuo coffee machine, centrifusion technology, 5 cup sizes"),
    },
]

with dynamodb.batch_writer(overwrite_by_pkeys=["product_id"]) as batch:
    for item in items:
        # 向量属性用 B 类型（Binary）写入
        item_b = {
            "product_id": {"S": item["product_id"]},
            "category": {"S": item["category"]},
            "name": {"S": item["name"]},
            "price": {"N": str(item["price"])},
            "stock": {"N": str(item["stock"])},
            "tags": {"SS": list(item["tags"])},
            "description_embedding": {"B": item["description_embedding"]},
        }
        batch.put_item(TableName=TABLE, Item=item_b)

注意几个工程细节：

1. 向量属性用 B（Binary）类型而不是 L（列表）——省存储、省传输、减少序列化开销；
2. batch_writer 批量写入，减少 API 调用次数，配合 On-Demand 计费模式，写入成本是「一次请求一个 item」；
3. 如果你用 DynamoDB Streams 做下游事件（比如向量变更触发重索引），写入路径完全不用改——向量就是普通属性。

4.4 向量查询：找出语义最接近的商品

import struct

def query_vector(index_name: str, partition_key: str, query_vec: bytes, top_k: int = 5):
    resp = dynamodb.query(
        TableName=TABLE,
        IndexName=index_name,          # 向量索引名
        KeyConditionExpression="category = :pk",  # 向量索引分区键的等值条件
        ExpressionAttributeValues={":pk": {"S": partition_key}},
        VectorQuery={
            "VectorKey": "description_embedding",
            "QueryVector": {"B": query_vec},
            "TopK": top_k,
            "NumCandidates": top_k * 10,  # 候选数：越大召回率越高，延迟越高
        },
    )
    return resp["Items"]

# 用户搜「无线鼠标」
query_embedding = embed("wireless mouse for programming")
hits = query_vector("category-vector-index", "electronics", query_embedding, top_k=3)

for hit in hits:
    name = hit["name"]["S"]
    score = hit.get("VectorScore", {}).get("score", "N/A")
    print(f"{name}  similarity={score}")

几个关键点：

• KeyConditionExpression 必须指定向量索引分区键——这是「分区内 ANN」的入口，也是多租户隔离的强制手段；
• TopK：返回多少条；
• NumCandidates：ANN 搜索的候选池大小。这个参数就是「召回率 vs 延迟」旋钮：候选池越大，越接近暴力搜索，召回率越高，但延迟和 CPU 成本越高。经验值：TopK × 10 起步，压测后按需调；
• VectorScore：每条结果带相似度分数，用于后处理（比如过滤低于阈值的弱匹配）。

4.5 带过滤的向量查询：让业务条件参与检索

向量检索最怕的就是「先向量搜完再过滤」——如果过滤条件把 Top-K 里的结果全滤掉了，召回率直接崩掉。DynamoDB 的做法是把过滤条件下推到索引扫描阶段：

resp = dynamodb.query(
    TableName=TABLE,
    IndexName="category-vector-index",
    KeyConditionExpression="category = :pk",
    FilterExpression="price < :max_price AND stock > :min_stock",
    ExpressionAttributeValues={
        ":pk": {"S": "electronics"},
        ":max_price": {"N": "95.00"},
        ":min_stock": {"N": "0"},
    },
    VectorQuery={
        "VectorKey": "description_embedding",
        "QueryVector": {"B": query_embedding},
        "TopK": 5,
        "NumCandidates": 50,
    },
)

工程要点：过滤条件下推（filter pushdown） 意味着索引在导航图的同时评估过滤条件，而不是先取回 Top-K 再在应用层过滤。代价是候选池要相应放大（比如 NumCandidates 提到 50），否则过滤会吃掉召回率。这是所有「向量 + 过滤」场景的通用调优套路：过滤越严，候选池越大。

4.6 混合检索：向量 + 关键词的工程模式

DynamoDB 原生没有全文检索（那是 OpenSearch 的活），但工程上「向量召回 + 关键词精排」的混合检索可以这样落地：

def hybrid_search(query_text: str, keyword: str, partition: str, top_k: int = 5):
    """向量召回 + 关键词兜底 + 应用层融合"""
    query_embedding = embed(query_text)

    # 1. 向量召回
    vector_hits = query_vector(partition, query_embedding, top_k=top_k * 2)

    # 2. 关键词精确匹配（用 GSI 或 Scan + Filter，数据量小可用）
    kw_hits = dynamodb.scan(
        TableName=TABLE,
        FilterExpression="contains(#n, :kw)",
        ExpressionAttributeNames={"#n": "name"},
        ExpressionAttributeValues={":kw": {"S": keyword}},
    )["Items"]

    # 3. 融合（简单加权；生产可用 RRF 或学习排序）
    from collections import defaultdict
    scores = defaultdict(float)
    for rank, item in enumerate(vector_hits):
        scores[item["product_id"]["S"]] += 1.0 / (rank + 1)
    for rank, item in enumerate(kw_hits):
        scores[item["product_id"]["S"]] += 0.5 / (rank + 1)

    ranked = sorted(scores.items(), key=lambda x: -x[1])[:top_k]
    return [pid for pid, _ in ranked]

这套模式的价值：用 DynamoDB 一个库同时支撑语义召回和关键词兜底，把「向量库 + 搜索引擎」两套系统的融合逻辑，收敛成应用层一个函数。

五、性能优化：把 99% 召回率守住的生产手册

功能能跑和线上能扛是两码事。以下是 DynamoDB 向量搜索上生产前必须过一遍的检查清单。

5.1 向量索引分区键设计（最重要的决策）

分区键决定了搜索的「扇出」范围，直接决定延迟和成本：

• 按租户分区：多租户 SaaS 的首选。查询天然隔离，一个租户的流量洪峰不会拖垮别人（DynamoDB 的吞吐隔离在分区级别）；
• 按业务域分区：单租户但业务线多（电商的「电子产品」「家居」「图书」），按域分区可以让语义检索更聚焦——「无线鼠标」在电子产品域内搜索，不会被咖啡机干扰；
• 按时间分区：时序语义检索（新闻、日志、推荐流），按天/周分区，旧数据自动冷下来；
• 全局单分区：数据量小、语义检索需要全局视野（比如「全库知识问答」）。注意：单分区意味着单分区吞吐上限，量大了要拆。

反面教材：把高基数的属性（比如 product_id）当向量索引分区键——每个分区只有一两条向量，索引退化成精确匹配，ANN 的优势全没了。分区键的基数要「适中」：每个分区几百到几百万条向量是健康区间。

5.2 召回率调优三板斧

如果你发现线上召回率不达标（比如 RAG 检索经常漏掉正确答案），按这个顺序排查：

1. 放大 NumCandidates。这是性价比最高的旋钮。候选池从 TopK×10 提到 TopK×20，召回率通常能涨 1~3 个百分点，延迟代价在毫秒级；
2. 检查过滤条件。FilterExpression 越严，有效候选越少。把过滤条件里「可下推的」尽量写进查询（DynamoDB 会下推到索引扫描），把「不可下推的」（比如需要应用层计算的）挪到后处理；
3. 重排（re-rank）。对 ANN 返回的 Top-K（比如 K=100）用精确距离重排取 Top-10。DynamoDB 返回的相似度分数可以直接用，或者你在应用层用更精细的模型重排（交叉编码器）。这是 RAG 生产系统的标准姿势。

5.3 维度、量化和成本

• 维度选择：能用 384 维解决的别用 1536 维。每 100 万条 1536 维 float32 向量 ≈ 6GB 存储（不含索引开销）。维度砍一半，存储和计算成本都接近减半，召回率损失通常可以接受——用你自己的数据做消融实验，别拍脑袋；
• 二进制存储：向量用 B 类型存，比存 JSON 数组省 30%+ 的序列化开销；
• 投影裁剪：向量索引的 Projection 只投影查询需要的属性。查完 Top-K 如果需要完整 item，再按主键 GetItem——很多场景（比如只取 id 列表）根本不需要全属性投影；
• 按需计费 vs 预留容量：向量搜索是「读多写少、查询波动大」的负载，On-Demand（按请求计费）通常更划算；如果查询量稳定且巨大，预留容量有折扣。用 CloudWatch 监控 ConsumedReadCapacityUnits 后再决定。

5.4 监控与告警

上生产至少盯这几个指标（CloudWatch）：

- VectorSearch.Latency          # p95/p99 查询延迟
- VectorSearch.Recall           # 抽样对比暴力搜索的召回率（如果有影子流量）
- ThrottledRequests             # 限流：分区键倾斜的报警信号
- ConsumedReadCapacityUnits     # 成本走势

特别提醒：向量查询的延迟对「分区内向量数量」敏感。如果某个分区的向量数暴涨（比如租户 A 疯狂灌数据），要能通过监控发现「热点分区」，及时拆分区键。

六、实战案例：一个完整的 Agent 记忆系统

把前面所有东西串起来，我用 DynamoDB 向量搜索搭一个多租户 AI Agent 的长期记忆系统——这也是官方点名的核心场景。

需求：

• 每个租户的 Agent 有独立的记忆空间（隔离）；
• 记忆分「会话摘要」和「事实型记忆」两类；
• 检索时按语义找相关记忆，并支持按时间过滤；
• 记忆写入要有事务性（写入记忆的同时更新引用计数）。

import time, struct, json
import boto3

dynamodb = boto3.client("dynamodb")
TABLE = "agent-memory"

# 建表：向量索引分区键 = tenant_id（租户隔离），排序键 = created_at（时间范围）
# 向量索引: partition key = tenant_id, vector key = memory_embedding, metric = COSINE

def save_memory(tenant_id: str, memory_type: str, content: str):
    """写入一条记忆（业务属性 + 向量一次写入）"""
    vec = embed(content)
    dynamodb.put_item(
        TableName=TABLE,
        Item={
            "tenant_id": {"S": tenant_id},
            "memory_id": {"S": f"{memory_type}-{int(time.time()*1000)}"},
            "created_at": {"N": str(int(time.time()))},
            "memory_type": {"S": memory_type},
            "content": {"S": content},
            "importance": {"N": "0.8"},        # 记忆权重，检索后重排用
            "memory_embedding": {"B": vec},
        },
    )

def recall(tenant_id: str, query: str, memory_type: str | None = None, top_k: int = 5):
    """语义召回记忆，支持类型过滤"""
    vec = embed(query)
    expr_values = {":pk": {"S": tenant_id}}
    filter_expr = None
    if memory_type:
        filter_expr = "memory_type = :mt"
        expr_values[":mt"] = {"S": memory_type}

    resp = dynamodb.query(
        TableName=TABLE,
        IndexName="tenant-vector-index",
        KeyConditionExpression="tenant_id = :pk",
        FilterExpression=filter_expr,
        ExpressionAttributeValues=expr_values,
        VectorQuery={
            "VectorKey": "memory_embedding",
            "QueryVector": {"B": vec},
            "TopK": top_k,
            "NumCandidates": top_k * 20,   # 记忆检索对召回率更敏感，候选池放大
        },
    )
    # 按 importance 加权重排
    items = sorted(resp["Items"], key=lambda it: float(it["importance"]["N"]), reverse=True)
    return [{"content": it["content"]["S"], "score": it.get("VectorScore", {})} for it in items]

# 事务：写入记忆 + 更新用户画像计数（同一事务，原子生效）
def save_memory_with_counter(tenant_id: str, user_id: str, content: str):
    vec = embed(content)
    try:
        dynamodb.transact_write_items(
            TransactItems=[
                {
                    "Put": {
                        "TableName": TABLE,
                        "Item": {
                            "tenant_id": {"S": tenant_id},
                            "memory_id": {"S": f"fact-{int(time.time()*1000)}"},
                            "created_at": {"N": str(int(time.time()))},
                            "memory_type": {"S": "fact"},
                            "content": {"S": content},
                            "user_id": {"S": user_id},
                            "memory_embedding": {"B": vec},
                        },
                    }
                },
                {
                    "Update": {
                        "TableName": "user-profile",
                        "Key": {"user_id": {"S": user_id}, "tenant_id": {"S": tenant_id}},
                        "UpdateExpression": "ADD memory_count :one",
                        "ExpressionAttributeValues": {":one": {"N": "1"}},
                    }
                },
            ]
        )
    except dynamodb.exceptions.TransactionCanceledException as e:
        print(f"事务失败，记忆未写入: {e}")

这套系统的工程价值：

1. 租户隔离是架构级的——向量索引分区键强制每次查询都在租户自己的分区内进行，不会串数据；
2. 事务性记忆写入——transact_write_items 保证「记忆 + 画像计数」要么都成功要么都失败，这是独立向量数据库做不到的（跨系统无事务）；
3. 一个表搞定记忆的全部生命周期——语义检索、按类型过滤、按重要性重排、按时间排序（排序键），全部在 DynamoDB 内完成。

七、总结展望：数据库的「融合终局」已经来了

回头看，DynamoDB 向量搜索 GA 不是一个孤立的功能发布，它是数据库行业十年趋势的必然结果：数据形态越来越多样（结构化 + 文本 + 向量 + 图），而开发者越来越不想为每一种数据形态养一套数据库。

这条融合路线上，各家已经交出了自己的答卷：

• PostgreSQL：pgvector 扩展，关系模型 + 向量，中小规模的事实标准；
• MongoDB：Atlas Vector Search，文档模型 + 向量 + 混合搜索，开发者体验最好的一档；
• MySQL：HeatWave 向量存储，走分析型路线；
• DynamoDB：无服务器 + 数万亿向量 + 多租户隔离，在线事务型 AI 应用的底座；
• SQLite：本地向量扩展，边缘设备上的 RAG 也能跑。

这场融合的赢家，不是「向量检索最快」的库，而是「让 AI 应用架构最简单」的库。DynamoDB 的答卷是：业务数据就是向量数据，向量数据就是业务数据，一个表、一套 API、一份账单、零同步管道。

对你的工程决策，我的建议很直接：

1. 新项目：如果业务数据在 AWS，AI 应用需要在线语义检索（Agent 记忆、RAG、推荐），直接评估 DynamoDB 向量搜索，别再默认「业务库 + 独立向量库」了；
2. 存量项目：如果双库架构的同步管道已经让你头疼，把「向量搜索」作为 DynamoDB 数据底座迁移的第一个试点场景；
3. 纯检索场景（以图搜图、大规模去重、离线批量相似度计算），独立向量数据库依然是更专业的工具，别为了「统一」而强行迁。

最后说一句可能得罪人的话：过去两年，很多团队为了「AI 应用」这个标签，把架构搞复杂了——三个数据库、五条同步管道、七个微服务，其实核心需求只是「查一下相似的内容」。DynamoDB 向量搜索这类「数据库原生 AI 能力」的成熟，最大的价值不是省了多少钱，而是让 AI 应用回归简单。

架构的价值不是复杂，而是恰到好处。数据库融合的终局，就是让开发者少操一份心。