下面给出一套可复用的“在线 + 批量”推理服务模板与实践指南，围绕 FastAPI + Pydantic 构建，支持模型版本仓、A/B 分流、批量作业、监控与追踪、签名校验、数据漂移检测等。示例遵循 MIT 许可，仅使用常见开源依赖（BSD/MIT/Apache-2.0），不包含任何外部数据上传逻辑，适配容器化、S3 兼容对象存储、RabbitMQ、Prometheus/Grafana、Airflow 调度。
一、项目结构（建议）
- pyproject.toml # 依赖与构建（uv + pypi）
- uv.lock # uv 生成的锁定文件（可复现实验/构建）
- environment.yml # conda 环境（可选）
- Makefile # 本地开发脚手架
- .pre-commit-config.yaml # 统一代码风格/安全/质量
- docker/
- Dockerfile # 多阶段镜像构建（非 root）
- gunicorn_conf.py
- docker-compose.yml # 一键起服（App + Worker + RabbitMQ + MinIO）
- prometheus/
- prometheus.yml # 本地采集配置（简化示例）
- src/inference_service/
- init.py
- app.py # create_app() 组装
- api.py # FastAPI 路由与端点
- config.py # 配置中心（Pydantic Settings）
- schemas.py # I/O 模式校验（Pydantic v2）
- middleware.py # 请求 ID、计时、错误处理、限流（可选）
- metrics.py # Prometheus 指标
- logging.py # 结构化日志 + 脱敏过滤
- storage.py # S3 适配（aioboto3）
- crypto.py # 模型包签名校验（SHA256 + RSA）
- model_registry.py # 模型版本仓（按标签加载、热更新）
- routing.py # A/B/多臂流量切分
- preprocessing.py # 特征预处理（含 doctest）
- postprocessing.py # 结果后处理
- predictor.py # 推理器抽象（CPU/GPU 可选）
- batch_worker.py # 批量作业消费（RabbitMQ）
- drift.py # 数据漂移与阈值告警
- tests/
- test_preprocess.py
- test_postprocess.py
- test_api_smoke.py
- scripts/
- smoke_test.sh # 简易冒烟
- package_model.py # 模型打包示例
- sign_model.py # 生成签名
- load_test_locustfile.py # 压测脚本（可选 locust）
- README.md
- LICENSE
二、依赖与环境隔离
- pyproject.toml（使用 uv + PEP 621）
[project]
name = "inference-service"
version = "0.1.0"
description = "Reusable online & batch inference template"
license = { text = "MIT" }
requires-python = ">=3.10"
dependencies = [
"fastapi==0.115.5",
"uvicorn[standard]==0.32.0",
"pydantic==2.9.2",
"pydantic-settings==2.6.1",
"orjson==3.10.7",
"numpy==2.1.3",
"scikit-learn==1.5.2",
"aioboto3==13.1.1",
"boto3==1.35.57",
"aio-pika==9.4.1",
"prometheus-client==0.21.0",
"structlog==24.4.0",
"opentelemetry-api==1.27.0",
"opentelemetry-sdk==1.27.0",
"opentelemetry-instrumentation-fastapi==0.48b0",
"cryptography==43.0.1",
"xxhash==3.5.0"
]
[project.optional-dependencies]
gpu = ["torch==2.5.1"]
dev = ["pytest==8.3.3", "httpx==0.27.2", "pytest-asyncio==0.24.0", "ruff==0.7.3", "mypy==1.13.0", "bandit==1.7.9", "types-requests"]
s3 = ["aioboto3==13.1.1", "boto3==1.35.57"]
rabbit = ["aio-pika==9.4.1"]
[tool.uv] # uv 锁定与构建
dev-dependencies = ["pre-commit==4.0.1"]
- conda 环境（可选）
name: inference-service
channels: [conda-forge, defaults]
dependencies:
- python=3.10
- pip=24.2
- pip:
- inference-service @ file://.
- -r requirements.txt # 如需，也可以让 uv 生成 requirements 静态快照
- uv 使用
- 安装：pipx install uv
- 冻结依赖：uv lock
- 本地安装：uv sync --all-extras --frozen
- 运行：uv run uvicorn inference_service.app:create_app --factory --host 0.0.0.0 --port 8000
三、Docker 多阶段与非 root
docker/Dockerfile
# syntax=docker/dockerfile:1.7-labs
FROM python:3.11-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_DISABLE_PIP_VERSION_CHECK=1 \
UV_LINK_MODE=copy UV_SYSTEM_PYTHON=1
RUN apt-get update && apt-get install -y --no-install-recommends build-essential curl \
&& rm -rf /var/lib/apt/lists/*
FROM base AS builder
RUN curl -LsSf https://astral.sh/uv/install.sh | sh
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN /root/.cargo/bin/uv sync --frozen --no-dev
COPY src ./src
RUN /root/.cargo/bin/uv sync --frozen --no-dev
FROM base AS runtime
# 非 root
RUN useradd -m -u 10001 appuser
WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"
COPY src ./src
COPY docker/gunicorn_conf.py ./gunicorn_conf.py
ENV GUNICORN_CMD_ARGS="--config gunicorn_conf.py"
ENV OMP_NUM_THREADS=1 MKL_NUM_THREADS=1 # 避免过多并行
EXPOSE 8000
USER appuser
CMD ["uvicorn", "inference_service.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]
四、Docker Compose 一键起服
docker-compose.yml
version: "3.9"
services:
app:
build: ./docker
image: inference-service:latest
environment:
APP_ENV: dev
S3_ENDPOINT: http://minio:9000
S3_ACCESS_KEY: minioadmin
S3_SECRET_KEY: minioadmin
S3_BUCKET: models
RABBIT_URL: amqp://guest:guest@rabbitmq:5672/
PROMETHEUS_SCRAPE: "true"
MODEL_MANIFEST_S3_URI: s3://models/manifest.json
AB_CONFIG: '{"weights":{"v1":70,"v2":30}}'
depends_on: [rabbitmq, minio]
ports: ["8000:8000"]
worker:
image: inference-service:latest
command: ["uv", "run", "python", "-m", "inference_service.batch_worker"]
environment:
S3_ENDPOINT: http://minio:9000
S3_ACCESS_KEY: minioadmin
S3_SECRET_KEY: minioadmin
S3_BUCKET: datasets
RABBIT_URL: amqp://guest:guest@rabbitmq:5672/
depends_on: [rabbitmq, minio]
rabbitmq:
image: rabbitmq:3.13-management
ports: ["15672:15672", "5672:5672"]
minio:
image: minio/minio:RELEASE.2024-10-13T13-34-11Z
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
ports: ["9000:9000", "9001:9001"]
volumes: ["minio_data:/data"]
volumes:
minio_data:
五、核心代码片段
- config.py
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional
class Settings(BaseSettings):
app_env: str = Field(default="prod", alias="APP_ENV")
s3_endpoint: str = Field(..., alias="S3_ENDPOINT")
s3_access_key: str = Field(..., alias="S3_ACCESS_KEY")
s3_secret_key: str = Field(..., alias="S3_SECRET_KEY")
s3_bucket: str = Field(..., alias="S3_BUCKET")
rabbit_url: str = Field(..., alias="RABBIT_URL")
model_manifest_s3_uri: str = Field(..., alias="MODEL_MANIFEST_S3_URI")
ab_config: Optional[str] = Field(default=None, alias="AB_CONFIG")
prometheus_scrape: bool = Field(default=True, alias="PROMETHEUS_SCRAPE")
allow_gpu: bool = Field(default=False, alias="ALLOW_GPU")
class Config:
extra = "ignore"
settings = Settings()
- schemas.py（API 契约）
from pydantic import BaseModel, Field, conlist, AwareDatetime
from typing import List, Optional, Literal, Dict
class HealthResp(BaseModel):
status: Literal["ok"] = "ok"
version: str
model_active: str
class PredictRequest(BaseModel):
request_id: Optional[str] = None
model_tag: Optional[str] = None
features: conlist(float, min_length=3, max_length=1024) # 举例: 至少3维
extra: Optional[Dict[str, str]] = None
class PredictResponse(BaseModel):
request_id: str
model_tag: str
score: float = Field(ge=0.0, le=1.0)
latency_ms: float
breakdown_ms: Dict[str, float]
class BatchRequest(BaseModel):
job_id: Optional[str] = None
input_s3_uri: str # s3://bucket/path/input.csv
output_s3_uri: str
model_tag: Optional[str] = None
callback_url: Optional[str] = None
class BatchStatus(BaseModel):
job_id: str
status: Literal["queued", "running", "done", "failed"]
message: Optional[str] = None
class DriftReport(BaseModel):
model_tag: str
psi: float
threshold: float
exceeded: bool
ts: AwareDatetime
- logging.py（结构化日志 + 脱敏）
import structlog
import re
PII_KEYS = {"name", "email", "phone", "id_number"}
def _mask_value(v: str) -> str:
if not isinstance(v, str):
return v
if "@" in v:
return re.sub(r"(^.).*(@.*$)", r"*\2", v)
return v[:1] + "*" * max(0, len(v) - 2) + v[-1:] if len(v) > 2 else "**"
def pii_filter(logger, method, event_dict):
for k in list(event_dict.keys()):
if k.lower() in PII_KEYS:
event_dict[k] = _mask_value(str(event_dict[k]))
return event_dict
def configure_logging():
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="iso"),
pii_filter,
structlog.processors.JSONRenderer(),
],
)
return structlog.get_logger()
- middleware.py（请求 ID、计时、错误处理）
import time
import uuid
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from pydantic import ValidationError
from .metrics import REQ_LATENCY, REQ_COUNT, ERR_COUNT
class RequestContextMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
req_id = request.headers.get("x-request-id", str(uuid.uuid4()))
start = time.perf_counter()
try:
response = await call_next(request)
status = response.status_code
except ValidationError as ve:
ERR_COUNT.labels(type="validation").inc()
return Response(
status_code=422, content=f"validation_error:{ve.errors()}", media_type="text/plain"
)
except Exception as e:
ERR_COUNT.labels(type="server").inc()
return Response(status_code=500, content="internal_error", media_type="text/plain")
finally:
dur = (time.perf_counter() - start) * 1000
REQ_LATENCY.observe(dur)
REQ_COUNT.inc()
response.headers["x-request-id"] = req_id
return response
- metrics.py（Prometheus 指标）
from prometheus_client import Counter, Histogram, Gauge
REQ_COUNT = Counter("http_requests_total", "Total HTTP requests")
ERR_COUNT = Counter("errors_total", "Errors", labelnames=("type",))
LAT_HIST = Histogram("predict_latency_ms", "Predict latency ms", buckets=(1,5,10,20,40,60,80,100,200,500))
STAGE_HIST = Histogram("stage_latency_ms", "Stage latency ms", labelnames=("stage",))
MODEL_QPS = Counter("model_infer_total", "Model infer count", labelnames=("model",))
MODEL_ERR = Counter("model_infer_errors_total", "Model infer errors", labelnames=("model", "type"))
DRIFT_PSI = Gauge("model_drift_psi", "PSI value", labelnames=("model",))
- storage.py（S3 适配）
import aioboto3
import io
from urllib.parse import urlparse
from .config import settings
def _parse_s3_uri(uri: str):
p = urlparse(uri)
return (p.netloc, p.path.lstrip("/"))
def s3_session():
return aioboto3.Session()
async def get_object_bytes(s3_uri: str) -> bytes:
bucket, key = _parse_s3_uri(s3_uri)
async with s3_session().client(
"s3",
endpoint_url=settings.s3_endpoint,
aws_secret_access_key=settings.s3_secret_key,
aws_access_key_id=settings.s3_access_key,
) as s3:
obj = await s3.get_object(Bucket=bucket, Key=key)
return await obj["Body"].read()
async def put_object_bytes(s3_uri: str, data: bytes):
bucket, key = _parse_s3_uri(s3_uri)
async with s3_session().client(
"s3",
endpoint_url=settings.s3_endpoint,
aws_secret_access_key=settings.s3_secret_key,
aws_access_key_id=settings.s3_access_key,
) as s3:
await s3.put_object(Bucket=bucket, Key=key, Body=data)
- crypto.py（模型打包签名校验）
import hashlib
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding
def sha256(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def verify_signature(data: bytes, signature: bytes, public_pem: bytes) -> bool:
public_key = serialization.load_pem_public_key(public_pem)
public_key.verify(
signature,
data,
padding.PKCS1v15(),
hashes.SHA256(),
)
return True
- routing.py（A/B/多臂分流）
import random
import xxhash
from typing import Dict, Optional
class WeightedRouter:
def __init__(self, weights: Dict[str, int]):
self.weights = weights
self.variants = list(weights.keys())
self.buckets = []
s = 0
for v in self.variants:
s += weights[v]
self.buckets.append((s, v))
self.total = s
def pick(self, seed: Optional[str] = None) -> str:
r = random.randint(1, self.total) if seed is None else (xxhash.xxh32(seed).intdigest() % self.total) + 1
for up, v in self.buckets:
if r <= up:
return v
return self.variants[-1]
- model_registry.py（按标签加载与热更新）
import asyncio
import json
from typing import Dict, Optional
from .storage import get_object_bytes
from .crypto import sha256, verify_signature
from .predictor import SklearnPredictor
from .metrics import MODEL_ERR
from .config import settings
class ModelRegistry:
def __init__(self):
self._models: Dict[str, SklearnPredictor] = {}
self._lock = asyncio.Lock()
self._manifest = {}
async def refresh_manifest(self):
data = await get_object_bytes(settings.model_manifest_s3_uri)
self._manifest = json.loads(data.decode("utf-8"))
async def load(self, tag: str):
async with self._lock:
if tag in self._models:
return self._models[tag]
meta = self._manifest["models"].get(tag)
if not meta:
raise KeyError(f"model tag not found: {tag}")
blob = await get_object_bytes(meta["artifact"])
if "sha256" in meta and sha256(blob) != meta["sha256"]:
raise ValueError("checksum mismatch")
if "signature" in meta and "public_key" in self._manifest:
sig = await get_object_bytes(meta["signature"])
pub = await get_object_bytes(self._manifest["public_key"])
verify_signature(blob, sig, pub)
predictor = SklearnPredictor.from_bytes(blob, device="cpu")
self._models[tag] = predictor
return predictor
def get(self, tag: str) -> Optional[SklearnPredictor]:
return self._models.get(tag)
async def hot_swap(self, tag: str):
await self.load(tag) # 先加载到内存
# 调整活跃权重由 routing 控制；无需停机
registry = ModelRegistry()
manifest.json 示例（S3 上）：
{
"public_key": "s3://models/keys/public.pem",
"models": {
"v1": {"artifact": "s3://models/v1/model.pkl", "sha256": "abc123...", "signature": "s3://models/v1/model.sig"},
"v2": {"artifact": "s3://models/v2/model.pkl", "sha256": "def456...", "signature": "s3://models/v2/model.sig"}
}
}
- preprocessing.py（带 doctest）
import numpy as np
from typing import List
def normalize_minmax(x: List[float]):
"""
>>> normalize_minmax([0, 5, 10])
array([0. , 0.5, 1. ])
"""
arr = np.asarray(x, dtype=np.float32)
mn, mx = arr.min(), arr.max()
if mx == mn:
return np.zeros_like(arr)
return (arr - mn) / (mx - mn)
- predictor.py（CPU/GPU 可选）
import io, pickle, time
import numpy as np
from typing import Dict, Any, Optional
class SklearnPredictor:
def __init__(self, model):
self.model = model
@classmethod
def from_bytes(cls, blob: bytes, device: str = "cpu"):
model = pickle.loads(blob)
return cls(model)
def predict_proba(self, feats: np.ndarray) -> float:
# 这里假设二分类，取正类概率
proba = self.model.predict_proba(feats.reshape(1, -1))[0, 1]
return float(proba)
# GPU 可选（示意）
try:
import torch
class TorchPredictor:
def __init__(self, model, device="cpu"):
self.model = model.to(device)
self.device = device
@classmethod
def from_bytes(cls, blob: bytes, device="cpu"):
buffer = io.BytesIO(blob)
model = torch.jit.load(buffer, map_location=device)
model.eval()
return cls(model, device)
@torch.inference_mode()
def predict(self, feats: np.ndarray) -> float:
x = torch.from_numpy(feats).float().to(self.device)
y = self.model(x)
return float(torch.sigmoid(y).item())
except Exception:
TorchPredictor = None
- postprocessing.py
def clip_score(s: float, lo: float = 0.0, hi: float = 1.0) -> float:
return max(lo, min(hi, s))
- drift.py（简单 PSI）
import numpy as np
from .metrics import DRIFT_PSI
def psi(expected: np.ndarray, actual: np.ndarray, bins=10):
quantiles = np.linspace(0, 1, bins+1)
e_cuts = np.quantile(expected, quantiles)
a_cuts = e_cuts # 用 expected 分箱
e_hist, _ = np.histogram(expected, bins=e_cuts)
a_hist, _ = np.histogram(actual, bins=a_cuts)
e_rat = np.clip(e_hist / max(1, e_hist.sum()), 1e-6, 1)
a_rat = np.clip(a_hist / max(1, a_hist.sum()), 1e-6, 1)
val = float(np.sum((a_rat - e_rat) * np.log(a_rat / e_rat)))
return val
def report_and_export(model_tag: str, expected: np.ndarray, actual: np.ndarray, threshold=0.2) -> dict:
v = psi(expected, actual)
DRIFT_PSI.labels(model=model_tag).set(v)
return {"model_tag": model_tag, "psi": v, "threshold": threshold, "exceeded": v > threshold}
- api.py（HTTP 端点）
import time
import numpy as np
from fastapi import APIRouter, Depends
from fastapi.responses import ORJSONResponse, PlainTextResponse
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from .schemas import HealthResp, PredictRequest, PredictResponse, BatchRequest, BatchStatus
from .model_registry import registry
from .routing import WeightedRouter
from .preprocessing import normalize_minmax
from .postprocessing import clip_score
from .metrics import LAT_HIST, STAGE_HIST, MODEL_QPS, MODEL_ERR
from .config import settings
from .batch_worker import enqueue_job
router = APIRouter()
router_ab = WeightedRouter(weights={"v1":70, "v2":30})
@router.get("/health", response_model=HealthResp)
async def health():
return HealthResp(version="0.1.0", model_active="dynamic")
@router.post("/predict", response_model=PredictResponse)
async def predict(req: PredictRequest):
t0 = time.perf_counter()
tag = req.model_tag or router_ab.pick(seed=req.request_id or str(req.features[0]))
t_pre = time.perf_counter()
feats = normalize_minmax(req.features)
t_infer_s = time.perf_counter()
model = registry.get(tag)
if not model:
model = await registry.load(tag)
try:
score = model.predict_proba(np.asarray(feats))
except Exception as e:
MODEL_ERR.labels(model=tag, type="infer").inc()
raise
score = clip_score(score)
t1 = time.perf_counter()
MODEL_QPS.labels(model=tag).inc()
total_ms = (t1 - t0)*1000
LAT_HIST.observe(total_ms)
STAGE_HIST.labels("preprocess").observe((t_infer_s - t_pre)*1000)
STAGE_HIST.labels("inference").observe((t1 - t_infer_s)*1000)
return ORJSONResponse(
PredictResponse(
request_id=req.request_id or "",
model_tag=tag, score=score,
latency_ms=total_ms,
breakdown_ms={"preprocess": (t_infer_s-t_pre)*1000, "inference": (t1-t_infer_s)*1000}
).model_dump()
)
@router.post("/batch", response_model=BatchStatus)
async def batch(req: BatchRequest):
job_id = await enqueue_job(req)
return BatchStatus(job_id=job_id, status="queued")
@router.get("/metrics")
async def metrics():
data = generate_latest()
return PlainTextResponse(data, media_type=CONTENT_TYPE_LATEST)
- batch_worker.py（RabbitMQ 批量作业）
import asyncio, csv, io, json, uuid
import aio_pika
import numpy as np
from .config import settings
from .model_registry import registry
from .preprocessing import normalize_minmax
from .postprocessing import clip_score
from .storage import get_object_bytes, put_object_bytes
QUEUE = "batch_jobs"
async def enqueue_job(req) -> str:
job_id = req.job_id or str(uuid.uuid4())
payload = req.model_dump()
payload["job_id"] = job_id
conn = await aio_pika.connect_robust(settings.rabbit_url)
async with conn:
ch = await conn.channel()
q = await ch.declare_queue(QUEUE, durable=True)
await ch.default_exchange.publish(
aio_pika.Message(body=json.dumps(payload).encode(), delivery_mode=aio_pika.DeliveryMode.PERSISTENT),
routing_key=QUEUE,
)
return job_id
async def worker():
conn = await aio_pika.connect_robust(settings.rabbit_url)
async with conn:
ch = await conn.channel()
await ch.set_qos(prefetch_count=8)
q = await ch.declare_queue(QUEUE, durable=True)
async with q.iterator() as qiter:
async for msg in qiter:
async with msg.process():
job = json.loads(msg.body.decode())
await process_job(job)
async def process_job(job: dict):
tag = job.get("model_tag") or "v1"
model = registry.get(tag) or await registry.load(tag)
raw = await get_object_bytes(job["input_s3_uri"])
# 假设 CSV 一行一个样本，逗号分隔
reader = csv.reader(io.StringIO(raw.decode()))
scores = []
for row in reader:
feats = [float(x) for x in row]
feats = normalize_minmax(feats)
s = clip_score(model.predict_proba(np.asarray(feats)))
scores.append(s)
out = "\n".join(map(str, scores)).encode()
await put_object_bytes(job["output_s3_uri"], out)
# 可选：回调通知
# 若有 job["callback_url"]，在内网触发 HTTP 回调
if __name__ == "__main__":
asyncio.run(worker())
- app.py（组装应用与热加载）
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .api import router
from .middleware import RequestContextMiddleware
from .logging import configure_logging
from .model_registry import registry
from .config import settings
def create_app() -> FastAPI:
log = configure_logging()
app = FastAPI(title="Inference Service", version="0.1.0")
app.add_middleware(RequestContextMiddleware)
app.add_middleware(
CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
)
app.include_router(router)
@app.on_event("startup")
async def _startup():
await registry.refresh_manifest()
# 可选预热: 加载主流量模型
# await registry.load("v1")
return app
六、API 契约文档（概要）
- GET /health
- 200: {status:"ok", version:"0.1.0", model_active:"dynamic"}
- POST /predict
- Request: {request_id?: string, model_tag?: string, features: number[], extra?: object}
- Response: {request_id: string, model_tag: string, score: number[0..1], latency_ms: number, breakdown_ms: {preprocess:number, inference:number}}
- POST /batch
- Request: {job_id?: string, input_s3_uri: "s3://...", output_s3_uri: "s3://...", model_tag?: string, callback_url?: string}
- Response: {job_id: string, status: "queued"}
- GET /metrics
提示：FastAPI 自动生成 OpenAPI 文档，可在 /docs 查看契约细节。
七、灰度与回滚方案
- 基于 WeightedRouter 的权重文件（AB_CONFIG）驱动分流。先在 manifest.json 上线新版本 v2，后台执行 registry.hot_swap("v2") 预加载，确认健康后逐步调整 AB_CONFIG 权重 5% → 20% → 50% → 100%。
- 回滚即恢复权重至 v1:100%。无需重启；模型在内存中，切流即时生效。
- 支持多臂（A/B/C）策略；也可按规则（用户 ID 哈希、地区）做 deterministic 分流。
八、批量离线作业入口（Airflow 对接）
- 使用 BashOperator 或 PythonOperator 调用 worker 或直接发布消息到 RabbitMQ：
- 方式一：Airflow 推入消息，worker 消费
- 方式二：Airflow 直接运行 Python，复用 process_job()
- 结果回传：将结果与状态写回 S3（如 s3://datasets/jobs/{job_id}/result.csv 与 status.json），并可选回调内网 HTTP。
九、错误处理与数据校验
- Pydantic v2 模式约束（features 长度与类型、score 范围）
- 全局异常转化为 4xx/5xx，Prometheus 错误计数打点
- 日志脱敏中间件避免 PII 外泄
- 输入尺寸限制：建议在反向代理或 Uvicorn/gunicorn 配置 content-length 限制（如 1MB）
十、性能优化建议（目标：CPU p95 ≤ 60ms，200 RPS 单实例）
- 进程/线程
- gunicorn + uvicorn workers：workers ≈ CPU 核数，worker-class uvicorn.workers.UvicornWorker
- 每 worker 可开 1-2 线程；CPU 绑定库（NumPy/BLAS）设 OMP_NUM_THREADS=1
- 序列化
- 使用 ORJSONResponse；禁用不必要的字段计算
- 预热与缓存
- 启动时预加载主模型；常量、编码器预计算；向量化预处理
- 避免阻塞
- S3/RabbitMQ 走异步客户端；CPU 推理必要时 ThreadPoolExecutor
- 内存与 GC
- 减少临时对象；使用 NumPy float32；避免频繁创建大数组
- 内核优化
- uvloop + httptools（uvicorn[standard] 已内置）
- 冷启动 < 2s
- 镜像瘦身、避免 pip 源联网；uv.lock 确保快速安装；模型延迟加载或预热小模型
- 压测目标达成方法见“压力测试指南”
十一、监控与追踪
- Prometheus 指标
- QPS、总延迟直方图、阶段延迟（预处理/推理）、错误率、模型级计数/错误
- 数据漂移 PSI 指标（DRIFT_PSI）
- 追踪
- 可接入 OpenTelemetry（otlp exporter）对接内部可观测平台；把 request_id 写入日志与响应头
- 关键日志
- request_id、model_tag、latency、breakdown、错误类型、S3 错误码
十二、模型打包与签名
- scripts/package_model.py：将 sklearn 模型序列化为 pkl，上传 S3
- scripts/sign_model.py：私钥对 artifact 做 SHA256 签名，上传 signature；服务端按 public.pem 验签
- manifest.json 记录 sha256 与 signature 路径，加载时强制校验
十三、数据漂移与阈值告警
- drift.report_and_export 计算 PSI 并导出到 Prometheus
- 提供管理端点（可扩展）手动触发或定期在 Batch Worker/定时任务中更新
- 阈值超过在日志中打 WARN，并由 Grafana 告警规则触发通知
十四、单元测试与 doctest
tests/test_preprocess.py
from inference_service.preprocessing import normalize_minmax
import numpy as np
def test_normalize():
out = normalize_minmax([0, 5, 10])
assert np.isclose(out[1], 0.5)
tests/test_postprocess.py
from inference_service.postprocessing import clip_score
def test_clip():
assert clip_score(1.2) == 1.0
assert clip_score(-0.1) == 0.0
tests/test_api_smoke.py
import pytest, httpx, asyncio
from inference_service.app import create_app
from asgiref.wsgi import WsgiToAsgi
@pytest.mark.asyncio
async def test_predict_smoke():
app = create_app()
async with httpx.AsyncClient(app=app, base_url="http://test") as c:
resp = await c.post("/predict", json={"features":[1,2,3]})
assert resp.status_code == 200
十五、本地开发脚手架
Makefile
.PHONY: setup lint test run docker-build compose-up smoke
setup:
uv sync --all-extras --frozen
pre-commit install
lint:
ruff check .
mypy src
test:
pytest -q
python -m doctest -v src/inference_service/preprocessing.py
run:
uv run uvicorn inference_service.app:create_app --factory --reload
docker-build:
docker build -t inference-service:latest -f docker/Dockerfile .
compose-up:
docker compose up -d --build
smoke:
bash scripts/smoke_test.sh
.pre-commit-config.yaml
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.7.3
hooks: [{id: ruff}]
- repo: https://github.com/psf/black
rev: 24.10.0
hooks: [{id: black}]
- repo: https://github.com/PyCQA/bandit
rev: 1.7.9
hooks: [{id: bandit}]
scripts/smoke_test.sh
#!/usr/bin/env bash
set -e
curl -sf http://localhost:8000/health | jq .
curl -sf -X POST http://localhost:8000/predict -H 'content-type: application/json' \
-d '{"features":[1,2,3],"request_id":"smoke-1"}' | jq .
十六、压力测试指导
- 工具建议：hey、vegeta、wrk、locust（内网优先）
- 示例（hey）：
- 目标评估：
- RPS ≥ 200，p95 ≤ 60ms；观察 prometheus 指标 predict_latency_ms
- Tuning：
- CPU 核数 n：gunicorn -w n -k uvicorn.workers.UvicornWorker
- 限制 BLAS 线程 OMP_NUM_THREADS=1
- 批量压测时扩容副本、反向代理连接复用 keepalive=on
- 监控面板：Grafana 构建延迟/QPS/错误率/模型版本维度面板
十七、热更新无中断
- 通过 /admin（可扩展）触发 registry.refresh_manifest() 与 registry.hot_swap(tag)
- 路由权重即时更新（来自配置中心或环境变量变更+SIGHUP）
- 双缓冲：先加载新模型到内存，再调整路由；旧模型保留一段时间以便回滚
十八、安全与合规
- 严禁将业务数据上传至外网：未集成任何外部上报；S3/RabbitMQ 等均指向内网
- 日志脱敏；错误栈默认不回传客户端
- 输入大小限制、请求超时（由反向代理或 gunicorn 配置）
- 模型验签 + 校验和，防篡改
十九、GPU 分支（可选）
- extras "gpu" 安装 torch；在读取 manifest 时根据标签或配置选择 TorchPredictor 并设 device="cuda" if torch.cuda.is_available()
- Docker 需带 nvidia runtime 与 CUDA 基础镜像（不强依赖，默认 CPU）
二十、冷启动与复现
- 使用 uv.lock/conda environment.yml 固定版本
- 模型按需加载或轻量预热，确保启动时间 < 2s
- 镜像多阶段、非 root，保证可移植与安全
这套模板提供了端到端的落地骨架。你可以直接 clone（或复制上述文件结构与代码片段）后，接入你们的模型与特征逻辑，按需在 manifest.json 中管理版本与签名，在 docker-compose 环境完成一键起服与冒烟/压测，随后对接内网 Prometheus/Grafana 与 Airflow 即可。若你提供具体模型与特征定义，我可以进一步补全示例模型的打包脚本和 manifest 生成器，以及生产级 gunicorn/反向代理配置。
