mingxi_platform/docs/mingxi-vision-架构设计.md

# mingxi-vision 技术架构与设计分析

> 版本：v1.0 · 日期：2026-05-24  
> 硬件约束：联想 Y7000P · RTX 4060 Laptop 8GB · Python 3.8 · Windows 11  
> 上游依赖：`yolo_classification_system/yolo8/inference.py`（迁移基础）

---

## 1. 定位与职责边界

`mingxi-vision` 是整个明析平台的**推理微服务**，职责单一：

```
接收图像 → 前处理 → 模型推理 → 后处理 → 返回结构化缺陷JSON
```

它**不**做的事：
- 不持久化数据（无数据库依赖）
- 不做鉴权（内网服务，由 backend 代理）
- 不管理产线/批次业务（backend 的职责）
- 不直接控制相机（capture 的职责）

---

## 2. 框架选型：FastAPI

### 为什么不用 Django

现有系统用 Django，但推理服务有不同的特征：

| 维度 | Django（现有） | FastAPI（选用） |
|------|--------------|----------------|
| 定位 | 全功能Web框架 | 轻量API框架 |
| 启动时间 | ~3-5s | ~0.5s |
| ORM/DB | 必须配置 | 无需数据库 |
| 异步支持 | 有限（Django 4+ 部分支持） | 原生 async/await |
| 自动文档 | 需要额外配置 | 内置 /docs (Swagger) |
| 推理场景适配 | 过重 | 刚好合适 |

### Python 3.8 的约束

FastAPI 在 Python 3.8 下完全可用，需注意：
- 类型注解用 `Optional[X]` 而非 `X | None`（3.10+ 语法）
- `from __future__ import annotations` 可缓解部分问题
- Pydantic v1（随 FastAPI 早期版本）在 3.8 稳定

```
# requirements.txt 核心版本锁定
fastapi==0.104.1          # 3.8 兼容的最后稳定版系列
uvicorn[standard]==0.24.0
pydantic==1.10.13         # v1，Python 3.8 最稳定
ultralytics==8.0.235      # 训练阶段用
onnxruntime-gpu==1.16.3   # 推理部署阶段用
opencv-python-headless==4.8.1.78
numpy==1.24.4             # 3.8 + torch 2.0 兼容版本
python-multipart==0.0.6   # FastAPI 文件上传必须
```

---

## 3. 双运行时策略

这是 mingxi-vision 最核心的设计决策。

### 问题

- **训练阶段**：用 `ultralytics` 的 `.pt` 格式，便于迭代和验证
- **部署阶段**：用 `.onnx` 格式，跨平台、无需安装 PyTorch、性能更稳定
- 两套格式的推理 API 有差异，需要统一抽象

### 解决方案：运行时适配器（Adapter 模式）

```
               ┌─────────────────────────────┐
               │      InferenceEngine        │  ← 统一接口
               │  detect(image) → [Defect]   │
               └──────────┬──────────────────┘
                          │ 根据配置选择
          ┌───────────────┴────────────────┐
          ▼                                ▼
 UltralyticsAdapter               OnnxRuntimeAdapter
 (.pt 文件，训练验证用)            (.onnx 文件，生产部署用)
 ultralytics.YOLO                 onnxruntime.InferenceSession
```

### 代码设计

```python
# engine/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List
import numpy as np

@dataclass
class DefectBox:
    class_id: int
    class_name: str
    class_name_zh: str
    confidence: float
    severity: str          # fatal / major / minor / rework / none
    box_xyxy: List[float]  # [x1, y1, x2, y2]，像素坐标


@dataclass
class DetectResult:
    defects: List[DefectBox]
    duration_ms: float
    image_width: int
    image_height: int
    model_version: str

    @property
    def defect_count(self) -> int:
        return len(self.defects)

    @property
    def max_severity(self) -> str:
        order = ["fatal", "major", "minor", "rework", "none"]
        found = {d.severity for d in self.defects}
        for s in order:
            if s in found:
                return s
        return "none"

    @property
    def avg_confidence(self):
        if not self.defects:
            return None
        return round(sum(d.confidence for d in self.defects) / len(self.defects), 4)


class BaseInferenceEngine(ABC):
    @abstractmethod
    def detect(self, image: np.ndarray, conf: float) -> DetectResult:
        ...

    @abstractmethod
    def warmup(self) -> None:
        """启动时预热，避免第一次推理延迟"""
        ...
```

```python
# engine/ultralytics_adapter.py
import time
import numpy as np
from .base import BaseInferenceEngine, DefectBox, DetectResult
from .defect_classes import DEFECT_CLASSES

class UltralyticsAdapter(BaseInferenceEngine):
    """开发/训练验证阶段使用"""

    def __init__(self, weights_path: str):
        from ultralytics import YOLO
        self._model = YOLO(weights_path)
        self._weights_path = weights_path

    def warmup(self):
        dummy = np.zeros((640, 640, 3), dtype=np.uint8)
        self._model.predict(source=dummy, verbose=False, conf=0.01)

    def detect(self, image: np.ndarray, conf: float = 0.45) -> DetectResult:
        h, w = image.shape[:2]
        t0 = time.perf_counter()
        results = self._model.predict(source=image, verbose=False, conf=conf)
        duration_ms = (time.perf_counter() - t0) * 1000

        defects = []
        result = results[0]
        if result.boxes is not None and len(result.boxes):
            xyxy   = result.boxes.xyxy.cpu().numpy()
            confs  = result.boxes.conf.cpu().numpy()
            clsids = result.boxes.cls.cpu().numpy()
            for i in range(len(xyxy)):
                cid = int(clsids[i])
                meta = DEFECT_CLASSES.get(cid, {
                    "name": str(cid), "zh": str(cid), "severity": "minor"
                })
                defects.append(DefectBox(
                    class_id=cid,
                    class_name=meta["name"],
                    class_name_zh=meta["zh"],
                    confidence=round(float(confs[i]), 4),
                    severity=meta["severity"],
                    box_xyxy=[round(float(x), 1) for x in xyxy[i]],
                ))

        return DetectResult(
            defects=defects,
            duration_ms=round(duration_ms, 1),
            image_width=w,
            image_height=h,
            model_version=str(self._weights_path),
        )
```

```python
# engine/onnx_adapter.py
import time
import cv2
import numpy as np
import onnxruntime as ort
from .base import BaseInferenceEngine, DefectBox, DetectResult
from .defect_classes import DEFECT_CLASSES

class OnnxRuntimeAdapter(BaseInferenceEngine):
    """生产部署阶段使用，无 PyTorch 依赖"""

    def __init__(self, model_path: str, device: str = "cuda"):
        providers = (
            ["CUDAExecutionProvider", "CPUExecutionProvider"]
            if device == "cuda"
            else ["CPUExecutionProvider"]
        )
        self._session = ort.InferenceSession(model_path, providers=providers)
        self._input_name  = self._session.get_inputs()[0].name
        self._input_shape = self._session.get_inputs()[0].shape  # [1,3,640,640]
        self._imgsz = self._input_shape[2]  # 通常 640
        self._model_path = model_path

    def warmup(self):
        dummy = np.zeros(
            (1, 3, self._imgsz, self._imgsz), dtype=np.float32
        )
        self._session.run(None, {self._input_name: dummy})

    def _preprocess(self, image: np.ndarray):
        """BGR → RGB → letterbox → NCHW float32 [0,1]"""
        img = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        img, ratio, (dw, dh) = self._letterbox(img, self._imgsz)
        img = img.astype(np.float32) / 255.0
        img = np.transpose(img, (2, 0, 1))   # HWC → CHW
        img = np.expand_dims(img, 0)          # CHW → NCHW
        return img, ratio, dw, dh

    @staticmethod
    def _letterbox(img, new_size=640):
        h, w = img.shape[:2]
        ratio = min(new_size / h, new_size / w)
        nh, nw = int(h * ratio), int(w * ratio)
        img = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_LINEAR)
        dw = (new_size - nw) / 2
        dh = (new_size - nh) / 2
        top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1))
        left, right  = int(round(dw - 0.1)), int(round(dw + 0.1))
        img = cv2.copyMakeBorder(img, top, bottom, left, right,
                                  cv2.BORDER_CONSTANT, value=(114, 114, 114))
        return img, ratio, dw, dh

    def _postprocess(self, outputs, orig_h, orig_w, ratio, dw, dh, conf_thres):
        """YOLOv8 ONNX 输出解码：[1, 84, 8400] → DefectBox列表"""
        pred = outputs[0][0]              # [84, 8400]
        pred = pred.T                     # [8400, 84]
        boxes  = pred[:, :4]             # cx,cy,w,h
        scores = pred[:, 4:]             # [8400, num_classes]

        class_ids   = np.argmax(scores, axis=1)
        confidences = scores[np.arange(len(scores)), class_ids]

        mask = confidences > conf_thres
        boxes, class_ids, confidences = (
            boxes[mask], class_ids[mask], confidences[mask]
        )
        if len(boxes) == 0:
            return []

        # cx,cy,w,h → x1,y1,x2,y2
        x1 = boxes[:, 0] - boxes[:, 2] / 2
        y1 = boxes[:, 1] - boxes[:, 3] / 2
        x2 = boxes[:, 0] + boxes[:, 2] / 2
        y2 = boxes[:, 1] + boxes[:, 3] / 2

        # 去除 letterbox padding，还原到原始坐标
        x1 = np.clip((x1 - dw) / ratio, 0, orig_w)
        y1 = np.clip((y1 - dh) / ratio, 0, orig_h)
        x2 = np.clip((x2 - dw) / ratio, 0, orig_w)
        y2 = np.clip((y2 - dh) / ratio, 0, orig_h)

        # NMS
        nms_ids = cv2.dnn.NMSBoxes(
            np.stack([x1, y1, x2 - x1, y2 - y1], axis=1).tolist(),
            confidences.tolist(), conf_thres, iou_threshold=0.45
        )
        if len(nms_ids) == 0:
            return []

        defects = []
        for idx in nms_ids.flatten():
            cid  = int(class_ids[idx])
            meta = DEFECT_CLASSES.get(cid, {
                "name": str(cid), "zh": str(cid), "severity": "minor"
            })
            defects.append(DefectBox(
                class_id=cid,
                class_name=meta["name"],
                class_name_zh=meta["zh"],
                confidence=round(float(confidences[idx]), 4),
                severity=meta["severity"],
                box_xyxy=[
                    round(float(x1[idx]), 1), round(float(y1[idx]), 1),
                    round(float(x2[idx]), 1), round(float(y2[idx]), 1),
                ],
            ))
        return defects

    def detect(self, image: np.ndarray, conf: float = 0.45) -> DetectResult:
        orig_h, orig_w = image.shape[:2]
        inp, ratio, dw, dh = self._preprocess(image)

        t0 = time.perf_counter()
        outputs = self._session.run(None, {self._input_name: inp})
        duration_ms = (time.perf_counter() - t0) * 1000

        defects = self._postprocess(outputs, orig_h, orig_w, ratio, dw, dh, conf)

        return DetectResult(
            defects=defects,
            duration_ms=round(duration_ms, 1),
            image_width=orig_w,
            image_height=orig_h,
            model_version=str(self._model_path),
        )
```

---

## 4. 模型管理：单例 + 启动加载

### 问题根源

现有 `inference.py` 的 `_MODEL_CACHE` 是进程级字典，首次请求时才加载模型（懒加载）。对推理服务来说，这会导致第一个请求有数秒延迟，在演示现场是灾难性的。

### 解决方案：FastAPI 生命周期钩子

```python
# engine/loader.py
from typing import Optional
from .base import BaseInferenceEngine
from config import settings

_engine: Optional[BaseInferenceEngine] = None


def get_engine() -> BaseInferenceEngine:
    if _engine is None:
        raise RuntimeError("推理引擎未初始化，请检查启动日志")
    return _engine


def init_engine() -> BaseInferenceEngine:
    global _engine
    if settings.runtime == "onnx":
        from .onnx_adapter import OnnxRuntimeAdapter
        _engine = OnnxRuntimeAdapter(
            model_path=settings.model_path,
            device=settings.device,
        )
    else:
        from .ultralytics_adapter import UltralyticsAdapter
        _engine = UltralyticsAdapter(weights_path=settings.model_path)

    _engine.warmup()   # 预热，消除第一次推理的延迟
    return _engine
```

```python
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from engine.loader import init_engine
from api.detect import router as detect_router
from api.health import router as health_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载模型
    engine = init_engine()
    print(f"[mingxi-vision] 引擎就绪: {engine.__class__.__name__}")
    yield
    # 关闭时释放资源（ONNX session会自动GC）


app = FastAPI(
    title="明析推理服务",
    description="PCB缺陷检测推理接口",
    version="1.0.0",
    lifespan=lifespan,
)

app.include_router(detect_router, prefix="/api")
app.include_router(health_router, prefix="/api")
```

---

## 5. 图像接收与前处理

### 接收来源

mingxi-vision 接受两种图像来源，用同一个端点处理：

```python
# api/detect.py
import cv2
import numpy as np
from fastapi import APIRouter, File, Form, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from engine.loader import get_engine
from engine.annotator import draw_boxes
from schema import DetectResponse
import base64, uuid

router = APIRouter()


@router.post("/detect", response_model=DetectResponse)
async def detect(
    image: UploadFile = File(...),
    conf: float = Form(default=0.45, ge=0.01, le=0.99),
    line_id: str = Form(default=""),
    batch_id: str = Form(default=""),
    return_annotated: bool = Form(default=False),
):
    # 1. 读取图像字节
    raw = await image.read()
    if len(raw) > 20 * 1024 * 1024:   # 20MB 上限
        raise HTTPException(status_code=413, detail="图像文件过大（上限20MB）")

    # 2. 解码为 numpy BGR
    arr = np.frombuffer(raw, dtype=np.uint8)
    img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
    if img is None:
        raise HTTPException(status_code=422, detail="无法解码图像，请检查文件格式")

    # 3. 推理
    engine = get_engine()
    result = engine.detect(img, conf=conf)

    # 4. 可选：返回标注图
    annotated_b64 = None
    if return_annotated:
        annotated = draw_boxes(img.copy(), result.defects)
        _, buf = cv2.imencode(".jpg", annotated, [cv2.IMWRITE_JPEG_QUALITY, 85])
        annotated_b64 = base64.b64encode(buf.tobytes()).decode()

    return DetectResponse(
        task_id=str(uuid.uuid4()),
        line_id=line_id,
        batch_id=batch_id,
        duration_ms=result.duration_ms,
        image_width=result.image_width,
        image_height=result.image_height,
        defect_count=result.defect_count,
        max_severity=result.max_severity,
        avg_confidence=result.avg_confidence,
        defects=[d.__dict__ for d in result.defects],
        annotated_image_b64=annotated_b64,
        model_version=result.model_version,
    )
```

### 中文路径问题（继承自现有代码）

现有 `inference.py` 已有 `_copy_to_ascii_temp_input()` 解决中文路径导致 OpenCV 无法读图的问题。`mingxi-vision` 改用 **内存读取**（`np.frombuffer` + `cv2.imdecode`），从源头消除路径问题，不再需要这个 workaround。

---

## 6. 后处理：标注图生成

```python
# engine/annotator.py
import cv2
import numpy as np
from typing import List
from .base import DefectBox

SEVERITY_COLORS = {
    "fatal":  (0,   0,   220),   # 红（BGR）
    "major":  (0,   128, 255),   # 橙
    "minor":  (0,   215, 255),   # 黄
    "rework": (255, 165, 0  ),   # 蓝
    "none":   (180, 180, 180),   # 灰
}


def draw_boxes(image: np.ndarray, defects: List[DefectBox]) -> np.ndarray:
    for d in defects:
        x1, y1, x2, y2 = [int(v) for v in d.box_xyxy]
        color = SEVERITY_COLORS.get(d.severity, (180, 180, 180))
        cv2.rectangle(image, (x1, y1), (x2, y2), color, 2)

        label = f"{d.class_name_zh} {d.confidence:.2f}"
        (tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.55, 1)
        cv2.rectangle(image, (x1, y1 - th - 6), (x1 + tw + 4, y1), color, -1)
        cv2.putText(
            image, label, (x1 + 2, y1 - 4),
            cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA
        )
    return image
```

---

## 7. 缺陷类别定义

```python
# engine/defect_classes.py
# 与 HRIPCB 数据集的 6 类对齐，扩展至 10 类（含真实产线样本迁移学习后）

DEFECT_CLASSES = {
    # ── HRIPCB 基础 6 类（公开数据集，开箱即用）──
    0: {"name": "missing_hole",    "zh": "缺孔",   "severity": "fatal"},
    1: {"name": "mouse_bite",      "zh": "鼠咬",   "severity": "major"},
    2: {"name": "open_circuit",    "zh": "断路",   "severity": "fatal"},
    3: {"name": "short_circuit",   "zh": "短路",   "severity": "fatal"},
    4: {"name": "spur",            "zh": "毛刺",   "severity": "minor"},
    5: {"name": "spurious_copper", "zh": "余铜",   "severity": "major"},
    # ── 扩展类（真实产线样本微调后启用）──
    6: {"name": "oxidation",       "zh": "氧化",   "severity": "minor"},
    7: {"name": "solder_ball",     "zh": "锡珠",   "severity": "rework"},
    8: {"name": "scratch",         "zh": "划痕",   "severity": "minor"},
    9: {"name": "label_error",     "zh": "标签错贴","severity": "rework"},
}

SEVERITY_ORDER = ["fatal", "major", "minor", "rework", "none"]
```

---

## 8. 配置管理

```python
# config.py
import os
from typing import Literal
from pydantic import BaseSettings   # pydantic v1


class Settings(BaseSettings):
    # 运行时选择
    runtime: Literal["ultralytics", "onnx"] = "ultralytics"
    model_path: str = "./models/pcb_defect_v1.pt"
    device: Literal["cuda", "cpu"] = "cuda"

    # 推理默认参数
    default_conf: float = 0.45
    max_image_size_mb: int = 20

    # 服务配置
    host: str = "0.0.0.0"
    port: int = 8001
    workers: int = 1           # 推理服务单 worker，GPU 不支持多进程共享

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


settings = Settings()
```

`.env` 文件示例：

```bash
# 开发阶段（ultralytics .pt）
RUNTIME=ultralytics
MODEL_PATH=./models/pcb_defect_v1.pt
DEVICE=cuda

# 生产阶段（ONNX Runtime）
# RUNTIME=onnx
# MODEL_PATH=./models/pcb_defect_v1.onnx
# DEVICE=cuda
```

---

## 9. 健康检查与诊断接口

```python
# api/health.py
import platform
from fastapi import APIRouter
from engine.loader import get_engine

router = APIRouter()


@router.get("/health")
def health():
    try:
        engine = get_engine()
        return {
            "status": "ok",
            "runtime": engine.__class__.__name__,
            "platform": platform.system(),
        }
    except RuntimeError as e:
        return {"status": "error", "detail": str(e)}


@router.get("/health/gpu")
def gpu_info():
    """开发调试用，确认 GPU 是否被正确使用"""
    info = {"cuda_available": False, "onnx_providers": []}
    try:
        import torch
        info["cuda_available"] = torch.cuda.is_available()
        if torch.cuda.is_available():
            info["gpu_name"] = torch.cuda.get_device_name(0)
            info["vram_total_gb"] = round(
                torch.cuda.get_device_properties(0).total_memory / 1e9, 1
            )
    except ImportError:
        pass
    try:
        import onnxruntime as ort
        info["onnx_providers"] = ort.get_available_providers()
    except ImportError:
        pass
    return info
```

---

## 10. 完整目录结构

```
mingxi-vision/
├── api/
│   ├── __init__.py
│   ├── detect.py          # POST /api/detect（核心推理接口）
│   └── health.py          # GET  /api/health, /api/health/gpu
├── engine/
│   ├── __init__.py
│   ├── base.py            # DefectBox, DetectResult, BaseInferenceEngine
│   ├── loader.py          # 单例管理 + init_engine()
│   ├── ultralytics_adapter.py   # .pt 推理（开发阶段）
│   ├── onnx_adapter.py          # .onnx 推理（生产阶段）
│   ├── annotator.py       # 在图像上绘制标注框
│   └── defect_classes.py  # 缺陷类别 + 等级映射表
├── models/
│   ├── pcb_defect_v1.pt   # ultralytics 训练产物（gitignore）
│   └── pcb_defect_v1.onnx # 导出的 ONNX 模型（gitignore）
├── scripts/
│   ├── export_onnx.py     # pt → onnx 导出脚本
│   └── benchmark.py       # 本地推理性能测试
├── tests/
│   ├── test_detect_api.py
│   └── fixtures/          # 测试用 PCB 图片
├── schema.py              # Pydantic 响应模型
├── config.py              # Settings（pydantic BaseSettings）
├── main.py                # FastAPI app + lifespan
├── requirements.txt
├── .env.example
└── README.md
```

---

## 11. 响应 Schema

```python
# schema.py
from typing import List, Optional
from pydantic import BaseModel


class DefectItem(BaseModel):
    class_id: int
    class_name: str
    class_name_zh: str
    confidence: float
    severity: str          # fatal / major / minor / rework / none
    box_xyxy: List[float]  # [x1, y1, x2, y2]


class DetectResponse(BaseModel):
    task_id: str
    line_id: str
    batch_id: str
    duration_ms: float
    image_width: int
    image_height: int
    defect_count: int
    max_severity: str      # 本次检测中最高等级缺陷
    avg_confidence: Optional[float]
    defects: List[DefectItem]
    annotated_image_b64: Optional[str]   # 仅 return_annotated=true 时有值
    model_version: str
```

实际响应示例：

```json
{
  "task_id": "a3f1c2d4-...",
  "line_id": "LINE-01",
  "batch_id": "BAT-20260524-001",
  "duration_ms": 41.3,
  "image_width": 3072,
  "image_height": 2048,
  "defect_count": 2,
  "max_severity": "fatal",
  "avg_confidence": 0.8762,
  "defects": [
    {
      "class_id": 2,
      "class_name": "open_circuit",
      "class_name_zh": "断路",
      "confidence": 0.9134,
      "severity": "fatal",
      "box_xyxy": [234.5, 891.2, 312.8, 943.7]
    },
    {
      "class_id": 4,
      "class_name": "spur",
      "class_name_zh": "毛刺",
      "confidence": 0.8390,
      "severity": "minor",
      "box_xyxy": [1204.1, 456.3, 1251.9, 489.0]
    }
  ],
  "annotated_image_b64": null,
  "model_version": "./models/pcb_defect_v1.pt"
}
```

---

## 12. 性能分析（Y7000P RTX 4060）

### 推理延迟预估

| 运行时 | 模型 | 分辨率 | 单次推理 | 含预处理+后处理 |
|--------|------|--------|---------|--------------|
| ultralytics | YOLOv8n | 640×640 | ~12ms | ~20ms |
| ultralytics | YOLOv8s | 640×640 | ~20ms | ~30ms |
| ONNX Runtime GPU | YOLOv8n | 640×640 | ~8ms  | ~15ms |
| ONNX Runtime CPU | YOLOv8n | 640×640 | ~80ms | ~100ms |

> 数据来源：`pcb缺陷检测初步方案.md` 引用的 Y7000P 实测基准

**PCB 后道复判场景需求**：触发模式，每块板约 1-3 秒间隔，15ms 的推理延迟完全满足。

### 显存占用

| 模型 | 显存占用 |
|------|---------|
| YOLOv8n（ultralytics） | ~1.2GB |
| YOLOv8s（ultralytics） | ~2.4GB |
| YOLOv8n（ONNX CUDA EP） | ~0.8GB |

RTX 4060 8GB 剩余 6GB+ 可用，不构成瓶颈。

### 并发限制

```python
# main.py 启动参数
# workers=1：GPU 不支持多进程共享同一 CUDA Context
# 推理天然串行，配合 FastAPI 的 async 处理并发等待队列
uvicorn main:app --host 0.0.0.0 --port 8001 --workers 1
```

PCB 产线触发模式下，单工位请求天然串行，`workers=1` 完全够用。若需多产线并发，在 `mingxi-backend` 侧做请求队列即可。

---

## 13. 模型导出（pt → onnx）

```python
# scripts/export_onnx.py
from ultralytics import YOLO

model = YOLO("./models/pcb_defect_v1.pt")
model.export(
    format="onnx",
    imgsz=640,
    opset=12,          # onnxruntime 1.16 兼容
    simplify=True,     # onnx-simplifier 优化计算图
    dynamic=False,     # 固定 batch=1，推理服务不需要动态 batch
)
# 产物：pcb_defect_v1.onnx
```

切换到 ONNX 的时机：

```
1. YOLOv8n 在验证集 mAP@0.5 > 0.90 → 导出 ONNX
2. 运行 scripts/benchmark.py 对比两个运行时延迟
3. 确认输出一致后，修改 .env 的 RUNTIME=onnx
4. 重启服务，观察 /api/health 确认切换成功
```

---

## 14. 与现有代码的对应关系

| 现有 `inference.py` 函数 | mingxi-vision 对应位置 | 变化 |
|------------------------|----------------------|------|
| `_MODEL_CACHE` | `engine/loader.py` 单例 | 从懒加载改为启动时加载 |
| `_get_model()` | `loader.init_engine()` | 新增 ONNX 分支 |
| `_copy_to_ascii_temp_input()` | **已删除** | 改用内存读图，无需文件路径 |
| `_parse_result()` | `ultralytics_adapter.detect()` | 增加 severity 映射 |
| `run_image_inference()` | `engine/ultralytics_adapter.py` + `engine/onnx_adapter.py` | 拆成两个运行时 |
| `run_camera_inference()` | **不迁移** | 由 `mingxi-capture` 负责 |
| `run_video_inference()` | **暂不迁移** | 此版本不需要 |
| Django `MEDIA_ROOT` 文件落盘 | **已删除** | 推理服务无状态，不落盘 |

---

## 15. 关键设计决策汇总

| 决策 | 选择 | 理由 |
|------|------|------|
| Web 框架 | FastAPI | 轻量、无需 DB、原生 async、自动文档 |
| 模型运行时 | 双运行时（适配器模式） | 开发用 .pt 方便迭代，生产用 .onnx 稳定 |
| 模型加载时机 | 启动时加载 + warmup | 消除首次推理延迟，路演不卡顿 |
| 图像读取 | 内存读取（np.frombuffer） | 彻底消除中文路径问题 |
| 标注图返回 | 按需（return_annotated=true） | 默认不返回，减少响应体积和编码开销 |
| 并发模型 | workers=1，单进程 | GPU CUDA Context 不支持多进程共享 |
| 鉴权 | 无 | 内网服务，由 backend 代理调用 |
| 数据持久化 | 无 | 无状态推理服务，落库交给 backend |
| NMS 后处理 | ONNX 适配器自实现，ultralytics 自带 | 两路输出语义一致 |