生产环境

Ollama：简化LLM本地部署 Ollama是2026年最受欢迎的本地LLM部署工具之一。它以极简的命令行界面和自动化的模型管理，让在本地运行大模型变得前所未有的简单。但从"能跑"到"生产可用"之间，还有大量工程细节需要处理。 安装与环境准备 系统要求 # Linux安装 curl -fsSL https://ollama.com/install.sh | sh # 验证GPU支持 nvidia-smi # 确认GPU可用 ollama --version GPU显存规划 不同模型的显存需求： 模型 参数量 FP16显存 INT4显存 推荐GPU Qwen-3-7B 7B 14GB 5GB RTX 4060 8GB+ Llama-3-8B 8B 16GB 6GB RTX 4070 12GB+ Qwen-3-32B 32B 64GB 20GB RTX 4090 24GB+ Llama-3-70B 70B 140GB 40GB 2×A100 80GB Ollama服务配置 # 自定义模型存储路径 export OLLAMA_MODELS=/data/ollama/models # 监听所有网络接口（生产环境配合防火墙） export OLLAMA_HOST=0.0.0.0:11434 # 并发请求数 export OLLAMA_NUM_PARALLEL=4 # 上下文长度 export OLLAMA_CONTEXT_LENGTH=8192 # GPU层数（-1为全部卸载到GPU） export OLLAMA_NUM_GPU=-1 # 启动服务 ollama serve 模型管理 Modelfile自定义 # 基于Qwen-3创建自定义模型 FROM qwen3:32b # 系统提示词 SYSTEM """ 你是一个专业的技术助手。请提供准确、简洁的回答。 如果不确定，请明确说明。 """ # 参数调优 PARAMETER temperature 0.7 PARAMETER top_p 0.9 PARAMETER top_k 40 PARAMETER num_ctx 8192 PARAMETER num_gpu 50 PARAMETER stop "<|im_end|>" PARAMETER stop "<|endoftext|>" # 模板 TEMPLATE """ {{ if .System }}<|im_start|>system {{ .System }}<|im_end|> {{ end }}{{ range .Messages }}{{ if eq .Role "user" }}<|im_start|>user {{ .Content }}<|im_end|> {{ end }}{{ if eq .Role "assistant" }}<|im_start|>assistant {{ .Content }}<|im_end|> {{ end }}{{ end }}<|im_start|>assistant """ # 构建自定义模型 ollama create my-qwen -f Modelfile # 运行 ollama run my-qwen 模型量化 # Ollama自动选择量化级别 ollama pull llama3:70b # 默认INT4量化 ollama pull llama3:70b-q8_0 # 指定INT8 ollama pull llama3:70b-fp16 # FP16精度 # 从GGUF文件导入 ollama create my-model --file ./model.gguf API服务 REST API import requests # 基础对话 response = requests.post( "http://localhost:11434/api/chat", json={ "model": "my-qwen", "messages": [ {"role": "user", "content": "解释Transformer的注意力机制"} ], "stream": False, "options": { "temperature": 0.7, "num_ctx": 8192, } } ) print(response.json()["message"]["content"]) # 流式响应 response = requests.post( "http://localhost:11434/api/chat", json={"model": "my-qwen", "messages": [...], "stream": True}, stream=True ) for line in response.iter_lines(): if line: chunk = json.loads(line) print(chunk["message"]["content"], end="", flush=True) OpenAI兼容API Ollama提供OpenAI兼容接口，方便迁移现有应用： ...

前言：RAG从Demo到生产的鸿沟 写一个RAG Demo只需要30分钟——加载文档、Embedding、存向量库、检索、生成，完成。但把它放到生产环境，你会发现： 幻觉：LLM明明拿到了正确文档，还是编造了不存在的信息 漏检：库里明明有相关文档，检索就是找不到 延迟：用户等了8秒还没返回，体验崩溃 这三个问题构成了RAG生产环境的"不可能三角"。本文分享2026年我们在生产环境中踩过的坑和解决方案。 难题一：幻觉问题 幻觉的三种形态 类型 表现 根因 忠实性幻觉 答案与检索文档矛盾 LLM忽略上下文，依赖自身参数知识 编造型幻觉 答案包含文档中不存在的信息 LLM"脑补"细节 来源混淆 将多个文档的信息错误组合 多文档检索时上下文混淆 解决方案矩阵 方案1：强约束Prompt # ❌ 容易幻觉的Prompt prompt = f"""基于以下资料回答问题： {context} 问题：{question} """ # ✅ 抗幻觉的Prompt prompt = f"""你是一个严格的信息提取助手。请遵循以下规则： 1. **只使用**以下参考资料回答问题 2. 如果参考资料中没有相关信息，直接回答"根据现有资料无法回答此问题" 3. 不要添加任何参考资料中未提及的信息 4. 不要进行推理、猜测或补全 5. 回答中需要引用具体的资料来源 参考资料： --- {context} --- 问题：{question} 回答格式： [来源：文档名] 回答内容... """ 方案2：置信度校准 def answer_with_confidence(question, retrieved_docs): """带置信度的回答""" # 第一步：评估检索质量 relevance_prompt = f"""评估以下文档与问题的相关性(0-10): 问题: {question} 文档: {retrieved_docs[0].page_content[:500]} 只返回数字。""" relevance_score = int(llm.invoke(relevance_prompt).strip()) if relevance_score < 4: return { "answer": "抱歉，知识库中没有找到与您问题相关的信息。", "confidence": 0.2, "should_answer": False, } # 第二步：生成回答并自我验证 answer = rag_chain.invoke(question) # 第三步：验证回答是否忠于上下文 verify_prompt = f"""判断以下回答的每个陈述是否能在参考资料中找到支撑。 参考资料: {retrieved_docs} 回答: {answer} 返回JSON: {{"faithful": true/false, "unsupported_claims": [...]}}""" verification = json.loads(llm.invoke(verify_prompt)) return { "answer": answer if verification["faithful"] else "无法确认回答准确性", "confidence": relevance_score / 10.0, "should_answer": verification["faithful"], } 方案3：Citation机制 def generate_with_citations(question, retrieved_docs): """强制引用来源的生成""" # 给每个文档编号 numbered_context = "\n\n".join([ f"[{i+1}] {doc.page_content}" for i, doc in enumerate(retrieved_docs) ]) prompt = f"""基于以下编号的参考资料回答问题。每个陈述后必须标注来源编号[1]。 参考资料： {numbered_context} 问题：{question} 要求： - 每个事实陈述后标注来源编号，如"GraphRAG由微软提出[1]" - 如果某个陈述来自多个文档，标注所有来源，如"[1][3]" - 无法找到来源的陈述不要写出 回答：""" return llm.invoke(prompt) 幻觉抑制效果实测 方案 幻觉率↓ 答案完整度↓ 延迟增加 强约束Prompt 45% 15% +0s 置信度校准 68% 8% +1.5s Citation机制 72% 5% +0.8s 三者结合 85% 22% +2.3s 建议组合：强约束Prompt + Citation机制，性价比最高。 ...

引言 将 Agent 从 Demo 推向生产环境，需要跨越一道巨大的鸿沟。2025年，我们目睹了太多 Agent 在生产环境中翻车的案例：无限循环烧掉数万美元、Prompt 注入导致数据泄露、并发请求拖垮整个系统。这份 Checklist 是用真金白银换来的经验。 一、模型层 Checklist 1.1 模型选择与配置 模型版本锁定：生产环境使用明确的模型版本（如 gpt-5-2026-06-01），而非 gpt-5-latest Fallback 模型配置：主模型不可用时自动切换到备用模型 Token 限制设置：max_tokens 根据业务场景硬性设置，防止无限生成 Temperature 策略：事实性任务 T=0-0.3，创意任务 T=0.7-1.0，代码生成 T=0-0.2 上下文窗口管理：实现对话历史压缩策略，防止 Context Overflow PRODUCTION_MODEL_CONFIG = { "primary": { "model": "gpt-5-2026-06-01", "max_tokens": 4096, "temperature": 0.2, "timeout": 30, "retry": {"max_attempts": 3, "backoff": "exponential"} }, "fallback": { "model": "claude-opus-4-2026-04", "max_tokens": 4096, "temperature": 0.2, "timeout": 45, }, "emergency": { "model": "gpt-4o-2026-03", "max_tokens": 2048, "temperature": 0.0, } } 1.2 成本控制 Token 预算硬限：每个请求/用户/天的 Token 上限 模型路由策略：简单任务用小模型，复杂任务用大模型 缓存层：语义缓存命中率监控 成本告警：单次请求成本 > $0.1 时告警 class TokenBudget: """Token 预算管理器""" def __init__(self, redis_client): self.redis = redis_client async def check_budget( self, user_id: str, requested_tokens: int ) -> bool: daily_key = f"budget:{user_id}:{date.today()}" used = int(await self.redis.get(daily_key) or 0) limit = await self._get_user_limit(user_id) if used + requested_tokens > limit: await self._notify_overrun(user_id, used, requested_tokens, limit) return False return True async def consume( self, user_id: str, tokens_used: int, cost: float ): daily_key = f"budget:{user_id}:{date.today()}" cost_key = f"cost:{user_id}:{date.today()}" pipe = self.redis.pipeline() pipe.incrby(daily_key, tokens_used) pipe.incrbyfloat(cost_key, cost) pipe.expire(daily_key, 86400 * 2) pipe.expire(cost_key, 86400 * 2) await pipe.execute() 二、Prompt 层 Checklist System Prompt 注入防护：用户输入与系统指令严格分离 Prompt 版本管理：所有 Prompt 变更通过 Git 管理，支持 A/B 测试 Prompt 长度监控：System Prompt 不超过 Context Window 的 20% Few-shot 示例审计：示例中不包含敏感数据 输出格式约束：使用 JSON Mode 或 Structured Output class PromptSafetyValidator: """Prompt 安全验证器""" INJECTION_PATTERNS = [ r"ignore.*previous.*instructions", r"you.*are.*now.*a", r"system.*prompt.*is", r"reveal.*your.*instructions", ] def validate(self, user_input: str) -> ValidationResult: for pattern in self.INJECTION_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return ValidationResult( safe=False, reason=f"Potential prompt injection: matched {pattern}" ) if len(user_input) > 10000: return ValidationResult( safe=False, reason="Input exceeds maximum length" ) return ValidationResult(safe=True) 三、工具层 Checklist 工具输入校验：每个工具的输入参数用 Pydantic 模型校验 工具超时设置：每个工具调用设置独立超时 工具权限分级：只读工具自动执行，写操作需审批 工具结果截断：工具输出超过 N 字符时自动摘要 工具幂等性：关键工具支持重试不会产生副作用 from pydantic import BaseModel, Field from typing import Literal class ToolRegistry: """工具注册中心""" def register(self, tool: Tool): # 验证工具定义 assert tool.timeout_ms <= 30000, "Tool timeout must be < 30s" assert tool.input_schema is not None, "Input schema required" assert tool.permission_level in ["read", "write", "admin"] if tool.permission_level in ["write", "admin"]: assert tool.requires_confirmation == True self._tools[tool.name] = tool class WebSearchInput(BaseModel): query: str = Field(..., min_length=1, max_length=200) max_results: int = Field(default=5, ge=1, le=20) safe_search: bool = True class WebSearchTool(Tool): name = "web_search" description = "Search the web for information" input_schema = WebSearchInput timeout_ms = 10000 permission_level = "read" requires_confirmation = False idempotent = True 四、架构层 Checklist 4.1 并发与限流 请求限流：QPS / 并发数 / Token/s 三维度限流 队列隔离：不同优先级请求使用独立队列 背压机制：下游不可用时主动拒绝请求，而非堆积 from asyncio import Semaphore, Queue import asyncio class AgentRequestHandler: def __init__(self, max_concurrent: int = 10): self.semaphore = Semaphore(max_concurrent) self.priority_queue = Queue(maxsize=1000) self.normal_queue = Queue(maxsize=5000) async def handle(self, request, priority: str = "normal"): queue = self.priority_queue if priority == "high" else self.normal_queue if queue.full(): raise ServiceUnavailableError("Request queue full") await queue.put(request) async with self.semaphore: return await self._process(request) 4.2 状态管理 会话持久化：对话状态可持久化到外部存储 检查点机制：长流程 Agent 支持断点续传 幂等 ID：每个请求分配幂等 ID，防重复执行 4.3 容灾设计 多 Provider 热备：OpenAI / Anthropic / Azure 至少两家 降级策略：LLM 不可用时回退到规则引擎 熔断器：错误率 > 50% 时自动熔断 30s class CircuitBreaker: """Agent 熔断器""" def __init__( self, failure_threshold: int = 10, recovery_timeout: int = 30, half_open_max_calls: int = 3 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls self._state = "closed" # closed / open / half_open self._failures = 0 self._last_failure_time = None async def call(self, func, *args, **kwargs): if self._state == "open": if time.time() - self._last_failure_time > self.recovery_timeout: self._state = "half_open" else: raise CircuitOpenError("Agent temporarily unavailable") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise 五、安全层 Checklist PII 检测与脱敏：用户输入和 LLM 输出中的 PII 自动脱敏 输出过滤：有害内容、不当建议的过滤层 工具沙箱：代码执行工具在容器中运行，限制网络和文件访问 审计日志：所有 Agent 决策记录可审计 数据驻留：敏感数据不发送到 LLM，本地处理 class PIIRedactor: """PII 脱敏器""" PATTERNS = { "email": (r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]'), "phone": (r'\b1[3-9]\d{9}\b', '[PHONE]'), "id_card": (r'\b\d{17}[\dXx]\b', '[ID_CARD]'), "bank_card": (r'\b\d{16,19}\b', '[BANK_CARD]'), } def redact(self, text: str) -> tuple[str, dict]: """返回脱敏文本和映射表（用于恢复）""" mapping = {} redacted = text for pii_type, (pattern, replacement) in self.PATTERNS.items(): matches = re.finditer(pattern, redacted) for i, match in enumerate(matches): placeholder = f"{replacement}_{i}" mapping[placeholder] = match.group() redacted = redacted.replace(match.group(), placeholder) return redacted, mapping 六、可观测性 Checklist 全链路追踪：每个 Agent 步骤生成 Span 结构化日志：所有日志 JSON 格式，包含 trace_id 实时仪表盘：Grafana 仪表盘展示关键指标 告警规则：延迟、错误率、成本、Token 使用量告警 会话回放：支持根据 trace_id 回放完整 Agent 执行过程 七、运维层 Checklist 蓝绿部署：新版本 Agent 灰度发布，支持秒级回滚 配置热更新：Prompt、工具配置不重启即可更新 依赖版本锁定：所有依赖版本锁定，定期安全扫描 Secret 管理：API Key 通过 Vault/Secret Manager 管理 备份策略：对话历史、Agent 状态定期备份 八、合规层 Checklist 数据保留策略：用户对话数据保留期限明确 GDPR/PIPL 合规：支持用户数据删除请求 模型透明度：向用户说明使用了 AI 及其局限性 内容免责声明：输出中标注 AI 生成内容 审计合规：关键决策可追溯，满足监管要求 九、性能层 Checklist P95 延迟 < 5s：首 Token 延迟 < 2s，完整响应 < 30s 流式响应：长输出使用 SSE 流式返回 预计算：高频请求结果预计算缓存 连接池：LLM API 连接复用 CDN 加速：静态资源（Prompt 模板、配置）通过 CDN 分发 十、用户体验层 Checklist 加载状态：Agent 思考时展示进度提示 优雅降级：LLM 超时时返回友好提示而非错误码 多语言支持：Agent 输出跟随用户语言 反馈机制：用户可对 Agent 回答打分 边界处理：处理空输入、超长输入、非预期输入 十一、测试层 Checklist 单元测试覆盖率 > 80%：非 LLM 逻辑全覆盖 集成测试：关键业务流程端到端测试 混沌测试：主动注入 LLM 超时、工具失败 安全测试：Prompt 注入、数据泄露测试 负载测试：验证 10x 峰值流量下的表现 十二、上线前最终确认 □ 紧急关停开关（Kill Switch）可用且已测试 □ 回滚脚本已验证，可在 60s 内完成 □ on-call 排班已确认，告警通知链路畅通 □ 成本预算已设置硬性上限 □ 用户协议已更新，涵盖 AI 使用条款 □ 压测通过：P95 < 5s, 错误率 < 1% □ 安全审计已完成，无高危漏洞 □ 数据备份已验证可恢复 □ 运行手册（Runbook）已编写 □ 团队培训已完成 结语 这份 Checklist 看似冗长，但每一条都来自真实的生产事故。Agent 系统的复杂度远超传统应用——它不仅有代码的确定性风险，还有模型的不确定性风险、工具的副作用风险。上线的标准不是"能跑了"，而是"出事了能兜住"。祝你的 Agent 平稳上线。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

生产级 RAG 架构总览 生产环境中的 RAG 不是简单的"文档切块 + 向量搜索 + LLM 生成"三步走，而是一个包含数据接入、预处理、索引、检索、重排、生成、后处理等多环节的完整系统。 ┌─────────────────────────────────────────────────────────────────┐ │ RAG 生产架构 │ ├────────────────┬────────────────────────────────────────────────┤ │ 数据层 │ 文档加载 → 清洗 → 分块 → Embedding → 向量库 │ │ 检索层 │ 混合检索(向量+BM25) → 重排 → 上下文组装 │ │ 生成层 │ Prompt 模板 → LLM 调用 → 流式输出 → 后处理 │ │ 基础设施层 │ API网关 → 负载均衡 → 缓存 → 监控 → 日志 │ └────────────────┴────────────────────────────────────────────────┘ 向量数据库选型 三大主流向量库对比： 维度 Milvus Weaviate Qdrant 部署复杂度 高（依赖etcd/MinIO） 中 低（单二进制） 索引算法 HNSW/IVF/DiskANN HNSW HNSW 混合检索 需配合ES 内置BM25 内置稀疏向量 动态schema 支持 支持 支持 集群方案 成熟 支持 支持（一致性哈希） 性能(1M向量,768d) ~1200 QPS ~800 QPS ~1500 QPS 社区活跃度 高 中 高 适用场景 大规模(亿级) 中规模+GraphQL 中小规模+低延迟 选型建议： ...

部署架构设计 生产环境的聊天机器人不是"跑起来就行"，它需要处理高并发、保证可用性、支持快速回滚。一个经过验证的部署架构如下： 用户 → 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 小时窗口处理潜在问题。 ...