#!/usr/bin/env python3 """ 세 모델에 동일한 프레젠테이션 요청을 보내고 결과를 비교하는 스크립트. 각 모델이 create_presentation 도구를 호출하면 spec을 추출해 PPTX를 생성한다. """ import json, os, re, sys, time, subprocess import urllib.request, urllib.error # .env에서 API 키 로드 def _load_env(path="/srv/homeclaw/.env"): env = {} try: for line in open(path): line = line.strip() if line and not line.startswith("#") and "=" in line: k, _, v = line.partition("=") env[k.strip()] = v.strip() except FileNotFoundError: pass return env _env = _load_env() PEXELS_KEY = os.environ.get("PEXELS_API_KEY") or _env.get("PEXELS_API_KEY", "") UNSPLASH_KEY = os.environ.get("UNSPLASH_ACCESS_KEY") or _env.get("UNSPLASH_ACCESS_KEY", "") OLLAMA_URL = "http://localhost:11434/api/chat" WORKSPACE = "/srv/homeclaw/.smallclaw/users/papa/workspace" REPORT_TITLE = "한반도_기후_변화_100년_보고서" # 모든 모델 출력 폴더 고정 PPTX_GEN = "/srv/homeclaw/scripts/pptx_gen.py" PREVIEW = "/srv/homeclaw/scripts/pptx_preview.py" MODELS = [ "mistral-large-3:675b-cloud", "kimi-k2.6:cloud", "deepseek-v4-pro:cloud", ] USER_PROMPT = "한반도 기후 변화 100년 보고서를 10슬라이드로 만들어줘. 표 슬라이드와 차트 슬라이드를 반드시 포함하고, 사진이 어울리는 슬라이드에는 image_search 필드로 영문 검색어를 넣어줘. dark 테마로." SYSTEM_PROMPT = """\ 당신은 발표 자료 전문가입니다. `create_presentation` 도구를 사용해 전문적이고 시각적으로 설득력 있는 PowerPoint 자료를 만드세요. [필수] 언어 규칙: 사용자가 한국어로 말하면 반드시 한국어로만 답변하세요. 핵심 원칙: 1. 구조가 먼저: 목적·청중·핵심 메시지를 파악한 후 슬라이드 구성을 결정하세요 2. ONE CALL 원칙: 모든 슬라이드를 단 1번의 create_presentation 호출로 작성하세요 3. 7±2 원칙: 슬라이드당 글머리 기호 5~7개 이하 4. 차트 슬라이드는 반드시 categories와 series 데이터를 채워넣으세요 (비워두면 빈 슬라이드가 됩니다) 5. 사진 슬라이드: layout="split-right" 또는 "fullscreen"과 함께 image_search에 영문 검색어를 넣으세요 """ TOOL_SCHEMA = { "type": "function", "function": { "name": "create_presentation", "description": "PPTX 프레젠테이션을 생성합니다.", "parameters": { "type": "object", "properties": { "spec": { "type": "object", "properties": { "filename": {"type": "string"}, "title": {"type": "string"}, "template": {"type": "string", "description": "business, minimal, creative, dark, pastel, warm"}, "theme": {"type": "string", "description": "dark or light"}, "slides": { "type": "array", "items": { "type": "object", "properties": { "type": {"type": "string", "description": "title|content|section|image|table|chart|timeline"}, "layout": {"type": "string", "description": "split-right|split-left|text|fullscreen|compare"}, "title": {"type": "string"}, "subtitle": {"type": "string"}, "bullets": {"type": "array", "items": {"type": "string"}}, "body": {"type": "string"}, "image_search": {"type": "string"}, "headers": {"type": "array", "items": {"type": "string"}}, "rows": {"type": "array", "items": {"type": "array", "items": {"type": "string"}}}, "chart_type": {"type": "string"}, "categories": {"type": "array", "items": {"type": "string"}}, "series": {"type": "array", "items": {"type": "object"}}, "events": {"type": "array", "items": {"type": "object"}}, } } } }, "required": ["slides"] } }, "required": ["spec"] } } } def call_model(model: str, timeout: int = 180) -> dict | None: """Ollama chat API 호출 → tool call spec 반환""" payload = { "model": model, "stream": False, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": USER_PROMPT}, ], "tools": [TOOL_SCHEMA], } data = json.dumps(payload).encode() req = urllib.request.Request(OLLAMA_URL, data=data, headers={"Content-Type": "application/json"}) print(f"\n[{model}] 요청 중...", flush=True) t0 = time.time() try: with urllib.request.urlopen(req, timeout=timeout) as resp: result = json.loads(resp.read()) except Exception as e: print(f"[{model}] 오류: {e}", flush=True) return None elapsed = time.time() - t0 print(f"[{model}] 응답 완료 ({elapsed:.1f}초)", flush=True) msg = result.get("message", {}) # tool_calls 방식 for tc in (msg.get("tool_calls") or []): fn = tc.get("function", {}) if fn.get("name") == "create_presentation": args = fn.get("arguments") or fn.get("parameters") or {} if isinstance(args, str): try: args = json.loads(args) except Exception: pass spec = args.get("spec") if isinstance(args, dict) else None if isinstance(spec, str): try: spec = json.loads(spec) except Exception: pass if isinstance(spec, dict) and spec: return spec # content 안에 JSON 블록으로 spec을 반환하는 경우 fallback content = msg.get("content") or "" m = re.search(r'```(?:json)?\s*(\{.*?"slides".*?\})\s*```', content, re.DOTALL) if m: try: parsed = json.loads(m.group(1)) if "spec" in parsed: return parsed["spec"] if "slides" in parsed: return parsed except Exception: pass print(f"[{model}] tool call을 찾을 수 없음. content 미리보기:\n{content[:300]}", flush=True) return None def _search_pexels(query: str) -> str | None: if not PEXELS_KEY: return None url = f"https://api.pexels.com/v1/search?query={urllib.request.quote(query)}&per_page=3&orientation=landscape" req = urllib.request.Request(url, headers={"Authorization": PEXELS_KEY}) try: with urllib.request.urlopen(req, timeout=10) as r: data = json.loads(r.read()) photos = data.get("photos") or [] if photos: return photos[0].get("src", {}).get("large2x") or photos[0].get("src", {}).get("large") except Exception: pass return None def _search_unsplash(query: str) -> str | None: if not UNSPLASH_KEY: return None url = f"https://api.unsplash.com/search/photos?query={urllib.request.quote(query)}&per_page=3&orientation=landscape" req = urllib.request.Request(url, headers={"Authorization": f"Client-ID {UNSPLASH_KEY}"}) try: with urllib.request.urlopen(req, timeout=10) as r: data = json.loads(r.read()) results = data.get("results") or [] if results: return results[0].get("urls", {}).get("regular") except Exception: pass return None def download_image(query: str, dest_path: str) -> bool: """Pexels → Unsplash 순으로 이미지 검색 후 다운로드. 성공 시 True.""" img_url = _search_pexels(query) or _search_unsplash(query) if not img_url: return False try: req = urllib.request.Request(img_url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=15) as r: data = r.read() if len(data) < 100: return False with open(dest_path, "wb") as f: f.write(data) return True except Exception: return False def resolve_images(spec: dict, project_dir: str, model_tag: str) -> None: """spec 슬라이드의 image_search 필드를 image_path로 변환.""" for i, slide in enumerate(spec.get("slides", [])): query = slide.get("image_search") if not query or slide.get("image_path"): continue ext = ".jpg" fname = f"slide{i+1}_image{ext}" dest = os.path.join(project_dir, fname) print(f"[{model_tag}] 슬라이드 {i+1} 이미지 검색: {query!r}", flush=True) if download_image(query, dest): slide["image_path"] = fname print(f"[{model_tag}] 슬라이드 {i+1} 이미지 저장: {fname}", flush=True) else: print(f"[{model_tag}] 슬라이드 {i+1} 이미지 실패", flush=True) slide.pop("image_search", None) def build_pptx(spec: dict, model_tag: str, username: str = "papa") -> str | None: """spec → 이미지 다운로드 → PPTX 생성 → 경로 반환""" safe_tag = re.sub(r'[^a-zA-Z0-9_-]', '_', model_tag.split(":")[0]) spec["filename"] = f"climate_compare_{safe_tag}" spec["username"] = username spec["title"] = REPORT_TITLE # 모든 모델이 동일 폴더에 저장되도록 고정 spec.setdefault("template", "dark") spec.setdefault("theme", "dark") # 프로젝트 폴더 미리 생성 후 이미지 다운로드 title_slug = re.sub(r'[^a-zA-Z0-9가-힣_\-]', '_', REPORT_TITLE).strip("_")[:60] or "presentation" project_dir = os.path.join(WORKSPACE, title_slug) os.makedirs(project_dir, exist_ok=True) resolve_images(spec, project_dir, model_tag) spec_path = f"/tmp/compare_spec_{safe_tag}.json" with open(spec_path, "w", encoding="utf-8") as f: json.dump(spec, f, ensure_ascii=False, indent=2) print(f"[{model_tag}] PPTX 생성 중...", flush=True) r = subprocess.run( ["python3", PPTX_GEN, spec_path, WORKSPACE], capture_output=True, text=True, cwd="/srv/homeclaw" ) try: out = json.loads(r.stdout) except Exception: print(f"[{model_tag}] pptx_gen 오류:\n{r.stderr[-500:]}", flush=True) return None if not out.get("success"): print(f"[{model_tag}] 실패: {out.get('error')}", flush=True) return None pptx_path = out["path"] print(f"[{model_tag}] 완료: {pptx_path}", flush=True) if out.get("warnings"): print(f"[{model_tag}] 경고: {out['warnings']}", flush=True) return pptx_path def make_preview(pptx_path: str) -> list[str]: # LibreOffice requires .pptx extension to recognize format if not pptx_path.endswith(".pptx"): pptx_path = pptx_path + ".pptx" stem = os.path.splitext(os.path.basename(pptx_path))[0] prev_dir = os.path.join(os.path.dirname(pptx_path), f"preview_{stem}") os.makedirs(prev_dir, exist_ok=True) r = subprocess.run( ["python3", PREVIEW, pptx_path, prev_dir], capture_output=True, text=True, cwd="/srv/homeclaw" ) try: # MuPDF error messages may appear before the JSON line last_line = r.stdout.strip().rsplit("\n", 1)[-1] out = json.loads(last_line) return [os.path.join(prev_dir, img) for img in out.get("images", [])] except Exception: return [] if __name__ == "__main__": results = {} total_start = time.time() for model in MODELS: t0 = time.time() spec = call_model(model) llm_elapsed = time.time() - t0 if spec is None: print(f"[{model}] spec 없음, 건너뜀\n") continue pptx_path = build_pptx(spec, model) if pptx_path is None: continue total_elapsed = time.time() - t0 previews = make_preview(pptx_path) image_count = sum(1 for s in spec.get("slides", []) if s.get("image_search") or s.get("image_path")) results[model] = { "pptx": pptx_path, "previews": previews, "spec": spec, "llm_sec": round(llm_elapsed, 1), "total_sec": round(total_elapsed, 1), "image_count": image_count, } print(f"[{model}] 프리뷰 {len(previews)}장 생성 완료 (LLM {llm_elapsed:.1f}s / 전체 {total_elapsed:.1f}s)\n") print("\n===== 비교 결과 =====") for model, info in results.items(): slide_types = [s.get("type","?") for s in info["spec"].get("slides", [])] img_slides = [i+1 for i,s in enumerate(info["spec"].get("slides",[])) if s.get("image_search") or s.get("image_path")] print(f"{model}:") print(f" 슬라이드: {len(slide_types)}장 {slide_types}") print(f" 사진슬라이드: {len(img_slides)}장 (슬라이드 {img_slides})") print(f" 응답시간: LLM {info['llm_sec']}s / 전체(PPTX포함) {info['total_sec']}s") print(f" PPTX: {info['pptx']}") print(f" 미리보기: {info['previews'][:2]}{'...' if len(info['previews'])>2 else ''}")