AI应用部署上线：Docker打包+API服务+监控告警，我踩了4个坑

前11篇聊的都是本地跑AI应用——python app.py，终端里看输出，调试完了事儿。直到产品轻飘飘来一句："下周一上线，2000用户同时用。"

本地跑和上线，这俩事压根不在一个维度：

我第一次把AI应用部署到服务器，就被现实狠狠教育了一顿——踩了4个坑，每个都是"本地能跑≠线上能用"的经典教训。

先说结论

用Ja va背景的开发者来理解，AI应用部署其实就跟Spring Boot应用部署差不多，只不过多了一个特别慢的外部服务——LLM。所有Spring Boot部署踩过的坑，AI应用一样要踩，还得额外处理LLM的延迟和成本问题。

坑1：本地能跑，服务器跑不起来——环境不一致

翻车现场

bash

# 本地
python app.py  #  正常运行

# 服务器
pip install -r requirements.txt  #  编译chromadb报错：C++编译器版本不对
python app.py                    #  ModuleNotFoundError: No module named '_sqlite3'

Python的依赖管理，有时候真跟玄学似的。不同操作系统、不同Python版本、不同系统库，都可能让同样的代码跑不起来。尤其像chromadb、langchain这种依赖链很长的包，在Linux服务器上装一堆C++扩展，编译失败简直是家常便饭。

正确做法：Docker打包，环境跟着代码走

dockerfile

# Dockerfile
FROM python:3.11-slim

# 系统依赖（chromadb等需要）
RUN apt-get update && apt-get install -y 
    build-essential 
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# 先复制依赖文件，利用Docker缓存
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

yaml

# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - DASHSCOPE_API_KEY=${DASHSCOPE_API_KEY}
    env_file:
      - .env
    depends_on:
      - chromadb
  
  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8001:8000"
    volumes:
      - chromadb_data:/chroma/chroma

volumes:
  chromadb_data:

关键要点：

用Ja va背景来理解：Docker就是你的WAR包 + Tomcat + JDK全打包成一个镜像，不用担心服务器的JDK版本不对——镜像里自带。

坑2：2000用户同时调LLM，API直接打挂

翻车现场

用FastAPI搭了个RAG服务：

python

from fastapi import FastAPI

app = FastAPI()

@app.get("/chat")
def chat(question: str):
    result = rag.query(question)  # 同步调LLM，3-5秒
    return {"answer": result}

压测结果：

10个并发用户： 平均响应3.2秒
50个并发用户：️ 平均响应15秒，部分超时
200个并发用户： 90%超时，服务器CPU 100%

问题出在哪？LLM每次调用3-5秒，同步等待。200个并发等于200个线程同时等LLM返回——线程池爆了，CPU也爆了。而且LLM的API还有速率限制，并发太高直接给你429 Too Many Requests。

正确做法：异步+队列+限流

python

import asyncio
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import redis
import uuid

app = FastAPI()

# ============ 1. 异步LLM调用 ============
async def async_llm_call(prompt: str) -> str:
    """异步调LLM，不阻塞其他请求"""
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="qwen-plus")
    # 使用ainvoke异步调用
    result = await llm.ainvoke(prompt)
    return result.content

# ============ 2. 限流：控制LLM并发数 ============
MAX_LLM_CONCURRENT = 10  # 最多10个并发LLM调用
llm_semaphore = asyncio.Semaphore(MAX_LLM_CONCURRENT)

async def rate_limited_llm_call(prompt: str) -> str:
    """限流的LLM调用"""
    async with llm_semaphore:
        return await async_llm_call(prompt)

# ============ 3. 异步API ============
class ChatRequest(BaseModel):
    question: str

@app.post("/chat")
async def chat(request: ChatRequest):
    """异步聊天接口"""
    answer = await rate_limited_llm_call(request.question)
    return {"answer": answer}

# ============ 4. 长任务走队列 ============
@app.post("/report/submit")
async def submit_report(request: ChatRequest):
    """提交报告生成任务（长任务，走队列）"""
    task_id = str(uuid.uuid4())
    # 实际项目：把任务丢进Redis队列/Celery
    # redis_client.lpush("report_queue", json.dumps({"task_id": task_id, "question": request.question}))
    return {"task_id": task_id, "status": "processing"}

@app.get("/report/{task_id}")
async def get_report(task_id: str):
    """查询报告生成进度"""
    # 实际项目：从Redis/数据库查任务状态
    return {"task_id": task_id, "status": "processing", "progress": 60}

3层防御：

类比一下Ja va生态：异步就是 CompletableFuture，信号量就是 Semaphore，队列就是 RabbitMQ/Kafka。Spring Boot里怎么处理慢接口，AI应用就怎么处理LLM调用。

坑3：API Key写死在代码里，推到GitHub被人偷了

翻车现场

python

# main.py
OPENAI_API_KEY = "sk-xxxxxxxxxxxx"  # 写死在代码里
DASHSCOPE_API_KEY = "sk-yyyyyyyyyyyy"

llm = ChatOpenAI(api_key=OPENAI_API_KEY)

代码推到GitHub，2小时后收到邮件："你的API Key已被用于消费$200+"。

这可不是段子，这是真实发生过无数次的惨案。

正确做法：环境变量 + .env文件

python

# main.py
import os
from dotenv import load_dotenv

load_dotenv()  # 从.env文件加载环境变量

llm = ChatOpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),  # 从环境变量读取
    model="qwen-plus",
)

bash

# .env（不入Git！）
OPENAI_API_KEY=sk-xxxxxxxxxxxx
DASHSCOPE_API_KEY=sk-yyyyyyyyyyyy

bash

# .gitignore
.env
.env.*

3条铁律：

这就跟Spring的application.yml里不放数据库密码一个道理——用环境变量${DB_PASSWORD}注入。AI应用的API Key = 数据库密码，不能写死在代码里。

坑4：线上AI回答质量下降，一周后才发现

翻车现场

RAG上线运行2周，一切看起来正常。直到有用户投诉："你们AI的回答越来越离谱了，之前还能答对，现在全是废话。"

查了日志发现：知识库的向量索引损坏了一部分，检索结果不对，AI拿到的上下文就是错的——但应用没有报错，只返回了低质量回答。

关键是，没有任何监控和告警，线上问题全靠用户投诉才能发现。

正确做法：3层监控

python

import logging
import time
from datetime import datetime

# ============ 1. 结构化日志 ============
logger = logging.getLogger("ai_app")
logger.setLevel(logging.INFO)

# 结构化日志格式
formatter = logging.Formatter(
    '{"time": "%(asctime)s", "level": "%(levelname)s", "event": "%(message)s"}'
)

class AILogger:
    """AI应用专用日志"""

    @staticmethod
    def log_query(question: str, answer: str, latency: float, tokens: int = 0):
        """记录查询日志"""
        logger.info(
            f'query_completed | '
            f'question="{question[:50]}" | '
            f'answer_len={len(answer)} | '
            f'latency={latency:.2f}s | '
            f'tokens={tokens}'
        )

    @staticmethod
    def log_error(question: str, error: str):
        """记录错误日志"""
        logger.error(
            f'query_failed | '
            f'question="{question[:50]}" | '
            f'error="{error}"'
        )

    @staticmethod
    def log_llm_call(model: str, prompt_tokens: int, completion_tokens: int, latency: float):
        """记录LLM调用日志"""
        logger.info(
            f'llm_call | '
            f'model={model} | '
            f'prompt_tokens={prompt_tokens} | '
            f'completion_tokens={completion_tokens} | '
            f'latency={latency:.2f}s'
        )

# ============ 2. 健康检查端点 ============
@app.get("/health")
async def health_check():
    """健康检查：检查各组件是否正常"""
    checks = {}

    # 检查LLM
    try:
        start = time.time()
        await rate_limited_llm_call("ping")
        checks["llm"] = {"status": "ok", "latency": f"{time.time()-start:.2f}s"}
    except Exception as e:
        checks["llm"] = {"status": "error", "detail": str(e)}

    # 检查向量库
    try:
        docs = vector_store.similarity_search("test", k=1)
        checks["vector_store"] = {"status": "ok", "doc_count": len(docs)}
    except Exception as e:
        checks["vector_store"] = {"status": "error", "detail": str(e)}

    # 检查Redis（如果用了队列）
    try:
        # redis_client.ping()
        checks["redis"] = {"status": "ok"}
    except Exception as e:
        checks["redis"] = {"status": "error", "detail": str(e)}

    all_ok = all(c["status"] == "ok" for c in checks.values())
    return {"status": "ok" if all_ok else "degraded", "checks": checks}

# ============ 3. 关键指标统计 ============
from collections import defaultdict
from threading import Lock

class Metrics:
    """简单的指标收集器"""

    def __init__(self):
        self.data = defaultdict(list)
        self.lock = Lock()

    def record(self, name: str, value: float):
        with self.lock:
            self.data[name].append({
                "value": value,
                "timestamp": datetime.now().isoformat(),
            })
            # 只保留最近1000条
            if len(self.data[name]) > 1000:
                self.data[name] = self.data[name][-1000:]

    def get_stats(self, name: str) -> dict:
        values = [d["value"] for d in self.data.get(name, [])]
        if not values:
            return {"count": 0}
        return {
            "count": len(values),
            "a vg": sum(values) / len(values),
            "min": min(values),
            "max": max(values),
        }

metrics = Metrics()

# 在查询接口中记录指标
@app.post("/chat")
async def chat(request: ChatRequest):
    start = time.time()
    try:
        answer = await rate_limited_llm_call(request.question)
        latency = time.time() - start
        metrics.record("query_latency", latency)
        metrics.record("answer_length", len(answer))
        AILogger.log_query(request.question, answer, latency)
        return {"answer": answer}
    except Exception as e:
        metrics.record("query_error", 1)
        AILogger.log_error(request.question, str(e))
        return {"error": "服务暂时不可用，请稍后重试"}, 503

@app.get("/metrics")
async def get_metrics():
    """指标端点：供监控系统采集"""
    return {
        "query_latency": metrics.get_stats("query_latency"),
        "answer_length": metrics.get_stats("answer_length"),
        "error_count": metrics.get_stats("query_error"),
    }

3层监控：

必须告警的3个场景：

用Ja va的方式来理解，这就是Spring Boot Actuator + Prometheus + Grafana的AI应用版本。健康检查 = /actuator/health，指标 = /actuator/metrics，日志 = logback结构化日志。

完整项目结构

ai-rag-app/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI入口
│   ├── rag.py            # RAG核心逻辑
│   ├── config.py         # 配置（从环境变量读取）
│   ├── logger.py         # 日志工具
│   └── metrics.py        # 指标收集
├── tests/
│   ├── test_unit.py      # 单元测试
│   ├── test_integration.py # 集成测试
│   └── golden_dataset.json # 黄金数据集
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .env.example          # 环境变量模板（入Git）
├── .env                  # 真实环境变量（不入Git）
└── .gitignore

bash

# 部署命令
docker compose up -d

# 查看日志
docker compose logs -f app

# 健康检查
curl http://localhost:8000/health

# 指标查看
curl http://localhost:8000/metrics

4个坑的总结

从开发到上线的Checklist