部署架构设计 生产环境的聊天机器人不是"跑起来就行"，它需要处理高并发、保证可用性、支持快速回滚。一个经过验证的部署架构如下：
用户 → CDN/WAF → Nginx (SSL/反向代理) → API Gateway → Agent 服务集群 ↓ Redis (会话) + 向量DB (知识) + LLM API 架构组件职责 组件 职责 推荐方案 CDN/WAF 静态资源加速、DDoS 防护 Cloudflare / 阿里云 CDN Nginx SSL 终止、反向代理、负载均衡 Nginx / OpenResty API Gateway 鉴权、限流、路由 Kong / APISIX Agent 服务 业务逻辑、LLM 调用 FastAPI / Gin 会话存储 对话上下文 Redis Cluster 知识库 向量检索 Milvus / Qdrant 监控 指标采集、告警 Prometheus + Grafana 日志 日志聚合 ELK / Loki Docker 容器化 Dockerfile 最佳实践 # 多阶段构建：分离构建环境和运行环境 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"] Docker Compose 开发环境 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: 镜像优化要点 优化项 效果 方法 多阶段构建 镜像减小 40-60% 构建阶段和运行阶段分离 Slim 基础镜像 减少 200MB+ 用 python:slim 替代 python:full .dockerignore 加快构建 排除 .git, pycache, .venv 等 层缓存优化 加快重建 先 COPY 依赖文件再 COPY 源码 非 root 用户 安全加固 USER 指令切换运行用户 Nginx 反向代理与 SSL Nginx 配置 # /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; SSL 证书自动续期 # 使用 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" 负载均衡配置 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 模式也算就绪 监控告警 Prometheus 指标采集 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) Grafana 告警规则 告警名称 条件 持续时间 严重程度 服务不可用 up == 0 1m Critical 高错误率 rate(agent_requests_total{status=~"5.."}[5m]) / rate(agent_requests_total[5m]) > 0.05 2m Critical P99 延迟高 histogram_quantile(0.99, agent_request_latency_seconds_bucket) > 10 5m Warning 首字延迟高 histogram_quantile(0.95, agent_first_token_latency_seconds_bucket) > 2 5m Warning GPU 利用率高 gpu_utilization > 0.9 10m Warning 会话积压 agent_active_sessions > 500 5m Warning 日志收集 结构化日志 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"} ELK 日志管道 # 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) # 旧版本下线 灰度发布检查清单 检查项 阈值 不通过操作 新版本错误率 <1% 立即回滚 新版本 P99 延迟 <旧版本 1.2 倍 观察，持续超标则回滚 新版本 CPU/内存 <80% 扩容或回滚 用户负反馈 <基线水平 回滚 关键功能测试 全部通过 回滚 实战部署检查清单 部署前： □ Docker 镜像构建并推送到镜像仓库 □ 在 staging 环境完成功能测试 □ 数据库迁移脚本准备 □ 回滚方案准备 部署中： □ 逐节点滚动更新（一次一个） □ 每个节点更新后健康检查通过 □ 灰度流量切换 部署后： □ 监控指标 30 分钟无异常 □ 抽样验证核心功能 □ 日志无异常错误模式 □ 告警通道正常工作 实战建议 永远不要在周五下午部署。给团队 48 小时窗口处理潜在问题。
...
