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

最近我一直在想一件事情那些看起来“高大上”的 SaaS 工具真的是靠技术壁垒收那么贵还是只是把几个通用能力包装成了订阅于是这次我做了一个试验选了一款价格不低的订阅制文档处理 SaaS不破解、不扒源码而是用 AI 辅助开发的方式把它核心的文档解析、内容处理流程复刻成一个自部署服务并且把“按月付费”改成了“按 token 付费”。效果先说整套服务是能跑通的接口、批量任务、用户余额、扣费记录都有而且代码大部分是 Cursor AI Agent 生成的。如果你也好奇“SaaS 到底贵在哪”“本地能不能做一个差不多的”“token 计费到底怎么落地”这篇文章可以直接收藏。文章会从技术选型、项目初始化、AI 辅助生成代码、token 计费模块到功能测试、接口调用、批量任务、性能观察和常见问题排查完完整整走一遍。没有花里胡哨的营销包装只有能直接复制的命令和代码。1. 核心能力速览先把这次项目的核心信息放最前面方便你快速判断是否适合自己。能力项说明项目定位用 AI 辅助开发的方式复刻一款订阅制 SaaS 的核心功能并改为按 token 计费核心技术栈Python FastAPI SQLite/PostgreSQL 大模型 API或本地大模型AI 辅助开发工具Cursor、AI Agent 等编码工具主要功能文档上传、内容解析、智能提取、Markdown 导出、用户认证、Token 余额计费计费模式按 token 计费调用大模型 API 后根据真实 usage 扣费批量任务支持可上传多文件批量处理接口 API支持RESTful 风格返回 JSON部署方式本地命令行启动 / Docker 部署推荐环境Python 3.10内存 4GB 起使用云 API 时无需独立显卡是否支持本地模型可选本地模型需要根据显存和模型版本另行评估这里要特别说明本文的目标不是让你去盗用某家 SaaS 的源代码或绕过付费而是演示“用 AI 快速重写一个同类功能服务”的工程方法。直接用现成源码、破解授权都属于侵权风险不在讨论范围内。2. 适用场景与使用边界2.1 适合谁用个人开发者想做一个自己用的小工具不想每个月订阅高价 SaaS。中小企业内部需要一个文档解析或内容处理服务但不希望按人头订阅付费。做二次开发的团队先在本地跑通功能再封装成内部 API 供给业务系统调用。关注 token 计费模式的开发者想理解“按量付费”的产品逻辑和落地实现。2.2 能解决什么问题把固定订阅费用变成浮动 token 费用用多少花多少。把数据留在自己服务器上减少外部服务传文件的隐私顾虑。用 AI 辅助编程缩短开发周期一个人也能完成一个小型 SaaS 替代品。2.3 不适合什么场景复杂业务系统比如涉及多角色权限、审批流、企业合规审计不适合用这种快速克隆方案。高并发对外服务本文的实现属于轻量级自用没有做完备的网关、限流和分布式任务队列。需要做到像素级一致的用户体验前端 UI 可以相似但要完全复刻官方交互成本会迅速上升。2.4 版权、隐私与安全边界这部分的红线必须说清楚不能复制官方源码、不能破解官方客户端、不能绕过付费验证。本文的“克隆”指的是功能逻辑层面的重新实现而不是盗用代码。如果处理的是他人的文档、图片、音频必须确认有合法处理权限。涉及人脸、声音、身份信息时必须获得明确授权。如果调用了大模型 API数据会发送到模型服务方敏感数据要做脱敏或改为本地模型。部署对外接口时一定要加认证和限流否则容易被人刷接口产生大量费用。3. 技术选型与整体架构这类“克隆 SaaS token 计费”的项目架构并不复杂关键是先把模块边界理清楚。3.1 整体架构我的做法分三层接入层FastAPI 提供 RESTful 接口处理用户认证、上传文件、任务创建。业务层文档解析、内容提取、文本分块、调用大模型、组装 Markdown。计费层用户表、余额表、用量记录表每次模型调用完成后读取 usage 并扣费。前端不是重点可以直接做一个简单的 HTML 控制台用浏览器上传文件、看余额、看任务状态。如果你要做成产品再考虑 Vue 或 React。3.2 为什么选 FastAPIFastAPI 的优势是类型提示清晰、自动生成 Swagger 文档、异步支持好非常适合快速搭建 API 服务。配合 SQLAlchemy 做数据库操作整个后端骨架半天就能搭完。3.3 大模型选型这次的核心能力都交给大模型来做所以模型选型直接影响效果和成本追求效果优先选当前主流的多模态模型能直接处理图片、PDF 截图。追求成本可以用文本模型配合 OCR 先提取文字再交给模型整理。本地部署如果数据不能出域可以用本地模型但显存占用、推理速度、模型安装都要单独测试。一句话总结先确定数据能不能出域再决定用云 API 还是本地模型。4. 环境准备与项目初始化4.1 环境要求建议环境如下项目建议操作系统Windows 10/11、Ubuntu 20.04、macOSPython3.10 或更高版本数据库SQLite开发PostgreSQL生产内存4GB 以上显卡使用云 API 时不需要使用本地模型时按模型要求配置4.2 创建项目目录mkdir ai-saas-clone cd ai-saas-clone4.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_URLsqlite:///./app.db MODEL_NAMEgpt-4o-mini MODEL_API_KEYyour_api_key_here MODEL_API_BASEhttps://api.example.com/v1 TOKEN_PRICE_PER_1K0.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(titleAI SaaS Clone, version0.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 fuploads/{file.filename} with open(file_path, wb) as f: f.write(await file.read()) # 创建任务记录 task Task(user_idcurrent_user.id, file_pathfile_path, statuspending) 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: fdata:image/jpeg;base64,{img_data}}}, ], } ] else: with open(file_path, r, encodingutf-8) as f: text f.read() messages [ {role: user, content: f请将以下内容整理为 Markdown 格式\n\n{text[:12000]}} ] url f{API_BASE}/chat/completions headers {Authorization: fBearer {API_KEY}} payload { model: MODEL_NAME, messages: messages, temperature: 0.2, } resp requests.post(url, headersheaders, jsonpayload, timeout120) 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_typetext) task.output result task.status done task.total_tokens usage.get(total_tokens, 0) # 扣费 deduct_token( db, user_idtask.user_id, tokenstask.total_tokens, descriptionf任务 {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_keyTrue, indexTrue) username Column(String, uniqueTrue, indexTrue) hashed_password Column(String) balance Column(Float, default100.0) # 初始赠送余额 class Task(Base): __tablename__ tasks id Column(Integer, primary_keyTrue, indexTrue) user_id Column(Integer, ForeignKey(users.id)) file_path Column(String) output Column(Text, nullableTrue) status Column(String, defaultpending) total_tokens Column(Integer, default0) error Column(String, nullableTrue) created_at Column(DateTime, server_defaultfunc.now()) class UsageRecord(Base): __tablename__ usage_records id Column(Integer, primary_keyTrue, indexTrue) user_id Column(Integer, ForeignKey(users.id)) task_id Column(Integer, ForeignKey(tasks.id), nullableTrue) tokens Column(Integer) amount Column(Float) description Column(String, nullableTrue) created_at Column(DateTime, server_defaultfunc.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_code402, 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_code404, detail用户不存在) amount tokens / 1000 * TOKEN_PRICE_PER_1K if user.balance amount: raise HTTPException(status_code402, detail余额不足无法完成该任务) user.balance - amount record UsageRecord( user_iduser_id, tokenstokens, amountamount, descriptiondescription, ) 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默认端口可以设置为 8000uvicorn app.main:app --host 0.0.0.0 --port 8000启动后打开http://127.0.0.1:8000/docs可以看到 FastAPI 自动生成的 Swagger 文档。7.2 测试用户注册与登录在 Swagger 里创建用户或直接使用 curlcurl -X POST http://127.0.0.1:8000/api/register \ -H Content-Type: application/json \ -d {username: test_user, password: 123456}返回用户信息后再调用登录接口获取 tokencurl -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 常见失败原因现象可能原因注册成功但登录失败密码哈希方式不一致检查 auth 模块上传后任务一直 pendingWorker 未启动或任务处理逻辑没有被调用扣费没有发生任务成功后没有调用 deduct_token返回 402余额不足或单次处理成本超过余额模型调用超时API key 配置错误或文本太长改用流式或分段8. 接口 API 与批量任务这个服务跑通单文件处理后下一步就是批量任务。毕竟个人自用和团队使用都希望一次处理一批文件。8.1 API 列表方法路径功能是否需要认证POST/api/register用户注册否POST/api/login用户登录获取 token否POST/api/upload上传单个文件是POST/api/batch/upload批量上传文件是GET/api/task/{task_id}查询任务状态是GET/api/usage查询余额与扣费记录是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 fuploads/{file.filename} with open(file_path, wb) as f: f.write(await file.read()) task Task(user_idcurrent_user.id, file_pathfile_path, statuspending) 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: fBearer {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, headersheaders, filesfiles) 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 哪些因素影响性能和成本因素影响文本长度越长 token 越多费用越高处理时间越长图片分辨率高分辨率图片如果直接传给多模态模型token 消耗会明显增加批量数量并发任务越多内存和网络占用越高步数 / 温度对大模型来说temperature 影响随机性不直接影响 token 计费模型型号不同模型的单价差异很大选便宜模型测试选强模型做生产9.4 降低成本策略上传前做文件压缩比如把图片分辨率降到合理范围。先用 OCR 提取文字再只把文字传给模型比直接传整张图片省 token。对重复内容做缓存相同文件的处理结果直接复用。把长文本拆块处理按需合并避免一次传太多内容导致费用失控。10. 常见问题与排查方法在开发这个服务的过程中最典型的问题集中在启动失败、模型调用失败、计费逻辑异常三个方向。问题现象可能原因排查方式解决方案启动后页面打不开端口被占用或服务未启动检查控制台日志和端口换端口或杀掉占用进程注册接口 500数据库表未创建检查数据库初始化代码先执行建表逻辑再启动服务登录返回 token exchange failed自定义认证逻辑与前端不匹配检查 auth 模块返回字段统一 token 字段名修复过期时间配置文件上传后任务一直 pending任务处理器没有执行检查是否有后台线程消费任务手动触发处理函数或启动 Worker模型接口返回 401/403API key 错误或地区不支持查看服务商返回的错误码更换有效的 API key排除地区限制问题扣费金额和模型账单不一致计费单价配置错误核对 TOKEN_PRICE_PER_1K以模型服务商实际账单为准大文件处理超时单次请求内容过长查看模型服务商最大 token 限制拆块处理或增加超时时间余额扣成负数扣费前未做余额校验检查 check_balance 调用顺序扣费前强制校验余额批量任务部分失败模型接口偶发超时查看任务表的 error 字段加重试机制和错误日志清理上传目录后任务报错文件已被删除但任务未处理检查文件路径存在性任务创建前复制文件到独立目录这里特别提醒一下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 辅助开发已经把这层壳的成本降到了非常低真正值钱的是你对业务场景的理解。