从 Flask 到 FastAPI：我的 OCS 题库系统重构之路

前言：缘起

2025 年的冬天，我在 GitHub 上发现了一个有趣的项目 —— ai-ocs-question_bank by Miaozeqiu。这是一个基于 Flask 的 OCS（网课助手）题库查询系统，利用 AI 技术自动答题。作为一个正在学习异步编程的开发者，我看到了这个项目的潜力，也看到了优化的空间。

于是，我在 2025 年 12 月 27 日 fork 了项目，开始了一场为期三天的大规模架构重构。

这篇文章将记录我从接手项目到完成重构的完整心路历程，包括：

为什么选择重构而不是修补
如何设计新的架构
重构过程中遇到的坑
性能优化的具体成果
开源项目文档化的思考

一、现状分析：旧架构的痛点

1.1 技术债务

Fork 下来的项目使用的是经典的 Flask + 同步 Python 架构：

hljs python

# 旧架构示例（Flask）
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/query', methods=['GET'])
def query():
    title = request.args.get('title')
    # 同步数据库查询
    answer = db.query(title)
    # 同步 HTTP 请求
    ai_response = requests.post(ai_api, json=...)
    return jsonify(answer)

主要问题：

同步阻塞：每个请求都会阻塞整个线程
低并发能力：Flask 开发服务器只能处理 ~4 QPS
缺少类型提示：大量 Any 类型，IDE 无法提供智能补全
配置混乱：使用 .env 文件，不支持多 AI 平台配置
文档缺失：没有 API 文档，难以集成

1.2 性能瓶颈

在实际使用中，我发现：

当同时有 5 个请求时，响应时间从 50ms 飙升到 500ms
CPU 使用率低，但并发能力极差（典型的 IO 密集型任务）
数据库查询和 AI API 调用都是同步操作，浪费了大量时间等待

结论：这个项目需要一次彻底的架构升级，而不是简单的 bug 修复。

二、技术选型：为什么是 FastAPI？

2.1 对比分析

在决定重构方案时，我对比了几个主流框架：

特性	Flask	FastAPI	Sanic	Tornado
异步支持	❌ 需要扩展	✅ 原生	✅ 原生	✅ 原生
类型提示	⚠️ 部分	✅ 100%	⚠️ 部分	⚠️ 部分
自动文档	❌ 需要手动	✅ Swagger/ReDoc	❌ 需要手动	❌ 需要手动
数据验证	⚠️ 手动	✅ Pydantic	⚠️ 手动	⚠️ 手动
学习曲线	低	中	中	高
性能	基准	2-3x	2x	1.5x

2.2 选择 FastAPI 的理由

1. 原生异步支持

hljs python

from fastapi import FastAPI
from httpx import AsyncClient

app = FastAPI()

@app.get("/query")
async def query(title: str):
    # 异步 HTTP 请求
    async with AsyncClient() as client:
        response = await client.post(ai_api, json={...})
    return response.json()

2. 类型安全和自动文档

hljs python

from pydantic import BaseModel

class QueryRequest(BaseModel):
    title: str
    options: list[str]
    type: Literal["single", "multiple", "judgement", "fill"]

@app.post("/api/v1/query")
async def query(req: QueryRequest) -> QueryResponse:
    """
    自动生成 Swagger 文档！
    """
    ...

3. Pydantic 数据验证

自动类型转换
详细的错误提示
JSON Schema 生成

最终决定：使用 FastAPI + AsyncIO + SQLModel 作为新架构。

三、重构实战：三天三夜的代码马拉松

3.1 第一阶段：分层架构设计（12 月 27 日凌晨）

我采用了 Clean Architecture（整洁架构） 的思想，将代码分为四层：

app/
├── api/              # 路由层（处理 HTTP 请求）
│   ├── deps.py       # 依赖注入
│   └── v1/           # API v1 版本
├── core/             # 核心配置
│   ├── config.py     # Pydantic Settings
│   ├── db.py         # 数据库连接池
│   └── logger.py     # Loguru 日志
├── models/           # SQLModel 数据模型
├── schemas/          # Pydantic 请求/响应 Schema
├── repositories/     # 数据访问层（Repository 模式）
├── services/         # 业务逻辑层（Service 层）
└── providers/        # 外部服务提供商（AI、缓存等）

设计原则：

单一职责：每层只做一件事
- API 层：处理 HTTP 请求/响应
- Service 层：业务逻辑
- Repository 层：数据访问
依赖倒置：高层不依赖低层，都依赖抽象

hljs python

# providers/base.py
class BaseAIProvider(ABC):
    @abstractmethod
    async def generate_answer(self, prompt: str) -> str:
        pass

# providers/siliconflow.py
class SiliconFlowProvider(BaseAIProvider):
    async def generate_answer(self, prompt: str) -> str:
        # 具体实现
        ...

3.2 第二阶段：全链路异步改造（12 月 27 日上午）

核心挑战：将所有同步操作改为异步

1. 数据库异步化

hljs python

# 旧代码（同步）
from sqlalchemy import create_engine
engine = create_engine("sqlite:///./.db")
result = engine.execute(query)

# 新代码（异步）
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("sqlite+aiosqlite:///./.db")
async with AsyncSession(engine) as session:
    result = await session.execute(query)

2. HTTP 客户端异步化

hljs python

# 旧代码（同步）
import requests
response = requests.post(url, json=data)

# 新代码（异步）
import httpx
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=data)

3. 缓存系统异步化

hljs python

from functools import lru_cache

# 同步 LRU 缓存
@lru_cache(maxsize=1000)
def get_cache(key):
    ...

# 异步缓存（自定义实现）
class AsyncCache:
    def __init__(self):
        self._cache = {}

    async def get(self, key: str):
        return self._cache.get(key)

    async def set(self, key: str, value: Any, ttl: int = 3600):
        self._cache[key] = value
        # 可选：使用 asyncio.create_task 实现过期删除

3.3 第三阶段：配置系统重构（12 月 27 日下午）

问题：原来的 .env 文件难以管理多个 AI 平台的配置

旧配置（.env）：

hljs bash

AI_API_KEY=sk-xxx
AI_API_URL=https://api.siliconflow.cn
AI_MODEL=Qwen/Qwen2.5-7B-Instruct

新配置（config.json）：

hljs json

{
  "ai": {
    "default_provider": "siliconflow",
    "providers": {
      "siliconflow": {
        "enabled": true,
        "api_key": "sk-xxx",
        "base_url": "https://api.siliconflow.cn/v1",
        "model": "Qwen/Qwen2.5-7B-Instruct"
      },
      "ali_bailian": {
        "enabled": false,
        "api_key": "sk-yyy"
      },
      "zhipu": {
        "enabled": false,
        "api_key": "sk-zzz"
      },
      "google": {
        "enabled": false,
        "api_key": "AIza..."
      },
      "openai": {
        "enabled": false,
        "api_key": "sk-..."
      }
    }
  }
}

优势：

支持多个 AI 平台同时配置
JSON 格式更直观
易于添加新平台
可以快速切换默认平台

通用 AI Provider 设计：

hljs python

class UniversalAIProvider:
    def __init__(self, provider_name: str, config: ProviderConfig):
        self.provider_name = provider_name
        self.config = config

        if provider_name == "google":
            self.client = GoogleProvider(config)
        else:
            # OpenAI 兼容格式（SiliconFlow、Ali Bailian、Zhipu、OpenAI）
            self.client = OpenAICompatibleProvider(config)

    async def generate_answer(self, prompt: str) -> str:
        return await self.client.generate_answer(prompt)

3.4 第四阶段：Bug 修复与优化（12 月 27 日晚 - 12 月 28 日）

Bug #1：OCS 答案格式不匹配

问题描述： AI 返回的答案格式不统一，有时是 A.北京，有时只是 北京，导致 OCS 无法正确匹配。

解决方案：

hljs python

def match_option(answer: str, options: list[str]) -> str:
    """
    智能匹配答案到选项
    """
    # 如果答案已包含选项前缀（如 "A.北京"），直接返回
    for opt in options:
        if answer in opt or opt in answer:
            return opt

    # 否则，尝试根据内容匹配
    answer_lower = answer.lower().strip()
    for opt in options:
        content = opt.split(".", 1)[-1].strip()
        if content.lower() == answer_lower:
            return opt

    # 实在匹配不到，返回原答案
    return answer

Bug #2：SOCKS 代理错误

问题：使用 SOCKS5 代理时报错 Missing dependencies for SOCKS support

解决：

hljs bash

# 在 pyproject.toml 中添加
dependencies = [
    "httpx[socks]",  # 添加 socksio 支持
]

Bug #3：缓存系统 AttributeError

问题：配置结构更新后，缓存服务还在使用旧的 settings.CACHE_TYPE

解决：

hljs python

# 旧代码
cache_type = settings.CACHE_TYPE
cache_ttl = settings.CACHE_TTL

# 新代码
cache_type = settings.cache.type
cache_ttl = settings.cache.ttl

3.5 第五阶段：文档工程（12 月 28 日 - 12 月 29 日）

我发现很多开源项目代码写得很好，但文档一塌糊涂，导致用户难以上手。因此，我决定把文档工程作为重构的重要组成部分。

文档结构：

docs/
├── INSTALL.md          # 安装指南（uv、pip、Docker 三种方式）
├── DOCKER.md           # Docker 部署指南（1200+ 行，超详细！）
├── API.md              # API 文档（请求/响应格式、错误处理、SDK 示例）
├── DEVELOPMENT.md      # 开发指南（架构、工作流、编码规范）
├── OCS集成指南.md      # OCS 插件配置教程
└── 配置指南.md         # AI 平台配置指南

Docker 部署指南亮点：

多平台 Docker 安装（Linux/macOS/Windows）
三种部署方式（Docker Compose / CLI / 多容器）
生产环境优化（HTTPS、反向代理、安全配置）
备份、恢复和迁移指南
常见问题排查

API 文档亮点：

完整的请求/响应示例
错误码对照表
Python / JavaScript SDK 封装示例
与 OCS、批量导入、聊天应用的集成案例

四、性能优化：从 4 QPS 到 200+ QPS

4.1 优化前 vs 优化后

指标	Flask 版本	FastAPI 版本	提升倍数
并发能力（QPS）	~4	200+	50x
平均响应时间	~100ms	<50ms	2x
内存占用	80MB	60MB	25% ↓
代码行数	239 行	140 行	40% ↓
类型覆盖率	~30%	100%	-

4.2 优化手段详解

1. 异步 I/O 多路复用

hljs python

# 旧版本：同步阻塞，一个请求一个线程
@app.route('/query')
def query():
    answer = db.query()  # 阻塞 10ms
    ai_resp = requests.post()  # 阻塞 500ms
    return jsonify(ai_resp)  # 总共 510ms

# 新版本：异步非阻塞，一个线程处理多个请求
@app.get("/api/v1/query")
async def query():
    answer = await db.query()  # 释放控制权
    ai_resp = await httpx.post()  # 释放控制权
    return ai_resp  # 总共 <50ms（并发处理）

2. 三级缓存系统

hljs python

class QueryService:
    def __init__(self, db_repo, cache_repo, ai_service):
        self.db = db_repo
        self.cache = cache_repo
        self.ai = ai_service

    async def query(self, title: str, options: list[str], type: str):
        # L1: 内存缓存（最快）
        cached = await self.cache.get(title)
        if cached:
            return cached

        # L2: 数据库缓存
        db_answer = await self.db.find_by_question(title)
        if db_answer:
            await self.cache.set(title, db_answer)
            return db_answer

        # L3: AI API（最慢）
        ai_answer = await self.ai.generate_answer(...)
        await self.db.create(...)
        await self.cache.set(title, ai_answer)
        return ai_answer

3. 智能重试机制

hljs python

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_ai_api(prompt: str):
    try:
        return await httpx.post(api_url, json={"prompt": prompt})
    except httpx.TimeoutError:
        logger.warning(f"AI API timeout, retrying...")
        raise

4.3 压测数据

测试环境：

CPU: 4 核心
内存: 8GB
数据库: SQLite
AI API: 本地 Mock（延迟 50ms）

压测工具：locust -f load_test.py --host=http://localhost:8000

结果：

Flask 版本:
- QPS: 4.2
- 平均响应时间: 238ms
- 失败率: 12%

FastAPI 版本:
- QPS: 217.5
- 平均响应时间: 46ms
- 失败率: 0%

五、踩坑总结：给后来者的建议

5.1 异步编程的常见陷阱

陷阱 #1：混用同步和异步代码

hljs python

# ❌ 错误：在异步函数中调用同步库
async def query():
    result = requests.get(url)  # 阻塞整个事件循环！
    return result

# ✅ 正确：使用异步库
async def query():
    async with httpx.AsyncClient() as client:
        result = await client.get(url)
    return result

陷阱 #2：忘记 await

hljs python

# ❌ 错误：没有 await，协程不会执行
async def query():
    result = db.query()  # 返回一个协程对象，不是结果！
    return result

# ✅ 正确：使用 await
async def query():
    result = await db.query()  # 等待协程完成
    return result

陷阱 #3：在异步函数中使用阻塞操作

hljs python

# ❌ 错误：time.sleep 是阻塞的
async def query():
    time.sleep(1)  # 阻塞整个事件循环！

# ✅ 正确：使用 asyncio.sleep
async def query():
    await asyncio.sleep(1)  # 释放控制权

5.2 FastAPI 实践建议

1. 使用依赖注入管理数据库连接

hljs python

from fastapi import Depends

async def get_db():
    async with AsyncSession(engine) as session:
        yield session

@app.get("/query")
async def query(db: AsyncSession = Depends(get_db)):
    result = await db.execute(...)

2. 合理使用 Pydantic 模型

hljs python

# ✅ 定义清晰的 Schema
class QueryRequest(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    options: list[str] = Field(..., min_items=2)
    type: QuestionType

    class Config:
        json_schema_extra = {
            "example": {
                "title": "中国的首都是哪里？",
                "options": ["A.北京", "B.上海", "C.广州", "D.深圳"],
                "type": "single"
            }
        }

3. 统一错误处理

hljs python

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import http_exception_handler

app = FastAPI()

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"code": 0, "msg": str(exc), "data": None}
    )

5.3 开源项目维护心得

1. 文档与代码同等重要

代码写得好，用户不一定用得上
文档写得好，用户才能快速上手
我的经验：每写 100 行代码，就花 30 分钟写文档

2. 及时更新 README

每次大功能更新，都要更新 README
添加 badge 让项目更专业
提供 Quick Start 让用户 5 分钟内跑起来

3. 善用 Git 历史

hljs bash

# 查看重构前的代码
git show Miaozeqiu:main.py

# 对比新旧实现
git diff Miaozeqiu/main..HEAD -- main.py

4. 开源协议要明确

我选择了 MIT 协议（最宽松）
在 LICENSE 文件和 pyproject.toml 中都声明
README 添加 badge

六、成果展示与未来规划

6.1 当前成果

项目地址：wchiways/question-bank

核心特性：

全链路异步处理（FastAPI + AsyncIO）
类型安全（100% 类型注解）
自动文档（Swagger UI + ReDoc）
多 AI 平台支持（5 个平台）
三级缓存系统（内存 + 数据库 + AI）
智能重试机制
完善的文档（6 篇详细指南）

技术栈：

Web 框架:  FastAPI 0.127+
ORM:       SQLModel (Pydantic + SQLAlchemy)
数据库:    SQLite (aiosqlite 异步驱动)
HTTP 客户端: httpx (异步)
日志:      loguru
包管理:    uv

6.2 性能对比

与原项目相比：

并发能力提升 50 倍（4 QPS → 200+ QPS）
响应时间减半（100ms → <50ms）
代码量减少 40%（239 行 → 140 行）
文档从 0 到 6000+ 行

6.3 未来规划

短期计划（1-2 个月）：

支持更多 AI 平台（DeepSeek、Moonshot 等）
添加 PostgreSQL/MySQL 支持
实现题目批量导入功能
添加用户系统（使用量统计）
提供 Python / JavaScript SDK

中期计划（3-6 个月）：

Web 管理界面（Nuxt 3 + Nuxt UI）
题目质量评分系统
多语言支持（英文、日文）
题目分享社区

七、个人成长与反思

7.1 技术收获

通过这次重构，我深入理解了：

1. 异步编程的本质

协程 vs 线程 vs 进程
事件循环的原理
如何避免阻塞事件循环

2. 架构设计的重要性

分层架构的优势
依赖倒置原则（DIP）
Repository 模式的实践

3. FastAPI 的最佳实践

依赖注入的使用
Pydantic 数据验证
自动生成 API 文档

4. 开源项目维护

文档工程的重要性
Git 历史的管理
开源协议的选择

7.2 软技能提升

1. 时间管理

三天完成重构（每天 8+ 小时）
合理安排优先级（架构 → 功能 → 文档）

2. 问题解决能力

遇到 bug 不慌张
学会使用 GitHub Issues 搜索
善用日志和调试工具

3. 文档写作能力

技术文档要简洁明了
多用示例和图表
从用户角度思考

7.3 遗憾与反思

遗憾：

没有写单元测试（时间紧迫）
没有做 CI/CD（GitHub Actions）
Docker 镜像没有推送到 Docker Hub
没有做性能压测（只用 locust 简单测试）

反思：

下次重构前，先写测试（TDD）
使用 pytest-asyncio 编写异步测试
集成 GitHub Actions 自动化测试
使用 Docker Hub 自动构建镜像

八、总结：重构，值得吗？

8.1 重构的价值

这次重构投入了大约 24 小时（三天 × 8 小时），但收获是巨大的：

技术收益：

性能提升 50 倍
代码更简洁（-40%）
类型安全（100% 覆盖）
易于维护和扩展

个人成长：

深入理解异步编程
掌握 FastAPI 最佳实践
提升架构设计能力
学会了文档工程

开源社区：

为开源生态贡献力量
帮助更多开发者学习异步编程

8.2 给重构者的建议

什么时候该重构？

代码还能跑，别动它（过犹不及）
性能成为瓶颈（如本项目的并发问题）
技术债务严重（每次修改都头疼）
有新的需求（如本项目需要支持多 AI 平台）

重构的黄金法则：

先写测试（保证重构不破坏原有功能）
小步快跑（分阶段重构，不要一次性改太多）
保留历史（用 Git 分支，随时可以回退）
文档同步（代码和文档一起更新）

重构的心态：

不要追求完美，先让它能跑
接受临时的"丑代码"（标记 TODO，以后优化）
重构是一个持续的过程，不是一锤子买卖

九、致谢

感谢以下项目和开发者：

原项目作者 Miaozeqiu 的 ai-ocs-question_bank
FastAPI 框架作者 Sebastián Ramírez
SQLModel 作者 Tiangolo
uv 包管理器团队 Astral

同时感谢所有为本项目提 Issue 和 PR 的开发者！

十、参考资源

官方文档：

推荐阅读：

项目地址：

GitHub: wchiways/question-bank
博客: chiway.blog

结语

从 Flask 到 FastAPI，从同步到异步，从单文件到分层架构，这三天三夜的重构之旅，让我深刻体会到了代码之美和工程之术。

重构不仅仅是重写代码，更是对架构设计的思考、对性能优化的探索、对工程实践的理解。如果你也在犹豫是否要重构，我的建议是：

Just do it! 但要循序渐进。

希望这篇文章能对你的项目重构有所帮助。如果你有任何问题，欢迎在 GitHub Issues 中讨论！

Happy Coding!

作者简介：Chiway Wang，硬件工程师、业余全栈开发者，热爱开源，专注于 Python、Vue.js 和C/C++。

Email: wchiway@163.com

Blog: chiway.blog

GitHub: @wchiways