Case 09 : 실패한 평가를 찾아봅시다

페르소나

Mia는 고객 지원 챗봇의 RAG 파이프라인을 평가하는 AI 엔지니어입니다.

그녀는 20개 테스트 프롬프트에 대해 관련성, 충실도, 답변 길이를 평가합니다.

Mia가 원하는 것은 평균 점수 하나가 아닙니다.

• 어떤 프롬프트가 실패했나요?
• 실패한 프롬프트는 어떤 카테고리에 속하나요?
• 평균 점수가 품질 게이트를 통과해도 개별 실패를 찾을 수 있나요?
• 샘플별 평가 기록을 나중에 다시 조회할 수 있나요?

상황

20개 프롬프트의 평균 관련성 점수는 0.745입니다.

품질 게이트 기준을 통과하지만 7 / 12 / 17번 프롬프트는 관련성 점수가 낮아 실패했습니다.

평균만 보면 세 개의 치명적인 실패가 묻힙니다.

이 케이스의 핵심은 집계 메트릭만 보지 않고, 프롬프트별 샘플 관측과 메트릭을 저장해 실패 샘플을 찾는 것입니다.

Contexta 없이 해결하려면

Contexta 없이 실패 프롬프트를 찾으려면 보통 아래 작업을 직접 해야 합니다.

1. 평가 스크립트가 남긴 JSON 파일이나 DataFrame을 찾습니다.
2. 프롬프트 ID, 카테고리, 점수를 연결하는 파싱 코드를 작성합니다.
3. 관련성 점수가 낮은 샘플을 필터링합니다.
4. 실패 샘플의 원문 프롬프트와 답변을 다시 찾아봅니다.
5. 평가 스위트가 커질 때마다 분석 코드를 유지보수합니다.

평균 점수만 저장하면 어떤 고객 질문이 실제로 실패했는지 답할 수 없습니다.

Contexta로 해결하기

Mia는 각 프롬프트를 SampleObservation으로 기록하고, 샘플별 메트릭을 남깁니다.

run:rag-eval
└─ stage:evaluate
   ├─ sample:prompt-07  relevance=0.12  category=product
   ├─ sample:prompt-12  relevance=0.05  category=escalation
   └─ sample:prompt-17  relevance=0.19  category=account

이 해결 흐름은 세 단계입니다.

단계	Contexta API	얻는 정보
샘플 관측 저장	SampleObservation	프롬프트별 입력과 메타데이터
샘플별 메트릭 저장	MetricRecord	각 프롬프트의 관련성, 충실도, 길이
실패 샘플 조회	ctx.get_run_snapshot(RUN_REF)	낮은 점수의 프롬프트 목록

예제 코드

아래 코드는 20개 프롬프트의 평가 기록 중에서 관련성 점수가 낮은 샘플을 찾습니다.

analyze_prompt_evaluation.py

"""Find failed prompts from previously recorded sample-level metrics."""

from pathlib import Path

from contexta import Contexta
from contexta.config import UnifiedConfig, WorkspaceConfig


PROJECT_NAME = "support-chatbot-rag-eval"
RUN_REF = f"run:{PROJECT_NAME}.eval-run-v1"
CATEGORIES = ["account", "billing", "product", "escalation"]


def prompt_category(sample_name: str) -> str:
    prompt_index = int(sample_name.removeprefix("prompt-"))
    return CATEGORIES[(prompt_index - 1) % len(CATEGORIES)]

ctx = Contexta(
    config=UnifiedConfig(
        project_name=PROJECT_NAME,
        workspace=WorkspaceConfig(root_path=Path(".contexta")),
    )
)

store = ctx.metadata_store
try:
    snapshot = ctx.get_run_snapshot(RUN_REF)
    samples_by_time = {sample.observed_at: sample for sample in snapshot.samples}
    relevance = [
        record
        for record in snapshot.records
        if record.record_type == "metric" and record.key == "relevance"
    ]
    mean_relevance = sum(float(record.value) for record in relevance) / len(relevance)
    failures = [record for record in relevance if float(record.value) < 0.3]

    print(f"Mean relevance: {mean_relevance:.3f}")
    print("Failed prompts:")
    for record in failures:
        sample = samples_by_time[record.observed_at]
        print(
            f"  {sample.name} ({prompt_category(sample.name)}): "
            f"relevance={record.value:.3f}, record={record.record_id}"
        )
finally:
    store.close()

실행하면 다음과 같은 결과를 얻습니다.

Mean relevance: 0.745
Failed prompts:
  prompt-07 (product): relevance=0.120, record=record:support-chatbot-rag-eval.eval-run-v1.r00020
  prompt-12 (escalation): relevance=0.050, record=record:support-chatbot-rag-eval.eval-run-v1.r00035
  prompt-17 (account): relevance=0.190, record=record:support-chatbot-rag-eval.eval-run-v1.r00050

코드 조각별로 이해하기

1. 실행 스냅샷과 샘플 목록 읽기

snapshot = ctx.get_run_snapshot(RUN_REF)
samples_by_time = {sample.observed_at: sample for sample in snapshot.samples}

분석 코드는 평가 실행의 스냅샷을 읽고, 샘플 관측을 observed_at 기준으로 찾을 수 있게 준비합니다.

이렇게 하면 낮은 점수의 metric record를 다시 prompt-07 같은 샘플 이름과 연결할 수 있습니다.

2. 관련성 메트릭과 실패 샘플 필터링하기

relevance = [
    record
    for record in snapshot.records
    if record.record_type == "metric" and record.key == "relevance"
]
failures = [record for record in relevance if float(record.value) < 0.3]

분석 코드는 relevance 메트릭만 모은 뒤 0.3 미만인 record를 실패로 분류합니다.

따라서 평균이 통과해도 낮은 점수의 개별 프롬프트를 찾을 수 있습니다.

3. 실패 프롬프트 이름과 기록 ID 출력하기

for record in failures:
    sample = samples_by_time[record.observed_at]
    print(
        f"  {sample.name} ({prompt_category(sample.name)}): "
        f"relevance={record.value:.3f}, record={record.record_id}"
    )

출력에는 프롬프트 이름, 카테고리, 관련성 점수, record id가 함께 표시됩니다.

따라서 나중에 같은 record id로 실패 평가를 다시 추적할 수 있습니다.

최종 답변

Contexta를 통해, Mia는 앞서 제시된 질문에 이런 식으로 답변할 수 있습니다.

• Q1. 어떤 프롬프트가 실패했나요?

  • A1. 관련성 0.3 미만으로 실패한 프롬프트는 prompt-07, prompt-12, prompt-17가 있습니다.

• Q2. 실패한 프롬프트는 어떤 카테고리에 속하나요?

  • A2. prompt-07은 product, prompt-12는 escalation, prompt-17은 account 카테고리입니다.

• Q3. 평균 점수가 품질 게이트를 통과해도 개별 실패를 찾을 수 있나요?

  • A3. 네. 평균 관련성은 0.745이지만, 샘플별 메트릭을 보면 관련성 0.120, 0.050, 0.190인 실패 프롬프트 3개를 찾을 수 있습니다.

• Q4. 샘플별 평가 기록을 나중에 다시 조회할 수 있나요?

  • A4. 네. 실패 프롬프트는 각각 기록 ID로 다시 조회할 수 있습니다.

따라서, Mia는 평가 리뷰에서 다음과 같이 답할 수 있습니다.

평균 관련성만 보면 괜찮아 보이지만, prompt-07, prompt-12, prompt-17이 관련성 기준에서 실패했습니다.
이 평가 실행은 그대로 승인하지 않고, 해당 프롬프트와 retriever 품질을 먼저 수정해야 합니다.

선택: 예제 데이터 생성

이 섹션은 직접 코드를 실행해 보고 싶은 경우에만 필요합니다.

아래 데이터 준비 코드는 .contexta 워크스페이스에 20개 프롬프트의 샘플 관측과 샘플별 평가 메트릭을 생성합니다.

seed_prompt_evaluation_data.py

"""Create per-prompt evaluation records used by the prompt case study."""

from __future__ import annotations

import tempfile
from pathlib import Path
from typing import Any

from contexta import Contexta
from contexta.config import UnifiedConfig, WorkspaceConfig
from contexta.contract import (
    MetricPayload,
    MetricRecord,
    Project,
    RecordEnvelope,
    Run,
    SampleObservation,
    StageExecution,
    StructuredEventPayload,
    StructuredEventRecord,
)


PROJECT_NAME = "support-chatbot-rag-eval"
STAGE_NAME = "evaluate"
NUM_PROMPTS = 20
# Prompts that have bad relevance (1-based index matching sample names)
FAILING_PROMPT_INDICES = {7, 12, 17}

_REC_COUNTER = 0


def _next_rid() -> str:
    global _REC_COUNTER
    _REC_COUNTER += 1
    return f"r{_REC_COUNTER:05d}"


def _prompt_metrics(idx: int) -> tuple[float, float, int]:
    """Return (relevance, faithfulness, answer_length) for prompt index (1-based)."""
    if idx in FAILING_PROMPT_INDICES:
        # Escalation scenarios -- retriever completely misses
        relevance = round(0.05 + (idx % 3) * 0.07, 3)
        faithfulness = round(0.12 + (idx % 5) * 0.04, 3)
        answer_length = 18 + (idx % 4) * 3
    else:
        # Normal prompts
        relevance = round(0.78 + (idx % 7) * 0.02 + (idx % 3) * 0.01, 3)
        faithfulness = round(0.81 + (idx % 5) * 0.02, 3)
        answer_length = 45 + (idx % 8) * 5
    return relevance, faithfulness, answer_length


def _category(idx: int) -> str:
    categories = ["account", "billing", "product", "escalation"]
    return categories[(idx - 1) % 4]


def run_example(workspace: Path | str | None = None) -> dict[str, Any]:
    """Create 1 run with 20 per-prompt SampleObservations."""

    if workspace is None:
        root = Path(tempfile.mkdtemp(prefix="contexta-case09-"))
        workspace_path = root / ".contexta"
    else:
        workspace_path = Path(workspace)

    ctx = Contexta(
        config=UnifiedConfig(
            project_name=PROJECT_NAME,
            workspace=WorkspaceConfig(root_path=workspace_path),
        )
    )

    store = ctx.metadata_store
    try:
        store.projects.put_project(
            Project(
                project_ref=f"project:{PROJECT_NAME}",
                name=PROJECT_NAME,
                created_at="2025-04-10T00:00:00Z",
                description="RAG customer support chatbot evaluation suite",
            )
        )

        run_name = "eval-run-v1"
        run_ref = f"run:{PROJECT_NAME}.{run_name}"
        started_at = "2025-04-10T10:00:00Z"
        ended_at = "2025-04-10T10:30:00Z"

        store.runs.put_run(
            Run(
                run_ref=run_ref,
                project_ref=f"project:{PROJECT_NAME}",
                name=run_name,
                status="completed",
                started_at=started_at,
                ended_at=ended_at,
            )
        )

        stage_ref = f"stage:{PROJECT_NAME}.{run_name}.{STAGE_NAME}"
        store.stages.put_stage_execution(
            StageExecution(
                stage_execution_ref=stage_ref,
                run_ref=run_ref,
                stage_name=STAGE_NAME,
                status="completed",
                started_at=started_at,
                ended_at=ended_at,
                order_index=0,
            )
        )

        # Register a structured event describing the evaluation suite
        record_store = ctx.record_store
        record_store.append(
            StructuredEventRecord(
                envelope=RecordEnvelope(
                    record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                    record_type="event",
                    recorded_at=started_at,
                    observed_at=started_at,
                    producer_ref="contexta.case09",
                    run_ref=run_ref,
                    stage_execution_ref=stage_ref,
                    completeness_marker="complete",
                    degradation_marker="none",
                ),
                payload=StructuredEventPayload(
                    event_key="eval.suite-registered",
                    level="info",
                    message=f"Evaluation suite: {NUM_PROMPTS} prompts across 4 categories.",
                    attributes={"prompt_count": NUM_PROMPTS, "categories": "account,billing,product,escalation"},
                    origin_marker="explicit_capture",
                ),
            )
        )

        # Create 20 SampleObservations (one per prompt) with metric records
        for idx in range(1, NUM_PROMPTS + 1):
            sample_name = f"prompt-{idx:02d}"
            # sample_observation_ref must equal stage_execution_ref + "." + sample_name
            sample_ref = f"sample:{PROJECT_NAME}.{run_name}.{STAGE_NAME}.{sample_name}"
            obs_ts = f"2025-04-10T10:{idx:02d}:00Z"

            store.samples.put_sample_observation(
                SampleObservation(
                    sample_observation_ref=sample_ref,
                    run_ref=run_ref,
                    stage_execution_ref=stage_ref,
                    sample_name=sample_name,
                    observed_at=obs_ts,
                )
            )

            relevance, faithfulness, answer_length = _prompt_metrics(idx)
            for metric_key, metric_val in [
                ("relevance", relevance),
                ("faithfulness", faithfulness),
                ("answer-length", float(answer_length)),
            ]:
                record_store.append(
                    MetricRecord(
                        envelope=RecordEnvelope(
                            record_ref=f"record:{PROJECT_NAME}.{run_name}.{_next_rid()}",
                            record_type="metric",
                            recorded_at=obs_ts,
                            observed_at=obs_ts,
                            producer_ref="contexta.case09",
                            run_ref=run_ref,
                            stage_execution_ref=stage_ref,
                            sample_observation_ref=sample_ref,
                            completeness_marker="complete",
                            degradation_marker="none",
                        ),
                        payload=MetricPayload(
                            metric_key=metric_key,
                            value=metric_val,
                            value_type="float64",
                        ),
                    )
                )

        return {
            "run_id": run_ref,
            "total_prompts": NUM_PROMPTS,
        }
    finally:
        store.close()


def main() -> None:
    from contextlib import redirect_stdout
    import io

    with redirect_stdout(io.StringIO()):
        run_example(Path(".contexta"))

    print(f"Seeded {PROJECT_NAME} data in .contexta.")


if __name__ == "__main__":
    main()

코드를 seed_prompt_evaluation_data.py로 저장한 뒤, Contexta가 설치된 환경에서 실행하세요.

uv run seed_prompt_evaluation_data.py