实践指南

为什么需要流式输出 LLM 生成完整回答可能需要 5-15 秒。用户盯着空白屏幕等待体验极差。流式输出让用户像看打字机一样实时看到生成内容，首 token 延迟降到 200-500ms，体感速度提升 10 倍。 SSE vs WebSocket 对比 维度 SSE (Server-Sent Events) WebSocket 协议 HTTP/1.1 或 HTTP/2 独立协议(ws://) 方向 服务器到客户端(单向) 双向 自动重连 浏览器内置 需手动实现 代理兼容性 好(就是HTTP) 差(需特殊配置) 适合场景 LLM 流式输出(推荐) 实时对话/多Agent通信 实现复杂度 低 中 结论：LLM 流式输出首选 SSE，需要双向通信才用 WebSocket。 SSE 后端实现（FastAPI） from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx, json, asyncio app = FastAPI() LLM_API_URL = "https://api.openai.com/v1/chat/completions" @app.post("/api/chat/stream") async def chat_stream(request: Request): body = await request.json() async def event_generator(): headers = { "Authorization": f"Bearer {body['api_key']}", "Content-Type": "application/json" } payload = { "model": body["model"], "messages": body["messages"], "stream": True, "max_tokens": 4096 } try: async with httpx.AsyncClient(timeout=120) as client: async with client.stream("POST", LLM_API_URL, headers=headers, json=payload) as resp: async for line in resp.aiter_lines(): if not line.startswith("data: "): continue data = line[6:] if data.strip() == "[DONE]": yield "data: " + json.dumps({"type": "done"}) + "\n\n" break chunk = json.loads(data) delta = chunk["choices"][0]["delta"] if "content" in delta: token = delta["content"] yield "data: " + json.dumps({"type": "token", "content": token}) + "\n\n" except httpx.ReadTimeout: yield "data: " + json.dumps({"type": "error", "msg": "LLM timeout"}) + "\n\n" except Exception as e: yield "data: " + json.dumps({"type": "error", "msg": str(e)}) + "\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"} ) SSE 前端实现 class SSEClient { constructor(url) { this.url = url; this.controller = null; } async stream(messages, onToken, onDone, onError) { this.controller = new AbortController(); try { const resp = await fetch(this.url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages }), signal: this.controller.signal }); const reader = resp.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); const lines = buffer.split('\n'); buffer = lines.pop(); for (const line of lines) { if (!line.startsWith('data: ')) continue; const data = JSON.parse(line.slice(6)); if (data.type === 'token') onToken(data.content); else if (data.type === 'done') onDone(); else if (data.type === 'error') onError(data.msg); } } } catch (e) { if (e.name !== 'AbortError') onError(e.message); } } abort() { this.controller?.abort(); } } // 使用 const client = new SSEClient('/api/chat/stream'); let fullText = ''; await client.stream( [{ role: 'user', content: '解释 RAG 架构' }], (token) => { fullText += token; renderMarkdown(fullText); }, () => console.log('完成'), (err) => console.error('错误:', err) ); WebSocket 方案 需要双向通信（用户中途打断、实时修正）时使用： ...

为什么需要 Prompt 管理平台 当团队有 3 个以上 LLM 应用时，Prompt 管理就会失控：Prompt 散落在代码里、改一个字要重新部署、无法 A/B 测试、没有版本回滚。Prompt 管理平台把 Prompt 当作独立资产来管理。 核心功能需求 功能模块 说明 优先级 Prompt CRUD 创建/读取/更新/删除 Prompt P0 版本管理 每次修改生成新版本，支持回滚 P0 变量模板 支持 {{variable}} 变量插值 P0 A/B 测试 流量分割对比不同 Prompt 效果 P1 权限控制 团队成员角色与审批流 P1 效果监控 Prompt 调用的成功率/延迟/成本 P1 批量测试 用测试集自动评估 Prompt P2 架构设计 +----------------------------------------------+ | Prompt 管理平台架构 | | | | +------------+ +------------------+ | | | Web UI | | API Gateway | | | | (React) | | (FastAPI) | | | +-----+------+ +-------+----------+ | | | | | | +-----+-----------------+----------------+ | | | Prompt Service Layer | | | | CRUD | Version | AB Test | RBAC | | | +-------------------+--------------------+ | | | | | +---------+---------+---------+----------+ | | |PostgreSQL| Redis缓存 |S3存储 | | | |(元数据) | (热Prompt) |(备份) | | | +---------+--------------+----------+ | +----------------------------------------------+ 数据库设计 -- Prompt 主表 CREATE TABLE prompts ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(128) NOT NULL UNIQUE, description TEXT, category VARCHAR(64), current_version_id UUID REFERENCES prompt_versions(id), created_at TIMESTAMP DEFAULT NOW(), updated_at TIMESTAMP DEFAULT NOW() ); -- Prompt 版本表 CREATE TABLE prompt_versions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), prompt_id UUID NOT NULL REFERENCES prompts(id) ON DELETE CASCADE, version INT NOT NULL, content TEXT NOT NULL, variables JSONB DEFAULT '[]', model_config JSONB DEFAULT '{}', change_log TEXT, created_by UUID NOT NULL REFERENCES users(id), created_at TIMESTAMP DEFAULT NOW(), UNIQUE(prompt_id, version) ); -- A/B 测试表 CREATE TABLE ab_tests ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(128) NOT NULL, prompt_id UUID NOT NULL REFERENCES prompts(id), variant_a_version_id UUID NOT NULL REFERENCES prompt_versions(id), variant_b_version_id UUID NOT NULL REFERENCES prompt_versions(id), traffic_split JSONB DEFAULT '{"a": 50, "b": 50}', status VARCHAR(16) DEFAULT 'running', metrics JSONB DEFAULT '{}', created_at TIMESTAMP DEFAULT NOW(), ended_at TIMESTAMP ); 核心 API 实现 from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel import asyncpg, json app = FastAPI() class PromptCreate(BaseModel): name: str description: str = "" content: str variables: list[dict] = [] model_config: dict = {} class PromptUpdate(BaseModel): content: str change_log: str = "" variables: list[dict] = [] @app.post("/api/prompts") async def create_prompt(data: PromptCreate, db=Depends(get_db)): async with db.transaction(): prompt = await db.fetchrow( "INSERT INTO prompts (name, description) VALUES ($1, $2) RETURNING *", data.name, data.description ) version = await db.fetchrow( "INSERT INTO prompt_versions (prompt_id, version, content, variables, model_config, created_by) VALUES ($1, 1, $2, $3, $4, $5) RETURNING *", prompt["id"], data.content, json.dumps(data.variables), json.dumps(data.model_config), current_user_id ) await db.execute( "UPDATE prompts SET current_version_id = $1 WHERE id = $2", version["id"], prompt["id"] ) return {"prompt": dict(prompt), "version": dict(version)} @app.put("/api/prompts/{prompt_id}") async def update_prompt(prompt_id: str, data: PromptUpdate, db=Depends(get_db)): async with db.transaction(): current = await db.fetchrow( "SELECT version FROM prompt_versions WHERE prompt_id = $1 ORDER BY version DESC LIMIT 1", prompt_id ) new_version = (current["version"] + 1) if current else 1 version = await db.fetchrow( "INSERT INTO prompt_versions (prompt_id, version, content, variables, change_log, created_by) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *", prompt_id, new_version, data.content, json.dumps(data.variables), data.change_log, current_user_id ) await db.execute( "UPDATE prompts SET current_version_id = $1, updated_at = NOW() WHERE id = $2", version["id"], prompt_id ) return {"version": dict(version)} @app.post("/api/prompts/{prompt_id}/rollback/{version}") async def rollback_prompt(prompt_id: str, version: int, db=Depends(get_db)): target = await db.fetchrow( "SELECT id FROM prompt_versions WHERE prompt_id = $1 AND version = $2", prompt_id, version ) if not target: raise HTTPException(404, "Version not found") await db.execute( "UPDATE prompts SET current_version_id = $1 WHERE id = $2", target["id"], prompt_id ) return {"rolled_back_to": version} A/B 测试实现 import random class ABTestRouter: def __init__(self, db, redis): self.db = db self.redis = redis async def get_variant(self, test_id: str, user_id: str) -> str: cache_key = f"ab:{test_id}:{user_id}" cached = await self.redis.get(cache_key) if cached: return cached test = await self.db.fetchrow( "SELECT * FROM ab_tests WHERE id = $1 AND status = 'running'", test_id ) if not test: return "a" split = json.loads(test["traffic_split"]) variant = "a" if random.random() * 100 < split["a"] else "b" await self.redis.setex(cache_key, 86400, variant) return variant async def record_metric(self, test_id, variant, metric, value): await self.db.execute( "INSERT INTO ab_metrics (test_id, variant, metric, value) VALUES ($1, $2, $3, $4)", test_id, variant, metric, value ) 与现有工具对比 工具 优点 缺点 适用场景 LangSmith LangChain 生态集成好 绑定 LangChain，贵 已用 LangChain 的团队 PromptHub UI 好用，有协作功能 自定义能力弱 非技术团队 自建平台 完全可控，可定制 开发成本高 大团队/特殊需求 建议：团队 < 5 人且用 LangChain，直接上 LangSmith。团队 > 10 人或有特殊需求，自建平台 ROI 更高。 ...

生产级 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 中小规模+低延迟 选型建议： ...

多语言 LLM 的核心挑战 部署一个支持多语言的 LLM 应用不是简单地把 prompt 翻译成不同语言。真正的挑战在于： Embedding 跨语言对齐：中文问题和英文文档之间的语义匹配 文化适配：同一个概念在不同文化语境下含义不同 混合语言输入：用户一句话中混用中英文（“帮我看看这个 function 的 implementation”） 成本控制：多语言模型通常更大更贵 语言检测与路由 第一步是准确检测用户输入的语言，用于后续路由： import langdetect from langdetect import detect, DetectorFactory # 设置种子保证检测结果稳定 DetectorFactory.seed = 42 def detect_language(text: str) -> str: """检测文本语言，返回 ISO 639-1 代码""" # 预处理：去除代码块、URL 等干扰内容 import re clean = re.sub(r'```.*?```', '', text, flags=re.DOTALL) clean = re.sub(r'https?://\S+', '', clean) if len(clean.strip()) < 10: return "en" # 太短，默认英文 try: lang = detect(clean) # 映射到支持的语言集合 supported = {"zh-cn", "zh-tw", "en", "ja", "ko", "fr", "de", "es"} if lang in supported: return lang elif lang.startswith("zh"): return "zh-cn" else: return "en" # 不支持的语言回退到英文 except: return "en" def route_by_language(lang: str, config: dict) -> dict: """根据语言路由到不同的模型/配置""" routing = { "zh-cn": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "zh"}, "zh-tw": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "zh-tw"}, "en": {"model": "llama-3.1-70b", "embedding": "bge-m3", "system_lang": "en"}, "ja": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "ja"}, "default": {"model": "llama-3.1-70b", "embedding": "bge-m3", "system_lang": "en"} } return routing.get(lang, routing["default"]) 多语言 Embedding 选型 模型 维度 支持语言 MTEB 均分 备注 bge-m3 1024 100+ 66.3 推荐，多语言SOTA multilingual-e5-large 1024 100+ 64.8 稳定可靠 paraphrase-multilingual 768 50+ 61.5 轻量级 OpenAI text-embedding-3-large 3072 100+ 65.0 贵但方便 Cohere embed-multilingual-v3 1024 100+ 64.9 API调用 结论：自部署用 bge-m3，API 调用用 OpenAI 或 Cohere。bge-m3 在中英跨语言检索任务上表现最好。 ...

需求分析：客服系统的核心指标 构建 AI 客服系统前，先明确要解决什么问题。客服系统的价值体现在三个维度： 维度 指标 目标值 衡量方式 效率 自动解决率 ≥60% 无需人工介入的会话占比 效率 平均响应时间 <2s 用户发送到首字的时间 质量 回答准确率 ≥90% 人工抽检评分 质量 用户满意度 ≥4.0/5 会话后评分 成本 单会话成本 <¥0.5 API 调用 + 基础设施 / 会话数 体验 人工转接等待 <30s 转人工后接通时间 客服场景分类 不同场景需要不同的处理策略： from enum import Enum from dataclasses import dataclass class QueryType(Enum): """客服查询类型""" FAQ = "faq" # 常见问题（退货政策、运费说明） TROUBLESHOOT = "trouble" # 故障排查（设备不工作、登录失败） TRANSACTION = "transaction" # 交易查询（订单状态、退款进度） COMPLAINT = "complaint" # 投诉建议 CHITCHAT = "chitchat" # 闲聊（超出业务范围） @dataclass class QueryStrategy: """不同查询类型的处理策略""" query_type: QueryType needs_rag: bool needs_tools: bool max_turns: int can_self_serve: bool escalation_threshold: float # 置信度低于此值则转人工 STRATEGIES = { QueryType.FAQ: QueryStrategy( query_type=QueryType.FAQ, needs_rag=True, needs_tools=False, max_turns=3, can_self_serve=True, escalation_threshold=0.6 ), QueryType.TROUBLESHOOT: QueryStrategy( query_type=QueryType.TROUBLESHOOT, needs_rag=True, needs_tools=True, max_turns=8, can_self_serve=True, escalation_threshold=0.5 ), QueryType.TRANSACTION: QueryStrategy( query_type=QueryType.TRANSACTION, needs_rag=False, needs_tools=True, max_turns=5, can_self_serve=True, escalation_threshold=0.7 ), QueryType.COMPLAINT: QueryStrategy( query_type=QueryType.COMPLAINT, needs_rag=False, needs_tools=False, max_turns=2, can_self_serve=False, escalation_threshold=0.8 ), QueryType.CHITCHAT: QueryStrategy( query_type=QueryType.CHITCHAT, needs_rag=False, needs_tools=False, max_turns=2, can_self_serve=True, escalation_threshold=0.9 ), } 知识库构建 知识库是 AI 客服的核心。垃圾进、垃圾出——知识库的质量直接决定回答质量。 ...

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

为什么选 vLLM Ollama 适合本地开发，但生产环境需要高吞吐：vLLM 是目前最快的开源 LLM 推理引擎。 引擎 吞吐量 并发 显存利用 适用场景 Ollama 1x 低 中 本地开发 vLLM 5-10x 高 极高 生产部署 TGI 3-5x 高 高 生产部署 TensorRT-LLM 8-12x 高 极高 极致性能 核心技术：PagedAttention 传统 KV Cache: ┌──────────────────────────────────┐ │ Request A: [████████░░░░░░░░░░░] │ 预分配，大量浪费 │ Request B: [██████████████░░░░░] │ │ Request C: [██░░░░░░░░░░░░░░░░░] │ └──────────────────────────────────┘ 显存利用率：~40% PagedAttention: ┌──┬──┬──┬──┬──┬──┬──┬──┬──┬──┐ │A1│A2│B1│B2│B3│C1│A3│B4│A4│B5│ 按需分配，零浪费 └──┴──┴──┴──┴──┴──┴──┴──┴──┴──┘ 显存利用率：~95% 快速部署 Docker 部署 docker run --gpus all \ -v /models:/models \ -p 8000:8000 \ --ipc=host \ vllm/vllm-openai:latest \ --model /models/Qwen3-7B-Instruct \ --tensor-parallel-size 1 \ --max-model-len 8192 \ --gpu-memory-utilization 0.9 Python 部署 from vllm import LLM, SamplingParams llm = LLM( model="/models/Qwen3-7B-Instruct", tensor_parallel_size=1, # GPU 数量 gpu_memory_utilization=0.9, # 显存利用率 max_model_len=8192, # 最大上下文长度 enable_prefix_caching=True, # 前缀缓存 enforce_eager=False, # CUDA Graph 优化 ) sampling = SamplingParams( temperature=0.7, top_p=0.9, max_tokens=1024, ) # 批量推理 outputs = llm.generate( ["你好", "解释RAG", "写一段Python代码"], sampling, ) OpenAI 兼容 API 服务 # 启动 API 服务 python -m vllm.entrypoints.openai.api_server \ --model /models/Qwen3-7B-Instruct \ --port 8000 \ --tensor-parallel-size 1 \ --max-model-len 8192 \ --gpu-memory-utilization 0.9 \ --enable-prefix-caching \ --served-model-name qwen3-7b # 客户端调用（与 OpenAI SDK 完全兼容） from openai import OpenAI client = OpenAI(base_url="http://localhost:8000/v1", api_key="vllm") response = client.chat.completions.create( model="qwen3-7b", messages=[{"role": "user", "content": "你好"}], stream=True, ) 性能调优 1. 批处理配置 llm = LLM( model="/models/Qwen3-7B-Instruct", # 批处理 max_num_seqs=256, # 最大并发序列数 max_num_batched_tokens=8192, # 每批最大 token 数 # KV Cache gpu_memory_utilization=0.9, # 显存利用率（0.8-0.95） swap_space=4, # CPU 交换空间 (GB) # 量化 quantization="awq", # AWQ 量化（省 50% 显存） ) 2. 前缀缓存 # 启用前缀缓存：相同 system prompt 的请求共享 KV Cache llm = LLM( model="/models/Qwen3-7B-Instruct", enable_prefix_caching=True, ) # 效果： # 第一个请求：1.2s（生成 KV Cache） # 后续相同 system prompt 的请求：0.3s（复用 KV Cache） 3. Speculative Decoding # 用小模型猜，大模型验 llm = LLM( model="/models/Qwen3-7B-Instruct", speculative_model="/models/Qwen3-0.5B", # 草稿模型 num_speculative_tokens=5, # 每轮猜 5 个 token ) # 效果：吞吐量提升 1.5-2x # 原理：小模型快速生成 5 个候选 token，大模型一次性验证 4. 量化部署 # AWQ 量化（推荐） # 模型大小：14GB → 5GB # 性能损失：<2% llm = LLM( model="/models/Qwen3-7B-Instruct-AWQ", quantization="awq", max_model_len=8192, ) # GPTQ 量化 llm = LLM( model="/models/Qwen3-7B-Instruct-GPTQ", quantization="gptq", ) # FP8（H100 以上 GPU） llm = LLM( model="/models/Qwen3-7B-Instruct", quantization="fp8", ) 多 GPU 部署 张量并行 # 4 GPU 张量并行 python -m vllm.entrypoints.openai.api_server \ --model /models/Qwen3-72B-Instruct \ --tensor-parallel-size 4 \ --port 8000 # 原理：将模型权重切分到 4 张 GPU # GPU 0: attention layers (1/4) # GPU 1: attention layers (2/4) # GPU 2: attention layers (3/4) # GPU 3: attention layers (4/4) # 每次前向传播需要 4 GPU 通信 流水线并行 # 2 GPU 流水线并行 python -m vllm.entrypoints.openai.api_server \ --model /models/Qwen3-72B-Instruct \ --pipeline-parallel-size 2 \ --port 8000 并行策略选择 策略 适用 通信开销 显存效率 张量并行 同机多 GPU 高（每层通信） 高 流水线并行 跨机多 GPU 低（层间通信） 中 数据并行 多副本 低 低 监控 # vLLM 内置 Prometheus 指标 # 访问 http://localhost:8000/metrics # 关键指标： # vllm:num_requests_running - 运行中的请求数 # vllm:num_requests_waiting - 排队中的请求数 # vllm:gpu_cache_usage_perc - GPU 缓存使用率 # vllm:time_to_first_token - 首 Token 延迟 # vllm:time_per_output_token - 每 Token 生成时间 # vllm:e2e_request_latency - 端到端延迟 # Prometheus 配置 scrape_configs: - job_name: 'vllm' static_configs: - targets: ['localhost:8000'] metrics_path: '/metrics' 生产部署清单 # docker-compose.yml version: '3.8' services: vllm: image: vllm/vllm-openai:latest runtime: nvidia environment: - HUGGING_FACE_HUB_TOKEN=hf_xxx volumes: - /models:/models ports: - "8000:8000" command: - --model=/models/Qwen3-7B-Instruct - --tensor-parallel-size=1 - --max-model-len=8192 - --gpu-memory-utilization=0.9 - --enable-prefix-caching - --served-model-name=qwen3-7b - --uvicorn-log-level=info healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8000/health"] interval: 30s timeout: 10s retries: 3 deploy: resources: reservations: devices: - capabilities: ["gpu"] 成本对比 以 Qwen3-7B 为例，处理 100 万 Token： ...

为什么必须流式 AI 生成一个完整回答可能需要 5-30 秒。如果等全部生成完再返回，用户体验极差。流式输出让用户看到"AI 正在打字"，将感知等待时间从 15 秒降到 0.5 秒。 SSE vs WebSocket 维度 SSE WebSocket 协议 HTTP 独立协议 方向 服务端→客户端（单向） 双向 重连 自动 需手动实现 兼容性 所有现代浏览器 所有现代浏览器 复杂度 低 中 适用场景 AI 回复流式输出 实时协作、语音对话 结论：AI 对话场景用 SSE 足够，语音/多模态交互用 WebSocket。 SSE 实现 服务端（Python FastAPI） from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() @app.post("/api/chat") async def chat(request: dict): async def stream(): # 调用 LLM 流式 API async for chunk in llm.stream( model="gpt-4o", messages=request["messages"] ): # SSE 格式：data: <content>\n\n content = chunk.choices[0].delta.content if content: yield f"data: {json.dumps({'content': content})}\n\n" # 发送结束标记 yield f"data: {json.dumps({'done': True})}\n\n" return StreamingResponse( stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # Nginx 不缓冲 } ) 客户端（JavaScript） class AIChat { async sendMessage(text) { const response = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages: [...this.history, { role: 'user', content: text }] }) }); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // 解析 SSE 数据 const lines = buffer.split('\n\n'); buffer = lines.pop(); // 保留不完整的行 for (const line of lines) { if (line.startsWith('data: ')) { const data = JSON.parse(line.slice(6)); if (data.done) { this.onComplete(); } else { this.onToken(data.content); } } } } } onToken(content) { document.getElementById('response').textContent += content; } } 进阶：流式 + 工具调用 当 Agent 需要调用工具时，流式输出变得复杂： ...

一个真实案例 我们运营一个客服 Agent，日均处理 5000 次对话。优化前月成本约 $3,200，优化后降至 $680——降本 78%。以下是完整的优化路径。 成本构成分析 优化前月成本分布： ├── LLM API 调用 72% ($2,304) │ ├── 主模型 (GPT-4o) 58% │ ├── 嵌入模型 8% │ └── 审核模型 6% ├── 向量数据库 12% ($384) ├── 服务器/带宽 8% ($256) └── 监控/日志 8% ($256) 第一层：模型分级路由 不是所有请求都需要最强模型： class ModelRouter: def __init__(self): self.routes = { "simple_qa": { "model": "gpt-4o-mini", "criteria": lambda q: len(q) < 50 and not q.requires_tools }, "standard": { "model": "claude-3.5-sonnet", "criteria": lambda q: q.complexity < 5 }, "complex": { "model": "gpt-4o", "criteria": lambda q: True # fallback } } def route(self, query): for level, config in self.routes.items(): if config["criteria"](query): return config["model"] return "gpt-4o" # 效果：62% 的请求被路由到 mini 模型 # 月节省：$1,180 第二层：语义缓存 相似问题命中缓存，避免重复调用 LLM： ...

不同 Agent 任务该选哪个 LLM？本文从推理能力、工具调用、成本、上下文窗口等维度全面对比 2026 年主流大模型。