Ragas Ragas
stable · 中文译文
中文译文 · 原文：https://docs.ragas.io/en/stable/howtos/integrations/_opik/ · 许可证 Apache-2.0

Comet Opik

本 notebook 展示如何将 Opik 与 Ragas 结合，用于监控和评测 RAG（Retrieval-Augmented Generation）pipeline。

将 Opik 与 Ragas 一起使用主要有两种方式：

  1. 使用 Ragas 指标为 traces 打分
  2. 使用 Ragas evaluate 函数为数据集打分

设置

Comet 提供托管版 Opik 平台，创建账号后即可获取 API Key。

你也可以在本地运行 Opik 平台，更多信息见安装指南。

import getpass
import os

os.environ["OPIK_API_KEY"] = getpass.getpass("Opik API Key: ")
os.environ["OPIK_WORKSPACE"] = input(
    "Comet workspace (often the same as your username): "
)

若在本地运行 Opik 平台，只需设置：

# import os
# os.environ["OPIK_URL_OVERRIDE"] = "http://localhost:5173/api"

准备环境

首先安装必要库，配置 OpenAI API key，并创建一个新的 Opik 数据集。

%pip install opik --quiet

import getpass
import os

os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

将 Opik 与 Ragas 集成

使用 Ragas 指标为 traces 打分

Ragas 提供一组可用于评测 RAG pipeline 质量的指标，包括但不限于：answer_relevancy、answer_similarity、answer_correctness、context_precision、context_recall、context_entity_recall、summarization_score。完整指标列表见 Ragas 文档。

这些指标可以即时计算，并记录到 Opik 的 traces 或 spans。本例中，我们先创建一条简单的 RAG pipeline，再用 answer_relevancy 指标为其打分。

创建 Ragas 指标

若要在不使用 evaluate 函数的情况下使用 Ragas 指标，需要用 RunConfig 对象和 LLM provider 初始化该指标。本例使用 LangChain 作为 LLM provider，并启用 Opik tracer。

我们先初始化 Ragas 指标：

# Import the metric
# Import some additional dependencies
from langchain_openai.chat_models import ChatOpenAI
from langchain_openai.embeddings import OpenAIEmbeddings

from ragas.embeddings import LangchainEmbeddingsWrapper
from ragas.llms import LangchainLLMWrapper
from ragas.metrics import AnswerRelevancy

# Initialize the Ragas metric
llm = LangchainLLMWrapper(ChatOpenAI())
emb = LangchainEmbeddingsWrapper(OpenAIEmbeddings())

answer_relevancy_metric = AnswerRelevancy(llm=llm, embeddings=emb)

指标初始化后，即可用它为一道示例问题打分。由于指标打分是异步的，需要用 asyncio 库来运行打分函数。

# Run this cell first if you are running this in a Jupyter notebook
import nest_asyncio

nest_asyncio.apply()

import asyncio

from ragas.dataset_schema import SingleTurnSample
from ragas.integrations.opik import OpikTracer


# Define the scoring function
def compute_metric(metric, row):
    row = SingleTurnSample(**row)

    opik_tracer = OpikTracer()

    async def get_score(opik_tracer, metric, row):
        score = await metric.single_turn_ascore(row, callbacks=[OpikTracer()])
        return score

    # Run the async function using the current event loop
    loop = asyncio.get_event_loop()

    result = loop.run_until_complete(get_score(opik_tracer, metric, row))
    return result


# Score a simple example
row = {
    "user_input": "What is the capital of France?",
    "response": "Paris",
    "retrieved_contexts": ["Paris is the capital of France.", "Paris is in France."],
}

score = compute_metric(answer_relevancy_metric, row)
print("Answer Relevancy score:", score)

Answer Relevancy score: 1.0

现在打开 Opik，你会看到 Default Project 项目中已创建一条新的 trace。

为 traces 打分

你可以使用 update_current_trace 函数获取当前 trace，并把 feedback scores 传给该函数，从而为 traces 打分。

这种方法的优点是打分 span 会加到这条 trace 上，便于对 RAG pipeline 做更细粒度的分析。不过它会同步运行 Ragas 指标计算，因此可能不适合生产场景。

from opik import track
from opik.opik_context import update_current_trace


@track
def retrieve_contexts(question):
    # Define the retrieval function, in this case we will hard code the contexts
    return ["Paris is the capital of France.", "Paris is in France."]


@track
def answer_question(question, contexts):
    # Define the answer function, in this case we will hard code the answer
    return "Paris"


@track(name="Compute Ragas metric score", capture_input=False)
def compute_rag_score(answer_relevancy_metric, question, answer, contexts):
    # Define the score function
    row = {"user_input": question, "response": answer, "retrieved_contexts": contexts}
    score = compute_metric(answer_relevancy_metric, row)
    return score


@track
def rag_pipeline(question):
    # Define the pipeline
    contexts = retrieve_contexts(question)
    answer = answer_question(question, contexts)

    score = compute_rag_score(answer_relevancy_metric, question, answer, contexts)
    update_current_trace(
        feedback_scores=[{"name": "answer_relevancy", "value": round(score, 4)}]
    )

    return answer


rag_pipeline("What is the capital of France?")

'Paris'

from datasets import load_dataset

from ragas import evaluate from ragas.metrics import answer_relevancy, context_precision, faithfulness

fiqa_eval = load_dataset("vibrantlabsai/fiqa", "ragas_eval")

Reformat the dataset to match the schema expected by the Ragas evaluate function

dataset = fiqa_eval["baseline"].select(range(3))

dataset = dataset.map( lambda x: { "user_input": x["question"], "reference": x["ground_truth"], "retrieved_contexts": x["contexts"], } )

opik_tracer_eval = OpikTracer(tags=["ragas_eval"], metadata={"evaluation_run": True})

result = evaluate( dataset, metrics=[context_precision, faithfulness, answer_relevancy], callbacks=[opik_tracer_eval], )

print(result)

from datasets import load_dataset

from ragas import evaluate
from ragas.metrics import answer_relevancy, context_precision, faithfulness

fiqa_eval = load_dataset("vibrantlabsai/fiqa", "ragas_eval")

# Reformat the dataset to match the schema expected by the Ragas evaluate function
dataset = fiqa_eval["baseline"].select(range(3))

dataset = dataset.map(
    lambda x: {
        "user_input": x["question"],
        "reference": x["ground_truth"],
        "retrieved_contexts": x["contexts"],
    }
)

opik_tracer_eval = OpikTracer(tags=["ragas_eval"], metadata={"evaluation_run": True})

result = evaluate(
    dataset,
    metrics=[context_precision, faithfulness, answer_relevancy],
    callbacks=[opik_tracer_eval],
)

print(result)

Evaluating:   0%|          | 0/6 [00:00<?, ?it/s]


{'context_precision': 1.0000, 'faithfulness': 0.7375, 'answer_relevancy': 0.9889}
← OpenlayerTonic Validate →