企业级私有化 AI Agent 多租户隔离架构实战：组织架构树对齐、模型算力配额划分与向量知识库租户级安全隔离

【集团多级租户上下文三要素规范】

• 1. 租户路径编码（Tenant Path / Materialized Path）：
  采用物化路径算法表示租户层级，如 root.energy_corp.sub_factory_01。在查询权限与知识继承时，利用前缀匹配（path LIKE 'root.energy_corp%'）即可实现向下继承与向上溯源，无需递归查询树状结构。
• 2. 客户端不可见的服务端鉴权注入：
  租户身份严禁由前端 HTTP Header 或用户自然语言传入。必须由企业统一身份网关（OAuth2 / OIDC / CAS）在校验集团 LDAP/AD 凭据后，将已验证的 tenant_id 与 org_path 签名写入不可变认证主体（Claims）。
• 3. 协程与上下文穿透（Context Propagation）：
  在 Python/Node.js 运行时中，利用 contextvars.ContextVar 或 AsyncLocalStorage 封装租户上下文对象。无论调度链跨越多少层异步回调，底层的向量查询器、工具执行器和日志器均直接读取当前协程的不可变租户属性。

# 文件位置：bxagent/core/middleware/tenant.py
from contextvars import ContextVar
from dataclasses import dataclass
from typing import Optional, List

@dataclass(frozen=True)
class TenantContext:
    # 租户唯一标识，如 "corp_fin_001"
    tenant_id: str
    # 组织架构树物化路径，如 "root/finance/corp_fin_001"
    org_path: str
    # 该租户获准访问的知识库命名空间列表
    allowed_namespaces: List[str]
    # 算力配额组等级：STANDARD, VIP, CRITICAL
    tier: str = "STANDARD"

# 协程隔离的安全上下文容器
_current_tenant_ctx: ContextVar[Optional[TenantContext]] = ContextVar("current_tenant", default=None)

class TenantContextManager:
    @classmethod
    def set_context(cls, tenant: TenantContext):
        return _current_tenant_ctx.set(tenant)

    @classmethod
    def get_current(cls) -> TenantContext:
        ctx = _current_tenant_ctx.get()
        if ctx is None:
            raise PermissionError("CRITICAL: 缺失合法的租户上下文，拒绝执行任何 Agent 动作！")
        return ctx

【租户级 GPU 算力治理三级流控矩阵】

• 1. 并发请求槽位（Max In-flight Concurrency Slots）：
  限制特定租户同时处于 LLM 推理执行阶段的请求数。例如：VIP 租户保障 20 个并发槽位，普通租户限制 5 个。超出槽位直接进入该租户专属排队队列，不挤占全局 GPU 显存。
• 2. 速率令牌桶（Token-per-Minute Rate Limiting）：
  基于漏桶/令牌桶算法（Leaky Bucket），按租户维度监控每分钟生成的 Prompt + Completion Token 总数。避免单个租户高频发送超长上下文请求。
• 3. 优雅降级与借调策略（Quota Borrowing & Degradation）：
  当全局集群空闲度高于 70% 时，允许普通租户临时“超配借调（Burst Borrowing）”算力；一旦核心业务或高优先级租户流量上升，系统在下一个推理迭代步强制回收借调资源，保障高优先级 SLA。

# 文件位置：bxagent/core/security/tool_guard.py
from functools import wraps
from bxagent.core.middleware.tenant import TenantContextManager

def enforce_tenant_scope(func):
    """装饰器：强制绑定当前协程安全上下文中的租户 ID，覆写或校验模型参数"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        current_tenant = TenantContextManager.get_current()

        # 防御场景：恶意 Prompt 诱导 LLM 生成了 kwargs["target_tenant_id"] = "corp_audit_secret"
        if "tenant_id" in kwargs:
            target_tenant = kwargs["tenant_id"]
            if target_tenant != current_tenant.tenant_id:
                # 记录严重安全审计违规告警
                audit_log_security_violation(
                    current_tenant=current_tenant.tenant_id,
                    attempted_tenant=target_tenant,
                    tool=func.__name__
                )
                raise PermissionError(
                    f"安全阻断：当前租户 {current_tenant.tenant_id} 严禁越权操作租户 {target_tenant} 的业务系统！"
                )

        # 强制将合法租户凭据绑定到底层数据库或第三方 API Client
        kwargs["verified_tenant_ctx"] = current_tenant
        return await func(*args, **kwargs)
    return wrapper

【多租户审计与计量日志标准字段（JSON Schema）】

• tenant_id 与 org_path：记录发起本次交互的租户全局编码及组织层级路径。
• trace_id 与 span_id：符合 OpenTelemetry 分布式链路标准，串联从网关、LLM 推理、RAG 检索到工具调用的全生命周期。
• resource_metrics：精确记录本次交互消耗的 Prompt Tokens、Completion Tokens、Cached Tokens 以及各推理阶段首字延迟（TTFT）。
• audit_events：敏感知识召回命中记录（仅记录文档哈希与切片 ID，脱敏敏感明文内容）及外部工具调用输入输出指纹。