VideoAgentTrek-ScreenFilter打造SaaS服务多租户架构设计与API计费系统实现最近和几个做内容平台的朋友聊天他们都在头疼同一个问题用户上传的视频里时不时会冒出一些不太合适的画面人工审核吧成本高、效率低还容易漏。正好我之前在折腾一个叫VideoAgentTrek-ScreenFilter的视频内容过滤模型效果挺不错。我就琢磨着能不能把它包装成一个服务让有同样需求的团队直接调用按需付费这不就是个现成的SaaS点子吗说干就干。但要把一个单机跑通的模型变成稳定、可商用、能赚钱的服务光有模型可不行。核心得解决两个大问题怎么让成百上千个客户安全、隔离地使用同一个服务以及怎么清晰、公平地统计使用量并收费这就是多租户架构和API计费系统要干的活儿。今天我就把自己从零搭建这套系统的思路和关键实现分享出来给想技术创业或者做产品化的朋友一些参考。1. 整体蓝图一个SaaS视频过滤平台长什么样在动手写代码之前我们先得把整个服务的轮廓画出来。我们的目标不是做一个演示Demo而是一个真正能对外服务的产品。核心价值很简单客户通过API把视频传给我们我们调用VideoAgentTrek-ScreenFilter模型进行分析返回过滤结果比如标记出问题帧或者直接生成“洁净版”视频。客户按调用次数付费。基于这个核心流程系统需要几个关键部分来支撑多租户隔离确保A公司的数据绝对不会泄露给B公司他们的处理请求也不会相互干扰。用户与权限管理让客户能注册、登录、管理自己的API密钥。API网关与计费接收请求验证客户身份和权限统计调用次数并执行扣费或配额检查。异步任务队列视频处理比较耗时不能让它阻塞API响应需要异步处理。监控与运维随时了解服务健康状态、客户使用情况。下面这张图描绘了系统的核心架构[用户/客户端] | | (1. 发起API请求) v [API网关] - [认证/鉴权] - [计费/配额检查] | | (2. 创建异步任务) v [消息队列] (如Redis, RabbitMQ) | | (3. Worker消费任务) v [任务处理Worker] - [调用 VideoAgentTrek-ScreenFilter] | | (4. 存储结果) v [对象存储 数据库] | | (5. 通知用户/提供结果查询) v [用户/客户端]这个流程保证了API的快速响应和系统的可扩展性。接下来我们深入两个最核心的模块。2. 多租户架构设计让数据与计算“各回各家”多租户是SaaS的基石。我们的设计目标是逻辑隔离清晰实现相对简单未来易于扩展。主要有三种常见模式独立数据库每个客户一个独立的数据库。隔离性最强但成本高运维复杂。共享数据库独立Schema所有客户共用同一个数据库实例但每个客户有自己的一套表Schema。隔离性较好。共享数据库共享Schema所有客户的数据都存在同一套表里用一个tenant_id字段来区分。成本最低但需要在应用层小心处理所有数据查询确保隔离。对于起步阶段的视频处理SaaS我推荐第三种模式共享Schema。它的优势在于运维简单资源利用率高。只要我们能在代码层面牢牢守住“数据边界”就完全可行。2.1 如何在代码中实现租户隔离关键在于要在请求处理的最早阶段就识别出当前请求属于哪个租户并将这个信息贯穿整个处理链路。通常我们用客户的API Key来关联其租户身份。这里给出一个基于Python FastAPI的简单示例展示如何实现租户上下文管理# middleware/tenant_middleware.py import logging from fastapi import Request, HTTPException from starlette.middleware.base import BaseHTTPMiddleware from database import get_tenant_db_session # 假设的数据库会话管理 logger logging.getLogger(__name__) class TenantMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 1. 从请求头中提取API Key api_key request.headers.get(X-API-Key) if not api_key: raise HTTPException(status_code401, detailAPI Key缺失) # 2. 验证API Key并获取租户信息 tenant await validate_api_key_and_get_tenant(api_key) if not tenant: raise HTTPException(status_code403, detail无效的API Key或租户已禁用) # 3. 将租户信息存储在请求状态中供后续使用 request.state.tenant tenant logger.info(fProcessing request for tenant: {tenant.id}) # 4. 为当前租户设置数据库会话确保查询范围限定 # 这里假设你的ORM支持类似“schema”或“row level security”的租户隔离 # 以SQLAlchemy tenant_id字段为例我们通常通过scoped session或查询过滤器实现 # 此处简化演示将tenant_id存入请求状态在数据库操作时自动附加条件 request.state.tenant_id tenant.id response await call_next(request) return response # database.py (部分示例) from sqlalchemy import create_engine, Column, Integer, String, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, scoped_session from contextvars import ContextVar Base declarative_base() # 定义上下文变量来存储当前租户ID tenant_id_context_var: ContextVar[str] ContextVar(tenant_id, defaultNone) class Tenant(Base): __tablename__ tenants id Column(Integer, primary_keyTrue) name Column(String, uniqueTrue) api_key Column(String, uniqueTrue) is_active Column(Boolean, defaultTrue) # 所有需要租户隔离的表都应有tenant_id字段 class VideoJob(Base): __tablename__ video_jobs id Column(Integer, primary_keyTrue) tenant_id Column(Integer, ForeignKey(tenants.id), nullableFalse) # 关键字段 original_video_url Column(String) status Column(String) result_url Column(String) # ... 其他字段 # 创建一个自定义的Session类在查询时自动过滤tenant_id from sqlalchemy.orm import Query class TenantScopedQuery(Query): _tenant_id None def set_tenant(self, tenant_id): self._tenant_id tenant_id def get(self, ident): # 重写get方法确保获取的对象属于当前租户 obj super().get(ident) if obj and hasattr(obj, tenant_id) and obj.tenant_id ! self._tenant_id: return None return obj def filter_by(self, **kwargs): # 在所有查询中自动加入tenant_id过滤 if self._tenant_id is not None and tenant_id not in kwargs: kwargs[tenant_id] self._tenant_id return super().filter_by(**kwargs) # 创建绑定到自定义Query类的sessionmaker SessionLocal sessionmaker(query_clsTenantScopedQuery) # 使用scoped_session管理会话生命周期并与上下文变量关联 from sqlalchemy.orm import scoped_session engine create_engine(your-database-url) db_session_factory scoped_session(SessionLocal, scopefunclambda: tenant_id_context_var.get()) def get_db(): db db_session_factory() try: # 在生成db会话后从请求上下文中设置当前查询的租户ID current_tenant_id tenant_id_context_var.get() if current_tenant_id: db.set_tenant(current_tenant_id) yield db finally: db.remove()通过这样的中间件和数据库会话管理我们确保了从API入口开始每一个数据库操作都自动带上了tenant_id的过滤条件从根本上避免了数据跨租户泄露。2.2 资源隔离计算与存储除了数据计算和存储资源也需要隔离计算资源可以为不同套餐的客户分配不同优先级的任务队列。例如VIP客户的视频处理任务进入高优先级队列Worker会优先处理。这可以通过消息队列如RabbitMQ的priority属性或设置多个队列来实现。存储资源处理后的视频文件、截图等在对象存储如AWS S3、MinIO中应使用不同的路径前缀进行隔离。例如s3://your-bucket/tenant_{id}/videos/job_{job_id}/output.mp4。3. API计费与配额系统清晰透明才能长久计费系统是SaaS的“收银台”必须准确、可靠、可审计。它的核心功能是记录每一次API调用并根据预设的计费规则进行扣费或配额检查。3.1 核心数据模型设计我们需要几张核心表来支撑这个系统-- 用户/租户表 (tenants) 已在上文定义 -- 套餐计划表 (subscription_plans) CREATE TABLE subscription_plans ( id INT PRIMARY KEY, name VARCHAR(100), -- 如“基础版”、“专业版” price_per_unit DECIMAL(10,2), -- 每单位如千次调用的价格 monthly_quota INT, -- 每月调用次数配额 features JSON -- 存储该套餐的其他特性如并发限制、支持视频最大时长等 ); -- 租户订阅表 (tenant_subscriptions) CREATE TABLE tenant_subscriptions ( id INT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id), plan_id INT FOREIGN KEY REFERENCES subscription_plans(id), current_period_start DATETIME, -- 当前计费周期开始时间 current_period_end DATETIME, -- 当前计费周期结束时间 is_active BOOLEAN DEFAULT TRUE ); -- API调用记录表 (api_call_logs) CREATE TABLE api_call_logs ( id BIGINT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id), endpoint VARCHAR(255), -- 调用的API端点 job_id VARCHAR(255), -- 关联的异步任务ID input_params JSON, -- 调用参数便于审计和复现 status_code INT, -- 处理状态码 cost_units INT DEFAULT 1, -- 本次调用消耗的“单位数”如视频时长不同成本不同 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_tenant_created (tenant_id, created_at) -- 重要索引用于按租户和时间范围查询 ); -- 账户余额/信用表 (tenant_credits) CREATE TABLE tenant_credits ( id INT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id) UNIQUE, balance DECIMAL(10,2) DEFAULT 0.00, -- 当前账户余额预付费 total_usage_units INT DEFAULT 0, -- 当前周期总使用单位数后付费 last_reset_at DATETIME, -- 上次重置usage的时间用于月结 updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );3.2 计费逻辑的实现计费可以在两个时机触发调用前检查预扣费/配额检查和调用后确认扣费。对于按次计费的API我倾向于采用“先检查后扣费”的模式在异步任务完成后再精确扣费因为视频处理的成本可能与视频时长、复杂度有关。在API网关的认证通过后我们立即进行配额和余额检查# services/billing_service.py from datetime import datetime, timedelta from sqlalchemy.orm import Session from models import Tenant, TenantSubscription, TenantCredits, SubscriptionPlan, ApiCallLog from fastapi import HTTPException async def check_quota_and_billing(db: Session, tenant_id: int, estimated_cost_units: int 1): 检查租户配额和余额。 estimated_cost_units: 预估本次调用会消耗的单位数可根据请求参数初步计算。 # 1. 获取租户当前有效订阅 subscription db.query(TenantSubscription).filter( TenantSubscription.tenant_id tenant_id, TenantSubscription.is_active True, TenantSubscription.current_period_end datetime.utcnow() ).first() if not subscription: raise HTTPException(status_code403, detail无有效订阅套餐) plan db.query(SubscriptionPlan).get(subscription.plan_id) # 2. 检查月度配额 (如果套餐有配额限制) if plan.monthly_quota: # 计算本月已使用量 period_start subscription.current_period_start current_usage db.query(func.sum(ApiCallLog.cost_units)).filter( ApiCallLog.tenant_id tenant_id, ApiCallLog.created_at period_start ).scalar() or 0 if current_usage estimated_cost_units plan.monthly_quota: raise HTTPException(status_code429, detail本月调用配额已用尽) # 3. 检查账户余额 (对于预付费模式) credit db.query(TenantCredits).filter(TenantCredits.tenant_id tenant_id).first() if credit and credit.balance (estimated_cost_units * plan.price_per_unit): # 余额不足可以尝试触发自动充值或直接拒绝 raise HTTPException(status_code402, detail账户余额不足请充值) # 所有检查通过允许请求继续 return True # 在任务处理Worker完成工作后进行最终扣费 async def finalize_billing(db: Session, job_id: str, actual_cost_units: int): 根据任务实际消耗进行最终扣费和记录。 # 通过job_id找到对应的API调用记录和租户 call_log db.query(ApiCallLog).filter(ApiCallLog.job_id job_id).first() if not call_log: return tenant_id call_log.tenant_id call_log.cost_units actual_cost_units # 更新实际消耗 call_log.status_code 200 # 标记为成功完成 # 获取套餐单价 subscription ... # 获取租户当前订阅 plan db.query(SubscriptionPlan).get(subscription.plan_id) # 扣费逻辑 (以预付费为例) credit db.query(TenantCredits).filter(TenantCredits.tenant_id tenant_id).with_for_update().first() # 加锁防止并发扣费 if credit: cost actual_cost_units * plan.price_per_unit credit.balance - cost credit.total_usage_units actual_cost_units db.add(credit) db.add(call_log) db.commit() logger.info(fBilling finalized for job {job_id}, cost: {actual_cost_units} units.)3.3 给客户一个清晰的后台客户需要清楚地知道钱花在哪了。一个简单的管理后台应包含用量仪表盘显示本月已用配额/总额度、当前余额、每日调用量趋势图。账单明细列出所有API调用记录包括时间、端点、消耗单位、成本。套餐管理查看当前套餐升级或降级。API密钥管理可以生成、查看、禁用API密钥。4. 把碎片拼成整体从设计到可运行服务有了多租户和计费这两个核心模块我们再将其它部分串联起来形成一个完整的服务闭环。1. 用户发起请求客户使用他的API Key调用我们的视频过滤接口。2. API网关与预处理网关用Nginx或API网关软件进行限流、路由然后将请求转发给我们的核心应用如FastAPI服务。3. 认证、鉴权与计费检查应用内的中间件如TenantMiddleware验证API Key识别租户并调用check_quota_and_billing进行预检查。4. 创建异步任务检查通过后API立即响应一个job_id并将视频处理任务包含视频地址、租户ID、job_id等推送到消息队列如Redis或RabbitMQ。这一步必须快速不能让用户等待。5. 异步处理后台的Worker进程从队列中取出任务下载视频调用本地的VideoAgentTrek-ScreenFilter模型进行处理将结果上传到对象存储并更新数据库中的任务状态。6. 最终计费与通知处理完成后Worker调用finalize_billing进行精确扣费。同时可以通过Webhook或消息队列通知客户方系统“任务job_id已完成结果在这里”。7. 结果查询客户可以使用job_id随时查询任务状态和获取结果。5. 总结从单机模型到可商用的SaaS服务最大的挑战往往不在算法本身而在这些“外围”的系统性工程问题上。多租户架构确保了服务的稳定与安全是客户信任的基础而清晰、准确的计费系统则是商业模式的体现直接关系到服务的可持续性。这套设计并不是唯一解而是一个兼顾了复杂度、成本和功能的起步方案。在实际开发中你还会遇到更多细节如何做灰度发布、如何监控每个租户的资源消耗、如何设计更灵活的计费策略如按视频时长阶梯计价等等。但只要你抓住了“租户隔离”和“计量计费”这两个核心并像我们上面讨论的那样在架构和代码层面将其落实你就已经为你的VideoAgentTrek-ScreenFilter服务搭建起了一个坚实可靠的商业骨架。剩下的就是不断打磨产品体验丰富模型能力并找到你的第一批客户了。这条路我还在走希望这些分享能帮你少踩一些坑。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
VideoAgentTrek-ScreenFilter打造SaaS服务：多租户架构设计与API计费系统实现
发布时间：2026/5/20 4:10:03
VideoAgentTrek-ScreenFilter打造SaaS服务多租户架构设计与API计费系统实现最近和几个做内容平台的朋友聊天他们都在头疼同一个问题用户上传的视频里时不时会冒出一些不太合适的画面人工审核吧成本高、效率低还容易漏。正好我之前在折腾一个叫VideoAgentTrek-ScreenFilter的视频内容过滤模型效果挺不错。我就琢磨着能不能把它包装成一个服务让有同样需求的团队直接调用按需付费这不就是个现成的SaaS点子吗说干就干。但要把一个单机跑通的模型变成稳定、可商用、能赚钱的服务光有模型可不行。核心得解决两个大问题怎么让成百上千个客户安全、隔离地使用同一个服务以及怎么清晰、公平地统计使用量并收费这就是多租户架构和API计费系统要干的活儿。今天我就把自己从零搭建这套系统的思路和关键实现分享出来给想技术创业或者做产品化的朋友一些参考。1. 整体蓝图一个SaaS视频过滤平台长什么样在动手写代码之前我们先得把整个服务的轮廓画出来。我们的目标不是做一个演示Demo而是一个真正能对外服务的产品。核心价值很简单客户通过API把视频传给我们我们调用VideoAgentTrek-ScreenFilter模型进行分析返回过滤结果比如标记出问题帧或者直接生成“洁净版”视频。客户按调用次数付费。基于这个核心流程系统需要几个关键部分来支撑多租户隔离确保A公司的数据绝对不会泄露给B公司他们的处理请求也不会相互干扰。用户与权限管理让客户能注册、登录、管理自己的API密钥。API网关与计费接收请求验证客户身份和权限统计调用次数并执行扣费或配额检查。异步任务队列视频处理比较耗时不能让它阻塞API响应需要异步处理。监控与运维随时了解服务健康状态、客户使用情况。下面这张图描绘了系统的核心架构[用户/客户端] | | (1. 发起API请求) v [API网关] - [认证/鉴权] - [计费/配额检查] | | (2. 创建异步任务) v [消息队列] (如Redis, RabbitMQ) | | (3. Worker消费任务) v [任务处理Worker] - [调用 VideoAgentTrek-ScreenFilter] | | (4. 存储结果) v [对象存储 数据库] | | (5. 通知用户/提供结果查询) v [用户/客户端]这个流程保证了API的快速响应和系统的可扩展性。接下来我们深入两个最核心的模块。2. 多租户架构设计让数据与计算“各回各家”多租户是SaaS的基石。我们的设计目标是逻辑隔离清晰实现相对简单未来易于扩展。主要有三种常见模式独立数据库每个客户一个独立的数据库。隔离性最强但成本高运维复杂。共享数据库独立Schema所有客户共用同一个数据库实例但每个客户有自己的一套表Schema。隔离性较好。共享数据库共享Schema所有客户的数据都存在同一套表里用一个tenant_id字段来区分。成本最低但需要在应用层小心处理所有数据查询确保隔离。对于起步阶段的视频处理SaaS我推荐第三种模式共享Schema。它的优势在于运维简单资源利用率高。只要我们能在代码层面牢牢守住“数据边界”就完全可行。2.1 如何在代码中实现租户隔离关键在于要在请求处理的最早阶段就识别出当前请求属于哪个租户并将这个信息贯穿整个处理链路。通常我们用客户的API Key来关联其租户身份。这里给出一个基于Python FastAPI的简单示例展示如何实现租户上下文管理# middleware/tenant_middleware.py import logging from fastapi import Request, HTTPException from starlette.middleware.base import BaseHTTPMiddleware from database import get_tenant_db_session # 假设的数据库会话管理 logger logging.getLogger(__name__) class TenantMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # 1. 从请求头中提取API Key api_key request.headers.get(X-API-Key) if not api_key: raise HTTPException(status_code401, detailAPI Key缺失) # 2. 验证API Key并获取租户信息 tenant await validate_api_key_and_get_tenant(api_key) if not tenant: raise HTTPException(status_code403, detail无效的API Key或租户已禁用) # 3. 将租户信息存储在请求状态中供后续使用 request.state.tenant tenant logger.info(fProcessing request for tenant: {tenant.id}) # 4. 为当前租户设置数据库会话确保查询范围限定 # 这里假设你的ORM支持类似“schema”或“row level security”的租户隔离 # 以SQLAlchemy tenant_id字段为例我们通常通过scoped session或查询过滤器实现 # 此处简化演示将tenant_id存入请求状态在数据库操作时自动附加条件 request.state.tenant_id tenant.id response await call_next(request) return response # database.py (部分示例) from sqlalchemy import create_engine, Column, Integer, String, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, scoped_session from contextvars import ContextVar Base declarative_base() # 定义上下文变量来存储当前租户ID tenant_id_context_var: ContextVar[str] ContextVar(tenant_id, defaultNone) class Tenant(Base): __tablename__ tenants id Column(Integer, primary_keyTrue) name Column(String, uniqueTrue) api_key Column(String, uniqueTrue) is_active Column(Boolean, defaultTrue) # 所有需要租户隔离的表都应有tenant_id字段 class VideoJob(Base): __tablename__ video_jobs id Column(Integer, primary_keyTrue) tenant_id Column(Integer, ForeignKey(tenants.id), nullableFalse) # 关键字段 original_video_url Column(String) status Column(String) result_url Column(String) # ... 其他字段 # 创建一个自定义的Session类在查询时自动过滤tenant_id from sqlalchemy.orm import Query class TenantScopedQuery(Query): _tenant_id None def set_tenant(self, tenant_id): self._tenant_id tenant_id def get(self, ident): # 重写get方法确保获取的对象属于当前租户 obj super().get(ident) if obj and hasattr(obj, tenant_id) and obj.tenant_id ! self._tenant_id: return None return obj def filter_by(self, **kwargs): # 在所有查询中自动加入tenant_id过滤 if self._tenant_id is not None and tenant_id not in kwargs: kwargs[tenant_id] self._tenant_id return super().filter_by(**kwargs) # 创建绑定到自定义Query类的sessionmaker SessionLocal sessionmaker(query_clsTenantScopedQuery) # 使用scoped_session管理会话生命周期并与上下文变量关联 from sqlalchemy.orm import scoped_session engine create_engine(your-database-url) db_session_factory scoped_session(SessionLocal, scopefunclambda: tenant_id_context_var.get()) def get_db(): db db_session_factory() try: # 在生成db会话后从请求上下文中设置当前查询的租户ID current_tenant_id tenant_id_context_var.get() if current_tenant_id: db.set_tenant(current_tenant_id) yield db finally: db.remove()通过这样的中间件和数据库会话管理我们确保了从API入口开始每一个数据库操作都自动带上了tenant_id的过滤条件从根本上避免了数据跨租户泄露。2.2 资源隔离计算与存储除了数据计算和存储资源也需要隔离计算资源可以为不同套餐的客户分配不同优先级的任务队列。例如VIP客户的视频处理任务进入高优先级队列Worker会优先处理。这可以通过消息队列如RabbitMQ的priority属性或设置多个队列来实现。存储资源处理后的视频文件、截图等在对象存储如AWS S3、MinIO中应使用不同的路径前缀进行隔离。例如s3://your-bucket/tenant_{id}/videos/job_{job_id}/output.mp4。3. API计费与配额系统清晰透明才能长久计费系统是SaaS的“收银台”必须准确、可靠、可审计。它的核心功能是记录每一次API调用并根据预设的计费规则进行扣费或配额检查。3.1 核心数据模型设计我们需要几张核心表来支撑这个系统-- 用户/租户表 (tenants) 已在上文定义 -- 套餐计划表 (subscription_plans) CREATE TABLE subscription_plans ( id INT PRIMARY KEY, name VARCHAR(100), -- 如“基础版”、“专业版” price_per_unit DECIMAL(10,2), -- 每单位如千次调用的价格 monthly_quota INT, -- 每月调用次数配额 features JSON -- 存储该套餐的其他特性如并发限制、支持视频最大时长等 ); -- 租户订阅表 (tenant_subscriptions) CREATE TABLE tenant_subscriptions ( id INT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id), plan_id INT FOREIGN KEY REFERENCES subscription_plans(id), current_period_start DATETIME, -- 当前计费周期开始时间 current_period_end DATETIME, -- 当前计费周期结束时间 is_active BOOLEAN DEFAULT TRUE ); -- API调用记录表 (api_call_logs) CREATE TABLE api_call_logs ( id BIGINT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id), endpoint VARCHAR(255), -- 调用的API端点 job_id VARCHAR(255), -- 关联的异步任务ID input_params JSON, -- 调用参数便于审计和复现 status_code INT, -- 处理状态码 cost_units INT DEFAULT 1, -- 本次调用消耗的“单位数”如视频时长不同成本不同 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_tenant_created (tenant_id, created_at) -- 重要索引用于按租户和时间范围查询 ); -- 账户余额/信用表 (tenant_credits) CREATE TABLE tenant_credits ( id INT PRIMARY KEY, tenant_id INT FOREIGN KEY REFERENCES tenants(id) UNIQUE, balance DECIMAL(10,2) DEFAULT 0.00, -- 当前账户余额预付费 total_usage_units INT DEFAULT 0, -- 当前周期总使用单位数后付费 last_reset_at DATETIME, -- 上次重置usage的时间用于月结 updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );3.2 计费逻辑的实现计费可以在两个时机触发调用前检查预扣费/配额检查和调用后确认扣费。对于按次计费的API我倾向于采用“先检查后扣费”的模式在异步任务完成后再精确扣费因为视频处理的成本可能与视频时长、复杂度有关。在API网关的认证通过后我们立即进行配额和余额检查# services/billing_service.py from datetime import datetime, timedelta from sqlalchemy.orm import Session from models import Tenant, TenantSubscription, TenantCredits, SubscriptionPlan, ApiCallLog from fastapi import HTTPException async def check_quota_and_billing(db: Session, tenant_id: int, estimated_cost_units: int 1): 检查租户配额和余额。 estimated_cost_units: 预估本次调用会消耗的单位数可根据请求参数初步计算。 # 1. 获取租户当前有效订阅 subscription db.query(TenantSubscription).filter( TenantSubscription.tenant_id tenant_id, TenantSubscription.is_active True, TenantSubscription.current_period_end datetime.utcnow() ).first() if not subscription: raise HTTPException(status_code403, detail无有效订阅套餐) plan db.query(SubscriptionPlan).get(subscription.plan_id) # 2. 检查月度配额 (如果套餐有配额限制) if plan.monthly_quota: # 计算本月已使用量 period_start subscription.current_period_start current_usage db.query(func.sum(ApiCallLog.cost_units)).filter( ApiCallLog.tenant_id tenant_id, ApiCallLog.created_at period_start ).scalar() or 0 if current_usage estimated_cost_units plan.monthly_quota: raise HTTPException(status_code429, detail本月调用配额已用尽) # 3. 检查账户余额 (对于预付费模式) credit db.query(TenantCredits).filter(TenantCredits.tenant_id tenant_id).first() if credit and credit.balance (estimated_cost_units * plan.price_per_unit): # 余额不足可以尝试触发自动充值或直接拒绝 raise HTTPException(status_code402, detail账户余额不足请充值) # 所有检查通过允许请求继续 return True # 在任务处理Worker完成工作后进行最终扣费 async def finalize_billing(db: Session, job_id: str, actual_cost_units: int): 根据任务实际消耗进行最终扣费和记录。 # 通过job_id找到对应的API调用记录和租户 call_log db.query(ApiCallLog).filter(ApiCallLog.job_id job_id).first() if not call_log: return tenant_id call_log.tenant_id call_log.cost_units actual_cost_units # 更新实际消耗 call_log.status_code 200 # 标记为成功完成 # 获取套餐单价 subscription ... # 获取租户当前订阅 plan db.query(SubscriptionPlan).get(subscription.plan_id) # 扣费逻辑 (以预付费为例) credit db.query(TenantCredits).filter(TenantCredits.tenant_id tenant_id).with_for_update().first() # 加锁防止并发扣费 if credit: cost actual_cost_units * plan.price_per_unit credit.balance - cost credit.total_usage_units actual_cost_units db.add(credit) db.add(call_log) db.commit() logger.info(fBilling finalized for job {job_id}, cost: {actual_cost_units} units.)3.3 给客户一个清晰的后台客户需要清楚地知道钱花在哪了。一个简单的管理后台应包含用量仪表盘显示本月已用配额/总额度、当前余额、每日调用量趋势图。账单明细列出所有API调用记录包括时间、端点、消耗单位、成本。套餐管理查看当前套餐升级或降级。API密钥管理可以生成、查看、禁用API密钥。4. 把碎片拼成整体从设计到可运行服务有了多租户和计费这两个核心模块我们再将其它部分串联起来形成一个完整的服务闭环。1. 用户发起请求客户使用他的API Key调用我们的视频过滤接口。2. API网关与预处理网关用Nginx或API网关软件进行限流、路由然后将请求转发给我们的核心应用如FastAPI服务。3. 认证、鉴权与计费检查应用内的中间件如TenantMiddleware验证API Key识别租户并调用check_quota_and_billing进行预检查。4. 创建异步任务检查通过后API立即响应一个job_id并将视频处理任务包含视频地址、租户ID、job_id等推送到消息队列如Redis或RabbitMQ。这一步必须快速不能让用户等待。5. 异步处理后台的Worker进程从队列中取出任务下载视频调用本地的VideoAgentTrek-ScreenFilter模型进行处理将结果上传到对象存储并更新数据库中的任务状态。6. 最终计费与通知处理完成后Worker调用finalize_billing进行精确扣费。同时可以通过Webhook或消息队列通知客户方系统“任务job_id已完成结果在这里”。7. 结果查询客户可以使用job_id随时查询任务状态和获取结果。5. 总结从单机模型到可商用的SaaS服务最大的挑战往往不在算法本身而在这些“外围”的系统性工程问题上。多租户架构确保了服务的稳定与安全是客户信任的基础而清晰、准确的计费系统则是商业模式的体现直接关系到服务的可持续性。这套设计并不是唯一解而是一个兼顾了复杂度、成本和功能的起步方案。在实际开发中你还会遇到更多细节如何做灰度发布、如何监控每个租户的资源消耗、如何设计更灵活的计费策略如按视频时长阶梯计价等等。但只要你抓住了“租户隔离”和“计量计费”这两个核心并像我们上面讨论的那样在架构和代码层面将其落实你就已经为你的VideoAgentTrek-ScreenFilter服务搭建起了一个坚实可靠的商业骨架。剩下的就是不断打磨产品体验丰富模型能力并找到你的第一批客户了。这条路我还在走希望这些分享能帮你少踩一些坑。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
