简介基于AI大模型API实现的聚合模型服务资源包面向需要在单一平台内灵活调度多家模型的开发者解决多模型切换成本高、接口不统一的问题。资源覆盖DeepSeek、月之暗面、豆包、OpenAI、Claude3、文心一言、通义千问、讯飞星火、智谱清言、腾讯混元等主流模型内置一键切换逻辑同时整合Ollama、Langchain本地模型与知识库问答能力并预留扣子、Dify、FastGPT、Gitee AI等在线API扩展适合用于模型管理、能力对比、私有化知识库等场景。包体共1215个文件压缩后约7.15MB轻量易部署内容以705个Java后端代码、111个Vue前端页面、90个JavaScript与49个TypeScript脚本为主辅以SVG图标、Less/SCSS样式、SQL脚本及Docker/Nginx/Redis等配置文件组成一套可运行、可二次开发的前后端一体工程。目前已有493人浏览学习可直接参照接入多家大模型API省去逐一对接的重复工作快速搭建统一的模型服务入口。1. 聚合模型服务把十个大模型 API 收编成一个网关值得吗做 AI 大模型应用开发最烦的不是效果而是厂商太多。接了几家模型请求格式、鉴权方式、流式返回各不一样代码里于是长出一堆 if else 适配函数。需求还总在变今天说 DeepSeek 便宜明天说 Claude3 写代码强——换一次模型就改一遍调用逻辑血泪经验全是这么攒出来的。这套基于 AI 大模型 API 实现的聚合模型服务把脏活集中到一个网关对外统一暴露 OpenAI 兼容接口对内适配 DeepSeek、月之暗面、豆包、OpenAI、Claude3、文心一言、通义千问、讯飞星火、智谱清言(ChatGLM)、腾讯混元等主流厂商。调用方只改 model 字段就能一键切换业务代码不用动。适合正在做多模型接入的应用开发团队、需要模型冗余降级的线上服务以及想低成本横向对比各家模型效果的独立开发者。下面按设计、实现、部署、踩坑、进阶五层拆开讲。2. 统一协议与模型路由为什么选 OpenAI 兼容层如何做一键切换先解决最核心的设计问题十个厂商协议、鉴权、模型 ID 全不一样网关拿什么当统一语言我的结论很直接选 OpenAI 的 chat/completions 格式做中间层这是聚合类服务通行做法代价最小。原因是 ChatGPT 把这个调用方式带成了事实标准国内主流厂商要么原生兼容要么提供了兼容模式的 endpoint只有 Claude3 需要单独做一次字段转换。2.1 统一请求协议OpenAI 兼容格式是各家厂商都认的普通话OpenAI 格式的核心就三块路径固定为 /v1/chat/completions请求体必须有 model 和 messagesmessages 是[{role: system|user|assistant, content: ...}]结构需要逐字渲染时加 stream: true。各家真正的差异不在结构而在 endpoint 和模型 ID 的命名上所以我习惯把所有差异收进一张路由表厂商兼容 OpenAI 格式常见 endpoint 风格网关内模型 ID 示例DeepSeek是api.deepseek.com/v1/chat/completionsdeepseek-chat、deepseek-reasoner月之暗面是api.moonshot.cn/v1/chat/completionsmoonshot-v1-8k豆包是ark.cn-beijing.volces.com/api/v3/chat/completionsdoubao-pro-32kOpenAI原生api.openai.com/v1/chat/completionsgpt-4oClaude3否需做字段转换api.anthropic.com/v1/messagesclaude-3-5-sonnet文心一言是Qianfan v2qianfan.baidubce.com/v2/chat/completionsernie-4.0-8k通义千问是dashscope.aliyuncs.com/compatible-mode/v1/chat/completionsqwen-plus讯飞星火是新版开放接口spark-api-open.xf-yun.com/v1/chat/completionsspark-max智谱清言是open.bigmodel.cn/api/paas/v4/chat/completionsglm-4-plus腾讯混元是api.hunyuan.cloud.tencent.com/v1/chat/completionshunyuan-turbo这张表有两点要特别注意。第一Claude3 是唯一不兼容 OpenAI 格式的Anthropic 的 messages 接口把 system 放在顶层参数而不是塞进 messages 数组网关要对它单独做转换——把 OpenAI 请求里的 system 消息抽出来映射成顶层 system其余消息重组成 Anthropic 的 user/assistant 结构。第二讯飞星火新版接口在鉴权上有特殊要求Authorization 头要传apiKey:apiSecret的冒号拼接串鉴权模块要对它单独放行。这两家是接入时最容易卡住的地方。提示表里的 endpoint 是我日常在用的地址各家平台偶尔调整版本后缀。接进网关后换 endpoint 只需要改配置表不需要改代码这正是路由表存在的意义。2.2 模型路由表一键切换的本质是改一行配置一键切换听着玄拆开看就是一张路由表加一个路由函数。请求进来网关用 model 字段查表拿到该模型对应的上游 endpoint、上游模型名、默认参数和厂商 Key再做协议转换与转发。调用方视角只有一个变化model 从 deepseek-chat 改成 glm-4-plus其他什么都不用动。# gateway/model_registry.py MODEL_ROUTES { deepseek-chat: { provider: deepseek, endpoint: https://api.deepseek.com/v1/chat/completions, upstream_model: deepseek-chat, default_params: {temperature: 0.7, max_tokens: 2048}, }, deepseek-reasoner: { provider: deepseek, endpoint: https://api.deepseek.com/v1/chat/completions, upstream_model: deepseek-reasoner, default_params: {max_tokens: 8192}, }, glm-4-plus: { provider: zhipu, endpoint: https://open.bigmodel.cn/api/paas/v4/chat/completions, upstream_model: glm-4-plus, default_params: {temperature: 0.3}, }, hunyuan-turbo: { provider: tencent, endpoint: https://api.hunyuan.cloud.tencent.com/v1/chat/completions, upstream_model: hunyuan-turbo, default_params: {}, }, } def resolve_route(model: str) - dict: route MODEL_ROUTES.get(model) if not route: raise ValueError(funknown model: {model}) # 返回副本防止调用方篡改全局路由表 return dict(route) def build_upstream_payload(route: dict, payload: dict) - dict: # 默认参数与请求参数合并请求参数优先 merged {**route.get(default_params, {}), **payload} # 用上游模型 ID 覆盖网关内模型名 merged[model] route[upstream_model] return merged这段代码有三个设计点值得说。第一路由表的 key 是网关内模型名和上游模型名解耦上游升级模型时只改 upstream_model 一个字段全站调用不用动。第二default_params 按模型单配因为不同模型对 temperature、max_tokens 的敏感度差别很大——智谱我习惯压到 0.3DeepSeek 用 0.7 才不呆。第三build_upstream_payload 合并参数时让请求参数优先调用方传了 temperature 就能覆盖默认值这是默认参数不干扰显式传参的惯例。resolve_route 返回副本防止多租户场景下某个调用方把全局路由表污染掉。有了这张表一键切换的完整链路就是业务侧把请求发到网关网关取 model 字段查表替换 endpoint 和上游模型名带上该厂商的 Key 转发。业务代码里没有任何厂商分支新增模型 加一行路由配置 在 .env 里加一个 Key不需要发版。2.3 AI 交互逻辑封装为什么用异步中间件而不是各业务各写一段这类网关的技术栈我的选择是 FastAPI httpx.AsyncClient消息走 SSE前端用 fetch 的 ReadableStream 消费。选 FastAPI 因为它原生支持 StreamingResponseasync 语法处理多个上游并发调用很顺手而且自动生成 OpenAPI 文档业务方直接对着 /docs 联调。httpx 的 AsyncClient 自带连接池和厂商之间的 TCP 连接可以复用避免每个请求重新握手。为什么强调封装而不是各业务各写一段因为 AI 交互逻辑里有三个必须收敛的共性流式转发、鉴权、超时重试。如果每个业务模块自己写一遍调用代码最终一定出现三套不一致的行为有人忘了传 stream有人把 Key 硬编码进代码有人不处理 429 直接重试把厂商打限流。收敛成网关中间层之后这些行为变成可配置项# gateway/app.py 关键骨架 from fastapi import FastAPI, Request, HTTPException from fastapi.responses import StreamingResponse import httpx app FastAPI(titlellm-aggregate-gateway) app.post(/v1/chat/completions) async def chat_completions(payload: dict, request: Request): model payload.pop(model) route resolve_route(model) # 对外鉴权校验网关 Key而不是厂商 Key auth request.headers.get(Authorization, ) if not auth.endswith(sk-agg-local): raise HTTPException(status_code401, detailinvalid gateway key) # 对内取厂商 Key转发时使用 api_key get_upstream_key(route[provider]) headers {Authorization: fBearer {api_key}} upstream_payload build_upstream_payload(route, payload) async def event_stream(): async with httpx.AsyncClient(timeouthttpx.Timeout(600.0, connect10.0)) as client: async with client.stream( POST, route[endpoint], jsonupstream_payload, headersheaders, ) as resp: if resp.status_code ! 200: error_body (await resp.aread()).decode() yield fevent: error\ndata: {error_body}\n\n return async for line in resp.aiter_lines(): if line.startswith(data:): yield f{line}\n\n return StreamingResponse( event_stream(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )这里三个参数值得展开。timeout 主超时设 600 秒是为了兜住 deepseek-reasoner 这类推理模型的沉思过程connect 单独设 10 秒防止厂商域名解析或握手阶段长时间挂起。media_type 必须是 text/event-stream否则浏览器端 EventSource 不认X-Accel-Buffering: no 是写给 nginx 的指令告诉它别缓冲这个响应否则流式会被攒成一次性返回。这些值我全部放到配置里而不是硬编码生产环境调参不用动代码。3. SSE 流式输出、abort 与鉴权核心实现和参数调优网关能不能用就看三个点流式输出是否逐字到达、停止生成是否真的停、鉴权是否能挡住所有不该进的人。这一章把这三块拆开每个都给到可抄的代码。3.1 服务端流式转发逐行透传保持 [DONE] 结束标记SSE 协议本身不复杂响应体是一连串以 data: 开头、空行结尾的文本行服务端每产出一个增量 token 写一行客户端逐行读。OpenAI 兼容格式的流式 chunk 长这样data: {choices: [{delta: {content: 你好}, index: 0}]} data: {choices: [{delta: {content: 我是}, index: 0}]} data: [DONE]网关要做的是透传不是解析重组。透传比把流式体转成语义化 JSON再下发稳得多因为各家在流式细节上都有小动作有的在最后一个 chunk 塞 usage 统计有的会多发一个空 delta解析重组会让网关变成新的故障点。我在 2.3 的代码里用 aiter_lines 逐行读、line.startswith(data:) 判断、原样 yield就是这个思路。结束标记 data: [DONE] 也原样透传调用方 SDK 靠它判断流结束。这里有个隐蔽坑日志中间件把流读掉。FastAPI 项目常挂 logging 中间件它为了打访问日志会把 response body 读一遍流式响应被消费后下游一个字都收不到。解法是日志中间件对 text/event-stream 类型直接跳过或者只记录状态码和耗时不碰 body。我线上版就是这么处理的。3.2 前端消费与 abort停止生成必须一级一级取消前端消费 SSE 有两条路。EventSource API 简单但只支持 GET而我们调大模型必须 POST否则 prompt 全暴露在 URL 里所以生产上更常见的是 fetch ReadableStream// 前端stream-chat.js const controller new AbortController(); async function streamChat(messages, model) { const resp await fetch(/v1/chat/completions, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ model, messages, stream: true }), signal: controller.signal, }); const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // SSE 以空行分隔每条 data需要攒包再切分 const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (!line.startsWith(data:)) continue; const data line.slice(5).trim(); if (data [DONE]) return; const chunk JSON.parse(data); const delta chunk.choices?.[0]?.delta?.content ?? ; if (delta) renderDelta(delta); } } } function stop() { controller.abort(); // 用户点停止生成时调用 }这段代码的关键是 signal 和 buffer 处理。signal 让前端 abort 后浏览器立刻断开与网关的连接buffer 的 split(\n\n) 处理粘包——SSE 是文本流一次 read 可能拿到半条 data 或多条 data必须攒起来按空行切分。我在 renderDelta 里只处理 delta.content丢弃 role 变更和 usage 统计的行避免界面出现无意义刷新。abort 最坑的地方在服务端前端断开连接上游厂商还在继续生成等于花着钱算没人看的 token。所以网关的透传生成器必须监听取消信号# gateway/stream.py 支持取消的流式生成器 import asyncio async def event_stream(route, headers, upstream_payload): async with httpx.AsyncClient(timeouthttpx.Timeout(600.0)) as client: async with client.stream( POST, route[endpoint], jsonupstream_payload, headersheaders, ) as resp: if resp.status_code ! 200: yield fevent: error\ndata: {(await resp.aread()).decode()}\n\n return try: async for line in resp.aiter_lines(): if line.startswith(data:): yield f{line}\n\n except asyncio.CancelledError: # 客户端断开主动关闭上游连接停止计费 await resp.aclose() raise流式响应被客户端断开时FastAPI 会向生成器抛 asyncio.CancelledError。这里必须把上游的 httpx 响应对象 aclose 掉否则 TCP 连接还挂着上游会一直把内容生成完。这是停止生成真正生效的最后一公里漏了这一步abort 就只是个前端假动作。3.3 鉴权与 Key 管理401 的根源大部分在配置不在代码网关鉴权分两层对外校验调用方拿的网关 Key对内替调用方保管各厂商 Key。对外最简单比对 Authorization 头里的 sk-agg- 前缀对内才是 401 高发区。DeepSeek 的典型报错是unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****这类 401 九成不是厂商封号而是配置污染。我踩过的三种一是 .env 里 Key 带了引号或行尾换行os.getenv 原样读入二是转发时把 Bearer 拼成 Bearer Bearer sk-xxx三是同一个 Key 同时填进了两个厂商变量后加载的覆盖了前面。所以 Key 读取我做了强制清洗# gateway/keys.py import os def get_upstream_key(provider: str) - str: env_key f{provider.upper()}_API_KEY key os.getenv(env_key, ).strip().strip().strip() if not key: raise RuntimeError(f{env_key} is empty, check .env) return key.strip() 去换行和空白再剥掉引号空值直接报错而不是传一个空串让厂商返回迷之错误。这一层加上之后401 在我这边基本绝迹。另外讯飞星火的 Key 比较特殊它要求 Authorization 传apiKey:apiSecret冒号拼接串我在 get_upstream_key 之外单独为 spark 写了一个 key 组装函数其他厂商一律走 Bearer 单 Key。Key 轮换也不能省各家平台控制台都有管理页面我习惯 90 天轮一次线上环境的 Key 通过环境变量注入不写进镜像。4. 本地部署与调用从 .env 配置到 curl、OpenAI SDK 双通道验证网关拿到手第一件事是在本机跑起来而不是直接上生产。这一章给 Windows 10 下的完整落地路径Python 版本管理、虚拟环境、Key 配置、启动验证、双通道调用。4.1 Windows 10 环境准备Python 版本管理器与虚拟环境我建议用 Python 版本管理器来装解释器而不是直接装系统 Python。原因很现实同时维护多个项目时A 项目要 3.10B 项目要 3.12没有版本管理器就只能互相卸载重装早晚翻车。Windows 10 上我常用 pyenv-win命令行装完后把 bin 目录和 shims 目录加进 PATH然后就能一条命令装任意版本并全局切换。# pyenv-win 安装完成后一键装解释器并切换 pyenv install 3.11.9 pyenv global 3.11.9 python --version # 应输出 3.11.9版本确认后建虚拟环境再装依赖。虚拟环境是 Python 项目的底线Windows 下直接 pip install 到全局权限问题和 DLL 冲突能折腾一整晚python -m venv .venv .venv\Scripts\activate # Windows 激活虚拟环境 pip install -U pip pip install fastapi[standard] httpx python-dotenvfastapi[standard] 会带 uvicorn 和标准依赖python-dotenv 负责读 .envhttpx 是网关访问上游的 HTTP 客户端。装完验证一下python -c import fastapi, httpx; print(fastapi.__version__, httpx.__version__)能输出版本号说明环境没问题再继续往下走。4.2 .env 配置与启动把十家 Key 一次配齐环境就绪后先写 .env。这份文件是网关的钥匙串每家一行暂时用不到的可以留空但至少配 DeepSeek 和智谱先把链路跑通# .env.example DEEPSEEK_API_KEYsk-svcac-xxxxx MOONSHOT_API_KEYsk-xxxxx DOUBAO_API_KEYxxxxx OPENAI_API_KEYsk-proj-xxxxx ANTHROPIC_API_KEYsk-ant-xxxxx ERNIE_API_KEYxxxxx QWEN_API_KEYsk-xxxxx SPARK_API_KEYxxxxx ZHIPU_API_KEYxxxxx HUNYUAN_API_KEYxxxxx # 网关对外校验的 Key逗号分隔可配多个 GATEWAY_API_KEYSsk-agg-local启动网关# 开发调试模式--reload 让代码改动自动重启 uvicorn gateway.app:app --host 127.0.0.1 --port 8000 --reload看到 Application startup complete 之后先别急着调模型。访问 http://127.0.0.1:8000/docs确认 FastAPI 自动生成的接口文档里 /v1/chat/completions 存在这一步能直接排除路由挂载问题比盲调 curl 少一轮是不是路径写错的猜测。4.3 双通道验证curl 确认链路OpenAI SDK 确认兼容性网关是不是真OpenAI 兼容得用两个通道同时验证。curl 验证原始链路OpenAI SDK 验证协议兼容两个都过才算数。先 curl# -N 关掉 curl 缓冲区才能看到逐字输出 curl -N http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer sk-agg-local \ -d {model: deepseek-chat, messages: [{role: user, content: 用一句话介绍你自己}], stream: true}正常的话终端里会一行一行蹦出 data: 开头的文本。接着用 OpenAI 官方 SDK 打同一个路径验证协议兼容层# sdk_check.py from openai import OpenAI client OpenAI( base_urlhttp://127.0.0.1:8000/v1, # 指向网关不是官方 api_keysk-agg-local, # 网关 Key不是厂商 Key ) resp client.chat.completions.create( modeldeepseek-chat, # 网关内模型 ID messages[{role: user, content: 写一个 Python 快速排序}], streamTrue, ) for chunk in resp: delta chunk.choices[0].delta if delta and delta.content: print(delta.content, end)这段能跑通说明兼容层是合格的OpenAI 官方 SDK 都认业务侧接进来就不用改代码。两个参数是关键接线点base_url 指向网关根路径末尾必须带 /v1api_key 用网关签发的 Key 而不是上游厂商的 Key。这两个写错是新手上路最常见的翻车点。验证完 DeepSeek把 model 换成 glm-4-plus 再跑一次同一段代码就是完整的一键切换演示。5. 避坑指南聚合服务最常见的五个现场故障提示以下五条全部按现象 → 原因 → 解决整理可当排查手册直接抄。5.1 DeepSeek 401Key 被引号和换行污染现象调用 deepseek-chat 返回unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****Key 肉眼看着是对的复制到官网控制台发起请求却完全正常。原因.env 文件里 Key 被包了引号或行尾带了不可见字符os.getenv 原样读入转发时 Authorization 头带着脏字符。另外转发代码把 Bearer 拼了两遍、变成 Bearer Bearer sk-xxx 的情况也很常见。解决Key 读取统一走 3.3 的 get_upstream_key强制 strip 和剥引号还不行就在代码里 print(repr(key)) 看首尾有没有隐藏字符。改完必须重启 uvicorn因为 .env 的修改不会热加载。从那以后我没在 401 上浪费过时间。5.2 400 上下文超长最大上下文 1048576 tokens 的边界现象把长对话历史全量发给 deepseek-reasoner报api error: 400 this models maximum context length is 1048576 tokens明明没聊那么多轮。原因reasoner 模型上下文窗口很大但请求里 system prompt、工具定义、历史消息全塞进去后某些字段按 token 计算会迅速膨胀网关透传时又原样转发没做任何裁剪于是顶到上限。解决在网关层加滑动窗口裁剪保留 system 和最近若干轮丢弃最早的消息。我的经验值是普通对话保留最近 20 轮、总字符超过 120000 就从头丢system 永远保留。裁剪后这类 400 基本消失。注意 reasoner 类模型对推理连贯性敏感批量对话场景建议同时调小 max_tokens给上下文腾位置。5.3 Windows 下 Docker API 连不上npipe 报错与替代方案现象Windows 10 上执行 docker build 或 docker run报failed to connect to the docker api at npipe:////./pipe/docker-desktop-linuxDocker 命令全部失效。原因Docker Desktop 没启动或 WSL2 后端已经崩了npipe 管道不存在。Windows 的 Docker 走虚拟化后端和 Linux 原生环境不一样经常出现服务看起来在、管道却不通的假象。解决先打开 Docker Desktop 等它显示 Engine running再执行docker context ls确认当前 context 是 desktop-linux。如果还连不上本地调试完全绕开 Docker直接python -m uvicorn gateway.app:app --port 8000跑代码等逻辑稳定了再回 Docker 打包。别在这上面耗时间它就是个环境问题不是代码问题。5.4 流式输出被攒包nginx 缓冲把 SSE 变成一次性返回现象前端调用网关界面空白很久然后所有内容一次性全部渲染出来完全没有逐字效果绕过 nginx 直连 8000 端口却正常。原因网关前面挂了 nginx 做流量入口nginx 默认开启 proxy_buffering会把上游的流式响应攒满缓冲区才发给客户端SSE 就此失去流式意义。解决在 nginx 的 location 里关掉缓冲并把读超时调大防止长时间推理被掐断location /v1/chat/completions { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_buffering off; # 关键关掉缓冲 proxy_cache off; proxy_read_timeout 300s; # 推理时间长必须放大 proxy_set_header Connection ; }改完nginx -t校验语法再 reload。如果网关后面还套了 CDNCDN 的缓冲同样要关否则公网环境会在同一坑里再摔一次。5.5 abort 之后上游还在烧钱连接没有真正取消现象用户点了停止生成前端界面停了厂商后台的 token 用量却还在涨或者 abort 几次之后后续请求全部变慢。原因前端 abort 只断了浏览器与网关之间的连接网关生成器没监听 CancelledError上游 httpx 连接保持打开厂商继续生成直到自然结束。请求变慢则是频繁断连导致 TCP 连接反复重建。解决生成器里加 try/except asyncio.CancelledErrorexcept 分支中 await resp.aclose() 关闭上游响应再 re-raise代码在 3.2 节已给出。同时用 httpx.AsyncClient 做连接复用避免每个请求都重新握手高频 abort 场景下能省下可观的建连开销。6. 进阶把网关变成多模型评测台顺带把账算清楚网关跑稳之后我做的第一件事是给每个流式请求记录三个指标首字延迟、总耗时、输出 token 数。有了这三项数据网关就从转发工具升级成多模型评测台——切换模型之前先拿数据说话而不是听销售和社区吹。实现思路是在 StreamingResponse 外层包一层记录逻辑流结束后把统计写入内存或 SQLite。核心代码就两段一段在请求结束钩子里记录指标一段是对同一组 prompt 轮询所有模型的评测脚本# gateway/metrics.py 在流结束时记录指标 async def record_stream(route, event_stream): t0 time.time() first_token_at None total_chars 0 for line in event_stream: if first_token_at is None and line.startswith(data:): first_token_at time.time() total_chars len(line) yield line SESSION[requests].append({ provider: route[provider], first_token_ms: int((first_token_at - t0) * 1000), total_ms: int((time.time() - t0) * 1000), chars: total_chars, })# eval/benchmark.py 同题多模型对比 import asyncio, time from openai import OpenAI client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keysk-agg-local) MODELS [deepseek-chat, qwen-plus, glm-4-plus, hunyuan-turbo] PROMPTS [用 Python 写一个指数退避重试的 HTTP 客户端, 解释 SSE 与 WebSocket 的适用边界] async def run(): for model in MODELS: for prompt in PROMPTS: t0 time.time() first None n_chars 0 resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], streamTrue, ) for chunk in resp: delta chunk.choices[0].delta if delta and delta.content: if first is None: first time.time() - t0 n_chars len(delta.content) print(model, round(first, 2), round(time.time() - t0, 2), n_chars) asyncio.run(run())跑完会得到一张非常直观的对比表模型首字延迟(s)总耗时(s)输出字符数现场印象deepseek-chat0.43.1412快中文自然glm-4-plus0.62.8386稳代码一般hunyuan-turbo0.53.6405中规中矩把各家官网的每百万 token 价格配进路由表这次评测每次调用花了多少钱就是一道算术题再叠加首字延迟答案一目了然。完整的路由表、SSE 转发代码、评测脚本我都整理在资源包里下载后照着 .env.example 配 Key 就能跑起来。从那以后我每次把新模型接进网关都强制走一遍注册路由 → 单发冒烟 → 流式压测 → 评测对比这四步顺手把 Key 的过期时间写进提醒日历再也没有半夜被 401 叫醒过。希望帮到你。本文还有配套的精品资源点击获取 简介基于AI大模型API实现的聚合模型服务资源包面向需要在单一平台内灵活调度多家模型的开发者解决多模型切换成本高、接口不统一的问题。资源覆盖DeepSeek、月之暗面、豆包、OpenAI、Claude3、文心一言、通义千问、讯飞星火、智谱清言、腾讯混元等主流模型内置一键切换逻辑同时整合Ollama、Langchain本地模型与知识库问答能力并预留扣子、Dify、FastGPT、Gitee AI等在线API扩展适合用于模型管理、能力对比、私有化知识库等场景。包体共1215个文件压缩后约7.15MB轻量易部署内容以705个Java后端代码、111个Vue前端页面、90个JavaScript与49个TypeScript脚本为主辅以SVG图标、Less/SCSS样式、SQL脚本及Docker/Nginx/Redis等配置文件组成一套可运行、可二次开发的前后端一体工程。目前已有493人浏览学习可直接参照接入多家大模型API省去逐一对接的重复工作快速搭建统一的模型服务入口。1. 聚合模型服务把十个大模型 API 收编成一个网关值得吗做 AI 大模型应用开发最烦的不是效果而是厂商太多。接了几家模型请求格式、鉴权方式、流式返回各不一样代码里于是长出一堆 if else 适配函数。需求还总在变今天说 DeepSeek 便宜明天说 Claude3 写代码强——换一次模型就改一遍调用逻辑血泪经验全是这么攒出来的。这套基于 AI 大模型 API 实现的聚合模型服务把脏活集中到一个网关对外统一暴露 OpenAI 兼容接口对内适配 DeepSeek、月之暗面、豆包、OpenAI、Claude3、文心一言、通义千问、讯飞星火、智谱清言(ChatGLM)、腾讯混元等主流厂商。调用方只改 model 字段就能一键切换业务代码不用动。适合正在做多模型接入的应用开发团队、需要模型冗余降级的线上服务以及想低成本横向对比各家模型效果的独立开发者。下面按设计、实现、部署、踩坑、进阶五层拆开讲。2. 统一协议与模型路由为什么选 OpenAI 兼容层如何做一键切换先解决最核心的设计问题十个厂商协议、鉴权、模型 ID 全不一样网关拿什么当统一语言我的结论很直接选 OpenAI 的 chat/completions 格式做中间层这是聚合类服务通行做法代价最小。原因是 ChatGPT 把这个调用方式带成了事实标准国内主流厂商要么原生兼容要么提供了兼容模式的 endpoint只有 Claude3 需要单独做一次字段转换。2.1 统一请求协议OpenAI 兼容格式是各家厂商都认的普通话OpenAI 格式的核心就三块路径固定为 /v1/chat/completions请求体必须有 model 和 messagesmessages 是[{role: system|user|assistant, content: ...}]结构需要逐字渲染时加 stream: true。各家真正的差异不在结构而在 endpoint 和模型 ID 的命名上所以我习惯把所有差异收进一张路由表厂商兼容 OpenAI 格式常见 endpoint 风格网关内模型 ID 示例DeepSeek是api.deepseek.com/v1/chat/completionsdeepseek-chat、deepseek-reasoner月之暗面是api.moonshot.cn/v1/chat/completionsmoonshot-v1-8k豆包是ark.cn-beijing.volces.com/api/v3/chat/completionsdoubao-pro-32kOpenAI原生api.openai.com/v1/chat/completionsgpt-4oClaude3否需做字段转换api.anthropic.com/v1/messagesclaude-3-5-sonnet文心一言是Qianfan v2qianfan.baidubce.com/v2/chat/completionsernie-4.0-8k通义千问是dashscope.aliyuncs.com/compatible-mode/v1/chat/completionsqwen-plus讯飞星火是新版开放接口spark-api-open.xf-yun.com/v1/chat/completionsspark-max智谱清言是open.bigmodel.cn/api/paas/v4/chat/completionsglm-4-plus腾讯混元是api.hunyuan.cloud.tencent.com/v1/chat/completionshunyuan-turbo这张表有两点要特别注意。第一Claude3 是唯一不兼容 OpenAI 格式的Anthropic 的 messages 接口把 system 放在顶层参数而不是塞进 messages 数组网关要对它单独做转换——把 OpenAI 请求里的 system 消息抽出来映射成顶层 system其余消息重组成 Anthropic 的 user/assistant 结构。第二讯飞星火新版接口在鉴权上有特殊要求Authorization 头要传apiKey:apiSecret的冒号拼接串鉴权模块要对它单独放行。这两家是接入时最容易卡住的地方。提示表里的 endpoint 是我日常在用的地址各家平台偶尔调整版本后缀。接进网关后换 endpoint 只需要改配置表不需要改代码这正是路由表存在的意义。2.2 模型路由表一键切换的本质是改一行配置一键切换听着玄拆开看就是一张路由表加一个路由函数。请求进来网关用 model 字段查表拿到该模型对应的上游 endpoint、上游模型名、默认参数和厂商 Key再做协议转换与转发。调用方视角只有一个变化model 从 deepseek-chat 改成 glm-4-plus其他什么都不用动。# gateway/model_registry.py MODEL_ROUTES { deepseek-chat: { provider: deepseek, endpoint: https://api.deepseek.com/v1/chat/completions, upstream_model: deepseek-chat, default_params: {temperature: 0.7, max_tokens: 2048}, }, deepseek-reasoner: { provider: deepseek, endpoint: https://api.deepseek.com/v1/chat/completions, upstream_model: deepseek-reasoner, default_params: {max_tokens: 8192}, }, glm-4-plus: { provider: zhipu, endpoint: https://open.bigmodel.cn/api/paas/v4/chat/completions, upstream_model: glm-4-plus, default_params: {temperature: 0.3}, }, hunyuan-turbo: { provider: tencent, endpoint: https://api.hunyuan.cloud.tencent.com/v1/chat/completions, upstream_model: hunyuan-turbo, default_params: {}, }, } def resolve_route(model: str) - dict: route MODEL_ROUTES.get(model) if not route: raise ValueError(funknown model: {model}) # 返回副本防止调用方篡改全局路由表 return dict(route) def build_upstream_payload(route: dict, payload: dict) - dict: # 默认参数与请求参数合并请求参数优先 merged {**route.get(default_params, {}), **payload} # 用上游模型 ID 覆盖网关内模型名 merged[model] route[upstream_model] return merged这段代码有三个设计点值得说。第一路由表的 key 是网关内模型名和上游模型名解耦上游升级模型时只改 upstream_model 一个字段全站调用不用动。第二default_params 按模型单配因为不同模型对 temperature、max_tokens 的敏感度差别很大——智谱我习惯压到 0.3DeepSeek 用 0.7 才不呆。第三build_upstream_payload 合并参数时让请求参数优先调用方传了 temperature 就能覆盖默认值这是默认参数不干扰显式传参的惯例。resolve_route 返回副本防止多租户场景下某个调用方把全局路由表污染掉。有了这张表一键切换的完整链路就是业务侧把请求发到网关网关取 model 字段查表替换 endpoint 和上游模型名带上该厂商的 Key 转发。业务代码里没有任何厂商分支新增模型 加一行路由配置 在 .env 里加一个 Key不需要发版。2.3 AI 交互逻辑封装为什么用异步中间件而不是各业务各写一段这类网关的技术栈我的选择是 FastAPI httpx.AsyncClient消息走 SSE前端用 fetch 的 ReadableStream 消费。选 FastAPI 因为它原生支持 StreamingResponseasync 语法处理多个上游并发调用很顺手而且自动生成 OpenAPI 文档业务方直接对着 /docs 联调。httpx 的 AsyncClient 自带连接池和厂商之间的 TCP 连接可以复用避免每个请求重新握手。为什么强调封装而不是各业务各写一段因为 AI 交互逻辑里有三个必须收敛的共性流式转发、鉴权、超时重试。如果每个业务模块自己写一遍调用代码最终一定出现三套不一致的行为有人忘了传 stream有人把 Key 硬编码进代码有人不处理 429 直接重试把厂商打限流。收敛成网关中间层之后这些行为变成可配置项# gateway/app.py 关键骨架 from fastapi import FastAPI, Request, HTTPException from fastapi.responses import StreamingResponse import httpx app FastAPI(titlellm-aggregate-gateway) app.post(/v1/chat/completions) async def chat_completions(payload: dict, request: Request): model payload.pop(model) route resolve_route(model) # 对外鉴权校验网关 Key而不是厂商 Key auth request.headers.get(Authorization, ) if not auth.endswith(sk-agg-local): raise HTTPException(status_code401, detailinvalid gateway key) # 对内取厂商 Key转发时使用 api_key get_upstream_key(route[provider]) headers {Authorization: fBearer {api_key}} upstream_payload build_upstream_payload(route, payload) async def event_stream(): async with httpx.AsyncClient(timeouthttpx.Timeout(600.0, connect10.0)) as client: async with client.stream( POST, route[endpoint], jsonupstream_payload, headersheaders, ) as resp: if resp.status_code ! 200: error_body (await resp.aread()).decode() yield fevent: error\ndata: {error_body}\n\n return async for line in resp.aiter_lines(): if line.startswith(data:): yield f{line}\n\n return StreamingResponse( event_stream(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )这里三个参数值得展开。timeout 主超时设 600 秒是为了兜住 deepseek-reasoner 这类推理模型的沉思过程connect 单独设 10 秒防止厂商域名解析或握手阶段长时间挂起。media_type 必须是 text/event-stream否则浏览器端 EventSource 不认X-Accel-Buffering: no 是写给 nginx 的指令告诉它别缓冲这个响应否则流式会被攒成一次性返回。这些值我全部放到配置里而不是硬编码生产环境调参不用动代码。3. SSE 流式输出、abort 与鉴权核心实现和参数调优网关能不能用就看三个点流式输出是否逐字到达、停止生成是否真的停、鉴权是否能挡住所有不该进的人。这一章把这三块拆开每个都给到可抄的代码。3.1 服务端流式转发逐行透传保持 [DONE] 结束标记SSE 协议本身不复杂响应体是一连串以 data: 开头、空行结尾的文本行服务端每产出一个增量 token 写一行客户端逐行读。OpenAI 兼容格式的流式 chunk 长这样data: {choices: [{delta: {content: 你好}, index: 0}]} data: {choices: [{delta: {content: 我是}, index: 0}]} data: [DONE]网关要做的是透传不是解析重组。透传比把流式体转成语义化 JSON再下发稳得多因为各家在流式细节上都有小动作有的在最后一个 chunk 塞 usage 统计有的会多发一个空 delta解析重组会让网关变成新的故障点。我在 2.3 的代码里用 aiter_lines 逐行读、line.startswith(data:) 判断、原样 yield就是这个思路。结束标记 data: [DONE] 也原样透传调用方 SDK 靠它判断流结束。这里有个隐蔽坑日志中间件把流读掉。FastAPI 项目常挂 logging 中间件它为了打访问日志会把 response body 读一遍流式响应被消费后下游一个字都收不到。解法是日志中间件对 text/event-stream 类型直接跳过或者只记录状态码和耗时不碰 body。我线上版就是这么处理的。3.2 前端消费与 abort停止生成必须一级一级取消前端消费 SSE 有两条路。EventSource API 简单但只支持 GET而我们调大模型必须 POST否则 prompt 全暴露在 URL 里所以生产上更常见的是 fetch ReadableStream// 前端stream-chat.js const controller new AbortController(); async function streamChat(messages, model) { const resp await fetch(/v1/chat/completions, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ model, messages, stream: true }), signal: controller.signal, }); const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // SSE 以空行分隔每条 data需要攒包再切分 const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (!line.startsWith(data:)) continue; const data line.slice(5).trim(); if (data [DONE]) return; const chunk JSON.parse(data); const delta chunk.choices?.[0]?.delta?.content ?? ; if (delta) renderDelta(delta); } } } function stop() { controller.abort(); // 用户点停止生成时调用 }这段代码的关键是 signal 和 buffer 处理。signal 让前端 abort 后浏览器立刻断开与网关的连接buffer 的 split(\n\n) 处理粘包——SSE 是文本流一次 read 可能拿到半条 data 或多条 data必须攒起来按空行切分。我在 renderDelta 里只处理 delta.content丢弃 role 变更和 usage 统计的行避免界面出现无意义刷新。abort 最坑的地方在服务端前端断开连接上游厂商还在继续生成等于花着钱算没人看的 token。所以网关的透传生成器必须监听取消信号# gateway/stream.py 支持取消的流式生成器 import asyncio async def event_stream(route, headers, upstream_payload): async with httpx.AsyncClient(timeouthttpx.Timeout(600.0)) as client: async with client.stream( POST, route[endpoint], jsonupstream_payload, headersheaders, ) as resp: if resp.status_code ! 200: yield fevent: error\ndata: {(await resp.aread()).decode()}\n\n return try: async for line in resp.aiter_lines(): if line.startswith(data:): yield f{line}\n\n except asyncio.CancelledError: # 客户端断开主动关闭上游连接停止计费 await resp.aclose() raise流式响应被客户端断开时FastAPI 会向生成器抛 asyncio.CancelledError。这里必须把上游的 httpx 响应对象 aclose 掉否则 TCP 连接还挂着上游会一直把内容生成完。这是停止生成真正生效的最后一公里漏了这一步abort 就只是个前端假动作。3.3 鉴权与 Key 管理401 的根源大部分在配置不在代码网关鉴权分两层对外校验调用方拿的网关 Key对内替调用方保管各厂商 Key。对外最简单比对 Authorization 头里的 sk-agg- 前缀对内才是 401 高发区。DeepSeek 的典型报错是unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****这类 401 九成不是厂商封号而是配置污染。我踩过的三种一是 .env 里 Key 带了引号或行尾换行os.getenv 原样读入二是转发时把 Bearer 拼成 Bearer Bearer sk-xxx三是同一个 Key 同时填进了两个厂商变量后加载的覆盖了前面。所以 Key 读取我做了强制清洗# gateway/keys.py import os def get_upstream_key(provider: str) - str: env_key f{provider.upper()}_API_KEY key os.getenv(env_key, ).strip().strip().strip() if not key: raise RuntimeError(f{env_key} is empty, check .env) return key.strip() 去换行和空白再剥掉引号空值直接报错而不是传一个空串让厂商返回迷之错误。这一层加上之后401 在我这边基本绝迹。另外讯飞星火的 Key 比较特殊它要求 Authorization 传apiKey:apiSecret冒号拼接串我在 get_upstream_key 之外单独为 spark 写了一个 key 组装函数其他厂商一律走 Bearer 单 Key。Key 轮换也不能省各家平台控制台都有管理页面我习惯 90 天轮一次线上环境的 Key 通过环境变量注入不写进镜像。4. 本地部署与调用从 .env 配置到 curl、OpenAI SDK 双通道验证网关拿到手第一件事是在本机跑起来而不是直接上生产。这一章给 Windows 10 下的完整落地路径Python 版本管理、虚拟环境、Key 配置、启动验证、双通道调用。4.1 Windows 10 环境准备Python 版本管理器与虚拟环境我建议用 Python 版本管理器来装解释器而不是直接装系统 Python。原因很现实同时维护多个项目时A 项目要 3.10B 项目要 3.12没有版本管理器就只能互相卸载重装早晚翻车。Windows 10 上我常用 pyenv-win命令行装完后把 bin 目录和 shims 目录加进 PATH然后就能一条命令装任意版本并全局切换。# pyenv-win 安装完成后一键装解释器并切换 pyenv install 3.11.9 pyenv global 3.11.9 python --version # 应输出 3.11.9版本确认后建虚拟环境再装依赖。虚拟环境是 Python 项目的底线Windows 下直接 pip install 到全局权限问题和 DLL 冲突能折腾一整晚python -m venv .venv .venv\Scripts\activate # Windows 激活虚拟环境 pip install -U pip pip install fastapi[standard] httpx python-dotenvfastapi[standard] 会带 uvicorn 和标准依赖python-dotenv 负责读 .envhttpx 是网关访问上游的 HTTP 客户端。装完验证一下python -c import fastapi, httpx; print(fastapi.__version__, httpx.__version__)能输出版本号说明环境没问题再继续往下走。4.2 .env 配置与启动把十家 Key 一次配齐环境就绪后先写 .env。这份文件是网关的钥匙串每家一行暂时用不到的可以留空但至少配 DeepSeek 和智谱先把链路跑通# .env.example DEEPSEEK_API_KEYsk-svcac-xxxxx MOONSHOT_API_KEYsk-xxxxx DOUBAO_API_KEYxxxxx OPENAI_API_KEYsk-proj-xxxxx ANTHROPIC_API_KEYsk-ant-xxxxx ERNIE_API_KEYxxxxx QWEN_API_KEYsk-xxxxx SPARK_API_KEYxxxxx ZHIPU_API_KEYxxxxx HUNYUAN_API_KEYxxxxx # 网关对外校验的 Key逗号分隔可配多个 GATEWAY_API_KEYSsk-agg-local启动网关# 开发调试模式--reload 让代码改动自动重启 uvicorn gateway.app:app --host 127.0.0.1 --port 8000 --reload看到 Application startup complete 之后先别急着调模型。访问 http://127.0.0.1:8000/docs确认 FastAPI 自动生成的接口文档里 /v1/chat/completions 存在这一步能直接排除路由挂载问题比盲调 curl 少一轮是不是路径写错的猜测。4.3 双通道验证curl 确认链路OpenAI SDK 确认兼容性网关是不是真OpenAI 兼容得用两个通道同时验证。curl 验证原始链路OpenAI SDK 验证协议兼容两个都过才算数。先 curl# -N 关掉 curl 缓冲区才能看到逐字输出 curl -N http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer sk-agg-local \ -d {model: deepseek-chat, messages: [{role: user, content: 用一句话介绍你自己}], stream: true}正常的话终端里会一行一行蹦出 data: 开头的文本。接着用 OpenAI 官方 SDK 打同一个路径验证协议兼容层# sdk_check.py from openai import OpenAI client OpenAI( base_urlhttp://127.0.0.1:8000/v1, # 指向网关不是官方 api_keysk-agg-local, # 网关 Key不是厂商 Key ) resp client.chat.completions.create( modeldeepseek-chat, # 网关内模型 ID messages[{role: user, content: 写一个 Python 快速排序}], streamTrue, ) for chunk in resp: delta chunk.choices[0].delta if delta and delta.content: print(delta.content, end)这段能跑通说明兼容层是合格的OpenAI 官方 SDK 都认业务侧接进来就不用改代码。两个参数是关键接线点base_url 指向网关根路径末尾必须带 /v1api_key 用网关签发的 Key 而不是上游厂商的 Key。这两个写错是新手上路最常见的翻车点。验证完 DeepSeek把 model 换成 glm-4-plus 再跑一次同一段代码就是完整的一键切换演示。5. 避坑指南聚合服务最常见的五个现场故障提示以下五条全部按现象 → 原因 → 解决整理可当排查手册直接抄。5.1 DeepSeek 401Key 被引号和换行污染现象调用 deepseek-chat 返回unexpected status 401 unauthorized: incorrect api key provided: sk-svcac****Key 肉眼看着是对的复制到官网控制台发起请求却完全正常。原因.env 文件里 Key 被包了引号或行尾带了不可见字符os.getenv 原样读入转发时 Authorization 头带着脏字符。另外转发代码把 Bearer 拼了两遍、变成 Bearer Bearer sk-xxx 的情况也很常见。解决Key 读取统一走 3.3 的 get_upstream_key强制 strip 和剥引号还不行就在代码里 print(repr(key)) 看首尾有没有隐藏字符。改完必须重启 uvicorn因为 .env 的修改不会热加载。从那以后我没在 401 上浪费过时间。5.2 400 上下文超长最大上下文 1048576 tokens 的边界现象把长对话历史全量发给 deepseek-reasoner报api error: 400 this models maximum context length is 1048576 tokens明明没聊那么多轮。原因reasoner 模型上下文窗口很大但请求里 system prompt、工具定义、历史消息全塞进去后某些字段按 token 计算会迅速膨胀网关透传时又原样转发没做任何裁剪于是顶到上限。解决在网关层加滑动窗口裁剪保留 system 和最近若干轮丢弃最早的消息。我的经验值是普通对话保留最近 20 轮、总字符超过 120000 就从头丢system 永远保留。裁剪后这类 400 基本消失。注意 reasoner 类模型对推理连贯性敏感批量对话场景建议同时调小 max_tokens给上下文腾位置。5.3 Windows 下 Docker API 连不上npipe 报错与替代方案现象Windows 10 上执行 docker build 或 docker run报failed to connect to the docker api at npipe:////./pipe/docker-desktop-linuxDocker 命令全部失效。原因Docker Desktop 没启动或 WSL2 后端已经崩了npipe 管道不存在。Windows 的 Docker 走虚拟化后端和 Linux 原生环境不一样经常出现服务看起来在、管道却不通的假象。解决先打开 Docker Desktop 等它显示 Engine running再执行docker context ls确认当前 context 是 desktop-linux。如果还连不上本地调试完全绕开 Docker直接python -m uvicorn gateway.app:app --port 8000跑代码等逻辑稳定了再回 Docker 打包。别在这上面耗时间它就是个环境问题不是代码问题。5.4 流式输出被攒包nginx 缓冲把 SSE 变成一次性返回现象前端调用网关界面空白很久然后所有内容一次性全部渲染出来完全没有逐字效果绕过 nginx 直连 8000 端口却正常。原因网关前面挂了 nginx 做流量入口nginx 默认开启 proxy_buffering会把上游的流式响应攒满缓冲区才发给客户端SSE 就此失去流式意义。解决在 nginx 的 location 里关掉缓冲并把读超时调大防止长时间推理被掐断location /v1/chat/completions { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_buffering off; # 关键关掉缓冲 proxy_cache off; proxy_read_timeout 300s; # 推理时间长必须放大 proxy_set_header Connection ; }改完nginx -t校验语法再 reload。如果网关后面还套了 CDNCDN 的缓冲同样要关否则公网环境会在同一坑里再摔一次。5.5 abort 之后上游还在烧钱连接没有真正取消现象用户点了停止生成前端界面停了厂商后台的 token 用量却还在涨或者 abort 几次之后后续请求全部变慢。原因前端 abort 只断了浏览器与网关之间的连接网关生成器没监听 CancelledError上游 httpx 连接保持打开厂商继续生成直到自然结束。请求变慢则是频繁断连导致 TCP 连接反复重建。解决生成器里加 try/except asyncio.CancelledErrorexcept 分支中 await resp.aclose() 关闭上游响应再 re-raise代码在 3.2 节已给出。同时用 httpx.AsyncClient 做连接复用避免每个请求都重新握手高频 abort 场景下能省下可观的建连开销。6. 进阶把网关变成多模型评测台顺带把账算清楚网关跑稳之后我做的第一件事是给每个流式请求记录三个指标首字延迟、总耗时、输出 token 数。有了这三项数据网关就从转发工具升级成多模型评测台——切换模型之前先拿数据说话而不是听销售和社区吹。实现思路是在 StreamingResponse 外层包一层记录逻辑流结束后把统计写入内存或 SQLite。核心代码就两段一段在请求结束钩子里记录指标一段是对同一组 prompt 轮询所有模型的评测脚本# gateway/metrics.py 在流结束时记录指标 async def record_stream(route, event_stream): t0 time.time() first_token_at None total_chars 0 for line in event_stream: if first_token_at is None and line.startswith(data:): first_token_at time.time() total_chars len(line) yield line SESSION[requests].append({ provider: route[provider], first_token_ms: int((first_token_at - t0) * 1000), total_ms: int((time.time() - t0) * 1000), chars: total_chars, })# eval/benchmark.py 同题多模型对比 import asyncio, time from openai import OpenAI client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keysk-agg-local) MODELS [deepseek-chat, qwen-plus, glm-4-plus, hunyuan-turbo] PROMPTS [用 Python 写一个指数退避重试的 HTTP 客户端, 解释 SSE 与 WebSocket 的适用边界] async def run(): for model in MODELS: for prompt in PROMPTS: t0 time.time() first None n_chars 0 resp client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], streamTrue, ) for chunk in resp: delta chunk.choices[0].delta if delta and delta.content: if first is None: first time.time() - t0 n_chars len(delta.content) print(model, round(first, 2), round(time.time() - t0, 2), n_chars) asyncio.run(run())跑完会得到一张非常直观的对比表模型首字延迟(s)总耗时(s)输出字符数现场印象deepseek-chat0.43.1412快中文自然glm-4-plus0.62.8386稳代码一般hunyuan-turbo0.53.6405中规中矩把各家官网的每百万 token 价格配进路由表这次评测每次调用花了多少钱就是一道算术题再叠加首字延迟答案一目了然。完整的路由表、SSE 转发代码、评测脚本我都整理在资源包里下载后照着 .env.example 配 Key 就能跑起来。从那以后我每次把新模型接进网关都强制走一遍注册路由 → 单发冒烟 → 流式压测 → 评测对比这四步顺手把 Key 的过期时间写进提醒日历再也没有半夜被 401 叫醒过。希望帮到你。本文还有配套的精品资源点击获取