21 KiB
21 KiB
Invoice Master POC v2 - 总体架构审查报告
审查日期: 2026-02-01
审查人: Claude Code
项目路径: /Users/yiukai/Documents/git/invoice-master-poc-v2
架构概述
整体架构图
┌─────────────────────────────────────────────────────────────────┐
│ Frontend (React) │
│ Vite + TypeScript + TailwindCSS │
└─────────────────────────────┬───────────────────────────────────┘
│ HTTP/REST
┌─────────────────────────────▼───────────────────────────────────┐
│ Inference Service (FastAPI) │
│ ┌──────────────┬──────────────┬──────────────┬──────────────┐ │
│ │ Public API │ Admin API │ Training API│ Batch API │ │
│ └──────────────┴──────────────┴──────────────┴──────────────┘ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Service Layer │ │
│ │ InferenceService │ AsyncProcessing │ BatchUpload │ Dataset │ │
│ └────────────────────────────────────────────────────────────┘ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Data Layer │ │
│ │ AdminDB │ AsyncRequestDB │ SQLModel │ PostgreSQL │ │
│ └────────────────────────────────────────────────────────────┘ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Core Components │ │
│ │ RateLimiter │ Schedulers │ TaskQueues │ Auth │ │
│ └────────────────────────────────────────────────────────────┘ │
└─────────────────────────────┬───────────────────────────────────┘
│ PostgreSQL
┌─────────────────────────────▼───────────────────────────────────┐
│ Training Service (GPU) │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ CLI: train │ autolabel │ analyze │ validate │ │
│ └────────────────────────────────────────────────────────────┘ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ YOLO: db_dataset │ annotation_generator │ │
│ └────────────────────────────────────────────────────────────┘ │
│ ┌────────────────────────────────────────────────────────────┐ │
│ │ Processing: CPU Pool │ GPU Pool │ Task Dispatcher │ │
│ └────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
┌─────────┴─────────┐
▼ ▼
┌──────────────┐ ┌──────────────┐
│ Shared │ │ Storage │
│ PDF │ OCR │ │ Local/Azure/ │
│ Normalize │ │ S3 │
└──────────────┘ └──────────────┘
技术栈
| 层级 | 技术 | 评估 |
|---|---|---|
| 前端 | React + Vite + TypeScript + TailwindCSS | ✅ 现代栈 |
| API 框架 | FastAPI | ✅ 高性能，类型安全 |
| 数据库 | PostgreSQL + SQLModel | ✅ 类型安全 ORM |
| 目标检测 | YOLOv11 (Ultralytics) | ✅ 业界标准 |
| OCR | PaddleOCR v5 | ✅ 支持瑞典语 |
| 部署 | Docker + Azure/AWS | ✅ 云原生 |
架构优势
1. Monorepo 结构 ✅
packages/
├── shared/ # 共享库 - 无外部依赖
├── training/ # 训练服务 - 依赖 shared
└── inference/ # 推理服务 - 依赖 shared
优点:
- 清晰的包边界，无循环依赖
- 独立部署，training 按需启动
- 代码复用率高
2. 分层架构 ✅
API Routes (web/api/v1/)
↓
Service Layer (web/services/)
↓
Data Layer (data/)
↓
Database (PostgreSQL)
优点:
- 职责分离明确
- 便于单元测试
- 可替换底层实现
3. 依赖注入 ✅
# FastAPI Depends 使用得当
@router.post("/infer")
async def infer(
file: UploadFile,
db: AdminDB = Depends(get_admin_db), # 注入
token: str = Depends(validate_admin_token),
):
4. 存储抽象层 ✅
# 统一接口，支持多后端
class StorageBackend(ABC):
def upload(self, source: Path, destination: str) -> None: ...
def download(self, source: str, destination: Path) -> None: ...
def get_presigned_url(self, path: str) -> str: ...
# 实现: LocalStorageBackend, AzureStorageBackend, S3StorageBackend
5. 动态模型管理 ✅
# 数据库驱动的模型切换
def get_active_model_path() -> Path | None:
db = AdminDB()
active_model = db.get_active_model_version()
return active_model.model_path if active_model else None
inference_service = InferenceService(
model_path_resolver=get_active_model_path,
)
6. 任务队列分离 ✅
# 不同类型任务使用不同队列
- AsyncTaskQueue: 异步推理任务
- BatchQueue: 批量上传任务
- TrainingScheduler: 训练任务调度
- AutoLabelScheduler: 自动标注调度
架构问题与风险
1. 数据库层职责过重 ⚠️ 中风险
问题: AdminDB 类过大，违反单一职责原则
# packages/inference/inference/data/admin_db.py
class AdminDB:
# Token 管理 (5 个方法)
def is_valid_admin_token(self, token: str) -> bool: ...
def create_admin_token(self, token: str, name: str): ...
# 文档管理 (8 个方法)
def create_document(self, ...): ...
def get_document(self, doc_id: str): ...
# 标注管理 (6 个方法)
def create_annotation(self, ...): ...
def get_annotations(self, doc_id: str): ...
# 训练任务 (7 个方法)
def create_training_task(self, ...): ...
def update_training_task(self, ...): ...
# 数据集 (6 个方法)
def create_dataset(self, ...): ...
def get_dataset(self, dataset_id: str): ...
# 模型版本 (5 个方法)
def create_model_version(self, ...): ...
def activate_model_version(self, ...): ...
# 批处理 (4 个方法)
# 锁管理 (3 个方法)
# ... 总计 50+ 方法
影响:
- 类过大，难以维护
- 测试困难
- 不同领域变更互相影响
建议: 按领域拆分为 Repository 模式
# 建议重构
class TokenRepository:
def validate(self, token: str) -> bool: ...
def create(self, token: Token) -> None: ...
class DocumentRepository:
def find_by_id(self, doc_id: str) -> Document | None: ...
def save(self, document: Document) -> None: ...
class TrainingRepository:
def create_task(self, config: TrainingConfig) -> TrainingTask: ...
def update_task_status(self, task_id: str, status: TaskStatus): ...
class ModelRepository:
def get_active(self) -> ModelVersion | None: ...
def activate(self, version_id: str) -> None: ...
2. Service 层混合业务逻辑与技术细节 ⚠️ 中风险
问题: InferenceService 既处理业务逻辑又处理技术实现
# packages/inference/inference/web/services/inference.py
class InferenceService:
def process(self, image_bytes: bytes) -> ServiceResult:
# 1. 技术细节: 图像解码
image = Image.open(io.BytesIO(image_bytes))
# 2. 业务逻辑: 字段提取
fields = self._extract_fields(image)
# 3. 技术细节: 模型推理
detections = self._model.predict(image)
# 4. 业务逻辑: 结果验证
if not self._validate_fields(fields):
raise ValidationError()
影响:
- 难以测试业务逻辑
- 技术变更影响业务代码
- 无法切换技术实现
建议: 引入领域层和适配器模式
# 领域层 - 纯业务逻辑
@dataclass
class InvoiceDocument:
document_id: str
pages: list[Page]
class InvoiceExtractor:
"""纯业务逻辑，不依赖技术实现"""
def extract(self, document: InvoiceDocument) -> InvoiceFields:
# 只处理业务规则
pass
# 适配器层 - 技术实现
class YoloFieldDetector:
"""YOLO 技术适配器"""
def __init__(self, model_path: Path):
self._model = YOLO(model_path)
def detect(self, image: np.ndarray) -> list[FieldRegion]:
return self._model.predict(image)
class PaddleOcrEngine:
"""PaddleOCR 技术适配器"""
def __init__(self):
self._ocr = PaddleOCR()
def recognize(self, image: np.ndarray, region: BoundingBox) -> str:
return self._ocr.ocr(image, region)
# 应用服务 - 协调领域和适配器
class InvoiceProcessingService:
def __init__(
self,
extractor: InvoiceExtractor,
detector: FieldDetector,
ocr: OcrEngine,
):
self._extractor = extractor
self._detector = detector
self._ocr = ocr
3. 调度器设计分散 ⚠️ 中风险
问题: 多个独立调度器缺乏统一协调
# 当前设计 - 4 个独立调度器
# 1. TrainingScheduler (core/scheduler.py)
# 2. AutoLabelScheduler (core/autolabel_scheduler.py)
# 3. AsyncTaskQueue (workers/async_queue.py)
# 4. BatchQueue (workers/batch_queue.py)
# app.py 中分别启动
start_scheduler() # 训练调度器
start_autolabel_scheduler() # 自动标注调度器
init_batch_queue() # 批处理队列
影响:
- 资源竞争风险
- 难以监控和追踪
- 任务优先级难以管理
- 重启时任务丢失
建议: 使用 Celery + Redis 统一任务队列
# 建议重构
from celery import Celery
app = Celery('invoice_master')
@app.task(bind=True, max_retries=3)
def process_inference(self, document_id: str):
"""异步推理任务"""
try:
service = get_inference_service()
result = service.process(document_id)
return result
except Exception as exc:
raise self.retry(exc=exc, countdown=60)
@app.task
def train_model(dataset_id: str, config: dict):
"""训练任务"""
training_service = get_training_service()
return training_service.train(dataset_id, config)
@app.task
def auto_label_documents(document_ids: list[str]):
"""批量自动标注"""
for doc_id in document_ids:
auto_label_document.delay(doc_id)
# 优先级队列
app.conf.task_routes = {
'tasks.process_inference': {'queue': 'high_priority'},
'tasks.train_model': {'queue': 'gpu_queue'},
'tasks.auto_label_documents': {'queue': 'low_priority'},
}
4. 配置分散 ⚠️ 低风险
问题: 配置分散在多个文件
# packages/shared/shared/config.py
DATABASE = {...}
PATHS = {...}
AUTOLABEL = {...}
# packages/inference/inference/web/config.py
@dataclass
class ModelConfig: ...
@dataclass
class ServerConfig: ...
@dataclass
class FileConfig: ...
# 环境变量
# .env 文件
影响:
- 配置难以追踪
- 可能出现不一致
- 缺少配置验证
建议: 使用 Pydantic Settings 集中管理
# config/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class DatabaseSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='DB_')
host: str = 'localhost'
port: int = 5432
name: str = 'docmaster'
user: str = 'docmaster'
password: str # 无默认值，必须设置
class StorageSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='STORAGE_')
backend: str = 'local'
base_path: str = '~/invoice-data'
azure_connection_string: str | None = None
s3_bucket: str | None = None
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file='.env',
env_file_encoding='utf-8',
)
database: DatabaseSettings = DatabaseSettings()
storage: StorageSettings = StorageSettings()
# 验证
@field_validator('database')
def validate_database(cls, v):
if not v.password:
raise ValueError('Database password is required')
return v
# 全局配置实例
settings = Settings()
5. 内存队列单点故障 ⚠️ 中风险
问题: AsyncTaskQueue 和 BatchQueue 基于内存
# workers/async_queue.py
class AsyncTaskQueue:
def __init__(self):
self._queue = Queue() # 内存队列
self._workers = []
def enqueue(self, task: AsyncTask) -> None:
self._queue.put(task) # 仅存储在内存
影响:
- 服务重启丢失所有待处理任务
- 无法水平扩展
- 任务持久化困难
建议: 使用 Redis/RabbitMQ 持久化队列
6. 缺少 API 版本迁移策略 ❓ 低风险
问题: 有 /api/v1/ 版本，但缺少升级策略
当前: /api/v1/admin/documents
未来: /api/v2/admin/documents ?
建议:
- 制定 API 版本升级流程
- 使用 Header 版本控制
- 维护版本兼容性文档
关键架构风险矩阵
| 风险项 | 概率 | 影响 | 风险等级 | 优先级 |
|---|---|---|---|---|
| 内存队列丢失任务 | 中 | 高 | 高 | 🔴 P0 |
| AdminDB 职责过重 | 高 | 中 | 中 | 🟡 P1 |
| Service 层混合 | 高 | 中 | 中 | 🟡 P1 |
| 调度器资源竞争 | 中 | 中 | 中 | 🟡 P1 |
| 配置分散 | 高 | 低 | 低 | 🟢 P2 |
| API 版本策略 | 低 | 低 | 低 | 🟢 P2 |
改进建议路线图
Phase 1: 立即执行 (本周)
1.1 拆分 AdminDB
# 创建 repositories 包
inference/data/repositories/
├── __init__.py
├── base.py # Repository 基类
├── token.py # TokenRepository
├── document.py # DocumentRepository
├── annotation.py # AnnotationRepository
├── training.py # TrainingRepository
├── dataset.py # DatasetRepository
└── model.py # ModelRepository
1.2 统一配置
# 创建统一配置模块
inference/config/
├── __init__.py
├── settings.py # Pydantic Settings
└── validators.py # 配置验证
Phase 2: 短期执行 (本月)
2.1 引入消息队列
# docker-compose.yml 添加
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
celery_worker:
build: .
command: celery -A inference.tasks worker -l info
depends_on:
- redis
- postgres
2.2 添加缓存层
# 使用 Redis 缓存热点数据
from redis import Redis
redis_client = Redis(host='localhost', port=6379)
class CachedDocumentRepository(DocumentRepository):
def find_by_id(self, doc_id: str) -> Document | None:
# 先查缓存
cached = redis_client.get(f"doc:{doc_id}")
if cached:
return Document.parse_raw(cached)
# 再查数据库
doc = super().find_by_id(doc_id)
if doc:
redis_client.setex(f"doc:{doc_id}", 3600, doc.json())
return doc
Phase 3: 长期执行 (本季度)
3.1 数据库读写分离
# 配置主从数据库
class DatabaseManager:
def __init__(self):
self._master = create_engine(MASTER_DB_URL)
self._replica = create_engine(REPLICA_DB_URL)
def get_session(self, readonly: bool = False) -> Session:
engine = self._replica if readonly else self._master
return Session(engine)
3.2 事件驱动架构
# 引入事件总线
from event_bus import EventBus
bus = EventBus()
# 发布事件
@router.post("/documents")
async def create_document(...):
doc = document_repo.save(document)
bus.publish('document.created', {'document_id': doc.id})
return doc
# 订阅事件
@bus.subscribe('document.created')
def on_document_created(event):
# 触发自动标注
auto_label_task.delay(event['document_id'])
架构演进建议
当前架构 (适合 1-10 用户)
Single Instance
├── FastAPI App
├── Memory Queues
└── PostgreSQL
目标架构 (适合 100+ 用户)
Load Balancer
├── FastAPI Instance 1
├── FastAPI Instance 2
└── FastAPI Instance N
│
┌───────┴───────┐
▼ ▼
Redis Cluster PostgreSQL
(Celery + Cache) (Master + Replica)
总结
总体评分
| 维度 | 评分 | 说明 |
|---|---|---|
| 模块化 | 8/10 | 包结构清晰，但部分类过大 |
| 可扩展性 | 7/10 | 水平扩展良好，垂直扩展受限 |
| 可维护性 | 8/10 | 分层合理，但职责边界需细化 |
| 可靠性 | 7/10 | 内存队列是单点故障 |
| 性能 | 8/10 | 异步处理良好 |
| 安全性 | 8/10 | 基础安全到位 |
| 总体 | 7.7/10 | 良好的架构基础，需优化细节 |
关键结论
- 架构设计合理: Monorepo + 分层架构适合当前规模
- 主要风险: 内存队列和数据库职责过重
- 演进路径: 引入消息队列和缓存层
- 投入产出: 当前架构可支撑到 100+ 用户，无需大规模重构
下一步行动
| 优先级 | 任务 | 预计工时 | 影响 |
|---|---|---|---|
| 🔴 P0 | 引入 Celery + Redis | 3 天 | 解决任务丢失问题 |
| 🟡 P1 | 拆分 AdminDB | 2 天 | 提升可维护性 |
| 🟡 P1 | 统一配置管理 | 1 天 | 减少配置错误 |
| 🟢 P2 | 添加缓存层 | 2 天 | 提升性能 |
| 🟢 P2 | 数据库读写分离 | 3 天 | 提升扩展性 |
附录
关键文件清单
| 文件 | 职责 | 问题 |
|---|---|---|
inference/data/admin_db.py |
数据库操作 | 类过大，需拆分 |
inference/web/services/inference.py |
推理服务 | 混合业务和技术 |
inference/web/workers/async_queue.py |
异步队列 | 内存存储，易丢失 |
inference/web/core/scheduler.py |
任务调度 | 缺少统一协调 |
shared/shared/config.py |
共享配置 | 分散管理 |