用FastAPI与大模型API构建按token计费的自部署服务实践

今天聊一个很有意思的方向：把按月付费的 AI SaaS，理解功能、拆解流程、再用大模型 API 重新组装成按 token 计费的自部署服务。

先澄清边界。这里的“克隆”不是复制他人代码、扒前端、抄视觉稿或者盗用品牌素材，而是指通过公开的产品功能理解需求逻辑，用 AI 辅助工程化实现一套自己的轻量替代方案。你要做自用工具、内部系统、自动化实验，这条路非常实用；如果你想直接仿冒商业产品，那涉及版权和合规风险，不在本文讨论范围内。

从思路落地到工程，这篇文章会围绕一条完整技术链路展开：FastAPI 作为服务骨架，大模型 API 作为核心能力引擎，按 token 记账，再配合批量任务队列。内容按“环境准备 -> 部署启动 -> 功能测试 -> API 调用 -> token 成本核算 -> 问题排查”的顺序走一遍，末尾给一套可以照抄的最佳实践清单。

如果你正准备把“买 SaaS 订阅”改成“按用量付费的自建工具”，或者想搞懂 token 计费服务到底怎么搭，这篇可以直接收藏。

1. 核心能力速览

先把这套自部署方案的关键能力用一张表说清楚：

这里要强调一点：如果你全程调用云端大模型 API，其实对显卡要求很低，普通办公电脑都可以。真正吃资源的是后续想跑本地模型，那时才需要关注显存和 CPU 性能。

2. 适用场景与使用边界

2.1 适合做什么

这类“按 token 付费的自部署服务”最容易落地的场景有三个。

第一个是个人工具替代。很多 AI 笔记、AI 翻译、AI 写作助手，核心能力就一两项。你把这一两项抽出来，用 API 重新实现，日常使用成本往往比订阅费低，因为订阅费是为整套产品功能和服务渠道买单，而你只为自己真正用到的模型调用付费。

第二个是内部自动化。典型的例子：把客户邮件自动分类、把长篇文档自动生成摘要、把 Excel 表格里的描述字段批量翻译。这些任务走 SaaS 订阅很浪费，写成脚本按 token 计费反而划算。

第三个是学习研究。理解一个商业产品如何用大模型、如何设计提示词、如何控制成本，然后用代码复现核心链路，是很好的工程实践。

2.2 不适合做什么

不适合直接照着商业产品做仿冒。界面设计、图标、文案、专有数据、商标、后端逻辑代码，这些都不能直接搬。即使你只“参考功能”，也要注意平台服务条款：很多 API 服务商明确规定不能逆向工程、不能用于构建竞品。

另外，如果你的业务需要企业级 SLA、数据合规认证、售后支持，那自部署方案不一定合适。自己搭服务意味着自己负责监控、备份、故障恢复和可用性保障。

2.3 合规与隐私边界

使用大模型 API 处理数据时，要先确认服务商的数据使用政策。涉及客户信息、个人隐私、敏感业务数据时，要做到脱敏后才上传，并且优先选择承诺“不用于训练”的商业 API。

涉及人脸、声音、版权素材的功能，必须确认素材来源合法、获授权。不能拿未授权的图片、视频、音频做生成或克隆类实验，也不得用自建服务批量处理未授权数据。

3. 从“按月付费”到“按 token 付费”的架构设计

3.1 成本模型对比

传统 AI SaaS 的付费方式通常是按月订阅，比如每个月固定费用，包含一定次数或额度的使用量。问题在于，轻度用户用不满额度，重度用户又会超额。

按 token 付费的模式则完全不同：

按 token 付费需要自己处理记账、预算、监控，但好处是灵活，尤其适合批量任务和内部工具。

3.2 系统模块划分

整个服务可以拆成四个核心模块。

第一层是接入层。用 FastAPI 暴露 HTTP 接口，接收文本、文件路径、批量任务参数。这一层负责鉴权、参数校验、请求日志。

第二层是任务层。批量任务不能同步跑完，尤其当输入文件很多时，容易超时。所以要用简单队列：请求先进入队列，后台 Worker 逐个处理，前端轮询任务状态。

第三层是模型调用层。这里封装大模型 API，负责拼装提示词、调用模型、解析返回结果。后续如果换模型供应商，只需要改这一层。

第四层是记账层。每次调用成功后记录 token 消耗，写入本地数据库或日志，用于成本核算。

3.3 计费统计公式

按 token 计费的核心是统计公式：

总成本 = 输入 token 数 × 输入单价 + 输出 token 数 × 输出单价

大模型 API 通常按输入和输出分开计价，输出单价一般高于输入。所以控制成本的关键点有两个：减少输入 token（精简提示词、缩短上下文），减少无效输出（设置 max_tokens、禁止废话）。

4. 环境准备与前置条件

4.1 系统与软件要求

这里给一套通用环境清单，具体版本需要根据你选择的模型 API 和项目依赖调整。

4.2 API 密钥准备

在开始前，你需要准备一个可用的模型 API Key。不同服务商的申请方式不同，但基本流程都是注册账号、创建 API Key、充值或领取免费额度。

拿到 Key 后，建议先做一个最小验证，确认网络、Key、账户状态都正常。这一步非常重要，可以避免后面代码写好了才发现调用失败。

4.3 基础目录结构

推荐创建一个这样的项目目录：

ai-saas-clone/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── llm.py # 模型调用封装 │ ├── tasks.py # 批量任务队列 │ ├── token_usage.py # token 统计 │ └── config.py # 配置读取 ├── inputs/ # 批量处理输入目录 ├── outputs/ # 批量处理输出目录 ├── logs/ # 日志目录 ├── requirements.txt └── .env # 环境变量

这种结构适合中小型项目。模块划分越清晰，后续增加新功能越方便。

5. 安装部署与启动方式

5.1 安装依赖

创建一个虚拟环境并安装依赖：

python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install fastapi uvicorn requests pydantic python-dotenv

这里只列了最基础的依赖。实际使用中，如果要做文档解析，还需要安装pypdf、python-docx等库；如果要接特定模型 SDK，还需要按官方文档安装对应包。

5.2 配置文件

在项目根目录创建.env文件：

API_KEY=your_api_key_here API_BASE_URL=https://api.example.com/v1 MODEL_NAME=gpt-4o-mini MAX_TOKENS=1024 TEMPERATURE=0.7

API_BASE_URL和MODEL_NAME要根据你实际使用的模型服务商填写。不同服务商接口路径不一样，但绝大多数兼容 OpenAI 的接口格式。

5.3 FastAPI 服务骨架

创建一个最小可运行的 FastAPI 应用：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import os from dotenv import load_dotenv load_dotenv() app = FastAPI(title="AI SaaS Clone", version="0.1.0") class ChatRequest(BaseModel): prompt: str max_tokens: int = 512 class ChatResponse(BaseModel): response: str usage: dict @app.get("/health") def health_check(): return {"status": "ok"} @app.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest): # 这里在后续章节补充真实模型调用 return ChatResponse( response=f"你输入的 prompt 是：{req.prompt}", usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} )

这个骨架只做接口联通验证，不实际调用模型。先跑通 HTTP 链路，再接入大模型 API。

5.4 启动服务

uvicorn app.main:app --host 0.0.0.0 --port 8000

启动成功后，浏览器访问http://127.0.0.1:8000/health，应该能看到：

{"status":"ok"}

如果你只想本机访问，host用127.0.0.1；如果想让局域网内其他设备访问，用0.0.0.0。注意：暴露到公网时必须加鉴权，否则任何人都有可能调用你的服务并消耗你的 token。

6. 功能测试与效果验证

6.1 文本生成接口测试

先测试最基础的文本生成能力。调用/chat接口：

curl -X POST http://127.0.0.1:8000/chat \ -H "Content-Type: application/json" \ -d '{"prompt": "用一句话介绍 FastAPI"}'

这一步的目的是验证服务是否正常路由、参数是否解析成功、返回结构是否完整。如果这一步失败，先看服务日志，确定是网络问题、代码问题还是模型调用问题。

6.2 接入真实模型调用

把llm.py写成这样，封装模型 API：

import os import requests def call_llm(prompt, max_tokens=512, temperature=0.7): api_key = os.getenv("API_KEY") api_base = os.getenv("API_BASE_URL") model = os.getenv("MODEL_NAME") headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature, } response = requests.post(f"{api_base}/chat/completions", headers=headers, json=payload, timeout=60) response.raise_for_status() data = response.json() content = data["choices"][0]["message"]["content"] usage = data.get("usage", {}) return content, usage

再更新main.py中的/chat逻辑，把请求转发给模型 API，并返回 token 消耗：

from app.llm import call_llm @app.post("/chat", response_model=ChatResponse) def chat(req: ChatRequest): try: content, usage = call_llm( prompt=req.prompt, max_tokens=req.max_tokens ) return ChatResponse(response=content, usage=usage) except Exception as e: raise HTTPException(status_code=500, detail=str(e))

重启服务，再测试一次，应该能看到真实的模型返回和 token 统计。

6.3 文档批量摘要测试

做一次批量任务测试，验证服务在“多文件、多任务”场景下是否稳定。

操作步骤：

1. 在inputs目录下放入 5 个文本文件。
2. 写一个脚本读取每个文件，调用/chat接口生成摘要。
3. 把摘要写入outputs目录。

批量任务最核心的问题是超时和失败重试。同步循环会让大量请求排队，如果每个请求耗时 20 秒，5 个文件就要 100 秒。更合理的做法是使用异步队列。

6.4 判断成功标准

一个功能是否跑通，可以看四个指标：

只要这四个指标正常，说明这个功能可以进入正式使用。

7. 接口 API 与批量任务

7.1 设计一个简单的任务队列

批量任务建议使用“提交任务 + 查询状态”的模式，而不是同步等待。

在tasks.py中维护一个内存字典作为简单任务表：

import uuid from enum import Enum from typing import Dict class TaskStatus(str, Enum): PENDING = "pending" RUNNING = "running" DONE = "done" FAILED = "failed" tasks: Dict[str, dict] = {} def create_task(task_type: str, payload: dict) -> str: task_id = str(uuid.uuid4()) tasks[task_id] = { "id": task_id, "type": task_type, "payload": payload, "status": TaskStatus.PENDING, "result": None, "error": None, } return task_id def update_task(task_id: str, **kwargs): if task_id in tasks: tasks[task_id].update(kwargs)

这个实现只适合单机和中小批量场景。生产环境建议用 Redis + Celery，或者至少用 SQLite 持久化任务状态。

7.2 批量处理脚本示例

下面是一个串行处理的脚本，适合小批量验证：

import os import time import requests API_URL = "http://127.0.0.1:8000/chat" def process_file(filepath): with open(filepath, "r", encoding="utf-8") as f: text = f.read() response = requests.post(API_URL, json={ "prompt": f"请总结以下内容，输出 3 个要点：\n{text}", "max_tokens": 256, }, timeout=60) if response.status_code != 200: return None, response.text data = response.json() return data["response"], data["usage"] input_dir = "inputs" output_dir = "outputs" os.makedirs(output_dir, exist_ok=True) for filename in os.listdir(input_dir): if not filename.endswith(".txt"): continue filepath = os.path.join(input_dir, filename) result, usage = process_file(filepath) if result is None: print(f"FAILED: {filename}, error: {usage}") continue output_path = os.path.join(output_dir, f"summary_{filename}") with open(output_path, "w", encoding="utf-8") as f: f.write(result) print(f"OK: {filename}, token usage: {usage}") time.sleep(1) # 简单限速，避免触发服务商频率限制

这段脚本体现了批量任务的三个关键点：循环遍历、失败标记、延迟限速。真实项目中还要加上重试机制和 token 成本累计统计。

7.3 失败重试建议

批量任务失败的原因通常有两种：瞬时网络错误和服务商限流。

推荐重试策略：

重试时最好加指数退避，避免服务刚刚恢复时所有任务同时重试，把接口打挂。

8. token 成本核算与性能观察

8.1 token 费用计算方式

大多数模型 API 按输入 token 和输出 token 分别计价。假设你的模型服务商有两种价格：

输入：0.15 元 / 1K token 输出：0.60 元 / 1K token

一次调用的成本就是：

成本 = 输入 token 数 × 0.15 / 1000 + 输出 token 数 × 0.60 / 1000

长期来看，成本主要被三个变量放大：

8.2 控制 token 消耗的实用技巧

第一，提示词里只保留必要信息。不要把一个 10 万字的文档全文塞进 prompt，可以先做分块、抽取关键段落，再调用模型。

第二，为不同任务设置不同的max_tokens。摘要任务给 256，写作任务给 1024，对话任务给 512，避免模型无限制输出。

第三，开启历史对话截断。多轮对话场景中，只保留最近 N 轮消息，不要让上下文无限膨胀。

第四，开启流式输出。流式输出虽然不能减少 token 数，但可以降低用户等待感知，而且能在生成过程中提前中断低质量输出。

8.3 性能观察方法

资源占用方面重点看四类指标：

如果你是纯 API 调用方案，本地资源占用很低，瓶颈通常在网络延迟和服务商限流。如果你跑本地模型，才需要关注显存占用。

观察显存占用可以用nvidia-smi：

nvidia-smi -l 2

这个命令每 2 秒刷新一次，可以看到显存、GPU 利用率和显存温度。本地推理时，显存占用会随模型大小和输入长度变化，具体数值需按实际模型测试。

8.4 防止成本失控

给服务加一道成本保护，推荐三种方式：

1. 设置单次调用的最大 token 数。
2. 设置用户维度每日调用上限。
3. 每次调用后写入成本日志，超过阈值触发告警。

成本日志可以设计成简单的 JSON 文件，或者写入 SQLite：

{ "timestamp": "2025-01-01T12:00:00Z", "model": "gpt-4o-mini", "prompt_tokens": 1200, "completion_tokens": 200, "total_tokens": 1400, "estimated_cost": 0.0003 }

有了日志，每个月复盘成本时可以直接汇总，不用靠猜。

9. 常见问题与排查方法

9.1 token 403 类错误的处理思路

调用大模型 API 时，偶尔会遇到类似下面的错误：

sign-in could not be completed token exchange failed: token endpoint returned status 403 forbidden

这种错误通常出现在 OAuth 令牌交换环节，不是模型本身的问题。排查顺序是：

1. 检查 API Key 是否有效、是否过期。
2. 确认账号是否有权限调用对应模型。
3. 检查服务商是否支持当前所在区域。
4. 查看 API 文档中的具体错误码说明。

这里要特别注意：如果服务商对区域或组织有合规限制，应该按官方渠道申请开通，而不是尝试用绕过手段访问。绕过服务商限制可能存在账号封禁和合规风险。

9.2 依赖安装失败

安装 Python 依赖时，如果遇到网络较慢或部分包编译失败，可以尝试：

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

更换镜像源能解决大部分下载慢的问题。如果某个包需要编译且失败，优先检查 Python 版本是否匹配，再尝试安装该依赖的预编译版本。

10. 最佳实践与使用建议

10.1 工程化落地建议

第一，先小参数测试，再上批量。第一次运行服务时，用 1 到 2 条请求验证流程，确认输出质量和成本符合预期，再放开批量任务。

第二，保留一套最小可运行配置。把环境变量和依赖锁定到一个已知可用的版本，方便故障时快速回滚。

第三，模型、输入、输出分目录管理。输入目录、输出目录、日志目录分开，避免文件混乱影响任务处理。

第四，批量任务要加日志和失败重试。没有日志的批量任务是最难排查的，至少要让每个任务记录开始时间、结束时间、状态和错误信息。

第五，接口服务要限制访问范围。不暴露到公网，或者加 API Key 鉴权，避免被刷接口、消耗 token。

10.2 内容与版权合规建议

在实现类似“替代 SaaS”的项目时，有几条底线不能碰：

• 不复制商业产品的代码、设计稿、品牌标识。
• 不使用来源不明的数据集进行商业训练。
• 不处理未授权的人脸、声音、版权素材。
• 涉及企业内部数据时，先确认数据合规要求。

这些不只是技术问题，也是法律和道德风险。做技术实验没有问题，但发布和商用之前，必须自己复核一遍授权链条。

10.3 后续扩展方向

这套“按 token 付费的自部署服务”跑通之后，可以继续扩展的方向很多：

每一步扩展的成本都不高，因为核心架构已经拆好了：接口层负责入口，任务层负责调度，模型层负责能力，记账层负责成本。后面加功能只需要在对应层里增加模块。

10.4 最后的建议

这套方案的优点是很实际的：按 token 付费确实能让成本曲线贴合真实使用量，轻量用户不用为用不完的订阅额度买单，重度用户也不会因为超额而产生意外账单。服务搭好后，日常使用就是“启动服务 -> 调用接口 -> 查看成本日志”这么简单。

但也要想清楚一点：自己搭服务，意味着自己扛运维。模型 API 挂了你要知道怎么降级，批量任务跑坏了你要会重跑，token 成本超了你要会设预算。如果你的核心诉求只是“快速完成某个任务”，直接用成熟 SaaS 更省心；如果你享受技术掌控感、有明确的自动化需求，那就值得把这条路走下去。

建议收藏备用，动手的时候先跑通最小骨架，再逐步加功能。