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

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 的统一入口模式再逐步加固数据层和资源层的隔离。隔离做得越早代价越小——等到数据混了再拆分比脏了再洗难十倍。