【从0搭建AI智能体·11】把 Agent 部署上线：Docker + 限流 + 熔断 + 优雅降级

📚 本文是《从 0 搭建你的 AI 智能体》专栏第 11 篇。
上一篇：《Agent 可观测性：日志、追踪、成本监控》　|　专栏目录：点此查看全部

标签建议：部署 Docker 限流 熔断 FastAPI AI Agent 高可用

📌 前言：本地跑得好好的，一上线就崩

你的 Agent 在本地 python app.py 跑得欢，于是你信心满满地丢上服务器，然后：

• 密钥怎么安全传进容器？总不能写死在代码里带上去；
• 来了一波流量，并发请求把上游 API 的限额打爆，全线 429；
• 上游模型服务抽风了，你的服务跟着一起卡死、雪崩；
• 用户请求超时，前端转圈到天荒地老，没有任何兜底。

「能在本地跑」和「能扛住线上流量」之间，隔着一整套工程化。 这一篇就把 Agent 从「本地脚本」变成「生产级服务」——Docker 容器化、限流保护、熔断防雪崩、优雅降级、健康检查，一套打包给你，都是可直接用的配置和代码。

💡 本文适合谁：Agent 功能已完成、准备部署到生产的开发者。以 FastAPI + Docker 为例。阅读约 14 分钟。

目录

1. 上线前的检查清单
2. 第一步：Docker 容器化
3. 第二步：密钥与配置的安全注入
4. 第三步：限流（保护上游 API 额度）
5. 第四步：熔断（防止雪崩）
6. 第五步：超时与优雅降级
7. 第六步：健康检查与优雅关闭
8. 完整部署配置
9. 常见坑与建议
10. 总结

---

① 上线前的检查清单

先给你一张「上线体检表」，逐项对照，缺哪补哪：

项目	为什么	本文章节
容器化	环境一致、可复制、易扩缩容	②
密钥安全注入	绝不硬编码进镜像	③
限流	保护上游额度、防打爆	④
熔断	上游挂了别跟着雪崩	⑤
超时 + 降级	别让用户无限等	⑥
健康检查	让编排系统知道死活	⑦
日志 + 监控	出事看得见（上一篇）	第10篇

---

② 第一步：Docker 容器化

Docker 让「在我机器上能跑」变成「在哪都能跑」。写一个 Dockerfile：

# 用官方 slim 镜像，体积小
FROM python:3.11-slim

WORKDIR /app

# 先装依赖（利用 Docker 层缓存，依赖不变就不重装）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 再拷代码
COPY . .

# 非 root 用户运行，更安全
RUN useradd -m appuser && chown -R appuser /app
USER appuser

EXPOSE 8000

# 生产用多 worker 的 uvicorn/gunicorn，别用 --reload
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

配套 .dockerignore（别把垃圾和密钥打进镜像）：

__pycache__/
*.pyc
.env               # ⚠️ 关键：密钥文件绝不进镜像
.git/
*.log
chroma_db/

构建与运行：

docker build -t my-agent:latest .
docker run -d -p 8000:8000 --env-file .env my-agent:latest

🎯 两个高频坑：

1. .env 一定要写进 .dockerignore——否则密钥被打进镜像，镜像一分享就泄露；
2. 生产别用 --reload（那是开发用的），用 --workers N 起多进程扛并发。

---

③ 第二步：密钥与配置的安全注入

密钥怎么进容器？通过环境变量注入，绝不写进镜像（呼应第 1 篇的安全规范）。按环境从简到严：

# 方式1：--env-file（小项目/单机）
docker run --env-file .env my-agent

# 方式2：-e 单个注入（CI/CD 里从密文变量取）
docker run -e AGENT_KEY=$AGENT_KEY my-agent

生产集群（Kubernetes）用 Secret：

# k8s-secret.yaml
apiVersion: v1
kind: Secret
metadata:
  name: agent-secrets
type: Opaque
stringData:
  AGENT_KEY: "sk-xxx"
---
# Deployment 里引用
spec:
  containers:
    - name: agent
      image: my-agent:latest
      envFrom:
        - secretRef:
            name: agent-secrets

🚨 安全红线：密钥的三不原则——不进代码、不进镜像、不进 Git。永远通过运行时环境注入。云厂商的 Secrets Manager / Parameter Store 是更专业的选择。

---

④ 第三步：限流（保护上游 API 额度）

上游大模型 API 有 RPM/TPM 限额（第 1 篇讲过）。你的服务如果不限流，一波流量直接把额度打爆，全员 429。所以要在自己这一层先限流。

用 Redis 实现一个简单的滑动窗口限流（复用第 6 篇的 Redis）：

import os, time, redis
from fastapi import HTTPException

# 主机名从环境变量读：本地 localhost，容器里由 compose 注入 REDIS_HOST=redis
r = redis.Redis(host=os.getenv("REDIS_HOST", "localhost"),
                port=6379, decode_responses=True)


def rate_limit(user_id, max_requests=20, window=60):
    """每个用户每 window 秒最多 max_requests 次请求"""
    key = f"ratelimit:{user_id}"
    now = time.time()
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, now - window)   # 清掉窗口外的记录
    pipe.zadd(key, {str(now): now})               # 记录本次
    pipe.zcard(key)                               # 数窗口内请求数
    pipe.expire(key, window)
    count = pipe.execute()[2]
    if count > max_requests:
        raise HTTPException(status_code=429, detail="请求过于频繁，请稍后再试")

在接口里调用：

@app.post("/chat")
async def chat(body: dict):
    rate_limit(body["user_id"])      # 先过限流
    # ... 正常处理 ...

🎯 两层限流最稳：① 用户级（防单个用户刷爆，如上）；② 全局级（所有请求加总不超过上游 RPM）。再配合第 1 篇的指数退避重试兜底偶发 429，就很稳了。

---

⑤ 第四步：熔断（防止雪崩）

雪崩是分布式系统头号杀手：上游模型 API 挂了或变慢，你的请求全堆在那儿等超时，连接池被占满，最后你的服务也被拖垮——一个下游故障，引发全线崩溃。

熔断器（Circuit Breaker） 就是「保险丝」：当上游连续失败达到阈值，熔断器「跳闸」，之后的请求直接快速失败（不再傻等上游），过一段时间再«半开»试探恢复。

import time


class CircuitBreaker:
    def __init__(self, fail_threshold=5, recovery_time=30):
        self.fail_threshold = fail_threshold     # 连续失败多少次跳闸
        self.recovery_time = recovery_time       # 跳闸后多久尝试恢复
        self.failures = 0
        self.state = "closed"                    # closed(正常)/open(熔断)/half-open(试探)
        self.opened_at = 0

    def call(self, func, *args, **kwargs):
        # 熔断中：看是否到了试探时间
        if self.state == "open":
            if time.time() - self.opened_at > self.recovery_time:
                self.state = "half-open"         # 进入半开，放一个请求试探
            else:
                raise Exception("服务熔断中，请稍后再试")   # 快速失败，不等上游

        try:
            result = func(*args, **kwargs)
            self.failures = 0                    # 成功，重置
            self.state = "closed"
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.fail_threshold:
                self.state = "open"              # 连续失败，跳闸
                self.opened_at = time.time()
            raise e


breaker = CircuitBreaker()

# 用法：把上游调用包进熔断器
def call_llm_protected(messages):
    return breaker.call(call_llm, messages)

🎯 熔断的价值：上游挂了，熔断器让你的服务快速失败并返回友好提示，而不是让成千上万请求堆积等超时、最终把自己拖死。保护自己，也给上游喘息恢复的机会。

💡 生产可用成熟库如 pybreaker，或服务网格（Istio）在基础设施层做熔断，不必都手写。

---

⑥ 第五步：超时与优雅降级

超时是底线：任何外部调用都必须设超时，否则一个卡住的请求能拖垮整个服务。

import httpx

async def call_llm(messages):
    async with httpx.AsyncClient(timeout=30) as client:   # 必设超时！
        resp = await client.post(url, json=..., headers=...)
        return resp.json()[...]

优雅降级：出问题时给用户一个「兜底回复」，而不是抛个 500 错误页。

async def chat_with_fallback(messages):
    try:
        return await call_llm_protected(messages)         # 熔断+超时保护的调用
    except Exception as e:
        log_event("llm_fallback", error=str(e))           # 记日志（第10篇）
        # 降级：返回友好兜底，而非崩溃
        return "抱歉，服务暂时繁忙，请稍后再试。如问题紧急，请联系人工客服。"

🎯 降级的心法：用户要的是「有个说法」，不是「白屏 500」。哪怕给一句「稍后再试」，体验也远好过报错。关键路径都该有兜底。

---

⑦ 第六步：健康检查与优雅关闭

健康检查：让 K8s / 负载均衡器知道你的服务「还活着」，不健康就别往这台打流量。

@app.get("/health")
def health():
    # 可加检查：Redis 通不通、上游可达否
    try:
        r.ping()
        return {"status": "ok"}
    except Exception:
        raise HTTPException(status_code=503, detail="unhealthy")

优雅关闭：重启/扩缩容时，别粗暴掐断正在处理的请求，等它们处理完再退出。

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    yield                                    # 启动
    # 关闭时的清理：等待进行中的请求、关连接池
    log_event("shutdown", detail="graceful shutdown")

app = FastAPI(lifespan=lifespan)

K8s 里配上探针：

livenessProbe:                # 死了就重启
  httpGet: { path: /health, port: 8000 }
  initialDelaySeconds: 10
  periodSeconds: 15
readinessProbe:               # 没就绪就不给流量
  httpGet: { path: /health, port: 8000 }
  initialDelaySeconds: 5
  periodSeconds: 10

---

⑧ 完整部署配置

用 docker-compose 把 Agent + Redis 一键拉起（中小项目够用）：

# docker-compose.yml
version: "3.8"
services:
  agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env                     # 密钥从这注入，不进镜像
    environment:
      - REDIS_HOST=redis         # ⚠️ 容器内连 Redis 用服务名，不是 localhost！
    depends_on:
      - redis
    restart: always              # 挂了自动重启
    deploy:
      resources:
        limits:
          memory: 1G             # 限制内存，防单容器吃爆宿主机

  redis:
    image: redis:7-alpine
    restart: always
    volumes:
      - redis_data:/data         # 持久化会话数据

volumes:
  redis_data:

一键启动：

docker-compose up -d          # 后台启动全部服务
docker-compose logs -f agent  # 看日志
docker-compose down           # 停止

前面加个 Nginx 反代（记得第 4 篇的 proxy_buffering off; 让流式生效）：

location / {
    proxy_pass http://localhost:8000;
    proxy_buffering off;                    # 流式输出必须关缓冲（第4篇）
    proxy_read_timeout 120s;                # 给长回复留足时间
}

8.1 验证部署成功（别凭感觉，动手确认）

启动后，逐项验证各道防线真的生效，而不是"看起来跑起来了"：

# 1. 容器都在跑？（agent 和 redis 都应是 Up）
docker-compose ps

# 2. 健康检查通不通？应返回 {"status":"ok"}
curl http://localhost:8000/health

# 3. 正常对话通不通？
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"user_id":"u1","text":"你好"}'

# 4. 限流真的生效？连发 30 次，应该在第 20 次后开始返回 429
for i in $(seq 1 30); do
  curl -s -o /dev/null -w "%{http_code} " \
    -X POST http://localhost:8000/chat \
    -H "Content-Type: application/json" \
    -d '{"user_id":"u1","text":"hi"}'
done
echo

第 4 步的预期输出（前 20 个 200，之后被限流挡下变 429）：

200 200 200 ... 200 429 429 429 429 429 429 429 429 429 429

🔍 看到 200 变 429 那一刻，就是你的限流防线肉眼可见地生效了。同理可测熔断（把上游地址临时改错，看是否快速失败而非卡死）。防御性配置一定要主动验证，别等真实事故来"帮你测"。

---

⑨ 常见坑与建议

坑	说明	解决
.env 打进镜像	密钥泄露	写进 .dockerignore
用 --reload 上生产	性能差、不稳	用 --workers N
没设超时	一个卡请求拖垮全服务	所有外部调用设 timeout
不限流	打爆上游额度、被薅羊毛	用户级 + 全局级限流
无熔断	上游故障引发雪崩	熔断器快速失败
无健康检查	死了还在接流量	/health + K8s 探针
流式在线上失效	Nginx 缓冲	proxy_buffering off
容器里连不上 Redis	代码写死 localhost，容器内 localhost 是容器自己	用 compose 服务名（REDIS_HOST=redis）

🔍 上线前必做的一件事：做一次压测（用 locust、wrk 等模拟并发），看限流、熔断、降级是否真的生效。别等真实流量来了才发现兜底没兜住。

---

⑩ 总结

这一篇把 Agent 从「本地脚本」升级成了「生产级服务」：

能力	手段	防住什么
容器化	Docker + compose	环境不一致
密钥安全	环境变量 / Secret	凭证泄露
限流	Redis 滑动窗口	打爆额度、薅羊毛
熔断	Circuit Breaker	雪崩
超时降级	timeout + fallback	无限等待、白屏
健康检查	/health + 探针	死服务接流量

核心认知：生产环境的工作量，一半在功能，一半在「防止功能出问题」。 限流、熔断、降级这些「防御性工程」，平时看不出价值，出事时就是它们在保你不被叫醒。

到这里，你的 Agent 已经能安全、稳定地对外服务了。但还有最后一道关——安全。下一篇（本专栏收官）讲 AI 应用特有的安全威胁：幻觉、越权、提示词注入，以及怎么防。

---

🔜 下一篇预告（收官）：《常见幻觉 / 越权 / 注入攻击与防御》——AI 应用特有的安全威胁全解。
👍 如果本文帮到你，点赞 / 收藏 / 关注，追更不迷路。