用 Amazon Bedrock Managed Knowledge Base 构建多租户企业文档问答应用

企业文档问答真正难的部分，不只是让模型“会回答”，而是让它只回答当前用户有权访问的内容。基于 Amazon Bedrock Managed Knowledge Base，可以搭建一种多租户 agentic chat 应用：用户上传文档，系统异步完成索引，随后围绕企业数据进行有依据的问答，同时保持租户和用户之间的数据隔离。

一条完整的数据链路

这类应用通常包含四个阶段：

1. 上传：用户把文档写入 Amazon S3。对象路径中可以包含租户和用户标识，例如 tenants/acme/users/u-123/contracts/a.pdf。
2. 触发索引：应用调用知识库的数据源同步接口，或者由 S3 事件驱动后台任务。索引过程是异步的，上传成功并不代表文档已经可以检索。
3. 检索与生成：聊天请求携带租户上下文，应用先从 Knowledge Base 检索相关片段，再让模型基于这些片段生成答案。
4. 生命周期管理：前端展示 UPLOADED、INDEXING、AVAILABLE 和 FAILED 等状态，避免用户在索引完成前误以为文档丢失。

如果应用进一步引入 agent，agent 可以把“查文档”“调用业务 API”“整理结果”组合成一个工作流。但权限判断仍应由应用和数据层负责，不能把租户隔离寄托在模型提示词上。

异步索引不是一个瞬时动作

文档处理通常包括解析、切分、向量化和写入检索索引。这个过程需要时间，因此上传接口适合快速返回任务标识，而不是同步等待整个流程完成。

可以这样设计状态表：

后台任务应记录 document_id、tenant_id、user_id、S3 对象键、索引任务 ID、重试次数和错误信息。重试需要幂等：同一个文档版本重复触发时，不应产生无法追踪的重复记录。

下面是一个可改造的 Python 示例。它使用假设的应用层接口来展示上传后的异步编排；实际项目中可将 start_index_job 替换为队列生产者，再由 worker 调用 Bedrock Knowledge Base 的同步 API。

# Python 3.11+，示例：把上传事件放入异步索引队列
from dataclasses import dataclass
from datetime import datetime, timezone
from uuid import uuid4

@dataclass(frozen=True)
class UploadEvent:
    document_id: str
    tenant_id: str
    user_id: str
    s3_uri: str
    version: str
    created_at: str


def create_upload_event(tenant_id: str, user_id: str, filename: str) -> UploadEvent:
    document_id = str(uuid4())
    s3_uri = f"s3://enterprise-docs/tenants/{tenant_id}/users/{user_id}/{filename}"
    return UploadEvent(
        document_id=document_id,
        tenant_id=tenant_id,
        user_id=user_id,
        s3_uri=s3_uri,
        version="v1",
        created_at=datetime.now(timezone.utc).isoformat(),
    )


def start_index_job(event: UploadEvent) -> None:
    # 实际实现：写入 SQS/EventBridge，并由 worker 调用 Knowledge Base 同步接口。
    print({
        "status": "INDEXING",
        "document_id": event.document_id,
        "tenant_id": event.tenant_id,
        "s3_uri": event.s3_uri,
    })


if __name__ == "__main__":
    event = create_upload_event("acme", "u-123", "handbook.pdf")
    start_index_job(event)

生产环境中，worker 还应在处理前检查文档所属租户，使用文档版本作为幂等键，并在失败时把可诊断的错误写入日志和状态表。

多租户隔离要落在检索边界

最稳妥的做法是同时使用物理路径隔离、元数据过滤和服务端授权：

• S3 前缀按租户划分，并通过 IAM 限制访问范围。
• 文档元数据写入 tenant_id、owner_id、classification 等字段。
• 每次检索都根据经过认证的身份生成过滤条件，而不是直接信任浏览器传入的 tenant_id。
• 对跨用户共享文档，使用显式的 ACL 或共享组模型，不要用“没有过滤条件”表示共享全部数据。
• 在回答中保留检索片段的来源信息，便于审计和用户复核。

一个检索请求可以这样表达租户过滤逻辑。字段名需要根据实际 Knowledge Base 配置调整：

{
  "retrievalQuery": "公司差旅报销上限是多少？",
  "filter": {
    "equals": {
      "key": "tenant_id",
      "value": "acme"
    }
  },
  "sessionContext": {
    "user_id": "u-123",
    "groups": ["finance"]
  }
}

这里的 tenant_id 应来自服务端验证后的身份令牌或会话，而不是来自用户可以任意修改的请求字段。对于高敏感文档，还可以在检索前做二次授权检查，并将“允许访问的文档集合”转换为更严格的过滤条件。

Agentic chat 的边界与运行方式

一个实用的 agent 工作流可以分为三步：

1. 判断问题是否需要企业文档。
2. 使用带权限过滤的 Knowledge Base 检索相关内容。
3. 仅依据检索结果回答；证据不足时明确说明，而不是补写事实。

如果问题还需要订单、工单或员工系统的数据，agent 可以调用受控工具。工具需要定义输入校验、超时、权限和审计日志，尤其要防止模型把用户问题中的文字当成系统授权指令。

规模化运行时，重点关注以下指标：

• 文档从上传到 AVAILABLE 的端到端延迟。
• 索引失败率、重试次数和死信队列积压。
• 检索延迟、模型响应延迟和每次问答成本。
• 按租户统计的存储、索引和调用量，避免单个大租户影响其他租户。
• 无过滤检索、跨租户访问拒绝和异常下载等安全事件。

还应设置上传大小、文件类型、并发索引任务和单租户配额。对于经常更新的文档，使用版本号或内容哈希区分旧版本，避免用户在索引尚未完成时读到混合版本。

落地检查清单

上线前可以逐项确认：

• 上传接口是否只返回接收结果，并提供可查询的索引状态？
• 失败任务是否能够安全重试，并且具备幂等性？
• 每次检索是否强制带上服务端生成的租户过滤条件？
• S3、元数据、日志和检索结果中是否避免泄露不必要的敏感信息？
• 回答是否展示来源，且在没有证据时拒答或降级？
• 是否有按租户的配额、成本和延迟监控？

Managed Knowledge Base 可以减少文档解析、向量化和检索基础设施的维护工作，但它不会自动解决身份、授权、异步状态和运营问题。把这些边界放在应用架构中，多租户企业问答才会从演示功能变成可持续运行的服务。