从零搭建API封装SaaS：统一网关、Token计费与限流实践

把一个上游模型 API 包装成面向特定场景的 SaaS 服务核心不是把 HTTP 请求转发一遍而是要在转发之外解决鉴权、计量、计费、限流、错误隔离和用户管理。很多人看到“API 封装成 SaaS 能赚钱”这个方向第一反应是搭一个转发服务收差价实际操作时才发现账算不清楚、密钥暴露、上游一抖动用户就流失。这篇文章会从零开始搭建一个最小可运行的 API 封装 SaaS包括用户体系、API Key 签发、统一模型网关、token 计量、余额扣减、限流和常见报错排查。学完后你可以把这条链路直接用于接一个提示词生成服务、一个模型代理服务或者一个部门内部的公共模型网关。严格来说本文不承诺“一定会盈利”但会把收费闭环真正跑通让商业模型有被验证的基础注册、发 Key、调用模型、按 token 扣费、查调用日志。没有这套技术底座后面的定价、套餐、优惠和运营都无从谈起。1. 为什么“API 包装成 SaaS”不是搭一个转发服务那么简单1.1 一次 API 调用背后有 6 个隐藏环节用户调用你的 SaaS 接口时他看到的是一次普通 HTTP 请求。从服务端视角看一次请求至少要经过下面 6 个环节认证判断请求头里的 API Key 属于哪个用户。授权判断这个用户有没有权限调用某个模型是否触达套餐上限。参数校验检查模型名、消息格式、temperature、max_tokens 等参数是否合法。上游转发带着你的上游密钥去请求真实模型服务。计量与计费根据上游返回的 usage 信息计算 token 消耗和费用。日志与监控记录调用结果、错误信息、耗时和成本。只做转发的服务要么把这 6 个环节全部省略要么直接在代码里用 if 判断凑合最后账目一塌糊涂。真正值得做的 SaaS是在这 6 个环节上沉淀通用能力让用户不用自己维护某个模型厂商的 SDK、模型参数和密钥体系。1.2 这个模式成立的三个条件“把一个 API 包装成 SaaS 能不能赚钱”取决于三点。第一是价值差。如果用户直接调用上游 API 的成本是 100 元你的服务通过模型路由、缓存、上下文压缩把成本压到 60 元同时你只收 80 元中间 20 元就是你的毛利。价值差越大用户越愿意用。第二是使用门槛。很多用户不会写复杂的请求体不理解 temperature 和 top_p 的区别也不了解不同模型适合什么场景。你的 SaaS 如果提供了一个更友好的业务接口把“调用模型”变成“实现业务功能”用户会愿意付费。第三是错位竞争。直接在通用模型能力上和上游厂商竞争没有任何优势。更常见的做法是聚焦某个细分场景比如客服工单分类、小红书文案生成、招聘 JD 解析、合同要素抽取。你的抽象层做得越好越难被上游直接替代。1.3 从零到可运营需要先打通四条链路把项目落地到可运营状态至少需要四条链路。用户链路注册、登录、创建 API Key、查看余额、查看调用记录。 请求链路客户端携带 Key 请求网关网关鉴权、计量、转发上游再把结果返回。 计费链路上游返回 usage网关计算 token扣减用户余额写入调用日志。 观察链路管理员能看到每个用户、每个 Key、每个模型的调用量、成功率和成本。本文的架构图如下客户端 - FastAPI 网关 - 上游模型 API期间经过认证、限流、计费和日志。实际操作时可以用一张表把用户、Key、调用记录串起来先把最小闭环跑通再逐步增加功能。2. 技术选型和项目骨架先把服务跑起来2.1 技术选型FastAPI SQLite Redis 的取舍对于一个刚刚起步的 API 封装 SaaS技术选型不必复杂。下面这套组合适合学习环境和早期小规模运营组件选择说明Web 框架FastAPI异步性能好自带请求校验和 OpenAPI 文档写接口很快数据库SQLite - PostgreSQL学习阶段用 SQLite生产环境换成 PostgreSQL缓存Redis用于限流计数、短时令牌缓存、上游响应缓存HTTP 客户端httpx支持异步适合转发上游请求密钥管理环境变量 密钥哈希不在代码里硬编码上游密钥用户 Key 只存哈希FastAPI 的优点是自带参数校验能用 Pydantic 模型定义请求体出错时能返回清晰的 422 错误。对于 API 服务来说这一点能节省大量参数兼容工作。注意SQLite 不适合高并发写入如果单机 QPS 超过几十建议尽早切换到 PostgreSQL。早期用 SQLite 只是为了让项目跑通不代表它是生产选择。2.2 项目目录结构沿用分层思路目录结构如下api-saas/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── database.py │ ├── models.py │ ├── schemas.py │ ├── security.py │ ├── gateway.py │ ├── billing.py │ └── routers/ │ ├── __init__.py │ ├── auth.py │ ├── chat.py │ └── admin.py ├── requirements.txt ├── .env.example └── README.md把认证、网关、计费拆成独立模块是为了让日志、排错和后续扩展更方便。如果所有逻辑都堆在 main.py 里等你要加套餐、加模型路由时代码会很痛苦。requirements.txt 内容fastapi0.115.6 uvicorn[standard]0.34.0 sqlalchemy2.0.36 pydantic2.10.4 pydantic-settings2.7.1 httpx0.28.1 redis5.2.1 passlib1.7.4 python-jose[cryptography]3.3.0这里没有写死具体版本时落地前要重新确认依赖兼容性。FastAPI 和 Pydantic 的版本绑定比较紧建议直接按这个组合安装。2.3 数据库表设计最小闭环只需要 4 张表用户表、API Key 表、调用记录表、充值记录表。CREATE TABLE users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT UNIQUE NOT NULL, balance_cents INTEGER NOT NULL DEFAULT 0, is_active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL ); CREATE TABLE api_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, key_prefix TEXT NOT NULL, key_hash TEXT NOT NULL, revoked INTEGER NOT NULL DEFAULT 0, expires_at TEXT, created_at TEXT NOT NULL ); CREATE TABLE call_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key_id INTEGER NOT NULL, user_id INTEGER NOT NULL, model TEXT, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, cost_cents INTEGER DEFAULT 0, status INTEGER, error_message TEXT, created_at TEXT NOT NULL ); CREATE TABLE recharge_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, amount_cents INTEGER NOT NULL, created_at TEXT NOT NULL );余额字段使用整数类型单位用“分”。原因很简单浮点数在金额计算上会产生精度问题用分可以避免类似 0.1 0.2 的误差。所有涉及钱的计算都按整数处理最后展示时再除以 100。API Key 只存哈希不存明文。这样可以避免数据库泄露时用户 Key 全部暴露。2.4 用户注册与 API Key 签发用户注册接口先提供一个最简单的版本用户名或邮箱加密码即可。生产环境还需要邮箱验证、找回密码和风控这里先省略。# app/routers/auth.py import secrets import hashlib from datetime import datetime, timedelta from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.database import get_db from app.models import User, ApiKey from app.schemas import RegisterRequest, ApiKeyResponse router APIRouter() def hash_sha256(value: str) - str: return hashlib.sha256(value.encode()).hexdigest() router.post(/register, status_code201) def register(payload: RegisterRequest, db: Session Depends(get_db)): exists db.query(User).filter(User.email payload.email).first() if exists: raise HTTPException(status_code400, detailemail already registered) user User( emailpayload.email, balance_cents0, created_atdatetime.utcnow().isoformat(), ) db.add(user) db.commit() db.refresh(user) return {id: user.id, email: user.email} router.post(/api-keys, response_modelApiKeyResponse) def create_api_key(db: Session Depends(get_db), user_id: int Depends(get_current_user)): plain_key sk- secrets.token_urlsafe(32) key_entity ApiKey( user_iduser_id, key_prefixplain_key[:12], key_hashhash_sha256(plain_key), expires_at(datetime.utcnow() timedelta(days90)).isoformat(), created_atdatetime.utcnow().isoformat(), ) db.add(key_entity) db.commit() return {api_key: plain_key, prefix: key_entity.key_prefix, expires_at: key_entity.expires_at}这里的关键点有两个。第一token_urlsafe(32)生成的随机串足够长暴力猜解几乎不可能。第二key_prefix只存前 12 位用于在后台展示给用户识别是哪个 Key完整明文只在创建时返回一次。注意不要直接在注册接口里同时创建 API Key。注册和创建 Key 是两件事混在一起会让后续做套餐、试用额度时很难扩展。3. 实现统一模型网关把上游 API 安全地封装起来3.1 为什么不能直接把上游地址暴露给用户如果直接把上游 API 的 Base URL 和 Key 给用户表面上省掉了网关实际上损失了四个东西无法控制用户调用量、无法统计真实成本、无法统一处理错误、上游 Key 一旦泄露就是灾难。统一网关的价值是把上游密钥收在服务端用户只认你的 Key。你的 Key 可以定期轮换用户的 Key 也可以独立吊销两边互不影响。3.2 统一请求格式与模型白名单第一版请求体可以设计成主流模型厂商通用的 OpenAI 兼容风格{ model: demo-model, messages: [ {role: system, content: 你是一个专业的文案助手}, {role: user, content: 写一段咖啡店的介绍} ], temperature: 0.7, max_tokens: 500 }网关要做一次白名单校验。并不是上游支持什么模型你就开放什么模型而是你的套餐里包含哪些模型才放行哪些。模型白名单可以直接存在配置里# models.yaml 的示意结构 models: demo-model: upstream_model: actual-upstream-model-name context_window: 4096 price_per_million_tokens: 5 demo-fast-model: upstream_model: actual-fast-model-name context_window: 8192 price_per_million_tokens: 2这样一个对外模型名可以映射到不同的上游模型后续换厂商、换模型版本时不用通知用户改参数。3.3 调用上游 OpenAI 兼容接口的核心代码网关模块用 httpx 异步转发# app/gateway.py import httpx from fastapi import HTTPException from app.config import settings class UpstreamError(Exception): def __init__(self, status_code: int, message: str): self.status_code status_code self.message message async def call_upstream(payload: dict, timeout: float 120.0) - dict: headers { Authorization: fBearer {settings.upstream_api_key}, Content-Type: application/json, } url f{settings.upstream_base_url}/v1/chat/completions try: async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post(url, jsonpayload, headersheaders) except httpx.TimeoutException: raise UpstreamError(504, upstream request timeout) except httpx.TransportError as e: raise UpstreamError(502, fupstream transport error: {str(e)}) if resp.status_code 400: raise UpstreamError(resp.status_code, resp.text) return resp.json()注意超时时间。大模型生成回复通常比普通接口慢30 秒超时很容易误伤长文本任务。这里设置 120 秒是针对对话补全的保守值。流式模式下超时模型又不同需要在读取流的过程中不断重置超时而不是一次性等 120 秒。3.4 错误码映射和超时策略用户不需要看到上游原始的丑陋错误你的网关应该把这些错误转换为相对稳定的错误码。上游状态码网关返回说明400400参数错误需要客户端修改请求体401502上游密钥失效属于服务端故障不应暴露给用户402402余额不足可能是用户余额不足或上游欠费403403权限不足需要检查模型权限或工具权限429429触发上游限流应返回重试提示500502上游服务故障网关重试一次后仍失败则降级返回网关层适合做一次简单重试。只对 429、500、502、504 这类瞬时错误重试对 400、401、403 这类确定性错误不要重试。重试要加退避比如第一次等 1 秒第二次等 2 秒最多重试 2 次避免上游雪崩时你的网关也在加压力。4. 计费、配额和限流让服务从“能调用”变成“能运营”4.1 按 token 计费还是按次计费常见计费方式有三种方式优点缺点适用场景按 token 计费公平成本匹配度高用户难以预估费用面向开发者按量付费按调用次数计费用户理解成本低长上下文任务可能亏钱业务场景固定如每天 100 次文案生成套餐订阅现金流稳定需要做配额治理面向企业和长期用户建议第一版先用按 token 计费因为成本和收入直接对应不会出现“用户调用一次超长任务你亏掉一半套餐费”的问题。后续稳定后再把套餐做成折扣预充值而不是无限调用。4.2 余额扣减与调用日志计费逻辑要在上游响应返回后执行不能提前扣费。因为上游可能因为长度限制、内容过滤等原因返回失败提前扣费会造成大量投诉。# app/billing.py from datetime import datetime from sqlalchemy.orm import Session from app.models import User, CallLog def calculate_cost(total_tokens: int, price_per_million: float) - int: # 价格以“元/M tokens”为单位换算成“分” cost_yuan total_tokens / 1_000_000 * price_per_million return max(1, int(cost_yuan * 100)) def deduct_balance(db: Session, user_id: int, cost_cents: int): user db.query(User).filter(User.id user_id).first() if user.balance_cents cost_cents: raise ValueError(insufficient balance) user.balance_cents - cost_cents db.add(user) def write_call_log(db: Session, api_key_id: int, user_id: int, model: str, usage: dict, cost_cents: int, status: int, error_message: str None): log CallLog( api_key_idapi_key_id, user_iduser_id, modelmodel, prompt_tokensusage.get(prompt_tokens, 0), completion_tokensusage.get(completion_tokens, 0), total_tokensusage.get(total_tokens, 0), cost_centscost_cents, statusstatus, error_messageerror_message, created_atdatetime.utcnow().isoformat(), ) db.add(log)在扣费之前接口层应该提前检查用户余额避免请求已经发出去了才发现余额不足。完整的流程是鉴权后检查用户余额是否大于 0。调用上游。上游成功返回后根据 usage 计算费用。扣减余额。写入调用日志。返回上游响应给用户。如果第 3 步失败不扣费只记录一条 status 非 200 的日志。这样对账时才能看到成功调用和失败调用各自消耗了多少。4.3 限流和并发控制限流要分两层用户维度和 Key 维度。第一版可以按用户维度做最简单的固定窗口限制比如每分钟最多 60 次。用 Redis 做限流计数比较简单# app/limiter.py import redis import time from fastapi import HTTPException r redis.Redis.from_url(redis://localhost:6379/0) def check_rate_limit(user_id: int, limit: int 60, window_seconds: int 60): key fratelimit:user:{user_id}:{int(time.time()) // window_seconds} current r.incr(key) if current 1: r.expire(key, window_seconds 1) if current limit: raise HTTPException(status_code429, detailrate limit exceeded)固定窗口的缺点是在窗口边界可能瞬时通过两倍请求但对第一版足够。生产环境想更平滑可以用令牌桶或滑动窗口。注意限流器依赖 Redis如果 Redis 挂了网关不应该跟着挂。稳妥做法是 Redis 不可用时降级为本地内存计数或者直接放行但记录告警而不是让整个服务不可用。4.4 简单的管理后台统计没有统计就无法定价。至少需要三个查询每个用户最近 7 天的调用量和费用。每个模型的平均 token 数和平均延迟。失败请求的 error_message 分布。# app/routers/admin.py router.get(/stats/users/{user_id}) def get_user_stats(user_id: int, db: Session Depends(get_db)): logs db.query(CallLog).filter(CallLog.user_id user_id).all() total_calls len(logs) total_cost sum(log.cost_cents for log in logs) total_tokens sum(log.total_tokens for log in logs) return { user_id: user_id, total_calls: total_calls, total_tokens: total_tokens, total_cost_cents: total_cost, }有了这些原始数据你才能判断一个用户是不是高频低价值调用一个模型是不是成本过高一个上游渠道是不是经常失败。5. 上线后最常见的一批 API 错误按现象逐个排查5.1 400 系列参数、模型名和上下文长度实际运营中最常见的 400 错误有三类。第一类是请求参数不被上游识别。比如有些模型服务要求特定参数客户端传了不存在的字段上游返回类似“the thinking_budget parameter must be a positive integer”的错误。排查方式是看网关日志里的error_message如果上游返回原文直接定位是哪个字段。第二类是模型名不支持。上游会返回类似“the supported api model names are ... but ...”的提示。常见原因是对外模型名没有映射到正确的上游模型名或者上游已经下线了旧模型。解决方案是建立模型版本映射表并定期检查上游模型列表。第三类是上下文长度超过窗口。模型窗口是固定的如果用户持续追加历史消息很快会触达上限报错类似“maximum context length is ... tokens”。网关层可以主动做消息裁剪删掉最早的历史消息或者把超长文本分块后再调用。5.2 402 系列余额不足到底是谁的余额402 错误有两个来源你的用户余额不足或者你的上游账户欠费。排查时必须先分清是哪一层。看网关日志error_message里是否包含上游返回的原文。如果上游返回 402说明你部署服务时配置的上游 Key 余额不足这是服务端故障要把错误映射为 502而不是直接透传给用户。如果是在你自己扣费逻辑里抛出的insufficient balance说明用户没充钱这是业务逻辑返回 402 给用户并提示充值。5.3 403 系列权限、工具和隐私声明的边界403 常见于三种情况。第一种是上游账号权限不足某个模型没有开通或者某个工具类接口没有授权。排查时直接带上游 Key 用 curl 调一次如果 curl 同样 403说明是上游权限问题不是你的网关问题。第二种是用户请求里带了没有权限的工具名称例如让模型调用某个外部搜索工具。网关层应该维护工具白名单不在白名单内的工具直接返回参数错误。第三种是微信小程序或客户端上报的api scope is not declared in the privacy agreement类型错误这已经不在网关范围内是客户端所在平台的隐私政策配置问题。这类错误要引导用户去检查平台侧申请权限而不是在后端反复排查。5.4 连接中断和流式响应问题“connection lost mid-response”这类错误对封装类 SaaS 影响很大因为用户已经看到一半结果突然断流体验很差。可能原因有三个上游服务不稳定生成长文本时连接被中断。你的网关没有开启流式转发先攒完整响应再返回给用户超时时间不够。客户端断开了连接但网关还在继续请求上游浪费上游费用。排查方式看网关日志中status列和目标模型如果大量出现在某个模型上基本可以确定是上游问题。如果只出现在某个用户可能是该用户网络环境导致。应对措施是支持流式响应。FastAPI 下可以把上游的 SSE 流通过StreamingResponse转发给客户端同时把已产生的 token 计数记录下来。即使中途断开也能依据已记录 token 扣费。5.5 排错优先级和检查清单遇到一个用户报错按下面顺序排查优先级检查项命令或日志1用户 API Key 是否有效查 api_keys 表is_active 和 expires_at2请求参数是否合法看网关日志中的请求体3用户余额是否足够看 users.balance_cents4上游返回什么错误看 call_logs.error_message5模型白名单是否包含该模型查 models.yaml 配置6是不是上游整体故障用 curl 手动调上游接口测试这个清单能覆盖绝大多数 Day 1 问题。不要一上来就看代码先看数据数据会告诉你问题出在哪一层。6. 从学习项目到付费服务上线前需要做的生产化改造6.1 配置外置化与环境隔离本地开发时可以把密钥写在.env里但生产环境必须使用独立的配置来源比如环境变量或配置中心。# .env.example 示意 UPSTREAM_BASE_URLhttps://api.example.com UPSTREAM_API_KEYsk-upstream-secret DATABASE_URLsqlite:///./api_saas.db REDIS_URLredis://localhost:6379/0 PRICE_PER_MILLION_TOKENS5 DEFAULT_MODELdemo-model环境隔离至少需要 dev、test、prod 三套配置。dev 环境可以用免费低价模型prod 环境使用正式模型和正式 Key。最容易出问题的是把测试 Key 配到生产服务里导致上线后大量 401 错误。建议在启动脚本里增加环境校验如果环境是prod启动时必须检查上游 Key 的前缀是否符合生产 Key 规则。6.2 日志、监控和告警第一版最容易忽略的是日志结构化。Python 的logging默认输出纯文本排错时要在一堆文本里找字段很痛苦。建议写一个简单的 JSON 日志方法import json import logging logger logging.getLogger(gateway) def log_call(call_id: str, user_id: int, model: str, tokens: int, cost_cents: int, status: int): record { type: api_call, call_id: call_id, user_id: user_id, model: model, tokens: tokens, cost_cents: cost_cents, status: status, } logger.info(json.dumps(record, ensure_asciiFalse))每条请求都带一个call_id是最佳实践客户端报错时只需提供 call_id你就能在日志里找到完整链路。监控方面不需要一开始就上全链路追踪。先做到三点请求成功率、上游延迟 P95、错误分布。任何一个超过阈值就告警。优先关注上游 5xx 率和用户 429 率这两个指标直接关系到用户体验。6.3 数据备份和回滚方案余额数据不能丢。SQLite 在早期能用但生产环境必须换 PostgreSQL 并开启每日备份。至少做到数据库自动备份保留最近 7 天。余额变动通过recharge_logs和call_logs可审计。代码发布用 tag 标记出问题能快速回滚到上一个版本。特别建议不要在call_logs表里删数据只做软删除或归档。调用日志是定价和排错的重要依据删了就找不回成本分析了。6.4 发布检查清单上线前逐项确认检查项确认内容上游 Key使用生产 Key额度充足环境变量正确数据库从 SQLite 切换 PostgreSQL执行迁移模型白名单对外模型名和上游模型名映射正确超时设置普通请求和流式请求分别配置超时限流规则用户维度和 IP 维度限流已生效错误码映射400、401、402、403、429、5xx 映射正确日志每条请求有 call_id失败日志有 error_message监控告警成功率、延迟、错误率告警已配置回滚方案知道如何回滚代码和数据库结构7. 关于“能赚钱”的实话定价、成本和产品差异化7.1 三种常见的收费模型第一版的手动充值加按 token 计费最容易实现流程是用户充值 - 余额增加 - 每次调用扣费。管理后台手动为用户加额度即可。收费模型实现复杂度现金流风险预充值按量低先充值后使用用户可能一次性用完订阅套餐中稳定需要做用量配额按项目/企业合同高大额稳定需要售前支持不要一开始就做无穷套餐。没有用量数据的情况下任何定价都是拍脑袋。先按量收费跑一个月统计平均每个用户每天消耗多少 token成本是多少再考虑要不要出套餐。7.2 成本控制的四个抓手封装类 SaaS 的利润空间很薄成本控制直接影响毛利。第一个抓手是上下文裁剪。很多用户会把完整文档塞进上下文但大多数调用只用到了其中一小部分。可以在网关层按需截断或者提供“先检索再生成”的 RAG 模式。第二个抓手是模型路由。简单问题走便宜的小模型复杂问题才走贵的大模型。可以在请求参数里加一个priority字段由网关智能选择模型。第三个抓手是缓存。语义完全相同的请求可以命中缓存但要做相似度判断不能只做字符串精确匹配。缓存对固定格式的文案生成特别有效。第四个抓手是重试和熔断。上游不稳定时盲目重试只会放大成本。设置单次请求的成本上限超过上限直接放弃。7.3 哪些情况会让毛利变成负的没有限流就开放接口个别用户把你的服务当免费样本测试短时间打爆上游额度。长上下文调用没有额外收费一个用户连续调用超长文档总结你的实际成本远高于按 token 估算的价格。上游涨价但你没有同步调价。很多模型厂商调价比较频繁建议把价格做成配置项每周对账一次上游账单和你的调用日志发现毛利率下降立即调价。7.4 可以把产品做厚的扩展方向当基础付费闭环跑通后有几个值得扩展的方向。开放的开发者平台提供 API 文档、错误码说明、SDK 示例、调用详情页让用户自助排查。企业版功能发票、审计日志、子账号、按部门独立核算、SSO 登录。这些是 B 端客户付费意愿最高的功能。细分场景模板针对客服、翻译、文案、代码生成提供预置 prompt 模板和专用接口把“API 调用”变成“业务功能”。如果只是搭一个转发接口这门生意的门槛很低真正有价值的地方是把调用、计量、计费、监控和用户体验做成一个可运营的系统。建议先用手动充值加按量计费的方式跑通第一版再根据真实调用数据决定要不要按套餐收费。对新手来说最有价值的练习不是一开始就写完整后台而是把一条请求从客户端走到上游再回来的全过程记录完整。日志里有成本、有错误、有用户行为商业化决策的素材都在里面。