ContinuousEval

ContinuousEval

ContinuousEval은 LLM 기반 애플리케이션을 데이터 기반으로 평가하기 위한 오픈소스 파이썬 패키지예요. 특히 RAG(검색 증강 생성) 파이프라인 평가에 강점을 가진 라이브러리로, 파이프라인의 각 모듈(검색기·리랭커·생성기)을 모듈 단위로 나누어 각각에 맞는 맞춤형 메트릭으로 측정하는 것이 특징이에요.

출처: 문서

본문

ContinuousEval이 다른 평가 도구와 다른 점

• 모듈화된 평가(Modularized Evaluation): 전체 파이프라인을 한 번에 평가하지 않고, 각 모듈을 그에 맞는 메트릭으로 따로 측정해요.
• 종합적인 메트릭 라이브러리: RAG, 코드 생성, 에이전트 도구 사용, 분류 등 다양한 LLM 사용 사례를 아우르며, 결정론적(Deterministic)·의미론적(Semantic)·LLM 기반 메트릭을 자유롭게 조합할 수 있어요.
• 확률적 평가(Probabilistic Evaluation): 확률적 메트릭으로 파이프라인을 평가하기도 해요.

이런 평가에서 LLM 기반 메트릭은 골든 데이터셋(golden dataset)과 함께 소형 언어모델(LLM)을 심판(judge)처럼 활용해 답변의 정확성이나 관련성을 판단하는 방식으로 동작해요. 리트리벌 모듈에서는 Precision/K-재현율(Recall)/F1 같은 메트릭을, 생성 모듈에서는 답변 정확성(AnswerCorrectness) 같은 메트릭을 제공하죠.

설치

PyPI 패키지로 제공되며, 다음 명령으로 설치할 수 있어요.

python3 -m pip install continuous-eval

소스에서 설치하려면 이렇게 해요.

git clone https://github.com/relari-ai/continuous-eval.git && cd continuous-eval
poetry install --all-extras

LLM 기반 메트릭을 실행하려면 .env 파일에 LLM API 키가 하나 이상 필요해요. 예시는 .env.example 파일에서 확인할 수 있어요.

단일 메트릭 실행

하나의 데이터에 대해 단일 메트릭을 실행하는 기본 예시예요.

from continuous_eval.metrics.retrieval import PrecisionRecallF1

datum = {
    "question": "What is the capital of France?",
    "retrieved_context": [
        "Paris is the capital of France and its largest city.",
        "Lyon is a major city in France.",
    ],
    "ground_truth_context": ["Paris is the capital of France."],
    "answer": "Paris",
    "ground_truths": ["Paris"],
}

metric = PrecisionRecallF1()

print(metric(**datum))

데이터셋 전체 평가

EvaluationRunner 클래스를 쓰면 데이터셋 전체에 대해 평가를 돌릴 수 있어요.

from time import perf_counter

from continuous_eval.data_downloader import example_data_downloader
from continuous_eval.eval import EvaluationRunner, SingleModulePipeline
from continuous_eval.eval.tests import GreaterOrEqualThan
from continuous_eval.metrics.retrieval import (
    PrecisionRecallF1,
    RankedRetrievalMetrics,
)


def main():
    # Let's download the retrieval dataset example
    dataset = example_data_downloader("retrieval")

    # Setup evaluation pipeline (i.e., dataset, metrics and tests)
    pipeline = SingleModulePipeline(
        dataset=dataset,
        eval=[
            PrecisionRecallF1().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
            RankedRetrievalMetrics().use(
                retrieved_context=dataset.retrieved_contexts,
                ground_truth_context=dataset.ground_truth_contexts,
            ),
        ],
        tests=[
            GreaterOrEqualThan(
                test_name="Recall", metric_name="context_recall", min_value=0.8
            ),
        ],
    )

    # Start the evaluation manager and run the metrics (and tests)
    tic = perf_counter()
    runner = EvaluationRunner(pipeline)
    eval_results = runner.evaluate()
    toc = perf_counter()
    print("Evaluation results:")
    print(eval_results.aggregate())
    print(f"Elapsed time: {toc - tic:.2f} seconds\n")

    print("Running tests...")
    test_results = runner.test(eval_results)
    print(test_results)


if __name__ == "__main__":
    # It is important to run this script in a new process to avoid
    # multiprocessing issues
    main()

파이프라인(모듈 단위) 평가

시스템이 검색기(Retriever)→리랭커(Reranker)→생성기(LLM)처럼 여러 모듈로 이루어졌다면, 각 모듈을 정의하고 그에 맞는 메트릭을 붙여 파이프라인 단위로 평가할 수 있어요. 이때 Module 객체의 input과 output을 연결해 모듈 간 의존 관계를 표현하고, ModuleOutput으로 각 모듈에서 필요한 값을 추출해요.

from typing import Any, Dict, List

from continuous_eval.data_downloader import example_data_downloader
from continuous_eval.eval import (
    Dataset,
    EvaluationRunner,
    Module,
    ModuleOutput,
    Pipeline,
)
from continuous_eval.eval.result_types import PipelineResults
from continuous_eval.metrics.generation.text import AnswerCorrectness
from continuous_eval.metrics.retrieval import PrecisionRecallF1, RankedRetrievalMetrics


def page_content(docs: List[Dict[str, Any]]) -> List[str]:
    # Extract the content of the retrieved documents from the pipeline results
    return [doc["page_content"] for doc in docs]


def main():
    dataset: Dataset = example_data_downloader("graham_essays/small/dataset")
    results: Dict = example_data_downloader("graham_essays/small/results")

    # Simple 3-step RAG pipeline with Retriever->Reranker->Generation
    retriever = Module(
        name="retriever",
        input=dataset.question,
        output=List[str],
        eval=[
            PrecisionRecallF1().use(
                retrieved_context=ModuleOutput(page_content),  # specify how to extract what we need (i.e., page_content)
                ground_truth_context=dataset.ground_truth_context,
            ),
        ],
    )

    reranker = Module(
        name="reranker",
        input=retriever,
        output=List[Dict[str, str]],
        eval=[
            RankedRetrievalMetrics().use(
                retrieved_context=ModuleOutput(page_content),
                ground_truth_context=dataset.ground_truth_context,
            ),
        ],
    )

    llm = Module(
        name="llm",
        input=reranker,
        output=str,
        eval=[
            AnswerCorrectness().use(
                question=dataset.question,
                answer=ModuleOutput(),
                ground_truth_answers=dataset.ground_truth_answers,
            ),
        ],
    )

    pipeline = Pipeline([retriever, reranker, llm], dataset=dataset)
    print(pipeline.graph_repr())  # visualize the pipeline in marmaid format

    runner = EvaluationRunner(pipeline)
    eval_results = runner.evaluate(PipelineResults.from_dict(results))
    print(eval_results.aggregate())


if __name__ == "__main__":
    main()

참고: 병렬화가 제대로 동작하도록 코드를 main() 함수로 감싸고 if __name__ == "__main__": 가드를 꼭 넣어야 해요.

커스텀 메트릭 만들기

CustomMetric 클래스를 활용하면 LLM-as-a-Judge 방식의 커스텀 메트릭을 간단히 만들 수 있어요. 평가 기준(criteria)과 루브릭(rubric), 그리고 출력 형식을 정의하면 되는 구조예요.

from continuous_eval.metrics.base.metric import Arg, Field
from continuous_eval.metrics.custom import CustomMetric
from typing import List

criteria = "Check that the generated answer does not contain PII or other sensitive information."
rubric = """Use the following rubric to assign a score to the answer based on its conciseness:
- Yes: The answer contains PII or other sensitive information.
- No: The answer does not contain PII or other sensitive information.
"""

metric = CustomMetric(
    name="PIICheck",
    criteria=criteria,
    rubric=rubric,
    arguments={"answer": Arg(type=str, description="The answer to evaluate.")},
    response_format={
        "reasoning": Field(
            type=str,
            description="The reasoning for the score given to the answer",
        ),
        "score": Field(
            type=str, description="The score of the answer: Yes or No"
        ),
        "identifies": Field(
            type=List[str],
            description="The PII or other sensitive information identified in the answer",
        ),
    },
)

# Let's calculate the metric for the first datum
print(metric(answer="John Doe resides at 123 Main Street, Springfield."))

기타

• 포괄적인 메트릭 목록은 공식 문서에서 확인할 수 있어요.
• 사용 통계 추적은 기본 익명으로 켜져 있는데, CONTINUOUS_EVAL_DO_NOT_TRACK 플래그를 true로 설정하면 끌 수 있어요.
• 라이선스는 Apache 2.0이에요.

더 알아보기 (Learn more)

• GitHub 저장소