11f - RAG 评估与优化

用户查询
  │
  ▼
┌─────────────────────────────────────┐
│         检索阶段评估                  │
│  Context Precision ← 检索到的文档相关吗？│
│  Context Recall    ← 需要的信息都检索到了吗？│
│  Precision@K / Recall@K             │
└──────────────────┬──────────────────┘
                   ▼
┌─────────────────────────────────────┐
│         生成阶段评估                  │
│  Faithfulness      ← 回答忠实于上下文吗？│
│  Answer Relevancy  ← 回答切题吗？      │
└──────────────────┬──────────────────┘
                   ▼
┌─────────────────────────────────────┐
│         端到端评估                    │
│  Answer Correctness ← 最终回答正确吗？  │
│  Noise Sensitivity  ← 系统鲁棒吗？     │
└─────────────────────────────────────┘

# 黄金数据集格式：包含查询、标准答案、相关上下文
evaluation_dataset = [
    {
        "question": "RAG 系统中的分块策略有哪些？",
        "ground_truth": "常见的分块策略包括固定大小分块、递归分块、语义分块和文档结构分块...",
        "contexts": [
            "固定大小分块将文档按固定 token 数切分，简单但可能破坏语义...",
            "语义分块使用嵌入模型检测语义边界，保持段落完整性...",
        ]
    },
    {
        "question": "如何选择向量数据库？",
        "ground_truth": "选择向量数据库需要考虑数据规模、查询延迟、托管方式和成本...",
        "contexts": [
            "Pinecone 提供全托管服务，适合零运维需求...",
            "Qdrant 支持混合搜索，适合需要精确关键词匹配的场景...",
        ]
    },
    # 建议至少 50-100 条评估样本
]

你是一个 RAG 系统评估专家。请帮我设计评估方案：

**系统信息：**
- 知识库规模：[文档数量] 篇，[总 token 数]
- 查询类型：[事实查询/分析查询/对比查询/操作指导]
- 当前已有评估数据：[有/无] 黄金数据集
- 部署阶段：[开发/预发布/生产]

**请输出：**
1. 推荐的评估指标组合及优先级
2. 黄金数据集的构建策略（样本数、覆盖范围、标注方法）
3. 评估工具选择及配置建议
4. 基线指标的合理范围（各指标的目标值）
5. 持续评估的自动化方案

pip install ragas langchain-openai datasets

from ragas import evaluate
from ragas.metrics import (
    Faithfulness,
    AnswerRelevancy,
    ContextPrecision,
    ContextRecall,
)
from ragas.dataset_schema import SingleTurnSample, EvaluationDataset
from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI
 
# 配置评判模型
evaluator_llm = LangchainLLMWrapper(
    ChatOpenAI(model="gpt-4o-mini", temperature=0)
)
 
# 准备评估数据（从你的 RAG pipeline 收集）
samples = [
    SingleTurnSample(
        user_input="RAG 系统的核心组件有哪些？",
        retrieved_contexts=[
            "RAG 系统由文档摄入、分块、嵌入、向量存储、检索和生成六个核心组件组成。",
            "文档摄入负责从各种数据源加载原始文档，支持 PDF、Markdown、HTML 等格式。",
        ],
        response="RAG 系统的核心组件包括：1) 文档摄入模块，2) 分块策略，"
                 "3) 嵌入模型，4) 向量存储，5) 检索引擎，6) 生成模型。",
        reference="RAG 系统由六个核心组件组成：文档摄入、分块、嵌入、向量存储、检索和生成。",
    ),
    SingleTurnSample(
        user_input="如何优化 RAG 的检索精度？",
        retrieved_contexts=[
            "混合搜索结合 BM25 和向量搜索，可提升 10-30% 的召回率。",
            "重排序模型在初始检索后进行精排，显著提升排序质量。",
            "查询转换技术如 HyDE 可以弥合查询与文档之间的语义鸿沟。",
        ],
        response="优化 RAG 检索精度的主要方法包括：混合搜索（BM25 + 向量搜索）、"
                 "重排序（Cross-Encoder 精排）、查询转换（HyDE、查询分解）。"
                 "混合搜索可提升 10-30% 召回率，重排序可进一步提升精度。",
        reference="优化 RAG 检索精度的方法包括混合搜索、重排序和查询转换。",
    ),
]
 
# 创建评估数据集
eval_dataset = EvaluationDataset(samples=samples)
 
# 定义评估指标
metrics = [
    Faithfulness(llm=evaluator_llm),
    AnswerRelevancy(llm=evaluator_llm),
    ContextPrecision(llm=evaluator_llm),
    ContextRecall(llm=evaluator_llm),
]
 
# 执行评估
results = evaluate(
    dataset=eval_dataset,
    metrics=metrics,
)
 
# 查看结果
print(results)
# 输出示例：
# {'faithfulness': 0.92, 'answer_relevancy': 0.88,
#  'context_precision': 0.85, 'context_recall': 0.90}
 
# 转为 DataFrame 查看每条样本的详细分数
df = results.to_pandas()
print(df[['user_input', 'faithfulness', 'answer_relevancy',
           'context_precision', 'context_recall']])

# test_rag_quality.py — 用 pytest + RAGAS 做回归测试
import pytest
from ragas import evaluate
from ragas.metrics import Faithfulness, AnswerRelevancy
from ragas.dataset_schema import SingleTurnSample, EvaluationDataset
from ragas.llms import LangchainLLMWrapper
from langchain_openai import ChatOpenAI
import json
 
# 加载黄金数据集
def load_golden_dataset(path: str) -> list[dict]:
    with open(path) as f:
        return json.load(f)
 
# 运行 RAG pipeline 获取回答
def run_rag_pipeline(question: str) -> dict:
    """调用你的 RAG pipeline，返回 response 和 contexts"""
    # 替换为你的实际 RAG pipeline 调用
    from my_rag_app import query_rag
    result = query_rag(question)
    return {
        "response": result["answer"],
        "contexts": result["source_documents"],
    }
 
# 质量门：各指标的最低阈值
QUALITY_THRESHOLDS = {
    "faithfulness": 0.85,
    "answer_relevancy": 0.80,
}
 
def test_rag_quality():
    """RAG 质量回归测试 — 在 CI/CD 中运行"""
    golden_data = load_golden_dataset("tests/golden_dataset.json")
 
    samples = []
    for item in golden_data:
        result = run_rag_pipeline(item["question"])
        samples.append(SingleTurnSample(
            user_input=item["question"],
            retrieved_contexts=result["contexts"],
            response=result["response"],
            reference=item.get("ground_truth", ""),
        ))
 
    eval_dataset = EvaluationDataset(samples=samples)
    evaluator_llm = LangchainLLMWrapper(
        ChatOpenAI(model="gpt-4o-mini", temperature=0)
    )
 
    results = evaluate(
        dataset=eval_dataset,
        metrics=[
            Faithfulness(llm=evaluator_llm),
            AnswerRelevancy(llm=evaluator_llm),
        ],
    )
 
    # 断言质量门
    for metric_name, threshold in QUALITY_THRESHOLDS.items():
        score = results[metric_name]
        assert score >= threshold, (
            f"{metric_name} 分数 {score:.2f} 低于阈值 {threshold}"
        )

# test_rag_deepeval.py — DeepEval 提供更贴近开发者习惯的测试体验
from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    FaithfulnessMetric,
    AnswerRelevancyMetric,
    ContextualPrecisionMetric,
    ContextualRecallMetric,
)
 
def test_rag_faithfulness():
    """测试 RAG 回答的忠实度"""
    test_case = LLMTestCase(
        input="什么是向量数据库？",
        actual_output="向量数据库是专门存储和检索高维向量的数据库系统，"
                      "常用于 RAG 系统中存储文档嵌入。",
        retrieval_context=[
            "向量数据库是一种专门设计用于存储、索引和查询高维向量数据的数据库。",
            "在 RAG 系统中，向量数据库用于存储文档的嵌入表示。",
        ],
        expected_output="向量数据库是存储和检索高维向量的专用数据库。",
    )
 
    faithfulness = FaithfulnessMetric(threshold=0.8, model="gpt-4o-mini")
    relevancy = AnswerRelevancyMetric(threshold=0.8, model="gpt-4o-mini")
 
    assert_test(test_case, [faithfulness, relevancy])

你是一个 RAG 评估工程师。请帮我分析以下评估结果并给出优化建议：

**评估结果：**
- Faithfulness: [分数]
- Answer Relevancy: [分数]
- Context Precision: [分数]
- Context Recall: [分数]
- 样本数量: [数量]

**系统配置：**
- 嵌入模型: [模型名]
- 分块策略: [策略]
- 检索方式: [向量搜索/混合搜索]
- 生成模型: [模型名]

**请分析：**
1. 各指标的健康度评估（优秀/良好/需改进/严重问题）
2. 最需要优先改进的指标及原因
3. 针对每个低分指标的具体优化建议
4. 预期优化后的指标提升范围
5. 建议的下一步实验计划

from langchain_text_splitters import RecursiveCharacterTextSplitter
from ragas import evaluate
from ragas.metrics import ContextPrecision, ContextRecall
import itertools
 
# 定义参数搜索空间
chunk_sizes = [256, 512, 768, 1024]
chunk_overlaps = [0, 50, 100, 200]
 
results_log = []
 
for size, overlap in itertools.product(chunk_sizes, chunk_overlaps):
    if overlap >= size:
        continue  # 跳过无效组合
 
    # 使用当前参数分块
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", "。", ".", " "],  # 中文友好分隔符
    )
    chunks = splitter.split_documents(documents)
 
    # 重建索引并运行评估
    index = build_index(chunks)  # 你的索引构建函数
    eval_results = run_evaluation(index, golden_dataset)  # 你的评估函数
 
    results_log.append({
        "chunk_size": size,
        "chunk_overlap": overlap,
        "num_chunks": len(chunks),
        "context_precision": eval_results["context_precision"],
        "context_recall": eval_results["context_recall"],
        "faithfulness": eval_results["faithfulness"],
    })
 
# 分析结果，找到最优参数组合
import pandas as pd
df = pd.DataFrame(results_log)
print(df.sort_values("faithfulness", ascending=False).head(5))

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
 
# 语义分块：基于嵌入相似度检测语义边界
semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",  # 或 "standard_deviation"
    breakpoint_threshold_amount=75,  # 相似度低于 75 分位时切分
)
 
semantic_chunks = semantic_splitter.split_documents(documents)
 
# 对比固定分块和语义分块的效果
print(f"固定分块: {len(fixed_chunks)} 块, 平均 {avg_size(fixed_chunks)} tokens")
print(f"语义分块: {len(semantic_chunks)} 块, 平均 {avg_size(semantic_chunks)} tokens")

你是一个 RAG 分块优化专家。请根据以下信息推荐最佳分块策略：

**文档特征：**
- 文档类型：[技术文档/法律合同/客服FAQ/学术论文/产品手册]
- 平均文档长度：[token 数]
- 文档结构：[有清晰标题层级/纯文本/混合格式]
- 语言：[中文/英文/多语言]
- 是否包含代码块：[是/否]
- 是否包含表格：[是/否]

**当前问题：**
- Context Precision: [分数]（目标: [目标值]）
- Context Recall: [分数]（目标: [目标值]）

**请输出：**
1. 推荐的分块策略及理由
2. 具体参数建议（chunk_size, overlap, 分隔符）
3. 是否需要元数据增强（标题、摘要、关键词）
4. 预期的指标提升范围
5. 实验计划（参数搜索空间和评估方法）

# 生成训练数据：(query, positive_doc, negative_doc) 三元组
from openai import OpenAI
import json
 
client = OpenAI()
 
def generate_training_pairs(chunks: list[str], n_queries: int = 3) -> list[dict]:
    """为每个文档块生成训练查询"""
    training_data = []
 
    for i, chunk in enumerate(chunks):
        # 用 LLM 为文档块生成可能的查询
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "system",
                "content": (
                    f"阅读以下文档段落，生成 {n_queries} 个用户可能会问的问题。"
                    "这些问题应该能通过该段落回答。输出 JSON 数组。"
                )
            }, {
                "role": "user",
                "content": chunk
            }],
            response_format={"type": "json_object"},
        )
        questions = json.loads(response.choices[0].message.content).get("questions", [])
 
        for q in questions:
            training_data.append({
                "query": q,
                "positive": chunk,  # 正样本：能回答该查询的文档
                "negative": chunks[(i + 1) % len(chunks)],  # 简单负采样
            })
 
    return training_data
 
training_pairs = generate_training_pairs(document_chunks)
print(f"生成了 {len(training_pairs)} 个训练样本")

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader
 
# 加载基础模型
model = SentenceTransformer("BAAI/bge-small-en-v1.5")
 
# 准备训练数据
train_examples = [
    InputExample(
        texts=[pair["query"], pair["positive"], pair["negative"]]
    )
    for pair in training_pairs
]
 
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
 
# 使用 TripletLoss 微调
train_loss = losses.TripletLoss(model=model)
 
# 训练
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3,
    warmup_steps=100,
    output_path="./fine_tuned_embedding_model",
    show_progress_bar=True,
)
 
# 评估微调效果
from sentence_transformers import evaluation
evaluator = evaluation.InformationRetrievalEvaluator(
    queries=eval_queries,
    corpus=eval_corpus,
    relevant_docs=eval_relevant_docs,
)
score = evaluator(model)
print(f"微调后 NDCG@10: {score['ndcg@10']:.4f}")

# 对比微调前后的 RAG 评估指标
from ragas import evaluate
from ragas.metrics import ContextPrecision, ContextRecall, Faithfulness
 
# 使用原始模型构建索引
original_index = build_index(chunks, model="BAAI/bge-small-en-v1.5")
original_results = run_evaluation(original_index, golden_dataset)
 
# 使用微调模型构建索引
finetuned_index = build_index(chunks, model="./fine_tuned_embedding_model")
finetuned_results = run_evaluation(finetuned_index, golden_dataset)
 
# 对比
print("指标对比：")
print(f"  Context Precision: {original_results['context_precision']:.2f}"
      f" → {finetuned_results['context_precision']:.2f}")
print(f"  Context Recall:    {original_results['context_recall']:.2f}"
      f" → {finetuned_results['context_recall']:.2f}")
print(f"  Faithfulness:      {original_results['faithfulness']:.2f}"
      f" → {finetuned_results['faithfulness']:.2f}")

你是一个嵌入模型微调专家。请评估是否需要微调以及如何执行：

**当前系统：**
- 嵌入模型：[模型名]
- 领域：[技术/医疗/法律/金融/通用]
- 领域术语密度：[高/中/低]
- 当前 Context Precision: [分数]
- 可用训练数据：[文档数量] 篇

**请评估：**
1. 是否需要微调（vs 换用更好的通用模型）
2. 推荐的微调方法（全量微调/LoRA/对比学习）
3. 训练数据准备策略（正负样本生成方法）
4. 训练参数建议（epochs, batch_size, learning_rate）
5. 预期的检索精度提升范围
6. 微调的计算成本和时间估算

┌─────────────────────────────────────────────────────┐
│  Layer 1: 查询结果缓存（精确匹配）                      │
│  命中率: 高（重复查询）  延迟: <1ms  成本节省: 最大       │
│  实现: Redis/Memcached，key=hash(query)              │
├─────────────────────────────────────────────────────┤
│  Layer 2: 语义缓存（相似查询匹配）                      │
│  命中率: 中  延迟: ~10ms  成本节省: 大                  │
│  实现: GPTCache，基于嵌入相似度匹配                      │
├─────────────────────────────────────────────────────┤
│  Layer 3: 嵌入缓存（避免重复嵌入计算）                   │
│  命中率: 高（文档不变时）  延迟: <1ms  成本节省: 中       │
│  实现: CacheBackedEmbeddings                         │
├─────────────────────────────────────────────────────┤
│  Layer 4: 检索结果缓存（缓存 Top-K 文档）               │
│  命中率: 中  延迟: <5ms  成本节省: 中                   │
│  实现: Redis，key=hash(query_embedding)              │
└─────────────────────────────────────────────────────┘

from gptcache import cache
from gptcache.adapter import openai as gptcache_openai
from gptcache.embedding import Onnx
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation
 
# 初始化语义缓存
onnx = Onnx()  # 轻量级嵌入模型用于缓存匹配
cache_base = CacheBase("sqlite")  # 缓存元数据存储
vector_base = VectorBase(
    "faiss",
    dimension=onnx.dimension,
)
data_manager = get_data_manager(cache_base, vector_base)
 
cache.init(
    embedding_func=onnx.to_embeddings,
    data_manager=data_manager,
    similarity_evaluation=SearchDistanceEvaluation(),
)
cache.set_openai_key()
 
# 使用缓存的 OpenAI 调用
# 第一次调用：正常请求 API
response1 = gptcache_openai.ChatCompletion.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "什么是 RAG？"}],
)
 
# 第二次调用（语义相似查询）：命中缓存，零延迟零成本
response2 = gptcache_openai.ChatCompletion.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "RAG 是什么意思？"}],
)
# response2 直接从缓存返回，不调用 API

from langchain_openai import OpenAIEmbeddings
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
 
# 创建本地文件缓存
store = LocalFileStore("./embedding_cache/")
 
# 包装嵌入模型，自动缓存计算结果
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    document_embedding_cache=store,
    namespace="text-embedding-3-small",  # 模型变更时自动失效
)
 
# 第一次嵌入：调用 API 并缓存
vectors = cached_embeddings.embed_documents(["RAG 系统架构...", "向量数据库..."])
 
# 第二次嵌入相同文档：直接从缓存读取，零 API 调用
vectors_cached = cached_embeddings.embed_documents(["RAG 系统架构...", "向量数据库..."])

import redis
import hashlib
import json
from typing import Optional
 
class RAGCache:
    """RAG 系统多层缓存"""
 
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.default_ttl = 3600  # 1 小时过期
 
    def _hash_key(self, prefix: str, content: str) -> str:
        return f"{prefix}:{hashlib.sha256(content.encode()).hexdigest()[:16]}"
 
    # Layer 1: 精确查询缓存
    def get_answer(self, query: str) -> Optional[str]:
        key = self._hash_key("answer", query)
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None
 
    def set_answer(self, query: str, answer: str, contexts: list[str]):
        key = self._hash_key("answer", query)
        self.redis.setex(
            key, self.default_ttl,
            json.dumps({"answer": answer, "contexts": contexts})
        )
 
    # Layer 4: 检索结果缓存
    def get_retrieval(self, query_embedding: list[float]) -> Optional[list]:
        key = self._hash_key("retrieval", str(query_embedding[:8]))
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None
 
    def set_retrieval(self, query_embedding: list[float], docs: list[str]):
        key = self._hash_key("retrieval", str(query_embedding[:8]))
        self.redis.setex(key, self.default_ttl, json.dumps(docs))
 
    # 缓存统计
    def get_stats(self) -> dict:
        info = self.redis.info("stats")
        hits = info.get("keyspace_hits", 0)
        misses = info.get("keyspace_misses", 0)
        total = hits + misses
        return {
            "hit_rate": hits / total if total > 0 else 0,
            "total_keys": self.redis.dbsize(),
            "memory_used_mb": self.redis.info("memory")["used_memory"] / 1024 / 1024,
        }
 
# 使用示例
rag_cache = RAGCache()
 
# 查询时先检查缓存
cached = rag_cache.get_answer(user_query)
if cached:
    return cached  # 缓存命中，直接返回
else:
    answer = run_rag_pipeline(user_query)  # 缓存未命中，执行完整 pipeline
    rag_cache.set_answer(user_query, answer["response"], answer["contexts"])
    return answer

你是一个 RAG 缓存架构师。请设计缓存策略：

**系统特征：**
- 日均查询量：[数量]
- 查询重复率估计：[百分比]
- 平均查询延迟要求：[毫秒]
- 知识库更新频率：[每天/每周/每月]
- 当前月度 API 成本：[$金额]

**请输出：**
1. 推荐的缓存层次和每层的实现方案
2. 缓存失效策略（TTL、主动失效、版本控制）
3. 语义缓存的相似度阈值建议
4. 预期的成本节省和延迟改善
5. 缓存预热策略（高频查询预加载）
6. 监控指标（命中率、内存使用、过期率）

from openai import OpenAI
from enum import Enum
 
class QueryComplexity(Enum):
    SIMPLE = "simple"      # 简单事实查询
    MODERATE = "moderate"  # 中等复杂度
    COMPLEX = "complex"    # 复杂多步推理
 
def classify_query(query: str) -> QueryComplexity:
    """用轻量模型分类查询复杂度"""
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # 用最便宜的模型做分类
        messages=[{
            "role": "system",
            "content": (
                "将查询分类为 simple/moderate/complex。"
                "simple: 单一事实查询。moderate: 需要综合多个信息。"
                "complex: 需要推理、对比或多步分析。只输出分类词。"
            )
        }, {"role": "user", "content": query}],
        max_tokens=10,
    )
    level = response.choices[0].message.content.strip().lower()
    return QueryComplexity(level)
 
def cost_aware_rag(query: str) -> dict:
    """根据查询复杂度选择不同的处理路径"""
    complexity = classify_query(query)
 
    if complexity == QueryComplexity.SIMPLE:
        # 简单查询：小模型 + 少量检索 + 无重排序
        docs = retrieve(query, k=3)
        answer = generate(query, docs, model="gpt-4o-mini")
        # 预估成本: ~$0.001
 
    elif complexity == QueryComplexity.MODERATE:
        # 中等查询：小模型 + 混合搜索 + 轻量重排序
        docs = hybrid_retrieve(query, k=10)
        docs = rerank(query, docs, top_n=5)
        answer = generate(query, docs, model="gpt-4o-mini")
        # 预估成本: ~$0.005
 
    else:
        # 复杂查询：强模型 + 全流程 + 重排序 + 质量检查
        docs = hybrid_retrieve(query, k=20)
        docs = rerank(query, docs, top_n=8)
        answer = generate(query, docs, model="gpt-4o")
        answer = verify_and_retry(query, answer, docs)
        # 预估成本: ~$0.02
 
    return {"answer": answer, "complexity": complexity.value}

import time
from dataclasses import dataclass, field
 
@dataclass
class CostTracker:
    """RAG 系统成本追踪器"""
    daily_budget: float = 50.0  # 每日预算 $50
    monthly_budget: float = 1000.0  # 每月预算 $1000
 
    daily_cost: float = 0.0
    monthly_cost: float = 0.0
    query_count: int = 0
    _costs: list = field(default_factory=list)
 
    # 各组件单价（每千 tokens）
    PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "text-embedding-3-small": {"input": 0.00002},
        "cohere-rerank": {"per_query": 0.002},
    }
 
    def track(self, model: str, input_tokens: int, output_tokens: int = 0):
        pricing = self.PRICING.get(model, {})
        cost = (
            pricing.get("input", 0) * input_tokens / 1000
            + pricing.get("output", 0) * output_tokens / 1000
            + pricing.get("per_query", 0)
        )
        self.daily_cost += cost
        self.monthly_cost += cost
        self.query_count += 1
        self._costs.append({"model": model, "cost": cost, "time": time.time()})
 
        # 预算告警
        if self.daily_cost > self.daily_budget * 0.8:
            self._alert(f"日成本已达预算 80%: ${self.daily_cost:.2f}/${self.daily_budget}")
        if self.monthly_cost > self.monthly_budget * 0.8:
            self._alert(f"月成本已达预算 80%: ${self.monthly_cost:.2f}/${self.monthly_budget}")
 
    def _alert(self, message: str):
        print(f"⚠️ 成本告警: {message}")
        # 实际生产中发送到 Slack/PagerDuty/邮件
 
    @property
    def avg_cost_per_query(self) -> float:
        return self.monthly_cost / max(self.query_count, 1)
 
    def report(self) -> dict:
        return {
            "daily_cost": f"${self.daily_cost:.2f}",
            "monthly_cost": f"${self.monthly_cost:.2f}",
            "query_count": self.query_count,
            "avg_cost_per_query": f"${self.avg_cost_per_query:.4f}",
            "daily_budget_usage": f"{self.daily_cost/self.daily_budget*100:.1f}%",
        }
 
# 使用示例
tracker = CostTracker(daily_budget=50, monthly_budget=1000)
tracker.track("text-embedding-3-small", input_tokens=500)
tracker.track("gpt-4o-mini", input_tokens=2000, output_tokens=500)
print(tracker.report())

from litellm import completion
 
def smart_generate(query: str, context: str, max_budget: float = 0.01) -> str:
    """智能模型路由：根据预算和查询复杂度选择模型"""
 
    # 模型优先级列表（从强到弱）
    models = [
        {"name": "gpt-4o", "cost_per_1k": 0.0025, "quality": "high"},
        {"name": "gpt-4o-mini", "cost_per_1k": 0.00015, "quality": "medium"},
        {"name": "claude-3-5-haiku-latest", "cost_per_1k": 0.00025, "quality": "medium"},
    ]
 
    # 估算 token 数
    estimated_tokens = len(query + context) // 4  # 粗略估算
 
    # 选择预算内最强的模型
    for model in models:
        estimated_cost = model["cost_per_1k"] * estimated_tokens / 1000
        if estimated_cost <= max_budget:
            response = completion(
                model=model["name"],
                messages=[
                    {"role": "system", "content": f"基于以下上下文回答问题。\n\n上下文：{context}"},
                    {"role": "user", "content": query},
                ],
            )
            return response.choices[0].message.content
 
    # 所有模型都超预算，使用最便宜的
    response = completion(
        model=models[-1]["name"],
        messages=[
            {"role": "system", "content": f"基于以下上下文简洁回答。\n\n上下文：{context[:2000]}"},
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content

你是一个 RAG 成本优化顾问。请分析当前成本并提出优化方案：

**当前成本数据：**
- 月度总成本：[$金额]
- 日均查询量：[数量]
- 平均每次查询成本：[$金额]
- 成本构成：嵌入 [%]，存储 [%]，生成 [%]，重排序 [%]

**系统配置：**
- 生成模型：[模型名]
- 嵌入模型：[模型名]
- 向量数据库：[数据库名]
- 是否使用缓存：[是/否]
- 是否使用重排序：[是/否]

**目标：**
- 目标月度成本：[$金额]
- 质量底线：Faithfulness ≥ [分数]

**请输出：**
1. 成本瓶颈分析（哪个环节最贵、为什么）
2. 短期优化（1 周内可实施，预期节省 [%]）
3. 中期优化（1 月内可实施，预期节省 [%]）
4. 长期优化（需要架构调整，预期节省 [%]）
5. 质量-成本权衡分析（降低成本对质量的影响）
6. 月度成本预测模型

阶段 1（第 1 周）：分块优化
  - 从固定 512 token → 递归分块（chunk_size=800, overlap=150）
  - 为产品规格表使用文档结构分块，保持表格完整
  - 效果：Context Precision 0.55 → 0.71

阶段 2（第 2 周）：检索优化
  - 从纯向量搜索 → 混合搜索（alpha=0.7）
  - 添加 Cohere Rerank v3.5（Top-20 → Top-5）
  - 效果：Context Precision 0.71 → 0.82, Context Recall 0.60 → 0.78

阶段 3（第 3 周）：嵌入微调
  - 用 5000 条客服 QA 对微调 bge-small-en-v1.5
  - 效果：Context Recall 0.78 → 0.87

阶段 4（第 4 周）：生成优化 + 缓存
  - 在 system prompt 中强调"仅基于上下文回答，不确定时说明"
  - 部署 Redis 精确缓存 + GPTCache 语义缓存
  - 效果：Faithfulness 0.72 → 0.93, 成本降低 45%