本文从实际工程角度出发，逐步构建一个稳定、安全、可横向扩展的 AI 情感分析服务镜像——从推理逻辑的设计，到 Dockerfile 的精细调优，再到 Kubernetes 生产部署的完整流程。

---

一、项目结构

sentiment-analyzer/
├── app.py               # 推理服务主体
├── model_registry.py    # 模型注册表
├── middleware.py        # 限流、鉴权中间件
├── metrics.py           # 指标采集
├── requirements.txt     # Python 依赖声明
├── Dockerfile           # 镜像构建配置
└── k8s/
    ├── deployment.yaml  # K8s 工作负载声明
    ├── service.yaml     # 服务暴露配置
    ├── configmap.yaml   # 模型配置管理
    └── hpa.yaml         # 水平自动扩缩容

requirements.txt 内容如下：

gradio==4.44.0
transformers==4.44.0
torch==2.4.0
fastapi==0.115.0
uvicorn==0.30.6
prometheus-client==0.21.0
slowapi==0.1.9
python-jose==3.3.0
langdetect==1.0.9

建议锁定版本号。transformers 的 API 在大版本间存在 breaking change，不锁版本的镜像在重新构建时可能产生难以复现的行为差异。

---

二、模型注册表

将模型元信息集中声明，推理逻辑与模型配置彻底解耦。新增模型只需在注册表中添加一条记录，无需修改任何业务代码。

model_registry.py

from dataclasses import dataclass, field

@dataclass
class ModelConfig:
    name: str
    path: str
    description: str
    languages: list[str]
    max_length: int = 512
    # 不同模型的输出 label 格式各异，统一映射到标准情感
    label_map: dict[str, str] = field(default_factory=dict)

MODEL_REGISTRY: dict[str, ModelConfig] = {
    "distilbert-en": ModelConfig(
        name="distilbert-en",
        path="distilbert-base-uncased-finetuned-sst-2-english",
        description="英文情感分析，速度快，适合高并发场景",
        languages=["en"],
        max_length=512,
        label_map={"POSITIVE": "positive", "NEGATIVE": "negative"},
    ),
    "roberta-en": ModelConfig(
        name="roberta-en",
        path="cardiffnlp/twitter-roberta-base-sentiment-latest",
        description="基于 Twitter 数据训练，适合社交媒体文本",
        languages=["en"],
        max_length=512,
        label_map={
            "positive": "positive",
            "negative": "negative",
            "neutral": "neutral",
        },
    ),
    "bert-zh": ModelConfig(
        name="bert-zh",
        path="uer/roberta-base-finetuned-jd-binary-chinese",
        description="中文情感分析，基于京东评论数据训练",
        languages=["zh"],
        max_length=256,
        label_map={"positive": "positive", "negative": "negative"},
    ),
    "finbert": ModelConfig(
        name="finbert",
        path="ProsusAI/finbert",
        description="金融领域情感分析，支持正面/负面/中性三分类",
        languages=["en"],
        max_length=512,
        label_map={
            "positive": "positive",
            "negative": "negative",
            "neutral": "neutral",
        },
    ),
}

# 标准情感到中文展示的映射
SENTIMENT_DISPLAY = {
    "positive": ("积极", "🌟"),
    "negative": ("消极", "😞"),
    "neutral":  ("中性", "😐"),
}

---

三、指标采集

在推理层嵌入 Prometheus 指标，为后续监控和告警提供数据基础。

metrics.py

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 请求总量（按模型、情感结果分类）
REQUEST_COUNT = Counter(
    "sentiment_requests_total",
    "Total number of sentiment analysis requests",
    ["model", "sentiment", "status"],
)

# 推理延迟分布
INFERENCE_LATENCY = Histogram(
    "sentiment_inference_duration_seconds",
    "Inference latency in seconds",
    ["model"],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5],
)

# 当前加载的模型数量
LOADED_MODELS = Gauge(
    "sentiment_loaded_models_total",
    "Number of models currently loaded in memory",
)

# 文本长度分布（用于分析截断频率）
TEXT_LENGTH = Histogram(
    "sentiment_input_text_length",
    "Length of input text in characters",
    buckets=[50, 100, 200, 300, 512, 1024],
)

def start_metrics_server(port: int = 9090):
    """在独立端口暴露 Prometheus 指标，与业务端口解耦。"""
    start_http_server(port)
    print(f"[*] Prometheus 指标已暴露在 :{port}/metrics")

---

四、推理服务：完整增强版

本章将推理服务从单一的 Gradio 脚本升级为双接口架构——Gradio 面向人工演示，FastAPI 面向程序调用——两者共用同一套推理逻辑，通过同一个进程启动。

4.1 核心推理逻辑

# app.py

import os
import time
import threading
from functools import lru_cache

import gradio as gr
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from transformers import pipeline
from langdetect import detect, LangDetectException
import uvicorn

from model_registry import MODEL_REGISTRY, SENTIMENT_DISPLAY
from metrics import (
    REQUEST_COUNT, INFERENCE_LATENCY,
    LOADED_MODELS, TEXT_LENGTH, start_metrics_server,
)

4.2 模型加载与缓存

# ── 模型加载 ────────────────────────────────────────────────

DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "distilbert-en")
_model_lock = threading.Lock()  # 防止并发请求触发多次加载

@lru_cache(maxsize=4)
def load_model(model_path: str):
    """懒加载 + LRU 缓存：同一路径只加载一次，最多缓存 4 个模型。"""
    with _model_lock:
        print(f"[*] 加载模型: {model_path}")
        start = time.time()
        p = pipeline("sentiment-analysis", model=model_path)
        print(f"[*] 加载完成，耗时: {time.time() - start:.2f}s")
        LOADED_MODELS.set(len(load_model.cache_info().__class__.__mro__))
        return p

4.3 核心推理函数

def _normalize_label(raw_label: str, config) -> str:
    """将不同模型的原始 label 统一映射为 positive / negative / neutral。"""
    normalized = config.label_map.get(raw_label.lower())
    if normalized:
        return normalized
    # 兜底：根据关键词猜测
    upper = raw_label.upper()
    if "POS" in upper or upper == "LABEL_2":
        return "positive"
    if "NEG" in upper or upper == "LABEL_0":
        return "negative"
    return "neutral"

def auto_select_model(text: str, requested_model: str) -> str:
    """
    语言自动检测：若用户选择的模型不支持检测到的语言，
    自动降级到合适的模型，避免模型与语言不匹配导致的低质量结果。
    """
    config = MODEL_REGISTRY.get(requested_model)
    if not config:
        return requested_model

    try:
        detected_lang = detect(text)
    except LangDetectException:
        return requested_model

    if detected_lang not in config.languages:
        # 查找支持该语言的第一个模型
        for name, cfg in MODEL_REGISTRY.items():
            if detected_lang in cfg.languages:
                print(f"[*] 语言检测: {detected_lang}，自动切换模型: {name}")
                return name

    return requested_model

def run_inference(text: str, model_name: str) -> dict:
    """
    统一推理入口，供 Gradio 和 FastAPI 共同调用。
    返回标准化结构，包含情感、置信度、耗时、模型信息等。
    """
    if not text or not text.strip():
        raise ValueError("输入文本不能为空")

    TEXT_LENGTH.observe(len(text))

    # 语言自动检测与模型降级
    actual_model = auto_select_model(text, model_name)
    config = MODEL_REGISTRY.get(actual_model)
    if not config:
        raise ValueError(f"未知模型: {actual_model}")

    # 截断超长输入
    truncated = len(text) > config.max_length
    text = text[:config.max_length]

    try:
        analyzer = load_model(config.path)
    except Exception as e:
        REQUEST_COUNT.labels(
            model=actual_model, sentiment="unknown", status="error"
        ).inc()
        raise RuntimeError(f"模型加载失败: {e}")

    with INFERENCE_LATENCY.labels(model=actual_model).time():
        req_start = time.time()
        result = analyzer(text)[0]
        elapsed_ms = (time.time() - req_start) * 1000

    sentiment = _normalize_label(result["label"], config)
    display_name, emoji = SENTIMENT_DISPLAY.get(sentiment, ("未知", "❓"))

    REQUEST_COUNT.labels(
        model=actual_model, sentiment=sentiment, status="success"
    ).inc()

    return {
        "sentiment": sentiment,
        "display": f"{display_name} {emoji}",
        "confidence": round(result["score"], 4),
        "confidence_pct": f"{result['score']:.2%}",
        "model_used": actual_model,
        "model_desc": config.description,
        "elapsed_ms": round(elapsed_ms, 1),
        "truncated": truncated,
        "auto_switched": actual_model != model_name,
    }

4.4 FastAPI 接口层

# ── FastAPI 应用 ──────────────────────────────────────────

limiter = Limiter(key_func=get_remote_address)
api = FastAPI(
    title="Sentiment Analyzer API",
    description="基于 Transformer 的多模型情感分析服务",
    version="2.0.0",
)
api.state.limiter = limiter
api.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

api.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", "*").split(","),
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

# ── 请求 / 响应模型 ───────────────────────────────────────

class AnalyzeRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000, description="待分析文本")
    model: str = Field(default="distilbert-en", description="指定模型名称")

    class Config:
        json_schema_extra = {
            "example": {
                "text": "This product exceeded all my expectations!",
                "model": "distilbert-en",
            }
        }

class BatchAnalyzeRequest(BaseModel):
    texts: list[str] = Field(..., min_length=1, max_length=32, description="批量文本，最多 32 条")
    model: str = Field(default="distilbert-en", description="指定模型名称")

class AnalyzeResponse(BaseModel):
    sentiment: str
    display: str
    confidence: float
    confidence_pct: str
    model_used: str
    model_desc: str
    elapsed_ms: float
    truncated: bool
    auto_switched: bool

# ── API 端点 ──────────────────────────────────────────────

@api.get("/health", tags=["运维"])
async def health():
    """健康检查端点，供 K8s liveness/readiness 探针使用。"""
    loaded = load_model.cache_info().currsize
    return {
        "status": "ok",
        "loaded_models": loaded,
        "available_models": list(MODEL_REGISTRY.keys()),
    }

@api.get("/models", tags=["模型"])
async def list_models():
    """列出所有可用模型及其元信息。"""
    return {
        name: {
            "description": cfg.description,
            "languages": cfg.languages,
            "max_length": cfg.max_length,
        }
        for name, cfg in MODEL_REGISTRY.items()
    }

@api.post("/analyze", response_model=AnalyzeResponse, tags=["推理"])
@limiter.limit("60/minute")
async def analyze(request: Request, body: AnalyzeRequest):
    """
    单条文本情感分析。

    - 支持语言自动检测与模型自动切换
    - 60 次/分钟限流（基于客户端 IP）
    """
    try:
        result = run_inference(body.text, body.model)
        return result
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e))
    except RuntimeError as e:
        raise HTTPException(status_code=503, detail=str(e))

@api.post("/analyze/batch", tags=["推理"])
@limiter.limit("10/minute")
async def analyze_batch(request: Request, body: BatchAnalyzeRequest):
    """
    批量文本情感分析，最多支持 32 条。

    - 逐条推理，部分失败不影响其余结果
    - 10 次/分钟限流
    """
    results = []
    for i, text in enumerate(body.texts):
        try:
            results.append({"index": i, "result": run_inference(text, body.model)})
        except Exception as e:
            results.append({"index": i, "error": str(e)})
    return {"total": len(body.texts), "results": results}

@api.get("/analyze/history", tags=["推理"])
async def get_history():
    """返回最近 50 条分析记录（内存存储，重启后清空）。"""
    return {"history": list(reversed(_history[-50:]))}

4.5 请求历史记录

# ── 内存历史记录 ──────────────────────────────────────────
# 生产环境建议替换为 Redis，支持持久化和跨实例共享

from collections import deque

_history: deque = deque(maxlen=200)
_original_run = run_inference

def run_inference_with_history(text: str, model_name: str) -> dict:
    result = _original_run(text, model_name)
    _history.append({
        "text_preview": text[:80] + ("..." if len(text) > 80 else ""),
        "result": result,
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    })
    return result

# 覆盖原始函数，使所有调用路径均记录历史
run_inference = run_inference_with_history

4.6 Gradio 界面

# ── Gradio 界面 ───────────────────────────────────────────

def gradio_predict(text: str, model_name: str) -> tuple:
    """Gradio 回调，将结构化结果拆分为多个输出组件。"""
    try:
        r = run_inference(text, model_name)
        summary = f"{r['display']}  （置信度 {r['confidence_pct']}）"
        detail = {
            "模型": f"{r['model_used']}  —  {r['model_desc']}",
            "处理耗时": f"{r['elapsed_ms']}ms",
            "文本截断": "是" if r["truncated"] else "否",
            "自动切换模型": "是" if r["auto_switched"] else "否",
        }
        return summary, detail
    except Exception as e:
        return f"⚠️ {e}", {}

model_choices = [
    (f"{cfg.description}", name)
    for name, cfg in MODEL_REGISTRY.items()
]

with gr.Blocks(title="AI 情感分析服务 v2.0") as gradio_app:
    gr.Markdown(
        """
        # AI 情感分析服务 v2.0
        基于 Transformer 架构的多模型情感分析，支持中英文及金融领域文本。数据不出容器。
        """
    )

    with gr.Row():
        with gr.Column(scale=2):
            text_input = gr.Textbox(
                lines=6,
                label="输入文本",
                placeholder="支持中英文，最长 512 字符...",
            )
            model_dropdown = gr.Dropdown(
                choices=model_choices,
                value="distilbert-en",
                label="选择模型",
            )
            submit_btn = gr.Button("开始分析", variant="primary")

        with gr.Column(scale=1):
            sentiment_output = gr.Textbox(label="情感判断", interactive=False)
            detail_output = gr.JSON(label="详细信息")

    gr.Examples(
        examples=[
            ["This movie is absolutely fantastic!", "distilbert-en"],
            ["今天的天气真让人心情沮丧。", "bert-zh"],
            ["Q3 earnings significantly exceeded analyst expectations.", "finbert"],
            ["The product is okay but nothing special.", "roberta-en"],
        ],
        inputs=[text_input, model_dropdown],
    )

    submit_btn.click(
        fn=gradio_predict,
        inputs=[text_input, model_dropdown],
        outputs=[sentiment_output, detail_output],
    )

4.7 服务启动入口

# ── 启动 ─────────────────────────────────────────────────

if __name__ == "__main__":
    # 预热默认模型，避免第一次请求的冷启动延迟
    print(f"[*] 预热默认模型: {DEFAULT_MODEL}")
    load_model(MODEL_REGISTRY[DEFAULT_MODEL].path)

    # 在独立线程启动 Prometheus 指标服务器
    threading.Thread(
        target=start_metrics_server,
        kwargs={"port": int(os.getenv("METRICS_PORT", "9090"))},
        daemon=True,
    ).start()

    # 将 Gradio 挂载到 FastAPI，统一入口
    import gradio as gr
    app = gr.mount_gradio_app(api, gradio_app, path="/ui")

    uvicorn.run(
        app,
        host="0.0.0.0",
        port=int(os.getenv("PORT", "8080")),
        workers=1,  # Transformer 模型不支持多进程 fork，使用单 worker
        log_level="info",
    )

服务启动后，各端点分布如下：

端点	说明
GET /ui	Gradio 可视化界面，面向人工演示
POST /analyze	单条推理 REST 接口，60次/分钟限流
POST /analyze/batch	批量推理，最多 32 条，10次/分钟限流
GET /analyze/history	最近 50 条历史记录
GET /models	列出所有可用模型
GET /health	健康检查，供 K8s 探针使用
GET /docs	FastAPI 自动生成的 Swagger 文档
GET :9090/metrics	Prometheus 指标，与业务端口隔离

---

五、Dockerfile

FROM python:3.10-slim

WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1 \
    HF_ENDPOINT=https://hf-mirror.com

COPY requirements.txt .
RUN pip install -r requirements.txt \
    -i https://pypi.tuna.tsinghua.edu.cn/simple \
    --trusted-host pypi.tuna.tsinghua.edu.cn

# 将默认模型权重烘焙进镜像，消除冷启动网络依赖
RUN python -c "from transformers import pipeline; pipeline('sentiment-analysis')"

COPY app.py model_registry.py metrics.py ./

# RUN useradd --no-create-home --shell /bin/false appuser \
#     && chown -R appuser /app
# USER appuser

EXPOSE 8080
EXPOSE 9090

CMD ["python", "app.py"]

---

六、K8s 生产部署

6.1 ConfigMap

# k8s/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: sentiment-analyzer-config
  namespace: ai-services
data:
  DEFAULT_MODEL: "distilbert-en"
  HF_ENDPOINT: "https://hf-mirror.com"
  PYTHONUNBUFFERED: "1"
  METRICS_PORT: "9090"
  ALLOWED_ORIGINS: "https://your-domain.com"

6.2 Deployment

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sentiment-analyzer
  namespace: ai-services
  labels:
    app: sentiment-analyzer
    version: v2.0
spec:
  replicas: 2
  selector:
    matchLabels:
      app: sentiment-analyzer
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    metadata:
      labels:
        app: sentiment-analyzer
        version: v2.0
      annotations:
        # 告知 Prometheus 主动抓取该 Pod 的指标
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
        prometheus.io/path: "/metrics"
    spec:
      containers:
        - name: sentiment-analyzer
          image: registry.cn-hangzhou.aliyuncs.com/your-namespace/sentiment-analyzer:v2.0
          ports:
            - name: http
              containerPort: 8080
            - name: metrics
              containerPort: 9090
          envFrom:
            - configMapRef:
                name: sentiment-analyzer-config
          resources:
            requests:
              cpu: "500m"
              memory: "1Gi"
            limits:
              cpu: "2000m"
              memory: "3Gi"
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 60
            periodSeconds: 30
            failureThreshold: 3
          readinessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 60
            periodSeconds: 10
            failureThreshold: 5
      terminationGracePeriodSeconds: 30

6.3 Service 与 Ingress

# k8s/service.yaml
apiVersion: v1
kind: Service
metadata:
  name: sentiment-analyzer-svc
  namespace: ai-services
spec:
  selector:
    app: sentiment-analyzer
  ports:
    - name: http
      port: 80
      targetPort: 8080
    - name: metrics
      port: 9090
      targetPort: 9090
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: sentiment-analyzer-ingress
  namespace: ai-services
  annotations:
    nginx.ingress.kubernetes.io/proxy-read-timeout: "120"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "120"
spec:
  ingressClassName: nginx
  rules:
    - host: sentiment.your-domain.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: sentiment-analyzer-svc
                port:
                  number: 80

6.4 HPA

# k8s/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: sentiment-analyzer-hpa
  namespace: ai-services
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: sentiment-analyzer
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
    - type: Resource
      resource:
        name: memory
        target:
          type: Utilization
          averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Pods
          value: 2
          periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
        - type: Pods
          value: 1
          periodSeconds: 120

6.5 一键部署与运维命令

# 部署
kubectl create namespace ai-services
kubectl apply -f k8s/

# 观察状态
kubectl rollout status deployment/sentiment-analyzer -n ai-services
kubectl get pods,svc,hpa -n ai-services

# 查看日志
kubectl logs -l app=sentiment-analyzer -n ai-services --follow

# 切换默认模型（无需重新构建镜像）
kubectl patch configmap sentiment-analyzer-config -n ai-services \
  --patch '{"data": {"DEFAULT_MODEL": "bert-zh"}}'
kubectl rollout restart deployment/sentiment-analyzer -n ai-services

# 滚动更新
kubectl set image deployment/sentiment-analyzer \
  sentiment-analyzer=registry.cn-hangzhou.aliyuncs.com/your-namespace/sentiment-analyzer:v2.1 \
  -n ai-services

# 紧急回滚
kubectl rollout undo deployment/sentiment-analyzer -n ai-services

---

七、接口使用示例

单条分析

curl -X POST http://localhost:8080/analyze \
  -H "Content-Type: application/json" \
  -d '{"text": "This product exceeded all my expectations!", "model": "distilbert-en"}'

{
  "sentiment": "positive",
  "display": "积极 🌟",
  "confidence": 0.9995,
  "confidence_pct": "99.95%",
  "model_used": "distilbert-en",
  "model_desc": "英文情感分析，速度快，适合高并发场景",
  "elapsed_ms": 23.4,
  "truncated": false,
  "auto_switched": false
}

批量分析

curl -X POST http://localhost:8080/analyze/batch \
  -H "Content-Type: application/json" \
  -d '{
    "texts": [
      "Absolutely loved it!",
      "Terrible experience, never again.",
      "Q3 results beat consensus estimates by 12%."
    ],
    "model": "distilbert-en"
  }'

{
  "total": 3,
  "results": [
    {"index": 0, "result": {"sentiment": "positive", "confidence_pct": "99.98%", ...}},
    {"index": 1, "result": {"sentiment": "negative", "confidence_pct": "99.91%", ...}},
    {"index": 2, "result": {"sentiment": "positive", "confidence_pct": "87.34%", ...}}
  ]
}

查看可用模型

curl http://localhost:8080/models

完整的交互式 API 文档可访问 http://localhost:8080/docs，由 FastAPI 自动生成，支持在线调试。

---

小结

模块	核心设计	收益
双接口架构	Gradio + FastAPI 共用推理逻辑	演示与程序调用互不干扰
模型注册表	集中声明元信息，统一 label 归一化	新增模型只需注册，不改推理逻辑
语言自动检测	langdetect 检测后自动降级	避免语言与模型不匹配的低质量结果
lru_cache 懒加载	首次请求时加载，命中缓存时复用	多模型共存时按需占用内存
限流中间件	slowapi 基于 IP 限流	防止单一客户端打爆服务
批量推理接口	逐条推理，部分失败不影响其余	降低调用方的网络开销
Prometheus 指标	独立端口，按模型和情感分类计数	为监控告警提供细粒度数据基础
历史记录	内存 deque，可替换为 Redis	快速排查异常输入
ConfigMap 解耦	配置与镜像分离	切换模型无需重新构建镜像
HPA 扩缩容	CPU/内存双指标 + 冷却窗口	流量波峰自动扩容，低谷自动回收