实践指南

AI 应用为什么需要专门的监控 传统 APM（应用性能监控）关注的是 CPU、内存、响应时间和错误率。但 AI 应用引入了全新的维度： Token 维度：每次请求消耗的 Token 数直接影响成本 模型质量：响应不仅要快，还要准确、安全、无幻觉 长尾延迟：流式输出可能持续数分钟，P99 指标失真 多模型路由：请求分散在不同模型上，需要统一视图 如果你的监控面板还只有"QPS + 延迟 + 错误率"三件套，你实际上是在"盲飞"。 监控指标体系 四层指标架构 ┌─────────────────────────────────────────────┐ │ 业务指标层 (Business) │ │ 任务完成率 · 用户满意度 · 成本/请求 │ ├─────────────────────────────────────────────┤ │ AI 质量指标层 (Quality) │ │ 幻觉率 · 意图准确率 · 安全违规率 │ ├─────────────────────────────────────────────┤ │ 模型性能指标层 (Model) │ │ 首 Token 延迟 · 生成速率 · KV Cache 命中率 │ ├─────────────────────────────────────────────┤ │ 基础设施指标层 (Infra) │ │ GPU 利用率 · 显存使用 · CPU/内存 · 网络 │ └─────────────────────────────────────────────┘ 完整指标清单 层级 指标名 类型 说明 告警阈值 Infra gpu_utilization Gauge GPU 计算利用率 <20% 或 >95% Infra gpu_memory_used_ratio Gauge GPU 显存使用率 >0.95 Infra gpu_temperature Gauge GPU 温度 (°C) >85 Model ttft_seconds Histogram 首 Token 延迟 (TTFT) P95 >2s Model tokens_per_second Gauge Token 生成速率 <30 tokens/s Model kv_cache_hit_rate Gauge KV Cache 命中率 <0.3 Model active_sequences Gauge 活跃推理序列数 >max_seqs×0.9 Model queue_depth Gauge 排队请求数 >10 Quality hallucination_score Gauge 幻觉评分 (0-1) >0.15 Quality safety_violation_count Counter 安全违规次数 >0/h Quality user_feedback_score Histogram 用户评分 (1-5) 均值 <3.5 Business cost_per_request Gauge 单请求成本 ($) >$0.05 Business task_completion_rate Gauge 任务完成率 <80% Business daily_spend Counter 日累计花费 >预算 80% Prometheus 指标采集 应用侧指标暴露 from prometheus_client import Counter, Histogram, Gauge, generate_latest from prometheus_client import CollectorRegistry, CONTENT_TYPE_LATEST from fastapi import FastAPI, Response import time app = FastAPI() registry = CollectorRegistry() # === 基础设施指标 === gpu_utilization = Gauge( "ai_gpu_utilization", "GPU utilization percentage", ["gpu_id", "model_name"], registry=registry ) gpu_memory_used = Gauge( "ai_gpu_memory_used_bytes", "GPU memory used in bytes", ["gpu_id"], registry=registry ) # === 模型性能指标 === request_duration = Histogram( "ai_request_duration_seconds", "Total request duration", ["model_name", "endpoint"], buckets=[0.5, 1, 2, 5, 10, 30, 60, 120, 300], registry=registry, ) ttft = Histogram( "ai_time_to_first_token_seconds", "Time to first token", ["model_name"], buckets=[0.1, 0.25, 0.5, 1, 2, 5, 10], registry=registry, ) tokens_generated = Counter( "ai_tokens_generated_total", "Total tokens generated", ["model_name", "type"], # type: input|output registry=registry, ) # === 质量指标 === hallucination_score = Gauge( "ai_hallucination_score", "Hallucination score (0=good, 1=bad)", ["model_name"], registry=registry, ) safety_violations = Counter( "ai_safety_violations_total", "Safety violation count", ["model_name", "violation_type"], registry=registry, ) # === 业务指标 === request_cost = Gauge( "ai_request_cost_usd", "Cost per request in USD", ["model_name"], registry=registry, ) daily_spend = Counter( "ai_daily_spend_usd", "Daily total spend in USD", ["model_name"], registry=registry, ) # === 中间件：自动采集请求指标 === @app.middleware("http") async def metrics_middleware(request, call_next): start_time = time.time() response = await call_next(request) duration = time.time() - start_time model = request.headers.get("X-Model-Name", "unknown") request_duration.labels( model_name=model, endpoint=request.url.path, ).observe(duration) return response # === 业务逻辑中手动埋点 === async def chat_completion(messages, model="gpt-4o-mini"): start = time.time() # 调用 LLM response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, stream_options={"include_usage": True}, ) first_token_time = None total_output_tokens = 0 async for chunk in response: if first_token_time is None and chunk.choices[0].delta.content: first_token_time = time.time() ttft.labels(model_name=model).observe(first_token_time - start) if chunk.usage: total_output_tokens = chunk.usage.completion_tokens input_tokens = chunk.usage.prompt_tokens tokens_generated.labels(model_name=model, type="input").inc(input_tokens) tokens_generated.labels(model_name=model, type="output").inc(total_output_tokens) # 计算成本 cost = calculate_cost(model, input_tokens, total_output_tokens) request_cost.labels(model_name=model).set(cost) daily_spend.labels(model_name=model).inc(cost) return response # === 异步质量评估 === async def evaluate_quality(response: str, context: str, model: str): """异步评估响应质量""" # 幻觉检测 score = await hallucination_detector.check(response, context) hallucination_score.labels(model_name=model).set(score) # 安全检查 violations = safety_checker.check(response) for v_type, count in violations.items(): safety_violations.labels( model_name=model, violation_type=v_type, ).inc(count) # === Prometheus 指标暴露端点 === @app.get("/metrics") async def metrics(): return Response( generate_latest(registry), media_type=CONTENT_TYPE_LATEST, ) GPU 指标采集（DCGM Exporter） # gpu-metrics-stack.yaml apiVersion: v1 kind: Namespace metadata: name: gpu-monitoring --- apiVersion: apps/v1 kind: DaemonSet metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter template: metadata: labels: app: dcgm-exporter spec: nodeSelector: accelerator: nvidia-t4 # 调度到 GPU 节点 tolerations: - key: nvidia.com/gpu operator: Exists containers: - name: dcgm-exporter image: nvcr.io/nvidia/k8s/dcgm-exporter:3.3.0-ubuntu22.04 ports: - containerPort: 9400 name: metrics env: - name: DCGM_EXPORTER_LISTEN value: ":9400" - name: DCGM_EXPORTER_KUBERNETES value: "true" securityContext: capabilities: add: ["SYS_ADMIN"] --- # ServiceMonitor 让 Prometheus 自动发现 apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter endpoints: - port: metrics path: "/metrics" interval: 10s DCGM 关键 GPU 指标 指标 说明 用途 DCGM_FI_DEV_GPU_UTIL GPU 计算利用率 (%) 判断是否充分使用 GPU DCGM_FI_DEV_MEM_COPY_UTIL 显存带宽利用率 (%) 判断是否显存带宽瓶颈 DCGM_FI_DEV_FB_USED 已用显存 (MB) 判断显存是否够用 DCGM_FI_DEV_FB_FREE 空闲显存 (MB) 可调度的余量 DCGM_FI_DEV_GPU_TEMP GPU 温度 (°C) 散热健康 DCGM_FI_DEV_POWER_USAGE 功耗 (W) 能耗成本 DCGM_FI_PROF_PIPE_TENSOR_ACTIVE Tensor Core 利用率 推理效率 Grafana 仪表盘构建 仪表盘布局设计 ┌──────────────────────────────────────────────────────────┐ │ AI Application Overview │ ├────────────┬────────────┬────────────┬───────────────────┤ │ QPS │ P95 TTFT │ Error Rate│ Daily Spend │ │ (Stat) │ (Stat) │ (Stat) │ (Stat) │ ├────────────┴────────────┼────────────┴───────────────────┤ │ Request Latency │ Token Throughput │ │ (Time Series) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ GPU Utilization │ GPU Memory Usage │ │ (Time Series, per GPU) │ (Time Series, per GPU) │ ├──────────────────────────┼───────────────────────────────┤ │ Model Distribution │ Queue Depth │ │ (Pie Chart) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ Hallucination Score │ Safety Violations │ │ (Gauge + Time Series) │ (Alert Table) │ ├──────────────────────────┴───────────────────────────────┤ │ Request Logs (Table) │ └──────────────────────────────────────────────────────────┘ Grafana Dashboard JSON { "dashboard": { "title": "AI Application Monitor", "tags": ["ai", "llm", "production"], "timezone": "browser", "refresh": "10s", "panels": [ { "id": 1, "title": "Requests Per Second", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 0, "y": 0}, "targets": [{ "expr": "sum(rate(ai_request_duration_seconds_count[1m]))", "legendFormat": "QPS" }], "fieldConfig": { "defaults": { "thresholds": { "steps": [ {"color": "red", "value": null}, {"color": "yellow", "value": 10}, {"color": "green", "value": 50} ] } } } }, { "id": 2, "title": "P95 Time to First Token", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 6, "y": 0}, "targets": [{ "expr": "histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le))", "legendFormat": "P95 TTFT" }], "fieldConfig": { "defaults": { "unit": "s", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 1}, {"color": "red", "value": 2} ] } } } }, { "id": 3, "title": "Request Latency Distribution", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 0, "y": 4}, "targets": [ { "expr": "histogram_quantile(0.50, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P50 {{model_name}}" }, { "expr": "histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P95 {{model_name}}" }, { "expr": "histogram_quantile(0.99, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P99 {{model_name}}" } ], "fieldConfig": { "defaults": {"unit": "s"} } }, { "id": 4, "title": "GPU Utilization & Memory", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 12, "y": 4}, "targets": [ { "expr": "DCGM_FI_DEV_GPU_UTIL", "legendFormat": "GPU {{gpu}} Util %" }, { "expr": "DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL * 100", "legendFormat": "GPU {{gpu}} Mem %" } ], "fieldConfig": { "defaults": { "min": 0, "max": 100, "custom": {"fillOpacity": 10} } } }, { "id": 5, "title": "Daily Spend by Model", "type": "bargauge", "gridPos": {"h": 6, "w": 12, "x": 0, "y": 12}, "targets": [{ "expr": "sum by (model_name) (increase(ai_daily_spend_usd_total[24h]))", "legendFormat": "{{model_name}}" }], "fieldConfig": { "defaults": { "unit": "currencyUSD", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 50}, {"color": "red", "value": 100} ] } } } }, { "id": 6, "title": "Hallucination Score Trend", "type": "timeseries", "gridPos": {"h": 6, "w": 12, "x": 12, "y": 12}, "targets": [{ "expr": "avg_over_time(ai_hallucination_score[10m])", "legendFormat": "Hallucination Score" }], "fieldConfig": { "defaults": { "min": 0, "max": 1, "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 0.1}, {"color": "red", "value": 0.2} ] } } } } ] } } 告警体系 多级告警规则 # alerting-rules.yaml groups: - name: ai_infra_alerts interval: 30s rules: # === 基础设施告警 === - alert: GPUHighTemperature expr: DCGM_FI_DEV_GPU_TEMP > 85 for: 5m labels: severity: critical team: infra annotations: summary: "GPU 温度过高: {{ $value }}°C" description: "GPU {{ $labels.gpu }} on {{ $labels.instance }} 温度超过 85°C" - alert: GPUHighMemoryUsage expr: DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL > 0.95 for: 2m labels: severity: warning team: infra annotations: summary: "GPU 显存使用率 >95%" - alert: GPULowUtilization expr: avg_over_time(DCGM_FI_DEV_GPU_UTIL[10m]) < 10 for: 10m labels: severity: info team: infra annotations: summary: "GPU 利用率过低 (<10%)，可能资源浪费" - name: ai_model_alerts interval: 15s rules: # === 模型性能告警 === - alert: HighTTFT expr: | histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le, model_name) ) > 2 for: 5m labels: severity: warning team: ai annotations: summary: "首 Token 延迟 P95 > 2s ({{ $labels.model_name }})" - alert: HighErrorRate expr: | sum(rate(ai_request_duration_seconds_count{status="error"}[5m])) by (model_name) / sum(rate(ai_request_duration_seconds_count[5m])) by (model_name) > 0.05 for: 3m labels: severity: critical team: ai annotations: summary: "错误率 >5% ({{ $labels.model_name }})" - alert: ModelQueueBacklog expr: ai_queue_depth > 20 for: 2m labels: severity: warning team: ai annotations: summary: "请求排队 >20 ({{ $labels.model_name }})" - name: ai_quality_alerts interval: 60s rules: # === 质量告警 === - alert: HighHallucinationRate expr: avg_over_time(ai_hallucination_score[15m]) > 0.15 for: 10m labels: severity: critical team: ai annotations: summary: "幻觉评分 >0.15 持续 10 分钟" - alert: SafetyViolationDetected expr: increase(ai_safety_violations_total[1h]) > 0 for: 0m labels: severity: critical team: security annotations: summary: "检测到安全违规 ({{ $labels.violation_type }})" - name: ai_business_alerts interval: 60s rules: # === 业务告警 === - alert: DailyBudgetExceeded expr: sum(increase(ai_daily_spend_usd_total[24h])) > 500 for: 1m labels: severity: critical team: business annotations: summary: "日预算超出 (${{ $value }})" - alert: CostPerRequestSpike expr: ai_request_cost_usd > 0.10 for: 5m labels: severity: warning team: business annotations: summary: "单请求成本 >$0.10 ({{ $labels.model_name }})" 告警通知渠道配置 # alertmanager-config.yaml apiVersion: v1 kind: ConfigMap metadata: name: alertmanager-config namespace: monitoring data: config.yml: | route: group_by: ["alertname", "model_name"] group_wait: 30s group_interval: 5m repeat_interval: 4h receiver: "default" routes: - matchers: ["severity=critical"] receiver: "critical" group_wait: 10s repeat_interval: 1h - matchers: ["team=security"] receiver: "security" group_wait: 0s receivers: - name: "default" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-alerts" send_resolved: true - name: "critical" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-critical" send_resolved: true pagerduty_configs: - routing_key: "..." severity: critical - name: "security" webhook_configs: - url: "https://internal.security.com/ai-alert-webhook" 告警分级与响应 级别 响应时间 通知渠道 升级策略 示例 P0 Critical 5 分钟 PagerDuty + 电话 15分钟升级到团队负责人 安全违规、服务宕机 P1 Warning 30 分钟 Slack @channel 2小时升级 延迟超标、队列积压 P2 Info 4 小时 Slack 普通消息 无升级 GPU 利用率低 P3 Notice 24 小时 邮件 无升级 日花费趋势异常 日志聚合与链路追踪 结构化日志标准 import structlog import json logger = structlog.get_logger() async def handle_chat_request(request: ChatRequest): """带全链路日志的请求处理""" request_id = generate_request_id() logger.info("chat_request_start", request_id=request_id, user_id=request.user_id, model=request.model, message_length=len(request.message), conversation_id=request.conversation_id, ) try: # 模型路由 routed_model = router.route(request.message) logger.info("model_routed", request_id=request_id, original_model=request.model, routed_model=routed_model, ) # RAG 检索 if needs_retrieval(request.message): retrieved = await retriever.search(request.message) logger.info("rag_retrieval_complete", request_id=request_id, chunks_retrieved=len(retrieved), top_score=retrieved[0]["score"] if retrieved else 0, ) # LLM 调用 response = await llm_client.chat.completions.create(...) logger.info("chat_request_complete", request_id=request_id, model=routed_model, input_tokens=response.usage.prompt_tokens, output_tokens=response.usage.completion_tokens, ttft_ms=first_token_latency, total_latency_ms=total_latency, cost_usd=cost, ) return response except Exception as e: logger.error("chat_request_failed", request_id=request_id, error_type=type(e).__name__, error_message=str(e), ) raise 成本异常检测 class CostAnomalyDetector: """基于统计方法的成本异常检测""" def __init__(self, window_size: int = 168): # 7 天小时数据 self.window_size = window_size self.cost_history: list[float] = [] def update(self, hourly_cost: float) -> dict | None: self.cost_history.append(hourly_cost) if len(self.cost_history) < self.window_size: return None # 使用滑动窗口统计 recent = self.cost_history[-self.window_size:] mean = sum(recent) / len(recent) variance = sum((x - mean) ** 2 for x in recent) / len(recent) std = variance ** 0.5 # Z-score 异常检测 z_score = abs(hourly_cost - mean) / max(std, 0.01) if z_score > 3: return { "type": "cost_spike", "current_cost": hourly_cost, "expected_mean": mean, "z_score": z_score, "deviation_pct": (hourly_cost - mean) / mean * 100, } return None 结语 AI 应用监控的核心理念是从基础设施到业务价值的全链路可观测。GPU 利用率告诉你硬件是否健康，TTFT 告诉你用户是否在等待，幻觉率告诉你 AI 是否可信，成本指标告诉你系统是否可持续。 ...

为什么 LLM 需要专门的 K8s 部署方案 LLM 推理服务与传统 Web 服务有本质区别： 显存约束：7B 模型需要 ~14GB 显存，70B 模型需要 ~140GB 显存，且显存是最大的资源瓶颈 冷启动慢：模型加载到 GPU 需要 30-120 秒 请求特性：单次请求可能持续 30-300 秒（流式输出），与 HTTP 常规超时机制冲突 异构硬件：不同模型需要不同 GPU 类型（推理用 T4/A10，训练用 A100/H100） 这些特性决定了标准 K8s 部署方式（HPA + 滚动更新）并不适用。 整体架构 ┌─────────────────────────────────────────────────────┐ │ Ingress / Gateway │ ├─────────────────────────────────────────────────────┤ │ LLM Gateway (路由/限流) │ │ (LiteLLM / APISIX / Higress) │ ├──────────────┬──────────────┬──────────────────────┤ │ GPU Pool 1 │ GPU Pool 2 │ CPU Pool (兜底) │ │ (7B Models) │ (70B Models) │ (小模型/重写) │ │ T4/A10×N │ A100×N │ gpt-4o-mini proxy │ ├──────────────┴──────────────┴──────────────────────┤ │ GPU Operator + NVIDIA Device Plugin │ ├─────────────────────────────────────────────────────┤ │ K8s Control Plane │ └─────────────────────────────────────────────────────┘ GPU 节点池配置 节点池规划 池名称 GPU 型号 显存 用途 节点数 单节点副本数 gpu-small T4 (16GB) 16GB 7B 以下模型 3 2 gpu-medium A10 (24GB) 24GB 7B-14B 模型 2 1 gpu-large A100 (80GB) 80GB 70B 模型 2 1 cpu-fallback 无 - 预处理/重写 5 10 GPU 节点 Label 与 Taint 配置 # GPU 节点打标签 apiVersion: v1 kind: Node metadata: name: gpu-node-1 labels: accelerator: nvidia-t4 gpu.memory: "16Gi" node.kubernetes.io/instance-type: "g4dn.xlarge" pool: gpu-small spec: {} --- # 专用 GPU 节点设置 Taint（防止非 GPU Pod 调度上去） apiVersion: v1 kind: Node metadata: name: gpu-node-1 spec: taints: - key: nvidia.com/gpu value: "true" effect: NoSchedule - key: pool value: gpu-small effect: NoSchedule NVIDIA GPU Operator 部署 # 安装 NVIDIA GPU Operator helm repo add nvidia https://helm.ngc.nvidia.com/nvidia helm repo update helm install gpu-operator nvidia/gpu-operator \ --namespace gpu-operator --create-namespace \ --set driver.enabled=true \ --set toolkit.enabled=true \ --set devicePlugin.enabled=true \ --set dcgmExporter.enabled=true \ --set nodeStatusExporter.enabled=true # 验证 GPU 可用 kubectl get nodes -o wide kubectl describe node gpu-node-1 | grep nvidia.com/gpu 模型服务部署 vLLM 推理服务部署 vLLM 是目前性能最好的开源 LLM 推理框架，支持 PagedAttention、连续批处理和 Tensor Parallel。 ...

引言 当 LLM 应用从原型走向生产，成本往往是继效果之后的第二大瓶颈。一个日活 10 万的对话应用，如果不做任何优化，每月 API 费用可以轻松突破 5 万美元。本文总结 10 种经过实战验证的成本优化方法，覆盖从模型层到架构层的完整链路。 方法一：模型分级路由 不同复杂度的请求使用不同级别的模型。简单问题用小模型，复杂问题才用大模型。 import os from typing import Dict, Any class ModelRouter: """根据请求复杂度路由到不同模型""" def __init__(self): self.routes = { "simple": { "model": "gpt-4o-mini", "max_tokens": 512, "cost_per_1k": 0.00015, # input $0.15/1M }, "medium": { "model": "gpt-4o", "max_tokens": 1024, "cost_per_1k": 0.0025, }, "complex": { "model": "o3", "max_tokens": 4096, "cost_per_1k": 0.015, }, } def classify(self, query: str) -> str: """基于规则快速分类""" query_lower = query.lower() # 简单查询：问候、翻译短句、简单事实 if any(kw in query_lower for kw in ["你好", "翻译", "what is", "hi ", "hello"]): return "simple" # 复杂查询：代码生成、多步推理、长文写作 if any(kw in query_lower for kw in ["写代码", "分析", "设计", "架构", "debug"]): return "complex" # 默认中等 return "medium" def route(self, query: str) -> Dict[str, Any]: level = self.classify(query) return self.routes[level] # 使用示例 router = ModelRouter() config = router.route("帮我写一个 Python 排序算法") print(f"路由到: {config['model']}, 预估成本: ${config['cost_per_1k'] * 2:.6f}/1K tokens") 路由效果量化 查询类型 占比 原方案 分级路由后 节省 简单问答 50% $0.0025/1K $0.00015/1K 94% 中等对话 35% $0.0025/1K $0.0025/1K 0% 复杂推理 15% $0.0025/1K $0.015/1K -500% 加权总计 100% $0.0025/1K $0.0021/1K 16% 关键洞察：虽然复杂查询费用上升，但简单查询的大幅节省使总成本下降。实际场景中简单查询占比往往超过 60%，节省可达 30-50%。 ...

为什么需要 Prompt 工程化 在原型阶段，Prompt 通常是一个写在代码里的字符串常量。但当应用走向生产，问题开始浮现： 改了一个词，线上效果突然变差，却不知道回退到哪个版本 A/B 测试靠手动切换环境变量，数据散落在日志文件里 新来的同事改了 Prompt，破坏了之前精心设计的 Few-shot 格式 同一个功能有 5 个 Prompt 变体，没人知道哪个在跑 Prompt 工程化的核心目标：让 Prompt 成为可追踪、可测试、可回滚的一等公民。 Prompt 版本管理 目录结构设计 prompts/ ├── config.yaml # 全局配置 ├── chatbot/ # 功能模块 │ ├── meta.yaml # 模块元数据 │ ├── v1.0.0/ # 语义化版本 │ │ ├── system.txt # 系统提示 │ │ ├── few_shot.jsonl # Few-shot 示例 │ │ └── config.yaml # 模型参数 │ ├── v1.1.0/ │ │ ├── system.txt │ │ ├── few_shot.jsonl │ │ └── config.yaml │ └── v2.0.0/ # 大版本变更 │ ├── system.txt │ ├── few_shot.jsonl │ └── config.yaml └── classifier/ └── ... Prompt 注册中心实现 import yaml import json from pathlib import Path from dataclasses import dataclass, field from typing import Any @dataclass class PromptVersion: """单个 Prompt 版本""" name: str version: str system: str few_shot: list[dict] = field(default_factory=list) model: str = "gpt-4o-mini" temperature: float = 0.7 max_tokens: int = 1024 status: str = "active" # draft | testing | active | archived def render(self, user_input: str) -> list[dict]: """渲染为 API 消息格式""" messages = [{"role": "system", "content": self.system}] for example in self.few_shot: messages.append({"role": example["role"], "content": example["content"]}) messages.append({"role": "user", "content": user_input}) return messages class PromptRegistry: """Prompt 版本注册中心""" def __init__(self, base_dir: str = "prompts"): self.base_dir = Path(base_dir) self._cache: dict[str, PromptVersion] = {} def load(self, name: str, version: str = "latest") -> PromptVersion: """加载指定版本的 Prompt""" if version == "latest": version = self._get_latest_version(name) cache_key = f"{name}@{version}" if cache_key in self._cache: return self._cache[cache_key] version_dir = self.base_dir / name / f"v{version}" # 加载系统提示 system = (version_dir / "system.txt").read_text(encoding="utf-8") # 加载 Few-shot few_shot = [] few_shot_path = version_dir / "few_shot.jsonl" if few_shot_path.exists(): for line in few_shot_path.read_text(encoding="utf-8").strip().split("\n"): few_shot.append(json.loads(line)) # 加载配置 config_path = version_dir / "config.yaml" config = yaml.safe_load(config_path.read_text(encoding="utf-8")) pv = PromptVersion( name=name, version=version, system=system, few_shot=few_shot, model=config.get("model", "gpt-4o-mini"), temperature=config.get("temperature", 0.7), max_tokens=config.get("max_tokens", 1024), status=config.get("status", "active"), ) self._cache[cache_key] = pv return pv def _get_latest_version(self, name: str) -> str: """获取最新 active 版本""" module_dir = self.base_dir / name versions = [] for d in module_dir.iterdir(): if d.is_dir() and d.name.startswith("v"): config = yaml.safe_load((d / "config.yaml").read_text(encoding="utf-8")) if config.get("status") == "active": versions.append(d.name[1:]) # 去掉 'v' 前缀 versions.sort(key=lambda v: [int(x) for x in v.split(".")]) return versions[-1] if versions else "1.0.0" def diff(self, name: str, v1: str, v2: str) -> dict: """对比两个版本的差异""" p1 = self.load(name, v1) p2 = self.load(name, v2) return { "system_changed": p1.system != p2.system, "few_shot_changed": p1.few_shot != p2.few_shot, "model_changed": p1.model != p2.model, "temperature_changed": p1.temperature != p2.temperature, } # 使用示例 registry = PromptRegistry("prompts") prompt = registry.load("chatbot", "latest") messages = prompt.render("你好，帮我查一下订单") 版本管理规范 版本类型 变更内容 示例 Major (x.0.0) Prompt 结构重构、角色定义变更 从单轮改为多轮对话 Minor (1.x.0) Few-shot 增删、指令逻辑调整 新增 2 个示例 Patch (1.0.x) 文案微调、错别字修正 “请” → “请帮我” A/B 测试框架 import random import hashlib from dataclasses import dataclass, field from collections import defaultdict import time @dataclass class ABTestConfig: """A/B 测试配置""" test_name: str variants: dict[str, PromptVersion] # variant_name -> Prompt traffic_split: dict[str, float] # variant_name -> 流量比例 metrics: list[str] = field(default_factory=lambda: [ "user_satisfaction", "response_length", "latency_ms", "cost" ]) min_sample_size: int = 100 def assign(self, user_id: str) -> str: """基于用户 ID 确定性分配变体（同一用户始终进入同一组）""" hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) ratio = (hash_val % 10000) / 10000.0 cumulative = 0.0 for variant, weight in self.traffic_split.items(): cumulative += weight if ratio < cumulative: return variant return list(self.traffic_split.keys())[-1] @dataclass class ExperimentResult: variant: str user_id: str metric: str value: float timestamp: float = field(default_factory=time.time) class ABTestRunner: """A/B 测试运行器""" def __init__(self): self.results: list[ExperimentResult] = [] def run( self, test_config: ABTestConfig, user_id: str, user_input: str, execute_fn, # callable: (PromptVersion, str) -> dict ) -> dict: """执行一次 A/B 测试请求""" variant_name = test_config.assign(user_id) prompt = test_config.variants[variant_name] # 执行并收集指标 result = execute_fn(prompt, user_input) # 记录指标 for metric in test_config.metrics: if metric in result: self.results.append(ExperimentResult( variant=variant_name, user_id=user_id, metric=metric, value=result[metric], )) return {"variant": variant_name, "result": result} def analyze(self, test_name: str) -> dict: """分析实验结果""" stats = defaultdict(lambda: defaultdict(list)) for r in self.results: stats[r.variant][r.metric].append(r.value) report = {} for variant, metrics in stats.items(): report[variant] = {} for metric, values in metrics.items(): vals = sorted(values) report[variant][metric] = { "count": len(vals), "mean": sum(vals) / len(vals), "median": vals[len(vals) // 2], "p95": vals[int(len(vals) * 0.95)] if len(vals) > 20 else None, } return report def is_significant(self, test_name: str, metric: str, alpha: float = 0.05) -> bool: """简单的统计显著性检验（Z 检验）""" import math variants = [r.variant for r in self.results if r.metric == metric] if len(set(variants)) < 2: return False # 按 variant 分组 groups = defaultdict(list) for r in self.results: if r.metric == metric: groups[r.variant].append(r.value) if len(groups) < 2: return False v1, v2 = list(groups.keys())[:2] s1, s2 = groups[v1], groups[v2] if len(s1) < 30 or len(s2) < 30: return False # 样本不足 m1, m2 = sum(s1) / len(s1), sum(s2) / len(s2) var1 = sum((x - m1) ** 2 for x in s1) / len(s1) var2 = sum((x - m2) ** 2 for x in s2) / len(s2) se = math.sqrt(var1 / len(s1) + var2 / len(s2)) if se == 0: return False z = abs(m1 - m2) / se return z > 1.96 # 95% 置信度 # 使用示例 runner = ABTestRunner() test_config = ABTestConfig( test_name="chatbot_tone_v2", variants={ "control": registry.load("chatbot", "1.0.0"), "treatment": registry.load("chatbot", "1.1.0"), }, traffic_split={"control": 0.5, "treatment": 0.5}, ) def execute_fn(prompt: PromptVersion, user_input: str) -> dict: # 实际调用 LLM return { "user_satisfaction": 4.5, # 用户评分 "response_length": 320, "latency_ms": 850, "cost": 0.003, } # 模拟 200 次请求 for i in range(200): runner.run(test_config, f"user-{i}", "帮我查订单", execute_fn) # 分析结果 report = runner.analyze("chatbot_tone_v2") for variant, metrics in report.items(): print(f"\n=== {variant} ===") for metric, stats in metrics.items(): print(f" {metric}: mean={stats['mean']:.2f}, p95={stats['p95']}") print(f"\n统计显著: {runner.is_significant('chatbot_tone_v2', 'user_satisfaction')}") 回归评测体系 每次 Prompt 变更前，必须通过回归测试集的验证。 ...

引言 RAG（Retrieval-Augmented Generation）是当前最流行的 LLM 应用架构之一。然而，从 Demo 到生产之间横亘着巨大的鸿沟。本文基于多个 RAG 生产项目的实战经验，总结 12 个最常见、最致命的坑，并给出经过验证的解决方案。 坑 1：文档分块策略不当 问题 天真地按固定长度分块（如每 512 字符），导致： 语义被截断（一个完整的段落从中间切断） 关键信息分散在多个块中，检索时只命中一部分 表格和列表被拆碎，失去结构信息 解决方案：语义分块 + 重叠窗口 from dataclasses import dataclass from typing import List import re @dataclass class Chunk: text: str metadata: dict token_count: int = 0 class SemanticChunker: """基于语义边界的智能分块器""" def __init__( self, target_size: int = 400, # 目标块大小（tokens） min_size: int = 100, # 最小块大小 max_size: int = 600, # 最大块大小 overlap: int = 50, # 重叠区间 ): self.target_size = target_size self.min_size = min_size self.max_size = max_size self.overlap = overlap def chunk_document(self, text: str, source: str = "") -> List[Chunk]: """分块主流程""" # Step 1: 按结构边界切分 sections = self._split_by_structure(text) # Step 2: 对每个 section 按 paragraph 切分 paragraphs = [] for section in sections: paragraphs.extend(self._split_by_paragraph(section)) # Step 3: 合并过小的段落，拆分过大的段落 chunks = self._merge_and_split(paragraphs) # Step 4: 添加重叠 chunks = self._add_overlap(chunks) # Step 5: 附加元数据 return [ Chunk( text=c, metadata={"source": source, "chunk_index": i, "total_chunks": len(chunks)}, token_count=len(c) // 2, # 粗略估算 ) for i, c in enumerate(chunks) ] def _split_by_structure(self, text: str) -> List[str]: """按标题、分隔符等结构边界切分""" # 按 Markdown 标题切分 pattern = r'(?=^#{1,6}\s)' sections = re.split(pattern, text, flags=re.MULTILINE) return [s.strip() for s in sections if s.strip()] def _split_by_paragraph(self, text: str) -> List[str]: """按段落（双换行）切分""" paras = text.split("\n\n") return [p.strip() for p in paras if p.strip()] def _merge_and_split(self, paragraphs: List[str]) -> List[str]: """合并过小段落，拆分过大段落""" chunks = [] buffer = [] buffer_size = 0 for para in paragraphs: para_size = len(para) // 2 # 粗略 token 估算 if buffer_size + para_size > self.max_size and buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if para_size > self.max_size: # 单段落过长，按句子切分 if buffer: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 sentences = re.split(r'(?<=[。！？.!?])\s+', para) sent_buffer = [] sent_size = 0 for sent in sentences: if sent_size + len(sent) // 2 > self.max_size and sent_buffer: chunks.append(" ".join(sent_buffer)) sent_buffer = [] sent_size = 0 sent_buffer.append(sent) sent_size += len(sent) // 2 if sent_buffer: chunks.append(" ".join(sent_buffer)) else: buffer.append(para) buffer_size += para_size if buffer_size >= self.target_size: chunks.append("\n\n".join(buffer)) buffer = [] buffer_size = 0 if buffer: chunks.append("\n\n".join(buffer)) return chunks def _add_overlap(self, chunks: List[str]) -> List[str]: """为相邻块添加重叠""" if self.overlap <= 0 or len(chunks) <= 1: return chunks result = [chunks[0]] for i in range(1, len(chunks)): prev_text = chunks[i - 1] overlap_text = prev_text[-self.overlap * 2:] # 粗略取后半段 result.append(overlap_text + " " + chunks[i]) return result # 使用示例 chunker = SemanticChunker(target_size=400, overlap=50) document = open("knowledge_base/product_manual.md").read() chunks = chunker.chunk_document(document, source="product_manual.md") print(f"分块完成: {len(chunks)} 个块") 分块策略对比 策略 优点 缺点 适用场景 固定长度 实现简单 语义截断 不推荐 按段落 保持语义 块大小不均 短文档 语义分块 语义完整 实现复杂 通用推荐 按文档结构 保持层级 需要结构化输入 Markdown/HTML 递归分块 灵活适配 可控性差 混合内容 坑 2：Embedding 模型与 LLM 不匹配 问题 用 OpenAI 的 text-embedding-3-large 做向量，但生成用的是 Claude 模型。两者对语义的理解不同，可能导致检索到的内容并非生成模型"认为"最相关的。 ...

引言 “我们应该微调还是用 RAG？"——这是 2024-2026 年 AI 工程师被问得最多的问题。两者都能让 LLM 适配特定领域，但适用场景截然不同。本文不给标准答案，而是给一个决策框架。 核心差异：先理解本质 微调（Fine-tuning） 微调是改变模型内部权重，让模型"学会"新的知识或行为模式。 原始模型 → 注入领域数据 → 新权重 → 直接推理 RAG（检索增强生成） RAG是不改变模型，在推理时"查阅"外部知识库。 用户问题 → 检索知识库 → 拼接到 Prompt → 模型生成答案 本质对比 维度 微调 RAG 知识存储位置 模型权重中 外部数据库中 知识更新方式 重新训练 更新数据库 推理时依赖 无外部依赖 需要检索系统 “记忆"方式 隐式（权重） 显式（文本） 幻觉风险 较高（知识编码不精确） 较低（有原文参考） 知识溯源 无法溯源 可追溯到来源文档 决策框架：7 个关键问题 问题 1：知识更新频率多高？ 更新频率 │ ├─ 每日/每周更新 ──────→ RAG ✅ │ （产品目录、新闻、工单） │ ├─ 每月/每季度更新 ─────→ RAG 或混合 │ （技术文档、政策） │ └─ 几乎不变 ───────────→ 微调 ✅ （行业术语、品牌语调） 原理：微调每次更新知识都需要重新训练（数小时到数天），RAG 只需更新数据库（秒级）。 ...

为什么 Agent 需要记忆 没有记忆的 Agent 就像金鱼——每次对话都从零开始。你告诉它你的名字，下一轮它就忘了。你纠正过它的错误，它下次照犯。 记忆系统让 Agent 能够： 记住用户偏好和历史交互 从过去经验中学习 保持跨会话的上下文一致性 在长任务中不丢失关键信息 记忆类型体系 人类认知科学启发 记忆类型 持续时间 容量 Agent 对应 存储方案 工作记忆 秒-分钟 4±2 项 当前对话上下文 LLM Context Window 短期记忆 分钟-小时 有限 当前会话摘要 内存/Redis 长期记忆 天-永久 近无限 用户偏好、知识 向量库 + 关系库 情景记忆 事件级 大 具体交互记录 时序向量库 架构总览 ┌──────────────────────────────────────────────┐ │ Agent 主循环 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌──────────────┐ │ │ │ 工作记忆 │ │ 短期记忆 │ │ 长期记忆 │ │ │ │ Context │←→│ Session │←→│ Persistent │ │ │ │ Window │ │ Summary │ │ Store │ │ │ └─────────┘ └─────────┘ └──────────────┘ │ │ │ │ │ │ │ │ │ ┌────────┼────────┐ │ │ │ │ │ │ │ │ │ │ │ ┌──┴──┐ ┌───┴───┐ ┌──┴──┐ │ │ │ │语义 │ │情景 │ │程序 │ │ │ │ │记忆 │ │记忆 │ │记忆 │ │ │ │ │向量 │ │时序 │ │图 │ │ │ │ └─────┘ └───────┘ └─────┘ │ │ │ │ ┌────┴────────────┴────────────────────────┐│ │ │ 记忆管理器 ││ │ │ 写入 → 压缩 → 检索 → 遗忘 → 巩固 ││ │ └────────────────────────────────────────────┘│ └──────────────────────────────────────────────┘ 工作记忆：上下文窗口管理 from dataclasses import dataclass, field from typing import List, Dict @dataclass class WorkingMemory: """工作记忆: 管理 LLM 上下文窗口""" max_tokens: int = 128000 reserved_for_response: int = 4096 messages: List[dict] = field(default_factory=list) def add(self, role: str, content: str): self.messages.append({"role": role, "content": content}) self._evict_if_needed() def add_system(self, content: str): """系统消息放在最前面，不被驱逐""" self.messages.insert(0, {"role": "system", "content": content}) def _evict_if_needed(self): """当超过 token 限制时，驱逐最旧的对话""" while self._count_tokens() > self.max_tokens - self.reserved_for_response: if len(self.messages) <= 1: break # 找到第一个非 system 消息并移除 for i, msg in enumerate(self.messages): if msg["role"] != "system": evicted = self.messages.pop(i) # 触发压缩: 被驱逐的消息需要被压缩到短期记忆 self._on_evict(evicted) break def _count_tokens(self) -> int: import tiktoken enc = tiktoken.encoding_for_model("gpt-4o") return sum(len(enc.encode(m["content"])) for m in self.messages) def _on_evict(self, evicted_msg: dict): """被驱逐的消息回调，交给短期记忆处理""" # 由 MemoryManager 注册 pass def get_context(self) -> List[dict]: """获取当前上下文""" return self.messages.copy() 短期记忆：会话级摘要 class ShortTermMemory: """短期记忆: 当前会话的压缩摘要""" def __init__(self): self.summary = "" self.key_points: List[str] = [] self.token_budget = 2000 # 摘要不超 2000 token async def compress(self, messages: List[dict]) -> str: """将多轮对话压缩为摘要""" conversation = "\n".join( f"{m['role']}: {m['content'][:200]}" for m in messages ) prompt = f"""请将以下对话压缩为简洁摘要，保留: 1. 用户的核心需求 2. 已确定的关键事实 3. 未解决的问题 4. 用户的明确偏好 对话: {conversation} 摘要:""" response = await llm_call(prompt, max_tokens=self.token_budget) self.summary = response return response async def extract_key_points(self, messages: List[dict]) -> List[str]: """提取关键信息点""" prompt = f"""从以下对话中提取关键信息点，每点一行: {chr(10).join(m['content'][:200] for m in messages)} 关键信息:""" response = await llm_call(prompt, max_tokens=500) self.key_points = response.strip().split('\n') return self.key_points def to_context(self) -> str: """转换为可注入上下文的文本""" parts = [] if self.summary: parts.append(f"之前的对话摘要:\n{self.summary}") if self.key_points: parts.append(f"关键信息:\n" + "\n".join(f"- {p}" for p in self.key_points)) return "\n\n".join(parts) 长期记忆：持久化存储 语义记忆（向量库） class SemanticMemory: """语义记忆: 存储事实和知识，向量检索""" def __init__(self, vector_store, embed_model): self.store = vector_store self.embed = embed_model async def remember(self, content: str, metadata: dict = None): """记住一条信息""" embedding = await self.embed(content) await self.store.upsert({ "id": self._generate_id(content), "values": embedding, "metadata": { "content": content, "type": "semantic", "created_at": datetime.now().isoformat(), **(metadata or {}) } }) async def recall(self, query: str, top_k: int = 5) -> List[dict]: """检索相关记忆""" query_embedding = await self.embed(query) results = await self.store.query( vector=query_embedding, top_k=top_k, filter={"type": {"$eq": "semantic"}} ) return [ { "content": r.metadata["content"], "score": r.score, "metadata": r.metadata } for r in results.matches ] async def forget(self, content_id: str): """主动遗忘""" await self.store.delete(ids=[content_id]) 情景记忆（时序事件） class EpisodicMemory: """情景记忆: 记录具体交互事件，支持时间检索""" def __init__(self, db): self.db = db # PostgreSQL + pgvector async def record(self, event: dict): """记录一个交互事件""" await self.db.execute( """ INSERT INTO episodic_memory (event_id, user_id, timestamp, input, output, outcome, context_snapshot, embedding) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) """, event["id"], event["user_id"], datetime.now(), event["input"], event["output"], event["outcome"], json.dumps(event.get("context", {})), event.get("embedding") ) async def recall_by_time(self, user_id: str, start: datetime, end: datetime) -> List[dict]: """按时间范围回忆""" rows = await self.db.fetch( """SELECT * FROM episodic_memory WHERE user_id = $1 AND timestamp BETWEEN $2 AND $3 ORDER BY timestamp DESC""", user_id, start, end ) return [dict(r) for r in rows] async def recall_similar(self, user_id: str, query_embedding: list, top_k: int = 5) -> List[dict]: """检索相似的历史事件""" rows = await self.db.fetch( """SELECT *, embedding <=> $1 as distance FROM episodic_memory WHERE user_id = $2 ORDER BY embedding <=> $1 LIMIT $3""", query_embedding, user_id, top_k ) return [dict(r) for r in rows] 程序记忆（知识图谱） class ProceduralMemory: """程序记忆: 存储技能、流程、因果关系（图结构）""" def __init__(self, graph_db): self.graph = graph_db # Neo4j async def learn_procedure(self, name: str, steps: List[str], triggers: List[str]): """学习一个新流程""" await self.graph.execute( """ CREATE (p:Procedure {name: $name}) WITH p UNWIND $steps as step_text CREATE (s:Step {text: step_text}) CREATE (p)-[:HAS_STEP]->(s) WITH p UNWIND $triggers as trigger CREATE (t:Trigger {text: trigger}) CREATE (t)-[:ACTIVATES]->(p) """, name=name, steps=steps, triggers=triggers ) async def recall_procedure(self, context: str) -> List[dict]: """根据上下文检索相关流程""" result = await self.graph.execute( """ MATCH (t:Trigger)-[:ACTIVATES]->(p:Procedure)-[:HAS_STEP]->(s:Step) WHERE $context CONTAINS t.text RETURN p.name as procedure, collect(s.text) as steps """, context=context ) return [dict(r) for r in result] 记忆管理器：统一调度 class MemoryManager: """统一记忆管理器""" def __init__(self): self.working = WorkingMemory() self.short_term = ShortTermMemory() self.semantic = SemanticMemory(...) self.episodic = EpisodicMemory(...) self.procedural = ProceduralMemory(...) self.recall_count = 0 async def think(self, user_input: str, user_id: str) -> str: """带记忆的思考过程""" # 1. 从长期记忆检索相关信息 semantic_results = await self.semantic.recall(user_input, top_k=3) episodic_results = await self.episodic.recall_similar( user_id, await self.embed(user_input), top_k=3 ) procedures = await self.procedural.recall_procedure(user_input) # 2. 组装上下文 context_parts = [] if self.short_term.summary: context_parts.append(self.short_term.to_context()) if semantic_results: context_parts.append("相关知识:\n" + "\n".join(f"- {r['content']}" for r in semantic_results)) if episodic_results: context_parts.append("历史交互:\n" + "\n".join(f"- {r['input']} → {r['output']}" for r in episodic_results)) if procedures: context_parts.append("可用流程:\n" + "\n".join(f"- {p['procedure']}: {p['steps']}" for p in procedures)) context = "\n\n".join(context_parts) # 3. 注入工作记忆 if context: self.working.add("system", f"记忆上下文:\n{context[:4000]}") self.working.add("user", user_input) # 4. 生成回复 response = await llm_call(self.working.get_context()) self.working.add("assistant", response) # 5. 异步写入记忆 asyncio.create_task(self._consolidate(user_input, response, user_id)) return response async def _consolidate(self, input: str, output: str, user_id: str): """记忆巩固: 筛选重要信息写入长期记忆""" # 判断是否值得记住 importance = await self._assess_importance(input, output) if importance > 0.7: # 高重要性: 写入语义记忆 + 情景记忆 await self.semantic.remember( f"用户说了: {input}\n助手回答: {output}", metadata={"importance": importance, "user_id": user_id} ) # 总是记录情景记忆 await self.episodic.record({ "id": str(uuid4()), "user_id": user_id, "input": input, "output": output, "outcome": "success", "embedding": await self.embed(input), }) async def _assess_importance(self, input: str, output: str) -> float: """评估信息重要性 0-1""" prompt = f"""评估以下交互的信息重要性(0-1): 输入: {input[:200]} 输出: {output[:200]} 评分标准: - 0.9+: 用户明确要求记住/涉及核心偏好 - 0.7-0.8: 包含重要事实或决策 - 0.5-0.6: 有一定参考价值 - 0.3-0.4: 闲聊但含少量信息 - 0.0-0.2: 纯闲聊无信息 只输出数字:""" return float(await llm_call(prompt, max_tokens=10)) 遗忘机制 class ForgettingMechanism: """遗忘机制: 防止记忆膨胀""" async def decay(self, user_id: str): """基于时间的记忆衰减""" await self.db.execute( """ UPDATE semantic_memory SET importance = importance * 0.95 -- 每天衰减5% WHERE user_id = $1 AND last_accessed < NOW() - INTERVAL '1 day' """, user_id ) # 删除低重要性记忆 await self.db.execute( """ DELETE FROM semantic_memory WHERE user_id = $1 AND importance < 0.1 """, user_id ) async def deduplicate(self, user_id: str): """去重: 合并语义相似的记忆""" # 找到相似度 > 0.9 的记忆对 duplicates = await self.db.fetch( """ SELECT a.id as id_a, b.id as id_b, a.content as content_a, b.content as content_b FROM semantic_memory a JOIN semantic_memory b ON a.id < b.id WHERE a.user_id = $1 AND b.user_id = $1 AND a.embedding <=> b.embedding < 0.1 """, user_id ) for dup in duplicates: # 合并: 保留更详细的，删除另一个 longer = dup['content_a'] if len(dup['content_a']) > len(dup['content_b']) else dup['content_b'] shorter_id = dup['id_b'] if len(dup['content_a']) > len(dup['content_b']) else dup['id_a'] await self.db.execute("DELETE FROM semantic_memory WHERE id = $1", shorter_id) async def compress_old_memories(self, user_id: str, days: int = 30): """压缩旧记忆: 把多条相关记忆合并为一条摘要""" old_memories = await self.db.fetch( """ SELECT content FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days' ORDER BY created_at """, user_id, days ) if len(old_memories) < 10: return # 用 LLM 合并 memories_text = "\n".join(m['content'] for m in old_memories) summary = await llm_call( f"将以下记忆合并为简洁摘要，保留关键信息:\n{memories_text}" ) # 删除旧记忆，写入摘要 await self.db.execute( "DELETE FROM semantic_memory WHERE user_id = $1 AND created_at < NOW() - INTERVAL '$2 days'", user_id, days ) await self.semantic.remember(summary, metadata={"compressed": True}) MemGPT / Letta 架构参考 MemGPT (现 Letta) 的核心思想是把操作系统的虚拟内存概念引入 LLM： ...

Agent 模式全景 复杂度 → 单Agent ──→ Router ──→ Fan-out/Fan-in ──→ Hierarchical ──→ Multi-Agent │ │ │ │ │ │ 简单任务 │ 按需分发 │ 并行处理 │ 管理者+执行者 │ 自主协作 │ │ │ │ │ └────────────┴──────────────┴──────────────────┴────────────────┘ 模式 适用场景 复杂度 延迟 错误恢复 单 Agent 简单问答、单一任务 低 低 简单重试 Router 多领域问答、任务分类 中 低+路由 单分支重试 Fan-out/Fan-in 并行研究、批量处理 中 高(并行) 部分失败容忍 Hierarchical 复杂项目、多步骤流程 高 高 子任务级恢复 Multi-Agent 开放式协作、辩论 极高 最高 最复杂 单 Agent 架构 最基础的形态：一个 Agent + 一组工具。 ...

成本结构分析 在深入优化之前，先搞清楚钱花在哪里。一个典型的 RAG 应用每次调用的成本构成： 成本项 占比 计算方式 可优化空间 LLM 生成 Token 50-65% output_tokens × price_per_1k 高 LLM 输入 Token 20-35% input_tokens × price_per_1k 高 Embedding 5-10% chunks × chunk_size × embed_price 中 向量检索 2-5% 按次/按量计费 低 基础设施 5-15% GPU/CPU/内存/带宽 中 关键洞察：输出 Token 通常比输入 Token 贵 3-5 倍，优化输出端收益更大。 Prompt 压缩 1. 去除冗余信息 # 冗长 Prompt（382 tokens） prompt = """ 你是一个专业的客服助手，你的主要职责是回答用户关于产品的问题。 你需要始终保持礼貌和专业，回答要简洁明了。 如果用户的问题超出了你的知识范围，请诚实告知用户你会将问题转接给人工客服。 在回答之前，请先理解用户的核心诉求，然后给出针对性的回答。 请使用以下格式回答： 1. 直接回答问题 2. 如果需要补充信息，请提供 3. 如果无法回答，请说明原因 """ # 精简 Prompt（89 tokens） prompt = "客服助手。简洁专业。不知则转人工。格式：直接回答→补充→说明原因。" 2. Context Compression 对长文档先做摘要再喂给大模型： ...

监控分层架构 AI 应用监控需要三层视角，每层关注不同问题： ┌─────────────────────────────────────────────────┐ │ Layer 3: 模型质量监控 │ │ - 回答正确率 / 幻觉率 / 毒性检测 / 偏见 │ │ - 用户满意度反馈 │ ├─────────────────────────────────────────────────┤ │ Layer 2: 应用性能监控 │ │ - 首字延迟 (TTFT) / 完整延迟 / 吞吐量 │ │ - Token 消耗 / 成本 / 缓存命中率 │ ├─────────────────────────────────────────────────┤ │ Layer 1: 基础设施监控 │ │ - GPU 利用率 / 内存 / 网络 / 磁盘 │ │ - API 错误率 / 限流次数 / 超时 │ └─────────────────────────────────────────────────┘ 指标设计 核心指标清单 层级 指标 类型 告警阈值 模型 回答质量评分 Gauge < 0.8 模型 幻觉率 Gauge > 5% 模型 用户点赞率 Gauge < 70% 应用 TTFT (首字延迟) Histogram P99 > 2s 应用 完整延迟 Histogram P99 > 15s 应用 Token 消耗 Counter 日预算 120% 应用 缓存命中率 Gauge < 15% 基础设施 GPU 利用率 Gauge > 90% 基础设施 API 错误率 Gauge > 1% 基础设施 限流次数 Counter > 100/min Prometheus 指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # === 模型质量指标 === llm_quality_score = Gauge( "llm_quality_score", "Average answer quality score (0-1)", ["model", "task_type"] ) llm_hallucination_rate = Gauge( "llm_hallucination_rate", "Hallucination rate detected", ["model"] ) llm_user_feedback = Counter( "llm_user_feedback_total", "User feedback counts", ["model", "feedback"] # feedback: positive/negative ) # === 应用性能指标 === llm_ttft = Histogram( "llm_time_to_first_token_seconds", "Time to first token in seconds", ["model"], buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10] ) llm_total_latency = Histogram( "llm_total_latency_seconds", "Total request latency in seconds", ["model"], buckets=[0.5, 1, 2, 5, 10, 20, 30, 60] ) llm_tokens_total = Counter( "llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) llm_cost_total = Counter( "llm_cost_usd_total", "Total cost in USD", ["model"] ) llm_cache_hit_rate = Gauge( "llm_cache_hit_rate", "Cache hit rate", ["cache_type"] # cache_type: exact/semantic ) # === 基础设施指标 === gpu_utilization = Gauge( "gpu_utilization_percent", "GPU utilization percentage", ["gpu_id", "node"] ) api_error_rate = Gauge( "llm_api_error_rate", "LLM API error rate", ["provider", "error_type"] ) 指标采集 中间件式采集 import time from contextlib import asynccontextmanager class LLMInstrumentation: def __init__(self, registry): self.registry = registry @asynccontextmanager async def trace_request(self, model: str, task_type: str): request_id = generate_id() start = time.monotonic() first_token_time = None async def on_first_token(): nonlocal first_token_time first_token_time = time.monotonic() try: yield {"request_id": request_id, "on_first_token": on_first_token} # 请求成功 elapsed = time.monotonic() - start llm_total_latency.labels(model=model).observe(elapsed) if first_token_time: ttft = first_token_time - start llm_ttft.labels(model=model).observe(ttft) except Exception as e: api_error_rate.labels( provider=model, error_type=type(e).__name__ ).inc() raise def record_tokens(self, model, input_tokens, output_tokens): llm_tokens_total.labels( model=model, direction="input" ).inc(input_tokens) llm_tokens_total.labels( model=model, direction="output" ).inc(output_tokens) # 成本计算 cost = self._calc_cost(model, input_tokens, output_tokens) llm_cost_total.labels(model=model).inc(cost) def _calc_cost(self, model, in_tok, out_tok): prices = { "gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006), "claude-sonnet": (0.003, 0.015), } in_price, out_price = prices.get(model, (0, 0)) return (in_tok * in_price + out_tok * out_price) / 1000 使用方式 instrument = LLMInstrumentation(registry) async def chat_completion(messages, model="gpt-4o"): async with instrument.trace_request(model, "chat") as ctx: response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, ) async for chunk in response: if chunk.choices[0].delta.content: if not ctx["first_token_called"]: await ctx["on_first_token"]() ctx["first_token_called"] = True yield chunk.choices[0].delta.content instrument.record_tokens(model, input_count, output_count) 质量监控：LLM-as-a-Judge class QualityMonitor: def __init__(self, judge_model="gpt-4o", sample_rate=0.1): self.judge_model = judge_model self.sample_rate = sample_rate async def maybe_evaluate(self, query, response, context=None): """按采样率随机评估回答质量""" import random if random.random() > self.sample_rate: return score, issues = await self._judge(query, response, context) llm_quality_score.labels( model=self.judge_model, task_type="auto" ).set(score) if "hallucination" in issues: llm_hallucination_rate.labels( model=self.judge_model ).inc() async def _judge(self, query, response, context): prompt = f"""评估以下回答的质量。 问题：{query} 回答：{response} 参考资料：{context or '无'} 请检查： 1. 事实准确性（是否有幻觉） 2. 回答完整性 3. 逻辑一致性 输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}""" result = await llm_client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) data = json.loads(result.choices[0].message.content) return data["score"], data.get("issues", []) Grafana 仪表板 关键 Panel 配置 Panel 1: 延迟分布（P50/P90/P99） ...