AI网关：面向多模型时代的轻量级中间层实践

1. 为什么今天聊AI网关，不是讲技术参数，而是讲“你家应用门口那道门”

AI网关这个词最近在技术群里刷屏，但很多人点开文章一看——满屏是Kubernetes、OpenAPI、JWT鉴权、负载均衡策略……立刻关掉。不是不想学，是根本没搞清：我写个调用Qwen的Flask接口，或者用Streamlit搭个内部工具，凭什么要多加一层？这层到底挡谁？放谁？又凭什么让我改代码、加配置、扛运维？

我去年帮三家中小团队落地AI能力，从零开始做智能客服、文档摘要、图像审核。最早的做法很朴素：前端直接调后端，后端硬编码写死模型URL，比如requests.post("https://api.qwen.com/v1/chat", json=payload)。结果呢？三个月后，业务方说“能不能把回答风格调得更正式一点”，技术一查——得换模型；半年后，法务说“所有图片必须本地处理，不能外传”，得切到本地部署的SDXL；又过两个月，销售说“客户想自己选模型”，我们只能临时加个下拉框，后端写if-else硬分支……最后那个app.py文件里，光模型路由逻辑就占了200多行，还全是重复的重试、超时、日志埋点。

这就是没中间层的真实代价：模型不是插件，是活的变量；而你的应用，正在被它牵着鼻子走。
AI网关不是给大厂准备的“高大上基建”，它是给每个真实业务线写的“模型适配器”——就像USB-C接口，不管手机是华为、苹果还是折叠屏，只要插进去，就能充电、传数据、投屏。AI网关干的就是这事：把千差万别的模型（文本、语音、图像、多模态）、千奇百怪的部署方式（云API、私有GPU集群、边缘小盒子）、五花八门的协议（REST、gRPC、WebSocket、SSE）全统一成你后端能理解的一套语言。它不替你写业务逻辑，但它让你写一次业务逻辑，就能随时切换底层模型，不用动一行核心代码。

所以别被“网关”俩字吓住。它不是Nginx加个转发规则，也不是K8s Ingress换个yaml。它是一套面向AI服务生命周期的抽象层：模型注册像添加打印机驱动，路由策略像设置快递收货地址，熔断限流像给水管装压力阀，审计日志像给每笔交易记流水账。你不需要成为K8s专家，但得知道——当Qwen突然返回503，当多模态模型处理一张图耗时从800ms飙到4.2秒，当销售临时要求对某类客户屏蔽某个模型——这些事，不该由你凌晨三点改代码来救火。

关键词“多模型”和“中间层”背后，本质是两个现实：第一，没有哪个模型能通吃所有场景；第二，业务迭代速度远快于模型升级周期。AI网关就是这两股力之间的缓冲垫。它不解决“怎么训好模型”，但解决“怎么让业务不被模型拖垮”。

2. 多模型时代的真实困境：不是选择困难，而是切换成本太高

很多人以为多模型只是“选哪个更好”的问题，其实真正卡脖子的是模型切换带来的系统性摩擦。这种摩擦不是理论上的，而是每天在开发、测试、上线、监控环节真实发生的。我整理了过去一年帮客户踩过的典型坑，按发生频率排序：

2.1 接口契约撕裂：同一个“生成摘要”，不同模型要你交三份“身份证”

假设你要实现“上传PDF→生成300字摘要”。用Qwen API时，你发JSON：

{ "model": "qwen-max", "input": {"documents": ["...base64..."]}, "parameters": {"max_length": 300} }

换成本地部署的Phi-3，接口变成：

{ "prompt": "请为以下文档生成300字摘要：\n\n[文档内容]", "max_tokens": 300, "temperature": 0.3 }

再切到多模态模型（比如Qwen-VL），你发现它根本不接PDF，得先调OCR服务抽文本，再喂给模型，还得处理图片里的表格识别——整个流程链路变了。

提示：这不是API设计水平问题，而是模型能力边界决定的。文本模型天生不理解二进制PDF，多模态模型需要像素级输入。网关要做的，不是强行统一输入格式，而是在入口处做语义归一化：你只传{"file": base64_pdf, "task": "summary", "length": 300}，网关自动拆解、调用OCR、拼提示词、选模型、合并结果。

2.2 错误处理失序：500报错背后，可能是网络、模型、还是你传错了字段？

Qwen返回{"code": 401, "message": "Invalid API key"}，你立刻知道是密钥问题；
但本地Llama-3返回{"error": "CUDA out of memory"}，你得查GPU显存、batch size、甚至模型量化精度；
而多模态模型遇到模糊图片，可能静默返回空结果，或者返回乱码，连HTTP状态码都是200。

没有网关时，你的业务代码里塞满了if "CUDA" in str(e): retry_with_smaller_batch()这类补丁。网关则把错误分类：基础设施类（超时、连接拒绝）、模型能力类（输入不支持、输出截断）、业务规则类（长度超限、敏感词拦截），再映射成统一错误码（如MODEL_RESOURCE_EXHAUSTED）和用户友好提示（“当前请求负载较高，请稍后重试”）。

2.3 流量治理失控：一个模型挂了，不该让整个AI功能瘫痪

去年有家教育公司，把作文批改、题目解析、知识点推荐全绑在一个模型实例上。某天模型服务因GPU故障不可用，客服系统报警：“AI批改失败率100%”。运维查了一小时才发现，是批改模块的健康检查探针没配，导致K8s没触发重启。更糟的是，题目解析请求还在持续打进来，把本就不堪重负的节点彻底压垮——一个功能故障，引发雪崩式连锁反应。

AI网关的熔断机制不是简单“5秒失败3次就停”，而是分层设计：

• 模型级熔断：针对单个模型实例（如qwen-7b-gpu-01），连续5次超时即隔离；
• 能力级熔断：当“文本生成”能力整体错误率＞15%，自动降级到备用模型（如切到GLM-4）；
• 租户级熔断：某客户API Key调用量突增300%，自动限流，不影响其他客户。

这背后是实时指标采集：每毫秒记录请求耗时、token消耗、错误类型，用滑动窗口计算P95延迟和错误率。没有网关，这些数据散落在各模型服务的日志里，等你发现异常，业务已经损失两小时。

2.4 模型灰度发布无从谈起：新模型上线=全量切换，风险不可控

某电商团队上线多模态商品图描述模型，想先对1%流量灰度。没网关时怎么做？要么改Nginx配置做随机分流（但无法按用户ID哈希，导致同一用户反复切模型），要么在业务代码里加判断逻辑（if user_id % 100 == 0: use_new_model()），结果测试期间发现新模型对某些品牌logo识别不准，紧急回滚——但已有2000个用户看到错误描述，客服电话被打爆。

网关的灰度发布是声明式的：

routes: - name: product-caption match: "user_tier == 'premium'" backend: qwen-vl-prod weight: 100 - name: product-caption-fallback match: "true" backend: clip-vit-l weight: 100

它支持按用户属性、设备类型、地域、甚至请求内容特征（如图片是否含文字）做精准分流，并内置AB测试效果对比面板：新模型的描述准确率、人工复核通过率、用户点击率变化一目了然。

2.5 审计与合规形同虚设：你说“没传用户隐私”，证据在哪？

金融客户要求：所有AI请求必须留存原始输入、模型输出、调用时间、操作人账号，保留180天。没网关时，每个模型服务自己记日志，格式五花八门：Qwen用JSON，本地Llama用纯文本，多模态模型甚至只打DEBUG日志。合规部门要查某次投诉，得登录4台服务器，grep不同路径，再人工比对时间戳——平均耗时3小时。

网关的日志是结构化的审计事件流：

{ "event_id": "req_abc123", "timestamp": "2024-06-15T08:22:14.882Z", "user_id": "U7890", "tenant_id": "bank-xyz", "model": "qwen-vl-prod", "input_hash": "sha256:...", "output_truncated": false, "tokens_in": 128, "tokens_out": 42, "latency_ms": 1240, "status": "success" }

输入哈希值确保不可篡改，output_truncated字段明确标识是否因长度限制被截断（避免隐瞒关键信息），所有字段可直接对接SIEM系统。这才是真正的“留痕可溯”。

3. AI网关的核心能力拆解：它到底在中间层里干了什么

把AI网关想象成机场的航站楼——飞机（模型）各不相同（波音、空客、支线小飞机），乘客（请求）带着不同行李（文本、图片、音频），但所有人在出发前都经过同一套流程：值机（认证）、安检（校验）、登机口分配（路由）、广播通知（错误处理）、行李追踪（审计）。网关的每一项能力，都是这个流程中的标准化环节。

3.1 模型注册与元数据管理：不是加个URL，而是建“模型档案”

很多团队以为网关配置就是填个API地址，实际远不止。一个健壮的模型注册包含至少7个维度：

我见过最典型的错误：把qwen-max和qwen-plus注册成两个独立模型，但它们共享同一套后端服务。结果网关健康检查时，一个探针失败，另一个也跟着下线——因为底层是同一组Pod。正确做法是注册为同一model_id下的不同version，并配置weight做灰度。

注意：input_schema不是摆设。当业务传入{"prompt": 123}（数字而非字符串），网关直接返回400，而不是把错误透传给模型。这省去大量后端类型校验代码。

3.2 智能路由：不是轮询或哈希，而是“按需匹配”

传统API网关路由看URL路径或Header，AI网关路由看语义意图。例如：

• 请求带Content-Type: image/jpeg且X-Task: caption→ 路由到多模态模型集群
• 请求JSON中"task": "code-review"且"language": "python"→ 路由到CodeLlama专用节点
• 请求含"stream": true→ 过滤掉不支持SSE的模型

更进一步，路由可结合实时指标：

# 伪代码：动态权重路由 def select_backend(request): candidates = get_models_by_capability(request.capabilities) # 排除错误率＞5%的模型 candidates = [m for m in candidates if m.error_rate < 0.05] # 优先选P95延迟＜800ms的 candidates.sort(key=lambda m: m.p95_latency) return candidates[0]

实测下来，这种策略让整体P95延迟降低37%，错误率下降22%。因为网关永远把请求导向“此刻状态最好”的模型，而不是静态配置的“理论上最优”模型。

3.3 协议转换与提示词工程：让老系统无缝接入新模型

这是最容易被低估的能力。很多遗留系统用SOAP协议，而新模型只支持REST；有些IoT设备只能发MQTT消息，但模型API要HTTP。网关在中间做协议桥接：

• MQTT → HTTP：将MQTT payload解析为JSON，注入标准headers，转发到模型API
• SOAP → REST：提取SOAP Body中的<Prompt>字段，构造成REST body，调用模型
• 二进制图片 → Base64：自动将multipart/form-data中的图片转Base64，适配模型要求

更关键的是提示词（Prompt）的自动注入。比如业务只传{"text": "总结这篇文章"}，网关根据目标模型自动补全：

• 对Qwen："请用中文总结以下文章，不超过300字：\n\n{text}"
• 对Llama-3："<|begin_of_text|><|start_header_id|>system<|end_header_id|>你是一个专业编辑，用简洁语言总结文章。<|eot_id|><|start_header_id|>user<|end_header_id|>请总结：{text}<|eot_id|>"
• 对多模态模型：先OCR抽文本，再拼"请分析以下图文内容并总结：[OCR文本] + [图片base64]"

这避免了业务代码里充斥着不同模型的提示词模板，也保证了提示词质量的一致性——毕竟不是每个工程师都懂怎么写高质量system prompt。

3.4 流量治理：不只是限流，而是“智能节流”

限流（Rate Limiting）常被简化为“每秒100次”，但AI场景需要更精细的控制：

• Token级限流：不限制请求数，而限制每分钟总token消耗（如用户每月最多用50万tokens）
• 模型级配额：qwen-max每天最多调用1万次，超限自动切到qwen-plus
• 突发流量保护：检测到某IP 1秒内发起50次请求，立即启用leaky bucket算法平滑流量，而非粗暴429

网关的限流器必须和模型监控深度集成。例如，当检测到qwen-vl的GPU显存使用率＞90%，自动将新请求的priority降级，让高优请求（如VIP客户）先处理，普通请求排队等待——这比单纯“拒绝”更合理。

3.5 统一可观测性：不是堆监控，而是建“AI服务仪表盘”

网关的监控面板必须回答三个问题：

1. 现在谁在用？—— 按租户、模型、API Key维度统计QPS、成功率、平均延迟
2. 哪里出问题了？—— 错误分布热力图（哪些模型/哪些错误码集中爆发）
3. 要不要扩容？—— 基于token消耗预测未来24小时资源需求（如“当前token消耗速率，3小时后GPU显存将达95%”）

我给客户部署时，强制要求开启trace_id透传。每个请求从网关生成唯一trace_id，注入到所有下游模型调用的headers中。这样当qwen-vl返回异常，你能直接关联到上游哪个业务请求、哪个用户、哪段前端代码触发的——而不是在日志里大海捞针。

4. 从零搭建一个轻量级AI网关：用Python+FastAPI实现实战版

别被“网关”吓住。一个生产可用的最小可行网关（MVP），核心代码不到500行。下面是我给初创团队落地的方案，兼顾轻量、可扩展、易维护。

4.1 技术选型逻辑：为什么选FastAPI而不是Envoy或Kong？

• Envoy/Kong：企业级，但学习成本高，配置复杂，小团队运维负担重
• 自研Go网关：性能好，但开发周期长，调试难
• FastAPI：Python生态成熟，异步支持好，Pydantic Schema验证天然契合AI请求校验，且可直接用uvicorn单进程跑，Docker镜像仅80MB

关键决策点：

• 模型注册存储：用SQLite而非Redis。理由：模型元数据变更频率低（每天＜10次），但需要ACID事务（如更新weight时必须原子性），SQLite更轻量可靠
• 路由引擎：不依赖外部规则引擎，用Python字典+函数式编程实现。实测10万QPS下延迟＜2ms
• 协议转换：用aiohttp做HTTP客户端，paho-mqtt做MQTT桥接，避免C扩展依赖

4.2 核心代码结构（目录树）

ai-gateway/ ├── main.py # FastAPI主应用 ├── models/ # 数据模型定义 │ ├── model.py # ModelConfig, RouteRule │ └── request.py # InputRequest, OutputResponse ├── registry/ # 模型注册中心 │ ├── sqlite_registry.py # SQLite实现 │ └── __init__.py ├── router/ # 智能路由引擎 │ ├── engine.py # 核心路由逻辑 │ └── strategies/ # 不同路由策略 │ ├── capability.py # 按能力匹配 │ └── latency.py # 按延迟优选 ├── protocol/ # 协议转换器 │ ├── rest_adapter.py # REST通用适配 │ └── multimodal.py # 多模态专用处理（OCR+提示词） └── utils/ ├── metrics.py # Prometheus指标收集 └── logger.py # 结构化日志

4.3 关键模块实现详解

模型注册（registry/sqlite_registry.py）

import sqlite3 from typing import List, Optional from models.model import ModelConfig class SQLiteRegistry: def __init__(self, db_path: str = "models.db"): self.db_path = db_path self._init_db() def _init_db(self): # 创建表，注意capabilities是JSON数组，便于WHERE查询 with sqlite3.connect(self.db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS models ( id TEXT PRIMARY KEY, name TEXT NOT NULL, endpoint TEXT NOT NULL, protocol TEXT NOT NULL DEFAULT 'rest', capabilities TEXT NOT NULL, -- JSON array like '["text-gen"]' input_schema TEXT, output_schema TEXT, health_check_path TEXT DEFAULT '/healthz', weight INTEGER DEFAULT 100, status TEXT DEFAULT 'active' -- active/inactive/degraded ) """) def list_models(self, capability: Optional[str] = None) -> List[ModelConfig]: """按能力查询可用模型""" where_clause = " WHERE status = 'active'" params = [] if capability: where_clause += " AND capabilities LIKE ?" params.append(f'%"{capability}"%') with sqlite3.connect(self.db_path) as conn: rows = conn.execute( f"SELECT * FROM models {where_clause}", params ).fetchall() return [ModelConfig.from_row(row) for row in rows] def update_weight(self, model_id: str, weight: int): """原子性更新权重，用于灰度发布""" with sqlite3.connect(self.db_path) as conn: conn.execute( "UPDATE models SET weight = ? WHERE id = ?", (weight, model_id) )

实操心得：capabilities字段用JSON字符串存储而非单独建表，是因为查询频率远高于更新频率，且SQLite的LIKE查询对短JSON足够快。如果模型数超1000，再考虑用FTS5全文索引。

智能路由引擎（router/engine.py）

from typing import List, Dict, Any from models.model import ModelConfig from registry.sqlite_registry import SQLiteRegistry from router.strategies.capability import match_by_capability from router.strategies.latency import select_by_latency class RoutingEngine: def __init__(self, registry: SQLiteRegistry): self.registry = registry def route(self, request: Dict[str, Any]) -> ModelConfig: """ 核心路由逻辑：先按能力过滤，再按延迟优选 """ # Step 1: 按请求能力需求筛选候选模型 capabilities = self._extract_capabilities(request) candidates = self.registry.list_models() filtered = match_by_capability(candidates, capabilities) if not filtered: raise ValueError(f"No model supports capabilities {capabilities}") # Step 2: 按实时延迟指标排序（这里简化为读取预存指标） # 实际生产中，从Prometheus拉取最新P95延迟 sorted_models = select_by_latency(filtered) # Step 3: 按weight做加权随机选择（支持灰度） total_weight = sum(m.weight for m in sorted_models) import random rand = random.randint(0, total_weight - 1) for model in sorted_models: if rand < model.weight: return model rand -= model.weight return sorted_models[0] # fallback def _extract_capabilities(self, request: Dict[str, Any]) -> List[str]: """从请求中提取语义能力标签""" caps = [] if request.get("stream"): caps.append("streaming") if "image" in request or request.get("content_type", "").startswith("image/"): caps.append("multimodal") if request.get("task") in ["summary", "translate", "code-review"]: caps.append("text-generation") return caps

多模态协议转换（protocol/multimodal.py）

import base64 from io import BytesIO from PIL import Image import requests def preprocess_multimodal_request(request: dict) -> dict: """ 处理多模态请求：OCR抽文本 + 构造提示词 """ # 1. 提取图片（支持base64或URL） image_data = request.get("image") if isinstance(image_data, str) and image_data.startswith("http"): # 下载远程图片 img_bytes = requests.get(image_data).content elif isinstance(image_data, str): # base64解码 img_bytes = base64.b64decode(image_data.split(",")[-1]) else: raise ValueError("Invalid image format") # 2. OCR识别（调用本地Tesseract或云OCR服务） ocr_text = _call_ocr_service(img_bytes) # 3. 构造多模态提示词 prompt = f"请分析以下图文内容：\n\n【OCR识别文本】\n{ocr_text}\n\n【图片描述】\n{request.get('caption_hint', '')}" # 4. 注入到标准请求体 return { "prompt": prompt, "max_tokens": request.get("max_tokens", 512), "temperature": request.get("temperature", 0.5) } def _call_ocr_service(img_bytes: bytes) -> str: """ 实际项目中替换为Tesseract或百度OCR SDK 这里用mock模拟 """ # 真实OCR会返回结构化文本+坐标 return "发票日期：2024-06-15，金额：¥12,800.00，收款方：XX科技有限公司"

4.4 部署与配置实战

Dockerfile（精简版）

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0:8000", "--port", "8000", "--workers", "4"]

启动命令

# 初始化数据库 python -c "from registry.sqlite_registry import SQLiteRegistry; SQLiteRegistry()" # 启动网关（生产环境加--reload） uvicorn main:app --host 0.0.0.0:8000 --port 8000 --workers 4 # 注册第一个模型（curl示例） curl -X POST http://localhost:8000/models \ -H "Content-Type: application/json" \ -d '{ "id": "qwen-7b-chat", "name": "Qwen 7B Chat", "endpoint": "https://api.qwen.com/v1/chat", "capabilities": ["text-generation", "streaming"], "input_schema": "{\"type\":\"object\",\"properties\":{\"prompt\":{\"type\":\"string\"}}}", "weight": 100 }'

生产环境必配项

• HTTPS终止：在网关前加Nginx，处理SSL卸载，网关内部走HTTP
• 健康检查：K8s liveness probe指向/healthz，该接口检查SQLite连接+至少一个模型可用
• 日志归集：uvicorn日志输出JSON格式，用Filebeat推送到ELK
• 配置热更新：监听SQLite文件变更，自动重载模型列表（避免重启）

注意：不要在网关里做模型推理！它的唯一职责是调度、转换、治理。所有模型服务必须独立部署，网关只做HTTP客户端。

5. 常见问题与避坑指南：那些文档里不会写的实战细节

5.1 “我的模型是gRPC，网关能接吗？”——协议适配的边界在哪？

能接，但要注意gRPC的特殊性。gRPC不像HTTP有明确的RESTful语义，其方法名（如/inference.InferenceService/Generate）和protobuf schema才是关键。网关需：

• 动态加载proto文件：启动时读取.proto定义，生成Python stub
• 请求体转换：将JSON请求按proto schema序列化为二进制
• 流式响应处理：gRPC streaming需在网关层转成SSE或WebSocket，否则前端无法消费

避坑点：不要试图在网关里写gRPC server。正确做法是让模型服务同时暴露gRPC和REST接口（用grpc-gateway），网关只走REST。这样既保持网关轻量，又避免gRPC版本兼容问题。

5.2 “模型返回的token数不准，限流失效怎么办？”

这是高频痛点。Qwen API返回usage.total_tokens，但本地Llama模型可能只返回generated_tokens（不含prompt tokens）。网关限流若只看模型返回值，会导致：

• 对Qwen：严格按总tokens限流
• 对Llama：漏计prompt tokens，实际超限

解决方案：网关主动计算prompt tokens。用tiktoken库（Qwen用cl100k_base，Llama用llama3）对输入文本分词，再加模型返回的generated_tokens。实测误差＜3 tokens。

from tiktoken import get_encoding def count_prompt_tokens(text: str, model_name: str) -> int: if "qwen" in model_name.lower(): enc = get_encoding("cl100k_base") elif "llama" in model_name.lower(): enc = get_encoding("llama3") else: enc = get_encoding("gpt2") # fallback return len(enc.encode(text))

5.3 “灰度发布时，新旧模型输出格式不一致，下游炸了”

典型场景：旧模型返回{"result": "text"}，新模型返回{"response": {"text": "text", "confidence": 0.92}}。业务代码没适配，直接res["result"]报KeyError。

网关必须做输出标准化。在路由后、返回前插入转换层：

def standardize_output(model_id: str, raw_output: dict) -> dict: if "qwen" in model_id: return {"text": raw_output.get("text", ""), "tokens": raw_output.get("usage", {}).get("total_tokens", 0)} elif "llama" in model_id: return {"text": raw_output.get("generated_text", ""), "tokens": raw_output.get("tokens", 0)} else: return raw_output # passthrough

实操心得：标准化规则必须版本化。v1标准要求text字段，v2增加metadata对象。网关根据请求HeaderAccept-Version: v2决定用哪套规则，避免一刀切破坏旧客户端。

5.4 “网关成了单点故障，怎么保证高可用？”

AI网关本身必须无状态。所有状态（模型注册、路由策略、限流计数）存外部系统：

• 模型注册：SQLite可主从复制，或直接用PostgreSQL（推荐）
• 限流计数：用Redis Cluster，key为rate_limit:{tenant_id}:{model_id}
• 健康检查状态：用Consul或etcd做分布式心跳

部署时，至少2个网关实例，前面挂负载均衡（如AWS ALB）。实测：单实例可扛5000 QPS，双实例+ALB自动故障转移，RTO＜30秒。

5.5 “审计日志太大，磁盘爆了怎么办？”

结构化日志虽好，但1000 QPS下每天产生20GB日志。必须分级：

• Level 1（必存）：event_id,timestamp,user_id,model_id,status,latency_ms,tokens_in/out—— 存Elasticsearch，保留30天
• Level 2（采样存）：1%请求存完整input_hash和output_truncated—— 用于合规抽查
• Level 3（不存）：原始input/output文本 —— 除非X-Audit-Full: trueHeader显式要求

用Logstash做日志采样：if [level] == "2" and (rand() * 100) > 99 { ... }

6. 多模态模型落地的特殊考量：网关如何应对“图文音”混合战场

多模态不是“文本模型+图片模型”的简单叠加，而是能力维度的指数级增长。网关在此场景下，角色从“调度员”升级为“交响乐指挥”。

6.1 输入组合爆炸：一张图+一段语音+一段文字，该怎么路由？

传统路由看单一特征，多模态必须做联合特征分析。例如：

• 图片+文字 → 可能是“图文问答”，路由到Qwen-VL
• 语音+文字 → 可能是“会议纪要”，需ASR转文字再摘要，路由到Whisper+Qwen流水线
• 图片+语音 → 可能是“视频帧分析”，需抽帧+语音转录+联合推理

网关需内置轻量特征提取器：

• 用python-magic识别文件类型（避免image/jpegHeader被伪造）
• 用ffprobe快速获取音频时长、采样率（判断是否需分段）
• 用PIL.Image读取图片尺寸、模式（RGB/RGBA，排除损坏图片）

然后生成联合路由标签：["multimodal", "audio", "long-context"]，匹配预设规则。

6.2 输出格式碎片化：文本、JSON、二进制图片、音频流，怎么统一返回？

多模态模型输出五花八门：

• Qwen-VL：返回JSON含text和image_url
• Stable Diffusion：返回PNG二进制
• Whisper：返回VTT字幕文件

网关必须做输出协议协商：

• 请求HeaderAccept: application/json→ 返回结构化JSON（图片转base64）
• Accept: image/png→ 若模型输出图片，直接透传；否则返回406
• Accept: text/vtt→ 对ASR结果做VTT封装

核心原则：网关不生成内容，只转换容器。图片base64编码开销大，生产环境应配置CDN回源，网关只返回{"image_url": "https://cdn.example.com/xxx.png"}。

6.3 计算资源异构：GPU、CPU、NPU，网关怎么感知并调度？

模型部署环境差异巨大：

• Qwen-VL：需A10G GPU
• Whisper-small：CPU即可，但需大内存
• Phi-3-vision：NPU加速，Linux驱动特殊

网关的健康检查必须包含资源探测：

# GPU探测 nvidia-smi --query-gpu=utilization.gpu,temperature.gpu --format=csv,noheader,nounits # CPU探测 lscpu | grep "CPU\(s\)"

将结果存入模型元数据resource_profile字段。路由时，若请求标记"gpu_required": true，则只选resource_profile.gpu_util < 70%的模型。

注意：不要在网关里执行