Оценка RAG: RAGAS метрики

pip install ragas datasets langchain-openai

import json
from openai import OpenAI

client = OpenAI()

CLAIMS_PROMPT = """Извлеки все фактические утверждения из ответа.
Каждое утверждение — отдельный атомарный факт, без интерпретации.

Ответ: {answer}

Верни JSON-массив: ["утверждение 1", "утверждение 2", ...]"""

VERIFY_PROMPT = """Можно ли вывести следующее утверждение из контекста?
Отвечай строго по контексту, не используй внешние знания.

Контекст:
{context}

Утверждение: {claim}

Верни JSON: {{"verdict": "yes" | "no", "reason": "кратко"}}"""


def extract_claims(answer: str) -> list[str]:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": CLAIMS_PROMPT.format(answer=answer)}],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    # LLM может вернуть {"claims": [...]} или просто [...]
    return data if isinstance(data, list) else data.get("claims", [])


def verify_claim(claim: str, context: str) -> bool:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": VERIFY_PROMPT.format(
            context=context, claim=claim
        )}],
        response_format={"type": "json_object"},
    )
    result = json.loads(resp.choices[0].message.content)
    return result.get("verdict") == "yes"


def faithfulness(answer: str, contexts: list[str]) -> float:
    context_text = "\n\n---\n\n".join(contexts)
    claims = extract_claims(answer)
    if not claims:
        return 1.0  # Нет клеймов — нечего проверять
    verified = sum(1 for c in claims if verify_claim(c, context_text))
    return verified / len(claims)


# Использование
score = faithfulness(
    answer="RAG появился в 2020 году. Метод улучшает точность на 30%.",
    contexts=["Lewis et al. (2020) из Facebook AI..."]
)
print(f"Faithfulness: {score:.2f}")  # 0.50

from ragas.metrics import faithfulness
from ragas import evaluate
from datasets import Dataset

data = Dataset.from_dict({
    "question":     ["Когда вышел RAG?"],
    "answer":       ["RAG появился в 2020 году. Метод улучшает точность на 30%."],
    "contexts":     [["Lewis et al. (2020) из Facebook AI..."]],
    "ground_truth": ["RAG появился в 2020 году в работе Lewis et al."],
})

result = evaluate(data, metrics=[faithfulness])
print(result)  # {'faithfulness': 0.50}

import numpy as np
from sentence_transformers import SentenceTransformer

embed_model = SentenceTransformer("intfloat/multilingual-e5-large")

REVERSE_Q_PROMPT = """По этому ответу придумай {n} разных вопросов, на которые он отвечает.
Вопросы должны быть разнообразными, но все — релевантными ответу.

Ответ: {answer}

Верни JSON-массив строк: ["вопрос 1", "вопрос 2", ...]"""


def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))


def answer_relevancy(question: str, answer: str, n: int = 3) -> float:
    # 1. Генерируем обратные вопросы
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": REVERSE_Q_PROMPT.format(
            answer=answer, n=n
        )}],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    gen_questions = data if isinstance(data, list) else list(data.values())[0]

    # 2. Embeddings
    q_emb = embed_model.encode(question, normalize_embeddings=True)
    gen_embs = embed_model.encode(gen_questions, normalize_embeddings=True)

    # 3. Среднее cosine similarity
    sims = [cosine(q_emb, g) for g in gen_embs]
    return float(np.mean(sims))

ATTRIBUTION_PROMPT = """Для каждого предложения из эталонного ответа определи,
можно ли его вывести из контекста. Используй только контекст.

Контекст:
{context}

Предложения эталонного ответа (JSON-массив):
{sentences}

Верни JSON-массив объектов:
[{{"sentence": "...", "attributed": true | false}}, ...]"""


def split_sentences(text: str) -> list[str]:
    """Простая разбивка по '. ' — для продакшена используй spacy/nltk."""
    return [s.strip() for s in text.replace(".\n", ". ").split(". ") if s.strip()]


def context_recall(contexts: list[str], ground_truth: str) -> float:
    sentences = split_sentences(ground_truth)
    if not sentences:
        return 1.0

    context_text = "\n\n---\n\n".join(contexts)
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": ATTRIBUTION_PROMPT.format(
            context=context_text,
            sentences=json.dumps(sentences, ensure_ascii=False),
        )}],
        response_format={"type": "json_object"},
    )
    data = json.loads(resp.choices[0].message.content)
    results = data if isinstance(data, list) else list(data.values())[0]

    attributed = sum(1 for r in results if r.get("attributed", False))
    return attributed / len(results)

RELEVANCE_PROMPT = """Полезен ли этот фрагмент для ответа на вопрос,
принимая во внимание правильный ответ?

Вопрос: {question}
Правильный ответ: {ground_truth}
Фрагмент: {chunk}

Верни JSON: {{"useful": true | false, "reason": "кратко"}}"""


def is_chunk_useful(question: str, chunk: str, ground_truth: str) -> bool:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": RELEVANCE_PROMPT.format(
            question=question, ground_truth=ground_truth, chunk=chunk
        )}],
        response_format={"type": "json_object"},
    )
    result = json.loads(resp.choices[0].message.content)
    return result.get("useful", False)


def context_precision(question: str, contexts: list[str], ground_truth: str) -> float:
    # Оцениваем каждый чанк
    useful = [is_chunk_useful(question, ctx, ground_truth) for ctx in contexts]

    if not any(useful):
        return 0.0

    # Average precision: считаем P@k только на позициях релевантных чанков
    relevant_count = 0
    precision_sum = 0.0
    for k, is_rel in enumerate(useful, start=1):
        if is_rel:
            relevant_count += 1
            precision_sum += relevant_count / k  # precision@k

    return precision_sum / relevant_count

from datasets import Dataset

data = {
    # Обязательно для всех метрик
    "question": [
        "Что такое RAG?",
        "Как работает vector search?",
        "Чем отличается dense от sparse retrieval?",
    ],
    "answer": [
        "RAG — это подход, который совмещает поиск по базе знаний и генерацию LLM.",
        "Vector search находит похожие документы через cosine similarity эмбеддингов.",
        "Dense retrieval использует нейронные эмбеддинги, sparse — TF-IDF и BM25.",
    ],
    # Обязательно для всех метрик
    "contexts": [
        [
            "Retrieval-Augmented Generation (RAG) — архитектура, предложенная Lewis et al.",
            "RAG повышает точность LLM за счёт поиска релевантных документов.",
        ],
        [
            "ANN (Approximate Nearest Neighbor) позволяет искать похожие векторы за O(log n).",
            "FAISS, Qdrant, Chroma реализуют vector search для эмбеддингов.",
        ],
        [
            "BM25 — статистический метод, не требующий обучения.",
            "Dense retrieval требует предобученной модели эмбеддингов.",
        ],
    ],
    # Нужен только для context_recall и context_precision
    "ground_truth": [
        "RAG — метод, сочетающий поиск документов и генерацию ответа языковой моделью.",
        "Vector search использует ANN для поиска ближайших векторов по cosine similarity.",
        "Dense retrieval — нейронные эмбеддинги; sparse — TF-IDF / BM25 без обучения.",
    ],
}

dataset = Dataset.from_dict(data)

from ragas.testset.generator import TestsetGenerator
from ragas.testset.evolutions import simple, reasoning, multi_context
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader

# Загружаем документы
loader = DirectoryLoader("./docs/", glob="**/*.md")
documents = loader.load()

# Создаём генератор
generator = TestsetGenerator.with_openai(
    generator_llm=ChatOpenAI(model="gpt-4o-mini"),  # генерирует вопросы
    critic_llm=ChatOpenAI(model="gpt-4o"),           # оценивает качество
    embeddings=OpenAIEmbeddings(),
)

# Генерируем тестовый датасет
testset = generator.generate_with_langchain_docs(
    documents,
    test_size=100,
    distributions={
        simple: 0.5,         # Простые фактологические вопросы
        reasoning: 0.25,     # Требующие рассуждений
        multi_context: 0.25, # Ответ в нескольких чанках одновременно
    },
    with_debugging_logs=True,
)

df = testset.to_pandas()
print(df[["question", "ground_truth", "evolution_type"]].head())
#                     question         ground_truth  evolution_type
# 0  Что такое chunking?  Chunking — разбивка...       simple
# 1  Почему важен overlap?  Перекрытие позволяет...   reasoning

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)
from ragas.run_config import RunConfig
from datasets import Dataset

# Датасет с ответами вашей RAG-системы
dataset = Dataset.from_dict({
    "question":     [...],
    "answer":       [...],   # ответы вашего RAG
    "contexts":     [...],   # чанки, которые retriever вернул для каждого вопроса
    "ground_truth": [...],
})

# RunConfig управляет параллелизмом и retry-политикой
run_config = RunConfig(
    max_workers=16,     # параллельных LLM-вызовов
    max_wait=120,       # таймаут в секундах на один запрос
    max_retries=3,      # повторных попыток при rate limit
)

result = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
    run_config=run_config,
    raise_exceptions=False,  # продолжать при ошибках отдельных сэмплов
)

print(result)
# {'faithfulness': 0.85, 'answer_relevancy': 0.79,
#  'context_recall': 0.91, 'context_precision': 0.73}

# Детальный анализ по каждому вопросу
df = result.to_pandas()
print(df[["question", "faithfulness", "answer_relevancy"]].sort_values("faithfulness"))

from dataclasses import dataclass
from typing import Any
import asyncio


@dataclass
class RAGSample:
    question: str
    answer: str
    contexts: list[str]
    ground_truth: str | None = None


async def run_rag(question: str, rag_pipeline) -> tuple[str, list[str]]:
    """Запускает RAG и возвращает (ответ, список контекстов)."""
    contexts = await rag_pipeline.retrieve(question)
    answer = await rag_pipeline.generate(question, contexts)
    return answer, [c.page_content for c in contexts]


async def collect_eval_data(
    questions: list[str],
    ground_truths: list[str],
    rag_pipeline,
) -> list[RAGSample]:
    tasks = [run_rag(q, rag_pipeline) for q in questions]
    results = await asyncio.gather(*tasks)

    return [
        RAGSample(
            question=q,
            answer=answer,
            contexts=contexts,
            ground_truth=gt,
        )
        for q, gt, (answer, contexts) in zip(questions, ground_truths, results)
    ]


def build_dataset(samples: list[RAGSample]) -> Dataset:
    return Dataset.from_dict({
        "question":     [s.question for s in samples],
        "answer":       [s.answer for s in samples],
        "contexts":     [s.contexts for s in samples],
        "ground_truth": [s.ground_truth or "" for s in samples],
    })


# Запуск
async def evaluate_rag(rag_pipeline, test_questions, test_answers):
    samples = await collect_eval_data(test_questions, test_answers, rag_pipeline)
    dataset = build_dataset(samples)
    result = evaluate(dataset, metrics=[faithfulness, answer_relevancy,
                                         context_recall, context_precision])
    return result

# Оценка без ground truth — для первого запуска
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy
from datasets import Dataset

result = evaluate(
    Dataset.from_dict({
        "question": questions,
        "answer":   answers,
        "contexts": contexts,
    }),
    metrics=[faithfulness, answer_relevancy],
)
print(result)  # {'faithfulness': 0.85, 'answer_relevancy': 0.79}