前11篇聊的都是本地跑AI应用——python app.py，终端里看输出，调试完了事儿。

直到产品轻飘飘来一句："下周一上线，2000用户同时用。"

本地开发和生产上线，不是一个难度级别

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

维度	本地开发	生产上线
运行环境	你的电脑	服务器（Linux）
并发	1个人用	2000人同时用
稳定性	挂了就重启	挂了要告警+自动恢复
安全	API Key写在代码里	必须加密+权限控制
依赖	pip装在本机	Docker镜像打包

我第一次把AI应用部署到服务器，就被现实狠狠教育了一顿。

一共踩了4个坑。每个坑都在说明一件事：本地能跑≠线上能用。

---

先说结论

坑	一句话
环境不一致	Docker打包，别裸奔部署
API并发扛不住	异步+队列+限流，别让LLM调用成为瓶颈
API Key泄露	环境变量+密钥管理，别写死在代码里
线上出问题不知道	日志+监控+告警，别等用户投诉才发现

用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:

关键要点：

• 用slim镜像：python:3.11-slim比python:3.11小5倍。
• 先复制requirements.txt：Docker分层缓存，依赖不变时不用重新pip install。
• 系统依赖显式安装：chromadb等C++扩展需要的库要手动装。
• 环境变量不写死：用.env文件或环境变量注入API Key。
• 数据持久化：chromadb数据挂载到Docker Volume，容器重建不丢数据。

用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层防御：

• 异步调用：ainvoke代替invoke，一个请求等LLM时不阻塞其他请求。
• 信号量限流：asyncio.Semaphore(10)，最多10个LLM并发，防止打爆API。
• 队列异步：长任务丢队列，返回task_id。报告生成30秒+，不能让用户等。

类比一下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条铁律：

• 代码里不写Key：所有密钥用os.getenv()读取。
• .env不入Git：.gitignore里加.env。
• 服务器用系统环境变量：生产环境不要用.env文件，用Docker/K8s环境变量或密钥管理服务。

这就跟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层监控：

• 结构化日志：监控每次查询的详细信息，可用ELK/Grafana Loki 搜日志。
• 健康检查：监控各组件是否正常，通过/health端点 + 外部探针发现异常。
• 指标统计：监控延迟、Token用量、错误率，通过/metrics端点 + Prometheus + Grafana采集。

必须告警的3个场景：

• LLM调用延迟飙升：a vg latency > 10s，说明API可能被限流或服务异常。
• 错误率上升：error rate > 5%，说明上游服务可能挂了。
• 回答质量下降：a vg answer_length < 20字，说明AI可能开始答非所问或返回空。

用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个坑的总结

#	坑	错误做法	正确做法	一句话
1	环境不一致	裸奔部署	Docker打包	本地能跑≠线上能用
2	并发扛不住	同步调LLM	异步+队列+限流	LLM是3-5秒的慢IO，必须异步
3	API Key泄露	写死在代码里	环境变量+.env	Key推GitHub=给黑客送钱
4	出问题不知道	无监控无告警	日志+健康检查+指标	别等用户投诉才发现

---

从开发到上线的Checklist

• 打包
  • Dockerfile能正常build
  • docker-compose能正常启动
• 安全
  • API Key不在代码里
  • .env在.gitignore里
• 性能
  • 异步API
  • LLM调用限流
• 可观测
  • 结构化日志
  • /health健康检查
  • /metrics指标端点
• 测试
  • 黄金数据集测试通过
  • 质量回归测试通过

一句话收尾：

AI应用上线，不是把代码传到服务器就完了。

真正要解决的是环境一致性、并发控制、密钥安全和线上可观测性。

这4件事做好了，AI应用才能从“能跑”走到“能稳定上线”。