AI Agent 往往同时需要两类数据：一类是订单、用户、库存等结构化业务数据，另一类是文档、知识库和历史记录中的语义信息。传统方案通常把业务数据放在 DynamoDB，把向量放在独立的向量数据库，再通过多个数据访问层拼接结果。这样虽然灵活，却增加了同步、权限和运维成本。
借助 Amazon DynamoDB 的原生向量搜索能力，可以把向量嵌入与业务字段放在同一张表中，再让 Amazon Bedrock Agent 通过统一的数据访问入口完成结构化查询和语义检索。DynamoDB Streams 则负责捕获数据变更，触发嵌入生成和更新流程，使向量数据与原始内容保持同步。
统一数据模型：业务字段和向量放在一起
一个可行的记录模型如下：
{
"pk": "TENANT#customer-001",
"sk": "DOC#refund-policy-v3",
"entity_type": "knowledge_document",
"title": "退款政策",
"content": "未发货订单可以在下单后七天内申请退款。",
"status": "published",
"updated_at": "2025-01-15T10:30:00Z",
"embedding": [0.0123, -0.0841, 0.2311]
}
其中，pk 和 sk 继续服务于 DynamoDB 的主键访问模式，status、tenant_id、category 等字段可用于结构化过滤，embedding 则用于语义搜索。实际建模时，需要根据租户隔离、文档类型和访问权限设计分区键，避免所有向量数据集中到少数分区。
这种设计的价值不只是减少一个数据库。结构化字段和向量结果属于同一条业务记录，Agent 在回答问题时可以直接拿到文档内容、状态、租户和权限信息，减少跨系统关联造成的数据不一致。
Bedrock Agent 如何访问统一数据层
Bedrock Agent 可以通过 Action Group 调用 Lambda 或后端 API。建议把数据访问动作拆成清晰的业务操作，例如：
get_customer_orders：按客户、时间范围和订单状态查询订单。search_knowledge：根据自然语言问题执行向量搜索，并支持租户和发布状态过滤。get_document：根据文档 ID读取完整内容和元数据。
Agent 不需要知道 DynamoDB 的分区键细节，只需要根据 OpenAPI Schema 选择合适的动作。后端负责校验租户、用户身份、过滤条件和返回字段。
一个简化的 Action Group 请求可以设计成这样：
{
"actionGroup": "KnowledgeActions",
"function": "search_knowledge",
"parameters": [
{
"name": "query",
"type": "string",
"value": "客户在什么情况下可以申请退款？"
},
{
"name": "tenant_id",
"type": "string",
"value": "customer-001"
},
{
"name": "top_k",
"type": "integer",
"value": "5"
}
]
}
在生产环境中，tenant_id 不应完全信任 Agent 传入的参数。更稳妥的做法是从 Bedrock Agent 的调用身份、API Gateway authorizer 或 Lambda 上下文中取得租户信息，并在服务端覆盖请求中的同名字段。
用 DynamoDB Streams 保持嵌入同步
当文档内容发生新增或修改时，系统需要重新生成 embedding。DynamoDB Streams 可以捕获 INSERT 和 MODIFY 事件，并将事件交给 Lambda。Lambda 调用 Amazon Bedrock Embeddings 模型，随后把结果写回同一条 DynamoDB 记录。
下面的示例展示了核心同步逻辑。它使用 Amazon Titan Text Embeddings V2 的调用形式；模型 ID、区域和向量维度需要根据实际账户和索引配置调整。示例假设原始文本保存在 content 字段，向量保存在 embedding 字段。
运行前需要为 Lambda 配置 TABLE_NAME、EMBEDDING_MODEL_ID 和 AWS IAM 权限，包括读取 Streams、调用 Bedrock 模型以及更新 DynamoDB 项目。
import json
import os
from decimal import Decimal
import boto3
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
bedrock = boto3.client("bedrock-runtime")
MODEL_ID = os.environ.get("EMBEDDING_MODEL_ID", "amazon.titan-embed-text-v2:0")
def make_embedding(text: str) -> list[float]:
response = bedrock.invoke_model(
modelId=MODEL_ID,
contentType="application/json",
accept="application/json",
body=json.dumps({
"inputText": text,
"dimensions": 1024,
"normalize": True,
}),
)
payload = json.loads(response["body"].read())
return payload["embedding"]
def lambda_handler(event, context):
updated = 0
for record in event.get("Records", []):
if record.get("eventName") not in {"INSERT", "MODIFY"}:
continue
image = record.get("dynamodb", {}).get("NewImage", {})
content = image.get("content", {}).get("S")
pk = image.get("pk", {}).get("S")
sk = image.get("sk", {}).get("S")
if not content or not pk or not sk:
continue
embedding = make_embedding(content)
table.update_item(
Key={"pk": pk, "sk": sk},
UpdateExpression="SET embedding = :embedding",
ExpressionAttributeValues={
":embedding": [Decimal(str(value)) for value in embedding]
},
)
updated += 1
return {"updated": updated}
上面的代码需要修正一个排版细节后再部署：table = ... 前不能有额外缩进。正确版本如下：
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
还需要考虑两个同步问题。第一，Lambda 更新 embedding 字段本身可能再次产生 Streams 事件，因此处理函数应检查变更是否只包含向量字段，或者使用 embedding_status、内容哈希等字段避免重复生成。第二，Bedrock 调用失败时不要覆盖旧向量，可以记录重试状态，让事件源映射或补偿任务再次处理。
读取路径：结构化查询和语义查询分开表达
统一存储不意味着所有查询都使用同一种方式。
结构化查询适合使用主键、索引和条件表达式。例如查询某个租户下已发布的退款政策，可以先根据分区键定位候选记录，再用索引或过滤条件限制状态。对于订单、库存等高频访问路径，应继续按明确的访问模式设计 GSI，而不是依赖向量搜索。
语义查询则需要使用 DynamoDB 的向量索引配置，提交查询向量，并结合租户、状态、类别等过滤条件。查询流程通常是：
- 使用与写入时相同的 Bedrock Embeddings 模型生成问题向量。
- 调用 DynamoDB 的向量搜索接口，指定向量属性、索引和
top_k。 - 应用服务端过滤和权限校验。
- 将标题、正文、来源和相关性分数整理成 Agent 可消费的结果。
不同 AWS SDK 版本对 DynamoDB 向量搜索 API 的封装可能不同，因此向量查询代码应以当前区域和 SDK 文档中的接口名称为准。下面是一个可改造的查询层伪代码，展示了边界位置，而不是假定某个固定 SDK 方法名：
def search_knowledge(query: str, tenant_id: str, top_k: int = 5):
query_vector = make_embedding(query)
# 将这里替换为当前 SDK 对 DynamoDB 原生向量搜索的调用。
# 关键参数包括：表名、向量属性、向量索引、查询向量和结果数量。
candidates = dynamodb_vector_search(
table_name=os.environ["TABLE_NAME"],
index_name=os.environ["VECTOR_INDEX_NAME"],
vector_attribute="embedding",
query_vector=query_vector,
top_k=top_k,
filter_expression={
"tenant_id": tenant_id,
"status": "published",
},
)
return [
{
"document_id": f'{item["pk"]}:{item["sk"]}',
"title": item.get("title"),
"content": item.get("content"),
"score": item.get("score"),
}
for item in candidates
]
过滤条件尤其重要。没有租户过滤时，语义相似度可能把另一个客户的内容返回给 Agent；没有 status = published 时，草稿或已撤回文档也可能进入上下文。权限控制应在数据访问层完成，而不是交给模型自行判断。
一张表带来的收益和边界
这种架构适合业务数据和知识数据联系紧密、访问权限一致、团队希望减少基础设施数量的场景。主要收益包括：
- 原始内容、业务元数据和 embedding 的生命周期更容易绑定。
- Streams 可以把内容变更转化为可追踪的异步处理任务。
- Agent 的 Action Group 只需要面对一个统一的数据服务层。
- 租户、状态和业务条件可以与语义检索结果一起处理。
它也有明确边界。向量维度、索引配置、查询吞吐和成本需要在真实数据规模下验证。大规模文档通常还需要分块，否则单条记录过长会降低召回质量。重新选择 embedding 模型或维度时，不能直接覆盖旧向量，应该通过版本字段或新属性进行迁移，待新索引验证完成后再切换查询路径。
此外，Agent 不应直接获得整张 DynamoDB 表的访问权限。把 DynamoDB 封装在窄接口之后，可以集中处理输入校验、审计日志、限流、脱敏和结果截断，也能避免把过多内容直接塞入模型上下文。
落地检查清单
可以按以下顺序推进：
- 列出 Agent 的结构化和语义访问模式，先确定主键、GSI 和租户边界。
- 选择 embedding 模型和向量维度，固定写入与查询使用的模型版本。
- 创建包含向量属性和向量索引的 DynamoDB 表配置。
- 打开 DynamoDB Streams，部署嵌入同步 Lambda，并处理重试和幂等。
- 为 Bedrock Agent 定义最小化的 Action Group 和 OpenAPI Schema。
- 在服务端强制执行租户、权限、发布状态和结果数量限制。
- 使用真实问题评估召回率、延迟、吞吐和每次查询成本。
统一架构的核心不是把所有查询都变成向量搜索，而是让同一条业务记录同时拥有结构化访问能力和语义访问能力。DynamoDB 负责可靠存储与索引，Streams 负责数据变化传播，Bedrock 负责嵌入和 Agent 推理；三者边界清晰时，系统才更容易扩展和排障。