用AI复刻SaaS文档处理服务：从订阅制到Token计费实战

最近我一直在想一件事情：那些看起来“高大上”的 SaaS 工具，真的是靠技术壁垒收那么贵，还是只是把几个通用能力包装成了订阅？

于是这次我做了一个试验：选了一款价格不低的订阅制文档处理 SaaS，不破解、不扒源码，而是用 AI 辅助开发的方式，把它核心的文档解析、内容处理流程复刻成一个自部署服务，并且把“按月付费”改成了“按 token 付费”。

效果先说：整套服务是能跑通的，接口、批量任务、用户余额、扣费记录都有，而且代码大部分是 Cursor + AI Agent 生成的。如果你也好奇“SaaS 到底贵在哪”“本地能不能做一个差不多的”“token 计费到底怎么落地”，这篇文章可以直接收藏。

文章会从技术选型、项目初始化、AI 辅助生成代码、token 计费模块，到功能测试、接口调用、批量任务、性能观察和常见问题排查，完完整整走一遍。没有花里胡哨的营销包装，只有能直接复制的命令和代码。

1. 核心能力速览

先把这次项目的核心信息放最前面，方便你快速判断是否适合自己。

这里要特别说明：本文的目标不是让你去盗用某家 SaaS 的源代码或绕过付费，而是演示“用 AI 快速重写一个同类功能服务”的工程方法。直接用现成源码、破解授权都属于侵权风险，不在讨论范围内。

2. 适用场景与使用边界

2.1 适合谁用

• 个人开发者：想做一个自己用的小工具，不想每个月订阅高价 SaaS。
• 中小企业内部：需要一个文档解析或内容处理服务，但不希望按人头订阅付费。
• 做二次开发的团队：先在本地跑通功能，再封装成内部 API 供给业务系统调用。
• 关注 token 计费模式的开发者：想理解“按量付费”的产品逻辑和落地实现。

2.2 能解决什么问题

• 把固定订阅费用变成浮动 token 费用，用多少花多少。
• 把数据留在自己服务器上，减少外部服务传文件的隐私顾虑。
• 用 AI 辅助编程缩短开发周期，一个人也能完成一个小型 SaaS 替代品。

2.3 不适合什么场景

• 复杂业务系统：比如涉及多角色权限、审批流、企业合规审计，不适合用这种快速克隆方案。
• 高并发对外服务：本文的实现属于轻量级自用，没有做完备的网关、限流和分布式任务队列。
• 需要做到像素级一致的用户体验：前端 UI 可以相似，但要完全复刻官方交互，成本会迅速上升。

2.4 版权、隐私与安全边界

这部分的红线必须说清楚：

• 不能复制官方源码、不能破解官方客户端、不能绕过付费验证。
• 本文的“克隆”指的是功能逻辑层面的重新实现，而不是盗用代码。
• 如果处理的是他人的文档、图片、音频，必须确认有合法处理权限。
• 涉及人脸、声音、身份信息时，必须获得明确授权。
• 如果调用了大模型 API，数据会发送到模型服务方，敏感数据要做脱敏或改为本地模型。
• 部署对外接口时，一定要加认证和限流，否则容易被人刷接口产生大量费用。

3. 技术选型与整体架构

这类“克隆 SaaS + token 计费”的项目，架构并不复杂，关键是先把模块边界理清楚。

3.1 整体架构

我的做法分三层：

1. 接入层：FastAPI 提供 RESTful 接口，处理用户认证、上传文件、任务创建。
2. 业务层：文档解析、内容提取、文本分块、调用大模型、组装 Markdown。
3. 计费层：用户表、余额表、用量记录表，每次模型调用完成后读取 usage 并扣费。

前端不是重点，可以直接做一个简单的 HTML 控制台，用浏览器上传文件、看余额、看任务状态。如果你要做成产品，再考虑 Vue 或 React。

3.2 为什么选 FastAPI

FastAPI 的优势是类型提示清晰、自动生成 Swagger 文档、异步支持好，非常适合快速搭建 API 服务。配合 SQLAlchemy 做数据库操作，整个后端骨架半天就能搭完。

3.3 大模型选型

这次的核心能力都交给大模型来做，所以模型选型直接影响效果和成本：

• 追求效果：优先选当前主流的多模态模型，能直接处理图片、PDF 截图。
• 追求成本：可以用文本模型配合 OCR 先提取文字，再交给模型整理。
• 本地部署：如果数据不能出域，可以用本地模型，但显存占用、推理速度、模型安装都要单独测试。

一句话总结：先确定数据能不能出域，再决定用云 API 还是本地模型。

4. 环境准备与项目初始化

4.1 环境要求

建议环境如下：

4.2 创建项目目录

mkdir ai-saas-clone cd ai-saas-clone

4.3 创建虚拟环境并安装依赖

python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate

核心依赖：

pip install fastapi uvicorn sqlalchemy pydantic python-multipart requests python-dotenv

如果你需要做本地 OCR 预处理，可以额外安装：

pip install paddleocr paddlepaddle

这套依赖足够支撑一个轻量级的文档解析服务。我的建议是先在虚拟环境里一步步装，不要直接装到系统 Python 里，避免污染环境。

4.4 项目目录结构

ai-saas-clone/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── database.py # 数据库连接 │ ├── models.py # ORM 模型 │ ├── auth.py # 用户认证与 token 校验 │ ├── billing.py # token 计费模块 │ ├── tasks.py # 任务处理逻辑 │ └── llm.py # 大模型调用封装 ├── uploads/ # 上传文件目录 ├── outputs/ # 输出结果目录 ├── .env # 环境变量 ├── requirements.txt └── run.py # 启动入口

在开始写代码之前，先把.env建好：

touch .env

写入环境变量：

DATABASE_URL=sqlite:///./app.db MODEL_NAME=gpt-4o-mini MODEL_API_KEY=your_api_key_here MODEL_API_BASE=https://api.example.com/v1 TOKEN_PRICE_PER_1K=0.002

注意：这里的MODEL_API_BASE和MODEL_API_KEY需要替换成大模型服务商提供的真实地址和密钥。不同模型的单价不一样，最终计费金额以你的模型服务商账单为准。

5. 用 AI 辅助开发核心代码

这个项目的重点不是手写每一行代码，而是让 AI 帮你把骨架搭起来，你负责审查和调整。

5.1 给 AI 的 Prompt 写法

我在 Cursor 里的第一个 Prompt 是这样写的：

我用 FastAPI 搭建一个文档解析服务，实现用户注册、登录、上传 PDF 或图片，调用大模型提取内容并返回 Markdown。按 token 计费，每次调用大模型后读取 usage 并扣除用户余额。请生成完整的项目目录结构和代码。

注意这里要把需求拆成几个点：

• 用户认证方式。
• 文件上传方式。
• 大模型调用方式。
• 计费触发时机。
• 返回格式。

AI 生成代码之后，不要直接跑，先做一次人工审查。因为 AI 生成的代码经常有 import 遗漏、字段命名不一致、异步调用问题。

5.2 FastAPI 入口代码

下面是一份简化后的入口代码，你可以直接参考：

from fastapi import FastAPI, File, UploadFile, Depends, HTTPException from sqlalchemy.orm import Session from .database import get_db from .auth import get_current_user from .billing import deduct_token, check_balance from .llm import process_document from .models import User, Task app = FastAPI(title="AI SaaS Clone", version="0.1.0") @app.post("/api/upload") async def upload_document( file: UploadFile = File(...), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): # 检查余额，防止用户提交后才发现余额不足 check_balance(db, current_user) # 保存上传文件 file_path = f"uploads/{file.filename}" with open(file_path, "wb") as f: f.write(await file.read()) # 创建任务记录 task = Task(user_id=current_user.id, file_path=file_path, status="pending") db.add(task) db.commit() db.refresh(task) return {"task_id": task.id, "status": task.status}

这个接口做的事情很简单：验证用户、检查余额、保存文件、创建任务记录。

5.3 大模型调用封装

文档内容处理的核心在llm.py，代码大致如下：

import os import requests from dotenv import load_dotenv load_dotenv() API_BASE = os.getenv("MODEL_API_BASE") API_KEY = os.getenv("MODEL_API_KEY") MODEL_NAME = os.getenv("MODEL_NAME") def extract_content(file_path: str, file_type: str = "text"): """ 通用文本提取函数：根据文件类型做不同处理 """ if file_type == "image": # 图片场景：可以使用多模态模型，将图片 base64 传入 import base64 with open(file_path, "rb") as f: img_data = base64.b64encode(f.read()).decode("utf-8") messages = [ { "role": "user", "content": [ {"type": "text", "text": "请提取图片中的所有文字，并整理为 Markdown 格式。"}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_data}"}}, ], } ] else: with open(file_path, "r", encoding="utf-8") as f: text = f.read() messages = [ {"role": "user", "content": f"请将以下内容整理为 Markdown 格式：\n\n{text[:12000]}"} ] url = f"{API_BASE}/chat/completions" headers = {"Authorization": f"Bearer {API_KEY}"} payload = { "model": MODEL_NAME, "messages": messages, "temperature": 0.2, } resp = requests.post(url, headers=headers, json=payload, timeout=120) resp.raise_for_status() data = resp.json() completion = data["choices"][0]["message"]["content"] usage = data.get("usage", {}) return completion, usage

这个封装有一个关键点：把模型返回的usage原样带回，这样才能在业务层做精确扣费。

5.4 任务处理逻辑

任务处理可以选择同步执行，也可以选择后台队列。先做同步版本，跑通再改异步。

from fastapi import BackgroundTasks from .llm import extract_content from .billing import deduct_token def process_task(db, task_id: int): task = db.query(Task).filter(Task.id == task_id).first() if not task: return task.status = "processing" db.commit() try: # 这里简化逻辑：默认按 text 处理 result, usage = extract_content(task.file_path, file_type="text") task.output = result task.status = "done" task.total_tokens = usage.get("total_tokens", 0) # 扣费 deduct_token( db, user_id=task.user_id, tokens=task.total_tokens, description=f"任务 {task.id} 模型调用" ) db.commit() except Exception as e: task.status = "failed" task.error = str(e) db.commit()

实际项目中，任务处理逻辑应该放到 Worker 里执行，避免阻塞 API 请求。但在做功能验证阶段，同步处理也够用。

6. Token 计费模块设计与实现

Token 计费是整个项目的核心差异点。

传统 SaaS 按月收费，不管用户用多用少，价格固定。按 token 计费之后，用户的实际成本与模型调用量直接相关。

6.1 计费模型设计

我设计了三个核心数据库表：

• users：用户信息。
• tasks：任务记录，保存每次处理的 token 消耗。
• usage_records：扣费流水，记录每次扣费的 token 数和金额。

from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Text from sqlalchemy.sql import func from .database import Base class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String, unique=True, index=True) hashed_password = Column(String) balance = Column(Float, default=100.0) # 初始赠送余额 class Task(Base): __tablename__ = "tasks" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, ForeignKey("users.id")) file_path = Column(String) output = Column(Text, nullable=True) status = Column(String, default="pending") total_tokens = Column(Integer, default=0) error = Column(String, nullable=True) created_at = Column(DateTime, server_default=func.now()) class UsageRecord(Base): __tablename__ = "usage_records" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, ForeignKey("users.id")) task_id = Column(Integer, ForeignKey("tasks.id"), nullable=True) tokens = Column(Integer) amount = Column(Float) description = Column(String, nullable=True) created_at = Column(DateTime, server_default=func.now())

6.2 扣费逻辑

扣费函数需要做两件事：校验余额、扣减余额、写入流水。

from sqlalchemy.orm import Session from fastapi import HTTPException from .models import User, UsageRecord TOKEN_PRICE_PER_1K = float(os.getenv("TOKEN_PRICE_PER_1K", "0.002")) def check_balance(db: Session, user: User): if user.balance <= 0: raise HTTPException(status_code=402, detail="余额不足，请先充值") def deduct_token(db: Session, user_id: int, tokens: int, description: str = ""): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="用户不存在") amount = tokens / 1000 * TOKEN_PRICE_PER_1K if user.balance < amount: raise HTTPException(status_code=402, detail="余额不足，无法完成该任务") user.balance -= amount record = UsageRecord( user_id=user_id, tokens=tokens, amount=amount, description=description, ) db.add(record) db.commit() return record

这里有几个细节值得注意：

• 扣费要在任务成功之后做，失败任务不应该扣用户钱。
• 任务开始时先检查一次余额，防止处理到一半才发现没钱。
• 大模型返回的total_tokens是真实消耗，不要自己预估，因为预估往往不准。
• 单价用环境变量配置，方便随时调整。

6.3 Token 费用展示

用户最关心的不是扣了多少次，而是每次任务花了多少钱。可以加一个接口返回用户余额和最近流水。

@app.get("/api/usage") def get_usage(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): records = db.query(UsageRecord).filter(UsageRecord.user_id == current_user.id).all() return { "balance": current_user.balance, "records": [ { "task_id": r.task_id, "tokens": r.tokens, "amount": r.amount, "description": r.description, "created_at": str(r.created_at), } for r in records ], }

这个接口在后期接入前端控制台时非常有用，用户可以实时看到每一笔 token 消耗。

7. 功能测试与效果验证

项目跑起来之后，不要急着加更多功能，先按下面这组测试用例验证核心链路。

7.1 启动服务

python run.py

默认端口可以设置为 8000：

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

启动后打开http://127.0.0.1:8000/docs，可以看到 FastAPI 自动生成的 Swagger 文档。

7.2 测试用户注册与登录

在 Swagger 里创建用户，或直接使用 curl：

curl -X POST http://127.0.0.1:8000/api/register \ -H "Content-Type: application/json" \ -d '{"username": "test_user", "password": "123456"}'

返回用户信息后，再调用登录接口获取 token：

curl -X POST http://127.0.0.1:8000/api/login \ -H "Content-Type: application/json" \ -d '{"username": "test_user", "password": "123456"}'

7.3 测试文件上传与任务处理

curl -X POST http://127.0.0.1:8000/api/upload \ -H "Authorization: Bearer YOUR_TOKEN" \ -F "file=@./test.txt"

预期返回：

{ "task_id": 1, "status": "pending" }

7.4 测试 token 扣费

任务完成后，查询用户流水：

curl -X GET http://127.0.0.1:8000/api/usage \ -H "Authorization: Bearer YOUR_TOKEN"

预期结果里能看到一条扣费记录，包含 token 数和金额。如果其他功能正常但这里没有记录，说明任务处理链路里没有触发deduct_token，需要检查任务回调逻辑。

7.5 测试余额不足场景

把用户余额改成 0，再上传文件，预期接口直接返回 402，提示“余额不足”。

这一步很重要，因为很多 SaaS 替代品最容易出问题的就是“用户已经消耗了资源，最后才发现余额不够扣”。

7.6 判断成功标准

• 注册、登录成功。
• 上传文件后任务状态从 pending 变为 done。
• 输出内容是模型整理后的 Markdown。
• 用户余额正确减少。
• 扣费记录与模型返回的usage.total_tokens一致。

7.7 常见失败原因

8. 接口 API 与批量任务

这个服务跑通单文件处理后，下一步就是批量任务。毕竟个人自用和团队使用都希望一次处理一批文件。

8.1 API 列表

8.2 批量任务接口示例

批量任务的思路是：一次性接收多个文件，每个文件生成一个独立任务，然后逐个处理。

@app.post("/api/batch/upload") async def batch_upload( files: list[UploadFile] = File(...), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): task_ids = [] for file in files: file_path = f"uploads/{file.filename}" with open(file_path, "wb") as f: f.write(await file.read()) task = Task(user_id=current_user.id, file_path=file_path, status="pending") db.add(task) db.commit() db.refresh(task) task_ids.append(task.id) return {"task_ids": task_ids, "count": len(task_ids)}

批量任务的处理建议使用后台 Worker，否则大量文件会阻塞接口响应。

8.3 Python 批量调用脚本

import requests API_BASE = "http://127.0.0.1:8000" TOKEN = "YOUR_TOKEN" headers = {"Authorization": f"Bearer {TOKEN}"} files = [] for file_path in ["doc1.txt", "doc2.txt", "doc3.txt"]: with open(file_path, "rb") as f: files.append(("files", (file_path, f))) resp = requests.post(f"{API_BASE}/api/batch/upload", headers=headers, files=files) print(resp.json())

批量场景下，建议加两个机制：

• 每个任务处理完成后，把结果写入独立文件，记录 task_id。
• 定时扫描失败任务并重试，重试次数建议 2 到 3 次，避免模型接口偶发超时导致任务失败。

8.4 任务队列的下一步设计

当前版本的批量任务只是循环处理，适合个人使用。如果要做成正式服务，建议把任务写入 Redis 队列，由多个 Worker 并发消费。那个时候，任务表需要增加worker_id、started_at、finished_at字段，方便追踪每个任务的处理进度。

9. 资源占用与性能观察

这部分很多人会关心，尤其是用本地模型时，显存和内存直接决定能不能跑。

9.1 云 API 模式

如果大模型调用走云 API，服务的资源瓶颈主要在三个地方：

• 文件上传与临时存储：上传目录会不断增长，需要定期清理。
• 内存：大文档读取到内存时，如果文件超过几十 MB，内存容易飙升。
• 网络延迟：每个任务都要请求模型服务，任务越多，等待时间越长。

观察方法：

htop # 看 CPU 和内存

9.2 本地模型模式

如果数据不能出域，要改用本地模型：

• 下载模型权重，需要预留足够磁盘空间。
• 推理时显存占用由模型大小、上下文长度、批次大小决定。
• 没有独立显卡时，CPU 推理理论上可行，但处理速度会明显下降。

观察命令：

nvidia-smi # 查看 GPU 显存占用

不要凭感觉判断显存够不够，以实际跑任务的占用为准。第一次测试时用小文本、小图片，逐步加大输入，观察显存变化趋势。

9.3 哪些因素影响性能和成本

9.4 降低成本策略

• 上传前做文件压缩，比如把图片分辨率降到合理范围。
• 先用 OCR 提取文字，再只把文字传给模型，比直接传整张图片省 token。
• 对重复内容做缓存，相同文件的处理结果直接复用。
• 把长文本拆块处理，按需合并，避免一次传太多内容导致费用失控。

10. 常见问题与排查方法

在开发这个服务的过程中，最典型的问题集中在启动失败、模型调用失败、计费逻辑异常三个方向。

这里特别提醒一下：token exchange failed这类错误在对接外部模型服务时偶发出现，常见原因包括 token 过期、请求头拼错、密钥无效，或者服务端限制。遇到时先抓接口返回的原始错误，不要只看状态码。

11. 最佳实践与合规建议

项目能跑通只是一个开始。如果你想把它变成日常可用的工具，下面的工程化建议能帮你少踩很多坑。

11.1 开发流程建议

• 先做最小闭环：注册、登录、上传、处理、扣费，全部跑通后再加批量功能。
• 每次改代码后，先跑一遍核心测试用例，不要攒到最后一起验证。
• 保留一套最小可运行配置，方便随时回滚到稳定版本。

11.2 数据管理建议

• uploads/、outputs/、logs/目录分开，用.gitignore排除。
• 数据库文件要定期备份，尤其是线上环境。
• 每次任务生成的中间文件，任务结束后按策略清理。

11.3 安全建议

• 部署到公网时，必须加 HTTPS。
• API token 要设置过期时间，长期 token 有泄漏风险。
• 接口要加限流，防止被刷掉几十美金。
• 用户密码不要明文存储，用passlib或bcrypt做哈希。
• 大模型调用密钥要放在服务端环境变量里，不要写进前端代码。

11.4 合规建议

• 如果复刻的是别人已有产品，只允许做功能层面的重写，不要盗用界面素材、图标、宣传文案。
• 不要用仿冒名称混淆用户，产品名要和你自己的项目匹配。
• 处理第三方文档、图片、音频，必须有合法的处理权限。
• 商用之前，确认大模型服务条款是否允许你的调用场景。

虽然这次项目叫“克隆”，但从工程角度看，本质是“用 AI 辅助快速实现一个同类功能的替代品”。这套方法的价值不在“模仿”，而在“快速验证一个产品想法”。

12. 总结与下一步

这次最值得尝试的点是完整的业务闭环：AI 辅助生成代码、文档处理、用户认证、token 精确计费，一个服务串起来。按 token 计费之后，成本变得非常透明，用户每一笔消耗都能查到记录。

建议你先验证最小闭环，不要一上来就追求批量任务和并发。把单文件上传、模型处理、余额扣减、流水查询这一条链路跑通，整个服务的骨架就稳了。

最容易踩的坑有两个：

• 模型调用失败导致任务卡在 pending，却没有错误日志可查。
• 扣费时机不对，任务失败了还扣了用户费用。

这两点只需要在代码里加上异常处理和日志输出就能有效避免。

下一步的扩展方向很明确：

• 接入 PostgreSQL 替代 SQLite，适配多人使用。
• 增加 Redis 队列，让批量任务真正并发执行。
• 做一个简单的 Web 控制台，让用户直接在浏览器里上传文件、看余额、下载结果。
• 把计费模块独立成通用中间件，复用到其他 AI 工具上。

如果你正在纠结要不要订阅某个 SaaS，不如先按这套方法做一个小范围验证。用多少付多少，数据留在自己手里，核心业务逻辑完全可控。这套流程跑通后，你会发现很多“贵得离谱”的工具，本质上就是把几个 API 调用包了一层壳。现在 AI 辅助开发已经把这层壳的成本降到了非常低，真正值钱的，是你对业务场景的理解。