Comet Opik

Comet Opik

이 노트북에서는 Opik을 Ragas와 함께 사용해 RAG(Retrieval-Augmented Generation) 파이프라인을 모니터링·평가하는 방법을 보여줘요. Opik을 Ragas와 함께 쓰는 방법은 크게 두 가지예요.

• Ragas 메트릭으로 trace에 점수 매기기
• Ragas evaluate 함수로 데이터셋에 점수 매기기

출처: 문서

본문

설정

Comet은 Opik 플랫폼의 호스팅 버전을 제공해요. 계정을 만들고 API 키를 받기만 하면 됩니다. 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 키를 구성하고, 새 Opik 데이터셋을 만들어요.

%pip install opik --quiet

import getpass
import os

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

Opik을 Ragas와 통합

Ragas 메트릭으로 trace에 점수 매기기

Ragas는 RAG 파이프라인의 품질을 평가하는 데 쓸 수 있는 메트릭 집합을 제공해요. 여기에는 answer_relevancy, answer_similarity, answer_correctness, context_precision, context_recall, context_entity_recall, summarization_score 등이 포함되지만 이에 국한되지 않아요. 전체 메트릭 목록은 Ragas 문서에서 찾을 수 있어요.

이 메트릭들은 즉석(on the fly)으로 계산해 Opik의 trace나 스팬에 로깅할 수 있어요. 이 예시에서는 간단한 RAG 파이프라인을 만들고 answer_relevancy 메트릭으로 점수를 매겨 볼게요.

Ragas 메트릭 만들기

evaluate 함수를 사용하지 않고 Ragas 메트릭을 사용하려면 RunConfig 객체와 LLM 프로바이더로 메트릭을 초기화해야 해요. 이 예시에서는 Opik tracer를 활성화한 상태로 LangChain을 LLM 프로바이더로 사용할 거예요.

먼저 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가 생성된 것을 볼 수 있어요.

trace에 점수 매기기

update_current_trace 함수를 사용해 현재 trace를 가져오고 피드백 점수를 그 함수에 전달하면 trace에 점수를 매길 수 있어요. 이 접근법의 장점은 trace에 점수 스팬이 추가되어 RAG 파이프라인을 더 세밀하게 분석할 수 있다는 거예요. 다만 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)

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

{'context_precision': 1.0000, 'faithfulness': 0.7375, 'answer_relevancy': 0.9889}

더 알아보기 (Learn more)

• 통합 목록