feat: report original text and judgment criteria for review

검토자가 침해 여부를 눈으로 판별하려면 원문이 있어야 하고, 어떤 조건으로
걸렸는지 알아야 한다. 두 가지를 워크북에 싣는다.

- extract_suspected_excerpts.py: 코퍼스 DB(읽기 전용)에서 질의·상대 원문과
  일치 구간을 뽑는다. 킹서버 기본 python 이 3.6 이라 이 파일만 구문을 맞췄다.
- '판정 기준' 시트: 3개 OR 조건과 결합유사도 가중치, 조건별 충족 현황
- '원문 대조' 시트: 판정 사유 컬럼으로 각 건이 걸린 조건을 표시

판례 표기도 바로잡는다. USE_LLM_LEGAL_JUDGE=false 는 LLM 법적 판단만
끈 것이고 판례 검색은 규칙 기반으로 늘 동작한다. legal_risk.assess() 가
점수 하한 없이 상위 5건을 붙이므로 매칭 미탐지 건에도 판례가 채워진다.
한 줄에 뭉쳐 두면 모순으로 읽혀 판례 검색과 법적 판단을 분리해 적었다.

원문이 담긴 산출물은 gitignore 로 제외한다. 필요하면 위 스크립트로
코퍼스에서 다시 뽑을 수 있어 저장소에 영구 보존할 이유가 없다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
hbyang 2026-09-03 09:49:50 +09:00
parent aed8cd6b48
commit e89cd23ed6
4 changed files with 362 additions and 3 deletions

6
.gitignore vendored
View File

@ -49,3 +49,9 @@ logs/
# Hugging Face 모델 캐시 # Hugging Face 모델 캐시
.cache/ .cache/
~/.cache/huggingface/ ~/.cache/huggingface/
# 원문(개인 자서전 본문)이 담긴 산출물은 저장소에 두지 않는다.
# 필요하면 scripts/extract_suspected_excerpts.py 로 킹서버 코퍼스에서 다시 뽑는다.
reports/excerpts_*.jsonl
reports/침해판별_원문대조_*.xlsx
reports/침해판별_의심건검토_*.xlsx

View File

@ -47,7 +47,7 @@ DETAIL_COLUMNS = [
("lemma 점수", "어휘 유사도", 11), ("lemma 점수", "어휘 유사도", 11),
("주 법령태그", "주 침해유형", 18), ("주 법령태그", "주 침해유형", 18),
("보조 법령태그", "보조 침해유형", 16), ("보조 법령태그", "보조 침해유형", 16),
("관련 판례 사건번호", "관련 판례 사건번호", 46), ("관련 판례 사건번호", "관련 판례 사건번호 (관련성 미검증)", 46),
("judgment_summary", "판단 요약", 70), ("judgment_summary", "판단 요약", 70),
] ]
@ -121,6 +121,15 @@ def build_detail_sheet(wb: Workbook, rows: list[dict]) -> None:
) )
append_rows(sheet, suspected, DETAIL_COLUMNS) append_rows(sheet, suspected, DETAIL_COLUMNS)
sheet.auto_filter.ref = f"A1:{get_column_letter(len(DETAIL_COLUMNS))}{sheet.max_row}" sheet.auto_filter.ref = f"A1:{get_column_letter(len(DETAIL_COLUMNS))}{sheet.max_row}"
sheet.append([])
for line in (
"관련 판례는 침해 판정의 근거가 아닙니다. 엔진이 법령태그와 어휘 유사도로 뽑은 상위 5건이며, "
"점수 하한이 없어 관련성이 낮은 판례도 항상 채워집니다.",
"실제로 매칭 미탐지 건에도 같은 방식으로 판례가 붙습니다. 판례가 있다는 사실만으로 "
"침해 가능성을 판단하시면 안 되며, 사건번호를 직접 확인하셔야 합니다.",
):
sheet.append([line])
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
def build_summary_sheet(wb: Workbook, rows: list[dict], *, backend: str, notes: list[str]) -> None: def build_summary_sheet(wb: Workbook, rows: list[dict], *, backend: str, notes: list[str]) -> None:
@ -146,9 +155,11 @@ def build_summary_sheet(wb: Workbook, rows: list[dict], *, backend: str, notes:
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True) sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
sheet.cell(row=sheet.max_row, column=2).font = Font(bold=True) sheet.cell(row=sheet.max_row, column=2).font = Font(bold=True)
for label, value in ( for label, value in (
("판정 방식", "규칙 기반 (판례 매칭, LLM 법률판단 미사용)"), ("침해 판별", "규칙 기반 (유사도·연속일치·커버리지 3개 조건)"),
("판례 검색", "규칙 기반. 법령태그·어휘 유사도로 상위 5건을 뽑으며 점수 하한이 없음"),
("법적 판단", "미수행 (LLM 법률판단 비활성 — 원문을 외부로 전송하지 않음)"),
("검색 백엔드", backend), ("검색 백엔드", backend),
("원문 텍스트", "결과 파일에 저장하지 않음 (가명 식별자만 수록)"), ("원문 텍스트", "'원문 대조' 시트에 수록 (그 밖의 시트에는 가명 식별자만)"),
): ):
sheet.append([label, value]) sheet.append([label, value])
@ -162,6 +173,199 @@ def build_summary_sheet(wb: Workbook, rows: list[dict], *, backend: str, notes:
sheet.column_dimensions[get_column_letter(index)].width = width sheet.column_dimensions[get_column_letter(index)].width = width
CRITERIA_ROWS = [
(1, "결합유사도", "≥ 0.65", "persistent_similarity_threshold",
"네 가지 유사도를 가중 합산한 값이 임계값 이상"),
(2, "최장 연속 일치 길이", "≥ 80자", "persistent_min_exact_span",
"두 글이 끊기지 않고 그대로 이어지는 최장 구간"),
(3, "문서 단위 일치 커버리지", "≥ 0.30 (30%)", "persistent_min_coverage",
"질의 문서 전체에서 일치 구간이 차지하는 비율(중복 제외)"),
]
WEIGHT_ROWS = [
("어휘 유사도 (lemma)", 0.45, "형태소 원형 기준. 어미·조사 변형을 흡수"),
("표현 유사도 (text)", 0.30, "표층 문자열 기준"),
("문자 유사도 (character)", 0.15, "문자 3~4-gram 기준"),
("모티프 유사도 (motif)", 0.10, "사건·소재 요소 일치"),
]
PARAMETER_ROWS = [
("후보 재정렬 대상", "상위 20건", "persistent_rerank_top_k",
"정밀 비교에 참여하는 후보 수"),
("증거 스팬 최소 길이", "12자", "_evidence_spans(min_match)",
"이보다 짧은 일치는 증거로 세지 않음"),
("증거 스팬 표시 개수", "최대 10개", "_evidence_spans(limit)",
"커버리지 계산은 전량, 표시만 제한"),
("자기 문서 제외", "적용", "source_group 필터",
"동일 문서 및 동일 원천 그룹은 후보에서 제외"),
("판례 검색", "상위 5건 고정", "legal_risk.assess()",
"법령태그 0.55 + 어휘 0.20 + 판시기준 0.10 + 유형 0.08 + 충실도 0.02 로 정렬"),
("판례 점수 하한", "없음", "legal_risk.assess()",
"하한이 없어 관련성이 낮아도 항상 5건이 붙는다. 매칭 미탐지 건에도 판례가 채워지는 이유"),
("법적 판단", "미수행", "USE_LLM_LEGAL_JUDGE=false",
"판례를 근거로 침해 여부를 판단하는 단계는 실행하지 않음. 외부 LLM에 원문 미전송"),
]
def build_criteria_sheet(wb: Workbook, rows: list[dict], excerpts: list[dict] | None = None) -> None:
sheet = wb.create_sheet("판정 기준")
def header(*labels: str) -> None:
sheet.append(list(labels))
for index in range(1, len(labels) + 1):
cell = sheet.cell(row=sheet.max_row, column=index)
cell.font = Font(bold=True)
cell.fill = HEADER_FILL
def title(text: str) -> None:
sheet.append([text])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True, size=12)
def note(text: str) -> None:
sheet.append([text])
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
title("1. 침해 의심 판정 조건")
note("세 조건 중 하나 이상을 충족하면 침해 의심으로 분류합니다 (OR 조건). 세 조건 모두를 요구하지 않습니다.")
header("번호", "조건", "기준값", "설정 키", "설명")
for row in CRITERIA_ROWS:
sheet.append(list(row))
sheet.append([])
title("2. 결합유사도 산식")
note("결합유사도 = 어휘×0.45 + 표현×0.30 + 문자×0.15 + 모티프×0.10 (실측 기반 가중치)")
header("구성 요소", "가중치", "설명")
for label, weight, description in WEIGHT_ROWS:
sheet.append([label, weight, description])
sheet.cell(row=sheet.max_row, column=2).number_format = "0.00"
sheet.append([])
title("3. 조건별 충족 현황 (침해 의심 건 기준)")
suspected = [row for row in rows if row["침해 의심"]]
longest_by_key = {}
if excerpts:
for item in excerpts:
spans = item.get("일치 구간") or []
longest_by_key[item["query_key"]] = max((len(span) for span in spans), default=0)
# 조건별 충족 수와 조합별 건수는 서로 다른 집계다. 한 카운터에 섞으면
# 단독 사유 건이 양쪽에 잡혀 합계가 전체 건수를 넘는다.
condition_counts = Counter()
combo_counts = Counter()
for row in suspected:
reasons = judgment_reasons(row, longest_by_key.get(row["query_key"], 0))
combo_counts[" + ".join(reasons) if reasons else "(사유 미상)"] += 1
for reason in reasons:
condition_counts[reason] += 1
header("구분", "건수", "비고")
sheet.append(["침해 의심 전체", len(suspected), ""])
note("아래 세 줄은 중복 집계입니다. 한 건이 여러 조건을 동시에 충족할 수 있어 합계가 103을 넘습니다.")
sheet.append(["조건 ① 유사도 (0.65 이상) 충족", condition_counts["①유사도"], ""])
if excerpts:
sheet.append(["조건 ② 연속일치 (80자 이상) 충족", condition_counts["②연속일치"], "원문 대조로 실측"])
sheet.append(["조건 ③ 커버리지 (30% 이상) 충족", condition_counts["③커버리지"], ""])
sheet.append([])
sheet.append(["판정 사유 조합별 건수", "", "'원문 대조' 시트의 판정 사유 컬럼과 같은 값. 합계 = 103"])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
for key, value in sorted(combo_counts.items(), key=lambda item: (-item[1], item[0])):
single = " + " not in key and not key.startswith("(")
sheet.append([
key + (" 단독" if single else ""), value,
"이 조건 하나만으로 걸린 건" if single else "",
])
sheet.append(["합계", sum(combo_counts.values()), ""])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
if not excerpts:
note("조건 ②(최장 연속 일치 80자)는 원문 발췌 파일이 있어야 집계할 수 있습니다. --excerpts-jsonl 을 주면 실측값이 채워집니다.")
sheet.append([])
title("4. 그 밖의 동작 기준")
header("항목", "", "설정 키", "설명")
for row in PARAMETER_ROWS:
sheet.append(list(row))
sheet.append([])
title("5. 기준값 신뢰도에 관한 한계")
for line in (
"결합유사도 임계값 0.65는 실데이터의 오탐 분포를 측정해 도출한 값이 아닌 잠정값입니다 "
"(similarity_threshold_calibrated = false). API 응답에도 provisional = true 로 표기됩니다.",
"따라서 현재 결과로 정확도(정밀도·재현율)를 제시할 수 없습니다. "
"원문 대조로 정답 라벨을 만든 뒤 임계값을 재조정하는 것이 다음 단계입니다.",
"'매칭 미탐지'는 등록 코퍼스 안에서 일치 증거를 찾지 못했다는 뜻이며 비침해 확정이 아닙니다.",
):
note(line)
for index, width in enumerate((26, 24, 18, 30, 62), start=1):
sheet.column_dimensions[get_column_letter(index)].width = width
SCORE_MIN = 0.65
SPAN_MIN = 80
COVERAGE_MIN = 0.30
def judgment_reasons(row: dict, longest_span: int) -> list[str]:
"""어느 조건으로 의심 판정이 났는지. detector.py 의 OR 조건과 같은 기준."""
reasons = []
if (row.get("결합유사도") or 0) >= SCORE_MIN:
reasons.append("①유사도")
if longest_span >= SPAN_MIN:
reasons.append("②연속일치")
if (row.get("union_coverage") or 0) >= COVERAGE_MIN:
reasons.append("③커버리지")
return reasons
EXCERPT_COLUMNS = [
("가명 제목", 20), ("침해 여부", 12), ("판정 사유", 26), ("매칭 상대 제목", 20),
("결합유사도", 11), ("일치 구간 길이", 13),
("일치 구간 (똑같은 문장)", 70), ("검사 대상 원문", 70), ("매칭 상대 원문", 70),
]
def build_excerpt_sheet(wb: Workbook, excerpts: list[dict], records: dict[str, dict]) -> None:
sheet = wb.create_sheet("원문 대조")
sheet.append([label for label, _ in EXCERPT_COLUMNS])
for index, (_, width) in enumerate(EXCERPT_COLUMNS, start=1):
cell = sheet.cell(row=1, column=index)
cell.font = Font(bold=True)
cell.fill = HEADER_FILL
sheet.column_dimensions[get_column_letter(index)].width = width
sheet.freeze_panes = "D2"
legend = (
"판정 사유: ①유사도 = 결합유사도 0.65 이상 / ②연속일치 = 똑같은 글자가 80자 이상 이어짐 "
"/ ③커버리지 = 문서의 30% 이상이 겹침. 하나만 충족해도 침해 의심입니다."
)
order = {"침해 의심": 0, "매칭 미탐지": 1}
ordered = sorted(
excerpts,
key=lambda r: (order.get(r.get("침해 여부"), 9), -(r.get("결합유사도") or 0)),
)
for row in ordered:
spans = row.get("일치 구간") or []
longest = max((len(span) for span in spans), default=0)
record = records.get(row["query_key"], {})
if row.get("침해 여부") == "침해 의심":
reasons = " + ".join(judgment_reasons(record, longest)) or "(사유 미상)"
else:
reasons = "-"
sheet.append([
row.get("가명 제목"), row.get("침해 여부"), reasons,
row.get("매칭 상대 제목") or "-", row.get("결합유사도"), longest,
"\n---\n".join(spans) if spans else "(일치 구간 없음)", row.get("검사 대상 원문"),
row.get("매칭 상대 원문") or "(매칭된 상대 글 없음)",
])
for column in (7, 8, 9):
sheet.cell(row=sheet.max_row, column=column).alignment = Alignment(
wrap_text=True, vertical="top"
)
sheet.cell(row=sheet.max_row, column=5).number_format = "0.0000"
sheet.auto_filter.ref = f"A1:{get_column_letter(len(EXCERPT_COLUMNS))}{sheet.max_row}"
sheet.append([])
sheet.append([legend])
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
COMPARE_COLUMNS = [ COMPARE_COLUMNS = [
"가명 제목", "원천 구분", "매칭 상대 제목", "가명 제목", "원천 구분", "매칭 상대 제목",
"1차 유사도", "2차 유사도", "근거 스팬(1차)", "근거 스팬(2차)", "1차 유사도", "2차 유사도", "근거 스팬(1차)", "근거 스팬(2차)",
@ -233,6 +437,8 @@ def main() -> int:
parser.add_argument("--compare-jsonl", type=Path, help="비교 시트를 붙일 이전 회차 JSONL") parser.add_argument("--compare-jsonl", type=Path, help="비교 시트를 붙일 이전 회차 JSONL")
parser.add_argument("--backend", default="", help="검색 백엔드 표기 (미지정 시 JSONL 값 사용)") parser.add_argument("--backend", default="", help="검색 백엔드 표기 (미지정 시 JSONL 값 사용)")
parser.add_argument("--note", action="append", default=[], help="요약 시트에 남길 유의사항") parser.add_argument("--note", action="append", default=[], help="요약 시트에 남길 유의사항")
parser.add_argument("--excerpts-jsonl", type=Path,
help="원문 대조 시트를 붙일 발췌 JSONL (extract_suspected_excerpts.py 산출물)")
args = parser.parse_args() args = parser.parse_args()
records = load(args.jsonl) records = load(args.jsonl)
@ -243,6 +449,15 @@ def main() -> int:
wb.remove(wb.active) wb.remove(wb.active)
build_result_sheet(wb, rows) build_result_sheet(wb, rows)
build_detail_sheet(wb, rows) build_detail_sheet(wb, rows)
excerpts = None
if args.excerpts_jsonl:
excerpts = [
json.loads(line)
for line in args.excerpts_jsonl.read_text(encoding="utf-8").splitlines()
if line.strip()
]
build_excerpt_sheet(wb, excerpts, records)
build_criteria_sheet(wb, rows, excerpts if args.excerpts_jsonl else None)
build_summary_sheet(wb, rows, backend=backend, notes=args.note) build_summary_sheet(wb, rows, backend=backend, notes=args.note)
if args.compare_jsonl: if args.compare_jsonl:
build_compare_sheet(wb, load(args.compare_jsonl), records) build_compare_sheet(wb, load(args.compare_jsonl), records)

View File

@ -0,0 +1,138 @@
"""판별 대상의 원문과 일치 구간을 뽑아 JSONL로 쓴다.
배치 결과 파일에는 원문이 없다. 검토자가 실제 침해인지 눈으로 판별하려면
질의 원문과 매칭 상대 원문, 그리고 사이의 일치 구간이 필요하다.
코퍼스 DB는 읽기 전용으로만 연다.
킹서버의 기본 python3 구버전이라 파일만은 표준 라이브러리와 구문
호환 범위 안에서 쓴다 (타입 표기·f-string 미사용, __future__ import 없음).
저장소의 다른 모듈과 스타일이 다른 이유가 이것이다.
원문이 포함된 산출물이므로 취급에 주의한다.
"""
import argparse
import json
import sqlite3
from difflib import SequenceMatcher
# app.engine.persistent_index._evidence_spans 와 같은 기준을 쓴다.
MIN_MATCH = 12
SPAN_LIMIT = 10
def open_readonly(path):
con = sqlite3.connect("file:%s?mode=ro" % path, uri=True)
con.row_factory = sqlite3.Row
return con
def fetch_segments(con, segment_ids):
texts = {}
ids = [sid for sid in segment_ids if sid]
for start in range(0, len(ids), 500):
chunk = ids[start:start + 500]
placeholders = ",".join("?" * len(chunk))
rows = con.execute(
"SELECT segment_id, text FROM segments WHERE segment_id IN (%s)" % placeholders,
chunk,
)
for row in rows:
texts[row["segment_id"]] = row["text"]
return texts
def fetch_document_texts(con, document_ids):
"""문서 단위 질의(생활수기)용. 세그먼트를 순서대로 이어 붙인다."""
texts = {}
for document_id in document_ids:
rows = con.execute(
"SELECT text FROM segments WHERE document_id = ? ORDER BY ordinal, segment_id",
(document_id,),
).fetchall()
if rows:
texts[document_id] = "\n".join(row["text"] for row in rows)
return texts
def matching_spans(query, reference):
if not query or not reference:
return []
blocks = SequenceMatcher(None, query, reference, autojunk=False).get_matching_blocks()
useful = [b for b in blocks if b.size >= MIN_MATCH]
useful.sort(key=lambda b: (-b.size, b.a))
selected = sorted(useful[:SPAN_LIMIT], key=lambda b: b.a)
return [query[b.a:b.a + b.size] for b in selected]
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--database", required=True)
parser.add_argument("--jsonl", required=True, help="배치 결과 JSONL")
parser.add_argument("--out-jsonl", required=True)
parser.add_argument("--max-chars", type=int, default=6000,
help="원문 컬럼 길이 상한. 0 이면 제한 없음")
parser.add_argument("--scope", choices=("all", "suspected"), default="all",
help="all=전 건(미탐지 포함), suspected=침해 의심 건만")
args = parser.parse_args()
with open(args.jsonl, encoding="utf-8") as handle:
all_rows = [json.loads(line) for line in handle if line.strip()]
if args.scope == "all":
targets = all_rows
else:
targets = [row for row in all_rows if row.get("침해 의심")]
flagged = sum(1 for row in targets if row.get("침해 의심"))
print("대상 %d건 (침해 의심 %d / 매칭 미탐지 %d)"
% (len(targets), flagged, len(targets) - flagged))
con = open_readonly(args.database)
try:
segment_ids = set()
document_ids = set()
for row in targets:
for key in ("query_segment_id", "매칭 상대 segment"):
if row.get(key):
segment_ids.add(row[key])
if not row.get("query_segment_id"):
document_ids.add(row["doc_id"])
segment_texts = fetch_segments(con, segment_ids)
document_texts = fetch_document_texts(con, document_ids)
finally:
con.close()
def clip(text):
if args.max_chars and len(text) > args.max_chars:
return text[:args.max_chars] + "\n…(이하 %d자 생략)" % (len(text) - args.max_chars)
return text
written = 0
missing = 0
with open(args.out_jsonl, "w", encoding="utf-8") as handle:
for row in targets:
query_key = row["query_segment_id"] or row["doc_id"]
query_text = segment_texts.get(query_key) or document_texts.get(query_key, "")
source_text = segment_texts.get(row.get("매칭 상대 segment") or "", "")
# 미탐지 건은 매칭 상대가 없는 것이 정상이므로 질의 원문만 확인한다.
if not query_text:
missing += 1
record = {
"query_key": row["query_key"],
"가명 제목": row["가명 제목"],
"침해 여부": row["침해 여부"],
"매칭 상대 제목": row.get("매칭 상대 제목"),
"원천 구분": row["원천 구분"],
"결합유사도": row["결합유사도"],
"근거 스팬 수": row.get("evidence 스팬 수"),
"일치 구간": matching_spans(query_text, source_text),
"검사 대상 원문": clip(query_text),
"매칭 상대 원문": clip(source_text),
}
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
written += 1
print("wrote %s rows=%d 원문누락=%d" % (args.out_jsonl, written, missing))
return 0
if __name__ == "__main__":
raise SystemExit(main())