Agent 多租户隔离设计：不同客户的任务别混在同一个上下文里

一、"客户 A 的 Agent 调出了客户 B 的数据"

SaaS 平台的 AI 助手上线后，客户 A 的售后客服问 Agent"查看最近订单"，Agent 返回了客户 B 的订单明细。排查发现 Agent 的知识库检索范围是全库而非租户级别的——不同客户的数据混在了同一个检索空间里。

Agent 多租户的本质不只是数据隔离，还有上下文隔离、任务隔离和成本隔离。租户 A 的对话上下文不能污染租户 B、租户 A 的任务失败不影响租户 B、租户 A 的成本需要独立核算。

二、多租户隔离的四个维度

flowchart TD
    subgraph Data[数据隔离]
        D1[向量库: 按 TenantID 分区]
        D2[数据库: Row-Level Security]
        D3[缓存: 按 TenantID 做 Key 前缀]
    end
    
    subgraph Context[上下文隔离]
        C1[会话上下文绑定 TenantID]
        C2[Agent 记忆按租户隔离]
        C3[工具调用权限按租户配置]
    end
    
    subgraph Task[任务隔离]
        T1[任务队列: 按租户的独立队列?]
        T2[并发限制: 每租户最大并发数]
        T3[失败重试: 租户粒度而非全局]
    end
    
    subgraph Cost[成本隔离]
        CO1[Token 使用量按租户归集]
        CO2[模型调用次数独立统计]
        CO3[成本账单按租户生成]
    end
    
    API[API 请求] --> Auth[认证: 提取 TenantID]
    Auth --> C1
    Auth --> D1
    Auth --> CO1

多租户隔离的核心原则：所有数据操作必须携带 TenantID，所有任务调度都要考虑租户的资源配额，所有成本统计都要归属到具体租户。

三、Go 实现：多租户 Agent 中间件

package multitenant

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// ========== 租户上下文 ==========

type contextKey string

const (
    KeyTenantID  contextKey = "tenant_id"
    KeyTenantTier contextKey = "tenant_tier" // free, pro, enterprise
)

// TenantInfo 租户信息
type TenantInfo struct {
    ID            string
    Name          string
    Tier          string    // free, pro, enterprise
    MaxConcurrent int       // 最大并发任务数
    MaxTokensPerDay int64   // 每日 Token 配额
    AllowedTools  []string  // 允许使用的工具列表
    RateLimitQPS  int       // 租户级别 QPS 限制
}

// WithTenant 给 context 注入租户信息
func WithTenant(ctx context.Context, info *TenantInfo) context.Context {
    ctx = context.WithValue(ctx, KeyTenantID, info.ID)
    ctx = context.WithValue(ctx, KeyTenantTier, info.Tier)
    return ctx
}

// TenantIDFromContext 从 context 中获取租户 ID
func TenantIDFromContext(ctx context.Context) (string, bool) {
    id, ok := ctx.Value(KeyTenantID).(string)
    return id, ok
}

// ========== 租户管理器 ==========

// TenantManager 租户管理器
type TenantManager struct {
    tenants map[string]*TenantInfo
    mu      sync.RWMutex
}

func NewTenantManager() *TenantManager {
    return &TenantManager{
        tenants: make(map[string]*TenantInfo),
    }
}

func (tm *TenantManager) Register(info *TenantInfo) {
    tm.mu.Lock()
    defer tm.mu.Unlock()
    tm.tenants[info.ID] = info
}

// ========== 数据隔离中间件 ==========

// TenantDataFilter 数据层租户过滤器
type TenantDataFilter struct {
    db *mockDB // 实际项目中是真实的数据库连接
}

// QueryDocuments 查询文档时自动注入 TenantID 过滤
func (tf *TenantDataFilter) QueryDocuments(ctx context.Context, query string) ([]string, error) {
    tenantID, ok := TenantIDFromContext(ctx)
    if !ok {
        return nil, fmt.Errorf("未找到租户 ID，拒绝查询")
    }
    
    // 构造带租户过滤的查询
    filter := map[string]string{
        "tenant_id": tenantID,
        "query":     query,
    }
    
    // 实际查询
    _ = filter
    return nil, nil
}

// ========== 上下文隔离 ==========

// TenantContextStore 租户级别的上下文存储
type TenantContextStore struct {
    sessions map[string]map[string]interface{} // tenantID → sessionID → context
    mu       sync.RWMutex
}

func NewTenantContextStore() *TenantContextStore {
    return &TenantContextStore{
        sessions: make(map[string]map[string]interface{}),
    }
}

// GetContext 获取租户的会话上下文
func (cs *TenantContextStore) GetContext(
    ctx context.Context, sessionID string,
) (map[string]interface{}, error) {
    tenantID, ok := TenantIDFromContext(ctx)
    if !ok {
        return nil, fmt.Errorf("未找到租户 ID")
    }
    
    cs.mu.RLock()
    defer cs.mu.RUnlock()
    
    if sessions, ok := cs.sessions[tenantID]; ok {
        if sessionData, ok := sessions[sessionID]; ok {
            return sessionData.(map[string]interface{}), nil
        }
    }
    
    return make(map[string]interface{}), nil
}

// SaveContext 保存租户的会话上下文
func (cs *TenantContextStore) SaveContext(
    ctx context.Context, sessionID string, data map[string]interface{},
) error {
    tenantID, ok := TenantIDFromContext(ctx)
    if !ok {
        return fmt.Errorf("未找到租户 ID")
    }
    
    cs.mu.Lock()
    defer cs.mu.Unlock()
    
    if _, ok := cs.sessions[tenantID]; !ok {
        cs.sessions[tenantID] = make(map[string]interface{})
    }
    
    cs.sessions[tenantID][sessionID] = data
    return nil
}

// ========== 并发限制中间件 ==========

// TenantRateLimiter 租户级别并发限制
type TenantRateLimiter struct {
    tokens map[string]chan struct{} // 令牌桶
    mu     sync.Mutex
}

func NewTenantRateLimiter() *TenantRateLimiter {
    return &TenantRateLimiter{
        tokens: make(map[string]chan struct{}),
    }
}

// Acquire 获取租户的执行槽位
func (rl *TenantRateLimiter) Acquire(ctx context.Context, tenantID string, maxConcurrent int) error {
    rl.mu.Lock()
    ch, ok := rl.tokens[tenantID]
    if !ok {
        ch = make(chan struct{}, maxConcurrent)
        rl.tokens[tenantID] = ch
    }
    rl.mu.Unlock()
    
    select {
    case ch <- struct{}{}:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

// Release 释放租户的执行槽位
func (rl *TenantRateLimiter) Release(tenantID string) {
    rl.mu.Lock()
    defer rl.mu.Unlock()
    
    if ch, ok := rl.tokens[tenantID]; ok {
        <-ch
    }
}

// ========== 成本隔离 ==========

// TenantCostTracker 租户成本追踪
type TenantCostTracker struct {
    usage map[string]*TenantUsage // 租户 ID → 用量
    mu    sync.RWMutex
}

type TenantUsage struct {
    TenantID     string
    TotalTokens   int64
    TotalCost     float64
    DailyTokens   int64
    DailyCost     float64
    DailyDate     string
    MonthlyTokens int64
    MonthlyCost   float64
    MonthlyKey    string
}

func NewTenantCostTracker() *TenantCostTracker {
    return &TenantCostTracker{
        usage: make(map[string]*TenantUsage),
    }
}

// RecordUsage 记录租户的 Token 使用量
func (ct *TenantCostTracker) RecordUsage(tenantID string, inputTokens, outputTokens int64, cost float64) error {
    ct.mu.Lock()
    defer ct.mu.Unlock()
    
    if _, ok := ct.usage[tenantID]; !ok {
        ct.usage[tenantID] = &TenantUsage{TenantID: tenantID}
    }
    
    usage := ct.usage[tenantID]
    usage.TotalTokens += inputTokens + outputTokens
    usage.TotalCost += cost
    
    // 每日统计
    today := time.Now().Format("2006-01-02")
    if usage.DailyDate != today {
        usage.DailyDate = today
        usage.DailyTokens = 0
        usage.DailyCost = 0
    }
    usage.DailyTokens += inputTokens + outputTokens
    usage.DailyCost += cost
    
    return nil
}

// GenerateBill 生成租户账单
func (ct *TenantCostTracker) GenerateBill(tenantID string, startTime, endTime time.Time) *TenantBill {
    // 实际项目中从数据库查询区间用量
    return &TenantBill{
        TenantID:    tenantID,
        TotalCost:   0,
        ModelBreakdown: make(map[string]float64),
    }
}

type TenantBill struct {
    TenantID       string
    TotalCost      float64
    ModelBreakdown map[string]float64 // 各模型的费用明细
}

// ========== 工具权限控制 ==========

// ToolPermissionChecker 工具权限检查
type ToolPermissionChecker struct {
    tenantTools map[string][]string // 租户 ID → 允许的工具列表
    mu          sync.RWMutex
}

func (tc *ToolPermissionChecker) CanUseTool(ctx context.Context, toolName string) bool {
    tenantID, ok := TenantIDFromContext(ctx)
    if !ok {
        return false
    }
    
    tc.mu.RLock()
    defer tc.mu.RUnlock()
    
    allowed, ok := tc.tenantTools[tenantID]
    if !ok {
        return false
    }
    
    for _, t := range allowed {
        if t == toolName || t == "*" {
            return true
        }
    }
    
    return false
}

// ========== 接口抽象 ==========

type mockDB struct{}

// ========== 使用示例 ==========

func ProcessTenantRequest(ctx context.Context, tenantID, message string) (string, error) {
    // 1. 加载租户配置
    tenantInfo := &TenantInfo{
        ID:            tenantID,
        Tier:          "pro",
        MaxConcurrent: 5,
        AllowedTools:  []string{"query_order", "check_status"},
        RateLimitQPS:  10,
    }
    
    // 2. 注入租户上下文
    ctx = WithTenant(ctx, tenantInfo)
    
    // 3. 后续所有操作都自动携带租户 ID
    // 数据查询 → 自动过滤
    // 上下文存储 → 自动隔离
    // 成本记录 → 自动归集
    
    _ = ctx
    _ = message
    
    return "处理完成", nil
}

四、多租户隔离的边界与陷阱

不要用数据库字段隔离代替真正的主键隔离。 在 SQL 中加 WHERE tenant_id = ? 是对的，但不能只靠这一层。如果开发忘了加这个条件，数据就泄露了。更安全的做法是数据库层面的 Row-Level Security 策略。

共享知识的租户特殊处理。 有些场景下多个租户共享同一套知识库（如同一个行业的多租户共享法规知识）。这时需要区分"共享知识"和"私有数据"，共享部分不需要租户过滤，私有数据严格要求。

租户级别的资源配额要留缓冲。 如果 10 个租户共享一个 Agent 集群，每个租户分到 10% 的资源配额，总共 100%。实际运行中资源利用率很少达到 100%，总会有一个租户的配额浪费。用弹性配额而非硬配额——未使用的配额可以被其他租户临时借用。

隔离带来的运维成本。 如果每个租户有独立的向量库索引、独立的 Agent 实例，隔离性最好但运维成本也最高。大多数场景下，轻量隔离（数据级 + 上下文级）已经足够，不需要做到实例级隔离。

五、总结

多租户隔离的四个核心：数据操作必须带 TenantID、上下文存储按租户分区、任务配额按租户限制、成本统计按租户独立。实施策略：先用 Context 传递 TenantID 的统一入口模式，再逐步加固数据层和资源层的隔离。隔离做得越早代价越小——等到数据混了再拆分，比脏了再洗难十倍。