"""대본을 읽고 **분해 규칙**을 저작한다 — 경계는 저작하지 않는다. 설계 = docs/superpowers/specs/2026-08-08-scene-segmentation-rule-authoring-design.md ## 왜 규칙만 쓰게 하나 경계를 LLM 이 직접 찍으면 두 가지가 따라온다. ①형식이 튀는 씬을 의미로 판단하다 놓친다(실측: 같은 대본에서 같은 씬 셋을 두 실행이 똑같이 놓쳤다). ②실행마다 결과가 흔들린다. 규칙만 쓰게 하면 판단은 한 번이고, 그 뒤로는 코드가 같은 입력에 같은 출력을 낸다. ## ★정리본에서만 저작한다 PDF 화면·PDF 직접 추출·`text_cleanup` 정리본은 서로 다른 텍스트다. 한 표현에서 저작해 다른 표현에 실행하면 규칙이 맞아도 실패한다. 실측: PDF 직접 추출에는 쪽 번호가 남고 헤딩이 두 줄로 갈리는데, 정리본에서는 둘 다 사라진다. ## ★대본을 자르지 않는다 프로젝트 절대 규칙이다. 규칙 저작은 특히 그렇다 — 앞부분만 보면 뒤에 나오는 다른 형식(회상·몽타주·부록)을 못 본다. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional from app.modules.pipeline.segment_rule import SegmentRule logger = logging.getLogger(__name__) STEP = "segment_rule_author" SYSTEM = """\ 당신은 대본 한 편을 처음부터 끝까지 읽고, **씬이 시작하는 줄을 찾는 정규식**을 씁니다. 씬 경계를 직접 찍지 않습니다 — 규칙만 씁니다. 그 규칙은 파이썬 `regex` 모듈이 MULTILINE 으로 실행합니다. ## 무엇이 씬의 시작인가 대본마다 다릅니다. 번호가 앞에 오기도, 기호가 붙기도, 번호 없이 장소 표기로만 시작하기도 합니다. **당신이 받은 이 대본에서 실제로 쓰인 표기**를 찾아 그것을 정규식으로 옮기세요. 다른 대본에서 본 형식을 가정하지 마세요. ## 반드시 가려야 하는 두 가지 **쪽 번호** — 번호만 있고 그 줄에 다른 내용이 없는 줄이 규칙적으로 나오면 쪽 번호일 수 있습니다. 씬 헤딩이라면 같은 줄이나 바로 다음 줄에 장소·시간 같은 내용이 따라옵니다. 쪽 번호는 본문이 그냥 이어집니다. **씬 안의 항목 번호** — 여러 장면을 나열하는 씬 안에서 번호가 다시 매겨지는 경우가 있습니다. 겉모습이 씬 헤딩과 같아 번호만으로는 못 가릅니다. 그 대본에서 **씬 헤딩에만 붙는 표기**(장소 구분, 시간 표기, 구분 기호 따위)를 찾아 규칙에 담으세요. 그것이 이 일의 핵심입니다. 코드는 걸린 매치를 하나도 버리지 않습니다. 번호가 되돌아가면 그 후보를 실패로 판정하고 당신에게 다시 물어봅니다. ## 후보를 2~3개 쓰세요 서로 다른 신호를 쓰는 것이 좋습니다(예: 번호 표기를 쓰는 것 하나, 장소 표기를 쓰는 것 하나). 코드가 각 후보를 실행해 기계로 재고 더 나은 쪽을 고릅니다. 확신이 없으면 좁은 것과 넓은 것을 함께 내세요. ## rationale 에는 그 대본에서 **실제로 걸리는 줄**을 원문 그대로 2~3개 옮기세요. 지어내지 마세요. 사람이 그 줄을 보고 규칙이 맞는지 판단합니다. """ SCHEMA: Dict[str, Any] = { "type": "object", "properties": { "candidates": { "type": "array", "minItems": 1, "maxItems": 3, "items": { "type": "object", "properties": { "name": {"type": "string", "description": "이 규칙이 쓰는 신호를 짧게"}, "pattern": {"type": "string", "description": "파이썬 regex. MULTILINE 으로 실행된다"}, "number_group": { "type": ["integer", "null"], "description": "씬 번호를 담은 캡처 그룹 번호. 없으면 null", }, "rationale": { "type": "string", "description": "이 대본에서 실제로 걸리는 줄 2~3개 (원문 그대로)", }, }, "required": ["name", "pattern", "number_group", "rationale"], "additionalProperties": False, }, }, "observed_format": { "type": "string", "description": "이 대본의 씬 표기가 어떻게 생겼는지 한두 문장", }, }, "required": ["candidates", "observed_format"], "additionalProperties": False, } def author_and_choose( text: str, *, model: str = "gpt", max_rounds: int = 3, step: str = STEP, project_config: Optional[Dict[str, Any]] = None, project_id: Optional[str] = None, episode_id: Optional[str] = None, ) -> tuple[Optional[SegmentRule], Optional[Any], List[Dict[str, Any]]]: """저작 → 기계 검증 → 후보 간 차이로 재저작. `(고른 규칙, 판정, 회차 기록)`. ★실험대와 프로덕션이 **같은 루프를 쓴다.** 두 곳에 따로 두면 실험에서 찾은 결함 대응(넓은 후보 신호·탈락 후보 제외·거짓 성공 막기)이 한쪽에만 남아 조용히 갈린다. 회차 기록은 갤러리가 읽는 자료다. 프로덕션은 무시해도 된다. """ from app.modules.pipeline.segment_rule import apply_rule, choose_rule, verify rounds: List[Dict[str, Any]] = [] hint = "" for rnd in range(max_rounds): try: cands, observed = author_rules( (text if not hint else f"{hint}\n\n---\n\n{text}"), model=model, step=step, project_config=project_config, project_id=project_id, episode_id=episode_id) except Exception as exc: # noqa: BLE001 — 부르는 쪽이 이어가게 둔다 rounds.append({"round": rnd + 1, "error": str(exc)[:200]}) break verds = [] for c in cands: try: verds.append(verify(text, apply_rule(text, c), c)) except ValueError: verds.append(None) rounds.append({ "round": rnd + 1, "observed": observed, "candidates": [ {"name": c.name, "pattern": c.pattern, "number_group": c.number_group, "rationale": c.rationale, # ★자르지 않는다 "count": v.match_count if v else 0, "ok": bool(v and v.ok), "failures": (v.failures if v else ["거부"]), "content": round(v.heading_content, 3) if v else None, "continuity": round(v.number_continuity, 3) if v else None, # ★탈락 이유를 화면에서 재검증할 수 있어야 한다 — 이게 없으면 # "✗C5" 만 보이고 무엇이 몇 번 되돌아갔는지 모른다. "detail": (v.detail if v else {})} for c, v in zip(cands, verds)], }) chosen, verdict = choose_rule(text, cands) if not chosen: hint = retry_prompt(cands, verds) continue # ★골랐다고 끝이 아니다. 기계 계약은 후보끼리의 차이를 못 본다(실측: # 41/42 누락이 모든 계약을 통과했다). 그 차이를 **양쪽 다** 되돌려준다 — # 한쪽만 보면 정답을 통째로 포함하는 넓은 규칙이 조용히 이긴다. # # ★계약을 통과한 후보끼리만 견준다. 탈락한 후보는 "쓰지 않기로" 이미 # 판정된 것이라 기준이 될 수 없다. peers = [c for c, v in zip(cands, verds) if v and v.ok] missed = diff_lines(text, chosen, peers, want_extra=False) extra = diff_lines(text, chosen, peers, want_extra=True) rounds[-1]["missed_after_choice"] = missed[:20] rounds[-1]["extra_after_choice"] = extra[:20] # ★견줄 상대가 있었는지 먼저 본다(Codex 지적, 실측으로 확인함). # 통과한 후보가 **하나뿐**이면 "차이 없음"은 맞다고 확인한 것이 아니라 # **아무도 안 봤다**는 뜻이다. 그런데 코드는 둘을 구별 못 해 그대로 # 확정했다 — 이 설계가 막으려던 "개수는 맞는데 경계가 틀림"이 다시 # 통과한다. # # 실제로 gemini-flash 저작에서 그 일이 났다: 후보 셋 중 둘이 탈락해 # 하나만 남았고, 견줄 상대가 없어 씬 하나를 놓친 규칙이 확정됐다. # # gpt 저작 22개는 **전부** 통과 후보가 둘 이상이었다(15개는 답까지 # 같았고 7개는 갈려서 재저작으로 풀렸다). 그래서 이 요구는 정상 대본의 # 비용을 거의 안 늘린다. # # ★재는 것은 **통과한 후보 수**다. 서로 다른 경계가 둘 이상이어야 한다고 # 재면 안 된다 — 답이 같은 것도 견준 것이고(오히려 일치 확인이다), # 그렇게 재면 22개 중 15개가 헛되이 재저작을 돈다. rounds[-1]["compared_candidates"] = len(peers) if not missed and not extra and len(peers) >= 2: return chosen, verdict, rounds if not missed and not extra: # 차이는 없지만 견준 적이 없다 — 다른 신호를 쓰는 후보를 요구한다. if rnd == max_rounds - 1: # 끝내 못 구했다. 그 규칙을 쓰되 **확인 못 했다는 사실을 남긴다** — # 여기서 실패시키면 후보가 하나뿐인 대본은 아예 안 돈다. rounds[-1]["unverified_alone"] = True logger.warning( "%s: 견줄 후보가 끝내 하나뿐이라 상대 비교 없이 확정한다 " "(pattern=%s)", step, chosen.pattern[:60]) return chosen, verdict, rounds rounds[-1]["retry_reason"] = "견줄 후보가 하나뿐 — 다른 신호로 하나 더" hint = alone_prompt(chosen) continue if rnd == max_rounds - 1: # ★상한을 다 썼다고 성공이 아니다. 코드가 이미 아는 차이를 남긴 채 # 확정하면 **거짓 성공**이 기록된다. rounds[-1]["unresolved"] = {"missed": len(missed), "extra": len(extra)} return None, None, rounds why = [] if missed: why.append(f"다른 후보가 잡은 {len(missed)}줄 놓침") if extra: why.append(f"저 혼자 잡은 {len(extra)}줄") rounds[-1]["retry_reason"] = " · ".join(why) hint = diff_prompt(chosen, missed, extra) return None, None, rounds def alone_prompt(chosen: SegmentRule) -> str: """통과한 후보가 하나뿐일 때 — 다른 신호를 쓰는 후보를 하나 더 요구한다. ★"차이 없음"이 **확인된 것**이 되려면 견줄 상대가 있어야 한다. 하나뿐이면 그 규칙이 옳은지 그른지 코드가 알 방법이 없다. """ return "\n".join([ f"낸 후보 중 기계 검증을 넘은 것이 `{chosen.pattern}` 하나뿐입니다.", "", "하나만으로는 그 규칙이 맞는지 견줄 수가 없습니다.", "**다른 신호를 쓰는** 후보를 하나 더 쓰세요 — 앞 것이 번호를 썼다면", "장소·시간 표기를, 표기를 썼다면 번호를 쓰는 식으로.", "앞의 후보도 그대로 다시 내세요.", ]) def diff_lines(text: str, chosen: SegmentRule, peers: List[SegmentRule], *, want_extra: bool) -> List[str]: """고른 규칙과 다른 후보가 갈린 자리 — 원문 그대로. `want_extra=False` 면 **다른 후보는 잡았는데 이 규칙이 놓친** 줄, `True` 면 **이 규칙만 잡고 아무도 안 잡은** 줄. ``peers`` 는 계약을 통과한 후보만 받는다(부르는 쪽 책임). 탈락한 후보를 섞으면 멀쩡한 규칙이 진짜 헤딩을 "혼자 잡았다"고 나온다. """ from app.modules.pipeline.segment_rule import apply_rule try: mine = {s.start_char: s.heading for s in apply_rule(text, chosen)} except ValueError: return [] others: Dict[int, str] = {} seen_any = False for c in peers: if c.pattern == chosen.pattern: continue try: for s in apply_rule(text, c): others.setdefault(s.start_char, s.heading) seen_any = True except ValueError: continue if not seen_any: # 견줄 후보가 없으면 신호도 없다 return [] if want_extra: return [mine[k] for k in sorted(mine) if k not in others] return [others[k] for k in sorted(others) if k not in mine] def author_rules( cleaned_text: str, *, model: str = "gpt", step: str = STEP, project_config: Optional[Dict[str, Any]] = None, project_id: Optional[str] = None, episode_id: Optional[str] = None, ) -> tuple[List[SegmentRule], str]: """정리본을 통째로 주고 후보 규칙을 받는다. ``model`` 은 alias 다(`gpt` / `gpt-terra` …). 저작 모델을 실측으로 고르려고 인자로 뺐다 — 이름으로 등급을 추측하지 않는다. 돌려주는 것은 (후보 목록, 관찰한 형식 설명)이다. 검증은 부르는 쪽이 `segment_rule.choose_rule` 로 한다 — 저작과 판정을 한 곳에 두면 저작이 자기 답을 채점하게 된다. ★되돌리기(fallback)는 끈다. 기본 경로는 막히면 입력을 바꿔 다시 부르고 끝에는 모델을 gpt 로 바꾼다. 그러면 `--author gemini` 로 잰 결과가 실은 gpt 것인데 기록에는 gemini 로 남아, **모델 비교가 통째로 틀린다**(호출 횟수도 같이 어긋난다). 막히면 막힌 대로 두고 그 사실을 남긴다. """ from app.modules.llm.llm_client import _resolve_model, call_structured # ★프로덕션은 자기 스텝 이름으로 설정을 찾는다. 실험대는 모델을 직접 # 고르므로 그 alias 로 만든다 — 이름으로 등급을 추측하지 않는다. cfg = (project_config if project_config is not None else {step: {"model": model}}) # ★기록에는 **실제로 고른 모델**을 남긴다(Codex 지적). 인자로 받은 model 을 # 그대로 적으면, project_config 가 있을 때 라우팅은 manifest 기본값을 # 쓰는데 기록만 인자 값으로 남아 **모델별 결과를 잘못 읽는다** — 실제로 # 프로덕션은 gemini-flash 로 저작하는데 기록은 gpt 였다. resolved = _resolve_model(step, cfg) result = call_structured( step=step, system_prompt=SYSTEM, user_prompt=cleaned_text, # ★자르지 않는다 response_schema=SCHEMA, project_config=cfg, schema_name="segment_rules", opik_metadata={"project_id": project_id, "episode_id": episode_id, "step": step, "author_model": resolved}, enable_fallback=False, ) rules: List[SegmentRule] = [] for c in (result.get("candidates") or []): pattern = (c.get("pattern") or "").strip() if not pattern: continue rules.append(SegmentRule( name=str(c.get("name") or "이름 없음"), pattern=pattern, number_group=c.get("number_group"), rationale=str(c.get("rationale") or ""), )) if not rules: logger.warning("%s: 후보를 하나도 못 받았다 (model=%s)", STEP, model) return rules, str(result.get("observed_format") or "") def diff_prompt(chosen: SegmentRule, missed: List[str], extra: Optional[List[str]] = None) -> str: """후보끼리 갈린 자리를 **양쪽 다** 보여준다. ★기계 계약만으로는 이걸 못 잡는다(실측). 한 대본에 번호 붙은 헤딩 41개와 번호 없는 헤딩 1개가 섞여 있었는데, 번호를 필수로 요구한 규칙이 41개만 잡고도 **모든 계약을 통과했다** — 번호 있는 것만 세니 연속이고, 헤딩에 글자도 있었다. 놓친 씬은 앞 구간에 합쳐져 흔적도 안 남는다. ★한쪽만 물으면 반대쪽으로 새어 나간다(재현함). 정답을 통째로 포함하는 넓은 규칙은 놓친 자리가 0 이라 신호가 안 뜨는데, 순위에서는 매치 수가 많아 이긴다. 그래서 "저 혼자 잡은 자리"도 같이 돌려준다. 어느 쪽이 옳은지는 여기서 정하지 않는다 — 실측에서 41개보다 42개가 옳았고, 반대로 본문 줄을 다 잡은 규칙은 틀렸다. 대본을 본 모델이 판단할 일이다. 후보 전부가 같은 예외를 놓치면 이 방법도 못 잡는다 — 그건 남는 한계다. """ lines = [f"고른 규칙: `{chosen.pattern}`", ""] if missed: lines.append("① 다른 후보는 잡았는데 이 규칙이 놓친 줄 (원문 그대로):") lines += [f" {m}" for m in missed[:20]] if len(missed) > 20: lines.append(f" … 그리고 {len(missed) - 20}줄 더") lines.append("") if extra: lines.append("② 이 규칙만 잡고 다른 후보는 아무도 안 잡은 줄:") lines += [f" {m}" for m in extra[:20]] if len(extra) > 20: lines.append(f" … 그리고 {len(extra) - 20}줄 더") lines.append("") lines += [ "①이 씬의 시작이라면 그것까지 잡는 규칙을, ②가 씬이 아니라면 그것을", "빼는 규칙을 다시 쓰세요.", "그대로가 맞다고 보면 그렇게 판단한 이유를 rationale 에 적고 같은 규칙을", "다시 내세요.", ] return "\n".join(lines) def retry_prompt(previous: List[SegmentRule], verdicts: List[Any]) -> str: """재저작용 — 무엇이 왜 실패했는지 되돌려준다. ★실패 이유만 주고 금지를 쌓지 않는다. 금지를 얹으면 모델에게 남는 재료가 줄어 더 나쁜 답이 나온다(이 프로젝트의 실측 교훈). 무엇을 봤는지 알려주고 다시 보게 한다. """ lines = ["앞서 낸 후보가 기계 검증을 못 넘었습니다. 각각 이렇게 나왔습니다.", ""] for rule, v in zip(previous, verdicts): if v is None: lines.append(f"- {rule.name}: 실행 자체가 거부됐습니다 (패턴이 안전 검사에 걸림)") continue why = [] if "C1" in v.failures: why.append(f"걸린 줄이 {v.match_count}개뿐입니다") if "C5" in v.failures: back = v.detail.get("C5_backward_count") unread = v.detail.get("C5_unreadable") if back: why.append(f"씬 번호가 {back}번 되돌아갑니다" " (씬 안의 항목 번호를 잡고 있을 수 있습니다)") if unread: why.append(f"매치 {unread}개에서 번호 그룹을 못 읽었습니다") if v.heading_content < 0.5: why.append(f"걸린 줄 중 {round((1 - v.heading_content) * 100)}%가" " 번호·기호뿐이고 내용이 없습니다 (쪽 번호일 수 있습니다)") lines.append(f"- {rule.name} (`{rule.pattern}`): " + ("; ".join(why) if why else "통과했으나 더 나은 후보가 필요합니다")) lines += ["", "대본을 다시 보고 후보를 새로 쓰세요."] return "\n".join(lines)