☰
自建AI算力站:统一API网关与模型路由实现高性价比大模型调用
2026/10/10 13:29:08 网站建设 项目流程

1. 为什么我要搭一个自己的算力站

先交代一下背景。我手里有几个AI相关的应用在跑,平时需要调用大模型做推理、embedding、摘要生成这些事情。一开始图省事,直接买了各家云厂商的按量付费API,每个月账单下来,心里总咯噔一下。后来算了一笔账,发现我这种有一定固定调用量的场景,按量付费其实很亏。于是我开始琢磨一件事:能不能自己搭一个“算力中转站”,对外暴露一个统一的API接口,对内把请求调度到多个后端模型上,既能摊薄成本,又能灵活切换模型?

这个想法最终落地的产物,就是标题里说的“Praka API 算力站”。说白了,它不是一个物理上的机房,而是一套软件架构:一个统一的API网关层、一层请求调度与路由逻辑、一组后端模型适配器,再加一套鉴权与计量计费模块。通过这套东西,我可以把同一个OpenAI风格的请求,转发给智谱、DeepSeek、讯飞星火或者其他兼容接口的大模型,然后按自己的成本策略把最合适的请求分给最合适的模型。

文章适合谁看?如果你也是以下情况之一,这篇内容应该能帮你少走不少弯路:

  • 手里有多家大模型API的Key,想统一管理、统一调用;
  • 每个月API账单太高,想通过调度和缓存把成本降下来;
  • 给团队或客户提供AI能力,需要一个带鉴权和用量统计的API服务;
  • 纯粹对AI基础设施感兴趣,想理解“算力站/API聚合层”是怎么运作的。

关于名字我多说一句。网上搜“Praka”可能查不到什么官方背景,我理解它更像是一个自造的项目代号。所以本文不带你搭某个特定品牌的收费服务,而是教你从零构建一套具备同等能力、同样叫“算力站”架构模式的系统。这样更有实操价值。

在展开之前,先给一个我个人的结论:所谓“高性价比的AI基础设施”,核心不在于买多便宜的GPU,而在于把“模型调用”这件小事工程化。省下来的钱和精力,远比追着算力价格跑更有意义。

2. 算力站的架构拆分:先看全局,再抠细节

要搭这样一个站,你不能上来就写代码。先把架构想清楚,后面每一步都是在填框架。我的落地架构是这样的:

  • 网关层:接收客户端请求,统一鉴权、限流、记录日志;
  • 路由层:根据模型名、成本、可用性把请求分发给真实后端;
  • 适配层:把各家OpenAI风格兼容/不兼容的接口做归一化;
  • 计量层:统计每次调用的token数、耗时、费用,用于账单分析;
  • 缓存层:对可缓存的请求做结果复用,减少重复调用。

这五个层次对应到代码上,就是一个Python写的异步服务。为什么用Python?因为生态里现成的OpenAI SDK、Anthropic SDK都很好接,写起来快;异步框架用FastAPI,性能也够用。如果你对性能有更高要求,之后可以用Go或Rust重写核心路由部分,但前期完全没必要。

来看一个最简单的请求流转示意:

客户端 │ POST /v1/chat/completions ▼ API 网关(FastAPI) │ 校验 API Key、检查限流 ▼ 路由策略模块 │ 根据 model 字段选择后端(例如 deepseek-chat / glm-4) ▼ 后端模型适配器 │ 调用智谱/DeepSeek/讯飞星火等真实接口 ▼ 响应归一化 │ 转成 OpenAI 格式返回给客户端

图里每一个环节都能展开很多内容,但大部分教程都只讲其中一环。我在下面几节会把每个环节的坑和选择都讲透,尤其是路由策略和成本控制这两个地方,这是我在实际使用中觉得回报最大的部分。

2.1 为什么需要“统一API层”而不是直接各家分开调

可能有人会问:我代码里直接写好多个模型厂商的SDK,每次调用if model == "glm" 走智谱,if model == "deepseek" 走DeepSeek,不也一样吗?

如果你的项目只有一个脚本,这样确实行,甚至更简单。但只要有第二个项目、第三个项目,或者要开放给团队其他人调用,你就面临一个很现实的问题:每个项目都要写一遍模型厂商的SDK接入代码,每个项目都要各自配置Key、维护各自的超时重试逻辑。时间久了,代码里到处都是模型厂商SDK的碎片,哪一天某家的接口升级了,你要跑遍所有项目去改。

统一API层能带来三个很实际的收益:

  1. 调用方不关心后端是哪家模型。调用方只知道“我发一个OpenAI格式的请求出去,拿到OpenAI格式的响应”,至于后端是智谱还是DeepSeek,对调用方透明。万一某天智谱的接口不稳定,我切换到DeepSeek,调用方完全无感。
  2. 鉴权、限流、计量可以集中做。团队里每个人用自己的Key访问我的算力站,我能在网关层看到每个人的调用量、token消耗,方便内部结算。
  3. 切换模型成本趋近于零。今天GLM-4降价了,我把策略里的A/B切换一改,立刻切换;不需要改任何调用方代码。

所以我的建议是:只要你有可能面对“多项目共用模型能力”、“多人共用模型Key”、“模型厂商经常变动”这三个情况之一,统统一层是值得的。前期花一两天搭建,后面省下来的时间绝对远超投入。

3. 网关层设计:鉴权、限流和API Key管理

网关层是算力站的门面,所有请求都从这里进。如果这一层做得糙,后面的一切都会乱。

3.1 API Key的生成与校验

我采用的是最简单的方案:服务端生成随机字符串作为API Key,存哈希值,客户端请求时放在Authorization: Bearer <key>头里。别看简单,里面有两个容易踩坑的细节:

  • 不能明文存储API Key。用户在控制台创建Key时,只显示一次明文;数据库里存的是它的SHA-256哈希。这样即使数据库泄露,攻击者拿到的也是哈希,无法直接使用。
  • Key要有前缀。我会生成类似pk-live-xxxxxxxxxxxx这样的格式,前缀用于标识环境(live还是test),也方便日志里排查问题。

生成Key的代码大致这样:

import secrets import hashlib def create_api_key(env: str = "live") -> str: raw = f"pk-{env}-{secrets.token_hex(24)}" return raw def hash_key(raw: str) -> str: return hashlib.sha256(raw.encode()).hexdigest()

Key校验的地方用FastAPI的依赖注入做,每个请求进来先检查Authorization头,再用哈希比对数据库,未通过直接返回401。这个过程要放在路由处理之前,保证任何未认证请求都进不了业务逻辑。

3.2 限流:防止Key被滥用

没有限流的API站就像一个不设防的服务器,随时可能被脚本刷爆。我主要做两层限流:

  • 第一层:每个Key每分钟最多允许N次请求;
  • 第二层:每个Key每分钟最多允许消耗M个token(针对流式响应要单独处理)。

实现上我用的是令牌桶思路,可以用Redis来做这个统计。每次请求进来,从Redis里取当前Key的计数器,判断是否超过限制,没超过就放行并把计数器累加。这个逻辑看起来简单,但要注意并发场景下的原子性。建议用INCR+EXPIRE命令组合,或者直接用一个Lua脚本保证原子性,别用“先GET再SET”的方式,两个请求同时进来会竞争。

一个比较实用的限流中间件骨架:

import time import redis r = redis.Redis(host="localhost", port=6379, decode_responses=True) async def check_rate_limit(api_key: str, max_req: int, window: int = 60): key = f"ratelimit:{api_key}:{int(time.time() // window)}" count = r.incr(key) if count == 1: r.expire(key, window) if count > max_req: raise HTTPException(status_code=429, detail="Rate limit exceeded")

这里用“时间窗口取整”的方式做计数,窗口切分处会有一次突发放行,但对我们这种场景足够用了。如果你要做更平滑的限流,再上令牌桶算法的完整实现。

3.3 Key的层级设计

我建议不要把所有的Key都当成同一个权限级别。做个三级划分:

Key类型用途权限
管理员Key管理控制台、查看用量、修改策略全权限
项目Key某个项目专用,指定可用模型范围可配额
临时Key给测试或外部合作者短期使用限量且过期

这个层级设计在你规模小的时候看着多余,但只要开始有第二个项目接入,立刻就能体会到好处。我在搭的时候没有这个设计,后来为了给一个外包项目开权限,临时手改了数据库,那叫一个难受。

4. 路由策略:怎么把请求送到最合适的模型上

这一层是“算力站”的算力调度核心,也是性价比的关键。很多人理解的路由就是写死一个模型名,其实可以做得更聪明。

4.1 模型路由的核心属性

每次请求进来,路由层至少需要知道四个维度:

  • 用户的模型名:比如请求里写model: "auto",表示不指定具体模型;
  • 请求类型:chat、embedding、image,不同任务映射到不同的后端模型;
  • 成本预算:这个请求允许最多花多少钱;
  • 延迟要求:流式/非流式,同步/异步调用。

我设计的配置方式是用一个YAML文件定义路由策略,改动起来很方便,不用改代码:

routes: - match: model: "auto" purpose: "chat" budget: "low" backend: provider: "deepseek" model: "deepseek-chat" - match: model: "auto" purpose: "chat" budget: "medium" backend: provider: "zhipu" model: "glm-4" - match: model: "auto" purpose: "embedding" backend: provider: "zhipu" model: "embedding-2"

请求进来后,路由模块提取用户携带的预算信息和请求类型,匹配到第一个符合规则的后端,然后进入适配层执行真实调用。

4.2 “模型降级”与“失败转移”两个关键策略

这两个策略是画龙点睛的地方。我用文字描述一下它们解决的问题:

  • 模型降级:假设默认走GLM-4,但该厂商的接口正在高峰期,延迟飙到10秒,这时候路由层可以把请求自动切换到DeepSeek,保证用户侧响应时间不超预期。
  • 失败转移:假设智谱接口返回500错误,或者触发限流,路由层不要直接把错误抛给用户,而是记录错误后自动重试到下一个可用后端。

实现降级和转移的核心是“后端健康状态管理”。我会用一个后台任务定时探测每个provider的连通性、平均延迟、错误率,把这些数据实时更新到一个状态表里。路由层在选择后端时,先过滤掉不健康的provider,再按策略排序。

这个逻辑在代码里就是一个select_backend(model, purpose, budget, health)函数。健康状态由后台线程更新,不用每次请求都实时探测,否则探测请求本身就成了额外开销。

4.3 成本感知路由:我的省钱利器

这个要单独拿出来说。所谓“高性价比”,很大一部分就体现在成本感知路由上。

我到月底会拉一份账单,统计每个请求在每家模型的token单价、实际消耗。结果发现,我的流量里有很多是“摘要生成”“相似度计算”这类不需要强大模型的场景。之前我统一用最贵的旗舰模型跑,浪费严重。

后来我在路由策略里加了“cost_band”概念,把用户请求分为三个档位:

  • high:需要高质量强推理能力,比如代码生成、复杂分析,走GLM-4或DeepSeek-R1这类强模型;
  • medium:一般对话、简单问答,走DeepSeek-Chat或同级别模型;
  • low:分类、抽取、摘要等任务,走轻量模型,甚至本地小模型。

用户调用时如果不指定档位,网关会通过提示词长度和任务类型自动判断。实际用下来,medium和low档的请求占比超过60%,账单比之前下降了一半多。

注意:路由策略要遵守一个原则——不要为了省钱把用户的请求“缩水”处理。如果用户明确要求某个模型,或者请求本身确实需要强推理,路由层必须优先满足质量,而不是强行降级。省钱和省心要平衡,不能顾此失彼。

5. 后端模型适配层:兼容各家API的真实做法

如果所有模型厂商都提供OpenAI兼容接口,适配层就不存在了。现实情况是,智谱、DeepSeek、讯飞星火等各家接口虽然都在向OpenAI风格靠拢,但仍有差异。适配层就是把差异抹平。

5.1 统一请求格式与响应格式

我的算力站对外暴露的是OpenAI的chat/completions格式。这样做的好处是,调用方可以直接使用openai这个Python库,把base_url指向我的算力站就行。示例:

from openai import OpenAI client = OpenAI( api_key="pk-live-xxxx", base_url="https://your-gateway.example.com/v1" ) resp = client.chat.completions.create( model="auto", messages=[{"role": "user", "content": "你好"}], )

这个调用发出之后,我的网关内部会做:

  • 解析请求参数;
  • 根据路由策略选定真实后端模型;
  • 把OpenAI格式的请求转换成对应厂商的请求格式;
  • 调用厂商SDK;
  • 把厂商响应转换回OpenAI格式。

转换过程中最容易出问题的点有三个:

  • System Prompt的携带方式:有的厂商要求把system角色转成特殊字段,有的要求直接用messages传;
  • 参数命名差异:有的用temperature,有的用top_p,有的厂商对参数范围限制不同;
  • 返回字段差异:finish_reason的取值,有的返回stop,有的返回end;usage字段里的token统计口径也不同,有的统计总数,有的还要拆分prompt_tokens和completion_tokens。

这些细节只能逐个厂商踩坑后记录下来,建立一个映射表。我建议你在适配层用一个ProviderAdapter基类,每个厂商一个子类,里面实现build_request和parse_response两个方法,这样新增厂商时只需要写一个新的子类,不用动其他代码。

5.2 流式输出与代理

现在很多AI应用都用流式输出(SSE),一边生成一边打字。适配层如果只是简单地把整个响应转完再返回,用户会感觉特别慢。所以要支持流式转发。

流式转发的思路是:前端请求进来,网关立即开一个流式请求到后端厂商,厂商返回一个生成器逐条推送数据块,网关把每个数据块改写成OpenAI格式的SSE事件,再直接转发给客户端。这样做的好处是延迟几乎等同直连厂商。

坑在于:

  • 每个数据块的字段格式必须一致,否则客户端解析会报错;
  • 连接中断时,网关要能自动断开对应的后端请求,否则会漏掉一些数据块;
  • 数据块中choices[0].delta.content可能为空,只有finish_reason,这种也要正常转发。

我这里简单给一个流式转发的伪代码思路:

async def proxy_stream(request, backend_adapter): async for chunk in backend_adapter.stream(request): openai_chunk = normalize_stream_chunk(chunk) yield f"data: {json.dumps(openai_chunk)}\n\n" yield "data: [DONE]\n\n"

注意,[DONE]标记是OpenAI风格的流式结束符号,有些后端没有,你需要自己补上。

5.3 超时与重试机制

网络请求总会遇到超时,关键是设置合理的超时时间和重试策略。我的默认配置:

  • 连接超时:10秒;
  • 读取超时:50秒(流式请求读取超时加大到300秒);
  • 重试次数:2次;
  • 重试间隔:指数退避,1秒、2秒、4秒。

要在适配层做重试,但注意不要重试“不可重试的请求”。比如因为请求参数错误返回400,就不要重试;因为服务端5xx或网络中断,则要重试。这个判断逻辑用厂商返回的HTTP状态码区分就好。

async def call_with_retry(adapter, request, max_retries=2): for attempt in range(max_retries + 1): try: return await adapter.call(request) except RetryableError as e: if attempt == max_retries: raise await asyncio.sleep(2 ** attempt)

这里我专门定义了RetryableError异常,只有这类异常才触发重试。不要把超时和参数错误混为一谈,否则浪费的请求会拖慢整个网关。

6. 计量、缓存与成本优化实践

如果说路由层是决定“把请求发给谁”,那计量和缓存就是决定“这笔请求到底花多少、能不能不花”。实操下来,这两个模块对成本降低的贡献非常直接。

6.1 请求计量与token统计

我开发了三个维度去统计:

  • 按API Key维度:看每个Key的调用量、token数、费用;
  • 按模型维度:看每个模型占用的成本比例;
  • 按时段维度:按天/按小时看调用分布,方便发现低峰期。

实现上,每个请求完成之后,把计费信息异步写入数据库(或者先写入消息队列,再批量落库)。注意,流式请求的token数必须在流式结束后才能拿到,所以计量逻辑要挂在流式输出收尾之后,不能挂在请求进来时。

统计用量的数据库中,一张简单的表结构如下:

字段类型说明
request_idUUID请求ID
api_key_hashstringKey哈希
providerstring实际后端
modelstring实际模型名
prompt_tokensint输入token数
completion_tokensint输出token数
cost_centsdecimal本次费用(美分)
latency_msint总耗时
created_atdatetime时间

这张表要建立好索引。我一开始没加索引,统计一张100万行的表时查询慢得离谱,后来补了api_key_hash和created_at联合索引才好转。

6.2 语义缓存:高性价比的关键一招

这是全文最值得你直接抄走的一个优化点。

很多AI应用的请求是重复的。比如用户问“什么是大模型”,总共问了100次,你返回了100次,花了100次的钱;但如果第一次把结果缓存下来,后99次直接命中缓存,成本几乎为零。

这里要注意的坑是:不能用“精确文本匹配”做缓存,因为用户表达同一意思的文本可能不同。比如“怎么用Python”和“Python如何使用”意思相近但文本不同。我尝试过两种方案:

  • 方案一:把用户问题的embedding计算出来,存到向量数据库,新请求先算embedding,再查找余弦相似度大于0.95的已有缓存。这个方案效果好,但每次都要算embedding,本身就是一次调用成本。
  • 方案二:用短文本归一化,比如去掉标点、小写化、替换同义词,再做哈希匹配。这个方案成本更低,但命中率不如向量法。

我的推荐做法是“分级缓存”:

  1. 对于完全相同的请求文本(字符级相同),直接精确匹配缓存;
  2. 对于语义相似但文本不同的,用embedding缓存,但只对“低预算档位”的请求开启。
  3. 缓存TTL根据场景配置:事实性问答可以缓存24小时;需要实时性的问题不缓存。

缓存存储后端用Redis就够了,如果要存embedding向量,Redis也可以用redisearch模块或直接存JSON读出来算,前期量不大完全顶得住。

6.3 异步批处理:把同步请求降成本

如果你的使用场景里有大量非实时的AI调用,比如给一批历史文章生成摘要,没有必要每个请求都实时进出。我搭了一个异步任务队列:请求先进入队列,后台worker批量从队列里取任务,调用适配层并发发送给模型,生成完成后把结果存到存储里,用户通过回调或查询接口拿结果。

这样做有两个好处:

  • 批量调用可以拿到更好的并发效率,减少总体等待时间;
  • 可以选择低峰时段调用模型,有些厂商有夜间折扣,这样可以进一步压成本。

异步批处理不复杂,核心就是一个消息队列加几个worker。前期用Redis做队列就行,单机跑几十万条消息问题不大,不要一上来就上Kafka。

7. 部署与运维:记录我踩过的坑

代码写完只是第一步,部署运维才是确保长期稳定运行的关键。

7.1 部署形态选择

我推荐直接用Docker部署,一个容器里跑网关服务,一个容器跑Redis,MySQL/PostgreSQL用托管服务或者单独容器。这是最简单、最容易迁移的组合。

Docker镜像并不多复杂,重点在环境变量的配置上。建议把数据库连接、模型Key、限流参数全部放到环境变量里,不要让密钥出现在镜像里。

我的docker-compose.yml结构大致是这个样子:

services: gateway: build: . ports: - "8000:8000" environment: - DATABASE_URL=postgresql://... - REDIS_URL=redis://redis:6379 - ZHIPU_API_KEY=xxx - DEEPSEEK_API_KEY=xxx depends_on: - redis redis: image: redis:7 ports: - "6379:6379"

配合Nginx做反向代理和HTTPS终止。我强烈建议网关服务不要直接暴露到公网,一定要经过Nginx或云负载均衡。

7.2 监控告警的几个加分项

没有监控的系统,出问题了你是最后一个知道的。我搭建时先上了三样:

  • 请求成功率:低于99%就告警;
  • 平均延迟:P95延迟超过5秒就告警;
  • 账户余额/成本异常:每日成本环比超过20%就告警。

这几个指标从数据库或日志里很容易算出来,可以在服务里做个定时任务,把计算结果推到钉钉或飞书群机器人。别贪多,先把这三个守住,其他的后面再说。

7.3 遇到的典型案例:模型接口DPA与数据一致性

有一阵子,我的算力站频繁出现“用户请求超时”。排查时发现不是网络问题,而是某家模型的接口在特定时段会随机进入慢速状态。当时没有健康检查,所有请求都硬等50秒超时。

后来我在健康检查模块里加了一个“慢响应降级”规则:如果最近5分钟内某provider的P50延迟超过阈值,就自动把流量降到备选provider。这样再也不需要用户侧感受到超时。这个经验很直接:别等接口完全挂了才切换,延迟超过容忍线就要切。

8. 从小规模扩展到生产级:我建议的演进路径

如果你看完前面的内容开始动手,我给出一条适合大多数人的平滑路线,不要一上来就把所有模块做完。

第一步:只做简单的网关转发。一个API Key,一个固定后端,一个OpenAI格式转换就够了。能跑通,就能把项目用起来。

第二步:接入第二个模型厂商,写适配层。不要图省事直接写第三家。适配层一旦抽象好了,第三家就是复制粘贴。

第三步:加入路由策略和健康检查。这时候你的算力站才算得上“算力调度”。

第四步:加入计量和缓存,开始优化成本。这一步就是要看账单下降的时候了。

第五步:加入限流、审计、监控告警,让它变成一个可以对外提供服务的稳定产品。

我走完这套路线大约用了一个月业余时间,真正的冲刺是在周末。每一个阶段的代码量都不大,但每走一步,系统能力上升一个台阶,非常上头。

9. 最后分享几点真实体会

搭建这套算力站,我很深的体会是:AI基础设施的“高性价比”不完全是钱的问题,更是灵活性和掌控感的问题。

我实际使用中最大的收获不是账单降了多少,而是我再也不怕某一家模型厂商涨价或不稳定了。对我来说,所有模型都是“插件”,今天这个不行就换那个,明天那个便宜就多走那个,主动权始终在自己手里。这种感觉,用钱不一定买得到。

另外,一个非常实际的提醒:不要把模型厂商的免费额度当作架构设计的基石。免费额度再多,一旦你依赖它,总有一天会被收费的通知弄个措手不及。正确的做法是搭好路由和降级机制,任何一家都可以随时换掉,免费额度只是锦上添花。

如果你也打算搭一套自己的算力站,我的建议很简单:先把手头一个真实项目接上来跑,按上面的五个阶段慢慢演进。重要的是先把“网关转发”跑通,你很快就能看到它的价值。后面的路由和缓存,都是在实际使用中才会越想越清楚的优化。祝你也早日拥有自己的高性价比AI基础设施。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询