企业级 AI 中间件架构设计

企业AI服务化的痛点

随着AI应用在企业中的普及，一个典型场景是：市场部接入了GPT-4做文案生成，研发部在用DeepSeek做代码补全，客服系统使用了自训练的意图识别模型。这些AI服务各自为政，缺乏统一管理——API Key散落各处、调用量无法监控、成本难以分摊、模型切换需要改代码。AI中间件的目标就是解决这些碎片化问题，构建一个统一的AI能力接入层。

AI网关核心能力

AI中间件的核心是AI网关，它应具备以下能力：统一API（所有模型通过统一的OpenAI兼容接口访问，上游应用无需关心后端模型）、智能路由（根据请求类型自动选择最优模型——简单问答走廉价模型，复杂推理走高级模型）、限流与配额（按租户/应用进行速率限制和Token配额管理）、协议转换（将OpenAI格式的请求转换为不同供应商的原生格式）、缓存与降级（高频请求缓存、模型不可用时降级到备用模型）。

AI网关核心实现

import time, hashlib, json
from collections import defaultdict
from openai import OpenAI

class AIGateway:
    def __init__(self):
        self.models = {}           # 模型注册表
        self.rate_limits = {}      # 租户限流配置
        self.usage = defaultdict(list)  # 使用统计
        self.cache = {}

    def register_model(self, name, provider, model_id, api_key, base_url, cost_per_1k):
        """注册模型"""
        self.models[name] = {
            "provider": provider, "model_id": model_id,
            "client": OpenAI(api_key=api_key, base_url=base_url),
            "cost_per_1k": cost_per_1k
        }

    def set_rate_limit(self, tenant_id, rpm, tpm):
        """设置租户限流（每分钟请求数、每分钟Token数）"""
        self.rate_limits[tenant_id] = {"rpm": rpm, "tpm": tpm, "requests": [], "tokens": 0}

    def _check_limit(self, tenant_id):
        if tenant_id not in self.rate_limits:
            return True
        lim = self.rate_limits[tenant_id]
        now = time.time()
        lim["requests"] = [t for t in lim["requests"] if now - t < 60]
        if len(lim["requests"]) >= lim["rpm"]:
            return False
        return True

    def route(self, messages, tenant_id, max_cost=None):
        """智能路由——根据任务复杂度选择模型"""
        msg_text = json.dumps(messages, ensure_ascii=False)
        if len(msg_text) < 200:
            preferred = ["deepseek-chat", "gpt-3.5-turbo"]  # 简单任务用廉价模型
        else:
            preferred = ["gpt-4", "deepseek-chat"]  # 复杂任务用高级模型

        for model in preferred:
            if model in self.models and self._check_limit(tenant_id):
                return model
        return list(self.models.keys())[0]  # 回退

    def chat(self, tenant_id, model, messages, **kwargs):
        """执行带限流和缓存的LLM调用"""
        if not self._check_limit(tenant_id):
            return {"error": "rate limited", "retry_after": 60}

        # 缓存检查
        cache_key = hashlib.md5(
            json.dumps({"model":model,"msgs":messages},sort_keys=True).encode()
        ).hexdigest()
        if cache_key in self.cache:
            return self.cache[cache_key]

        m = self.models[model]
        resp = m["client"].chat.completions.create(
            model=m["model_id"], messages=messages, **kwargs
        )
        result = resp.choices[0].message.content

        # 更新统计
        self.rate_limits.setdefault(tenant_id,
            {"rpm":9999,"tpm":999999,"requests":[],"tokens":0}
        )["requests"].append(time.time())

        self.cache[cache_key] = result
        return result

    def cost_report(self, tenant_id):
        """生成成本报告"""
        usage = self.usage.get(tenant_id, [])
        return {"total_calls": len(usage), "estimated_cost": sum(u.get("cost",0) for u in usage)}

gw = AIGateway()
gw.register_model("deepseek-chat", "deepseek", "deepseek-chat",
                  "sk-xxx", "https://api.deepseek.com", 0.001)
gw.set_rate_limit("tenant-1", rpm=100, tpm=100000)
print(gw.chat("tenant-1", "deepseek-chat", [{"role":"user","content":"Hello"}]))

生产级部署考量

AI网关在生产环境中需要额外关注：高可用（多实例部署+健康检查+自动故障切换）、请求队列（突发流量时排队而非直接拒绝，设置队列长度和超时）、流式响应（支持SSE透传实现打字机效果，需处理流中断和重连）、日志与审计（记录所有请求的完整信息用于成本核算和问题排查，注意脱敏处理用户数据）、多区域部署（就近路由减少延迟，跨区域故障切换）。

开源AI网关方案对比

如果你不想从零构建AI网关，以下开源方案值得关注：LiteLLM——最流行的AI网关代理，支持100+模型提供商的统一OpenAI格式接口，内置负载均衡、速率限制和花费追踪。Portkey——增强版AI网关，额外提供Prompt管理、A/B测试、缓存和金丝雀发布。Helicone——侧重可观测性的AI网关，提供请求日志、成本分析和用量仪表盘。三者的选择建议：只需统一API和基础限流→LiteLLM；需要完整的LLMOps能力→Portkey；核心需求是可观测性和成本分析→Helicone。我们最终选择了LiteLLM+自研路由模块的组合——LiteLLM处理协议统一和基础能力，自研模块负责复杂的多模型路由和降级策略。