AI 算法平台项目开发文档：FastAPI+Vue 全栈实现

1. 项目前言在大模型、计算机视觉、NLP 算法落地过程中普遍存在一个痛点算法模型分散部署、调用协议不统一、缺少任务生命周期管理、没有可视化运维面板。算法工程师交付模型脚本之后业务开发人员想要调用 AI 能力需要重复编写接口、处理异步任务、管理推理日志接入成本高、难以管控流量与权限。为了解决这个问题本文设计并实现一套轻量化 AI 算法中台平台采用当下主流的前后端分离架构后端基于 FastAPI 搭建高性能 API 服务天然自带类型校验、OpenAPI 接口文档、异步支持前端基于 Vue3 Vite Element Plus Pinia 开发管理控制台。平台定位不是重新训练算法而是算法的统一托管、调度、权限管控、任务追踪平台。平台支持两种推理模式短耗时同步推理OCR、文本分类、长耗时异步推理批量图片处理、大模型长文本、文档解析。算法可以是本地 PyTorch 脚本、TensorRT 推理服务也可以封装第三方大模型 API。新增算法遵循统一入参出参 Schema不需要修改平台主流程代码做到配置化接入。本项目文档适合作为毕业设计、企业内部中台原型、个人全栈实战项目完整覆盖从需求、架构、编码、联调到容器化部署全流程。技术栈总览表格分层组件说明后端框架FastAPI高性能异步 Web 框架自动生成 Swagger 文档ORMSQLAlchemy 2.0异步 ORM适配 async 数据库操作数据库MySQL 8.0持久化用户、模型、任务元数据缓存 / 消息中间件RedisCelery Broker、限流缓存、分布式锁异步任务Celery承载耗时 AI 推理任务支持多 Worker 横向扩容任务监控FlowerCelery 可视化面板监控 Worker、任务状态前端框架Vue3 Vite前端工程化框架UI 组件库Element Plus表单、表格、弹窗等基础组件状态管理PiniaVue3 全局状态管理替代 Vuex路由Vue Router 4前端路由、路由权限守卫请求库Axios请求拦截、统一 Token 处理、异常捕获图表ECharts调用量、成功率统计看板认证方案JWT账号登录鉴权配套 APIKey 给第三方系统调用部署Docker Docker Compose开发、测试环境一键编排2. 需求分析与角色设计2.1 用户角色划分平台一共四类使用者权限隔离采用 RBAC 基于角色的权限控制模型超级管理员最高权限管理全部用户、角色、系统配置查看所有任务与算法数据算法管理员负责算法注册、版本维护、上下线、配置参数 Schema调试算法普通业务用户浏览已授权算法提交推理任务查看、导出自己的任务结果第三方调用方业务系统通过 APIKey 调用平台接口无前端页面操作权限。2.2 核心功能需求2.2.1 用户与权限模块用户注册、登录、密码重置、账号启用 / 禁用JWT 签发、过期校验、刷新 TokenRBAC 权限绑定角色绑定权限点用户绑定角色APIKey 生成、启用 / 禁用、过期时间配置APIKey 独立绑定权限操作日志记录登录、新增算法、提交任务、修改配置等操作留痕。2.2.2 算法模型管理模块算法注册算法唯一标识、名称、描述、同步 / 异步类型版本管理同一个算法支持多版本并存支持切换线上生效版本参数 Schema 配置使用 JSON Schema 约束入参、出参格式前端自动生成表单算法上下线下线算法拒绝新的推理请求资源配置单任务超时时间、最大重试次数、并发配额在线调试页面填写参数直接发起测试调用。2.2.3 AI 推理任务模块同步推理短耗时场景HTTP 请求阻塞等待推理结果返回异步任务提交任务后立刻返回 task_id业务方轮询或者配置回调地址接收结果任务管理分页列表、多条件筛选状态、算法、时间范围任务详情原始入参、推理结果、异常堆栈、运行耗时、日志结果导出结构化结果导出 JSON 文件回调通知任务成功 / 失败后自动 POST 请求到预设回调地址。2.2.4 监控统计模块按小时、按天统计接口调用次数、成功量、失败量单算法调用排行、平均推理耗时 P50/P90基础组件状态监控Redis、MySQL、Celery Worker 在线状态。2.3 非功能需求性能同步接口 P95 响应时间小于 500ms异步任务支持 Worker 横向扩容容错任务支持可配置重试Worker 异常宕机未完成任务不丢失安全密码 bcrypt 加密存储、参数校验、SQL 注入防护、CORS 限制、接口限流扩展性新增算法只需要实现标准处理函数注册配置不改动平台主链路可观测统一异常捕获、结构化日志、标准化错误码。2.4 需求边界说明本方案不包含 Kubernetes GPU 调度、容器虚拟化资源隔离。大规模 GPU 集群场景可以后续扩展对接 K8s当前方案采用 Celery Worker 绑定显卡依靠 Redis 队列做任务排队。3. 系统总体架构设计整体采用分层解耦架构分为五层接入层、API 业务层、任务调度层、算法服务层、数据持久层。接入层前端 Vue 控制台 第三方业务系统通过 HTTP 访问API 业务层FastAPI路由分发、参数校验、JWT/APIKey 鉴权、权限判断、业务逻辑、数据组装任务调度层Redis Celery接收长耗时推理任务放入队列Worker 消费持久化任务状态算法服务层统一抽象算法处理器支持本地脚本、模型推理、封装外部大模型 API数据持久层MySQL RedisMySQL 存储业务元数据Redis 做消息队列、热点缓存、限流。调用流程分为两条链路同步推理链路前端 / 第三方请求 → FastAPI 鉴权校验 → 直接调用算法处理器 → 返回结果异步推理链路请求提交任务 → FastAPI 校验参数 → 写入任务记录 → 投递 Celery 队列 → 返回 task_id → Celery Worker 执行推理 → 更新任务状态 → 可选回调通知。架构优势同步异步一套算法注册规范流量削峰、长耗时任务不占用 Web 服务进程FastAPI 主线程不会被阻塞。4. 后端 FastAPI 模块详细设计4.1 项目目录结构backend/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── core/ # 核心配置 │ │ ├── config.py # 环境配置读取 │ │ ├── security.py # JWT、密码加密 │ │ ├── celery_app.py # celery 实例初始化 │ │ └── exceptions.py # 全局异常处理 │ ├── api/ │ │ ├── deps.py # 依赖注入鉴权、权限校验 │ │ ├── v1/ │ │ │ ├── auth.py # 登录、刷新token │ │ │ ├── user.py # 用户管理 │ │ │ ├── model.py # 算法模型管理 │ │ │ ├── task.py # 推理任务接口 │ │ │ └── statistic.py # 统计接口 │ ├── models/ # SQLAlchemy ORM 模型 │ ├── schemas/ # Pydantic 请求/响应结构体 │ ├── services/ │ │ ├── model_service.py # 算法业务逻辑 │ │ ├── task_service.py # 任务业务逻辑 │ │ └── algorithm/ # 各个算法实现 │ └── db/ # 数据库连接、会话管理 ├── requirements.txt └── docker-compose.yml4.2 全局统一返回体非常关键前后端约定所有接口统一返回格式方便前端 Axios 拦截器统一处理# app/schemas/response.py from pydantic import BaseModel from typing import Any, Optional class Result(BaseModel): code: int msg: str data: Optional[Any] None classmethod def success(cls, dataNone, msgsuccess): return cls(code200, msgmsg, datadata) classmethod def fail(cls, code400, msgfail, dataNone): return cls(codecode, msgmsg, datadata)约定错误码200成功401未登录 / Token 失效403权限不足404资源不存在422参数校验失败500服务内部异常4.3 全局异常捕获中间件# app/core/exceptions.py from fastapi import Request, FastAPI from fastapi.responses import JSONResponse from app.schemas.response import Result def register_exception_handler(app: FastAPI): app.exception_handler(Exception) async def global_exception(request: Request, exc: Exception): return JSONResponse(Result.fail(code500, msgstr(exc)).model_dump())4.4 路由拆分与依赖注入FastAPI 依赖注入是本项目的核心特性把鉴权逻辑抽成 Depends所有需要登录、权限校验的接口直接挂载。# app/api/deps.py from fastapi import Depends, HTTPException from app.core.security import verify_token from app.models.user import User async def get_current_user(token: str Depends(verify_token)) - User: if not token: raise HTTPException(status_code401, detailtoken invalid) return token5. 数据库表结构设计这里列出核心表基于 SQLAlchemy 2.0 异步模型。5.1 user 用户表字段id、username、hashed_password、is_active、role_id、api_key、create_time5.2 role 角色表字段id、role_name、permissionsJSON 存储权限标识数组5.3 ai_model 算法模型表字段id、model_key唯一标识、name、desc、is_syncbool、is_onlinebool、version、input_schema、output_schema、timeout、retry_count、callback_url5.4 inference_task 推理任务表字段id、task_id、model_key、user_id、statuspending/running/success/fail、input_data、result_data、error_msg、cost_time、create_time、finish_time5.5 operation_log 操作日志表字段id、user_id、operate_type、content、ip、create_time核心设计点model_key 作为算法唯一主键业务代码通过 model_key 匹配对应的算法处理器task_id 对外暴露给前端和第三方主键 id 内部使用。ORM 示例任务表# app/models/task.py from sqlalchemy import Column, Integer, String, Text, Float, DateTime from sqlalchemy.ext.declarative import declarative_base import datetime Base declarative_base() class InferenceTask(Base): __tablename__ inference_task id Column(Integer, primary_keyTrue, autoincrementTrue) task_id Column(String(64), uniqueTrue, indexTrue) model_key Column(String(64), indexTrue) user_id Column(Integer) status Column(String(20)) input_data Column(Text) result_data Column(Text) error_msg Column(Text) cost_time Column(Float) create_time Column(DateTime, defaultdatetime.datetime.now) finish_time Column(DateTime)6. 权限认证与安全方案JWTRBACAPIKey平台两套鉴权体系JWT给前端页面用户登录使用APIKey给第三方系统调用接口使用放在请求头X-API-Key。JWT 签发核心代码# app/core/security.py from jose import jwt from passlib.context import CryptContext from datetime import datetime, timedelta pwd_context CryptContext(schemes[bcrypt], deprecatedauto) SECRET_KEY your-secret-key ALGORITHM HS256 def create_access_token(subject: dict, expires_delta: timedelta): expire datetime.utcnow() expires_delta to_encode subject.copy() to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password)鉴权逻辑判断顺序优先判断请求头是否携带 X-API-Key有则走 APIKey 鉴权没有则校验 Authorization 的 Bearer Token。RBAC 权限点示例user:list、model:add、model:edit、task:submit接口装饰时校验用户角色是否包含该权限标识。安全补充策略CORS 配置只放行前端域名所有入参交给 Pydantic 强校验限流基于 Redis对 APIKey 和 IP 做 QPS 限制敏感密码不返回前端日志脱敏。7. Celery 异步任务调度核心实现Celery 是整个平台异步推理的核心Broker 使用 Redis。# app/core/celery_app.py from celery import Celery celery Celery( ai_task, brokerredis://redis:6379/0, backendredis://redis:6379/0 ) celery.conf.update( task_serializerjson, accept_content[json], result_serializerjson, timezoneAsia/Shanghai, enable_utcFalse ) # 注册推理任务 celery.task(bindTrue) def run_inference_task(self, task_id: str, model_key: str, input_data: dict): from app.services.task_service import execute_algorithm return execute_algorithm(task_id, model_key, input_data)业务层提交异步任务代码# app/services/task_service.py import uuid from app.core.celery_app import run_inference_task from app.models.task import InferenceTask from app.db.session import async_session async def submit_async_task(model_key: str, user_id: int, input_data: dict): task_id str(uuid.uuid4()) async with async_session() as session: db_task InferenceTask( task_idtask_id, model_keymodel_key, user_iduser_id, statuspending, input_datastr(input_data) ) session.add(db_task) await session.commit() # 投递celery队列 run_inference_task.delay(task_id, model_key, input_data) return task_id任务状态流转pending → running → success /failWorker 内部捕获异常自动更新 task 状态和错误信息。启动命令# 启动celery worker celery -A app.core.celery_app worker -l INFO # flower监控面板 celery -A app.core.celery_app flower8. AI 算法服务接入规范为了实现算法可插拔我们抽象统一的算法基类新增算法只需要继承基类实现 run 方法再注册到算法注册表。# app/services/algorithm/base.py from abc import ABC, abstractmethod class BaseAlgorithm(ABC): abstractmethod async def run(self, input_data: dict) - dict: 统一推理入口 :param input_data: 请求入参 :return: 推理结果字典 pass # 算法注册表 ALGORITHM_REGISTRY {} def register_algorithm(model_key: str): def decorator(cls): ALGORITHM_REGISTRY[model_key] cls() return cls return decorator示例一个文本分类算法# app/services/algorithm/text_classify.py from .base import BaseAlgorithm, register_algorithm register_algorithm(text_classify) class TextClassifyAlgorithm(BaseAlgorithm): async def run(self, input_data: dict): text input_data.get(text) # 这里替换成真实模型推理代码 return {label: positive, score: 0.92}9. Vue3 前端架构与核心模块实现前端工程Vue3 Vite Element Plus Pinia Vue Router4目录结构frontend/ ├── src/ │ ├── api/ # 接口请求封装 │ ├── components/ # 公共组件TaskTable、ModelForm等 │ ├── views/ # 页面登录、模型管理、任务列表、统计看板 │ ├── store/ # Pinia状态userStore │ ├── router/ # 路由表 路由守卫 │ ├── utils/ │ │ └── request.js # Axios封装 │ └── App.vue └── vite.config.jsAxios 请求封装request.jsimport axios from axios const service axios.create({ baseURL: import.meta.env.VITE_API_BASE_URL, timeout: 10000 }) // 请求拦截器自动携带token service.interceptors.request.use(config { const token localStorage.getItem(token) if(token){ config.headers.Authorization Bearer ${token} } return config }) // 响应拦截器统一处理返回体 service.interceptors.response.use(res{ const data res.data if(data.code ! 200){ ElMessage.error(data.msg) return Promise.reject(data) } return data.data }, err{ ElMessage.error(网络异常) return Promise.reject(err) }) export default service路由守卫实现前端权限控制// router/index.js const router createRouter({...}) router.beforeEach((to,from,next){ const token localStorage.getItem(token) if(to.meta.requireAuth !token){ next(/login) }else{ next() } })核心页面模型管理页动态根据 JSON Schema 渲染表单、任务列表页轮询异步任务状态、统计看板 ECharts。10. 前后端联调规范与统一返回体联调约定所有接口遵循统一 Result 返回结构异步任务提交接口只返回 task_id前端定时器轮询/api/v1/task/{task_id}获取状态文件上传统一走/api/v1/upload返回资源访问地址接口文档直接访问/docsFastAPI Swagger不需要额外维护文档时间字段统一 UTC 时间戳或者%Y-%m-%d %H:%M:%S。常见联调坑异步任务轮询间隔不要太短推荐 1.5s~3s防止打满接口。11. 单元测试与接口调试方案后端使用 pytest 做单元测试测试算法函数、权限校验、任务提交逻辑。FastAPI 自带 docs 页面是调试利器开发阶段直接在浏览器调试所有接口。测试重点无 Token 请求是否返回 401低权限账号访问管理员接口返回 403下线模型提交推理返回错误异步任务崩溃重试逻辑是否生效。12. Docker Compose 一键部署方案docker-compose.yml 编排服务fastapi、vue-nginx、mysql、redis、celery-worker、flower。version: 3.8 services: mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: root123 MYSQL_DATABASE: ai_platform ports: - 3306:3306 redis: image: redis:6-alpine ports: - 6379:6379 backend: build: ./backend ports: - 8000:8000 depends_on: - mysql - redis celery: build: ./backend command: celery -A app.core.celery_app worker -l INFO depends_on: - redis flower: build: ./backend command: celery -A app.core.celery_app flower ports: - 5555:5555 frontend: build: ./frontend ports: - 80:80生产环境建议增加 Nginx 反向代理开启 HTTPS关闭 flower 公网访问。13. 性能优化、限流与容错设计优化点清单Redis 缓存模型信息、权限配置减少频繁查询 MySQL异步任务队列拆分轻量队列、GPU heavy 队列避免长任务阻塞小任务接口限流基于 Redis lua 脚本实现令牌桶数据库加索引task_id、model_key、user_id大推理结果不要存在 MySQL存入对象存储数据库只存链接Worker 优雅重试幂等设计防止回调重复通知。限流伪代码思路根据 api_key path 作为 key单位时间内计数超过阈值直接返回 429。14. 踩坑总结与迭代规划踩坑记录Celery 默认后端存储结果过大大批量推理任务会把 Redis 撑爆业务上自己维护 task 表尽量不依赖 celery resultFastAPI 同步函数里面调用耗时模型推理会阻塞 event loop长任务必须丢 celeryJSON Schema 前端动态渲染表单时复杂嵌套结构容易报错做好 Schema 校验JWT 放在 localStorage 有 XSS 风险生产环境可以迁移到 HttpOnly CookieMySQL 连接池配置过小并发任务上来直接数据库连接耗尽。后续迭代方向接入对象存储 MinIO图片、文件托管对接 K8sGPU 资源调度模型热加载不用重启服务更新权重任务优先级队列模型 A/B 测试链路追踪 OpenTelemetry。15. 总结本文完整落地了一套 FastAPI Vue3 的 AI 算法平台解决了算法模型统一接入、任务调度、权限管控、可视化运维的问题。平台核心设计思想是业务中台和算法解耦中台负责流量、权限、任务生命周期算法只需要关心输入输出通过插件化注册方式接入。整套方案技术选型成熟、社区生态完善兼顾开发效率和性能既可以作为个人全栈实战项目学习也可以裁剪改造后直接落地中小企业内部 AI 能力平台。FastAPI 的异步能力、自动接口文档搭配 Vue3 现代化前端是目前 AI 应用中台非常合适的技术组合。