runner.py

"""v35 파이프라인(v34 + 문서 단위 chunk)을 텍스트 1건에 실행하는 프로그래매틱 러너.

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

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 pathlib import Path
from typing import Any

import yaml

PIPELINE_NAME = (
    "pipeline_dev_v35"  # = v34 + document_chunking (extends). 두 yaml 이 모두 있어야 한다
)
PROMPT_NAME = "prompt_dev_260825_combo"
DEFAULT_MODEL = "solar-pro4"

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


def find_config_dir() -> Path:
    here = Path(__file__).resolve().parent
    for cand in (here / "assets", here.parent / "03-evaluation"):
        if (cand / "pipelines" / f"{PIPELINE_NAME}.yaml").is_file():
            return cand
    raise FileNotFoundError(
        f"config 자산을 찾을 수 없습니다 — {here}/assets 또는 ../03-evaluation 에 "
        f"pipelines/{PIPELINE_NAME}.yaml 이 있어야 합니다 (build_space.py 참조)."
    )


def build_pipeline() -> Any:
    """v35 파이프라인 인스턴스를 만든다. 앱에서 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

    config_dir = find_config_dir()
    pipelines_dir = config_dir / "pipelines"

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

    raw = load_base(PIPELINE_NAME)
    composed = compose_pipeline(raw, load_base=load_base)
    prompts = load_step_prompts(config_dir / "prompts" / PROMPT_NAME)

    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, model: str, 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 (SP4 교체안 제약)
    result = await pipeline.run(
        sample,
        prompts="",
        provider=UpstageProvider(),
        model=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": PIPELINE_NAME,
        "prompt_key": PROMPT_NAME,
        "model": model,
    }


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

    `on_step` 은 스텝 시작/종료와 LLM 스텝의 bulk 진행마다 불린다 — UI 진행 표시용.
    """
    return asyncio.run(_run_async(pipeline, text, model, on_step))