runner.py

"""교열 구성(파이프라인 × 프롬프트 × 모델)을 텍스트 1건에 실행하는 프로그래매틱 러너.

교열 로직을 재구현하지 않고 solar-eval 엔진을 그대로 import 한다 — 데모가
보여주는 동작이 평가 run 과 바이트 단위로 같아야 하기 때문이다 (사본 로직은
반드시 갈라진다). 로컬에서는 uv 워크스페이스의 solar_eval 을, HF Space 에서는
`build_space.py` 가 벤더링한 사본을 쓴다 (sys.path 상 스크립트 옆이 먼저라
Space 에서는 벤더 사본이 이긴다).

**구성은 (pipeline, prompt, model) 한 묶음이다** — 모델만 갈아끼우는 선택지는
두지 않는다. mini4 의 이득은 모델 교체가 아니라 파이프라인·프롬프트 구조 변경에서
나왔고(step1 을 생성에서 판정으로, step2 를 어휘/문법으로 분할, 규칙 층 2개 추가),
옛 combo 프롬프트에 mini4 만 꽂으면 2026-09-16 에 이관 불가 판정을 받은 그 조합이
된다. 그래서 세 축을 `Config` 하나로 묶고, 데모는 그 구성만 노출한다.

config 자산(파이프라인 yaml·프롬프트·치환 사전·화이트리스트)은
  1. `./assets/`          — HF Space 레이아웃 (build_space.py 가 조립)
  2. `../03-evaluation/`  — 레포 레이아웃 (로컬 개발)
순서로 찾는다.
"""

from __future__ import annotations

import asyncio
import time
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml


@dataclass(frozen=True)
class Config:
    """데모가 고를 수 있는 한 구성. `base` 는 pipeline 이 extends 하는 파일(번들 대상)."""

    label: str
    pipeline: str
    prompt: str
    model: str
    note: str
    base: str | None = None


#: 데모에 노출되는 구성. 지금은 하나뿐이지만 tuple 로 둔다 — `build_space.required_assets`
#: 가 이 목록에서 번들 자산을 유도하므로, 구성이 늘어도 조립 코드를 고칠 필요가 없다.
CONFIGS: tuple[Config, ...] = (
    Config(
        label="solar-mini4 파이프라인",
        pipeline="pipeline_dev_v44",  # = v41 + document_chunking
        prompt="prompt_dev_260917_vsplit",
        model="solar-mini4-preview",
        note=(
            "FT 교열 → 판정형 step1(5KB) → 문맥 교정 어휘/문법 2스텝 → 결정론 규칙 6층. "
            "입력 3.3k 토큰."
        ),
        base="pipeline_dev_v40",
    ),
)

#: 라벨 → Config
BY_LABEL: dict[str, Config] = {c.label: c for c in CONFIGS}
DEFAULT_LABEL = CONFIGS[0].label

# 엔진의 StepCallback 과 같은 모양: (event, index, total, step_name, doc_chunk)
StepCallback = Callable[[str, int, int, str, tuple[int, int] | None], None]


def find_config_dir(pipeline: str) -> Path:
    """그 파이프라인 yaml 을 실제로 가진 자산 루트. Space 번들이 레포보다 먼저다."""
    here = Path(__file__).resolve().parent
    for cand in (here / "assets", here.parent / "03-evaluation"):
        if (cand / "pipelines" / f"{pipeline}.yaml").is_file():
            return cand
    raise FileNotFoundError(
        f"config 자산을 찾을 수 없습니다 — {here}/assets 또는 ../03-evaluation 에 "
        f"pipelines/{pipeline}.yaml 이 있어야 합니다 (build_space.py 참조)."
    )


def build_pipeline(label: str = DEFAULT_LABEL) -> Any:
    """구성 하나의 파이프라인 인스턴스. 앱에서 구성별로 1회 만들어 재사용한다."""
    import solar_eval.pipelines.steps  # noqa: F401 — STEP_REGISTRY 등록
    from solar_eval.core.dataset_loader import DatasetLoader
    from solar_eval.core.pipeline_compose import compose_pipeline
    from solar_eval.models.prompt_version import load_step_prompts
    from solar_eval.pipelines.registry import create_pipeline

    cfg = BY_LABEL[label]
    config_dir = find_config_dir(cfg.pipeline)
    pipelines_dir = config_dir / "pipelines"

    def load_base(name: str) -> dict[str, Any]:
        return yaml.safe_load((pipelines_dir / f"{name}.yaml").read_text(encoding="utf-8"))

    composed = compose_pipeline(load_base(cfg.pipeline), load_base=load_base)
    prompts = load_step_prompts(config_dir / "prompts" / cfg.prompt)

    return create_pipeline(
        "multi_step",
        input_fields=["original"],
        pipeline_config=composed,
        prompts=prompts,
        dataset_loader=DatasetLoader(),
        config_dir=config_dir,
    )


def step_names(pipeline: Any) -> list[str]:
    """UI 가 진행 표시를 미리 그릴 수 있게 스텝 이름을 순서대로."""
    return [step.name for step in pipeline.steps]


async def _run_async(
    pipeline: Any, text: str, cfg: Config, on_step: StepCallback | None
) -> dict[str, Any]:
    from solar_eval.models.sample import EvalSample
    from solar_eval.providers.upstage import UpstageProvider

    sample = EvalSample(input={"original": text})
    start = time.monotonic()
    # 평가 run 과 같은 조건: temp 0.0, reasoning off (60초 지연 예산 제약)
    result = await pipeline.run(
        sample,
        prompts="",
        provider=UpstageProvider(),
        model=cfg.model,
        temperature=0.0,
        max_tokens=8000,
        reasoning_effort=None,
        on_step=on_step,
    )
    elapsed = time.monotonic() - start
    return {
        "output": result.output,
        "step_outputs": result.artifacts.get("step_outputs", {}),
        "usage": result.artifacts.get("usage", {}),
        "elapsed_s": elapsed,
        "pipeline_key": cfg.pipeline,
        "prompt_key": cfg.prompt,
        "model": cfg.model,
        "config_label": cfg.label,
    }


def run_proofread(
    pipeline: Any,
    text: str,
    label: str = DEFAULT_LABEL,
    on_step: StepCallback | None = None,
) -> dict[str, Any]:
    """텍스트 1건 교열. {output, step_outputs, usage, elapsed_s, pipeline_key, prompt_key, model, config_label}.

    `pipeline` 은 같은 `label` 로 `build_pipeline` 한 인스턴스여야 한다 — 앱이
    구성별로 캐시한다. `on_step` 은 스텝 시작/종료와 chunk 진행마다 불린다(UI 진행 표시).
    """
    return asyncio.run(_run_async(pipeline, text, BY_LABEL[label], on_step))