13a8fca775
- viewclinic/01_readiness_report (진단 덱 v1.0·v2_표준캔버스·v1.1 + 빌더) / 02_poc_proposal (POC 정본·v3 + 빌더) / 03_question_bank (QB 120문항·실측·엑셀) / 04_supporters (GEO 계획·콘텐츠 기획·감사 엑셀) - taeha/ (태하 리포트 + 빌더), ibk/ (기업은행 블로그 감사), landing/ (랜딩 시안·스크린샷·HTML 내보내기) - README.md에 구조·정본/파생본·빌드 명령 안내 - 경로 참조 갱신: 핸드오버 문서 4개, QB 스크립트 3개, 빌더 docstring, discovery 페이지 주석 - 이전에 삭제된 POC v2_표준캔버스 2파일 삭제 반영 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
283 lines
11 KiB
Python
283 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""뷰성형외과 Question Bank 120문항 답변엔진 실측 러너 (OpenAI / Perplexity).
|
|
|
|
측정 방식:
|
|
- openai: Responses API + web_search 도구 (서울 위치 고정)
|
|
- perplexity: chat/completions, model=sonar (자체 웹검색 내장)
|
|
ChatGPT 소비자 UI 자동 조회는 약관 위반이므로 쓰지 않는다.
|
|
|
|
각 질문을 원문 그대로 1회 질의하고, 답변에서 언급된 병원명 상위 5개를
|
|
후처리 모델(OpenAI gpt-4o-mini, 없으면 Gemini 2.5 Flash)로 추출해 JSONL 기록.
|
|
재실행 시 완료된 문항은 건너뛴다.
|
|
|
|
사용:
|
|
python3 scripts/run_question_bank_openai.py --engine perplexity
|
|
OPENAI_API_KEY=... python3 scripts/run_question_bank_openai.py --engine openai
|
|
python3 scripts/run_question_bank_openai.py --engine openai --summarize
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
try:
|
|
import certifi
|
|
SSL_CTX = ssl.create_default_context(cafile=certifi.where())
|
|
except ImportError:
|
|
SSL_CTX = ssl.create_default_context()
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
QUESTIONS = os.path.join(ROOT, "scripts", "question_bank_viewclinic.json")
|
|
OUT_DIR = os.path.join(ROOT, "scripts", "out")
|
|
|
|
EXTRACT_PROMPT = """아래는 AI 검색의 답변이다. 이 답변에서 언급된 성형외과·의원·클리닉 이름을 등장 순서대로 최대 5개 추출하라.
|
|
- 병원 이름만 (플랫폼·앱·언론사 제외. 예: 강남언니, 바비톡, 네이버는 병원이 아님)
|
|
- "뷰성형외과"는 VIEW성형외과, View Plastic Surgery 등 표기 변형도 동일 병원으로 본다
|
|
- JSON만 출력: {"clinics": ["병원1", ...], "view_mentioned": true/false, "view_rank": 순번 또는 null}
|
|
|
|
답변:
|
|
"""
|
|
|
|
|
|
def load_dotenv():
|
|
path = os.path.join(ROOT, ".env")
|
|
if not os.path.exists(path):
|
|
return
|
|
for line in open(path):
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
|
|
|
|
|
|
def post(url, payload, headers, timeout=180):
|
|
req = urllib.request.Request(url, data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json", **headers})
|
|
with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r:
|
|
return json.loads(r.read())
|
|
|
|
|
|
def post_retry(url, payload, headers, tries=4):
|
|
for i in range(tries):
|
|
try:
|
|
return post(url, payload, headers)
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode(errors="replace")
|
|
if e.code in (429, 500, 502, 503) and i < tries - 1:
|
|
wait = 15 * (i + 1)
|
|
print(f" HTTP {e.code}, {wait}s 대기 후 재시도", flush=True)
|
|
time.sleep(wait)
|
|
continue
|
|
raise RuntimeError(f"HTTP {e.code}: {body[:500]}") from e
|
|
except (urllib.error.URLError, TimeoutError):
|
|
if i < tries - 1:
|
|
time.sleep(10)
|
|
continue
|
|
raise
|
|
|
|
|
|
# ---------- engines ----------
|
|
|
|
def openai_answer_text(resp):
|
|
texts, urls = [], []
|
|
for item in resp.get("output", []):
|
|
if item.get("type") != "message":
|
|
continue
|
|
for c in item.get("content", []):
|
|
if c.get("type") == "output_text":
|
|
texts.append(c.get("text", ""))
|
|
for a in c.get("annotations", []):
|
|
if a.get("type") == "url_citation" and a.get("url"):
|
|
urls.append(a["url"])
|
|
return "\n".join(texts).strip(), urls
|
|
|
|
|
|
class OpenAIEngine:
|
|
name = "openai"
|
|
|
|
def __init__(self, model, search_tool):
|
|
self.key = os.environ.get("OPENAI_API_KEY")
|
|
if not self.key:
|
|
sys.exit("OPENAI_API_KEY가 필요합니다 (.env 또는 환경변수)")
|
|
self.model = model
|
|
self.search_tool = search_tool
|
|
|
|
def ask(self, question):
|
|
tool = {"type": self.search_tool,
|
|
"user_location": {"type": "approximate", "country": "KR",
|
|
"city": "Seoul", "timezone": "Asia/Seoul"}}
|
|
try:
|
|
resp = post_retry("https://api.openai.com/v1/responses",
|
|
{"model": self.model, "tools": [tool], "input": question},
|
|
{"Authorization": f"Bearer {self.key}"})
|
|
except RuntimeError as e:
|
|
if self.search_tool == "web_search" and "web_search" in str(e):
|
|
print(" web_search 거부됨, web_search_preview로 폴백", flush=True)
|
|
self.search_tool = "web_search_preview"
|
|
return self.ask(question)
|
|
raise
|
|
return openai_answer_text(resp)
|
|
|
|
|
|
class PerplexityEngine:
|
|
name = "perplexity"
|
|
|
|
def __init__(self, model):
|
|
self.key = os.environ.get("PERPLEXITY_API_KEY")
|
|
if not self.key:
|
|
sys.exit("PERPLEXITY_API_KEY가 필요합니다 (.env)")
|
|
self.model = model
|
|
|
|
def ask(self, question):
|
|
resp = post_retry("https://api.perplexity.ai/chat/completions",
|
|
{"model": self.model,
|
|
"messages": [{"role": "user", "content": question}]},
|
|
{"Authorization": f"Bearer {self.key}"})
|
|
text = resp.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
|
|
urls = list(resp.get("citations") or [])
|
|
for s in resp.get("search_results") or []:
|
|
if s.get("url"):
|
|
urls.append(s["url"])
|
|
return text, urls
|
|
|
|
|
|
# ---------- extraction ----------
|
|
|
|
def extract_clinics(answer):
|
|
prompt = EXTRACT_PROMPT + answer[:8000]
|
|
text = None
|
|
okey = os.environ.get("OPENAI_API_KEY")
|
|
if okey:
|
|
resp = post_retry("https://api.openai.com/v1/responses",
|
|
{"model": "gpt-4o-mini", "input": prompt},
|
|
{"Authorization": f"Bearer {okey}"})
|
|
text, _ = openai_answer_text(resp)
|
|
else:
|
|
gkey = os.environ.get("GEMINI_API_KEY")
|
|
if not gkey:
|
|
sys.exit("추출용으로 OPENAI_API_KEY 또는 GEMINI_API_KEY가 필요합니다")
|
|
resp = post_retry(
|
|
"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key=" + gkey,
|
|
{"contents": [{"parts": [{"text": prompt}]}]}, {})
|
|
parts = resp.get("candidates", [{}])[0].get("content", {}).get("parts", [])
|
|
text = "\n".join(p.get("text", "") for p in parts)
|
|
m = re.search(r"\{.*\}", text or "", re.S)
|
|
if not m:
|
|
return {"clinics": [], "view_mentioned": False, "view_rank": None,
|
|
"parse_error": (text or "")[:200]}
|
|
try:
|
|
return json.loads(m.group(0))
|
|
except json.JSONDecodeError:
|
|
return {"clinics": [], "view_mentioned": False, "view_rank": None,
|
|
"parse_error": text[:200]}
|
|
|
|
|
|
# ---------- io / summary ----------
|
|
|
|
def out_path(engine):
|
|
return os.path.join(OUT_DIR, f"qb_{engine}_results.jsonl")
|
|
|
|
|
|
def load_done(engine):
|
|
done = {}
|
|
p = out_path(engine)
|
|
if os.path.exists(p):
|
|
for line in open(p):
|
|
if line.strip():
|
|
row = json.loads(line)
|
|
done[row["id"]] = row
|
|
return done
|
|
|
|
|
|
def summarize(engine):
|
|
done = load_done(engine)
|
|
if not done:
|
|
print(f"{engine}: 결과 없음")
|
|
return
|
|
label = {"openai": "OpenAI 웹검색(Responses API + web_search, 서울 고정)",
|
|
"perplexity": "Perplexity API (sonar)"}[engine]
|
|
qs = json.load(open(QUESTIONS))
|
|
freq, mentioned = {}, 0
|
|
lines = [f"# 뷰성형외과 Question Bank 실측 결과: {engine}", "",
|
|
f"측정일: {time.strftime('%Y-%m-%d')} / 방식: {label} / {len(done)}문항 완료", "",
|
|
"| # | 질문 | 사전판정 | 뷰 언급 | 뷰 순위 | 답변 내 상위 병원 (등장순) |",
|
|
"|---|---|---|---|---|---|"]
|
|
for q in qs:
|
|
r = done.get(q["id"])
|
|
if not r:
|
|
continue
|
|
clinics = [c for c in r.get("clinics", []) if isinstance(c, str)][:5]
|
|
for i, c in enumerate(clinics):
|
|
freq.setdefault(c, [0, 0])
|
|
freq[c][0] += 1
|
|
if i == 0:
|
|
freq[c][1] += 1
|
|
vm = r.get("view_mentioned")
|
|
if vm:
|
|
mentioned += 1
|
|
rank = r.get("view_rank")
|
|
lines.append(f"| {q['id']} | {q['question']} | {q['prior']} | {'O' if vm else 'X'} | "
|
|
f"{rank if rank else '-'} | {', '.join(clinics) if clinics else '(병원 언급 없음)'} |")
|
|
lines += ["", "## 집계", "",
|
|
f"- 뷰성형외과 언급: {mentioned}/{len(done)}문항 ({mentioned*100//max(len(done),1)}%)",
|
|
"", "### 병원별 언급 빈도 (상위 15)", "",
|
|
"| 병원 | 언급 질문 수 | 1순위 등장 수 |", "|---|---|---|"]
|
|
for name, (cnt, first) in sorted(freq.items(), key=lambda x: -x[1][0])[:15]:
|
|
lines.append(f"| {name} | {cnt} | {first} |")
|
|
out_md = os.path.join(ROOT, "docs", "reports", "viewclinic", "03_question_bank", f"Viewclinic_QB_{engine}_results.md")
|
|
open(out_md, "w").write("\n".join(lines) + "\n")
|
|
print(f"저장: {out_md}")
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--engine", choices=["openai", "perplexity"], default="openai")
|
|
ap.add_argument("--model", default=None)
|
|
ap.add_argument("--search-tool", default="web_search")
|
|
ap.add_argument("--limit", type=int, default=0)
|
|
ap.add_argument("--sleep", type=float, default=1.5)
|
|
ap.add_argument("--summarize", action="store_true")
|
|
args = ap.parse_args()
|
|
load_dotenv()
|
|
|
|
if args.summarize:
|
|
summarize(args.engine)
|
|
return
|
|
|
|
if args.engine == "openai":
|
|
eng = OpenAIEngine(args.model or "gpt-4o", args.search_tool)
|
|
else:
|
|
eng = PerplexityEngine(args.model or "sonar")
|
|
|
|
os.makedirs(OUT_DIR, exist_ok=True)
|
|
qs = json.load(open(QUESTIONS))
|
|
done = load_done(args.engine)
|
|
todo = [q for q in qs if q["id"] not in done]
|
|
if args.limit:
|
|
todo = todo[: args.limit]
|
|
print(f"[{args.engine}] 전체 {len(qs)} / 완료 {len(done)} / 이번 실행 {len(todo)}")
|
|
|
|
with open(out_path(args.engine), "a") as f:
|
|
for n, q in enumerate(todo, 1):
|
|
print(f"[{n}/{len(todo)}] {q['id']} {q['question'][:30]}", flush=True)
|
|
answer, urls = eng.ask(q["question"])
|
|
ext = extract_clinics(answer) if answer else {
|
|
"clinics": [], "view_mentioned": False, "view_rank": None}
|
|
row = {"id": q["id"], "question": q["question"], "prior": q["prior"],
|
|
"engine": args.engine, "model": eng.model,
|
|
"answer": answer, "citations": urls[:10], **ext,
|
|
"ts": time.strftime("%Y-%m-%dT%H:%M:%S")}
|
|
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
|
f.flush()
|
|
time.sleep(args.sleep)
|
|
summarize(args.engine)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|