Langfuse

Langfuse

Ragas와 Langfuse는 RAG(Retrieval-Augmented Generation) 파이프라인을 평가하고 모니터링하는 데 도움을 주는 강력한 조합이에요.

출처: 문서

본문

Langfuse란 무엇인가?

Langfuse ( GitHub )는 LLM 추적(tracing), 프롬프트 관리, 평가를 위한 오픈소스 플랫폼이에요. trace와 스팬에 점수를 매겨 RAG 파이프라인 성능에 대한 인사이트를 얻을 수 있어요. Langfuse는 OpenAI, LangChain 등을 포함한 다양한 통합을 지원해요.

Langfuse를 Ragas와 함께 사용했을 때의 주요 이점

  • Trace 점수 매기기 : trace와 스팬에 점수를 매겨 RAG 파이프라인 성능에 대한 인사이트 제공
  • 상세 분석 : trace를 세그먼트로 나누고 분석해 낮은 품질 점수를 식별하고 시스템 성능 개선
  • 점수 보고 : 특정 사용 사례와 사용자 세그먼트에 대한 상세 보고서로 드릴다운

Ragas ( GitHub )는 trace/스팬, 특히 RAG 파이프라인에 대한 Model-Based Evaluation을 실행하는 데 도움을 주는 오픈소스 도구예요. Ragas는 RAG 파이프라인의 다양한 측면에 대해 reference-free 평가를 수행할 수 있어요. reference-free이기 때문에 평가를 실행할 때 ground-truth가 필요 없고, Langfuse로 수집한 프로덕션 trace에 대해 실행할 수 있어요.

시작하기

이 가이드는 Ragas와 Langfuse를 사용한 RAG 평가의 end-to-end 예시를 안내해요.

환경

Langfuse에 가입해 API 키를 받으세요.

import os

# get keys for your project from https://cloud.langfuse.com
os.environ["LANGFUSE_SECRET_KEY"] = "sk-..."
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-..."

# your openai key
# os.environ["OPENAI_API_KEY"] = "sk-..."

%pip install datasets ragas llama_index python-dotenv --upgrade

데이터

이 예시에서는 RAG 시스템을 쿼리하고 그 출력을 수집해 이미 준비된 데이터셋을 사용할 거예요. Langfuse에서 프로덕션 데이터를 가져오는 방법은 아래를 참고하세요.

데이터셋은 다음 컬럼을 담아요.

  • question : list[str] - RAG 파이프라인이 평가될 질문들
  • answer : list[str] - RAG 파이프라인에서 생성되어 사용자에게 주어진 답변
  • contexts : list[list[str]] - 질문에 답하기 위해 LLM에 전달된 컨텍스트들
  • ground_truth : list[list[str]] - 질문에 대한 실제 정답. 단, 온라인 평가에서는 ground-truth 데이터에 접근할 수 없으므로 무시할 수 있어요.
from datasets import load_dataset

amnesty_qa = load_dataset("vibrantlabsai/amnesty_qa", "english_v2")["eval"]
amnesty_qa

Found cached dataset amnesty_qa (/home/jjmachan/.cache/huggingface/datasets/vibrantlabs___amnesty_qa/english_v2/2.0.0/d0ed9800191a31943ee52a5c22ee4305e28a33f5edcd9a323802112cff07cc24)

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

Dataset({
    features: ['question', 'ground_truth', 'answer', 'contexts'],
    num_rows: 20
})

메트릭

이 예시에서는 Ragas 라이브러리의 다음 메트릭을 사용할 거예요.

  • faithfulness : 제공된 컨텍스트에 대한 생성된 답변의 사실적 일관성 측정
  • answer_relevancy : 생성된 답변이 주어진 프롬프트에 얼마나 핵심적이고 관련 있는지 평가
  • context precision : 컨텍스트에 존재하는 모든 ground-truth 관련 항목이 더 높은 순위에 있는지 평가하는 메트릭. 이상적으로는 모든 관련 청크가 최상위 순위에 나타나야 해요. 이 메트릭은 질문과 컨텍스트로 계산되며 값의 범위는 0~1로, 점수가 높을수록 정밀도가 좋아요.
  • aspect_critique : 해로움(harmlessness), 정확성 같은 미리 정의된 측면에 따라 제출물을 평가하도록 설계됐어요. 또한 사용자는 자신의 특정 기준에 따라 제출물을 평가하는 측면을 자유롭게 정의할 수 있어요.

이 메트릭들과 동작 방식에 대해 더 알아보려면 문서를 확인하세요.

# import metrics
from ragas.metrics import faithfulness, answer_relevancy, context_precision
from ragas.metrics.critique import SUPPORTED_ASPECTS, harmfulness

# metrics you chose
metrics = [faithfulness, answer_relevancy, context_precision, harmfulness]

다음으로, 선택한 LLM과 임베딩으로 메트릭을 초기화해요. 이 예시에서는 OpenAI를 사용해요.

from ragas.run_config import RunConfig
from ragas.metrics.base import MetricWithLLM, MetricWithEmbeddings


# util function to init Ragas Metrics
def init_ragas_metrics(metrics, llm, embedding):
    for metric in metrics:
        if isinstance(metric, MetricWithLLM):
            metric.llm = llm
        if isinstance(metric, MetricWithEmbeddings):
            metric.embeddings = embedding
        run_config = RunConfig()
        metric.init(run_config)

from langchain_openai.chat_models import ChatOpenAI
from langchain_openai.embeddings import OpenAIEmbeddings

# wrappers
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper

llm = ChatOpenAI()
emb = OpenAIEmbeddings()

init_ragas_metrics(
    metrics,
    llm=LangchainLLMWrapper(llm),
    embedding=LangchainEmbeddingsWrapper(emb),
)

설정

Ragas로 model-based evaluation을 사용하는 방법은 2가지가 있어요.

  • 각 Trace 점수 매기기 : 각 trace 항목에 대해 평가를 실행해요. RAG 파이프라인에 대한 각 호출이 어떻게 수행되는지 훨씬 잘 알 수 있지만 비용이 들 수 있어요
  • 일괄 점수 매기기 : 이 방법에서는 주기적으로 trace의 임의 샘플을 가져와 점수를 매겨요. 비용을 줄이고 앱 성능의 대략적인 추정치를 주지만 중요한 샘플을 놓칠 수 있어요.

이 쿡북에서는 두 가지를 모두 설정하는 방법을 보여줄게요.

Trace 점수 매기기

단일 trace의 작은 예시를 가져와 Ragas로 점수를 매기는 방법을 볼게요. 먼저 데이터를 로드하세요.

row = amnesty_qa[0]
print("question: ", row["question"])
print("answer: ", row["answer"])

question:  What are the global implications of the USA Supreme Court ruling on abortion?
answer:  The global implications of the USA Supreme Court ruling on abortion can be significant, as it sets a precedent for other countries and influences the global discourse on reproductive rights. Here are some potential implications:
...

이제 앱을 계측하기 위해 Langfuse 클라이언트 SDK를 초기화해요.

from langfuse import Langfuse

langfuse = Langfuse()

선택한 메트릭으로 trace에 점수를 매기는 유틸리티 함수를 정의하고 있어요.

async def score_with_ragas(query, chunks, answer):
    scores = {}
    for m in metrics:
        print(f"calculating {m.name}")
        scores[m.name] = await m.ascore(
            row={"question": query, "contexts": chunks, "answer": answer}
        )
    return scores

question, contexts, answer = row["question"], row["contexts"], row["answer"]
await score_with_ragas(question, contexts, answer)

calculating faithfulness
calculating answer_relevancy

Using 'context_precision' without ground truth will be soon depreciated. Use 'context_utilization' instead

calculating context_precision
calculating harmfulness

{'faithfulness': 0.0,
 'answer_relevancy': 0.9999999999999996,
 'context_precision': 0.9999999999,
 'harmfulness': 0}

각 요청과 함께 점수를 계산해요. 아래에서는 다음 단계를 수행하는 더미 애플리케이션을 설명했어요.

  • 사용자로부터 질문 받기
  • 질문에 답하는 데 쓸 수 있는 데이터베이스나 벡터 스토어에서 컨텍스트 가져오기
  • LLM에 질문과 컨텍스트를 전달해 답변 생성하기

이 모든 단계는 Langfuse의 단일 trace 안에서 스팬으로 로깅돼요. trace와 스팬에 대해 더 자세히 알고 싶다면 Langfuse 문서를 읽어 보세요.

# the logic of the dummy application is
# given a question fetch the correspoinding contexts and answers from a dict

import hashlib


def hash_string(input_string):
    return hashlib.sha256(input_string.encode()).hexdigest()


q_to_c = {}  # map between question and context
q_to_a = {}  # map between question and answer
for row in amnesty_qa:
    q_hash = hash_string(row["question"])
    q_to_c[q_hash] = row["contexts"]
    q_to_a[q_hash] = row["answer"]

# if your running this in a notebook - please run this cell
# to manage asyncio event loops
import nest_asyncio

nest_asyncio.apply()

from langfuse.decorators import observe, langfuse_context
from asyncio import run


@observe()
def retriver(question: str):
    return q_to_c[question]


@observe()
def generator(question):
    return q_to_a[question]


@observe()
def rag_pipeline(question):
    q_hash = hash_string(question)
    contexts = retriver(q_hash)
    generated_answer = generator(q_hash)

    # score the runs
    score = run(score_with_ragas(question, contexts, answer=generated_answer))
    for s in score:
        langfuse_context.score_current_trace(name=s, value=score[s])
    return generated_answer

question, contexts, answer = row["question"], row["contexts"], row["answer"]
generated_answer = rag_pipeline(amnesty_qa[0]["question"])

calculating faithfulness
calculating answer_relevancy

Using 'context_precision' without ground truth will be soon depreciated. Use 'context_utilization' instead

calculating context_precision
calculating harmfulness

Langfuse에서 점수 분석

Langfuse UI에서 점수를 분석하고 각 질문·사용자별 점수로 드릴다운할 수 있어요.

→ 아직 Langfuse를 사용하지 않나요? 인터랙티브 데모에서 대시보드를 살펴보세요.

점수 매기기는 blocking 이라서 점수가 계산되기를 기다리기 전에 생성된 답변을 먼저 보냈는지 확인하세요. 또는 score_with_ragas() 를 별도 스레드에서 실행하고 trace_id를 전달해 점수를 로깅할 수도 있어요.

리소스

  • Ragas로 model-based evaluation을 실행하는 방법을 더 배우려면 Model-Based Evaluation 가이드를 확인하세요.
  • 여기에서 LLM 애플리케이션을 분석하고 개선하는 방법을 배워 보세요.

피드백

피드백이나 요청이 있다면 GitHub Issue를 만들거나 Discord 커뮤니티에서 작업을 공유해 주세요.

더 알아보기 (Learn more)