2025 年的冬天，我在 GitHub 上发现了一个有趣的项目 —— ai-ocs-question_bank by Miaozeqiu。这是一个基于 Flask 的 OCS（网课助手）题库查询系统，利用 AI 技术自动答题。作为一个正在学习异步编程的开发者，我看到了这个项目的潜力，也看到了优化的空间。
于是，我在 2025 年 12 月 27 日 fork 了项目，开始了一场为期三天的大规模架构重构。
这篇文章将记录我从接手项目到完成重构的完整心路历程，包括：
Fork 下来的项目使用的是经典的 Flask + 同步 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)主要问题：
Any 类型，IDE 无法提供智能补全.env 文件，不支持多 AI 平台配置在实际使用中，我发现：
结论：这个项目需要一次彻底的架构升级，而不是简单的 bug 修复。
在决定重构方案时，我对比了几个主流框架：
| 特性 | Flask | FastAPI | Sanic | Tornado |
|---|---|---|---|---|
| 异步支持 | ❌ 需要扩展 | ✅ 原生 | ✅ 原生 | ✅ 原生 |
| 类型提示 | ⚠️ 部分 | ✅ 100% | ⚠️ 部分 | ⚠️ 部分 |
| 自动文档 | ❌ 需要手动 | ✅ Swagger/ReDoc | ❌ 需要手动 | ❌ 需要手动 |
| 数据验证 | ⚠️ 手动 | ✅ Pydantic | ⚠️ 手动 | ⚠️ 手动 |
| 学习曲线 | 低 | 中 | 中 | 高 |
| 性能 | 基准 | 2-3x | 2x | 1.5x |
1. 原生异步支持
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. 类型安全和自动文档
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 数据验证
最终决定：使用 FastAPI + AsyncIO + SQLModel 作为新架构。
我采用了 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、缓存等）设计原则：
单一职责：每层只做一件事
依赖倒置：高层不依赖低层，都依赖抽象
# 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:
# 具体实现
...核心挑战：将所有同步操作改为异步
1. 数据库异步化
# 旧代码（同步）
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 客户端异步化
# 旧代码（同步）
import requests
response = requests.post(url, json=data)
# 新代码（异步）
import httpx
async with httpx.AsyncClient() as client:
response = await client.post(url, json=data)3. 缓存系统异步化
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 实现过期删除问题：原来的 .env 文件难以管理多个 AI 平台的配置
旧配置（.env）：
AI_API_KEY=sk-xxx
AI_API_URL=https://api.siliconflow.cn
AI_MODEL=Qwen/Qwen2.5-7B-Instruct新配置（config.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 Provider 设计：
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)Bug #1：OCS 答案格式不匹配
问题描述：
AI 返回的答案格式不统一，有时是 A.北京，有时只是 北京，导致 OCS 无法正确匹配。
解决方案：
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 answerBug #2：SOCKS 代理错误
问题：使用 SOCKS5 代理时报错 Missing dependencies for SOCKS support
解决：
# 在 pyproject.toml 中添加
dependencies = [
"httpx[socks]", # 添加 socksio 支持
]Bug #3：缓存系统 AttributeError
问题：配置结构更新后，缓存服务还在使用旧的 settings.CACHE_TYPE
解决：
# 旧代码
cache_type = settings.CACHE_TYPE
cache_ttl = settings.CACHE_TTL
# 新代码
cache_type = settings.cache.type
cache_ttl = settings.cache.ttl我发现很多开源项目代码写得很好，但文档一塌糊涂，导致用户难以上手。因此，我决定把文档工程作为重构的重要组成部分。
文档结构：
docs/
├── INSTALL.md # 安装指南（uv、pip、Docker 三种方式）
├── DOCKER.md # Docker 部署指南（1200+ 行，超详细！）
├── API.md # API 文档（请求/响应格式、错误处理、SDK 示例）
├── DEVELOPMENT.md # 开发指南（架构、工作流、编码规范）
├── OCS集成指南.md # OCS 插件配置教程
└── 配置指南.md # AI 平台配置指南Docker 部署指南亮点：
API 文档亮点：
| 指标 | Flask 版本 | FastAPI 版本 | 提升倍数 |
|---|---|---|---|
| 并发能力（QPS） | ~4 | 200+ | 50x |
| 平均响应时间 | ~100ms | <50ms | 2x |
| 内存占用 | 80MB | 60MB | 25% ↓ |
| 代码行数 | 239 行 | 140 行 | 40% ↓ |
| 类型覆盖率 | ~30% | 100% | - |
1. 异步 I/O 多路复用
# 旧版本：同步阻塞，一个请求一个线程
@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. 三级缓存系统
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_answer3. 智能重试机制
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测试环境：
压测工具：locust -f load_test.py --host=http://localhost:8000
结果：
Flask 版本:
- QPS: 4.2
- 平均响应时间: 238ms
- 失败率: 12%
FastAPI 版本:
- QPS: 217.5
- 平均响应时间: 46ms
- 失败率: 0%陷阱 #1：混用同步和异步代码
# ❌ 错误：在异步函数中调用同步库
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
# ❌ 错误：没有 await，协程不会执行
async def query():
result = db.query() # 返回一个协程对象，不是结果！
return result
# ✅ 正确：使用 await
async def query():
result = await db.query() # 等待协程完成
return result陷阱 #3：在异步函数中使用阻塞操作
# ❌ 错误：time.sleep 是阻塞的
async def query():
time.sleep(1) # 阻塞整个事件循环！
# ✅ 正确：使用 asyncio.sleep
async def query():
await asyncio.sleep(1) # 释放控制权1. 使用依赖注入管理数据库连接
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 模型
# ✅ 定义清晰的 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. 统一错误处理
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}
)1. 文档与代码同等重要
2. 及时更新 README
3. 善用 Git 历史
# 查看重构前的代码
git show Miaozeqiu:main.py
# 对比新旧实现
git diff Miaozeqiu/main..HEAD -- main.py4. 开源协议要明确
LICENSE 文件和 pyproject.toml 中都声明核心特性：
技术栈：
Web 框架: FastAPI 0.127+
ORM: SQLModel (Pydantic + SQLAlchemy)
数据库: SQLite (aiosqlite 异步驱动)
HTTP 客户端: httpx (异步)
日志: loguru
包管理: uv与原项目相比：
短期计划（1-2 个月）：
中期计划（3-6 个月）：
通过这次重构，我深入理解了：
1. 异步编程的本质
2. 架构设计的重要性
3. FastAPI 的最佳实践
4. 开源项目维护
1. 时间管理
2. 问题解决能力
3. 文档写作能力
遗憾：
反思：
pytest-asyncio 编写异步测试这次重构投入了大约 24 小时（三天 × 8 小时），但收获是巨大的：
技术收益：
个人成长：
开源社区：
什么时候该重构？
重构的黄金法则：
重构的心态：
感谢以下项目和开发者：
同时感谢所有为本项目提 Issue 和 PR 的开发者！
官方文档：
推荐阅读：
项目地址：
从 Flask 到 FastAPI，从同步到异步，从单文件到分层架构，这三天三夜的重构之旅，让我深刻体会到了代码之美和工程之术。
重构不仅仅是重写代码，更是对架构设计的思考、对性能优化的探索、对工程实践的理解。如果你也在犹豫是否要重构，我的建议是：
Just do it! 但要循序渐进。
希望这篇文章能对你的项目重构有所帮助。如果你有任何问题，欢迎在 GitHub Issues 中讨论！
Happy Coding!
作者简介：Chiway Wang，硬件工程师、业余全栈开发者，热爱开源，专注于 Python、Vue.js 和C/C++。
- Email: [email protected]
- Blog: chiway.blog
- GitHub: @wchiway