solar_eval/models/sample.py

"""EvalSample — 파이프라인 실행 한 건을 담는 레코드.

RAGAS 의 `EvalSample` 개념을 이 레포 필드명으로 옮긴 것이다. 교열 문맥에서
`user_input`/`response` 같은 RAGAS 이름은 어색해서 우리식(`input`/`output`)으로
바꿨다 (2026-08-09 결정, `.agents/01-plans/harness/2026-08-09-chosun-proofread-evalsample-migration.md` §9).

이 파일은 마이그레이션 단계 A 산출물이다. 단계 C 에서 `pipelines/*` 가, 단계 D/E 에서
`core/runner.py`·`evaluators/*` 가 이 클래스를 실제로 소비하기 시작했다 — `input`/
`output`/`artifacts["step_outputs"]` 는 파이프라인이 채우고, 러너가 저장 직전에
레거시 8+trace 키 dict 로 펼친다(`to_record()` 와는 키 이름이 다르다 — 저장 스키마
하위 호환 때문, `core/runner.py:build_eval_sample_from_result` 참고), 채점기는
`required_fields` 로 선언한 필드가 채워졌는지 확인한 뒤 이 샘플을 그대로 받는다.
`from_row`/`resolve_field_map` 은 아직 러너가 배선하지 않았다 (알려진 한계 —
`input_fields` 가 여러 개인 태스크에서 유도된 field_map 이 첫 필드만 담기 때문,
§5-C 단계 담당자 메모 참고).
"""

from dataclasses import dataclass, field
from typing import Any

# field_map 의 value 로 올 수 있는 두 형태:
#   - str: row 의 컬럼명 하나를 그대로 가리킨다.
#   - dict[str, str]: 여러 컬럼을 {서브키: 컬럼명} 으로 묶어 합성 정답(golden_fields)을
#     만든다. `runner.py` 의 `golden_fields` 처리와 동일한 규칙이다.
FieldMapValue = str | dict[str, str]


@dataclass
class EvalSample:
    """평가 파이프라인 한 샘플의 입력·중간 산출물·출력을 담는 레코드.

    Attributes:
        input: 교열이면 원문, RAG 면 질문, MCQ 면 문항+보기 등 태스크마다 shape 이 다르다.
        contexts: RAG 검색 결과 리스트. 교열/MCQ 는 쓰지 않아 None.
        output: 파이프라인이 채운다. **스텝마다 덮어쓴다** — 최종 스텝의 결과만 남는다.
            스텝별 스냅숏이 필요하면 `artifacts["step_outputs"]` 를 본다.
        reference: 정답(golden). 단일 값이거나(golden_field), 여러 컬럼을 묶은
            dict(golden_fields) 일 수 있다.
        artifacts: 중간 산출물 저장소. 관례상 `artifacts["step_outputs"]` 는
            {스텝 이름: 그 스텝 직후의 output} 로 누적된다.
    """

    input: Any = None
    contexts: list[Any] | None = None
    output: Any = None
    reference: Any = None
    artifacts: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_row(cls, row: dict[str, Any], field_map: dict[str, FieldMapValue]) -> "EvalSample":
        """데이터셋 행 + field_map 으로 샘플을 만든다.

        Args:
            row: 데이터셋 한 행 (jsonl 한 줄을 파싱한 dict).
            field_map: EvalSample 필드명(`input`/`reference`/`contexts`) -> 데이터셋
                컬럼명(str) 또는 {서브키: 컬럼명} dict. `core.project_loader.resolve_field_map`
                이 task 설정에서 이 값을 유도한다.

        Returns:
            field_map 에 있는 키만 채워진 EvalSample (없는 키는 dataclass 기본값 유지).

        Raises:
            TypeError: field_map 의 value 가 str/dict 가 아닌 경우 (조용히 무시하지 않는다).
        """
        kwargs: dict[str, Any] = {}

        for sample_field in ("input", "reference", "contexts"):
            if sample_field not in field_map:
                continue
            resolved = _resolve_field_value(field_map[sample_field], row)
            if sample_field == "contexts" and not isinstance(resolved, list):
                resolved = [resolved]
            kwargs[sample_field] = resolved

        return cls(**kwargs)

    def to_record(self) -> dict[str, Any]:
        """JsonlStore 저장용 직렬화. artifacts 를 포함해 평평한 dict 로 펼친다."""
        return {
            "input": self.input,
            "contexts": self.contexts,
            "output": self.output,
            "reference": self.reference,
            "artifacts": self.artifacts,
        }


def _resolve_field_value(mapping: FieldMapValue, row: dict[str, Any]) -> Any:
    """field_map 의 value 하나(str 또는 dict)를 row 에서 실제 값으로 풀어낸다."""
    if isinstance(mapping, dict):
        return {sub_key: row.get(column, "") for sub_key, column in mapping.items()}
    if isinstance(mapping, str):
        return row.get(mapping, "")
    raise TypeError(
        f"field_map value 는 str 또는 dict[str, str] 이어야 한다: {mapping!r} ({type(mapping)})"
    )