Phoenix

Phoenix (Arize)

RAG 파이프라인의 기준선(baseline)을 만드는 건 어렵지 않지만, 프로덕션에 적합하게 다듬고 응답 품질을 보장하는 건 거의 항상 어려워요. 선택지가 많을 때 옳은 도구와 파라미터를 고르는 것 자체도 도전이죠. 이 튜토리얼은 RAG를 구축하면서 올바른 선택을 하고 품질을 보장하는 견고한 워크플로우를 공유해요.

출처: 문서

본문

이 글은 오픈소스 라이브러리들을 조합해 RAG를 평가·시각화·분석하는 방법을 다뤄요. 다음을 사용할 거예요.

  • Ragas - 합성 테스트 데이터 생성과 평가
  • Arize AI의 Phoenix - 추적(tracing), 시각화, 클러스터 분석
  • LlamaIndex - RAG 파이프라인 구축

이 글에서는 RAG 파이프라인을 만들기 위해 프롬프트 엔지니어링에 관한 arXiv 논문 데이터를 사용할 거예요.

ℹ️ 이 노트북에는 OpenAI API 키가 필요해요.

2. 의존성 설치 및 라이브러리 import

아래 셀을 실행해 데이터셋 다운로드에 사용하는 Git LFS를 설치하세요.

!git lfs install

Python 의존성을 설치하고 import 하세요.

!pip install "ragas<0.1.1" pypdf arize-phoenix "openinference-instrumentation-llama-index<1.0.0" "llama-index<0.10.0" pandas

import pandas as pd

# Display the complete contents of DataFrame cells.
pd.set_option("display.max_colwidth", None)

3. OpenAI API 키 설정

이미 환경 변수로 설정되어 있지 않다면 OpenAI API 키를 설정하세요.

import os
from getpass import getpass
import openai

if not (openai_api_key := os.getenv("OPENAI_API_KEY")):
    openai_api_key = getpass("🔑 Enter your OpenAI API key: ")
openai.api_key = openai_api_key
os.environ["OPENAI_API_KEY"] = openai_api_key

4. 합성 테스트 데이터셋 생성

평가용 golden 테스트 데이터셋을 큐레이션하는 것은 길고 지루하고 비용이 드는 과정이에요. 특히 시작 단계이거나 데이터 소스가 계속 바뀔 때는 비현실적이죠. 이 문제는 고품질 데이터 포인트를 합성으로 생성해 개발자가 검증하게 하는 방식으로 해결할 수 있어요. 이렇게 하면 테스트 데이터 큐레이션의 시간과 노력을 90% 줄일 수 있어요.

아래 셀을 실행해 프롬프트 엔지니어링 논문 데이터셋을 PDF 형식으로 arXiv에서 다운로드하고 LlamaIndex로 이 문서들을 읽어요.

!git clone https://huggingface.co/datasets/vibrantlabsai/prompt-engineering-papers

from llama_index import SimpleDirectoryReader

dir_path = "./prompt-engineering-papers"
reader = SimpleDirectoryReader(dir_path, num_files_limit=2)
documents = reader.load_data()

이상적인 테스트 데이터셋은 프로덕션에서 관찰되는 분포와 유사한 분포에서 나온 고품질·다양한 데이터 포인트를 담아야 해요. Ragas는 독특한 진화(evolution) 기반 합성 데이터 생성 패러다임으로 가장 높은 품질을 보장하면서도 질문의 다양성까지 보장하는 질문을 생성해요. Ragas는 기본적으로 내부에서 OpenAI 모델을 사용하지만, 원하는 모델을 자유롭게 사용할 수 있어요. Ragas로 100개의 데이터 포인트를 생성해 보겠습니다.

from ragas.testset import TestsetGenerator
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

TEST_SIZE = 25

# generator with openai models
generator_llm = ChatOpenAI(model="gpt-4o-mini")
critic_llm = ChatOpenAI(model="gpt-4o")
embeddings = OpenAIEmbeddings()

generator = TestsetGenerator.from_langchain(generator_llm, critic_llm, embeddings)

# generate testset
testset = generator.generate_with_llamaindex_docs(documents, test_size=TEST_SIZE)
test_df = testset.to_pandas()
test_df.head()

필요에 따라 질문 유형 분포를 자유롭게 바꿀 수 있어요. 이제 테스트 데이터셋이 준비됐으니, LlamaIndex로 간단한 RAG 파이프라인을 만들어 볼게요.

5. LlamaIndex로 RAG 애플리케이션 구축

LlamaIndex는 RAG 애플리케이션 구축을 위한 사용하기 쉽고 유연한 프레임워크예요. 단순함을 위해 기본 LLM(gpt-3.5-turbo)과 임베딩 모델(openai-ada-2)을 사용해요.

Phoenix를 백그라운드로 실행하고 LlamaIndex 애플리케이션을 계측(instrument)해서 여러분의 OpenInference 스팬(span)과 trace가 Phoenix로 전송·수집되게 해요. OpenInference는 OpenTelemetry 위에 구축된 공개 표준으로, LLM 애플리케이션 실행을 캡처하고 저장해요. 벡터 스토어에서의 검색, 검색 엔진이나 API 같은 외부 도구 사용 같은 LLM 실행과 주변 애플리케이션 컨텍스트를 이해하는 데 쓰이는 텔레메트리(telemetry) 데이터 범주로 설계됐어요.

import phoenix as px
from llama_index import set_global_handler

session = px.launch_app()
set_global_handler("arize_phoenix")

쿼리 엔진을 만들어 보세요.

from llama_index.core import VectorStoreIndex, ServiceContext
from llama_index.embeddings.openai import OpenAIEmbedding


def build_query_engine(documents):
    vector_index = VectorStoreIndex.from_documents(
        documents,
        service_context=ServiceContext.from_defaults(chunk_size=512),
        embed_model=OpenAIEmbedding(),
    )
    query_engine = vector_index.as_query_engine(similarity_top_k=2)
    return query_engine


query_engine = build_query_engine(documents)

Phoenix를 확인해 보면 코퍼스 데이터가 인덱싱될 때 나온 임베딩 스팬이 보일 거예요. 나중에 시각화할 수 있도록 그 임베딩들을 DataFrame으로 내보내 저장하세요.

from phoenix.trace.dsl import SpanQuery

client = px.Client()
corpus_df = px.Client().query_spans(
    SpanQuery().explode(
        "embedding.embeddings",
        text="embedding.text",
        vector="embedding.vector",
    )
)
corpus_df.head()

누적된 trace를 지우려면 Phoenix를 재실행하세요.

px.close_app()
session = px.launch_app()

6. LLM 애플리케이션 평가

Ragas는 컴포넌트 단위와 end-to-end로 RAG 파이프라인을 모두 평가하는 데 쓸 수 있는 포괄적인 메트릭 목록을 제공해요.

Ragas를 사용하려면 먼저 질문, 생성된 답변, 검색된 컨텍스트, ground-truth 답변(질문에 대한 실제 기대 답변)으로 구성된 평가 데이터셋을 만들어요.

from datasets import Dataset
from tqdm.auto import tqdm
import pandas as pd


def generate_response(query_engine, question):
    response = query_engine.query(question)
    return {
        "answer": response.response,
        "contexts": [c.node.get_content() for c in response.source_nodes],
    }


def generate_ragas_dataset(query_engine, test_df):
    test_questions = test_df["question"].values
    responses = [generate_response(query_engine, q) for q in tqdm(test_questions)]

    dataset_dict = {
        "question": test_questions,
        "answer": [response["answer"] for response in responses],
        "contexts": [response["contexts"] for response in responses],
        "ground_truth": test_df["ground_truth"].values.tolist(),
    }
    ds = Dataset.from_dict(dataset_dict)
    return ds


ragas_eval_dataset = generate_ragas_dataset(query_engine, test_df)
ragas_evals_df = pd.DataFrame(ragas_eval_dataset)
ragas_evals_df.head()

Phoenix에서 LlamaIndex 애플리케이션 trace를 확인해 보세요.

print(session.url)

나중에 시각화할 임베딩 데이터를 담은 DataFrame과, Ragas로 평가할 내보낸 trace·스팬을 담은 DataFrame을 저장해 두어요.

# dataset containing embeddings for visualization
query_embeddings_df = px.Client().query_spans(
    SpanQuery().explode(
        "embedding.embeddings", text="embedding.text", vector="embedding.vector"
    )
)
query_embeddings_df.head()

from phoenix.session.evaluation import get_qa_with_reference

# dataset containing span data for evaluation with Ragas
spans_dataframe = get_qa_with_reference(client)
spans_dataframe.head()

Ragas는 LangChain을 사용해 LLM 애플리케이션 데이터를 평가해요. 내부에서 무엇이 일어나는지 볼 수 있도록 LangChain을 OpenInference로 계측해 볼게요.

from openinference.instrumentation.langchain import LangChainInstrumentor

LangChainInstrumentor().instrument()

LLM trace를 평가하고 평가 점수를 DataFrame 형식으로 확인하세요.

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_correctness,
    context_recall,
    context_precision,
)

evaluation_result = evaluate(
    dataset=ragas_eval_dataset,
    metrics=[faithfulness, answer_correctness, context_recall, context_precision],
)
eval_scores_df = pd.DataFrame(evaluation_result.scores)

여러분의 스팬에 어노테이션으로 보이도록 평가 결과를 Phoenix에 제출하세요.

from phoenix.trace import SpanEvaluations

# Assign span ids to your ragas evaluation scores (needed so Phoenix knows where to attach the spans).
eval_data_df = pd.DataFrame(evaluation_result.dataset)
assert eval_data_df.question.to_list() == list(
    reversed(spans_dataframe.input.to_list())  # The spans are in reverse order.
), "Phoenix spans are in an unexpected order. Re-start the notebook and try again."
eval_scores_df.index = pd.Index(
    list(reversed(spans_dataframe.index.to_list())), name=spans_dataframe.index.name
)

# Log the evaluations to Phoenix.
for eval_name in eval_scores_df.columns:
    evals_df = eval_scores_df[[eval_name]].rename(columns={eval_name: "score"})
    evals = SpanEvaluations(eval_name, evals_df)
    px.Client().log_evaluations(evals)

Phoenix를 확인하면 애플리케이션 스팬에 Ragas 평가가 어노테이션으로 붙어 있는 걸 볼 수 있어요.

7. 임베딩 시각화 및 분석

임베딩은 검색된 문서와 사용자 쿼리의 의미를 인코딩해요. RAG 시스템의 필수 부분일 뿐 아니라 LLM 애플리케이션 성능을 이해하고 디버깅하는 데도 아주 유용해요.

Phoenix는 RAG 애플리케이션의 고차원 임베딩을 가져와 차원을 줄이고 의미 있는 데이터 그룹으로 클러스터링해요. 그런 다음 원하는 메트릭(예: Ragas로 계산된 faithfulness나 answer correctness)을 선택해 애플리케이션 성능을 시각적으로 검사하고 문제 있는 클러스터를 드러낼 수 있어요. 이 접근법의 장점은 데이터의 세밀하면서도 의미 있는 부분 집합에 메트릭을 제공한다는 거예요. 데이터셋 전반의 전역(global) 성능이 아니라 국소(local) 성능을 분석하는 데 도움을 주죠. LLM 애플리케이션이 어떤 쿼리에서 어려움을 겪는지 직관을 얻는 데도 유용해요.

임베딩 시각화 도구로 Phoenix를 다시 실행해 테스트 데이터셋에서 애플리케이션 성능을 검사할 거예요.

query_embeddings_df = query_embeddings_df.iloc[::-1]
assert ragas_evals_df.question.tolist() == query_embeddings_df.text.tolist()
assert test_df.question.tolist() == ragas_evals_df.question.tolist()
query_df = pd.concat(
    [
        ragas_evals_df[["question", "answer", "ground_truth"]].reset_index(drop=True),
        query_embeddings_df[["vector"]].reset_index(drop=True),
        test_df[["evolution_type"]],
        eval_scores_df.reset_index(drop=True),
    ],
    axis=1,
)
query_df.head()

query_schema = px.Schema(
    prompt_column_names=px.EmbeddingColumnNames(
        raw_data_column_name="question", vector_column_name="vector"
    ),
    response_column_names="answer",
)
corpus_schema = px.Schema(
    prompt_column_names=px.EmbeddingColumnNames(
        raw_data_column_name="text", vector_column_name="vector"
    )
)
# relaunch phoenix with a primary and corpus dataset to view embeddings
px.close_app()
session = px.launch_app(
    primary=px.Dataset(query_df, query_schema, "query"),
    corpus=px.Dataset(corpus_df.reset_index(drop=True), corpus_schema, "corpus"),
)

Phoenix를 실행한 뒤, 다음 단계로 원하는 메트릭으로 데이터를 시각화할 수 있어요.

  • 벡터 임베딩을 선택하고,
  • Color By > dimension 을 선택한 뒤 데이터를 특정 필드로 색칠할 dimension을 선택해요. 예를 들어 faithfulness나 answer correctness 같은 Ragas 평가 점수로요,
  • 메트릭 드롭다운에서 원하는 메트릭을 선택해 클러스터별 집계 메트릭을 봐요.

8. 요약

축하해요! Ragas와 Phoenix로 LlamaIndex 쿼리 엔진을 구축하고 평가했어요. 배운 내용을 정리하면:

  • Ragas로 테스트 데이터셋을 부트스트랩하고 faithfulness, answer correctness 같은 메트릭을 계산해 LlamaIndex 쿼리 엔진을 평가했어요.
  • OpenInference로 쿼리 엔진을 계측해서 LlamaIndex와 Ragas의 내부 동작을 모두 관찰할 수 있었어요.
  • Phoenix로 스팬과 trace를 수집하고, 평가를 import 해 쉽게 검사하며, 임베딩된 쿼리와 검색된 문서를 시각화해 성능이 좋지 않은 부분을 식별했어요.

이 노트북은 Ragas와 Phoenix 기능의 소개일 뿐이에요. 더 배우려면 Ragas와 Phoenix 문서를 확인하세요. 이 튜토리얼이 마음에 들었다면 GitHub에 ⭐를 남겨 주세요.

  • Ragas
  • Phoenix
  • OpenInference

더 알아보기 (Learn more)