"""multi_step 파이프라인의 rule 스텝 구현체. `multi_step.py` 가 518줄이라 (`.claude/rules/code-style.md` 의 400줄 권장선을 넘어) 룰 구현만 따로 뽑아냈다. vocabulary/post_process/correction_filter 는 동작을 바꾸지 않고 `MultiStepPipeline._apply_*` 메서드에서 그대로 옮겨온 것이다 — 순수 텍스트 변환 로직이라 `self` 상태가 필요 없다(`apply_vocabulary` 만 예외 — 데이터셋에서 로드한 어휘 캐시를 인자로 받고 갱신된 캐시를 되돌려준다). `apply_judge_decisions`/`apply_change_rate_guard` 는 v24 계열 포팅 단계(§5-F)에서 추가됐다. 전자는 `EvalSample.artifacts` 를 읽어야 해서(judge 스텝의 구조화 결정, self-consistency 의 corrections 목록) `sample` 을 인자로 받는다 — 다른 rule 함수들과 시그니처가 다른 유일한 이유다. """ from __future__ import annotations import logging import re from pathlib import Path from typing import TYPE_CHECKING, Any from solar_eval.models.sample import EvalSample from solar_eval.pipelines.errors import KnowledgeAssetError if TYPE_CHECKING: from solar_eval.core.dataset_loader import DatasetLoader logger = logging.getLogger(__name__) def apply_vocabulary( text: str, config: dict[str, Any], vocabulary: list[dict[str, str]] | None, dataset_loader: DatasetLoader | None, config_dir: Path | None = None, ) -> tuple[str, list[dict[str, str]] | None]: """`vocabulary_sub`/`vocabulary_reapply` rule_type 구현. 호출자(`VocabularyStep`)가 run 수명 캐시를 들고 있으므로, 최초 호출 시 로드한 어휘 목록을 반환값으로 되돌려줘 캐시로 저장하게 한다. Args: config_dir: 레포의 config 정본 디렉토리(`03-evaluation`). 치환 사전은 데이터가 아니라 **설정**이라(바뀌면 같은 문제를 다르게 푼다) 여기를 먼저 찾는다. 데이터 루트 기준 레거시 선언도 계속 지원한다. Returns: (치환된 text, 갱신된 vocabulary 캐시) 튜플. Raises: KnowledgeAssetError: `vocab_dataset` 을 선언했는데 못 읽을 때. """ if vocabulary is None: vocabulary = _load_vocabulary(config, dataset_loader, config_dir) for row in vocabulary: wrong = row.get("wrong", row.get("before", "")) correct = row.get("correct", row.get("after", "")) if wrong and correct: text = text.replace(wrong, correct) return text, vocabulary def _load_vocabulary( config: dict[str, Any], dataset_loader: DatasetLoader | None, config_dir: Path | None, ) -> list[dict[str, str]]: """치환 사전을 로드한다 — config 정본 우선, 데이터 루트 폴백. 선언이 없으면 빈 목록(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다. """ vocab_path = config.get("vocab_dataset", "") if not vocab_path: return [] searched: list[str] = [] # 1. config 정본 (레포의 03-evaluation/) — 치환 사전은 설정이다. if config_dir is not None: candidate = Path(config_dir) / vocab_path searched.append(str(candidate)) if candidate.is_file() and dataset_loader is not None: return dataset_loader.load_csv(str(config_dir), vocab_path, source="local") # 2. 레거시 선언 (repo/source 로 데이터 루트나 HF 를 가리키는 경우) repo = config.get("repo", "") source = config.get("source", "huggingface") if dataset_loader is not None: searched.append(f"{source}:{repo}/{vocab_path}" if repo else f"{source}:{vocab_path}") try: return dataset_loader.load_csv(repo, vocab_path, source=source) except Exception as e: # noqa: BLE001 — 원인을 KnowledgeAssetError 에 실어 다시 던진다 logger.debug("vocabulary load failed (%s): %s", vocab_path, e) else: searched.append("(dataset_loader 없음)") raise KnowledgeAssetError(vocab_path, searched) def apply_pronoun_postprocess(text: str, replacements: dict[str, str]) -> str: """대명사/직함 치환 + 한국어 조사 교정. 단어를 치환할 때(예: "이재명 대표" → "이재명 대통령") 뒤따르는 조사도 새 단어의 받침에 맞춰 바꿔야 한다. """ # 모음으로 끝나는 조사 → 자음으로 끝나는 단어 뒤에 붙는 대응 조사 correction_map = { "는": "은", "가": "이", "를": "을", "와": "과", "로": "으로", "여": "이여", "라": "이라", "랑": "이랑", "다": "이다", "였다": "이었다", "라면": "이라면", "라서": "이라서", "로부터": "으로부터", } all_particles = ( list(correction_map.keys()) + list(correction_map.values()) + ["만", "도", "께서"] ) particle_pattern = "|".join(re.escape(p) for p in sorted(all_particles, key=len, reverse=True)) for old, new in replacements.items(): regex = re.compile(f"({re.escape(old)})({particle_pattern})?") def replacer(m: re.Match, new_word: str = new) -> str: particle = m.group(2) if particle: return new_word + correction_map.get(particle, particle) return new_word text = regex.sub(replacer, text) return text def apply_currency_compact(text: str) -> str: """통화 표현에서 숫자와 단위 사이 공백을 제거한다.""" _unit = r"(원|명|개|대|위안|달러|유로|엔|톤|권|건|회|장|마리|그루|송이|필)" _tail = r"(?=[\s.,!?()\"\'가-힣]|$)" # "3억 5000만 원" → "3억5000만원" text = re.sub( r"(\d+(?:\.\d+)?)(만|억|조)\s*(\d+)\s*" + _unit + _tail, r"\1\2\3\4", text, ) # "3억 원" → "3억원" text = re.sub( r"(\d+(?:\.\d+)?)(만|억|조)\s*" + _unit + _tail, r"\1\2\3", text, ) return text def apply_comma_removal(text: str) -> str: """숫자 사이의 천단위 콤마를 제거한다 (1,000 → 1000).""" return re.sub(r"(?<=\d),(?=\d)", "", text) def apply_correction_filter(text: str, config: dict[str, Any], original: str = "") -> str: """Filter corrections by comparing with original text. Keeps only 'safe' corrections (spacing changes, small character diffs). Reverts larger changes that are likely unnecessary expression substitutions. Config options: max_char_diff: Maximum character differences to allow (default: 2) allow_spacing: Keep spacing-only changes (default: true) """ from solar_eval.evaluators.scoring import compute_lcs if not original: return text max_cd = config.get("max_char_diff", 2) allow_spacing = config.get("allow_spacing", True) orig_tokens = original.split() pred_tokens = text.split() if orig_tokens == pred_tokens: return text lcs_pairs = compute_lcs(orig_tokens, pred_tokens) def is_safe(o_text: str, c_text: str) -> bool: if allow_spacing and o_text.replace(" ", "") == c_text.replace(" ", ""): return True if abs(len(o_text) - len(c_text)) <= 1: d = sum(1 for a, b in zip(o_text, c_text) if a != b) d += abs(len(o_text) - len(c_text)) if d <= max_cd: return True return False result_tokens: list[str] = [] prev_orig = -1 prev_pred = -1 for oi, pi in lcs_pairs: orig_gap = orig_tokens[prev_orig + 1 : oi] pred_gap = pred_tokens[prev_pred + 1 : pi] if orig_gap or pred_gap: o_text = " ".join(orig_gap) p_text = " ".join(pred_gap) if o_text == p_text or is_safe(o_text, p_text): result_tokens.extend(pred_gap) else: result_tokens.extend(orig_gap) result_tokens.append(orig_tokens[oi]) prev_orig = oi prev_pred = pi orig_tail = orig_tokens[prev_orig + 1 :] pred_tail = pred_tokens[prev_pred + 1 :] if orig_tail or pred_tail: o_text = " ".join(orig_tail) p_text = " ".join(pred_tail) if o_text == p_text or is_safe(o_text, p_text): result_tokens.extend(pred_tail) else: result_tokens.extend(orig_tail) return " ".join(result_tokens) def apply_judge_decisions(text: str, config: dict[str, Any], sample: EvalSample) -> str: """`tool_calling_judge` 의 verdict="revert" 판정을 텍스트에 되돌린다. `judge_step`(기본 "tool_calling_judge")의 판정 목록과 `corrections_step` (기본 "basic_correction")의 corrections 목록을 `sample.artifacts` 에서 읽는다 — 이 두 artifact 는 각각 `pipelines/judge.py::run_tool_calling_judge` 와 `MultiStepPipeline` 의 self-consistency 경로가 채운다. 어느 한쪽이라도 비어 있으면(judge 스텝이 없었거나 corrections 가 하나도 없었던 경우) 조용히 원문(text)을 그대로 돌려준다 — revert 할 대상 자체가 없는 정상 상태다. HF 구현(`huggingface/pipelines_v24.py::apply_judge_decisions_for_bulk`)을 포팅. bulk 단위였던 원본과 달리 여기서는 문서 전체 text 에 적용한다 (§5-F 설계 결정 — 이 엔진은 tool_calling_judge/basic_correction 을 이미 문서 단위로 처리하므로 bulk 정렬 문제 자체가 없다). """ judge_step = config.get("judge_step", "tool_calling_judge") corrections_step = config.get("corrections_step", "basic_correction") decisions = sample.artifacts.get("judge_decisions", {}).get(judge_step, []) corrections = sample.artifacts.get("corrections", {}).get(corrections_step, []) if not decisions or not corrections: return text revert_pairs: list[tuple[str, str]] = [] for d in decisions: if not isinstance(d, dict) or d.get("verdict") != "revert": continue idx = d.get("correction_index") if not isinstance(idx, int) or idx < 0 or idx >= len(corrections): continue entry = corrections[idx] if not isinstance(entry, dict): continue corrected = str(entry.get("corrected", "")) original = str(entry.get("original", "")) if corrected and original and corrected != original: revert_pairs.append((corrected, original)) # 긴 것부터 되돌려야 짧은 교정이 긴 교정의 부분 문자열일 때 잘못 되돌리지 않는다. revert_pairs.sort(key=lambda p: -len(p[0])) for corrected, original in revert_pairs: if corrected in text: text = text.replace(corrected, original, 1) return text def apply_change_rate_guard(text: str, original: str, config: dict[str, Any]) -> str: """원문 대비 토큰 변경 비율이 `max_change_rate` 를 넘으면 원문으로 되돌리는 안전망. v17 인사이트 문서(`insights/260504_solar-pro2_essentials.md`)가 "change_rate_guard (rule, bulk-level safety net, max_change_rate: 0.45)" 라고만 설명하고, 실제 구현은 HF 아카이브(`huggingface/pipelines.py`/`pipelines_v24.py`) 어디에도 없다 (레포 전체 검색으로 확인). 그래서 이 함수는 그 설명만 보고 새로 작성한 것이다 — v17 수치 재현을 약속하지 않는다는 전제로 포팅 대상에 포함됐다 (`.agents/01-plans/harness/2026-08-09-chosun-proofread-evalsample-migration.md` §9-3). 변경 비율은 `evaluators/scoring.compute_lcs` 로 원문·현재 텍스트의 토큰 단위 LCS 를 구해 `1 - |LCS| / |원문 토큰|` 으로 정의한다 — 이 프로젝트의 다른 모든 교열 diff 로직(`compute_proofread_metrics`, `apply_correction_filter`)과 같은 척도를 쓴다. """ from solar_eval.evaluators.scoring import compute_lcs max_rate = config.get("max_change_rate", 0.45) orig_tokens = original.split() if not orig_tokens: return text text_tokens = text.split() lcs_pairs = compute_lcs(orig_tokens, text_tokens) change_rate = 1 - (len(lcs_pairs) / len(orig_tokens)) if change_rate > max_rate: logger.warning( "[change_rate_guard] change_rate=%.3f > max=%.3f, reverting to original", change_rate, max_rate, ) return original return text def apply_unit_unicode(text: str) -> str: """단위 텍스트를 유니코드 결합 문자로 변환한다.""" replacements = { "mm²": "㎟", "cm²": "㎠", "m²": "㎡", "km²": "㎢", "mm": "㎜", "cm": "㎝", "km": "㎞", "mg": "㎎", "kg": "㎏", "ml": "㎖", "dl": "㎗", "cc": "㏄", } for k, v in replacements.items(): text = text.replace(k, v) return text def load_compound_whitelist(config: dict[str, Any], config_dir: Path | None) -> set[str]: """복합어 화이트리스트(탭 구분 ``표제어\\t빈도``)를 로드한다. 선언이 없으면 빈 set(정상적인 no-op)이고, 선언했는데 못 읽으면 예외다 — `KnowledgeAssetError` 계약(`errors.py`)을 치환 사전과 동일하게 따른다. 화이트리스트는 사람이 편집하는 지식 자산이라 config 정본(`03-evaluation/`) 기준 상대경로만 지원한다. """ wl_path = config.get("whitelist_path", "") if not wl_path: return set() searched: list[str] = [] if config_dir is not None: candidate = Path(config_dir) / wl_path searched.append(str(candidate)) if candidate.is_file(): entries: set[str] = set() for line in candidate.read_text(encoding="utf-8").splitlines(): form = line.split("\t")[0].strip() if form: entries.add(form) return entries else: searched.append("(config_dir 없음)") raise KnowledgeAssetError(wl_path, searched) def _has_whitelist_prefix(joined: str, split_at: int, whitelist: set[str]) -> bool: """결합형의 접두(분리점 너머 ~ 전체)가 화이트리스트 표제어인지. 조사·어미가 붙은 어절("삼성"+"전자는")도 표제어("삼성전자")로 잡기 위해 정확 일치가 아니라 접두 일치를 본다. 분리점(`split_at`) 안쪽에서 끝나는 접두는 첫 어절만으로 성립하므로 후보에서 뺀다. """ return any(joined[:k] in whitelist for k in range(split_at + 1, len(joined) + 1)) def apply_compound_whitelist_restore(text: str, original: str, whitelist: set[str]) -> str: """`compound_whitelist_restore` rule_type 구현 (v19/v32 계보의 현 엔진 포팅). 파이프라인(FT·step1/2/3)이 갈라놓은 화이트리스트 복합어를 다시 붙인다. 인접 어절 쌍의 결합이 (1) 화이트리스트 표제어를 접두로 갖고 — 분리점이 표제어 안쪽 — (2) 원문에 그대로 존재할 때만 붙인다. 조건 (2) 때문에 이 룰은 파이프라인이 만든 분리를 되돌릴 뿐, 새 교정을 만들지 않는다. 2026-09-02 재생 실측(combo 체인 출력 위): paragraph +2.91 f1 (redundant 408→342), ci-v0 +0.43, ci-v2/ci-v1/yul-ryul 무변화. """ if not whitelist: return text restored_lines: list[str] = [] for line in text.split("\n"): tokens = line.split() out: list[str] = [] i = 0 while i < len(tokens): if i + 1 < len(tokens): joined = tokens[i] + tokens[i + 1] if joined in original and _has_whitelist_prefix(joined, len(tokens[i]), whitelist): out.append(joined) i += 2 continue out.append(tokens[i]) i += 1 restored_lines.append(" ".join(out)) return "\n".join(restored_lines) #: 율/률 정규화가 "명사 + 조사" 문맥으로 인정하는 어절 꼬리(한글 잔여) 목록. #: 이 목록 밖의 한글이 이어지면(예: 인명 '김율희가'의 '희가') 건드리지 않는다. _YUL_RYUL_JOSA = frozenset( { "은", "는", "이", "가", "을", "를", "의", "에", "도", "만", "과", "와", "로", "으로", "보다", "부터", "까지", "조차", "마저", "밖에", "처럼", "대로", "만큼", "뿐", "나", "이나", "란", "이란", "라", "이라", "라고", "이라고", "라는", "이라는", "라며", "이라며", "며", "이며", "고", "이고", "다", "이다", "입니다", "에서", "에게", "에는", "에도", "에서는", "에서도", "로는", "으로는", "로도", "으로도", "로서", "으로서", "로써", "으로써", "와의", "과의", "과는", "와는", "은요", "요", "대비", } ) def apply_yul_ryul_normalize(text: str, exceptions: tuple[str, ...] = ()) -> str: """`yul_ryul_normalize` rule_type 구현 — 율/률 받침 규칙의 결정론적 적용. 한글 맞춤법 제11항: 앞 음절에 받침이 없거나 ㄴ 받침이면 '율'(비율·환율· 전환율), 그 외 받침이면 '률'(법률·확률·출석률). 받침은 유니코드 연산으로 판정되므로 LLM 없이 최종 텍스트에서 강제할 수 있다. 인명 오탐 방지 2중 가드: 1. 어절 꼬리 검사 — 율/률 뒤의 한글 잔여가 조사 목록(`_YUL_RYUL_JOSA`)에 없으면 접미사 용법이 아닌 것으로 보고 건너뛴다 (재생 실측의 유일한 오탐 '김율희가'→'김률희가' 가 여기서 걸러진다). 2. `exceptions` — 어절에 포함되면 무조건 건너뛰는 문자열 목록 (율/률로 끝나는 인명 '김석율' 류를 위한 훅. 기본 비어 있음). 2026-09-02 재생 실측(v33/combo 출력 위, 조사 가드 포함): yul-ryul 놓침 2건 전량 회수(+3.07), ci-v2 +0.59, ci-v1 +0.53, paragraph·ci-v0 변경 0건. """ out = list(text) n = len(text) for i, ch in enumerate(text): if ch not in ("율", "률"): continue prev = text[i - 1] if i > 0 else "" if not ("가" <= prev <= "힣"): continue jong = (ord(prev) - 0xAC00) % 28 want = "율" if jong in (0, 4) else "률" # 받침 없음(0)/ㄴ(4) → 율 if want == ch: continue j = i + 1 while j < n and not text[j].isspace(): j += 1 residue = "".join(c for c in text[i + 1 : j] if "가" <= c <= "힣") if residue and residue not in _YUL_RYUL_JOSA: continue if exceptions: k = i while k > 0 and not text[k - 1].isspace(): k -= 1 token = text[k:j] if any(e in token for e in exceptions): continue out[i] = want return "".join(out)