FastAPI多模型调度实战：构建企业级混合LLM网关
混合模型架构设计挑战与解决方案
在当今企业AI应用中，单一模型服务已无法满足复杂业务需求。混合使用云端LLM（如OpenAI）和本地部署模型（如ChatGLM、Llama2）成为技术新趋势，但这带来了接口不统一、流量分配不均、监控困难等工程挑战。
我们设计的多模型调度系统核心架构包含以下组件：
- 模型抽象层：通过工厂模式封装不同厂商API差异
- 流量控制层：基于Redis的分布式限流算法
- 动态路由层：支持运行时模型热切换
- 监控分析层：多维性能指标采集与可视化
# 架构示意图核心类 class ModelGateway: def __init__(self): self.model_factory = ModelFactory() self.rate_limiter = RedisRateLimiter() self.monitor = PrometheusMonitor() class ModelFactory: def get_model(self, model_type: str) -> BaseModel: if model_type == "openai": return OpenAIModel() elif model_type == "chatglm": return ChatGLMAdapter()统一接口工厂模式实现
工厂模式是解决多模型接口差异的关键设计模式。我们通过抽象基类定义统一接口，各模型实现具体逻辑。
基础抽象接口设计
from abc import ABC, abstractmethod class BaseModel(ABC): @abstractmethod async def generate(self, prompt: str, **kwargs) -> str: pass @abstractmethod def get_cost(self) -> float: pass具体模型实现示例
OpenAI模型适配器：
class OpenAIModel(BaseModel): def __init__(self, api_key: str): self.client = AsyncOpenAI(api_key=api_key) async def generate(self, prompt: str, **kwargs) -> str: response = await self.client.chat.completions.create( model=kwargs.get("model", "gpt-3.5-turbo"), messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content def get_cost(self) -> float: return 0.002 # 每千token成本本地模型适配器：
class ChatGLMAdapter(BaseModel): def __init__(self, endpoint: str): self.endpoint = endpoint async def generate(self, prompt: str, **kwargs) -> str: async with httpx.AsyncClient() as client: response = await client.post( f"{self.endpoint}/generate", json={"prompt": prompt}, timeout=30.0 ) return response.json()["answer"] def get_cost(self) -> float: return 0 # 本地模型无API调用费用工厂类实现
class ModelFactory: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._models = {} return cls._instance def register_model(self, name: str, model: BaseModel): self._models[name] = model def get_model(self, name: str) -> BaseModel: if name not in self._models: raise ValueError(f"未知模型: {name}") return self._models[name]分布式流量控制方案
在混合模型场景下，不同模型的API调用成本、性能特征差异显著，需要精细化的流量控制策略。
Redis限流器实现
import redis from fastapi import HTTPException class RedisRateLimiter: def __init__(self, redis_url: str): self.redis = redis.from_url(redis_url) async def check_limit(self, key: str, limit: int, window: int = 60): current = self.redis.get(key) or 0 if int(current) >= limit: raise HTTPException( status_code=429, detail=f"请求过于频繁，限制 {limit}次/{window}秒" ) pipe = self.redis.pipeline() pipe.incr(key) pipe.expire(key, window) pipe.execute()分层限流策略配置
| 层级 | 限制对象 | OpenAI限制 | 本地模型限制 | 目的 |
|---|---|---|---|---|
| 全局 | IP地址 | 100次/分钟 | 500次/分钟 | 防止滥用 |
| 用户 | API Key | 50次/分钟 | 200次/分钟 | 商业分级 |
| 模型 | 模型类型 | 20次/秒 | 100次/秒 | 负载均衡 |
# 在FastAPI路由中应用限流 @app.post("/generate") @limiter.limit("100/minute") async def generate_text( request: Request, model: str = "openai", prompt: str = Body(...) ): # 获取用户身份信息 user_token = request.headers.get("Authorization") # 分层限流检查 rate_limiter.check_limit(f"ip:{request.client.host}", 100) rate_limiter.check_limit(f"user:{user_token}", 50) rate_limiter.check_limit(f"model:{model}", 20 if model == "openai" else 100) # 处理请求...模型热切换与故障转移
生产环境需要在不重启服务的情况下切换模型配置，并具备自动故障转移能力。
动态配置加载
import yaml from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class ModelConfigHandler(FileSystemEventHandler): def on_modified(self, event): if event.src_path.endswith("models.yaml"): self.reload_models() def reload_models(self): with open("config/models.yaml") as f: config = yaml.safe_load(f) factory = ModelFactory() for name, params in config.items(): if params["type"] == "openai": factory.register_model(name, OpenAIModel(params["api_key"])) elif params["type"] == "chatglm": factory.register_model(name, ChatGLMAdapter(params["endpoint"]))故障转移策略实现
class FallbackModel(BaseModel): def __init__(self, primary: BaseModel, fallback: BaseModel): self.primary = primary self.fallback = fallback async def generate(self, prompt: str, **kwargs) -> str: try: return await self.primary.generate(prompt, **kwargs) except Exception as e: logging.warning(f"主模型失败: {e}, 尝试备用模型") return await self.fallback.generate(prompt, **kwargs)性能监控与智能路由
完善的监控系统是优化模型调度的基础，我们需要采集多维度的性能指标。
监控指标设计
核心监控维度：
- 响应时间分布（P50/P95/P99）
- 错误率与异常类型
- Token消耗与成本统计
- 并发请求量趋势
- 模型健康状态
from prometheus_client import Counter, Histogram class PrometheusMonitor: def __init__(self): self.request_count = Counter( 'model_requests_total', 'Total API requests', ['model', 'status'] ) self.latency = Histogram( 'model_request_latency_seconds', 'Request latency distribution', ['model'], buckets=[0.1, 0.5, 1, 2, 5, 10] ) def record_request(self, model: str, status: str, latency: float): self.request_count.labels(model, status).inc() self.latency.labels(model).observe(latency)智能路由算法
基于实时监控数据，我们可以实现动态路由策略：
def select_model(prompt: str) -> str: # 获取各模型当前状态 models = monitor.get_model_stats() # 规则引擎决策 if len(prompt) > 1000: return "gpt-4-turbo" # 长文本使用更强模型 elif models["gpt-4"]["latency"] < 2.0: return "gpt-4" if random.random() < 0.7 else "gpt-3.5-turbo" else: return "chatglm-pro" # OpenAI延迟高时降级部署架构与性能优化
生产级部署需要考虑高可用和性能扩展，以下是推荐架构：
组件部署方案：
graph TD A[客户端] --> B[负载均衡器] B --> C[FastAPI实例1] B --> D[FastAPI实例2] C --> E[Redis集群] D --> E E --> F[OpenAI API] E --> G[本地模型集群] C --> H[Prometheus] D --> H性能优化技巧：
连接池管理：
import httpx # 全局复用客户端实例 async_client = httpx.AsyncClient( limits=httpx.Limits( max_connections=100, max_keepalive_connections=20 ), timeout=30.0 )结果缓存：
from redis import Redis from hashlib import md5 def get_cache_key(prompt: str, model: str) -> str: return f"cache:{model}:{md5(prompt.encode()).hexdigest()}" async def cached_generate(prompt: str, model: str): cache_key = get_cache_key(prompt, model) cached = redis.get(cache_key) if cached: return cached result = await model.generate(prompt) redis.setex(cache_key, 3600, result) # 缓存1小时 return result批量处理优化：
async def batch_generate(prompts: List[str], model: str): if model.startswith("openai"): # 利用OpenAI的批处理API return await openai_batch_api(prompts) else: # 本地模型并行请求 return await asyncio.gather( *[local_model.generate(p) for p in prompts] )
在实际项目中，我们通过这种架构成功将端到端延迟从平均1.2秒降低到400毫秒，同时将成本节约了40%。关键在于持续监控和动态调整模型调度策略，而不是简单地选择最强或最便宜的模型。
