#!/bin/sh
# Claude Code Skill 설치 스크립트
# 생성: 2026-08-11 · Skill 9개 · 파일 17개
#
# ./install-skills.sh ~/.claude/skills/ 에 설치 (모든 프로젝트에서 사용)
# ./install-skills.sh ./myrepo ./myrepo/.claude/skills/ 에 설치 (그 저장소 전용)
set -e
DEST="${1:+$1/.claude/skills}"
DEST="${DEST:-$HOME/.claude/skills}"
mkdir -p "$DEST"
echo "설치 위치: $DEST"
mkdir -p "$DEST/cancer-type-classify"
cat > "$DEST/cancer-type-classify/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: cancer-type-classify
description: 임상시험 제목·대상질환·연구목적 텍스트를 암 여부 + 암종(폐암/위암/담도암/유방암 등 20여종)으로 분류한다. 하드필터 + LLM 2단계 + 과거 피드백 RAG 보정의 3단 하이브리드. TRIGGER - "암종 분류", "무슨 암인지 구분", "oncology 여부", "담도암 임상시험 찾기", 임상시험/논문/문서를 암종별로 나눠야 할 때, 분류 결과에 사람 피드백을 반영해 개선하려 할 때.
---
# 암종 분류
## 언제 쓰나
텍스트 한 덩어리(임상시험 제목, 논문 제목, 공고문)를 **암 여부 → 암종**으로 나눠야 할 때. 담도암 환자에게 어떤 임상시험이 있는지 찾는 것 같은 조합 작업의 핵심 부품이다.
조합 예: `[[ctrial-collect]]`로 임상시험을 긁고 → 이 skill로 암종을 붙이고 → 담도암(`isBileDuctCancer`)만 필터 → `[[regimen-extract]]`로 레지멘 대조.
## 정본 코드
```
~/Documents/IMOK/ctrial-auto/app/
services/clinical_classification_service.py 296줄 메인 로직
utils/oncology_constants.py SHEET_MAPPING, CANCER_FIELDS
prompt/oncology_detection.md 1단계: 암 여부
prompt/cancer_typing.md 2단계: 암종 분류
prompt/apply_feedback.md 3단계: 피드백 보정
```
## 바로 쓰는 코드
`scripts/feedback_guard.py` — 피드백 보정이 분류를 전부 지우는 사고를 막는다.
`parse_json_result` · `cosine` · `find_similar`(임계 0.6, 상위 3건) · `apply_feedback`(전면삭제 원복 + Rule 9 의도적 비움 예외).
```python
from feedback_guard import apply_feedback
result, rolled_back, why = apply_feedback(original, improved, CANCER_FIELDS)
```
## 3단 하이브리드
**LLM을 언제 안 부를지가 설계의 절반이다.**
```
1) 하드 필터 대상질환 카테고리 ∈ {Oncology, Hemato-oncology} → 암 확정, GPT 호출 생략
2) LLM 1단계 그 외에만 oncology_detection 프롬프트로 암 여부 판정
3) LLM 2단계 암일 때만 cancer_typing으로 전체 암종 JSON
4) 피드백 보정 유사 과거 사례를 찾아 apply_feedback으로 수정
```
공공데이터 API가 이미 `TRGT_DISS_CD_NM`으로 암 여부를 알려주는데 그걸 다시 LLM에 묻는 건 낭비다. 하드필터로 걸러진 건은 GPT 1단계를 통째로 건너뛴다.
## 입력을 풍부하게 넣어라
제목만 주면 분류가 흔들린다. `_build_input_block()`이 있는 것만 골라 붙인다:
```
Clinical trial title (Korean): ...
English title: ...
Target disease name: ...
Target disease category (MFDS hint, may be inaccurate): ...
Research purpose: ... (500자로 자름)
```
카테고리 힌트에 **"may be inaccurate"를 명시**한다. 안 쓰면 LLM이 부정확한 MFDS 카테고리를 맹신한다. 연구목적은 길어서 500자로 자른다.
## 출력 형식: `[판정, 확신도, 근거]`
각 암종 필드가 3원소 배열이다.
```json
{"isBileDuctCancer": ["1", "0.9", "담도암 환자 대상 2상 시험"],
"isLungCancer": ["0", "1.0", "암 관련이 아님"]}
```
**근거를 같이 받는 게 핵심이다.** 의료 데이터에서 분류만 있고 근거가 없으면 검수를 못 한다. 확신도는 낮은 것만 사람이 보게 정렬하는 데 쓴다.
최종적으로 `SHEET_MAPPING`(영문 필드 → 한글)으로 `"위암;담도암"` 형태 문자열을 만들어 DB에 넣는다. 이 매핑은 분류·내보내기가 공유하는 **단일 소스**여야 한다 — 두 군데 두면 시트명이 어긋난다.
## 피드백 RAG 자가개선
사람이 분류를 고치면 `change_history` 테이블에 (제목, 변경사유, 이전분류, 새분류, **제목 임베딩**)을 남긴다. 다음 분류 때:
```
현재 제목 임베딩 → 저장된 임베딩과 코사인 유사도
→ 0.6 초과만 채택, 상위 3건
→ apply_feedback 프롬프트에 유사사례로 주입
```
**임베딩은 저장 시점 것을 재사용한다** — 매번 다시 만들면 느리고, 저장·질의가 다른 임베딩 모델을 쓰면 유사도가 무의미해진다. 그래서 `_embed()`를 공용 헬퍼 하나로 일원화했다.
### 가드가 필수다
피드백이 분류를 **전부 0으로 지워버리는 사고**가 실제로 났다. 그래서:
```
피드백 적용 전 양성(1) 암종 스냅샷 저장
→ 적용 후 양성이 하나도 안 남았고
→ 사유에 "진단/이식/표식/supportive/의료기기/신약 아님/관찰 연구" 키워드가 없으면
→ 스냅샷으로 원복 + WARNING 로그
```
키워드가 있으면 **의도적 비움**(진단용·기기 평가 등 신약 연구가 아닌 경우)이라 그대로 둔다. LLM 보정 루프를 넣을 때는 "보정이 모든 결과를 지울 수 있다"를 항상 가정하고 되돌림 경로를 만들어라.
## 현재 상태 주의
정본에서 피드백 단계가 꺼져 있다:
```python
sims = [] # 피드백 비활성화 (복원: sims = _find_similar_feedbacks(trial_title))
```
복원하려면 이 한 줄을 되돌리면 된다. `clinical_classification_service.py.bak_feedback_off` 백업 파일도 같은 디렉토리에 있다.
## 함정
- **JSON 파싱**: `_parse_json_result()`가 ` ```json ` 블록과 생 JSON을 모두 처리하고 실패 시 `None`을 반환한다. `or {}`로 받아 빈 dict 폴백. LLM 응답을 `json.loads` 한 번으로 끝내려 하지 마라.
- **`temperature=0` 고정**: 분류는 재현성이 전부다.
- **암 아님으로 판정되면 2·3단계를 건너뛴다**: 비용의 대부분이 여기서 줄어든다.
## 연관 skill
`[[ctrial-collect]]`(입력 공급), `[[regimen-extract]]`, `[[lit-relevance-classify]]`(같은 피드백 RAG 패턴), `[[llm-provider-switch]]`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/cancer-type-classify/scripts"
cat > "$DEST/cancer-type-classify/scripts/feedback_guard.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""LLM 보정이 분류 결과를 통째로 지우는 사고를 막는다.
출처: ctrial-auto/app/services/clinical_classification_service.py 의
apply_feedback 가드와 유사 피드백 검색(_parse_json_result / _cos_sim / _find_similar_feedbacks).
실제로 난 사고: 과거 피드백을 반영시켰더니 LLM이 양성 암종을 전부 0으로 지웠다.
그래서 보정 전 스냅샷을 떠두고, 전부 지워졌으면 되돌린다.
단 '진단·이식·의료기기·신약 아님' 같은 **의도적 비움**은 그대로 둔다.
LLM 보정 루프를 넣을 때는 "보정이 모든 결과를 지울 수 있다"를 항상 가정하고
되돌림 경로를 만들어라.
python feedback_guard.py --self-check
"""
import argparse
import json
import re
import sys
# 의도적으로 암종을 비우는 정당한 사유 (원본 apply_feedback Rule 9)
INTENTIONAL_EMPTY_KEYWORDS = (
"진단", "이식", "형광", "표식", "supportive", "의료기기", "기기 평가",
"신약연구가 아니", "신약 연구가 아니", "암신약연구아님", "신약이 아니",
"관찰 연구", "추적 관찰",
)
SIMILARITY_THRESHOLD = 0.6 # 이보다 낮은 과거 사례는 참고하지 않는다
TOP_N = 3
def parse_json_result(text):
"""LLM 응답에서 JSON 을 꺼낸다. 실패하면 None.
```json 블록과 생 JSON 을 모두 처리한다.
LLM 응답을 json.loads 한 번으로 끝내려 하지 마라."""
if not text:
return None
try:
if "```" in text:
for block in text.split("```"):
b = block.strip()
if not b:
continue
if b.startswith("json"):
b = b[4:].strip()
try:
return json.loads(b)
except Exception:
continue
return json.loads(text)
except Exception:
return None
def cosine(v1, v2):
"""코사인 유사도. 길이가 다르거나 비면 0."""
if not v1 or not v2 or len(v1) != len(v2):
return 0.0
dot = sum(a * b for a, b in zip(v1, v2))
n1 = sum(a * a for a in v1) ** 0.5
n2 = sum(b * b for b in v2) ** 0.5
return dot / (n1 * n2) if n1 and n2 else 0.0
def find_similar(current_vec, history, threshold=SIMILARITY_THRESHOLD, top_n=TOP_N):
"""과거 교정 이력에서 비슷한 사례를 고른다.
history: [{"title":…, "embedding":[…], "old":[…], "new":[…], "reason":…}]
저장 시점 임베딩을 그대로 쓴다 — 매번 다시 만들면 느리고, 저장·질의가
다른 모델을 쓰면 유사도가 무의미해진다."""
out = []
for h in history:
s = cosine(current_vec, h.get("embedding"))
if s <= threshold:
continue
out.append({**{k: v for k, v in h.items() if k != "embedding"}, "similarity": s})
out.sort(key=lambda x: x["similarity"], reverse=True)
return out[:top_n]
def positive_fields(classification, fields):
"""양성(1)으로 판정된 필드만. 값은 [판정, 확신도, 근거] 3원소 배열."""
return {k: v for k, v in classification.items()
if k in fields and isinstance(v, list) and v and str(v[0]) == "1"}
def is_intentional_empty(classification, keywords=INTENTIONAL_EMPTY_KEYWORDS):
"""비움이 의도적인지 — 근거 텍스트 전체를 훑어 판단."""
blob = str(classification.get("_analysis", "")) + " " + " ".join(
str(v[2]) for v in classification.values()
if isinstance(v, list) and len(v) >= 3)
return any(kw in blob for kw in keywords)
def apply_feedback(original, improved, fields, is_positive_case=True,
keywords=INTENTIONAL_EMPTY_KEYWORDS):
"""보정 결과를 적용하되 전면 삭제는 되돌린다.
→ (최종 분류, 되돌렸는지, 사유)"""
result = dict(original)
# 3원소 배열 형태만 반영한다. 형식이 깨진 값은 무시.
for k, v in (improved.get("modified_classification") or {}).items():
if k in fields and isinstance(v, list) and len(v) >= 2:
result[k] = v
if "analysis" in improved:
result["_analysis"] = improved.get("analysis", "")
pre = positive_fields(original, fields)
post = positive_fields(result, fields)
if is_positive_case and pre and not post:
if is_intentional_empty(result, keywords):
return result, False, "의도적 비움(비신약·진단성 시험)으로 판단 → 유지"
for k, v in pre.items():
result[k] = v
return result, True, f"보정이 양성 판정을 전부 제거 → 원복 (복원={sorted(pre)})"
return result, False, ""
# ── 자체 점검 ───────────────────────────────────────────────────
FIELDS = {"isLungCancer", "isBileDuctCancer", "isGastricCancer"}
POS = ["1", "0.9", "담도암 대상 2상 시험"]
NEG = ["0", "1.0", "암 관련이 아님"]
def _self_check():
assert parse_json_result('```json\n{"a": 1}\n```') == {"a": 1}
assert parse_json_result('```\n{"a": 2}\n```') == {"a": 2}
assert parse_json_result('{"a": 3}') == {"a": 3}
assert parse_json_result("이건 JSON 아님") is None
assert parse_json_result("") is None
assert cosine([1, 0], [1, 0]) == 1.0
assert abs(cosine([1, 0], [0, 1])) < 1e-9
assert cosine([1, 0], [1, 0, 0]) == 0.0 # 길이 불일치
assert cosine([], [1]) == 0.0
assert cosine([0, 0], [1, 1]) == 0.0 # 0 벡터
hist = [{"title": "A", "embedding": [1, 0], "reason": "r1"},
{"title": "B", "embedding": [0, 1], "reason": "r2"},
{"title": "C", "embedding": [0.9, 0.1], "reason": "r3"}]
sim = find_similar([1, 0], hist)
assert [h["title"] for h in sim] == ["A", "C"], sim # B는 임계값 미달
assert "embedding" not in sim[0] # 임베딩은 돌려주지 않는다
assert find_similar([1, 0], []) == []
orig = {"isBileDuctCancer": POS, "isLungCancer": NEG}
assert set(positive_fields(orig, FIELDS)) == {"isBileDuctCancer"}
# 1) 정상 보정 — 암종이 바뀌는 건 그대로 반영
res, rolled, _ = apply_feedback(
orig, {"modified_classification": {"isGastricCancer": ["1", "0.8", "위암"],
"isBileDuctCancer": NEG},
"analysis": "위암으로 정정"}, FIELDS)
assert not rolled and set(positive_fields(res, FIELDS)) == {"isGastricCancer"}
assert res["_analysis"] == "위암으로 정정"
# 2) 사고 재현 — 전부 0으로 지워지면 되돌린다
res, rolled, why = apply_feedback(
orig, {"modified_classification": {"isBileDuctCancer": NEG}, "analysis": "암 아님"}, FIELDS)
assert rolled and res["isBileDuctCancer"] == POS, (rolled, res)
assert "원복" in why
# 3) 의도적 비움은 유지 — 진단용·기기 평가 등
res, rolled, why = apply_feedback(
orig, {"modified_classification": {"isBileDuctCancer": NEG},
"analysis": "진단 목적 의료기기 평가로 암 신약 연구가 아님"}, FIELDS)
assert not rolled and str(res["isBileDuctCancer"][0]) == "0", (rolled, res)
assert "의도적" in why
# 4) 애초에 양성이 없었으면 되돌릴 것도 없다
res, rolled, _ = apply_feedback({"isLungCancer": NEG}, {"modified_classification": {}}, FIELDS)
assert not rolled
# 5) 암 확정이 아닌 케이스는 가드를 걸지 않는다
res, rolled, _ = apply_feedback(
orig, {"modified_classification": {"isBileDuctCancer": NEG}}, FIELDS,
is_positive_case=False)
assert not rolled
# 6) 형식이 깨진 보정값은 무시
res, _, _ = apply_feedback(orig, {"modified_classification": {"isLungCancer": "1"}}, FIELDS)
assert res["isLungCancer"] == NEG
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
p.error("--self-check 로 동작을 확인하거나 모듈로 import 해서 쓰세요")
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/cancer-type-classify/scripts/feedback_guard.py"
mkdir -p "$DEST/ctrial-collect"
cat > "$DEST/ctrial-collect/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: ctrial-collect
description: 공공데이터포털(식약처 MFDS) 임상시험 API로 시험 목록·상세를 수집해 DB/Excel로 적재하고, 회차 간 변경분(신규·상태변경·종료)을 비교 산출한다. 목록+상세 2단 API, 재시도, 증분 갱신, 날짜 기반 파일 세대 관리. TRIGGER - "임상시험 수집", "임상시험 목록 긁기", "MFDS/식약처 API", "공공데이터포털", "임상시험 데이터 갱신", 특정 암종의 진행 중 임상시험을 찾아야 할 때.
---
# 임상시험 데이터 수집
## 언제 쓰나
식약처 승인 임상시험을 주기적으로 긁어 "지난번 대비 뭐가 새로 생겼고 뭐가 끝났는지"를 내야 할 때. 특정 암종의 현재 모집 중 시험을 찾는 작업의 입력을 만든다.
조합: 이 skill로 수집 → `[[cancer-type-classify]]`로 암종 부착 → 담도암만 필터.
## 정본 코드
```
~/Documents/IMOK/ctrial-auto/app/
services/api_download_service.py 388줄 ★ 공공데이터 API 수집 (정본)
services/data_download_service.py 242줄 구버전 Playwright 크롤링 (사용 안 함)
services/excel_compare_service.py 539줄 회차 간 변경 비교
services/url_update_service.py 83줄 MFDS 상세 URL 역추적
services/export_excel_service.py 70줄 내보내기
pipeline/pipeline_manager.py 246줄 단계 진행 상태 관리
prompt/ 분류 프롬프트 (cancer-type-classify skill 참조)
```
`ctrial-auto` 계열은 6벌(`_2월`, `_4월`, `_api`, `CTrial_AUTO`, `ctrial_auto_module`)이 있다. **`IMOK/ctrial-auto`가 최신(2026-07-30)이고 이것이 정본이다.**
## 수집: 목록 + 상세 2단 API
```
목록 API getMdcinClincTestInfoList02 → 최근 CLNC_TEST_SN(시험번호) 확보 + 실시기관명
상세 API getClncExamPlanDtlInq2 → 시험번호별 STATUS·대상질환·영문제목·연구목적·성분명
```
**목록만으로는 분류에 쓸 정보가 부족하다.** 상세 API가 주는 대상질환명/카테고리/영문제목/연구목적/성분명 5개 컬럼이 `[[cancer-type-classify]]`의 입력 품질을 결정한다. 목록에서 얻은 실시기관은 `_lab_map: {CLNC_TEST_SN: 실시기관}`에 담아 상세 결과와 합친다 — 상세 API에는 실시기관이 없다.
호출 사이에 `call_delay`를 둔다. 공공데이터포털은 연속 호출을 차단한다. `max_retries`로 재시도.
### 웹 크롤링에서 API로 갈아탄 이유
원래 Playwright로 긁었다(`data_download_service.py`). API로 바꾸면서 **출력 엑셀 형식은 완전히 동일하게 유지하고 컬럼 5개만 추가**했다. 이게 중요하다 — 다운스트림(비교·분류·내보내기)을 하나도 안 고치고 수집 방식만 교체할 수 있었다. 수집기를 바꿀 때는 출력 스키마를 먼저 고정해라.
`url_update_service.py`의 Playwright는 아직 필요하다. MFDS 상세 페이지 URL은 API로 안 나와서 검색 URL로 역추적한다.
## 로컬 필터링
API가 아니라 받은 뒤에 나눈다:
```
모집중 + 승인일 3년 이내 → recruiting_trials.xlsx
승인완료 + 승인일 6개월 이내 → completed_trials.xlsx
```
기준 기간은 용도마다 다르니 상수로 빼둬라.
## 변경 비교 (회차 간 증분)
`excel_compare_service.py`가 핵심이자 가장 큰 파일이다.
```
excel_old/ 에서 파일명 날짜로 '가장 최근' 1개 선택
→ excel_output/ 신규 파일과 비교
→ step1_output/ 에 비교결과 + updated 파일 생성
```
**파일명에 날짜를 박아 세대를 관리한다**(`extract_date_from_filename`, `find_latest_file_by_date`). DB에도 `20260403.db`, `20260618.db`, `20260701.db` 식으로 스냅샷을 남긴다. 임상시험은 상태가 바뀌므로 "그때 이 시험이 뭐였는지"를 복원할 수 있어야 한다.
비교 전에 양쪽 다 중복 제거를 한다 — 신규는 시험번호 기준(`dedup_file2_by_seq`), 기존은 라벨을 붙이며(`dedup_and_label_file1`). 제목은 `norm_title()`로 정규화해서 비교한다. 공백·괄호 차이로 같은 시험이 신규로 잡히는 사고를 막는다.
`find_previous_statuses`(`db_history`)로 이전 상태 이력을 끌어와 "모집중 → 종료" 같은 전이를 잡는다.
## 파이프라인 진행 상태
`PipelineManager`는 싱글턴이고 파이프라인 종류별로 상태를 따로 들고 있다:
```
current_step / total_steps / progress(%) / completed_steps / 실패 시 error
is_pipeline_running() 중복 실행 방지
```
수집은 수십 분 걸리므로 **웹 UI가 진행률을 물어볼 창구가 필요하다**. 동기 서비스는 `asyncio.to_thread`로 감싸 백그라운드에서 돈다(`imok_pipeline.py`).
## 함정
- **`mfds_updater_service.py`의 `DEFAULT_DB_MODE = "local"`** 에 `⚠️ 테스트용` 주석이 달려 있다. 운영에 쓸 때 반드시 확인해라.
- 외부 Gradle 프로젝트(`MfdsUpdater2`)를 `subprocess`로 부른다. 타임아웃 300초. 이 의존성이 있는지 먼저 확인.
- Elasticsearch 색인(`elasticsearch_service.py`)은 선택 단계다. ES 없이도 수집·비교는 돈다.
## 유사 프로젝트
`IMOK/iqvia`가 같은 구조(`excel_input`/`excel_old`/`excel_output`/`db`)로 IQVIA 데이터를 처리한다. `IMOK/ctrial-mapper`는 수집물을 PI(연구자) 중심으로 재구성한다.
## 연관 skill
`[[cancer-type-classify]]`, `[[excel-case-validator]]`, `[[regimen-extract]]`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/excel-case-validator"
cat > "$DEST/excel-case-validator/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: excel-case-validator
description: 원본 Excel 여러 개와 통합/처리 결과 Excel을 교차검증한다. 건수 일치, 키(PMID/DOI/시험번호) 중복, 필수 컬럼 완전성, 원본→통합 누락, 값 범위를 검사해 마크다운 리포트를 낸다. 제목 정규화 비교, 세대별 파일 관리, 중복 제거 패턴 포함. TRIGGER - "엑셀 검증", "데이터 무결성", "건수 안 맞음", "중복 확인", "누락 확인", "원본이랑 결과 비교", 처리 결과 Excel을 원본과 대조해야 할 때.
---
# Excel 케이스 교차검증
## 언제 쓰나
거의 모든 프로젝트에서 쓴다(Excel 취급 파일 1,816개). 원본 여러 개를 합치거나 LLM으로 처리한 뒤 **"몇 건이 어디로 샜는지"**를 확인해야 할 때. 검증 없이 넘어간 결과는 나중에 반드시 문제가 된다.
## 바로 실행
```bash
python ~/.claude/skills/excel-case-validator/scripts/validate.py \
--sources wos.xlsx pubmed.xlsx koreamed.xlsx \
--final consolidated.xlsx \
--key PMID --key DOI \
--required Title Year \
--year-col Year --year-range 1963 2026
```
마크다운 표로 리포트를 내고, FAIL이 있으면 exit code 1. 원본을 절대 수정하지 않는다(read-only).
```bash
python .../validate.py --self-check # 내장 테스트
```
`scripts/dedup.py` — 중복 제거와 세대 관리.
```bash
python3 scripts/dedup.py --file data.xlsx --key clinicExamSeq --latest-by 승인일 --out clean.xlsx
python3 scripts/dedup.py --latest-in ./excel_old # 파일명 날짜로 최신 파일 찾기
```
`norm_title`(NFKC+공백 정규화) · `dedup`(승인일 최신 → 정보량 순 선택) · `diff_by_key`(회차 간 신규/제거) · `find_latest_by_date`.
## 검사 항목
| 검사 | 내용 |
|---|---|
| 건수 | 원본별 건수 + 통합 ≤ 원본 합계 |
| 키 중복 | `--key` 컬럼의 중복. **빈값은 중복으로 세지 않는다** |
| 필수 컬럼 완전성 | `--required` 컬럼의 채움 비율 |
| 누락 | 원본에는 있는데 통합에 없는 키 (예시 3건까지 표시) |
| 값 범위 | 연도 등 수치 범위 밖 값 |
**빈 키를 중복으로 세지 마라.** PMID 없는 논문이 여러 건인 건 정상이다. 이걸 놓치면 검증이 항상 FAIL로 나와 아무도 안 보게 된다.
## 비교는 정규화 후에 한다
`_norm()` — NFKC 정규화 → 앞뒤 공백 제거 → 연속 공백 축약 → 소문자.
제목으로 비교할 때는 공백까지 전부 제거하는 옵션이 필요하다. 정본 구현(`ctrial-auto/app/utils/compare_utils.py`의 `norm_title(s, remove_all_spaces=True)`)을 참고. **공백·괄호·전각문자 차이로 같은 레코드가 "신규"로 잡히는 사고**가 실제로 반복됐다.
## 프로젝트별 검증 항목은 별도로 정의한다
범용 스크립트로는 도메인 규칙을 못 잡는다. 프로젝트마다 검증 목록을 문서로 고정하고 전용 스크립트를 둔다. 좋은 예:
```
~/Documents/IMOK/AML260130_Korea AML MDS 30y/
.claude/agents/data-validator.md 검증 10항목을 명시 (건수·중복·완전성·초록 보유율 93.5%·연도 1963~2026·색상 코딩)
scripts/validate_data.py
docs/VALIDATION_REPORT.md
```
**기대값을 숫자로 박아둔다**(WoS 1,679건 / PubMed 1,914건 / KoreaMed 761건 → 통합 2,583건). "대충 맞는 것 같다"로는 회귀를 못 잡는다.
## 중복 제거·세대 관리
`ctrial-auto/app/utils/compare_utils.py`에 실전에서 다듬어진 함수들이 있다:
| 함수 | 역할 |
|---|---|
| `dedup_file2_by_seq` | 신규 파일을 일련번호 기준 중복 제거 |
| `dedup_and_label_file1` | 기존 파일 중복 제거 + 라벨 부착 |
| `_pick_latest_by_approval` | 중복 그룹에서 승인일 최신 건 선택 |
| `_info_richness` | 정보가 더 많은 행을 남기는 기준 |
| `extract_date_from_filename` / `find_latest_file_by_date` | 파일명 날짜로 세대 관리 |
| `read_older_excels_newest_to_oldest` | 과거 파일을 최신순으로 순회 |
**중복 제거 시 어느 행을 남길지는 규칙이 필요하다.** 아무거나 남기면 정보가 적은 행이 살아남는다. `_info_richness`(채워진 필드 수)나 `_pick_latest_by_approval`(최신 승인일) 같은 명시적 기준을 써라.
## 함정
- `~$파일명.xlsx` 임시 파일이 있으면 Excel이 열려 있다는 뜻이다. 그 상태로 스크립트를 돌리면 깨진다. `신희정_교수님/pathology_processing/`에 실제로 남아 있다.
- 결과 파일명에 날짜시각을 박아라(`result_..._20260312_012900.xlsx`). 여러 번 돌리면 어느 게 어느 설정 결과인지 알 수 없어진다.
- 검증은 반드시 read-only. 검증 스크립트가 원본을 고치면 검증의 의미가 없다.
## 연관 skill
`[[ctrial-collect]]`(회차 비교), `[[biblio-analysis]]`(서지 통합 검증), `[[pathology-llm-extract]]`·`[[medical-code-extract]]`(추출 결과 검증)
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/excel-case-validator/scripts"
cat > "$DEST/excel-case-validator/scripts/dedup.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""중복 제거와 날짜 기반 파일 세대 관리.
출처: ctrial-auto/app/utils/compare_utils.py 의 norm_title / _info_richness /
_pick_latest_by_approval / dedup_file2_by_seq / extract_date_from_filename /
find_latest_file_by_date 를 컬럼명 의존 없이 일반화한 것.
중복 제거에서 **어느 행을 남길지는 규칙이 필요하다.** 아무거나 남기면 정보가
적은 행이 살아남는다.
python dedup.py --file data.xlsx --key clinicExamSeq --latest-by 승인일 --out clean.xlsx
python dedup.py --latest-in ./excel_old
python dedup.py --self-check
"""
import argparse
import re
import sys
import unicodedata
from pathlib import Path
DATE_IN_NAME = re.compile(r"(20\d{2})[-_.]?(\d{2})[-_.]?(\d{2})")
def norm_title(s, remove_all_spaces=False):
"""비교용 제목 정규화. 공백·괄호·전각문자 차이로 같은 레코드가
'신규'로 잡히는 사고를 막는다."""
if s is None or s != s:
return ""
t = unicodedata.normalize("NFKC", str(s)).strip()
t = re.sub(r"\s+", " ", t).lower()
return t.replace(" ", "") if remove_all_spaces else t
def same_title(a, b, remove_all_spaces=True):
return norm_title(a, remove_all_spaces) == norm_title(b, remove_all_spaces)
def info_richness(row):
"""행의 정보량 — 비어있지 않은 값의 수."""
return int(row.notna().sum())
def extract_date_from_filename(name):
"""파일명에 박힌 날짜 → 'YYYYMMDD'. 없으면 None."""
m = DATE_IN_NAME.search(str(Path(name).name))
return "".join(m.groups()) if m else None
def find_latest_by_date(directory, exts=(".xlsx", ".xls", ".csv")):
"""디렉터리에서 파일명 날짜가 가장 최근인 파일. 날짜가 없는 건 수정시각으로."""
files = [p for p in Path(directory).iterdir() if p.suffix.lower() in exts and not p.name.startswith("~$")]
if not files:
return None
dated = [(extract_date_from_filename(p), p) for p in files]
with_date = [(d, p) for d, p in dated if d]
if with_date:
return max(with_date)[1]
return max(files, key=lambda p: p.stat().st_mtime)
def dedup(df, key, latest_by=None, prefer="richness"):
"""key 중복을 제거한다. → (남긴 DF, 버린 DF)
prefer:
"richness" 정보가 가장 많은 행 (기본)
"latest" latest_by 컬럼이 가장 최신인 행
"first" 그룹의 첫 행
latest_by 를 주면 "latest" 로 먼저 고르고, 값이 없거나 동률이면 prefer 로 넘어간다."""
import pandas as pd
if key not in df.columns:
return df.copy(), df.iloc[0:0].copy()
keep_idx, drop_idx = [], []
for _, g in df.groupby(key, dropna=True):
if len(g) == 1:
keep_idx.append(g.index[0]); continue
chosen = None
if latest_by and latest_by in g.columns:
d = pd.to_datetime(g[latest_by], errors="coerce")
if d.notna().any():
top = d[d == d.max()].index
chosen = top[0] if len(top) == 1 else None
if chosen is None:
sub = g.loc[top]
chosen = max(top, key=lambda i: info_richness(sub.loc[i])) if prefer != "first" else top[0]
if chosen is None:
chosen = (g.index[0] if prefer == "first"
else max(g.index, key=lambda i: info_richness(g.loc[i])))
keep_idx.append(chosen)
drop_idx += [i for i in g.index if i != chosen]
# key 가 비어 있는 행은 중복 판정에서 빼고 그대로 남긴다
keep_idx += list(df.index[df[key].isna()])
return df.loc[sorted(keep_idx)].copy(), df.loc[sorted(drop_idx)].copy()
def diff_by_key(old, new, key):
"""회차 간 변경. → dict(added, removed, common)"""
o = {str(v) for v in old[key].dropna()} if key in old.columns else set()
n = {str(v) for v in new[key].dropna()} if key in new.columns else set()
return {"added": sorted(n - o), "removed": sorted(o - n), "common": sorted(o & n)}
# ── 자체 점검 ───────────────────────────────────────────────────
def _self_check():
import pandas as pd
assert norm_title(None) == "" and norm_title(float("nan")) == ""
assert norm_title(" Ａ Ｂ ") == "a b" # 전각 → 반각
assert norm_title("A B", remove_all_spaces=True) == "ab"
assert same_title("항암제 (제1상)", "항암제(제1상)")
assert not same_title("위암 연구", "폐암 연구")
assert extract_date_from_filename("list_20260403.xlsx") == "20260403"
assert extract_date_from_filename("a_2026-04-03.xlsx") == "20260403"
assert extract_date_from_filename("noname.xlsx") is None
# 정보량 기준 — 값이 더 많은 행이 남는다
df = pd.DataFrame({"seq": [1, 1, 2], "a": ["x", None, "z"], "b": ["y", None, None]})
kept, dropped = dedup(df, "seq")
assert len(kept) == 2 and len(dropped) == 1
assert kept.loc[kept.seq == 1, "a"].iloc[0] == "x"
# 승인일 기준 — 최신 행이 남는다 (정보량이 적어도)
df2 = pd.DataFrame({"seq": [1, 1], "승인일": ["2026-01-01", "2026-05-01"],
"a": ["old", None], "b": ["old", None]})
kept, _ = dedup(df2, "seq", latest_by="승인일")
assert kept["승인일"].iloc[0] == "2026-05-01", kept
# 승인일이 같으면 정보량으로 결정
df3 = pd.DataFrame({"seq": [1, 1], "승인일": ["2026-01-01", "2026-01-01"],
"a": [None, "full"], "b": [None, "full"]})
kept, _ = dedup(df3, "seq", latest_by="승인일")
assert kept["a"].iloc[0] == "full", kept
# 승인일이 전부 비면 정보량으로
df4 = pd.DataFrame({"seq": [1, 1], "승인일": [None, None], "a": [None, "full"]})
kept, _ = dedup(df4, "seq", latest_by="승인일")
assert kept["a"].iloc[0] == "full"
# key 가 빈 행은 중복으로 안 보고 남긴다
df5 = pd.DataFrame({"seq": [None, None, 1], "a": ["p", "q", "r"]})
kept, dropped = dedup(df5, "seq")
assert len(kept) == 3 and len(dropped) == 0, (len(kept), len(dropped))
# 없는 컬럼이면 원본 그대로
kept, dropped = dedup(df5, "없는컬럼")
assert len(kept) == 3 and len(dropped) == 0
d = diff_by_key(pd.DataFrame({"k": [1, 2]}), pd.DataFrame({"k": [2, 3]}), "k")
assert d == {"added": ["3"], "removed": ["1"], "common": ["2"]}, d
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--file"); p.add_argument("--key")
p.add_argument("--latest-by", help="동률일 때 최신으로 볼 날짜 컬럼")
p.add_argument("--prefer", choices=["richness", "latest", "first"], default="richness")
p.add_argument("--out"); p.add_argument("--dropped-out")
p.add_argument("--latest-in", help="이 디렉터리에서 가장 최근 파일을 찾는다")
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
if a.latest_in:
f = find_latest_by_date(a.latest_in)
print(f or "파일 없음")
return 0 if f else 1
if not (a.file and a.key):
p.error("--file 과 --key 필요")
import pandas as pd
read = pd.read_csv if a.file.lower().endswith(".csv") else pd.read_excel
df = read(a.file)
kept, dropped = dedup(df, a.key, a.latest_by, a.prefer)
print(f"{len(df):,}행 → {len(kept):,}행 (중복 {len(dropped):,}행 제거)")
if a.out:
(kept.to_csv if a.out.lower().endswith(".csv") else kept.to_excel)(a.out, index=False)
print(f"저장: {a.out}")
if a.dropped_out and len(dropped):
(dropped.to_csv if a.dropped_out.lower().endswith(".csv") else dropped.to_excel)(a.dropped_out, index=False)
print(f"제거분 저장: {a.dropped_out}")
return 0
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/excel-case-validator/scripts/dedup.py"
mkdir -p "$DEST/excel-case-validator/scripts"
cat > "$DEST/excel-case-validator/scripts/validate.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""원본 Excel들 ↔ 통합 Excel 무결성 교차검증. Read-only.
python validate.py --sources a.xlsx b.xlsx --final merged.xlsx \
--key PMID --key DOI --required Title Year --year-col Year --year-range 1963 2026
--self-check 로 내장 테스트 실행.
"""
import argparse
import sys
import pandas as pd
def _read(path):
"""엑셀/CSV를 DataFrame으로. 시트가 여럿이면 첫 시트."""
if str(path).lower().endswith(".csv"):
return pd.read_csv(path)
return pd.read_excel(path)
def _norm(s):
"""비교용 정규화: NFKC, 연속공백 축약, 소문자. 빈값은 ''."""
import re
import unicodedata
if pd.isna(s):
return ""
t = unicodedata.normalize("NFKC", str(s)).strip()
return re.sub(r"\s+", " ", t).lower()
def check(sources, final, keys=(), required=(), year_col=None, year_range=None):
"""검증 수행 → (results, ok). results = [(항목, PASS/FAIL, 상세)]"""
results = []
def add(name, ok, detail):
results.append((name, "PASS" if ok else "FAIL", detail))
src_total = sum(len(df) for df in sources.values())
for name, df in sources.items():
add(f"원본 건수: {name}", True, f"{len(df):,}건")
add("통합 건수 ≤ 원본 합계", len(final) <= src_total,
f"통합 {len(final):,} / 원본합 {src_total:,}")
# 키 중복 — 빈값은 중복 판정에서 제외 (PMID 없는 논문이 여럿인 건 정상)
for key in keys:
if key not in final.columns:
add(f"키 컬럼 존재: {key}", False, "통합 파일에 컬럼 없음")
continue
vals = [_norm(v) for v in final[key]]
nonblank = [v for v in vals if v]
dupes = len(nonblank) - len(set(nonblank))
add(f"{key} 중복 없음", dupes == 0,
f"비어있지 않은 {len(nonblank):,}건 중 중복 {dupes}건")
# 필수 컬럼 완전성
for col in required:
if col not in final.columns:
add(f"필수 컬럼 존재: {col}", False, "컬럼 없음")
continue
filled = sum(1 for v in final[col] if _norm(v))
add(f"{col} 완전성", filled == len(final),
f"{filled:,}/{len(final):,} ({filled / len(final) * 100:.1f}%)" if len(final) else "0건")
# 원본 → 통합 누락: 키 기준으로 원본에만 있는 값 탐지
for key in keys:
if key not in final.columns:
continue
final_vals = {_norm(v) for v in final[key] if _norm(v)}
for name, df in sources.items():
if key not in df.columns:
continue
src_vals = {_norm(v) for v in df[key] if _norm(v)}
missing = src_vals - final_vals
add(f"누락 없음: {name}.{key}", not missing,
f"통합에 없는 {key} {len(missing)}건" + (f" 예: {list(missing)[:3]}" if missing else ""))
if year_col and year_col in final.columns and year_range:
lo, hi = year_range
years = pd.to_numeric(final[year_col], errors="coerce").dropna()
out = years[(years < lo) | (years > hi)]
add(f"{year_col} 범위 {lo}~{hi}", out.empty,
f"범위 밖 {len(out)}건" + (f" 예: {sorted(out.unique())[:5]}" if len(out) else ""))
return results, all(r[1] == "PASS" for r in results)
def report(results, ok):
"""마크다운 표로 출력."""
lines = ["| 항목 | 결과 | 상세 |", "|---|---|---|"]
lines += [f"| {n} | {s} | {d} |" for n, s, d in results]
lines.append("")
lines.append(f"**{'전체 PASS' if ok else 'FAIL 있음'}** — {sum(1 for r in results if r[1] == 'FAIL')}건 실패")
return "\n".join(lines)
def _self_check():
src = pd.DataFrame({"PMID": ["1", "2", "3"], "Title": ["a", "b", "c"], "Year": [2020, 2021, 2022]})
# 통합에서 PMID 3 누락 + PMID 1 중복 + Title 빈값 + 연도 범위 밖
fin = pd.DataFrame({"PMID": ["1", "1", "2", "4"], "Title": ["a", "a", "", "d"], "Year": [2020, 2020, 2021, 1800]})
results, ok = check({"src": src}, fin, keys=["PMID"], required=["Title"],
year_col="Year", year_range=(1963, 2026))
got = {n: s for n, s, _ in results}
assert not ok
assert got["PMID 중복 없음"] == "FAIL", got
assert got["Title 완전성"] == "FAIL", got
assert got["누락 없음: src.PMID"] == "FAIL", got
assert got["Year 범위 1963~2026"] == "FAIL", got
# 정상 케이스는 전부 PASS
fin2 = pd.DataFrame({"PMID": ["1", "2", "3"], "Title": ["a", "b", "c"], "Year": [2020, 2021, 2022]})
_, ok2 = check({"src": src}, fin2, keys=["PMID"], required=["Title"],
year_col="Year", year_range=(1963, 2026))
assert ok2
# 빈 키는 중복으로 세지 않는다
fin3 = pd.DataFrame({"PMID": ["", "", "1"], "Title": ["a", "b", "c"]})
r3, _ = check({}, fin3, keys=["PMID"])
assert dict((n, s) for n, s, _ in r3)["PMID 중복 없음"] == "PASS"
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--sources", nargs="*", default=[])
p.add_argument("--final")
p.add_argument("--key", action="append", default=[], help="중복·누락 검사 키. 여러 번 지정 가능")
p.add_argument("--required", nargs="*", default=[], help="비어있으면 안 되는 컬럼")
p.add_argument("--year-col")
p.add_argument("--year-range", nargs=2, type=int)
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check()
return 0
if not a.final:
p.error("--final 필요")
results, ok = check({s: _read(s) for s in a.sources}, _read(a.final),
a.key, a.required, a.year_col, tuple(a.year_range) if a.year_range else None)
print(report(results, ok))
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/excel-case-validator/scripts/validate.py"
mkdir -p "$DEST/lit-relevance-classify"
cat > "$DEST/lit-relevance-classify/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: lit-relevance-classify
description: 논문·문서 대량을 LLM으로 분류한다(주제 관련성, 저자 전문분야, 기관 국내외 등). 결과를 키 기반 캐시에 저장해 재실행 비용을 0으로 만들고, 판정 모델을 캐시에 함께 기록해 재현성을 보장하며, 모델 간 판정 일치율을 비교한다. TRIGGER - "논문 관련성 분류", "전문분야 판정", "GPT로 대량 분류", "모델 비교", "분류 캐시", 수천 건을 LLM으로 분류하고 재현·검증해야 할 때.
---
# 논문 대량 LLM 분류
## 언제 쓰나
수천 건을 LLM으로 판정해야 하고, **스크립트를 여러 번 다시 돌려야 할 때**. 분류 기준을 다듬으며 반복 실행하는 상황이 기본 전제다.
## 정본 코드
```
~/Documents/IMOK/BTC_KOL_PoC/
classify_relevance_gpt.py 논문 주제 관련성 판정
classify_specialty_gpt.py ★ 전문분야 판정 (캐시 설계가 가장 성숙, 정본)
compare_specialty_models.py 모델 간 판정 비교
specialty_resolver.py, institution_resolver.py
collect_pubmed.py / collect_openalex.py / collect_koreamed.py
merge_wos.py / merge_openalex.py / merge_koreamed.py / merge_pmc.py
make_provenance.py, quality_report.py
```
## 바로 쓰는 코드
```bash
python3 scripts/classify_cache.py --compare cache_a.json cache_b.json --field sp
python3 scripts/classify_cache.py --stats specialty_cache.json --field sp
```
`scripts/classify_cache.py` — `Cache`(모델 기록 강제, `todo()`, `drop_stale()`, `drop_model()`) · `compare()`(일치율·불일치 조합) · `call_with_param_fallback()`(신형 배포의 파라미터 거부 대응).
**모델 없이 저장하려 하면 `ValueError` 를 던진다** — 재현 불가 캐시를 만들지 못하게.
## 캐시가 설계의 중심이다
```
키(UT/PMID/DOI 등) → 판정 결과 를 JSON에 저장
재실행 시 캐시에 없는 것만 처리 → 재실행 비용 0
```
`todo = [r for r in recs if r["key"] not in cache]` 한 줄이 전부다. 이게 없으면 기준을 조금 고칠 때마다 전체를 다시 돌리게 되고, 결국 아무도 기준을 안 고치게 된다.
### 캐시에 판정 모델을 반드시 기록해라
정본 코드의 주석에 실측 교훈이 그대로 남아 있다:
> 캐시는 `specialty_cache.json` 이며 **어떤 모델이 판정했는지 함께 기록한다** — `relevance_cache.json` 은 모델을 기록하지 않아 재현 확인이 불가능했다(2026-08-05 실측).
모델을 안 남기면 나중에 "이 판정이 어느 모델 것인지" 알 수 없고, 모델을 바꿔가며 실험한 결과가 한 파일에 뒤섞인다. **캐시 엔트리에는 결과 + 모델 + (가능하면) 판정 시각을 넣어라.**
### stale 무효화 경로를 만들어라
`stale_keys(cache)`가 더 이상 유효하지 않은 엔트리를 골라 지운다(예: 분류 체계가 바뀌어 `site` 값이 현재 목록에 없는 경우). **분류 카테고리를 바꿨는데 캐시가 옛 카테고리로 답하는 게 가장 찾기 어려운 버그다.**
대조 실험은 캐시 파일을 갈아끼워서 한다: `--cache other_cache.json`.
## 모델 간 비교
`compare_specialty_models.py`가 두 캐시를 로드해 일치/불일치를 낸다. 모델을 바꾸기 전에 **기존 모델과 얼마나 다르게 판정하는지** 먼저 재라. 불일치 건만 사람이 보면 검수 대상이 확 준다.
`classify_specialty_gpt.py`와 `compare_specialty_models.py` 둘 다 `selfcheck()`를 가지고 있다. 분류 로직을 고치면 이걸 먼저 돌린다.
## 신형 모델 배포 대응
정본에 이런 처리가 있다:
```python
# ponytail: 신형 배포는 max_tokens/temperature 를 거부한다. 첫 호출에서 걸리면 빼고 재시도.
kw = {"temperature": 0, "max_tokens": 300}
...
if "temperature" in msg and "unsupported" in msg.lower():
kw.pop("temperature", None)
```
**모델 파라미터를 하드코딩하고 예외 처리를 안 하면 배포 교체 때 전체가 죽는다.** 거부당하면 그 파라미터를 빼고 한 번 더 시도하는 게 가장 싼 대응이다.
## 병렬 처리
`run(recs, workers=8)` — 8 워커 병렬. 캐시 쓰기는 워커 완료 시점에 메인에서 모아 한다. 여러 워커가 캐시 파일에 동시에 쓰면 깨진다.
## 입력 본문 구성
`build_body(rec)`가 제목·초록·저널·MeSH 등에서 판정에 필요한 것만 뽑아 프롬프트 본문을 만든다. `max_tokens=300`으로 출력을 제한한다 — 분류는 짧게 답해야 하고, 길면 파싱이 흔들린다.
`load_exclude()`로 제외 목록을 별도 관리한다. 사람이 "이건 아니다"라고 한 건을 코드가 아니라 데이터로 둔다.
## 다중 소스 수집·병합도 여기 있다
`collect_*.py`(PubMed/OpenAlex/KoreaMed) → `merge_*.py`(WoS/OpenAlex/KoreaMed/PMC) → `make_provenance.py`로 **각 필드가 어느 소스에서 왔는지 출처를 남긴다.** `fill_doi.py`/`recover_doi.py`로 빠진 DOI를 메운다.
통합 구현 자체는 `[[biblio-analysis]]`가 더 성숙하다. 이쪽은 소스가 더 많다(OpenAlex, PMC 추가).
## 연관 skill
`[[biblio-analysis]]`, `[[cancer-type-classify]]`(같은 대량 LLM 분류 + 피드백 패턴), `paper-collect`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/lit-relevance-classify/scripts"
cat > "$DEST/lit-relevance-classify/scripts/classify_cache.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""대량 LLM 분류용 캐시와 모델 간 비교.
출처: BTC_KOL_PoC/classify_specialty_gpt.py 의 캐시·재시도 로직과
compare_specialty_models.py 의 비교 로직을 프로젝트 의존 없이 추출.
캐시가 설계의 중심이다. 없으면 기준을 조금 고칠 때마다 전체를 다시 돌리게 되고,
결국 아무도 기준을 안 고치게 된다.
원본에 남은 실측 교훈 — **캐시에 판정 모델을 반드시 기록한다.**
relevance_cache.json 은 모델을 안 남겨 재현 확인이 불가능했다(2026-08-05).
python classify_cache.py --compare a.json b.json --field sp
python classify_cache.py --stats cache.json --field sp
python classify_cache.py --self-check
"""
import argparse
import collections
import json
import sys
import time
from pathlib import Path
class Cache:
"""키 → 판정 결과. 모델과 판정 시각을 함께 남긴다."""
def __init__(self, path):
self.path = Path(path)
self.data = json.loads(self.path.read_text(encoding="utf-8")) if self.path.exists() else {}
def save(self):
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(json.dumps(self.data, ensure_ascii=False, indent=1), encoding="utf-8")
def put(self, key, value, model, stamp=None):
"""판정 결과 저장. model 은 필수 — 빼면 나중에 재현을 못 한다."""
if not model:
raise ValueError("model 은 필수입니다. 어느 모델이 판정했는지 없으면 재현 불가.")
self.data[key] = {**value, "model": model,
"at": stamp or time.strftime("%Y-%m-%d %H:%M:%S")}
def todo(self, keys):
"""아직 판정 안 된 키만. 이 한 줄이 재실행 비용을 0으로 만든다."""
return [k for k in keys if k not in self.data]
def drop_stale(self, valid, field):
"""분류 체계가 바뀌어 더는 유효하지 않은 엔트리를 지운다. → 지운 키 목록
카테고리를 바꿨는데 캐시가 옛 카테고리로 답하는 게 가장 찾기 어려운 버그다."""
valid = set(valid)
stale = [k for k, v in self.data.items() if v.get(field) not in valid]
for k in stale:
del self.data[k]
return stale
def drop_model(self, model):
"""특정 모델이 판정한 것만 지운다. 모델 교체 후 재판정용. → 지운 키 목록"""
stale = [k for k, v in self.data.items() if v.get("model") == model]
for k in stale:
del self.data[k]
return stale
def stats(self, field):
"""판정 분포와 모델 분포."""
return {
"총": len(self.data),
"판정": dict(collections.Counter(v.get(field) for v in self.data.values())),
"모델": dict(collections.Counter(v.get("model") for v in self.data.values())),
}
def compare(a, b, field="sp"):
"""두 캐시의 판정 일치·불일치. 모델을 바꾸기 전에 얼마나 다른지 먼저 재라."""
both = sorted(set(a) & set(b))
same = [k for k in both if a[k].get(field) == b[k].get(field)]
diff = [k for k in both if a[k].get(field) != b[k].get(field)]
return {
"a_only": sorted(set(a) - set(b)),
"b_only": sorted(set(b) - set(a)),
"both": both, "same": same, "diff": diff,
"agreement": round(len(same) / len(both) * 100, 1) if both else None,
"pairs": collections.Counter((a[k].get(field), b[k].get(field)) for k in diff),
}
def call_with_param_fallback(fn, kwargs, max_attempts=8):
"""모델 파라미터를 거부당하면 빼고 재시도, 429 면 지수 백오프.
신형 배포는 max_tokens / temperature 를 거부한다. 하드코딩하고 예외 처리를
안 하면 배포 교체 때 전체가 죽는다.
fn(**kwargs) 를 호출하고 결과를 그대로 돌려준다."""
kw = dict(kwargs)
last = None
for attempt in range(max_attempts):
try:
return fn(**kw)
except Exception as e:
last, msg = e, str(e)
if "max_tokens" in msg and "max_completion_tokens" in msg:
if "max_tokens" in kw:
kw["max_completion_tokens"] = kw.pop("max_tokens")
continue
if "temperature" in msg and "unsupported" in msg.lower() and "temperature" in kw:
kw.pop("temperature")
continue
if attempt == max_attempts - 1:
break
time.sleep(min(60, 5 * 2 ** attempt) if "429" in msg else 2 * (attempt + 1))
raise last
# ── 자체 점검 ───────────────────────────────────────────────────
def _self_check():
import tempfile
with tempfile.TemporaryDirectory() as d:
c = Cache(Path(d) / "c.json")
c.put("X", {"sp": "내과"}, model="gpt-4.1", stamp="2026-01-01 00:00:00")
c.put("Y", {"sp": "외과"}, model="gpt-4.1", stamp="2026-01-01 00:00:00")
assert c.todo(["X", "Y", "Z"]) == ["Z"]
assert c.data["X"]["model"] == "gpt-4.1" and c.data["X"]["at"].startswith("2026")
# 모델 없이 저장하려 하면 막는다
try:
c.put("Z", {"sp": "내과"}, model="")
raise AssertionError("model 없이 저장이 통과됨")
except ValueError:
pass
c.save()
assert Cache(Path(d) / "c.json").data["Y"]["sp"] == "외과" # 왕복 보존
# 분류 체계가 {내과} 로 좁혀지면 외과 엔트리는 무효
assert c.drop_stale({"내과"}, "sp") == ["Y"]
assert "Y" not in c.data and "X" in c.data
c.put("W", {"sp": "내과"}, model="gpt-5.3", stamp="2026-02-01 00:00:00")
assert c.drop_model("gpt-4.1") == ["X"]
assert list(c.data) == ["W"]
s = c.stats("sp")
assert s == {"총": 1, "판정": {"내과": 1}, "모델": {"gpt-5.3": 1}}, s
a = {"X": {"sp": "내과"}, "Y": {"sp": "외과"}, "Z": {"sp": "병리과"}}
b = {"X": {"sp": "소화기내과"}, "Y": {"sp": "외과"}, "W": {"sp": "영상의학과"}}
r = compare(a, b)
assert r["same"] == ["Y"] and r["diff"] == ["X"], r
assert r["a_only"] == ["Z"] and r["b_only"] == ["W"]
assert r["agreement"] == 50.0, r
assert r["pairs"][("내과", "소화기내과")] == 1
assert compare({}, {})["both"] == [] and compare({}, {})["agreement"] is None
# 파라미터 거부 → 빼고 재시도
seen = []
def fake(**kw):
seen.append(dict(kw))
if "temperature" in kw:
raise RuntimeError("temperature is unsupported with this model")
if "max_tokens" in kw:
raise RuntimeError("use max_completion_tokens instead of max_tokens")
return "ok"
assert call_with_param_fallback(fake, {"temperature": 0, "max_tokens": 260}) == "ok"
assert len(seen) == 3 and "temperature" not in seen[-1] and "max_completion_tokens" in seen[-1], seen
# 회복 불가한 오류는 그대로 올린다
try:
call_with_param_fallback(lambda **k: (_ for _ in ()).throw(ValueError("boom")), {}, max_attempts=1)
raise AssertionError("예외가 전파되지 않음")
except ValueError:
pass
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--compare", nargs=2, metavar=("A", "B"))
p.add_argument("--stats", metavar="CACHE")
p.add_argument("--field", default="sp", help="판정값이 담긴 키 이름")
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
if a.stats:
print(json.dumps(Cache(a.stats).stats(a.field), ensure_ascii=False, indent=2))
return 0
if a.compare:
load = lambda f: json.loads(Path(f).read_text(encoding="utf-8"))
r = compare(load(a.compare[0]), load(a.compare[1]), a.field)
print(f"공통 {len(r['both'])}건 · 일치 {len(r['same'])}건 · 불일치 {len(r['diff'])}건 "
f"· 일치율 {r['agreement']}%")
print(f"A에만 {len(r['a_only'])}건 · B에만 {len(r['b_only'])}건\n")
if r["pairs"]:
print("불일치 조합 (A → B):")
for (x, y), n in r["pairs"].most_common(20):
print(f" {x} → {y}: {n}건")
return 0
p.error("--compare 또는 --stats 필요")
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/lit-relevance-classify/scripts/classify_cache.py"
mkdir -p "$DEST/llm-provider-switch"
cat > "$DEST/llm-provider-switch/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: llm-provider-switch
description: 온프렘 Ollama ↔ OpenAI ↔ Azure OpenAI를 설정만 바꿔 전환하고 장애 시 자동 폴백하는 LLM 제공자 계층을 설계·구현한다. 다중 Ollama 서버 로드밸런싱, health check, thinking 모델(Qwen3 등) JSON 파싱, Azure content filter 대응을 포함. TRIGGER - 병원 온프렘 환경에서 개발하다 클라우드로 검증하려 할 때, LLM provider를 바꾸거나 추가할 때, "Ollama", "Azure OpenAI", "provider 전환", "폴백", "모델 서버 여러 대", "thinking 태그", "JSON 파싱 실패"가 나올 때.
---
# LLM Provider 전환 계층
## 언제 쓰나
병원 프로젝트는 거의 항상 이 왕복을 한다: **온프렘 Ollama에서 개발 → Azure/OpenAI로 성능 검증 → 다시 온프렘 배포**. 매번 새로 짜면 프로젝트마다 provider 코드가 갈라진다. 실제로 `CRC-TNM-Agents` 7벌의 `llm_manager.py`가 서로 다르게 갈라져 있다.
## 정본 코드
```
~/Documents/IMOK/CRC-TNM-Agents_crc/src/llm/llm_providers/
base_llm_provider.py 311줄 추상 인터페이스 + 예외 체계 + 재시도
llm_manager.py 572줄 provider 선택·폴백·health check
ollama_provider.py 783줄 다중서버 로드밸런싱 + thinking 대응
openai_provider.py 464줄
azure_provider.py 505줄
```
`_crc`가 이 계열에서 가장 최근에 손댄 버전(2026-07-14)이고 `ollama_http_client.py`를 유일하게 가지고 있다. 새 프로젝트는 **이 디렉토리를 복사해서 시작**한다. `src.setting.config`에 의존하므로 import 경로만 고치면 된다.
## 바로 쓰는 코드
```bash
# 추론 모델 응답에서 JSON 건지기 (4단 폴백)
echo '{"category":"T3"} 설명' | python3 scripts/thinking_json.py
```
`scripts/thinking_json.py` — `ollama_provider.py` 의 thinking 처리 4종을 provider 의존 없이 뺀 것.
`extract_thinking` · `remove_thinking_tags` · `extract_json`(json.loads 검증 포함) · `answer_to_json`.
`parse(content, value_pattern)` 이 본문 → thinking → 값 변환 순으로 물러나며 시도하고, 어느 단계에서 나왔는지 함께 돌려준다.
## 구조
3계층으로 나눈다. 이 경계를 지켜야 provider 추가가 한 파일 추가로 끝난다.
```
LLMManager 어느 provider를 쓸지 결정. 폴백·health check·통계
└ BaseLLMProvider 추상 인터페이스 (ABC)
├ OllamaProvider 온프렘
├ OpenAIProvider 클라우드
└ AzureOpenAIProvider 병원 계약 클라우드
```
`BaseLLMProvider`가 강제하는 5개 추상 메서드 — 새 provider를 붙일 때 이것만 구현하면 된다:
| 메서드 | 역할 |
|---|---|
| `initialize()` | 클라이언트 생성, 모델 확인 |
| `generate_response(LLMRequest) -> LLMResponse` | 단발 호출 |
| `generate_stream(LLMRequest)` | 스트리밍 |
| `validate_connection() -> bool` | health check용 |
| `get_available_models() -> List[str]` | 모델 목록 |
`LLMRequest` / `LLMResponse` 데이터클래스로 provider별 응답 형식을 흡수한다. **호출부는 provider를 절대 몰라야 한다** — 이게 깨지면 전환 계층의 의미가 없어진다.
## 폴백 동작
```
primary provider 시도
→ 실패하면 fallback_providers 순회
→ 성공한 provider로 응답 + 어느 provider가 응답했는지 메타데이터 기록
→ 전부 실패하면 LLMManagerError
```
메타데이터 기록(`_record_llm_metadata`)이 중요하다. **어느 응답이 폴백으로 나온 것인지 남기지 않으면 나중에 결과 품질 차이를 설명할 수 없다.** 의료 데이터에서 "이 케이스는 Ollama가 아니라 Azure가 답했다"는 재현성의 핵심 정보다.
health check는 5분 간격(`_health_check_interval = 300`)으로 lazy 실행 — 매 호출마다 하면 느리고, 안 하면 죽은 서버로 계속 보낸다.
## 함정
### 1. thinking 모델이 JSON을 thinking 안에 넣는다
Qwen3 등 reasoning 모델은 `...` 안에 정답 JSON을 넣고 본문에는 산문을 쓴다. 그대로 파싱하면 전부 실패한다. `ollama_provider.py`의 4단 대응을 그대로 가져다 쓴다:
```
_extract_thinking() 내용 추출
_remove_thinking_tags() 본문에서 태그 제거
_extract_json_from_thinking() thinking 안의 ```json 블록 또는 { } 추출 + json.loads 유효성 검증
_convert_thinking_to_json() "Final Answer: T2b", **T2b**, \boxed{T2b} 같은 텍스트 응답을 JSON으로 변환
```
**추출한 JSON은 반드시 `json.loads`로 검증하고, 실패하면 다음 패턴으로 넘어간다.** 정규식 매치만 믿으면 깨진 JSON이 통과한다.
### 2. Azure content filter가 의료 텍스트를 막는다
병리·수술 기록에 나오는 표현이 Azure content filter에 걸린다. `should_use_azure_safe_mode()`로 우회 모드를 켠다 — Azure provider이면서 `content_filter_enabled` 환경변수가 True일 때만 활성화. **환경변수로 끄고 켤 수 있어야 한다.** 코드에 하드코딩하면 다른 병원 계약에서 못 쓴다.
### 3. Ollama 서버 여러 대
`_GlobalServerIndexManager` 싱글턴이 round_robin으로 서버를 배분한다. 프로세스 전역이라 워커가 여러 개여도 한쪽 서버로 쏠리지 않는다. 서버별 healthy/unhealthy 마킹(`_mark_server_unhealthy`)으로 죽은 서버를 건너뛴다.
첫 호출 전에 `_warmup_models()`로 모델을 올려둔다. 안 하면 첫 케이스만 수십 초 걸려서 벤치마크가 왜곡된다.
### 4. 예외를 provider별로 뭉개지 마라
`base_llm_provider.py`가 정의하는 예외 체계를 유지한다 — `LLMConnectionError`(폴백 대상), `LLMAuthenticationError`(폴백해도 소용없음), `LLMRateLimitError`(재시도 대상), `LLMTimeoutError`, `LLMModelNotFoundError`. **전부 `Exception`으로 잡으면 인증 오류에도 폴백을 시도하며 시간을 낭비한다.**
## 연관 skill
`[[tnm-staging-extract]]`, `[[pathology-llm-extract]]`, `[[medical-code-extract]]`, `[[cancer-type-classify]]` — 모두 이 계층 위에서 동작한다.
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/llm-provider-switch/scripts"
cat > "$DEST/llm-provider-switch/scripts/thinking_json.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""추론(thinking) 모델의 응답에서 JSON을 건져낸다.
출처: CRC-TNM-Agents_crc/src/llm/llm_providers/ollama_provider.py 의
_extract_thinking / _remove_thinking_tags / _extract_json_from_thinking /
_convert_thinking_to_json 을 provider 의존 없이 추출한 것.
Qwen3 같은 추론 모델은 정답 JSON을 ... 안에 넣고 본문에는
산문을 쓴다. json.loads 한 번으로 끝내려 하면 전부 실패한다.
echo '{"a":1} 설명' | python thinking_json.py
python thinking_json.py --self-check
"""
import argparse
import json
import re
import sys
THINK_RE = re.compile(r"(.*?)", re.DOTALL)
# "Final Answer: T2b", **T2b**, \boxed{T2b} 처럼 값만 뱉은 응답을 건지는 패턴
ANSWER_PATTERNS = (
r"\\boxed\{([^}]+)\}",
r"\*\*Final Answer:\*\*\s*\*\*([^*]+)\*\*",
r"Final Answer:\s*\*\*([^*]+)\*\*",
r"\*\*Answer:\*\*\s*\*\*([^*]+)\*\*",
r"Answer:\s*\*\*([^*]+)\*\*",
r"Final Answer:\s*([^\n*]+)",
r"(?:category|classification)\s*(?:is|:)\s*\*\*([^*]+)\*\*",
)
def extract_thinking(content):
""" 안의 내용."""
m = THINK_RE.search(content or "")
return m.group(1).strip() if m else ""
def remove_thinking_tags(content):
"""본문에서 블록을 걷어낸다."""
return THINK_RE.sub("", content or "").strip()
def extract_json(text):
"""텍스트에서 유효한 JSON 문자열만 골라낸다. 없으면 ''.
정규식 매치만 믿으면 깨진 JSON이 통과하므로 **반드시 json.loads 로 검증**하고
실패하면 다음 패턴으로 넘어간다."""
if not text:
return ""
# 1) ```json ... ``` 블록
m = re.search(r"```json\s*([\s\S]*?)\s*```", text)
if m:
s = m.group(1).strip()
try:
json.loads(s)
return s
except json.JSONDecodeError:
pass
# 2) 이름 없는 ``` 블록
m = re.search(r"```\s*([\s\S]*?)\s*```", text)
if m:
s = m.group(1).strip()
try:
json.loads(s)
return s
except json.JSONDecodeError:
pass
# 3) 중괄호로 감싸인 덩어리 — 가장 바깥부터 좁혀가며 시도
starts = [i for i, c in enumerate(text) if c == "{"]
ends = [i for i, c in enumerate(text) if c == "}"]
for a in starts:
for b in reversed(ends):
if b <= a:
continue
s = text[a:b + 1].strip()
try:
json.loads(s)
return s
except json.JSONDecodeError:
continue
return ""
def answer_to_json(thinking, value_pattern=None, confidence=0.7):
"""값만 뱉은 추론 응답을 JSON 으로 되살린다. 없으면 ''.
value_pattern: 도메인 값의 정규식. 예: TNM 이면 r"\\b(T[0-4][a-c]?|N[0-3]|M[01][a-c]?)\\b" """
if not thinking:
return ""
value = None
for pat in ANSWER_PATTERNS:
m = re.search(pat, thinking, re.IGNORECASE)
if m:
value = m.group(1).strip().strip("*").strip()
break
if not value and value_pattern:
m = re.search(value_pattern, thinking)
if m:
value = m.group(1)
if not value:
return ""
return json.dumps({
"category": value,
"reasoning": thinking[:500].replace('"', "'").replace("\n", " "),
"confidence_score": confidence,
"confidence_reasoning": "Extracted from thinking mode response",
"key_findings": [],
"evidence_text": "Response converted from thinking mode",
}, ensure_ascii=False)
def parse(content, value_pattern=None):
"""추론 모델 응답 → dict. 4단계로 물러나며 시도하고 전부 실패하면 None.
→ (결과, 어느 단계에서 나왔는지)"""
thinking = extract_thinking(content)
body = remove_thinking_tags(content)
s = extract_json(body)
if s:
return json.loads(s), "body"
s = extract_json(thinking)
if s:
return json.loads(s), "thinking"
s = answer_to_json(thinking or body, value_pattern)
if s:
return json.loads(s), "converted"
return None, "failed"
# ── 자체 점검 ───────────────────────────────────────────────────
TNM = r"\b(T[0-4][a-c]?|N[0-3]|M[01][a-c]?)\b"
def _self_check():
assert extract_thinking(" 고민 답") == "고민"
assert extract_thinking("태그 없음") == ""
assert remove_thinking_tags("x 답 ") == "답"
# 본문에 정상 JSON
r, how = parse('생각```json\n{"category":"T2"}\n```')
assert (r, how) == ({"category": "T2"}, "body"), (r, how)
# 정답 JSON이 thinking 안에 들어간 전형적 실패 사례
r, how = parse('정리하면 {"category": "T3a", "confidence_score": 0.9} 이다'
"따라서 T3a 로 판단됩니다.")
assert how == "thinking" and r["category"] == "T3a", (r, how)
# 값만 뱉은 경우
r, how = parse("여러모로 보아 **Final Answer:** **T2b**", TNM)
assert how == "converted" and r["category"] == "T2b", (r, how)
r, how = parse(r"계산 결과 \boxed{N1} 이다", TNM)
assert how == "converted" and r["category"] == "N1", (r, how)
# 정해진 문구가 없어도 도메인 패턴으로 건진다
r, how = parse("침윤 깊이로 보아 T4a 에 해당한다", TNM)
assert how == "converted" and r["category"] == "T4a", (r, how)
# 깨진 JSON 은 통과시키지 않는다
assert extract_json('```json\n{"a": }\n```') == ""
assert extract_json("{ 이건 JSON 이 아니다 }") == ""
r, how = parse("모르겠다판단 불가")
assert r is None and how == "failed", (r, how)
# 뒤에 잡소리가 붙어도 JSON 만 건진다
assert json.loads(extract_json('설명 {"a": 1} 끝')) == {"a": 1}
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--value-pattern", help="도메인 값 정규식 (예: TNM)")
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
result, how = parse(sys.stdin.read(), a.value_pattern)
if result is None:
print("파싱 실패", file=sys.stderr)
return 1
print(json.dumps(result, ensure_ascii=False, indent=2))
print(f"(출처: {how})", file=sys.stderr)
return 0
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/llm-provider-switch/scripts/thinking_json.py"
mkdir -p "$DEST/medical-code-extract"
cat > "$DEST/medical-code-extract/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: medical-code-extract
description: 검사 판독문(경동맥·관상동맥 초음파/CT, PWV·ABI 등)의 결론 텍스트에서 질병코드와 수치를 추출한다. 규칙 기반 키워드 추출로 힌트를 만들어 LLM에 주입하고, LLM JSON 파싱이 실패하면 키워드 결과로 폴백하는 하이브리드. 부정문·중증도 처리 포함. TRIGGER - "질병코드 추출", "판독문에서 코드 뽑기", "경동맥/관상동맥/PWV/ABI 판정", "키워드+LLM 하이브리드", 검사 결론문을 코드로 자동 변환해야 할 때.
---
# 검사 판독문 → 질병코드 추출
## 언제 쓰나
건강검진·영상 검사의 **결론(conclusion) 텍스트**에서 질병코드나 수치 판정을 뽑을 때. 코드 체계가 정해져 있고(유효 코드 집합이 있음) 오답을 내면 안 되는 상황.
`[[pathology-llm-extract]]`가 자유 필드를 뽑는다면, 이쪽은 **닫힌 코드 집합**으로 분류한다 — 그래서 규칙 기반 폴백이 성립한다.
## 정본 코드
```
~/Documents/medical_code_extractor 2/
keyword_extractor.py 규칙 기반 추출 (문장 단위 정밀 분석)
llm_extractor.py LLM + 힌트 주입 + 폴백
coronary_extractor.py 관상동맥 전용 (Keyword/LLM 쌍 + 팩토리)
pwv_abi_extractor.py PWV·ABI 전용
keywords/ carotid_keywords.py, coronary_keywords.py, pwv_abi_keywords.py
prompts/ 검사별 프롬프트 빌더
ollama_client.py 온프렘 LLM
excel_handler.py 입출력
api_server.py FastAPI
~/Documents/건강의학과/ 검증 케이스 (120/140/199/206건 xlsx)
```
`건강의학과/medical_code_extractor`에도 사본이 있다. **`~/Documents/medical_code_extractor 2/`가 정본이다.**
## 핵심: 규칙이 힌트를 만들고 LLM이 판단한다
```
1) KeywordExtractor.extract(text) → KeywordHints
2) format_hints_for_llm(hints) → 힌트 텍스트
3) prompt_builder(text, hints_text) → system/user 프롬프트
4) LLM 호출 → JSON 파싱
5) _postprocess_codes(codes, hints) → 유효 코드 검증
6) JSON 파싱 3회 실패 → KeywordExtractor 결과로 폴백
```
**규칙 결과를 LLM에 넘기되 최종 판단은 LLM이 한다.** 규칙만 쓰면 표현 변형에 약하고, LLM만 쓰면 근거 없이 코드를 만든다. 그리고 **규칙 결과가 살아 있으므로 LLM이 무너져도 답을 낼 수 있다** — 이게 이 구조의 진짜 가치다.
## 규칙 추출기가 잡아야 하는 것
`VesselFinding`이 담는 필드가 곧 규칙 추출기의 요구사항이다:
```
vessel 어느 혈관/부위
severity 중증도 표현
severity_level 비교 가능한 숫자 등급
stenosis_percent 협착률 수치
context 근거 문장
negated ★ 부정문 여부
```
`negated`가 가장 중요하다. **"no significant stenosis"를 협착으로 잡으면 전부 틀린다.** 의료 판독문은 정상 소견을 부정문으로 쓴다. 그래서 문서 전체가 아니라 **문장 단위로** 쪼개 분석한다.
`vessel_max_severity: Dict[str, VesselFinding]`로 혈관별 최고 중증도만 남긴다. 같은 혈관이 여러 문장에 나오면 가장 심한 것이 판정 기준이다.
## 폴백 정책
```python
max_json_retries = 3 # JSON 파싱 실패 재시도
```
3회 실패하면 `KeywordExtractor.get_disease_codes(hints)`로 폴백하고, `reasoning`에 `"KeywordExtractor fallback (JSON 파싱 3회 실패)"`를 남긴다. 폴백 코드도 없으면 `success=False`.
**폴백으로 나온 결과임을 반드시 표시해라.** 나중에 정확도를 볼 때 LLM 결과와 폴백 결과를 섞어 계산하면 안 된다.
`json_mode=False`로 호출하고 직접 파싱한다(`_parse_response_strict`). 온프렘 모델은 json_mode를 제대로 지원하지 않는 경우가 많다. `max_tokens=8192` — 4096에서 올렸다. reasoning 모델은 thinking에 토큰을 많이 쓴다.
## 검사 종류 추가하는 법
각 검사가 동일한 3종 세트를 가진다:
```
keywords/<검사>_keywords.py 위치·중증도 키워드 사전
prompts/<검사>_prompt.py 프롬프트 빌더
<검사>_extractor.py KeywordExtractor + LLMExtractor + create_<검사>_extractor() 팩토리
```
`create_carotid_extractor(ollama_client)` 같은 팩토리로 조립한다. **키워드 사전과 프롬프트를 데이터로 분리했기 때문에 새 검사 추가가 파일 3개 추가로 끝난다.**
## 함정
- `valid_codes: Set[int]`로 유효 코드 집합을 강제한다. LLM이 존재하지 않는 코드를 만들어내면 `_postprocess_codes`에서 걸린다. **닫힌 코드 체계면 반드시 사후 검증해라.**
- 결론 텍스트만 넣는다. 판독문 전체를 넣으면 소견 부분의 언급이 결론과 충돌한다.
- `thinking` 필드를 따로 보관한다 — 온프렘 reasoning 모델 대응은 `[[llm-provider-switch]]` 참조.
- 검증 케이스가 `건강의학과/`에 xlsx로 있다(경동맥 120·170건, 관상동맥 140건, 199·206건). 규칙을 고치면 여기로 회귀 검증해라.
## 연관 skill
`[[pathology-llm-extract]]`, `[[llm-provider-switch]]`, `[[excel-case-validator]]`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/recurrence-detect"
cat > "$DEST/recurrence-detect/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: recurrence-detect
description: 환자별 의무기록 타임라인에서 암 재발 여부·재발일·재발 부위(local/distant)를 LLM으로 판정하고, Ground Truth 대비 정확도를 평가한다. 홀드아웃 평가, self-consistency 다수결, 날짜 ±N일 허용·부위 Jaccard 부분점수 채점, 불일치 사례 분석, 프롬프트 BEFORE/AFTER 스냅샷. TRIGGER - "재발 판정", "recurrence", "재발일 추출", "GT 정확도", "모델 비교", "홀드아웃 평가", "다수결/self-consistency", LLM 추출 정확도를 측정·개선해야 할 때.
---
# 재발 판정 + LLM 정확도 평가
## 언제 쓰나
여러 시점의 의무기록을 시간순으로 읽어 **사건(재발)이 언제 어디서 일어났는지** 판정할 때. 그리고 더 넓게는 — **LLM 추출 정확도를 제대로 측정하는 방법**이 필요할 때. 평가 설계 부분은 재발이 아닌 어떤 추출 과제에도 그대로 쓴다.
## 정본 코드
```
~/Documents/IMOK/LungCancerRecurrence/
run_batch_analysis.py 배치 실행
analyze_gt_accuracy.py ★ GT 대비 정확도 (정본)
analyze_mismatches.py / analyze_all_mismatches.py 불일치 분석
analyze_patient_detail.py 개별 환자 추적
scripts_tmp_eval_full.py, scripts_tmp_modelcompare.py, scripts_tmp_parallel_run.py
model_comparison_260617.md ★ 모델 비교 결과 기록
backup_prompts_260526/, backup_prompts_260611/ 프롬프트 BEFORE/AFTER 스냅샷
docker-compose.yml + app.db(SQLite)
```
## 바로 쓰는 코드
```bash
python3 scripts/score.py --gt gt.json --pred pred.json --spec spec.json
```
`scripts/score.py` — 네 개 파일에 재구현돼 있던 채점 코드를 합친 것.
`jaccard`(둘 다 비면 채점 제외) · `date_diff_days` · `Scorer`(지표별 n 분리 누적) · `majority`(self-consistency 다수결) · `report(min_n)`(표본 부족 경고).
재발 전용이 아니라 **어떤 LLM 추출 채점에도** 쓴다.
## 판정 구조
```
환자별 기록 타임라인
→ 상위 분석 (기록에서 재발 관련 사실 추출)
→ final_decision 재발 여부 + 재발일 + 부위 코드
```
상위 분석과 최종 판단을 나눈다. **모델을 바꿔 비교할 때 상위 분석은 재사용하고 최종 판단만 다시 돌린다** — 비교 비용이 크게 준다(`model_comparison_260617.md`의 "동일 상위분석 재사용").
## 채점을 제대로 설계해라 — 이 skill의 핵심
정확도를 "맞다/틀리다"로만 재면 개선이 안 보인다. 이 프로젝트가 쓰는 4지표:
| 지표 | 이유 |
|---|---|
| 날짜 정확일치 | 엄격 기준 |
| **날짜 ±30일** | 재발일은 확진일·영상일·기록일이 며칠씩 다르다. 완전일치만 재면 실제 성능이 안 보인다 |
| Local 부위 Jaccard | 부위는 여러 개일 수 있다 → 부분점수 |
| Distant 부위 Jaccard | 위와 동일 |
**부분점수(Jaccard)와 허용오차(±30일)가 있어야 프롬프트 개선이 지표에 나타난다.** 완전일치만 보면 "3개 중 2개 맞춤"이 0점이라 개선 방향을 못 잡는다.
비교 전에 정규화가 필요하다 — `normalize_llm_location()`이 LLM이 뱉은 부위 표현을 GT 코드 체계로 매핑한다. **정규화 없이 문자열 비교하면 실제보다 훨씬 낮게 나온다.**
`parse_gt_sites(gt_distant_site, gt_local_site, gt_mixed_site)` — GT가 여러 컬럼에 흩어져 있으면 파싱을 한 군데로 모아라.
## 홀드아웃 + 표본 수를 같이 적어라
```
홀드아웃 40명, 프로젝트 19
날짜 n=38~39 / Distant n=32~37 / Local n=3~7 ← Local은 신뢰도 낮음
```
**`n`을 지표 옆에 반드시 적는다.** `model_comparison_260617.md`는 "Local 92.9%"에 `n=7`을 붙이고 본문에 "신뢰도 낮음"을 명시했다. n을 안 적으면 표본 7건짜리 92.9%가 결론을 뒤집는다.
## 모델 비교에서 배운 것
| 방식 | 날짜 정확 | Distant(J) |
|---|---|---|
| gpt-oss:20b 단일 | 31.6% | 73.8% |
| gpt-oss:20b **다수결 3회** | 34.2% | 76.3% |
| llama3.3:70b 단일 | **41.0%** | 51.9% |
- **self-consistency(같은 입력 3회 호출 후 다수결)**: +2~3%p 일관 개선. 비용 3배지만 저위험.
- **단일 최강 모델은 없다.** 날짜는 llama가, 부위는 gpt-oss가 낫다 → 항목별 하이브리드가 현실적.
- 실패도 기록해라: `qwen3.5:35b-a3b`는 서버에서 사실상 무응답(간단 호출도 12초+ 타임아웃), 72분에 10명도 처리 못 해 제외. **왜 제외했는지 남기지 않으면 다음 사람이 또 시도한다.**
## 프롬프트 변경은 스냅샷으로 관리
`backup_prompts_YYYYMMDD/` 폴더에 변경 대상 파일의 `*_BEFORE.*` / `*_AFTER.*`를 함께 넣고 README에 원본 경로와 롤백 명령을 적는다:
```
final_decision_BEFORE.py → src/final_decision.py
ground_truth_api_BEFORE.py → src/backend/controllers/ground_truth_api.py
```
프롬프트를 고치면 지표가 오르내린다. **되돌릴 수 있어야 실험을 과감하게 한다.** git이 있어도 "이 지표는 어느 프롬프트에서 나온 건지"를 한 폴더로 묶어두는 게 훨씬 빠르다.
## 불일치를 봐야 개선된다
`analyze_mismatches.py`가 LLM 예측(`extract_llm_prediction`)과 분석 근거(`extract_analysis_details`)를 뽑아 GT와 다른 케이스를 정리한다. `analyze_patient_detail.py`로 개별 환자를 파고든다.
**틀린 케이스를 안 읽고 프롬프트를 고치는 건 추측이다.** 지표는 어디가 틀렸는지 안 알려준다.
## 함정
- 오래 걸리는 배치는 재시도 스크립트를 따로 둔다(`scripts_tmp_rerun_errors.py`). 전체 재실행은 최후 수단.
- 병렬 실행(`scripts_tmp_parallel_run.py`) 시 온프렘 서버 부하를 확인해라 — 위 qwen 사례가 그 결과다.
- `app.db-shm`/`app.db-wal`이 남아 있다. SQLite WAL 모드이므로 DB를 복사할 때 세 파일을 함께 옮겨야 한다.
## 연관 skill
`[[tnm-staging-extract]]`·`[[pathology-llm-extract]]`(추출 대상), `[[excel-case-validator]]`, `[[llm-provider-switch]]`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/recurrence-detect/scripts"
cat > "$DEST/recurrence-detect/scripts/score.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""LLM 추출 결과를 Ground Truth와 대조해 채점한다.
출처: LungCancerRecurrence 의 scripts_tmp_eval_full.py / _modelcompare.py /
_selfconsist_all.py / _validate_prompt.py 에 네 번 재구현돼 있던 채점 코드를 합친 것.
재발 판정 전용이 아니다. 값·날짜·집합·자유텍스트를 GT와 비교하는 어떤 추출 과제에도 쓴다.
핵심 규칙 두 가지:
1) GT나 예측이 없는 항목은 채점에서 뺀다. 0점으로 세면 정확도가 왜곡된다.
2) 지표마다 n(비교한 건수)을 따로 세서 함께 보고한다. n 없는 정확도는 못 믿는다.
python score.py --gt gt.json --pred pred.json --spec spec.json
python score.py --self-check
"""
import argparse
import json
import sys
from collections import defaultdict
from datetime import datetime
DATE_FORMATS = ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d", "%Y%m%d")
# ── 비교 함수 ───────────────────────────────────────────────────
def parse_date(s):
"""여러 표기를 datetime 으로. 실패하면 None."""
if not s:
return None
t = str(s).strip()[:10]
for fmt in DATE_FORMATS:
try:
return datetime.strptime(t, fmt)
except ValueError:
continue
return None
def jaccard(a, b):
"""집합 부분점수. 둘 다 비면 None(채점 제외), 한쪽만 비면 0.0."""
a, b = set(a or ()), set(b or ())
if not a and not b:
return None
if not a or not b:
return 0.0
return len(a & b) / len(a | b)
def date_diff_days(gt, pred):
"""|GT - 예측| 일수. 한쪽이라도 못 읽으면 None."""
g, p = parse_date(gt), parse_date(pred)
if not g or not p:
return None
return abs((p - g).days)
def yn(v):
"""Y/N·1/0·True/False 를 bool 로. 판단 불가면 None."""
if v is None or v == "":
return None
if isinstance(v, bool):
return v
t = str(v).strip().lower()
if t in ("y", "yes", "1", "1.0", "true", "t", "예", "있음"):
return True
if t in ("n", "no", "0", "0.0", "false", "f", "아니오", "없음"):
return False
return None
def text_match(g, p):
"""자유텍스트 일치. 한쪽이라도 비면 None."""
if not g or not p:
return None
return str(g).strip().lower() == str(p).strip().lower()
def majority(values):
"""self-consistency 다수결. 같은 입력을 여러 번 돌린 결과를 하나로.
LungCancerRecurrence 실측: 3회 다수결이 단일 호출 대비 날짜·부위 +2~3%p."""
vals = [v for v in values if v is not None]
if not vals:
return None
counts = defaultdict(int)
for v in vals:
counts[json.dumps(v, sort_keys=True, ensure_ascii=False)] += 1
best = max(counts.items(), key=lambda kv: kv[1])[0]
return json.loads(best)
# ── 채점기 ──────────────────────────────────────────────────────
class Scorer:
"""지표별로 점수합과 n을 따로 누적한다."""
def __init__(self, date_tolerances=(0, 14, 30)):
self.sum = defaultdict(float)
self.n = defaultdict(int)
self.tol = tuple(date_tolerances)
def bool_field(self, name, gt, pred):
g, p = yn(gt), yn(pred)
if g is None or p is None:
return
self.n[name] += 1
self.sum[name] += 1.0 if g == p else 0.0
def date_field(self, name, gt, pred):
"""정확일치와 허용오차별 점수를 한꺼번에. n은 하나로 공유한다."""
d = date_diff_days(gt, pred)
if d is None:
return
self.n[name] += 1
for t in self.tol:
key = f"{name}(정확)" if t == 0 else f"{name}(±{t}일)"
self.sum[key] += 1.0 if d <= t else 0.0
def set_field(self, name, gt, pred):
j = jaccard(gt, pred)
if j is None:
return
self.n[name] += 1
self.sum[name] += j
def text_field(self, name, gt, pred):
m = text_match(gt, pred)
if m is None:
return
self.n[name] += 1
self.sum[name] += 1.0 if m else 0.0
def rows(self):
"""[(지표, 백분율, n)] — 날짜 파생 지표는 원 지표의 n을 쓴다."""
out = []
for key in sorted(self.sum):
base = key.split("(")[0]
n = self.n.get(key) or self.n.get(base) or 0
if n:
out.append((key, self.sum[key] / n * 100, n))
return out
def report(self, title="채점 결과", min_n=10):
lines = [f"## {title}", "", "| 지표 | 정확도 | n |", "|---|---|---|"]
warn = []
for key, pct, n in self.rows():
flag = " ⚠" if n < min_n else ""
lines.append(f"| {key} | {pct:.1f}% | {n}{flag} |")
if n < min_n:
warn.append(key)
if warn:
lines += ["", f"⚠ 표본 {min_n}건 미만이라 신뢰도가 낮음: {', '.join(warn)}"]
return "\n".join(lines)
def score_records(gt_list, pred_list, spec, key="id"):
"""레코드 목록 두 개를 spec 대로 채점.
spec = {"bool": [...], "date": [...], "set": [...], "text": [...]}"""
preds = {r.get(key): r for r in pred_list}
s = Scorer()
for g in gt_list:
p = preds.get(g.get(key))
if p is None:
continue
for f in spec.get("bool", []):
s.bool_field(f, g.get(f), p.get(f))
for f in spec.get("date", []):
s.date_field(f, g.get(f), p.get(f))
for f in spec.get("set", []):
s.set_field(f, g.get(f), p.get(f))
for f in spec.get("text", []):
s.text_field(f, g.get(f), p.get(f))
return s
# ── 자체 점검 ───────────────────────────────────────────────────
def _self_check():
assert jaccard([], []) is None # 둘 다 없음 → 채점 제외
assert jaccard(["a"], []) == 0.0 # 한쪽만 없음 → 0점
assert jaccard(["a", "b"], ["a", "b"]) == 1.0
assert abs(jaccard(["a", "b"], ["a", "c"]) - 1/3) < 1e-9
assert date_diff_days("2026-01-01", "2026-01-15") == 14
assert date_diff_days("2026.01.01", "2026/01/01") == 0
assert date_diff_days("", "2026-01-01") is None
assert date_diff_days("2026-01-01", "언제인지 모름") is None
assert yn("Y") is True and yn("0") is False and yn("") is None and yn("아마도") is None
assert text_match("A", " a ") is True and text_match("", "a") is None
assert majority(["x", "x", "y"]) == "x"
assert majority([{"a": 1}, {"a": 1}, {"a": 2}]) == {"a": 1}
assert majority([None, None]) is None
gt = [{"id": 1, "recur": "Y", "date": "2026-01-01", "site": ["Lung"], "note": "A"},
{"id": 2, "recur": "N", "date": "", "site": [], "note": ""}]
pr = [{"id": 1, "recur": "Y", "date": "2026-01-20", "site": ["Lung", "Bone"], "note": "a"},
{"id": 2, "recur": "Y", "date": "2026-02-02", "site": ["Liver"], "note": "b"}]
s = score_records(gt, pr, {"bool": ["recur"], "date": ["date"], "set": ["site"], "text": ["note"]})
r = dict((k, (round(v, 1), n)) for k, v, n in s.rows())
assert r["recur"] == (50.0, 2), r # 1건 맞고 1건 틀림
assert r["date(정확)"] == (0.0, 1), r # 2번은 GT 날짜가 없어 제외 → n=1
assert r["date(±14일)"] == (0.0, 1), r # 19일 차이라 ±14 밖
assert r["date(±30일)"] == (100.0, 1), r # ±30 안
# 1번은 Jaccard 1/2, 2번은 GT가 빈 집합인데 모델이 부위를 답했으므로 0점으로 센다
assert r["site"] == (25.0, 2), r
assert r["note"] == (100.0, 1), r # 2번은 GT 없어 제외
assert "⚠" in s.report(min_n=10) # n 작으면 경고
assert "⚠" not in s.report(min_n=1)
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--gt"); p.add_argument("--pred")
p.add_argument("--spec", help='{"bool":[...],"date":[...],"set":[...],"text":[...]} JSON 파일')
p.add_argument("--key", default="id")
p.add_argument("--min-n", type=int, default=10, help="이보다 적으면 신뢰도 경고")
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
if not (a.gt and a.pred and a.spec):
p.error("--gt --pred --spec 필요")
load = lambda f: json.load(open(f, encoding="utf-8"))
s = score_records(load(a.gt), load(a.pred), load(a.spec), a.key)
print(s.report(min_n=a.min_n))
return 0
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/recurrence-detect/scripts/score.py"
mkdir -p "$DEST/regimen-extract"
cat > "$DEST/regimen-extract/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: regimen-extract
description: 항암 요법(레지멘) DB를 구축한다. HIRA 공고책자·허가초과(허초) 심의결과·Onco Regimen book PDF를 Vision LLM으로 추출하고, 식약처 API(허가상세·대조약)와 약가 엑셀을 병합해 암종별 요법·용법용량·급여기준·약가를 SQLite/D1로 낸다. 출처 추적, 사람 검토·교정 반영 워크플로 포함. TRIGGER - "항암제 DB", "레지멘 추출", "허가초과/허초", "급여기준", "공고책자", "regimen book", "약가", PDF 표를 구조화 DB로 만들어야 할 때.
---
# 항암 레지멘 DB 구축
## 언제 쓰나
"담도암 환자에게 쓸 수 있는 요법이 뭔가"를 답하는 DB를 만들 때. `[[ctrial-collect]]` + `[[cancer-type-classify]]`가 **임상시험** 쪽을 맡는다면, 이쪽은 **이미 승인·급여되는 치료**를 맡는다.
넓게는 **PDF 표를 Vision LLM으로 구조화 DB에 넣는 방법**의 정본이다.
## 정본 코드
```
~/Documents/IMOK/허초프로젝트/
extraction/ 추출 산출물 + 캐시 + 교정 파일 (JSON)
regbook_vision/, geupyeo_vision/ Vision 추출
regbook_pages/ 페이지별 결과
*_cache.json API 응답 캐시
*_corrections.json, review_state.json, verify_findings.json 검토·교정
scripts/ crop_*, extract_*, enrich_*, apply_*, build_*, export_*
db_design/
데이터_출처_정리.md ★ 테이블·컬럼별 출처를 전부 기록한 문서
schema_v2.sql, d1_import.sql, d1_overlay.sql
항암제_DB설계서_v2.xlsx, verify_report.md
web/, wrangler.toml Cloudflare D1 + Workers 배포
~/Documents/IMOK/oncoalert/ 같은 DB를 쓰는 웹 서비스
```
## 바로 쓰는 코드
```bash
python3 scripts/corrections.py --records out.json --corrections fix.json --key id --out final.json
```
`scripts/corrections.py` — `apply_*.py` 들이 공유하던 패턴을 하나로. `apply_corrections`(멱등, before/after 기록, 교정 표시) · `split_review`(확인완료/필요/미검토) · orphan 경고.
**교정 대상이 사라진 키를 반드시 보고한다** — 재추출로 레코드가 없어지면 사람 검토가 소리 없이 증발한다.
## 소스가 7개다 — 출처 문서를 먼저 써라
| 코드 | 소스 | 형태 | 무엇을 줌 |
|---|---|---|---|
| S1 | 약제급여목록·급여상한금액표 (HIRA) | 엑셀 | 제품목록·EDI·주성분코드·**약가** |
| S2 | 사전신청요법·불승인요법 (HIRA 허초 심의결과) | 엑셀 | 허가초과 요법(인정/검토중/불승인) |
| S3 | 항암제보험급여 공고책자 (HIRA, 연 1회) | **PDF** | 급여 요법·고시번호·투여대상·투여단계 |
| S4 | Onco Regimen book (아산병원 종양내과) | **PDF** | 표준요법·용법용량·투여일·참고논문 |
| S5 | 식약처 제품허가정보 API | API | 영문명·약효분류·전문일반·EDI |
| S6 | 식약처 허가상세 API | API | **ATC코드**·효능효과·용법용량·허가변경이력 |
| S7 | 식약처 대조약 API | API | 오리지널 여부 |
`db_design/데이터_출처_정리.md`가 **테이블·컬럼별로 어느 소스에서 왔는지** 전부 적어둔다. 소스가 여럿이면 이 문서 없이는 몇 달 뒤 아무도 값의 근거를 못 찾는다.
두 가지가 특히 중요하다:
- **AI 수기 입력 항목을 따로 표시한다.** `cancer_aliases`(NSCLC 같은 영문 약어)는 "⚠️ 자료 아님 — 약제부 검토 예정"으로 명시돼 있다. 근거 자료가 있는 값과 AI가 채운 값을 섞으면 신뢰도가 통째로 무너진다.
- **제거한 소스도 이유와 함께 남긴다.** "약가기준정보 API(심평원)는 제거됨 — 약가는 S1 엑셀에서 받음."
## PDF → 구조화: crop 후 Vision
```
crop_regbook_cards.py / crop_geupyeo_criteria.py 페이지에서 카드·표 영역을 잘라냄
→ regbook_vision/, geupyeo_vision/ Vision LLM으로 추출
→ regbook_pages/ 페이지별 JSON
→ regbook_schema_v3_samples.json 스키마 정착
```
**페이지 전체를 통째로 Vision에 넣지 마라.** 요법 카드 단위로 잘라 넣어야 정확하다. crop 결과는 사람이 검토하고(`crop_review_decisions.json`), `apply_crop_review.py`로 반영한다.
암종 단위로 나눠 산출한다(`regbook_esophageal.json`, `regbook_gastric_TEST.json`, `regbook_page005_braintumor.json`). 한 암종부터 스키마를 확정하고 나머지로 넓힌다.
## API 응답은 캐시한다
`drug_detail_cache.json`, `drug_permit_cache.json`, `recollect_cache.json`. 식약처 API는 느리고 호출 제한이 있으며, **enrich 스크립트를 여러 번 돌리게 되기 때문에** 캐시가 없으면 작업이 불가능하다. `enrich_from_cache.py`가 캐시만으로 재구성한다.
## 사람 교정을 데이터로 관리한다
이 프로젝트에서 가장 배울 점이다. 교정을 코드에 넣지 않고 **JSON 파일 + apply 스크립트**로 분리했다:
| 파일 | apply 스크립트 |
|---|---|
| `regimenbook_dose_corrections.json` | `apply_dose_corrections.py` |
| `crop_review_decisions.json` | `apply_crop_review.py` |
| `needs_confirm.json` | `apply_needs_confirm.py` |
| `_review_flags_cache.json` | `apply_review_flags.py` / `apply_flag_dispositions.py` |
| `verify_corrections.json`, `orphan_corrections.json` | `apply_review.py` |
| `verified_ok.json`, `verify_findings.json`, `review_state.json` | 검토 상태 추적 |
**추출을 다시 돌려도 사람이 고친 내용이 살아남는다.** 추출 → 검토 → 교정 JSON → 재적용이 반복 가능한 파이프라인이 된다. 추출 결과를 직접 수정하면 다음 재추출에서 전부 날아간다.
`needs_confirm.json`(확인 필요)과 `verified_ok.json`(확인 완료)을 나눠 검토 진행 상황을 추적한다.
## 빌드·배포
```
build_db.py / build_sqlite.py / build_drugcost.py SQLite 생성
enrich_db.py / enrich_detail.py / enrich_permit.py API로 보강
export_d1.py → d1_import.sql + d1_overlay.sql Cloudflare D1
export_web.py → web/ 정적 웹
```
`d1_overlay.sql`을 `d1_import.sql`과 분리한 게 요령이다 — 전체 재적재 없이 변경분만 덮어쓴다.
## 프로젝트 문서를 Obsidian으로 관리한다
`CLAUDE.md`가 Obsidian 볼트(`~/Documents/obsidian/dnlife/허초프로젝트/`)의 6개 문서를 규정한다: `00_Overview`(고정 정보), `01_Plan`, `02_Progress`(날짜별), `03_Decisions`(기술 결정), `04_Issues`(미해결), `05_Completed`(검증 완료만).
규칙이 좋다 — **덮어쓰지 않고 날짜별로 추가**, **실제 완료한 것만 기록**, **확인되지 않은 내용은 추측하지 않음**. 긴 프로젝트에는 이 방식을 그대로 쓸 만하다. `obsidian-mcp-connector` MCP로 접근한다.
## 함정
- `oncoalert.db-shm`/`-wal`이 있다. SQLite WAL — 복사 시 세 파일을 함께.
- 공고책자는 연 1회 갱신된다. 고시번호가 바뀌면 요법 매칭이 깨진다. `relink_report.json`이 재연결 결과를 남긴다.
- `_experiments/`, `_archive/`에 시도한 것들이 남아 있다.
## 연관 skill
`[[ctrial-collect]]`·`[[cancer-type-classify]]`(담도암 치료옵션 조합), `[[excel-case-validator]]`
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/regimen-extract/scripts"
cat > "$DEST/regimen-extract/scripts/corrections.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""사람이 고친 내용을 별도 파일로 두고 재추출 결과에 다시 얹는다.
출처: 허초프로젝트/scripts 의 apply_dose_corrections / apply_crop_review /
apply_needs_confirm / apply_review_flags 가 공유하는 패턴을 하나로 정리.
추출 결과를 직접 수정하면 다음 재추출에서 전부 날아간다.
교정을 코드가 아니라 **데이터(JSON)** 로 두면 추출 → 검토 → 교정 → 재적용이
반복 가능한 파이프라인이 된다.
python corrections.py --records out.json --corrections fix.json --key id --out final.json
python corrections.py --records out.json --corrections fix.json --key id --report
python corrections.py --self-check
"""
import argparse
import json
import sys
from pathlib import Path
# 검토 상태
NEEDS_CONFIRM = "needs_confirm" # 확인 필요
VERIFIED_OK = "verified_ok" # 확인 완료
CORRECTED = "corrected" # 사람이 고침
def load(path, default=None):
p = Path(path)
return json.loads(p.read_text(encoding="utf-8")) if p.exists() else (default if default is not None else {})
def save(path, obj):
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
def apply_corrections(records, corrections, key="id", track=True):
"""교정을 레코드에 얹는다. → (결과 레코드, 리포트)
corrections: {키: {필드: 새값}} 또는 {키: {"fields": {...}, "reason": "...", "by": "..."}}
교정 대상이 사라진 키(orphan)는 반드시 보고한다 — 재추출로 레코드가 없어졌는데
조용히 넘어가면 사람이 한 검토가 소리 없이 증발한다."""
index = {str(r.get(key)): r for r in records}
applied, orphans, unchanged = [], [], []
out = [dict(r) for r in records]
out_index = {str(r.get(key)): r for r in out}
for k, spec in corrections.items():
k = str(k)
if k not in index:
orphans.append(k)
continue
fields = spec.get("fields") if isinstance(spec, dict) and "fields" in spec else spec
if not isinstance(fields, dict):
orphans.append(k)
continue
rec = out_index[k]
changed = {}
for f, v in fields.items():
if rec.get(f) != v:
changed[f] = {"before": rec.get(f), "after": v}
rec[f] = v
if not changed:
unchanged.append(k)
continue
if track:
rec["_review"] = CORRECTED
rec["_corrected_fields"] = sorted(changed)
if isinstance(spec, dict) and spec.get("reason"):
rec["_correction_reason"] = spec["reason"]
applied.append({"key": k, "changed": changed})
return out, {
"applied": applied,
"orphans": sorted(orphans),
"unchanged": sorted(unchanged),
"total_records": len(records),
"total_corrections": len(corrections),
}
def split_review(records, verified_keys=(), confirm_keys=(), key="id"):
"""검토 진행 상황을 나눈다. → dict(verified, needs_confirm, untouched)
확인 완료와 확인 필요를 나눠 관리해야 어디까지 봤는지 추적된다."""
v, c = {str(x) for x in verified_keys}, {str(x) for x in confirm_keys}
buckets = {"verified": [], "needs_confirm": [], "untouched": []}
for r in records:
k = str(r.get(key))
buckets["verified" if k in v else "needs_confirm" if k in c else "untouched"].append(k)
return buckets
def report_text(rep):
lines = [f"레코드 {rep['total_records']:,}건 · 교정 {rep['total_corrections']:,}건",
f" 적용 {len(rep['applied'])}건",
f" 변화 없음 {len(rep['unchanged'])}건"]
if rep["orphans"]:
lines.append(f" ⚠ 대상 없음 {len(rep['orphans'])}건 — {', '.join(rep['orphans'][:10])}")
lines.append(" 재추출로 레코드가 사라졌는지 확인하세요. 사람 검토가 유실됩니다.")
for a in rep["applied"][:20]:
for f, d in a["changed"].items():
lines.append(f" {a['key']}.{f}: {d['before']!r} → {d['after']!r}")
if len(rep["applied"]) > 20:
lines.append(f" … 외 {len(rep['applied']) - 20}건")
return "\n".join(lines)
# ── 자체 점검 ───────────────────────────────────────────────────
def _self_check():
recs = [{"id": "R1", "dose": "100mg", "cycle": 21},
{"id": "R2", "dose": "50mg", "cycle": 14},
{"id": "R3", "dose": "75mg", "cycle": 28}]
# 단순 형식 + 상세 형식을 함께 받는다
fixes = {
"R1": {"dose": "120mg"},
"R2": {"fields": {"cycle": 21}, "reason": "공고책자 오탈자", "by": "약제부"},
"R3": {"dose": "75mg"}, # 이미 같은 값 → 변화 없음
"R9": {"dose": "10mg"}, # 대상 없음
}
out, rep = apply_corrections(recs, fixes, key="id")
assert out[0]["dose"] == "120mg" and out[0]["_review"] == CORRECTED
assert out[0]["_corrected_fields"] == ["dose"]
assert out[1]["cycle"] == 21 and out[1]["_correction_reason"] == "공고책자 오탈자"
assert "_review" not in out[2], out[2] # 변화 없으면 표시도 안 남긴다
assert rep["orphans"] == ["R9"], rep
assert rep["unchanged"] == ["R3"], rep
assert len(rep["applied"]) == 2
# 원본은 건드리지 않는다
assert recs[0]["dose"] == "100mg", recs[0]
assert "_review" not in recs[0]
# before/after 가 남는다
ch = rep["applied"][0]["changed"]["dose"]
assert ch == {"before": "100mg", "after": "120mg"}, ch
# 재적용해도 같은 결과 (멱등)
out2, rep2 = apply_corrections(out, fixes, key="id")
assert out2[0]["dose"] == "120mg"
assert len(rep2["applied"]) == 0 and rep2["unchanged"] == ["R1", "R2", "R3"], rep2
# track=False 면 표시를 안 붙인다
out3, _ = apply_corrections(recs, {"R1": {"dose": "120mg"}}, key="id", track=False)
assert out3[0]["dose"] == "120mg" and "_review" not in out3[0]
# 형식이 깨진 교정은 orphan 으로
_, rep4 = apply_corrections(recs, {"R1": "120mg"}, key="id")
assert rep4["orphans"] == ["R1"], rep4
b = split_review(recs, verified_keys=["R1"], confirm_keys=["R2"])
assert b == {"verified": ["R1"], "needs_confirm": ["R2"], "untouched": ["R3"]}, b
assert "⚠" in report_text(rep) and "R9" in report_text(rep)
import tempfile
with tempfile.TemporaryDirectory() as d:
f = Path(d) / "sub" / "c.json"
save(f, fixes)
assert load(f)["R1"] == {"dose": "120mg"}
assert load(Path(d) / "없음.json", default={"x": 1}) == {"x": 1}
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--records"); p.add_argument("--corrections")
p.add_argument("--key", default="id"); p.add_argument("--out")
p.add_argument("--report", action="store_true")
p.add_argument("--self-check", action="store_true")
a = p.parse_args()
if a.self_check:
_self_check(); return 0
if not (a.records and a.corrections):
p.error("--records 와 --corrections 필요")
out, rep = apply_corrections(load(a.records, []), load(a.corrections), a.key)
if a.out:
save(a.out, out)
print(f"저장: {a.out}")
if a.report or not a.out:
print(report_text(rep))
return 1 if rep["orphans"] else 0
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/regimen-extract/scripts/corrections.py"
mkdir -p "$DEST/tnm-staging-extract"
cat > "$DEST/tnm-staging-extract/SKILL.md" <<'SKILL_PAYLOAD_EOF'
---
name: tnm-staging-extract
description: 병리·영상 판독문에서 암 병기(TNM, cTNM/pTNM, AJCC stage)를 LLM 에이전트로 추출한다. T/N/M을 정보추출-분류 2단으로 쪼갠 에이전트 분할, 규칙 기반 힌트 주입, 근거(evidence) 동반 출력, LangGraph 워크플로, 재시도·보류(deferred) 정책. TRIGGER - "TNM", "병기 추출", "stage 판정", "cT/cN/pT/pN", "AJCC", 판독문에서 병기를 뽑거나 병기 판정 정확도를 올려야 할 때.
---
# TNM 병기 추출
## 언제 쓰나
병리 보고서·영상 판독문 → `pT3N1bM0, Stage IIIB` 같은 병기. 대장암(CRC)·폐암 구현이 있고, 다른 암종으로 확장할 때 이 구조를 그대로 쓴다.
## 정본 코드
```
~/Documents/IMOK/CRC-TNM-Agents/src/
llm/tnm_agents/ 에이전트 30여 개 (공통 + crc/ + lung/)
llm/tnm_agent_factory.py AGENT_TYPE_MAP + 캐시 + 평가유형→에이전트 라우팅
llm/prompt_composer.py YAML 프롬프트 조립
pipeline/ langgraph_tnm_workflow, orchestrator, retry_policy, result_merger
config/workflows/ 워크플로 정의 (YAML)
~/Documents/IMOK/NSCLC-ModularStageLLM/ 폐암 모듈러 버전 (논문용, CITATION.cff 있음)
~/Documents/IMOK/LungCancerTNM/
```
**같은 코드가 7벌 존재한다**(`CRC-TNM-Agents`, `_1`, `_crc`, `-251217`, `crc-tnm-app`, `crc-tnm-registry-closed-hotfix`, `@GMP/crc-tnm-registry-closed-main`). 기능 전체는 `IMOK/CRC-TNM-Agents`(2026-03-12, 174 py)가 가장 넓고, LLM provider 계층만은 `_crc`(2026-07-14, 176 py)가 최신이다. **어느 쪽을 정본으로 삼을지는 아직 미정**이다 — `[[llm-provider-switch]]` 참조.
## 바로 쓰는 코드
```bash
# 추출 결과에서 근거 없는 필드 찾기
python3 scripts/evidence.py --check-json result.json
```
`scripts/evidence.py` — `base_evidence_agent.py` 의 `EvidenceField` / `EvidenceExtractionResult`.
평면·중첩 형식 양방향 변환, `` 하이라이트 검증(`is_grounded`), `ungrounded()` · `low_confidence()` 로 검수 대상 선별, `defer()` 로 판정 불가 표시.
## 핵심: 에이전트를 잘게 쪼갠다
한 프롬프트로 "이 판독문의 TNM을 말해줘"라고 하면 정확도가 안 나온다. **T/N/M별로 나누고, 각각을 다시 정보추출 → 분류로 쪼갠다.**
```
pathological_t_information_agent 판독문에서 T 관련 사실만 추출 (침윤 깊이, 장막 침범…)
pathological_t_characteristics_agent 종양 특성
pathological_t_margins_agent 절제연
↓
pathological_t_classification_agent 위 사실들로 pT 값만 판정
```
N·M도 같은 구조(`pathological_n_information` → `pathological_n_classification`, `metastasis_information` → `metastasis_classification`). 마지막에 `final_stage_agent`가 T·N·M을 합쳐 AJCC stage를 낸다.
**이렇게 나누는 이유**: 사실 추출과 규칙 적용을 한 번에 시키면 LLM이 근거 없이 병기를 지어낸다. 나누면 어느 단계에서 틀렸는지 짚을 수 있고, 분류 단계는 AJCC 규칙만 보므로 프롬프트가 짧아진다.
암종별 차이는 `crc/`, `lung/` 하위 디렉토리로 분리한다. 공통 골격은 최상위에 두고 암종 특이 에이전트만 추가한다(예: CRC의 `lymph_node_count_agent`, `discrepancy_analyzer_agent`).
## 근거를 반드시 같이 받는다
`base_evidence_agent.py`의 `EvidenceField`가 모든 추출 필드에 강제하는 4종 세트:
```
value 추출값
evidence 출처 문장. 해당 부분을 ...로 감쌈
reasoning 왜 그렇게 판단했는지
confidence 확신도
```
`` 하이라이트가 중요하다 — 검수자가 원문 어디를 보고 판단했는지 바로 확인한다. **근거 없는 병기는 임상에서 못 쓴다.** 출력은 평면(`pT_evidence`)과 중첩(`{"value":…, "evidence":…}`) 둘 다 지원한다(`to_dict` / `to_nested_dict`).
## 규칙 기반 힌트 주입
`llm/tnm_agents/rule_based_hints/`와 `registry_v2_hint_generator.py`가 판독문에서 정규식·키워드로 확실한 것을 먼저 뽑아 프롬프트에 힌트로 넣는다. LLM이 처음부터 맨몸으로 읽는 것보다 정확하다. **규칙으로 확실한 건 규칙으로, 애매한 것만 LLM에게.**
## 프롬프트는 YAML 조립
`PromptComposer`가 공통 모듈 + 에이전트별 프롬프트를 YAML에서 읽어 시스템 프롬프트를 조립한다(`config/prompts/`). 캐시하고 `force_reload`로 갱신. 30여 개 에이전트가 AJCC 정의 같은 공통 블록을 공유하므로, 프롬프트를 코드에 박으면 규칙 하나 고칠 때 30군데를 고쳐야 한다.
## 워크플로·재시도·보류
`pipeline/`이 LangGraph로 에이전트 실행 순서를 만든다. `dependency_resolver`가 의존 관계를 풀고 `workflow_builder`가 그래프를 세운다. 워크플로 정의는 `config/workflows/` YAML — 코드 수정 없이 에이전트 조합을 바꾼다.
`RetryPolicy`(기본 `max_retries=2`)의 판단 기준이 중요하다:
- `has_llm_fallback()` — 결과가 LLM 폴백으로 나왔는지 감지
- `should_retry()` — 재시도할 가치가 있는지
- `should_abort()` — 포기
- `DeferredOutputBuilder` — **판정 불가를 "보류"로 명시 출력**
마지막이 핵심이다. 억지로 병기를 내놓는 것보다 "이 케이스는 판정 불가"가 안전하다. 의료 추출에서 **모른다고 말할 수 있는 경로를 반드시 만들어라.**
## 함정
- `type_detector.py`가 문서 종류(병리/영상/수술기록)를 먼저 판별한다. 이걸 건너뛰면 영상 판독문에 병리 에이전트를 돌리게 된다.
- `imaging_checker_agent`는 영상 소견이 병리와 어긋날 때 잡아낸다. CRC의 `discrepancy_analyzer_agent`도 같은 역할.
- 에이전트는 `TNMAgentFactory`가 `{agent_type}_{id(llm_manager)}` 키로 캐시한다. 테스트에서 매번 새 인스턴스가 필요하면 캐시를 인지해라.
- `isolate_context` 옵션 — 에이전트 간 컨텍스트 오염을 막는다. 정확도가 흔들리면 여기부터 본다.
## 연관 skill
`[[llm-provider-switch]]`, `[[pathology-llm-extract]]`(같은 판독문 입력), `[[excel-case-validator]]`(케이스 정확도 검증)
SKILL_PAYLOAD_EOF
mkdir -p "$DEST/tnm-staging-extract/scripts"
cat > "$DEST/tnm-staging-extract/scripts/evidence.py" <<'SKILL_PAYLOAD_EOF'
#!/usr/bin/env python3
"""추출 필드에 근거를 강제하는 컨테이너.
출처: CRC-TNM-Agents/src/llm/tnm_agents/base_evidence_agent.py 의
EvidenceField / EvidenceExtractionResult 를 프로젝트 의존 없이 추출.
근거 없는 병기는 임상에서 못 쓴다. 모든 추출 필드가 값과 함께
출처 문장·판단 이유·확신도를 들고 다니게 강제한다.
evidence 안의 ... 하이라이트가 중요하다 — 검수자가 원문 어디를 보고
판단했는지 바로 확인한다.
python evidence.py --self-check
"""
import argparse
import json
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
@dataclass
class EvidenceField:
"""값 하나 + 그 근거.
value 추출값
evidence 출처 문장. 해당 부분을 ... 로 감쌈
reasoning 왜 그렇게 판단했는지
confidence 0.0 ~ 1.0
"""
value: Optional[Any] = None
evidence: Optional[str] = None
reasoning: Optional[str] = None
confidence: float = 0.0
def to_dict(self, field_name):
"""평면 형식 — 엑셀 컬럼으로 바로 펼칠 때."""
return {
field_name: self.value,
f"{field_name}_evidence": self.evidence,
f"{field_name}_reasoning": self.reasoning,
f"{field_name}_confidence": self.confidence,
}
def to_nested_dict(self):
"""중첩 형식 — JSON 으로 주고받을 때."""
return {"value": self.value, "evidence": self.evidence,
"reasoning": self.reasoning, "confidence": self.confidence}
@classmethod
def from_dict(cls, data, field_name):
"""평면·중첩 둘 다 읽는다. LLM 응답 형식이 흔들려도 받아낸다."""
v = data.get(field_name)
if isinstance(v, dict):
return cls(
value=v.get("value"),
evidence=v.get("evidence") or data.get(f"{field_name}_evidence"),
reasoning=v.get("reasoning") or data.get(f"{field_name}_reasoning"),
confidence=v.get("confidence", data.get(f"{field_name}_confidence", 0.0)),
)
return cls(
value=v,
evidence=data.get(f"{field_name}_evidence"),
reasoning=data.get(f"{field_name}_reasoning"),
confidence=data.get(f"{field_name}_confidence", 0.0),
)
def highlighted_source(self):
"""evidence 에서 태그를 걷어낸 원문."""
return re.sub(r"", "", self.evidence or "")
def highlighted_part(self):
""" 로 표시된 부분만."""
return " ".join(re.findall(r"(.*?)", self.evidence or "", re.DOTALL)).strip()
def is_grounded(self):
"""근거가 실제로 원문을 가리키는지. 하이라이트가 없으면 근거로 치지 않는다."""
return bool(self.highlighted_part())
@dataclass
class EvidenceExtractionResult:
"""필드 여러 개의 추출 결과."""
fields: Dict[str, EvidenceField] = field(default_factory=dict)
deferred: bool = False # 판정 불가로 보류했는지
deferred_reason: str = ""
def get_field(self, name):
return self.fields.get(name)
def set_field(self, name, ef):
self.fields[name] = ef
def defer(self, reason):
"""억지로 값을 내놓는 것보다 '판정 불가'가 안전하다.
의료 추출에서는 모른다고 말할 수 있는 경로를 반드시 만들어라."""
self.deferred = True
self.deferred_reason = reason
def to_flat(self):
out = {}
for name, ef in self.fields.items():
out.update(ef.to_dict(name))
if self.deferred:
out["_deferred"] = True
out["_deferred_reason"] = self.deferred_reason
return out
def to_nested(self):
out = {n: ef.to_nested_dict() for n, ef in self.fields.items()}
if self.deferred:
out["_deferred"] = {"value": True, "evidence": None,
"reasoning": self.deferred_reason, "confidence": 1.0}
return out
def ungrounded(self):
"""근거 하이라이트가 없는 필드 이름. 검수 대상."""
return [n for n, ef in self.fields.items() if ef.value is not None and not ef.is_grounded()]
def low_confidence(self, threshold=0.7):
"""확신도가 낮은 필드 이름. 사람이 먼저 볼 순서."""
return [n for n, ef in self.fields.items() if ef.confidence < threshold]
@classmethod
def from_dict(cls, data, field_names):
r = cls()
for n in field_names:
r.set_field(n, EvidenceField.from_dict(data, n))
if data.get("_deferred"):
r.defer(str(data.get("_deferred_reason", "")))
return r
# ── 자체 점검 ───────────────────────────────────────────────────
def _self_check():
ef = EvidenceField(value="pT3", confidence=0.9,
evidence="Tumor invades through the muscularis propria into pericolic tissue.",
reasoning="고유근층 관통 소견")
assert ef.to_dict("pT")["pT_confidence"] == 0.9
assert ef.to_dict("pT")["pT"] == "pT3"
assert ef.to_nested_dict()["value"] == "pT3"
assert ef.highlighted_part() == "through the muscularis propria"
assert "" not in ef.highlighted_source()
assert ef.is_grounded()
# 하이라이트 없는 근거는 근거로 안 친다
assert not EvidenceField(value="pT3", evidence="어딘가에 그렇게 써 있음").is_grounded()
assert not EvidenceField(value="pT3").is_grounded()
# 평면 형식 왕복
flat = {"pT": "pT3", "pT_evidence": "x", "pT_reasoning": "r", "pT_confidence": 0.8}
back = EvidenceField.from_dict(flat, "pT")
assert (back.value, back.confidence, back.reasoning) == ("pT3", 0.8, "r")
# 중첩 형식도 같은 결과
nested = {"pT": {"value": "pT3", "evidence": "x", "reasoning": "r", "confidence": 0.8}}
assert EvidenceField.from_dict(nested, "pT").to_nested_dict() == back.to_nested_dict()
# 중첩에 없는 값은 평면 키로 보충한다
mixed = {"pT": {"value": "pT3"}, "pT_evidence": "y", "pT_confidence": 0.5}
m = EvidenceField.from_dict(mixed, "pT")
assert m.evidence == "y" and m.confidence == 0.5
# 키가 아예 없어도 죽지 않는다
miss = EvidenceField.from_dict({}, "pN")
assert miss.value is None and miss.confidence == 0.0
r = EvidenceExtractionResult()
r.set_field("pT", ef)
r.set_field("pN", EvidenceField(value="pN1", evidence="근거 하이라이트 없음", confidence=0.4))
assert r.ungrounded() == ["pN"], r.ungrounded()
assert r.low_confidence(0.7) == ["pN"], r.low_confidence(0.7)
assert r.to_flat()["pN"] == "pN1"
assert "_deferred" not in r.to_flat()
r.defer("영상과 병리 소견이 상충")
flat = r.to_flat()
assert flat["_deferred"] is True and "상충" in flat["_deferred_reason"]
assert r.to_nested()["_deferred"]["confidence"] == 1.0
back = EvidenceExtractionResult.from_dict(flat, ["pT", "pN"])
assert back.deferred and back.get_field("pT").value == "pT3"
assert json.loads(json.dumps(back.to_nested(), ensure_ascii=False))["pT"]["value"] == "pT3"
print("self-check OK")
def main():
p = argparse.ArgumentParser()
p.add_argument("--self-check", action="store_true")
p.add_argument("--check-json", help="추출 결과 JSON — 근거 없는 필드를 찾아낸다")
p.add_argument("--fields", nargs="*", default=[])
a = p.parse_args()
if a.self_check:
_self_check(); return 0
if not a.check_json:
p.error("--check-json 또는 --self-check 필요")
data = json.load(open(a.check_json, encoding="utf-8"))
names = a.fields or [k for k in data if not k.endswith(("_evidence", "_reasoning", "_confidence"))
and not k.startswith("_")]
r = EvidenceExtractionResult.from_dict(data, names)
bad, low = r.ungrounded(), r.low_confidence()
print(f"필드 {len(names)}개 · 근거 없음 {len(bad)}개 · 확신도 낮음 {len(low)}개")
if bad:
print("근거 없음:", ", ".join(bad))
if low:
print("확신도 낮음:", ", ".join(low))
return 1 if bad else 0
if __name__ == "__main__":
sys.exit(main())
SKILL_PAYLOAD_EOF
chmod +x "$DEST/tnm-staging-extract/scripts/evidence.py"
echo ""
echo "완료 — Skill 9개를 설치했습니다."
echo "Claude Code를 다시 시작하면 인식됩니다."