deploy: v35 document-level chunking (DP split + parallel chunks), progress view, Supabase logbook

README.md

@@ -10,7 +10,7 @@ pinned: false
 
 # 조선일보 본문 교열 데모
 
-Upstage Solar 기반 7+2-스텝 교열 파이프라인(`pipeline_dev_v34`) 데모입니다.
+Upstage Solar 기반 7+2-스텝 교열 파이프라인(`pipeline_dev_v35` — 문서 단위 chunk) 데모입니다.
 FT 교열 모델 → Solar 3-스텝 교정 → 결정론 규칙(어휘 치환 · 복합어 화이트리스트
 복원 · 율/률 받침 규칙) 순서로 흐르며, 평가 하네스와 같은 엔진 코드를 실행합니다.
 

app.py

@@ -1,7 +1,8 @@
 """조선일보 본문 교열 데모 (Streamlit / HF Space).
 
-v34 파이프라인(prod 7-스텝 + 화이트리스트 복원 + 율/률 규칙) × combo 프롬프트를
+v35 파이프라인(prod 7-스텝 + 화이트리스트 복원 + 율/률 규칙 + 문서 단위 chunk) × combo 프롬프트를
 solar-eval 엔진 그대로 실행한다. 로직은 전부 runner.py — 이 파일은 UI 만 담당.
+사용자 입력과 결과는 logbook.py 가 Supabase 에 남긴다 (시크릿 없으면 조용히 건너뜀).
 """
 
 from __future__ import annotations
@@ -10,19 +11,33 @@ import difflib
 import html
 import os
 
+import logbook
 import runner
 import streamlit as st
 
+PANE_STYLE = (
+    "border:1px solid rgba(128,128,128,.35);border-radius:8px;"
+    "padding:14px 16px;line-height:1.9;font-size:1.02rem;min-height:120px"
+)
+
 SAMPLE_TEXT = (
     "정부의 강력한 부동산 규제에도 불구하고 수도권 아파트 매매 거래 비률은 "
     "오히려 증가세를 보였다. 시중 은행의 주택 담보 대출 금리가 낮아지면서 "
     "실수요자의 구매 심리가 되살아났기 때문이라는 분석이 나온다."
 )
 
-PANE_STYLE = (
-    "border:1px solid rgba(128,128,128,.35);border-radius:8px;"
-    "padding:14px 16px;line-height:1.9;font-size:1.02rem;min-height:120px"
-)
+# 스텝 이름 → 사용자에게 보여줄 라벨 (pipeline_dev_v34 기준. 모르는 스텝은 이름 그대로)
+STEP_LABELS = {
+    "vocabulary_sub_pre": "용어 사전 치환",
+    "proofread": "1차 교열 (교열 전용 모델)",
+    "basic_correction": "기본 교정 (자기일관성 검증)",
+    "context_correction": "문맥 교정",
+    "style_correction": "문체 교정",
+    "vocabulary_reapply": "용어 사전 재적용",
+    "compound_whitelist_restore": "복합어 화이트리스트 복원",
+    "yul_ryul_normalize": "율/률 규칙 적용",
+    "post_process": "후처리",
+}
 
 
 def diff_panes(src: str, dst: str) -> tuple[str, str, int]:
@@ -46,6 +61,74 @@ def diff_panes(src: str, dst: str) -> tuple[str, str, int]:
     return " ".join(left), " ".join(right), n_changed
 
 
+class ProgressView:
+    """스텝 진행을 st.status 안에 그린다 — 단계별 출력은 보여주지 않고 '어디까지 왔나' 만.
+
+    문서 단위 chunk 모드에서는 chunk 들이 **병렬**로 돌아 이벤트가 섞여 들어오므로, chunk 마다 줄을 하나씩
+    두고(몇 단계까지 왔나) 전체 진행 바는 (끝난 단계 수) / (chunk 수 × 단계 수) 로 잡는다.
+    """
+
+    def __init__(self, names: list[str]) -> None:
+        self.names = names
+        self.total_steps = len(names)
+        self.n_chunks = 1
+        self.done: dict[int, int] = {1: 0}  # chunk 번호 → 끝난 단계 수
+        self.current: dict[int, str] = {}  # chunk 번호 → 진행 중 단계 라벨
+        self.status = st.status("교열 중입니다…", expanded=True)
+        self.bar = self.status.progress(0.0, text=f"0 / {self.total_steps} 단계")
+        self.lines = self.status.empty()
+        self._render()
+
+    def on_step(self, event: str, index: int, total: int, name: str, doc_chunk=None) -> None:
+        ci = doc_chunk[0] if doc_chunk else 1
+        if doc_chunk:
+            self.n_chunks = doc_chunk[1]
+        self.done.setdefault(ci, 0)
+        label = STEP_LABELS.get(name, name)
+        if event == "start":
+            self.current[ci] = label
+        elif event == "chunk":
+            self.current[ci] = f"{label} · 조각 {index}/{total}"
+        elif event == "done":
+            self.done[ci] = index
+            self.current.pop(ci, None)
+        finished = sum(self.done.values())
+        grand = self.n_chunks * self.total_steps
+        unit = "단계" if self.n_chunks == 1 else f"단계 (문서 조각 {self.n_chunks}개 병렬)"
+        self.bar.progress(min(finished / grand, 1.0), text=f"{finished} / {grand} {unit}")
+        self._render()
+
+    def _render(self) -> None:
+        if self.n_chunks == 1:
+            done = self.done.get(1, 0)
+            rows = []
+            for i, name in enumerate(self.names, 1):
+                label = STEP_LABELS.get(name, name)
+                if i <= done:
+                    rows.append(f"✅ {label}")
+                elif i == done + 1:
+                    rows.append(f"⏳ **{self.current.get(1, label)}**")
+                else:
+                    rows.append(f"▫️ {label}")
+        else:
+            rows = []
+            for ci in range(1, self.n_chunks + 1):
+                done = self.done.get(ci, 0)
+                if done >= self.total_steps:
+                    rows.append(f"✅ 조각 {ci}: 완료")
+                elif ci in self.current:
+                    rows.append(f"⏳ 조각 {ci}: **{self.current[ci]}** ({done}/{self.total_steps})")
+                else:
+                    rows.append(f"▫️ 조각 {ci}: 대기")
+        self.lines.markdown("  \n".join(rows))
+
+    def finish(self, elapsed_s: float) -> None:
+        self.status.update(label=f"교열 완��� · {elapsed_s:.1f}초", state="complete", expanded=False)
+
+    def fail(self) -> None:
+        self.status.update(label="교열 실패", state="error", expanded=True)
+
+
 @st.cache_resource
 def get_pipeline():
     return runner.build_pipeline()
@@ -74,14 +157,33 @@ if st.button("교열 실행", type="primary", disabled=not text.strip()):
             "UPSTAGE_API_KEY 가 필요합니다 — 사이드바에 입력하거나 Space Secrets 에 설정하세요."
         )
         st.stop()
-    with st.spinner("교열 중입니다…"):
-        try:
-            result = runner.run_proofread(get_pipeline(), text.strip(), model=model)
-        except Exception as e:  # noqa: BLE001 — 데모 표면에서는 원인 요약만 보여준다
-            st.error(f"실행 실패: {type(e).__name__}: {e}")
-            st.stop()
-
-    left, right, _n_changed = diff_panes(text.strip(), result["output"])
+
+    source = text.strip()
+    pipeline = get_pipeline()
+    # 입력은 실행 전에 남긴다 — 실행이 실패해도 무엇이 들어왔는지는 알 수 있게
+    article_id = logbook.save_article(source)
+
+    progress = ProgressView(runner.step_names(pipeline))
+    try:
+        result = runner.run_proofread(pipeline, source, model=model, on_step=progress.on_step)
+    except Exception as e:  # noqa: BLE001 — 데모 표면에서는 원인 요약만 보여준다
+        progress.fail()
+        st.error(f"실행 실패: {type(e).__name__}: {e}")
+        st.stop()
+    progress.finish(result["elapsed_s"])
+
+    logbook.save_run(
+        article_id,
+        pipeline_key=result["pipeline_key"],
+        prompt_key=result["prompt_key"],
+        model=result["model"],
+        output=result["output"],
+        processing_time_s=result["elapsed_s"],
+    )
+    if logbook.is_configured() and logbook.last_error():
+        st.caption(f"기록 실패 (결과에는 영향 없음): {logbook.last_error()}")
+
+    left, right, _n_changed = diff_panes(source, result["output"])
 
     c1, c2 = st.columns(2, gap="medium")
     with c1:

assets/pipelines/pipeline_dev_v35.yaml

@@ -0,0 +1,5 @@
+name: pipeline_dev_v35
+description: "v34 + 문서 단위 chunk — 입구에서 한 번 DP 분할해 chunk 마다 전 스텝(basic_correction 포함)을 돌리고 원장으로 접합. long-article 매칭 ΔF1 −0.9 → +3.5 (2026-09-10, insight 260910)."
+extends: pipeline_dev_v34
+overrides:
+  document_chunking: true

logbook.py

@@ -0,0 +1,96 @@
+"""사용자 입력·교열 결과 기록 — Supabase(PostgREST) 에 직접 REST 호출.
+
+옛 gradio 데모(`services/prj-chosun-gradio/blindtest/db.py`)의 `articles` · `pipeline_runs` 두 테이블을
+그대로 쓴다(스키마는 `schema.sql`). SDK 대신 httpx 로 REST 를 부르는 것도 그때와 같다 — `supabase`
+패키지가 끌고 오는 `realtime`/`websockets` 핀이 다른 의존성과 충돌했던 전력 때문.
+
+환경 변수:
+  SUPABASE_URL — 프로젝트 URL (https://xxx.supabase.co)
+  SUPABASE_KEY (또는 SUPABASE_ANON_KEY) — anon 키 (RLS 정책은 schema.sql)
+
+둘 중 하나라도 없으면 모든 쓰기는 no-op 이고 `is_configured()` 가 False 다. 기록 실패는 절대 데모를
+멈추지 않는다 — `last_error()` 에 한 줄 남기고 None 을 돌려준다.
+"""
+
+from __future__ import annotations
+
+import os
+
+import httpx
+
+TIMEOUT_S = 10.0
+_last_error: str | None = None
+
+
+def _url() -> str:
+    return os.environ.get("SUPABASE_URL", "").strip().rstrip("/")
+
+
+def _key() -> str:
+    # 옛 gradio 데모는 SUPABASE_KEY, 최근 .env 는 SUPABASE_ANON_KEY 를 쓴다 -- 둘 다 받는다
+    return (os.environ.get("SUPABASE_KEY") or os.environ.get("SUPABASE_ANON_KEY") or "").strip()
+
+
+def is_configured() -> bool:
+    return bool(_url() and _key())
+
+
+def last_error() -> str | None:
+    return _last_error
+
+
+def _post(table: str, row: dict) -> list[dict]:
+    headers = {
+        "apikey": _key(),
+        "Authorization": f"Bearer {_key()}",
+        "Content-Type": "application/json",
+        "Prefer": "return=representation",
+    }
+    with httpx.Client(timeout=TIMEOUT_S) as client:
+        resp = client.post(f"{_url()}/rest/v1/{table}", headers=headers, json=row)
+    if resp.status_code >= 300:
+        raise RuntimeError(f"{resp.status_code} {resp.text[:200]}")
+    return resp.json() if resp.content else []
+
+
+def _insert(table: str, row: dict) -> int | None:
+    global _last_error
+    if not is_configured():
+        return None
+    try:
+        data = _post(table, row)
+        _last_error = None
+        return data[0]["id"] if data else None
+    except Exception as exc:  # noqa: BLE001 — 기록은 데모의 부수 기능, 어떤 실패도 삼켜서 한 줄만 남긴다
+        _last_error = f"{table}: {type(exc).__name__}: {exc}"
+        return None
+
+
+def save_article(source_text: str) -> int | None:
+    """사용자 입력 원문 1건. 교열 실행 **전**에 부른다 — 실행이 실패해도 입력은 남게."""
+    return _insert("articles", {"source_text": source_text})
+
+
+def save_run(
+    article_id: int | None,
+    *,
+    pipeline_key: str,
+    prompt_key: str,
+    model: str,
+    output: str,
+    processing_time_s: float,
+) -> int | None:
+    """교열 결과 1건. `article_id` 가 없으면(입력 기록 실패) 결과도 기록하지 않는다."""
+    if article_id is None:
+        return None
+    return _insert(
+        "pipeline_runs",
+        {
+            "article_id": article_id,
+            "pipeline_key": pipeline_key,
+            "prompt_key": prompt_key,
+            "model": model,
+            "output": output,
+            "processing_time_s": round(processing_time_s, 2),
+        },
+    )

runner.py

@@ -1,4 +1,4 @@
-"""v34 파이프라인을 텍스트 1건에 실행하는 프로그래매틱 러너.
+"""v35 파이프라인(v34 + 문서 단위 chunk)을 텍스트 1건에 실행하는 프로그래매틱 러너.
 
 교열 로직을 재구현하지 않고 solar-eval 엔진을 그대로 import 한다 — 데모가
 보여주는 동작이 평가 run 과 바이트 단위로 같아야 하기 때문이다 (사본 로직은
@@ -16,15 +16,21 @@ 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_v34"
+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
@@ -38,7 +44,7 @@ def find_config_dir() -> Path:
 
 
 def build_pipeline() -> Any:
-    """v34 파이프라인 인스턴스를 만든다. 앱에서 1회 만들어 재사용한다."""
+    """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
@@ -65,7 +71,14 @@ def build_pipeline() -> Any:
     )
 
 
-async def _run_async(pipeline: Any, text: str, model: str) -> dict[str, Any]:
+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
 
@@ -80,6 +93,7 @@ async def _run_async(pipeline: Any, text: str, model: str) -> dict[str, Any]:
         temperature=0.0,
         max_tokens=8000,
         reasoning_effort=None,
+        on_step=on_step,
     )
     elapsed = time.monotonic() - start
     return {
@@ -87,9 +101,20 @@ async def _run_async(pipeline: Any, text: str, model: str) -> dict[str, Any]:
         "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) -> dict[str, Any]:
-    """텍스트 1건 교열. {output, step_outputs, usage, elapsed_s} 를 돌려준다."""
-    return asyncio.run(_run_async(pipeline, text, 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))

solar_eval/pipelines/pipeline.py

@@ -13,11 +13,13 @@ rule 후처리 블록이 끼어드는 구조가 이 모델의 원형이다.
   - 실행기가 스텝마다 `step_outputs[스텝이름] = ctx.text` 스냅숏을 찍는다.
 """
 
+import asyncio
 import logging
 import time
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from pathlib import Path
+from collections.abc import Callable
 from typing import Any
 
 from solar_eval.core.dataset_loader import DatasetLoader
@@ -35,6 +37,14 @@ class PipelineEnv:
     prompts: dict[str, str] = field(default_factory=dict)
     input_fields: list[str] = field(default_factory=list)
     text_split_size: int = 500
+    # bulk 분할 전략 -- "dp"(운영) | "greedy"(regex 사다리, ablation 전용). steps/chunking.py 참조.
+    text_split_strategy: str = "dp"
+    # 문서 단위 chunk (실험, 2026-09-10): 입구에서 한 번 DP 분할해 chunk 마다 **전 스텝**을 돌리고 원장으로 접합.
+    # 켜면 분할 없이 전문을 받던 스텝(basic_correction 의 user_template)도 chunk 의 원문만 본다.
+    document_chunking: bool = False
+    # 문서 단위 chunk 를 동시에 몇 개까지 돌릴지. chunk 마다 LLM 호출이 순차로 이어지므로 이 값 × 러너 워커 수가
+    # 동시 API 호출 상한이다 (solar-pro4 는 10 동시부터 429 가 났다 -- README "지뢰").
+    document_chunk_concurrency: int = 4
     dataset_loader: DatasetLoader | None = None
     # 데이터 루트의 프로젝트 디렉토리. v24 의 tool_calling_judge 가 pmi_lookup
     # (1.5MB 파생 통계 — 사람이 편집하지 않는 데이터 자산) 을 풀 때 쓰인다.
@@ -44,6 +54,15 @@ class PipelineEnv:
     config_dir: Path | None = None
 
 
+# 진행 콜백 -- (event, index, total, step_name, doc_chunk). 데모 UI 가 "지금 어느 단계인가" 를 보여주는 용도,
+# 결과에는 아무 영향이 없다.
+#   event: "start" | "done" (스텝), "chunk" (LLM 스텝의 bulk 진행), "document" (문서 단위 chunk 시작, step_name 은 "")
+#   doc_chunk: 문서 단위 chunk 모드에서 (chunk 번호, chunk 수), 아니면 None. chunk 들이 **병렬**로 돌므로
+#   이벤트가 섞여 들어온다 -- UI 는 이 값으로 줄을 나눠야 한다.
+DocChunk = tuple[int, int] | None
+StepCallback = Callable[[str, int, int, str, DocChunk], None]
+
+
 @dataclass
 class StepContext:
     """샘플 1건이 체인을 흐르는 동안의 가변 상태."""
@@ -64,6 +83,8 @@ class StepContext:
     usage: dict[str, int] = field(
         default_factory=lambda: {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
     )
+    on_step: StepCallback | None = None
+    doc_chunk: DocChunk = None  # 문서 단위 chunk 모드에서 이 컨텍스트가 담당하는 chunk
 
     def add_usage(self, usage: dict[str, Any]) -> None:
         for k in self.usage:
@@ -110,23 +131,46 @@ class Pipeline(BasePipeline):
         max_tokens: int = 8000,
         reasoning_effort: str | None = None,
         messages: list[dict[str, str]] | None = None,
+        on_step: StepCallback | None = None,
     ) -> EvalSample:
         input_data = sample.input if isinstance(sample.input, dict) else {}
         original_text = input_data.get("original", "")
-        ctx = StepContext(
-            provider=provider,
-            model=model,
-            temperature=temperature,
-            max_tokens=max_tokens,
-            reasoning_effort=reasoning_effort,
-            system_prompt=prompts,
-            messages=messages,
-            original_text=original_text,
-            text=original_text,
-            shared=self._shared,
-        )
         start = time.monotonic()
 
+        def make_ctx(text: str, doc_chunk: DocChunk = None) -> StepContext:
+            return StepContext(
+                provider=provider,
+                model=model,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                reasoning_effort=reasoning_effort,
+                system_prompt=prompts,
+                messages=messages,
+                original_text=text,
+                text=text,
+                shared=self._shared,
+                on_step=on_step,
+                doc_chunk=doc_chunk,
+            )
+
+        if self.env.document_chunking and len(original_text) > self.env.text_split_size:
+            await self._run_by_document_chunks(sample, input_data, original_text, make_ctx, on_step)
+        else:
+            ctx = make_ctx(original_text)
+            await self._run_chain(sample, ctx, on_step)
+            sample.output = ctx.text
+            sample.artifacts["usage"] = ctx.usage
+            sample.artifacts["step_outputs"] = ctx.step_outputs
+
+        sample.artifacts["inference_time_ms"] = (time.monotonic() - start) * 1000
+        # completion 스텝은 자기 finish_reason 을 직접 남긴다 -- 없을 때만 기본값.
+        sample.artifacts.setdefault("finish_reason", "stop")
+        return sample
+
+    async def _run_chain(
+        self, sample: EvalSample, ctx: StepContext, on_step: StepCallback | None
+    ) -> None:
+        """스텝 목록을 순서대로 한 번 돈다. `ctx.text` 가 결과, `ctx.step_outputs` 가 스냅숏."""
         from datetime import datetime
 
         for step_i, step in enumerate(self.steps):
@@ -135,6 +179,8 @@ class Pipeline(BasePipeline):
                 f"  [{ts}] step {step_i + 1}/{len(self.steps)} "
                 f"'{step.name}' ({step.config.get('type', '?')}) start"
             )
+            if on_step:
+                on_step("start", step_i + 1, len(self.steps), step.name, ctx.doc_chunk)
             try:
                 await step.run(sample, ctx)
             except Exception as e:
@@ -146,13 +192,64 @@ class Pipeline(BasePipeline):
             ctx.step_outputs[step.name] = ctx.text
             ts = datetime.now().strftime("%H:%M:%S")
             logger.info(f"  [{ts}] step {step_i + 1} '{step.name}' done, text_len={len(ctx.text)}")
+            if on_step:
+                on_step("done", step_i + 1, len(self.steps), step.name, ctx.doc_chunk)
 
-        elapsed_ms = (time.monotonic() - start) * 1000
+    async def _run_by_document_chunks(
+        self,
+        sample: EvalSample,
+        input_data: dict[str, Any],
+        original_text: str,
+        make_ctx: Callable[[str, DocChunk], StepContext],
+        on_step: StepCallback | None,
+    ) -> None:
+        """문서를 입구에서 한 번 나눠 chunk 마다 전 스텝을 돌리고, 출력·스냅숏·사용량을 원장으로 합친다."""
+        # steps/__init__ 가 llm.py 를 거쳐 이 모듈을 import 하므로 순환을 피해 지연 import
+        from solar_eval.pipelines.steps.chunking import join_chunks, plan_chunks
+
+        plan = plan_chunks(
+            original_text, self.env.text_split_size, strategy=self.env.text_split_strategy
+        )
+        n_chunks = len(plan.chunks)
+        limit = asyncio.Semaphore(max(1, int(self.env.document_chunk_concurrency)))
+
+        async def run_chunk(ci: int, chunk: str) -> EvalSample:
+            # chunk 끼리는 독립이라 동시에 돈다 -- 캐시(self._shared)는 읽기 전용으로만 공유
+            doc_chunk: DocChunk = (ci + 1, n_chunks)
+            async with limit:
+                if on_step:
+                    on_step("document", ci + 1, n_chunks, "", doc_chunk)
+                sub = EvalSample(input={**input_data, "original": chunk})
+                ctx = make_ctx(chunk, doc_chunk)
+                await self._run_chain(sub, ctx, on_step)
+                sub.output = ctx.text
+                sub.artifacts["usage"] = ctx.usage
+                sub.artifacts["step_outputs"] = ctx.step_outputs
+                return sub
+
+        subs: list[EvalSample] = list(
+            await asyncio.gather(*(run_chunk(ci, c) for ci, c in enumerate(plan.chunks)))
+        )
 
-        sample.output = ctx.text
-        sample.artifacts["usage"] = ctx.usage
-        sample.artifacts["inference_time_ms"] = elapsed_ms
-        # completion 스텝은 자기 finish_reason 을 직접 남긴다 -- 없을 때만 기본값.
-        sample.artifacts.setdefault("finish_reason", "stop")
-        sample.artifacts["step_outputs"] = ctx.step_outputs
-        return sample
+        sample.output = join_chunks([sub.output for sub in subs], plan)
+        sample.artifacts["step_outputs"] = {
+            name: join_chunks([sub.artifacts["step_outputs"][name] for sub in subs], plan)
+            for name in subs[0].artifacts["step_outputs"]
+        }
+        usage = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
+        for sub in subs:
+            for k in usage:
+                usage[k] += sub.artifacts["usage"].get(k, 0)
+        sample.artifacts["usage"] = usage
+        if any(sub.artifacts.get("finish_reason", "stop") != "stop" for sub in subs):
+            sample.artifacts["finish_reason"] = "length"
+        corrections: dict[str, list[Any]] = {}
+        for sub in subs:
+            for step_name, items in sub.artifacts.get("corrections", {}).items():
+                corrections.setdefault(step_name, []).extend(items)
+        if corrections:
+            sample.artifacts["corrections"] = corrections
+        sample.artifacts["document_chunks"] = {
+            "lens": [len(c) for c in plan.chunks],
+            "kinds": list(plan.kinds),
+        }

solar_eval/pipelines/registry.py

@@ -113,6 +113,9 @@ def create_pipeline(
         prompts=prompts or {},
         input_fields=input_fields,
         text_split_size=config.get("text_split_size", 500),
+        text_split_strategy=config.get("text_split_strategy", "dp"),
+        document_chunking=bool(config.get("document_chunking", False)),
+        document_chunk_concurrency=int(config.get("document_chunk_concurrency", 4)),
         dataset_loader=dataset_loader,
         project_dir=Path(project_dir) if project_dir else None,
         config_dir=Path(config_dir) if config_dir else None,

solar_eval/pipelines/steps/chunking.py

@@ -0,0 +1,294 @@
+"""bulk 분할 계획 -- 상한 안에서 문단 > 문장 > 어절 경계만 자르고, 구분자를 원장으로 남겨 접합 때 복원.
+
+왜 이 모듈인가: 옛 `_split_text` 는 500번째 **문자**에서 무조건 잘랐다. 절단면이 공백이면
+`.rstrip()` + `"".join` 이 공백을 삼키고(`미국 제너럴다이내믹스`→`미국제너럴다이내믹스`), 어절
+중간이면 모델이 뒤 조각을 완성하거나(`들이`→`이들이`) 앞 조각을 버렸고(`이뤄졌다. 지난`→
+`이뤄졌다.난`), `\\n`→`∥PARA∥` 왕복은 마커 한가운데를 잘라 리터럴을 출력에 남겼다. 상세는
+`accounts/chosun/body-proofread/03-evaluation/insights/260910_long_article_regression_set.md`.
+
+왜 greedy(상한까지 채우고 가장 가까운 경계) 가 아니라 DP 인가:
+- greedy 는 꼬리 chunk 를 수십 자로 남긴다 (1022자 → 500/500/22). 조각만 받은 모델은 헛손질한다.
+- 따옴표가 열린 채 끝나는 문장 경계는 약하다 -- 뒤 chunk 가 닫는 따옴표로 시작하면 모델이 "고친다".
+  후보 경계마다 종류별 비용을 두고 chunk 수·짧은 chunk 벌점까지 합쳐 전체 최소를 고르면 둘 다 풀린다.
+
+상한(`max_len`)은 파인튜닝 모델이 잘 도는 구간이라 **절대 넘기지 않는다**. 상한 이하 텍스트는 손대지
+않고 그대로 1개 chunk 로 돌려준다 (기존 평가셋 결과가 바이트 단위로 보존되는 조건).
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from dataclasses import dataclass
+
+WS_RUN = re.compile(r"\s+")
+SENTENCE_END = ".!?…。！？"
+# 문장 끝 부호 뒤에 붙을 수 있는 닫는 기호 -- `했다.”`, `있다)`, `한다.’` 도 문장 경계로 본다
+TRAILING_CLOSERS = "”’\"')]」』》〉"
+
+# 비용은 "선호 순서와 교환 비율" 이다. v2 (2026-09-10 ablation 반영): 문단 정렬이 호출 1회보다 우선한다 --
+# v1(chunk 1.0 > sent 0.15) 은 호출을 아끼려 문장 경계로 ~450자까지 채웠는데, 그 큰 다문단 chunk 가
+# 문단 단위 chunk 보다 recall 을 5점 깎았다(long-article 매칭 ΔF1 -1.4 vs greedy +0.4). 어절 절단·꼬리·
+# 따옴표 안 절단 회피는 그대로다.
+KIND_COST: dict[str, float] = {
+    "para": 0.0,  # 문단 경계 (공백 런에 개행 포함) -- 공짜. 문단이 상한에 들어가는 한 문단째로 자른다
+    "sent": 0.5,  # 문장 끝 부호 + 공백 -- 호출 하나(0.1)보다 비싸다: 문단을 쪼개느니 chunk 를 하나 더 쓴다
+    "quoted_sent": 1.0,  # 따옴표/괄호가 열린 채 만난 문장 끝 -- 뒤 chunk 가 닫는 기호로 시작한다
+    "word": 3.0,  # 어절 사이 공백 -- 문장 한가운데
+    "char": 5.0,  # 공백 없는 런이 상한보다 길 때만 쓰는 강제 절단
+}
+CHUNK_COST = 0.1  # chunk 하나 = LLM 호출 하나. 같은 경계 품질이면 호출이 적은 쪽 (tie-breaker 수준)
+SHORT_PENALTY = 3.0  # `min_len` 미만 chunk 벌점 계수 (부족 비율의 제곱에 곱한다) -- 꼬리 방지
+DEFAULT_MIN_LEN = 150
+
+
+@dataclass(frozen=True)
+class Boundary:
+    """chunk 사이 경계. `text[start:end]` 가 구분자(공백 런, 강제 절단이면 빈 문자열)."""
+
+    start: int
+    end: int
+    kind: str
+
+
+@dataclass(frozen=True)
+class ChunkPlan:
+    """`join_chunks(plan.chunks, plan)` 이 원문을 그대로 돌려주는 분할 계획."""
+
+    chunks: list[str]
+    seps: list[str]  # chunks 와 같은 길이. seps[i] 는 chunks[i] 뒤 구분자, 마지막은 trailing 공백
+    kinds: list[str]  # 내부 경계 종류 (len == len(chunks) - 1)
+    lead: str  # 원문 leading 공백
+
+    @property
+    def is_split(self) -> bool:
+        return len(self.chunks) > 1
+
+
+STRATEGIES = ("dp", "greedy")
+
+
+def plan_chunks(
+    text: str, max_len: int, min_len: int = DEFAULT_MIN_LEN, strategy: str = "dp"
+) -> ChunkPlan:
+    """`text` 를 `max_len` 이하 chunk 들로 나누는 계획을 세운다. 상한 이하면 그대로 1개.
+
+    `strategy="greedy"` 는 "상한까지 채우고 그 안에서 가장 뒤의 좋은 경계" 라는 단순 사다리 —
+    DP 와의 ablation 용이지 운영값이 아니다 (꼬리 chunk · 따옴표 안 절단을 못 막는다).
+    """
+    if max_len <= 0:
+        raise ValueError(f"max_len 은 양수여야 한다: {max_len}")
+    if strategy not in STRATEGIES:
+        raise ValueError(f"strategy 는 {STRATEGIES} 중 하나여야 한다: {strategy!r}")
+    if len(text) <= max_len:
+        return ChunkPlan(chunks=[text], seps=[""], kinds=[], lead="")
+
+    lead_len = len(text) - len(text.lstrip())
+    body = text.strip()
+    lead, trail = text[:lead_len], text[lead_len + len(body) :]
+    if len(body) <= max_len:
+        return ChunkPlan(chunks=[body], seps=[trail], kinds=[], lead=lead)
+
+    boundaries = _candidate_boundaries(body)
+    boundaries = _add_forced_cuts(body, boundaries, max_len)
+    if strategy == "greedy":
+        cut_ids = _solve_greedy(boundaries, len(body), max_len)
+    else:
+        cut_ids = _solve(body, boundaries, max_len, min_len)
+
+    chunks, seps, kinds = [], [], []
+    pos = 0
+    for b in (boundaries[i] for i in cut_ids):
+        chunks.append(body[pos : b.start])
+        seps.append(body[b.start : b.end])
+        kinds.append(b.kind)
+        pos = b.end
+    chunks.append(body[pos:])
+    seps.append(trail)
+    return ChunkPlan(chunks=chunks, seps=seps, kinds=kinds, lead=lead)
+
+
+def join_chunks(outputs: list[str], plan: ChunkPlan) -> str:
+    """LLM 을 거친 chunk 출력들을 원장의 구분자로 다시 잇는다. 출력 가장자리 공백은 모델 잡음이라 버린다."""
+    if len(outputs) != len(plan.chunks):
+        raise ValueError(f"chunk 수 불일치: outputs={len(outputs)} plan={len(plan.chunks)}")
+    if not plan.is_split:
+        return outputs[0]
+    return plan.lead + "".join(
+        out.strip() + sep for out, sep in zip(outputs, plan.seps, strict=True)
+    )
+
+
+# ── 출력 위생 ────────────────────────────────────────────────────────────
+
+# 토큰 앞뒤의 공백·개행까지 흡수한다 -- 모델이 토큰과 개행을 같이 내면 개행이 겹친다 (실측: 문단 +1)
+SEPARATOR_LITERAL = re.compile(r"\s*<paragraph_separator>\s*")
+LITERAL = "<paragraph_separator>"
+ANCHOR_LEN = 20
+
+
+def normalize_separator_literals(text: str, source: str | None = None) -> str:
+    """모델이 지어낸 `<paragraph_separator>` 리터럴을 그 자리의 원래 구분자로 되돌린다.
+
+    step1~3 프롬프트는 이 토큰으로 문단 경계를 표시하라고 하지만 엔진은 주입도 제거도 하지 않는다
+    (계약 불일치 -- insight 260910 참조). 그래서 모델이 가끔 토큰을 그대로 출력에 남기고, 문단 정렬
+    chunk(DP v2) 에서 빈도가 올라간다(실기사 6건 run 당 3건). 계약을 (A) 주입/복원으로 살리든 (B)
+    프롬프트에서 빼든, 최종 텍스트에 리터럴이 남는 건 어느 쪽에서도 옳지 않다 -- 이 가드는 그 결정과
+    독립이다.
+
+    무엇으로 바꾸나: `source`(그 chunk 의 입력)가 있으면 리터럴 직전 20자를 source 에서 찾아 그 뒤에 오는
+    원래 공백 런을 쓴다 -- 원문이 개행이면 개행, 문장 사이 공백이면 공백. 모델이 문장 사이에 토큰을
+    지어냈을 때 무조건 개행으로 바꾸면 문단이 늘어난다(실측: 기사 2건에서 +3·+2 문단). 앵커를 못 찾으면
+    개행(토큰의 계약상 의미).
+    """
+    if LITERAL not in text:
+        return text
+    if source is None:
+        return SEPARATOR_LITERAL.sub("\n", text)
+    parts = SEPARATOR_LITERAL.split(text)
+    out = parts[0]
+    for nxt in parts[1:]:
+        out += _original_separator(source, out) + nxt
+    return out
+
+
+def _original_separator(source: str, left: str) -> str:
+    anchor = left[-ANCHOR_LEN:].rstrip()
+    if anchor:
+        i = source.rfind(anchor)
+        if i >= 0:
+            m = WS_RUN.match(source, i + len(anchor))
+            if m:
+                return "\n" if ("\n" in m.group() or "\r" in m.group()) else " "
+            return " "  # 원문엔 공백조차 없던 자리 -- 붙여 쓰진 않고 최소 구분자
+    return "\n"
+
+
+# ── 후보 경계 ────────────────────────────────────────────────────────────
+
+
+def _candidate_boundaries(body: str) -> list[Boundary]:
+    open_depth = _open_quote_depth(body)
+    out: list[Boundary] = []
+    for m in WS_RUN.finditer(body):
+        start, end = m.span()
+        if start == 0 or end == len(body):
+            continue  # strip() 했으므로 없지만 방어
+        if "\n" in m.group() or "\r" in m.group():
+            kind = "para"
+        elif _ends_sentence(body, start, end):
+            kind = "quoted_sent" if open_depth[start] > 0 else "sent"
+        else:
+            kind = "word"
+        out.append(Boundary(start, end, kind))
+    return out
+
+
+def _ends_sentence(body: str, ws_start: int, ws_end: int) -> bool:
+    i = ws_start - 1
+    while i >= 0 and body[i] in TRAILING_CLOSERS:
+        i -= 1
+    if i < 0 or body[i] not in SENTENCE_END:
+        return False
+    # `3. 4.` 같은 번호 매김 -- 숫자 사이의 마침표는 문장 끝이 아니다
+    numbered = (
+        body[i] == "." and i > 0 and body[i - 1].isdigit() and body[ws_end : ws_end + 1].isdigit()
+    )
+    return not numbered
+
+
+def _open_quote_depth(body: str) -> list[int]:
+    """각 위치 직전까지 열려 있는 따옴표/괄호 수. 곧은따옴표(")는 홀수 개면 열린 것으로 본다."""
+    pairs = {"“": 1, "”": -1, "‘": 1, "’": -1, "(": 1, ")": -1, "「": 1, "」": -1}
+    depth: list[int] = [0] * (len(body) + 1)
+    curly = 0
+    straight = 0
+    for i, ch in enumerate(body):
+        depth[i] = max(curly, 0) + straight
+        if ch == '"':
+            straight ^= 1
+        else:
+            curly += pairs.get(ch, 0)
+    depth[len(body)] = max(curly, 0) + straight
+    return depth
+
+
+def _add_forced_cuts(body: str, boundaries: list[Boundary], max_len: int) -> list[Boundary]:
+    """공백 없는 런이 `max_len` 보다 길면 그 안에 강제 절단점을 넣는다 (URL 덩어리 같은 예외)."""
+    anchors = [0] + [b.end for b in boundaries]
+    starts = [b.start for b in boundaries] + [len(body)]
+    forced: list[Boundary] = []
+    for prev_end, next_start in zip(anchors, starts, strict=True):
+        pos = prev_end + max_len
+        while pos < next_start:
+            forced.append(Boundary(pos, pos, "char"))
+            pos += max_len
+    return sorted(boundaries + forced, key=lambda b: b.start)
+
+
+# ── DP ───────────────────────────────────────────────────────────────────
+
+
+def _solve(body: str, boundaries: list[Boundary], max_len: int, min_len: int) -> list[int]:
+    """전체 비용 최소인 경계 부분집합(인덱스, 오름차순)을 고른다.
+
+    노드 0 = 본문 시작, 노드 k(1..K) = boundaries[k-1], 노드 K+1 = 본문 끝.
+    chunk(i→j) 비용 = CHUNK_COST + 짧은 chunk 벌점 + (j 가 내부 경계면) 경계 종류 비용.
+    """
+    n_b = len(boundaries)
+    node_end = [0] + [b.end for b in boundaries]  # chunk 시작 위치 (노드별)
+    node_start = [b.start for b in boundaries] + [len(body)]  # chunk 끝 위치 (노드 1..K+1)
+    best = [math.inf] * (n_b + 2)
+    back = [-1] * (n_b + 2)
+    best[0] = 0.0
+    for j in range(1, n_b + 2):
+        end_pos = node_start[j - 1]
+        edge_cost = KIND_COST[boundaries[j - 1].kind] if j <= n_b else 0.0
+        for i in range(j - 1, -1, -1):
+            length = end_pos - node_end[i]
+            if length > max_len:
+                break
+            if length <= 0 or best[i] == math.inf:
+                continue
+            cost = best[i] + CHUNK_COST + _short_penalty(length, min_len) + edge_cost
+            if cost < best[j]:
+                best[j], back[j] = cost, i
+    if best[n_b + 1] == math.inf:
+        raise RuntimeError("분할 계획 실패 -- 강제 절단점이 부족하다 (버그)")
+    cut_ids: list[int] = []
+    j = n_b + 1
+    while back[j] > 0:
+        j = back[j]
+        cut_ids.append(j - 1)
+    return sorted(cut_ids)
+
+
+GREEDY_LADDER = (("para",), ("sent", "quoted_sent"), ("word",), ("char",))
+
+
+def _solve_greedy(boundaries: list[Boundary], n: int, max_len: int) -> list[int]:
+    """regex 사다리 재현: 창(pos, pos+max_len] 안에서 문단 > 문장 > 어절 > 강제 순으로, 가장 뒤 경계."""
+    cut_ids: list[int] = []
+    pos = 0
+    while n - pos > max_len:
+        window = [i for i, b in enumerate(boundaries) if pos < b.start <= pos + max_len]
+        pick = next(
+            (
+                cands[-1]
+                for kinds in GREEDY_LADDER
+                if (cands := [i for i in window if boundaries[i].kind in kinds])
+            ),
+            None,
+        )
+        if pick is None:
+            raise RuntimeError("greedy 분할 실패 -- 강제 절단점이 부족하다 (버그)")
+        cut_ids.append(pick)
+        pos = boundaries[pick].end
+    return cut_ids
+
+
+def _short_penalty(length: int, min_len: int) -> float:
+    if min_len <= 0 or length >= min_len:
+        return 0.0
+    deficit = 1.0 - length / min_len
+    return SHORT_PENALTY * deficit * deficit

solar_eval/pipelines/steps/llm.py

@@ -15,11 +15,26 @@ from solar_eval.models.sample import EvalSample
 from solar_eval.pipelines import judge
 from solar_eval.pipelines.pipeline import BaseStep, StepContext
 from solar_eval.pipelines.registry import register_step
+from solar_eval.pipelines.steps.chunking import (
+    ChunkPlan,
+    join_chunks,
+    normalize_separator_literals,
+    plan_chunks,
+)
 
 logger = logging.getLogger(__name__)
 
-# 문단 경계 보존용 마커 (bulk 분할 시) -- multi_step.py 원본 값 그대로
-PARAGRAPH_SEP = "∥PARA∥"
+# 출력 길이 가드 -- 교열은 길이를 거의 바꾸지 않는다. 입력 대비 ±25% 또는 40자(짧은 입력용) 를 넘는 변동은
+# 모델이 문장을 지어냈거나(실측 49자→321자) bulk 를 통째로 버린 것(490자→7자)이라 폴백/입력 유지로 처리한다.
+# 종전 규칙("5배 초과")은 길이 급감과 5배 미만의 환각을 그대로 통과시켰다 (insight 260910).
+LENGTH_TOL_REL = 0.25
+LENGTH_TOL_ABS = 40
+
+
+def plausible_length(source: str, output: str) -> bool:
+    if not source:
+        return True
+    return abs(len(output) - len(source)) <= max(LENGTH_TOL_ABS, LENGTH_TOL_REL * len(source))
 
 
 @register_step("llm")
@@ -51,13 +66,25 @@ class LLMStep(BaseStep):
         # Split into bulks if needed
         # For include_original/user_template steps, skip splitting
         if include_original or user_template:
-            bulks = [text]
+            plan = ChunkPlan(chunks=[text], seps=[""], kinds=[], lead="")
         else:
-            bulks = self._split_text(text)
+            plan = plan_chunks(
+                text, self.env.text_split_size, strategy=self.env.text_split_strategy
+            )
+        bulks = plan.chunks
         processed_bulks = []
         logger.debug(f"  [{ts}] {len(bulks)} bulk(s), text_len={len(text)}")
+        if plan.is_split:
+            # 절단면 진단용 -- 어디를 어떤 경계로 잘랐는지 (insight 260910 의 GLUED/LOST 분석이 읽는다)
+            chunk_plans = sample.artifacts.setdefault("chunk_plans", {})
+            chunk_plans[step_name] = {
+                "lens": [len(b) for b in bulks],
+                "kinds": list(plan.kinds),
+            }
 
         for bulk_i, bulk in enumerate(bulks):
+            if ctx.on_step and len(bulks) > 1:
+                ctx.on_step("chunk", bulk_i + 1, len(bulks), step_name, ctx.doc_chunk)
             if step_repeat > 1:
                 (
                     final_text,
@@ -119,16 +146,20 @@ class LLMStep(BaseStep):
             )
             ts3 = datetime.now().strftime("%H:%M:%S")
             raw_len = len(result.get("content", ""))
-            output = self._extract_output(result["content"]).rstrip()
+            output = normalize_separator_literals(
+                self._extract_output(result["content"]).rstrip(), source=bulk
+            )
 
-            # JSON 파싱 실패 시 fallback:
-            # output이 입력 대비 5배 이상 길면 파싱 실패로 간주
-            if len(output) > len(bulk) * 5 and len(bulk) > 0:
+            # 길이 가드: JSON 파싱 실패(설명이 통째로 output 이 됨) · 환각 · bulk 삭제를 한 규칙으로 잡는다
+            if not plausible_length(bulk, output):
                 raw_content = result.get("content", "")
                 logger.warning(
-                    f"  [{ts3}] [{step_name}] bulk {bulk_i + 1} output too long "
-                    f"({len(output)} vs input {len(bulk)}), "
-                    f"attempting fallback"
+                    f"  [{ts3}] [{step_name}] bulk {bulk_i + 1} output length implausible "
+                    f"(in={len(bulk)}, out={len(output)}), attempting fallback"
+                )
+                guard_log = sample.artifacts.setdefault("length_guard", [])
+                guard_log.append(
+                    {"step": step_name, "bulk": bulk_i + 1, "in": len(bulk), "out": len(output)}
                 )
 
                 # Try fallback LLM call if configured
@@ -156,10 +187,12 @@ class LLMStep(BaseStep):
                         reasoning_effort=step_re,
                         response_format=fallback_rf,
                     )
-                    fallback_output = self._extract_output(fallback_result["content"]).rstrip()
+                    fallback_output = normalize_separator_literals(
+                        self._extract_output(fallback_result["content"]).rstrip(), source=bulk
+                    )
                     ts4 = datetime.now().strftime("%H:%M:%S")
 
-                    if len(fallback_output) <= len(bulk) * 5:
+                    if plausible_length(bulk, fallback_output):
                         output = fallback_output
                         logger.info(
                             f"  [{ts4}] [{step_name}] fallback succeeded, len={len(output)}"
@@ -168,7 +201,7 @@ class LLMStep(BaseStep):
                     else:
                         output = bulk
                         logger.warning(
-                            f"  [{ts4}] [{step_name}] fallback also too long, using input text"
+                            f"  [{ts4}] [{step_name}] fallback length also implausible, using input text"
                         )
                 else:
                     output = bulk
@@ -191,33 +224,11 @@ class LLMStep(BaseStep):
                 step_corrs = sample.artifacts.setdefault("corrections", {})
                 step_corrs.setdefault(step_name, []).extend(step_corrections)
 
-        ctx.text = self._join_bulks(processed_bulks)
+        # 구분자 원장으로 접합 -- 절단면의 공백/개행을 그대로 복원한다. 분할·접합 규칙은 chunking.py.
+        ctx.text = join_chunks(processed_bulks, plan)
 
     # ── helpers (multi_step.py 에서 그대로 이동) ─────────────────────────
 
-    def _split_text(self, text: str) -> list[str]:
-        """Split text into bulks of max text_split_size chars, preserving line breaks."""
-        if len(text) <= self.env.text_split_size:
-            return [text]
-
-        text_with_markers = text.replace("\n", PARAGRAPH_SEP)
-        bulks = []
-        current = ""
-
-        for char in text_with_markers:
-            current += char
-            if len(current) >= self.env.text_split_size:
-                bulks.append(current.replace(PARAGRAPH_SEP, "\n"))
-                current = ""
-
-        if current:
-            bulks.append(current.replace(PARAGRAPH_SEP, "\n"))
-
-        return bulks
-
-    def _join_bulks(self, bulks: list[str]) -> str:
-        return "".join(bulks)
-
     def _extract_output(self, content: str) -> str:
         """Extract output from LLM response. Tries JSON parsing first.