聊天机器人生产部署全指南

用户 → CDN/WAF → Nginx (SSL/反向代理) → API Gateway → Agent 服务集群
                                                         ↓
                                              Redis (会话) + 向量DB (知识) + LLM API

# 多阶段构建：分离构建环境和运行环境
FROM python:3.12-slim AS builder

WORKDIR /app

# 安装依赖（利用 Docker 层缓存）
COPY pyproject.toml uv.lock ./
RUN pip install --no-cache-dir uv && \
    uv pip install --system --no-cache -r requirements.txt

# 复制源码
COPY . .

# 运行阶段：更小的镜像
FROM python:3.12-slim AS runtime

# 安装运行时依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl && \
    rm -rf /var/lib/apt/lists/*

# 创建非 root 用户
RUN useradd -m -u 1000 appuser
WORKDIR /app

# 从构建阶段复制
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /app /app

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# 切换用户
USER appuser

EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", \
     "--workers", "4", "--loop", "uvloop", "--no-access-log"]

version: "3.9"

services:
  agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379/0
      - VECTOR_DB_URL=http://qdrant:6333
      - LLM_API_KEY=${LLM_API_KEY}
      - LOG_LEVEL=debug
    depends_on:
      redis:
        condition: service_healthy
      qdrant:
        condition: service_started
    volumes:
      - ./app:/app  # 开发热重载
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 3s
      retries: 5
    volumes:
      - redis-data:/data

  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - qdrant-data:/qdrant/storage

volumes:
  redis-data:
  qdrant-data:

# /etc/nginx/conf.d/chatbot.conf

# 上游 Agent 服务
upstream agent_backend {
    least_conn;
    server 10.0.1.10:8000 weight=3;
    server 10.0.1.11:8000 weight=3;
    server 10.0.1.12:8000 weight=2;
    
    # 健康检查（Nginx Plus 或 OpenResty）
    keepalive 32;
    keepalive_timeout 60s;
}

# HTTP → HTTPS 重定向
server {
    listen 80;
    server_name chat.example.com;
    return 301 https://$server_name$request_uri;
}

# HTTPS 主服务
server {
    listen 443 ssl http2;
    server_name chat.example.com;

    # SSL 配置
    ssl_certificate     /etc/nginx/ssl/fullchain.pem;
    ssl_certificate_key /etc/nginx/ssl/privkey.pem;
    ssl_protocols       TLSv1.2 TLSv1.3;
    ssl_ciphers         ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256;
    ssl_prefer_server_ciphers on;
    ssl_session_cache   shared:SSL:10m;
    ssl_session_timeout 10m;

    # 安全头
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Content-Type-Options nosniff always;
    add_header X-Frame-Options DENY always;

    # 请求体大小限制（文件上传）
    client_max_body_size 10M;

    # SSE 流式输出专用配置
    location /api/chat/stream {
        proxy_pass http://agent_backend;
        
        # 关键：禁用缓冲，否则 SSE 不工作
        proxy_buffering off;
        proxy_cache off;
        
        # 支持长连接
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding on;
        
        # 超时设置（流式输出需要长超时）
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
        
        # 传递客户端信息
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    # 普通 API
    location /api/ {
        proxy_pass http://agent_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 限流
        limit_req zone=api burst=20 nodelay;
    }

    # 静态资源
    location / {
        root /var/www/chatbot-ui;
        try_files $uri $uri/ /index.html;
        
        # 静态资源缓存
        expires 1h;
        add_header Cache-Control "public, immutable";
    }

    # 健康检查端点
    location /health {
        proxy_pass http://agent_backend/health;
        access_log off;
    }
}

# 限流区域定义（放在 http 块中）
# limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;

# 使用 Let's Encrypt + certbot
certbot certonly --webroot -w /var/www/chatbot-ui -d chat.example.com

# 自动续期（crontab）
0 3 * * * certbot renew --quiet --post-hook "systemctl reload nginx"

# 策略1: 最少连接（推荐 Agent 场景）
upstream agent_backend {
    least_conn;
    server 10.0.1.10:8000;
    server 10.0.1.11:8000;
}

# 策略2: IP 哈希（会话亲和）
upstream agent_backend_sticky {
    ip_hash;
    server 10.0.1.10:8000;
    server 10.0.1.11:8000;
}

# 策略3: 加权轮询（异构节点）
upstream agent_backend_weighted {
    server 10.0.1.10:8000 weight=3;  # 高配机器
    server 10.0.1.11:8000 weight=2;  # 中配机器
    server 10.0.1.12:8000 weight=1;  # 低配机器
}

# 策略4: 一致性哈希（Nginx Plus 或第三方模块）
upstream agent_backend_consistent {
    hash $arg_session_id consistent;
    server 10.0.1.10:8000;
    server 10.0.1.11:8000;
}

# Agent 服务健康检查端点
from fastapi import FastAPI
import psutil
import asyncio

app = FastAPI()

@app.get("/health")
async def health_check():
    """轻量级健康检查"""
    return {"status": "healthy", "timestamp": asyncio.get_event_loop().time()}

@app.get("/health/ready")
async def readiness_check():
    """就绪检查：验证所有依赖"""
    checks = {
        "redis": await check_redis(),
        "vector_db": await check_vector_db(),
        "llm_api": await check_llm_api(),
        "gpu": check_gpu_available(),
    }
    
    all_ready = all(checks.values())
    return {
        "ready": all_ready,
        "checks": checks,
    }, 200 if all_ready else 503

async def check_redis() -> bool:
    try:
        await redis_client.ping()
        return True
    except:
        return False

def check_gpu_available() -> bool:
    try:
        import torch
        return torch.cuda.is_available()
    except:
        return True  # CPU 模式也算就绪

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from prometheus_client import make_asgi_app
import time

# 定义指标
REQUEST_COUNT = Counter(
    'agent_requests_total',
    'Total request count',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'agent_request_latency_seconds',
    'Request latency',
    ['endpoint'],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0]
)

FIRST_TOKEN_LATENCY = Histogram(
    'agent_first_token_latency_seconds',
    'Time to first token',
    buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0]
)

ACTIVE_SESSIONS = Gauge(
    'agent_active_sessions',
    'Number of active sessions'
)

LLM_TOKEN_USAGE = Counter(
    'agent_llm_tokens_total',
    'LLM token usage',
    ['direction', 'model']  # direction: input/output
)

# 中间件
@app.middleware("http")
async def metrics_middleware(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()
    
    REQUEST_LATENCY.labels(endpoint=request.url.path).observe(duration)
    
    return response

# 挂载 Prometheus 端点
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)

import structlog
import json
from datetime import datetime

# 配置 structlog
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(20),  # INFO+
)

logger = structlog.get_logger()

# 请求追踪中间件
@app.middleware("http")
async def logging_middleware(request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        session_id=request.headers.get("X-Session-ID", ""),
        client_ip=request.client.host,
        path=request.url.path,
    )
    
    logger.info("request_started")
    
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    logger.info("request_completed", 
                status_code=response.status_code,
                duration_ms=round(duration * 1000, 2))
    
    response.headers["X-Request-ID"] = request_id
    return response

# 日志输出示例：
# {"event":"request_started","request_id":"abc-123","session_id":"sess-456","client_ip":"10.0.0.1","path":"/api/chat","level":"info","timestamp":"2026-06-24T14:00:00Z"}
# {"event":"request_completed","request_id":"abc-123","status_code":200,"duration_ms":1523.45,"level":"info","timestamp":"2026-06-24T14:00:01Z"}

# Filebeat 配置：采集 Agent 日志
filebeat.inputs:
  - type: container
    paths:
      - /var/lib/docker/containers/*/*.log
    processors:
      - decode_json_fields:
          fields: ["message"]
          target: ""
    labels:
      service: chatbot-agent

output.logstash:
  hosts: ["logstash:5044"]

# Logstash 过滤规则
filter {
  if [labels][service] == "chatbot-agent" {
    json {
      source => "message"
    }
    if [request_id] {
      aggregate {
        task_id => "%{request_id}"
        code => "map['total_duration'] = event.get('duration_ms')"
        map_action => "create_or_update"
      }
    }
  }
}

"""灰度发布：逐步将流量切到新版本"""

class CanaryRouter:
    """基于会话 ID 的灰度路由"""
    
    def __init__(self):
        self.versions = {
            "stable": {"weight": 90, "endpoint": "http://agent-stable:8000"},
            "canary": {"weight": 10, "endpoint": "http://agent-canary:8000"},
        }
    
    def route(self, session_id: str) -> str:
        """根据会话 ID 和权重分配版本"""
        # 用会话 ID 哈希确保同一用户始终路由到同一版本
        hash_val = int(hashlib.md5(session_id.encode()).hexdigest(), 16) % 100
        
        cumulative = 0
        for version, config in self.versions.items():
            cumulative += config["weight"]
            if hash_val < cumulative:
                return config["endpoint"]
        
        return self.versions["stable"]["endpoint"]
    
    def update_weights(self, canary_weight: int):
        """调整灰度权重"""
        self.versions["stable"]["weight"] = 100 - canary_weight
        self.versions["canary"]["weight"] = canary_weight


# 灰度发布流程
canary = CanaryRouter()

# 阶段1: 10% 流量到新版本
canary.update_weights(10)
# 监控 30 分钟，检查错误率、延迟、用户反馈

# 阶段2: 50% 流量
canary.update_weights(50)
# 监控 1 小时

# 阶段3: 100% 流量
canary.update_weights(100)
# 旧版本下线

部署前：
□ Docker 镜像构建并推送到镜像仓库
□ 在 staging 环境完成功能测试
□ 数据库迁移脚本准备
□ 回滚方案准备

部署中：
□ 逐节点滚动更新（一次一个）
□ 每个节点更新后健康检查通过
□ 灰度流量切换

部署后：
□ 监控指标 30 分钟无异常
□ 抽样验证核心功能
□ 日志无异常错误模式
□ 告警通道正常工作