从模型到监控 “用Prometheus+Grafana搭建RAG系统运维看板”

# 多维度指标采集示例
rag_retrieval_latency_seconds{stage="vector_search", index="product_v2", shard="shard03"} 0.87
rag_llm_cost_usd{model="claude-3-opus", tier="200k", endpoint="/generate"} 2.31
rag_recall_rate{query_type="policy_search", index_version="202405"} 0.82

1. 多维度标签：通过stage/model/version等标签实现细粒度分析
2. 高效存储：每个样本仅占3-5字节，千万级指标日增存储<50GB
3. PromQL强大查询：支持跨指标关联分析（如延迟与成本相关性）
4. 生态集成：无缝对接Grafana/Alertmanager/Jaeger

- name: rag_retrieval_latency_distribution
  type: histogram
  buckets: [0.05, 0.1, 0.3, 0.5, 1, 2, 5]
  labels: [stage, index_type, shard_id]
  
- name: rag_recall_precision
  type: summary
  labels: [query_category, index_version]
  quantiles: {0.5: 0.05, 0.9: 0.01, 0.99: 0.001}
  
- name: rag_cache_efficiency
  type: gauge
  help: "缓存命中效率"

from prometheus_client import Histogram, Counter

EMBEDDING_LATENCY = Histogram('rag_embedding_latency_seconds', '嵌入延迟', ['model_version'])
EMBEDDING_ERRORS = Counter('rag_embedding_errors', '嵌入错误', ['error_code'])

def embed_text(text, model="text-embedding-ada-003"):
    start = time.time()
    try:
        # 模型加载检查
        if not model_loaded[model]:
            load_model(model)
        
        # 输入验证
        if len(text) > MAX_INPUT_LENGTH:
            raise ValueError("Input too long")
            
        # 执行嵌入
        vector = embedding_models[model](text)
        
        # 记录延迟
        EMBEDDING_LATENCY.labels(model_version=model).observe(time.time() - start)
        return vector
    except Exception as e:
        error_code = classify_error(e)
        EMBEDDING_ERRORS.labels(error_code=error_code).inc()
        raise

总成本 = Σ(输入token数 × 输入单价) + Σ(输出token数 × 输出单价) + 固定调用费

LLM_COST_USD = Counter('rag_llm_cost_usd', '累计成本', ['model', 'tier'])
LLM_TOKEN_USAGE = Counter('rag_llm_tokens_total', 'token用量', ['type'])

MODEL_PRICING = {
    "gpt-4-turbo": {"in": 0.01, "out": 0.03, "fixed": 0.001},
    "claude-3-sonnet": {"in": 0.003, "out": 0.015, "fixed": 0}
}

def calculate_llm_cost(model, input_tokens, output_tokens):
    if model not in MODEL_PRICING:
        model = "default"
    
    pricing = MODEL_PRICING[model]
    cost = (input_tokens/1000)*pricing["in"] + (output_tokens/1000)*pricing["out"] + pricing["fixed"]
    
    LLM_COST_USD.labels(model=model).inc(cost)
    LLM_TOKEN_USAGE.labels(type="input").inc(input_tokens)
    LLM_TOKEN_USAGE.labels(type="output").inc(output_tokens)
    return cost

# 人工反馈数据采集
FEEDBACK_SCORE = Gauge('rag_feedback_score', '用户评分', ['session_id'])
HALLUCINATION_FLAG = Counter('rag_hallucination_events', '幻觉事件')

def record_feedback(session_id, score, comment):
    FEEDBACK_SCORE.labels(session_id=session_id).set(score)
    
    # NLP检测幻觉关键词
    if detect_hallucination(comment):
        HALLUCINATION_FLAG.inc()

-- 检索延迟热力图
SELECT
  histogram_quantile(0.95, sum(rate(rag_retrieval_latency_seconds_bucket[5m])) as p95
FROM metrics
WHERE stage='vector_search'
GROUP BY time_bucket('1h'), index_version

-- 成本燃烧率预测
SELECT
  sum(rag_llm_cost_usd) as current_cost,
  integral(sum(rate(rag_llm_cost_usd[24h])) * 30 as predicted_monthly
FROM metrics

# 服务依赖图查询
sum by (service)(rate(request_duration_seconds_sum{namespace="rag-prod"}[5m]))
/
sum by (service)(rate(request_duration_seconds_count{namespace="rag-prod"}[5m]))

-- 召回率与用户评分关联分析
SELECT
  correlation(
    avg_over_time(rag_recall_rate[1h]),
    avg_over_time(rag_feedback_score[1h])
  ) as recall_satisfaction_corr
FROM metrics
WHERE query_type="technical_support"

- alert: RetrievalServiceDegradation
  expr: |
    # 基于基线自动调整阈值
    (rate(rag_retrieval_failures_total[10m]) 
     > (avg_over_time(rag_retrieval_failures_total[7d]) * 1.5))
    and
    (rate(rag_requests_total[10m]) > 5)
  for: 3m
  labels:
    severity: critical
    playbook: "/playbooks/retrieval_failure.md"
  annotations:
    summary: "检索服务异常率超过基线150%"
    impact: "用户请求超时率上升"

- alert: LLMCostAnomaly
  expr: |
    # 基于时间序列预测
    rag_llm_cost_usd - predict_linear(rag_llm_cost_usd[7d], 86400*30) > 1000
  for: 30m
  annotations:
    description: "当月成本预测超预算$1000"
    action: "检查高消耗端点：{{ $labels.endpoint }}"

- alert: KnowledgeCoverageDrop
  expr: |
    # 知识覆盖度下降检测
    avg(rag_recall_rate{index="knowledge_v3"}) 
    < 
    (avg_over_time(rag_recall_rate{index="knowledge_v3"}[7d]) * 0.85)
  for: 6h
  annotations:
    report: "知识库更新建议：{{ $labels.section }}"

SELECT topk(10, sum(rate(rag_retrieval_latency_seconds_count[5m])) by (query_hash)
FROM metrics WHERE shard="shard03"

def retrieve_with_cache(query, ttl=3600):
    cache_key = f"retrieval:{sha256(query)}"
    if cached := redis.get(cache_key):
        return cached
    
    results = vector_db.search(query)
    redis.setex(cache_key, ttl, pickle.dumps(results))
    return results

SELECT 
  model_version,
  avg(rag_recall_rate) as avg_recall
FROM metrics 
WHERE time > now() - 7d
GROUP BY model_version

def select_embedding_model(text):
    if len(text) > 100:
        return "text-embedding-ada-002"
    return "text-embedding-3-large"

SELECT 
  model,
  sum(rag_llm_cost_usd) as cost
FROM metrics
WHERE time > now() - 72h
GROUP BY model

def truncate_context(context, max_tokens=128000):
    tokens = tokenize(context)
    if len(tokens) > max_tokens:
        # 保留头尾关键信息
        head = tokens[:max_tokens//3]
        tail = tokens[-max_tokens//3:]
        return detokenize(head + ["..."] + tail)
    return context

def dynamic_sampling(error_rate):
    if error_rate > 0.1:
        return 1.0  # 全量采集
    elif error_rate > 0.01:
        return 0.5
    return 0.1

热数据：SSD存储，保留7天
温数据：高性能HDD，保留30天
冷数据：对象存储，保留1年

-- 原始查询（执行时间12s）
SELECT * 
FROM metrics 
WHERE model="gpt-4" 
AND time > now() - 7d

-- 优化后（0.8s）
SELECT /*+ MATERIALIZED */ cost, latency 
FROM daily_model_summary 
WHERE model="gpt-4"
AND date BETWEEN '2024-06-01' AND '2024-06-07'

def predict_capacity():
    query_growth = forecast(rag_requests_total[30d], horizon="7d")
    required_nodes = max(3, query_growth * 0.8 / 1000) 
    k8s.scale(deployment="embedding", replicas=required_nodes)

EXPLAIN SIMULATE 
SELECT sum(llm_cost) 
FROM production_traffic
WHERE model="claude-3.5-sonnet"

def check_knowledge_gaps():
    low_recall_queries = get_queries("recall_rate < 0.6")
    for query in low_recall_queries:
        if not exists_in_kb(query):
            jira.create_task(
                type="Knowledge Gap",
                priority="High",
                description=f"未覆盖查询: {query}"
            )

SELECT 
  rag_recall_rate as quality,
  rag_llm_cost_per_query as cost,
  quality / cost as roi
FROM metrics
ORDER BY roi DESC

当预测未来24小时成本超限时：
  1. 自动切换备用模型
  2. 发送预警告警
  3. 生成优化建议报告

传统监控：发生了什么 → 被动响应
智能监控：为什么发生 → 主动预防
业务监控：如何优化 → 价值创造