用 Amazon Bedrock AgentCore 构建多租户智能体：隔离、配置与落地实践

# tenant_config_tenant_a.yaml
tenant_id: "tenant-a"
agent:
  name: "CustomerServiceAgent-A"
  foundation_model: "anthropic.claude-3-5-sonnet"
  instruction: |
    你是 tenant-a 的客服助手。只回答 tenant-a 相关的产品问题。
    查询订单时，只返回 tenant-a 的订单数据。
    如果用户问到其他租户的信息，明确拒绝并说明原因。

tools:
  - name: "QueryOrders"
    description: "查询 tenant-a 的订单状态"
    # 工具执行时自动注入 tenant_id 作为查询过滤条件
    execution_params:
      tenant_filter: "tenant-a"
      max_results: 10
  - name: "SearchProductFAQ"
    description: "搜索 tenant-a 的产品常见问题"
    knowledge_base_id: "kb-tenant-a-products"

memory:
  store_id: "memory-tenant-a"
  session_ttl: 3600  # 1小时会话超时

rate_limit:
  max_invocations_per_minute: 30
  max_token_per_invocation: 4096

# tenant_config_tenant_b.yaml
tenant_id: "tenant-b"
agent:
  name: "CustomerServiceAgent-B"
  foundation_model: "anthropic.claude-3-5-sonnet"
  instruction: |
    你是 tenant-b 的客服助手。只处理 tenant-b 的退货和退款问题。
    不提供订单查询功能。

tools:
  - name: "ProcessRefund"
    description: "处理 tenant-b 的退款申请"
    execution_params:
      tenant_filter: "tenant-b"
      require_approval: true
  - name: "SearchRefundPolicy"
    description: "搜索 tenant-b 的退款政策"
    knowledge_base_id: "kb-tenant-b-refund"

memory:
  store_id: "memory-tenant-b"
  session_ttl: 7200  # 2小时会话超时

rate_limit:
  max_invocations_per_minute: 15
  max_token_per_invocation: 2048

import boto3
import json
from bedrock_agentcore import AgentRuntimeClient, TenantContext

# 初始化 AgentCore 客户端
client = AgentRuntimeClient(
    region="us-east-1",
    # 可选：指定统一的 IAM role，AgentCore 会按租户自动切换权限
    execution_role="arn:aws:iam::123456789012:role/AgentCoreExecutionRole"
)

def create_tenant_session(tenant_id: str, user_id: str) -> str:
    """为指定租户创建隔离的 Agent 会话"""

    # 加载租户配置（实际项目中可从 DynamoDB/S3 动态加载）
    config = load_tenant_config(tenant_id)

    # 创建租户上下文——这是隔离的关键
    tenant_context = TenantContext(
        tenant_id=tenant_id,
        user_id=user_id,
        # 租户级限流配置
        rate_limits=config["rate_limit"],
        # 租户级工具权限
        allowed_tools=[tool["name"] for tool in config["tools"]],
        # 租户级知识库绑定
        knowledge_base_ids=[
            tool["knowledge_base_id"]
            for tool in config["tools"]
            if "knowledge_base_id" in tool
        ],
    )

    # 创建 Agent 会话，自动注入租户上下文
    session = client.create_session(
        agent_name=config["agent"]["name"],
        model_id=config["agent"]["foundation_model"],
        instruction=config["agent"]["instruction"],
        tenant_context=tenant_context,
        memory_store_id=config["memory"]["store_id"],
        session_ttl=config["memory"]["session_ttl"],
    )

    return session.session_id


def invoke_agent(session_id: str, tenant_id: str, user_input: str) -> str:
    """在租户隔离会话中调用 Agent"""

    response = client.invoke(
        session_id=session_id,
        input_text=user_input,
        # 每次调用都携带 tenant_id，确保 Runtime 在工具调用和
        # 知识库检索时自动注入租户过滤
        tenant_id=tenant_id,
    )

    # 返回 Agent 的回复（已自动过滤跨租户数据）
    return response.output_text


# ---- 使用示例 ----
# tenant-a 的用户发起客服对话
session_a = create_tenant_session("tenant-a", "user-001")
reply = invoke_agent(session_a, "tenant-a", "我的订单 A-1234 现在什么状态？")
print(f"Agent 回复: {reply}")
# Agent 会调用 QueryOrders 工具，自动过滤 tenant_id=tenant-a

# tenant-b 的用户发起退款对话
session_b = create_tenant_session("tenant-b", "user-002")
reply = invoke_agent(session_b, "tenant-b", "我想退款订单 B-5678")
print(f"Agent 回复: {reply}")
# Agent 会调用 ProcessRefund 工具，自动过滤 tenant_id=tenant-b
# 注意：tenant-b 的 Agent 没有 QueryOrders 工具，无法查订单

# AgentCore 在内部自动执行类似以下逻辑
# 开发者通常不需要手动写这段代码，但理解它有助于排查隔离问题

import boto3

bedrock_kb = boto3.client("bedrock-agent-runtime")

def search_with_tenant_filter(kb_id: str, tenant_id: str, query: str):
    """按租户过滤的知识库检索"""

    response = bedrock_kb.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                # 关键：metadata filter 确保只返回当前租户的文档
                "filter": {
                    "andAll": [
                        {
                            "equals": {
                                "key": "tenant_id",
                                "value": tenant_id
                            }
                        }
                    ]
                },
                "numberOfResults": 5
            }
        }
    )

    return response["retrievalResults"]


# 验证隔离效果
results_a = search_with_tenant_filter("kb-tenant-a-products", "tenant-a", "退货政策")
results_b = search_with_tenant_filter("kb-tenant-a-products", "tenant-b", "退货政策")
# results_b 应返回空列表——tenant-b 无权访问 tenant-a 的知识库