问题：公司退款政策的期限是多久？
检索文档：退款申请须在购买后 30 天内提交。
模型回答：您可以在 60 天内申请退款，无需提供购买凭证。
Faithfulness = 0（"60天"和"无需凭证"均无文档支撑）

问题：如何重置密码？
答案：我们的系统使用了行业领先的加密技术，支持多种登录方式，包括邮箱、手机号和第三方账号。安全是我们的首要考虑...
Answer Relevancy = 低（答案没有回答"如何重置密码"）

pip install ragas langchain-openai datasets

from datasets import Dataset

# 构建评估数据集
# 实际使用时，这些数据来自你的 RAG 系统的实际输出
eval_data = {
    "question": [
        "LangGraph 的 Send API 是什么？",
        "RAG 系统中的 Chunking 有什么作用？",
        "向量数据库和关系型数据库有什么区别？",
    ],
    "contexts": [
        # 每个问题对应一个列表，包含检索到的文档片段
        [
            "Send API 是 LangGraph 的动态并行机制，允许在运行时向节点派发任意数量的任务。",
            "Send 对象的第一个参数是目标节点名称，第二个参数是输入状态。",
        ],
        [
            "Chunking 是将长文档切分为较小片段的过程，直接影响检索的粒度和准确性。",
            "合理的 chunk 大小通常在 200-500 token 之间，需要在检索精度和上下文完整性之间取得平衡。",
        ],
        [
            "向量数据库专为高维向量的相似性搜索设计，支持 ANN（近似最近邻）算法。",
            "关系型数据库使用 SQL 进行精确查询，不支持语义相似性搜索。",
            "向量数据库适合语义搜索场景，关系型数据库适合结构化数据的精确查询。",
        ],
    ],
    "answer": [
        "Send API 是 LangGraph 的动态并行派发机制，允许在节点运行时生成任意数量的 Send 对象，每个 Send 对象触发目标节点的一次独立执行。",
        "Chunking 将长文档切分为小片段，使向量检索能够精确命中相关内容。chunk 大小影响检索粒度，过小会丢失上下文，过大会引入噪音。",
        "向量数据库专门为语义相似性搜索设计，使用 ANN 算法；关系型数据库用于结构化数据的精确查询。两者适用场景不同，不是竞争关系。",
    ],
    "ground_truth": [
        "Send API 允许在 LangGraph 节点内部动态创建 Send 对象列表，实现运行时的并行任务派发，是实现 Map-Reduce 模式的核心机制。",
        "Chunking 是 RAG 系统中将文档切分为可检索片段的过程，chunk 的大小和策略直接影响检索质量和最终答案的准确性。",
        "向量数据库针对高维向量相似性搜索优化，关系型数据库针对结构化数据精确查询优化，两者在 RAG 系统中各司其职。",
    ],
}

dataset = Dataset.from_dict(eval_data)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 配置评估使用的 LLM 和 Embedding 模型
# RAGAS 本身也需要 LLM 来进行指标计算
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

ragas_llm = LangchainLLMWrapper(llm)
ragas_embeddings = LangchainEmbeddingsWrapper(embeddings)

# 运行评估
results = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ],
    llm=ragas_llm,
    embeddings=ragas_embeddings,
)

print(results)
# 输出示例：
# {'faithfulness': 0.92, 'answer_relevancy': 0.88,
#  'context_precision': 0.85, 'context_recall': 0.79}

import pandas as pd

# 转换为 DataFrame，查看每个样本的详细得分
df = results.to_pandas()
print(df[["question", "faithfulness", "answer_relevancy",
          "context_precision", "context_recall"]].to_string())

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

def collect_rag_outputs(rag_pipeline, questions: list[str]) -> list[dict]:
    """从 RAG 管道收集输入输出，构建评估样本"""
    samples = []
    for question in questions:
        # 假设 rag_pipeline 返回 (answer, source_documents)
        answer, source_docs = rag_pipeline.invoke(question)

        samples.append({
            "question": question,
            "answer": answer,
            "contexts": [doc.page_content for doc in source_docs],
            # ground_truth 需要人工标注，此处留空
            "ground_truth": None,
        })
    return samples

def generate_eval_questions(documents: list[str], n_questions: int = 50) -> list[str]:
    """从文档库自动生成评估问题"""
    llm = ChatOpenAI(model="gpt-4o-mini")
    questions = []

    for doc in documents[:n_questions]:
        response = llm.invoke([
            HumanMessage(content=f"""基于以下文档内容，生成2个高质量的评估问题：
1. 问题应该能用文档中的信息直接回答
2. 问题应该是用户可能真实提出的
3. 只输出问题，每行一个，不加序号

文档内容：
{doc[:500]}""")
        ])
        new_questions = response.content.strip().split("\n")
        questions.extend([q.strip() for q in new_questions if q.strip()])

    return questions[:n_questions]

from ragas.testset import TestsetGenerator
from langchain_community.document_loaders import DirectoryLoader
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 加载文档
loader = DirectoryLoader("./docs", glob="**/*.md")
documents = loader.load()

# 自动生成测试集（RAGAS 会自动生成问题和 ground truth）
generator = TestsetGenerator.from_langchain(
    generator_llm=ChatOpenAI(model="gpt-4o"),
    critic_llm=ChatOpenAI(model="gpt-4o"),
    embeddings=OpenAIEmbeddings(),
)

testset = generator.generate_with_langchain_docs(
    documents,
    test_size=50,
    distributions={"simple": 0.5, "reasoning": 0.25, "multi_context": 0.25},
)

eval_dataset = testset.to_dataset()

高 Faithfulness + 低 Context Recall
→ 模型只用检索到的文档回答，但检索不完整
→ 优先优化检索模块（分块、嵌入模型、检索策略）

低 Faithfulness + 高 Context Recall
→ 所需信息都检索到了，但模型没有严格遵循文档
→ 优先优化 prompt，增加忠实度约束，考虑换更强的 LLM

低 Context Precision + 低 Answer Relevancy
→ 检索引入大量噪音，导致模型产生了不相关的答案
→ 优先优化检索质量，引入重排序

# eval_pipeline.py
import json
import sys
from datetime import datetime
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# 质量阈值配置
QUALITY_THRESHOLDS = {
    "faithfulness": 0.85,
    "answer_relevancy": 0.80,
    "context_precision": 0.75,
    "context_recall": 0.70,
}

def load_eval_dataset(path: str) -> Dataset:
    with open(path) as f:
        data = json.load(f)
    return Dataset.from_dict(data)

def run_evaluation(dataset_path: str) -> dict:
    dataset = load_eval_dataset(dataset_path)

    results = evaluate(
        dataset=dataset,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )

    scores = dict(results)
    return scores

def check_quality_gates(scores: dict) -> tuple[bool, list[str]]:
    """检查是否通过质量门禁"""
    failures = []
    for metric, threshold in QUALITY_THRESHOLDS.items():
        if metric in scores and scores[metric] < threshold:
            failures.append(
                f"{metric}: {scores[metric]:.3f} < threshold {threshold}"
            )
    return len(failures) == 0, failures

def save_results(scores: dict, output_path: str):
    """保存评估结果用于趋势追踪"""
    record = {
        "timestamp": datetime.now().isoformat(),
        "scores": scores,
    }
    with open(output_path, "a") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

if __name__ == "__main__":
    dataset_path = sys.argv[1] if len(sys.argv) > 1 else "eval_dataset.json"

    print("Running RAGAS evaluation...")
    scores = run_evaluation(dataset_path)

    print("\nEvaluation Results:")
    for metric, score in scores.items():
        status = "PASS" if score >= QUALITY_THRESHOLDS.get(metric, 0) else "FAIL"
        print(f"  {metric}: {score:.3f} [{status}]")

    passed, failures = check_quality_gates(scores)
    save_results(scores, "eval_history.jsonl")

    if not passed:
        print("\nQuality gate FAILED:")
        for failure in failures:
            print(f"  - {failure}")
        sys.exit(1)  # 非零退出码，触发 CI 失败
    else:
        print("\nAll quality gates PASSED.")
        sys.exit(0)

# .github/workflows/rag-eval.yml
name: RAG Quality Evaluation

on:
  pull_request:
    paths:
      - 'rag/**'
      - 'prompts/**'
      - 'indexing/**'

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install ragas langchain-openai datasets
      - name: Run RAG evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python eval_pipeline.py eval_dataset.json