#!/bin/sh # Claude Code Skill 설치 스크립트 # 생성: 2026-08-11 · Skill 1개 · 파일 2개 # # ./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/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" echo "" echo "완료 — Skill 1개를 설치했습니다." echo "Claude Code를 다시 시작하면 인식됩니다."