LLM应用开发——多租户架构设计

LLM 应用开发系列

• Embedding 嵌入向量
• 向量数据库
• RAG 检索增强生成
• Prompt Engineering
• LangChain 框架
• Agent 智能体
• Function Calling
• 部署与优化
• 安全防护
• 对话记忆管理
• Streaming 流式输出
• LlamaIndex 框架
• 结构化输出
• 多模态 LLM 应用
• LLM 评估与测试
• 多模型路由
• 缓存策略
• 成本优化
• 监控与可观测性
• 高级 RAG 技术
• GraphRAG 知识图谱
• 长上下文处理
• LLM 微调实战
• 本地模型部署
• MCP 协议开发
• AI 搜索引擎
• 代码生成与执行
• 文档智能处理
• 语音交互系统
• 多租户架构
• Prompt 版本管理
• 混合检索策略
• 对话系统设计
• 内容审核与过滤
• 个性化推荐
• 知识库构建
• AI 写作助手
• 实时协作
• 边缘部署

# 使用 ContextVars 实现租户上下文在异步任务中的传播
from contextvars import ContextVar
from fastapi import Request, Depends

tenant_context: ContextVar[str] = ContextVar("tenant_id", default="public")

async def tenant_middleware(request: Request, call_next):
    # 从 Header 或 JWT 中提取租户 ID
    tenant_id = request.headers.get("X-Tenant-ID", "public")
    token = tenant_context.set(tenant_id)
    try:
        response = await call_next(request)
    finally:
        tenant_context.reset(token)
    return response

# 在 RAG 检索时自动应用过滤
def get_vector_store():
    current_tenant = tenant_context.get()
    return VectorStore(filter={"tenant_id": current_tenant})

import asyncio
from aioredis import Redis

async def track_usage(tenant_id: str, tokens: int):
    """异步记录 Token 消耗"""
    redis = await Redis.from_url("redis://localhost")
    # 使用 Redis 计数器进行实时配额检查
    await redis.incrby(f"usage:{tenant_id}:daily", tokens)
    await redis.expire(f"usage:{tenant_id}:daily", 86400)

---

#### 进阶：多租户微调与 Multi-LoRA 隔离

在高级场景下，不同租户可能需要不同的模型表现。例如，法律行业的租户需要严谨的语气，而创意行业的租户需要活泼的语气。

#### vLLM Multi-LoRA 方案
你不需要为每个租户部署一个完整的模型（那太贵了）。你可以部署一个**基础模型 (Base Model)**，并为每个租户动态加载一个微小的 **LoRA 适配器 (Adapter)**。

- **原理**：vLLM 支持在同一个推理实例中同时加载数百个 LoRA 适配器。
- **隔离**：请求时通过 `request_id` 指定对应的 `lora_name`。

```python
# vLLM 请求示例
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="token")

def call_tenant_model(tenant_id: str, prompt: str):
    # 假设每个租户的 LoRA 适配器名称就是其 tenant_id
    response = client.chat.completions.create(
        model=f"base-model:{tenant_id}", # 动态指定租户适配器
        messages=[{"role": "user", "content": prompt}]
    )
    return response

def get_tenant_prompt(tenant_id: str, base_prompt: str):
    # 从数据库加载租户自定义的约束
    tenant_config = db.get_tenant_config(tenant_id)
    custom_instructions = tenant_config.get("instructions", "")
    
    return f"{base_prompt}\n\n租户特定约束：\n{custom_instructions}"