Build a unified AI agent architecture with DynamoDB and Bedrock

使用 DynamoDB 和 Bedrock 构建统一的 AI 代理架构 | AWS 架构博客

使用 DynamoDB 和 Bedrock 构建统一的 AI 代理架构

在 AWS 上构建 AI 代理的团队通常面临碎片化的数据架构：运营数据存储在 Amazon DynamoDB 中，而用于语义搜索的向量嵌入则存储在单独的专用向量数据库中。这种数据冗余会增加基础设施成本、引入同步复杂性，并扩大检索结果过时的窗口。随着 Amazon DynamoDB 原生向量搜索功能的正式发布（2026 年 8 月 5 日上线），你现在可以将嵌入向量与运营数据存储在同一个表中。通过 SearchVectors API 操作即可查询这些数据。

在本文中，我将展示如何构建统一的 AI 代理架构：Amazon Bedrock 代理使用单个 DynamoDB 表同时支持结构化查询和语义相似性搜索。该代理调用 AWS Lambda 动作组，这些动作组通过 SearchVectors 实现自然语言检索，并通过标准 DynamoDB API 执行创建、读取、更新和删除（CRUD）操作。Amazon DynamoDB Streams 管道会在内容变更时自动使用 Amazon Titan Text Embeddings V2 生成嵌入向量，从而在无需人工干预的情况下保持向量索引同步。

应用场景

设想一个技术知识管理平台，其中团队维护着数百份内部文档：操作手册、架构决策记录和故障排除指南。团队成员通过对话代理查找相关内容（“我们对支付失败的重试策略是什么？”），通过 ID 检索特定文档，或更新现有条目。

在没有原生向量搜索功能的情况下，这种架构需要一个 DynamoDB 表用于文档存储，以及一个独立的向量数据库（或 Amazon OpenSearch 服务集群）用于语义检索。DynamoDB Streams 管道必须同时写入这两个存储，代理也必须将请求路由到正确的后端。使用 DynamoDB 向量搜索功能后，你可以将架构合并为单个表，从而减少运营开销。

解决方案概述

该方案采用 DynamoDB 的单表设计，支持两种访问模式：用于运营数据的键值查询，以及用于语义查询的近似最近邻（ANN）搜索。Bedrock 代理负责协调用户交互并路由请求到相应的动作组函数。

以下列表总结了核心组件：

• DynamoDB 表的向量索引将文档、元数据和 1024 维嵌入向量存储在同一个位置。

• Bedrock 代理负责对话协调、工具选择和响应合成。

• 动作组 Lambda 函数通过 SearchVectors 对同一张表执行语义搜索，并执行 CRUD 操作。

• 嵌入管道 Lambda（由 DynamoDB Streams 触发）使用 Amazon Titan Text Embeddings V2 为新内容或修改后的内容生成嵌入向量。

架构

下图展示了统一架构中的数据流。

图 1：使用 DynamoDB 向量搜索和 Amazon Bedrock 的统一 AI 代理架构

编号步骤描述了数据和请求流程：

• 用户向 Bedrock 代理发送自然语言查询。

• 代理分析请求并调用相应的动作组 Lambda 函数。

• 对于语义搜索，动作组 Lambda 使用 Amazon Titan Text Embeddings V2 生成查询嵌入向量。

• Lambda 函数调用 DynamoDB 的 SearchVectors API（或用于操作查询的标准 CRUD API），对具有向量索引的单表执行查询。

• 当新内容写入表时，DynamoDB Streams 会捕获该变更。

• DynamoDB Streams 触发嵌入管道 Lambda 函数。

• 嵌入管道 Lambda 函数调用 Amazon Titan Text Embeddings V2 为新内容生成向量，并将其写回同一 DynamoDB 项目，向量索引会自动对其进行索引。

先决条件

要在您的账户中实现此架构，您需要以下内容：

• 具有创建 DynamoDB 表、Lambda 函数、Bedrock 代理和 IAM 角色权限的 AWS 账户。

• 在表上启用 DynamoDB Streams，且 StreamViewType 设置为 NEW_AND_OLD_IMAGES（嵌入管道会比较旧内容和新内容以防止写入循环）。

• 对 Amazon Titan Text Embeddings V2 模型（amazon.titan-embed-text-v2:0）的访问权限，该模型已在 Amazon Bedrock 模型访问中启用。

• 对 Anthropic Claude 或 Amazon Nova 模型的访问权限，用于 Bedrock 代理基础模型（通过区域检查模型支持情况）。

• Python 3.12 或更高版本（用于 Lambda 函数代码）。

实现

本节将逐步介绍架构的关键组件。

设计单表模式

该表使用复合主键（entity_id 作为分区键，sk 作为排序键），并将嵌入存储为数字列表：

# 表模式概览
# PK: entity_id (S) - 唯一文档标识符
# SK: sk (S) - 项目版本控制的排序键
# 属性: title, content, category, metadata, embedding (L of N)

向量索引通过 category 属性对搜索结果进行分区。选择一个具有适中基数的分区键，与您的查询模式匹配。基数过低的键（仅几个值）会导致数据集中在少数分区中，限制吞吐量扩展，而每个项目唯一的键则没有可比较的邻居。对于多租户工作负载，tenant_id 通常是合适的分区键。有关详细信息，请参阅 DynamoDB 向量搜索最佳实践。

以下 AWS 命令行界面 (AWS CLI) 命令会在现有表上创建向量索引：

aws dynamodb update-table \
    --table-name unified-agent-data \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
    --attribute-definitions \
        AttributeName=category,AttributeType=S \
    --vector-index-updates \
    '[{"Create": {
        "IndexName": "content-embedding-index",
        "VectorAttribute": {"AttributeName": "embedding"},
        "Dimensions": 1024,
        "DistanceFunction": "COSINE",
        "SearchSchema": [
            {"AttributeName": "category", "SearchSchemaElementType": "HASH"}
        ],
        "Projection": {"ProjectionType": "INCLUDE", "NonKeyAttributes": ["title", "category"]}
    }}]'

创建索引后，等待其变为可搜索状态。轮询 DescribeTable 直到 IndexStatus 为 ACTIVE 且 Backfilling 不再为 true。索引报告 ACTIVE 后的前几次搜索仍可能返回 ValidationException，因为 SearchVectors 由专用搜索端点提供服务。请将这些情况视为可重试错误，而非失败。

aws dynamodb describe-table --table-name unified-agent-data \
    --query 'Table.VectorIndexes[?IndexName==`content-embedding-index`].[IndexStatus,Backfilling]'

需要记住的关键约束：

• DynamoDB 向量索引需要使用按需容量模式（不支持预置模式）。

• 每张表最多支持 5 个向量索引，每个索引最多支持 4,096 个维度。

• 在每个 SearchConditionExpression 中，SearchSchema HASH 属性是必需的。

• 搜索条件仅支持等值运算符。

• SearchVectors 响应结果限制为 16 MB 且不支持分页。请仅投影需要的属性并保持 TopK 值适中以符合此限制。

• 缺少 SearchSchema HASH 属性（本例中的 category）的项目会静默地从向量索引中排除，但仍然保留在基础表中。

构建操作组 Lambda

操作组 Lambda 同时处理语义搜索和操作查询。代理根据工具定义，通过函数名称和参数调用该 Lambda。

语义搜索函数生成查询嵌入向量并调用 SearchVectors。该索引使用余弦距离，分数越低表示相似度越高。请相应命名字段以避免代理反转排序：

def semantic_search(query: str, category: str, max_results: int = 5):
    embedding = generate_embedding(query)
    results = dynamodb.search_vectors(
        TableName=TABLE_NAME,
        IndexName=INDEX_NAME,
        SearchVector=[{"N": str(v)} for v in embedding],
        TopK=min(max_results, 100),
        SearchConditionExpression="category = :cat",
        ExpressionAttributeValues={":cat": {"S": category}},
    )
    return [
        {"entity_id": r["Item"]["entity_id"]["S"],
         "title": r["Item"].get("title", {}).get("S", ""),
         "distance": r["Score"]}  # COSINE: lower = more similar
        for r in results.get("SearchResults", [])
    ]

generate_embedding 辅助函数调用 Amazon Titan Text Embeddings V2：

def generate_embedding(text: str) -> list[float]:
    response = bedrock_runtime.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=json.dumps({
            "inputText": text,
            "dimensions": 1024,
            "normalize": True
        }),
    )
    return json.loads(response["body"].read())["embedding"]

Lambda 处理程序根据 Bedrock 代理传递的函数名称路由请求：

def handler(event, context):
    function = event.get("function")
    parameters = {p["name"]: p["value"] for p in event.get("parameters", [])}
    if function == "semantic_search":
        result = semantic_search(parameters["query"], parameters["category"])
        body = json.dumps({"results": result})
    elif function == "get_item_details":
        body = json.dumps(get_item_details(parameters["entity_id"]))
    else:
        body = json.dumps({"error": f"Unknown function: {function}"})
    return {
        "messageVersion": "1.0",
        "response": {
            "actionGroup": event["actionGroup"],
            "function": function,
            "functionResponse": {"responseBody": {"TEXT": {"body": body}}}
        }
    }

使用 DynamoDB 流自动处理嵌入向量

嵌入向量流水线 Lambda 触发于 INSERT 和 MODIFY 事件。它为新内容或已更改内容生成嵌入向量，并将其写回同一项目：

def handler(event, context):
    for record in event["Records"]:
        if record["eventName"] not in ("INSERT", "MODIFY"):
            continue
        new_image = record["dynamodb"]["NewImage"]
        old_image = record["dynamodb"].get("OldImage", {})
        content = new_image.get("content", {}).get("S")
        if not content:
            continue
        # 防止无限循环：如果内容未发生变化则跳过
        if "embedding" in new_image and old_image.get("content") == new_image.get("content"):
            continue
        embedding = generate_embedding(content)
        dynamodb.update_item(
            TableName=TABLE_NAME,
            Key={"entity_id": new_image["entity_id"], "sk": new_image["sk"]},
            UpdateExpression="SET embedding = :emb",
            ExpressionAttributeValues={
                ":emb": {"L": [{"N": str(v)} for v in embedding]}
            },
        )

无限循环防护机制至关重要。缺少该机制时，Lambda 会将嵌入向量写回数据库，这会触发新的流事件，进而再次生成嵌入向量，如此循环往复。该检查通过比较新旧记录的内容字段实现，当仅嵌入向量属性发生变化时跳过处理。此防护机制要求 StreamViewType = NEW_AND_OLD_IMAGES。若未设置该参数，OldImage 将为空且防护机制永远不会生效。

在生产环境中，应配置事件源映射的 ReportBatchItemFailures 参数，使仅失败记录需要重试。为反复失败的记录添加 Amazon Simple Queue Service（Amazon SQS）死信队列（或失败目的地）。对 Amazon Bedrock InvokeModel 调用实施指数退避重试策略，以应对限流问题。

定义代理工具模式

Bedrock 代理需要一个描述可用工具的函数模式。这告诉代理何时以及如何调用每个函数：

{
    "functions": [
        {
            "name": "semantic_search",
            "description": "通过自然语言按语义搜索文档，按余弦距离（数值越低越相似）对结果排序。",
            "parameters": {
                "query": {"type": "string", "required": true,
                          "description": "自然语言搜索查询"},
                "category": {"type": "string", "required": true,
                             "description": "要搜索的文档分类"}
            }
        },
        {
            "name": "get_item_details",
            "description": "通过唯一ID检索特定文档。",
            "parameters": {
                "entity_id": {"type": "string", "required": true,
                              "description": "唯一文档标识符"}
            }
        }
    ]
}

使用此模式的场景

当您的应用已将 DynamoDB 作为主要操作存储，并希望在不管理独立服务的情况下添加语义搜索时，这种统一架构效果最佳。请考虑以下决策点：

• 当满足以下条件时使用此模式：文档频繁更新且需要立即搜索；数据集符合 DynamoDB 向量索引约束；希望最小化基础设施组件。

• 当源数据存储在 Amazon Simple Storage Service（Amazon S3）中、需要托管分块和导入功能、或不需要与操作写入关联的实时索引更新时，使用 Amazon Bedrock 知识库。

• 在需要高级搜索功能（范围过滤器、聚合、面向搜索）、查询需要超出基于等值的过滤，或需要突破100项TopK限制的结果时，使用Amazon OpenSearch Service。

安全性考虑

以下列表突出了此架构的关键安全方面：

• 最小权限IAM策略：将dynamodb:SearchVectors作用域限定到特定索引ARN（arn:aws:dynamodb:{region}:{account}:table/{table}/index/{index}）。嵌入式Lambda只需dynamodb:UpdateItem权限，无需搜索权限。

• SearchVectors无细粒度访问控制：SearchVectors API不适用DynamoDB条件键（如dynamodb:LeadingKeys）。对于多租户工作负载，使用SearchSchema HASH分区键按租户限定查询，或使用独立表实现严格隔离。

• 静态加密：DynamoDB使用您选择的AWS托管密钥、AWS管理密钥或客户托管密钥（通过AWS KMS）对包括向量嵌入在内的数据进行加密。

• 传输加密：所有SearchVectors流量均使用TLS。API路由到AWS SDK自动处理的专用搜索端点。

• Bedrock模型访问：将bedrock:InvokeModel权限限制为解决方案所需的特定嵌入和代理基础模型ARN。

• 代理到Lambda的调用：在操作组Lambda上为bedrock.amazonaws.com授予lambda:InvokeFunction权限，并使用aws:SourceArn条件限定为代理ARN。缺少此资源策略时，代理将无法调用操作组。

清理

为避免持续计费，请按以下顺序删除资源：

• 删除Bedrock代理及其操作组。

• 删除嵌入式流水线Lambda函数及其事件源映射。

• 删除DynamoDB表（这也会删除向量索引）。如果需要保留表但删除向量索引，首先运行以下命令：aws dynamodb update-table \ --table-name unified-agent-data \ --vector-index-updates '[{"Delete": {"IndexName": "content-embedding-index"}}]'

• 删除操作组Lambda函数及相关IAM角色。

结论

通过此模式，您可以构建统一的AI代理架构，使用单个DynamoDB表同时存储操作数据和实现基于向量的语义搜索。DynamoDB原生向量搜索与Bedrock代理操作组的结合消除了对独立向量数据库的需求。由DynamoDB Streams驱动的嵌入式生成保持索引实时同步。

此模式为已依赖DynamoDB且需要添加对话式AI功能的应用程序减少了基础设施复杂性。自动化的嵌入式流水线使向量索引与操作写入保持同步，而操作组设计使代理能够同时访问语义和结构化查询路径。

根据您的领域调整表结构、嵌入维度和代理指令。克隆sample-dynamodb-vector-search-architecture仓库以部署完整实现。有关DynamoDB向量搜索功能和限制的更多信息，请参阅Amazon DynamoDB向量搜索文档。

参考资料

• Amazon DynamoDB向量搜索文档

• Amazon Bedrock代理文档

• Amazon Titan文本嵌入V2

作者信息

'"` /think