"""DA-12: 1단계 — Kei 실장 (꼭지 추출 + 분석). Kei API를 통해 Kei persona가 사고하여 꼭지를 추출한다. Kei API는 필수. fallback 없음. 성공할 때까지 무한 재시도. """ from __future__ import annotations import json import logging import re from typing import Any import httpx from src.config import settings from src.sse_utils import stream_sse_tokens logger = logging.getLogger(__name__) KEI_PROMPT = ( "다음 콘텐츠로 슬라이드 1장을 만들려고 해. 어떤 스토리로 구성할지 설계해줘.\n\n" "## 1단계: 핵심 메시지 파악\n" "- 이 콘텐츠가 전달하려는 **핵심 메시지**를 한 줄로 파악해줘.\n" "- 슬라이드를 본 사람이 기억해야 할 단 하나의 문장.\n" "- core_message 필드에 기록.\n\n" "## 2단계: 정보 구조 파악\n" "- 본문 흐름(flow)과 참조 정보(reference)로 분리되는 구조인가?\n" "- 독립적으로 참조되는 정보(용어 정의, 부록)가 있는가?\n" "- info_structure 필드에 기술.\n\n" "## 3단계: 슬라이드 스토리라인 설계\n" "핵심 메시지를 전달하기 위한 **흐름**을 설계해줘.\n" "각 꼭지에 purpose를 부여하고, topics 배열에 기록.\n\n" "## 4단계: 페이지 구조 판단 (비중 시스템)\n" "콘텐츠를 분석하여 이 페이지의 **구조와 비중**을 판단하라:\n\n" "- **본심**: 이 페이지가 말하려는 핵심. 가장 큰 공간을 차지해야 함.\n" " 비교라면 비교표, 관계라면 관계도, 프로세스라면 흐름도로 구조화.\n" " 비교 구조일 때 비교 목적(왜 비교하는가)을 summary에 명시.\n" "- **배경**: 본심을 이해하기 위한 도입/배경. 간결하게. 2-3줄이면 충분.\n" "- **첨부**: 본심을 보조하는 참조 정보 (용어 정의 등). sidebar 배치.\n" " role: 'reference'로 표시. 본문 흐름을 방해하지 않도록.\n" "- **결론**: 절대 잊으면 안 되는 핵심 한 줄. footer.\n\n" "각 역할에 해당하는 topic_ids와 **공간 비중(weight, 합계 1.0)**을 결정하라.\n" "**콘텐츠에 따라 비중은 매번 달라진다. 고정값이 아니다.**\n" "page_structure 필드에 기록.\n\n" "## 원본 텍스트 보존 원칙\n" "- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n" "- 원본 텍스트는 최대한 보존. 약간의 편집만.\n" "- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n" "- 각 꼭지의 source_hint에 원본의 어떤 부분이 가는지 명시\n\n" "## 배치 규칙\n" "- 참조 정보(용어 정의 등)는 role: 'reference'로 표시 → 사이드바 배치\n" "- 본문 흐름은 role: 'flow' → 메인 영역 배치\n" "- 결론은 layer: 'conclusion' → 하단 배치\n" "- detail_target: true는 정말로 별도로 봐야 하는 상세 데이터에만 사용\n" "- 이미지/표가 있으면 images[], tables[]에 기록\n" "- 1페이지 적정 꼭지: 5개. 분량 적으면 1페이지로.\n" "- **슬라이드 제목(title)과 첫 번째 꼭지 제목은 달라야 한다.** 슬라이드 제목은 전체 주제, 꼭지 제목은 해당 위치의 구체적 내용.\n\n" "## 출력 형식 (JSON만)\n" "```json\n" '{"title": "제목", ' '"core_message": "이 슬라이드의 핵심 메시지 한 줄", ' '"total_pages": 1, ' '"info_structure": "정보 구조 설명", ' '"page_structure": {' '"본심": {"topic_ids": [2, 3], "weight": 0.60}, ' '"배경": {"topic_ids": [1], "weight": 0.15}, ' '"첨부": {"topic_ids": [4], "weight": 0.15}, ' '"결론": {"topic_ids": [5], "weight": 0.10}}, ' '"topics": [' '{"id": 1, "title": "꼭지 제목", "summary": "요약", ' '"purpose": "문제제기|근거사례|핵심전달|용어정의|결론강조|구조시각화", ' '"source_hint": "원본에서 이 위치에 가져올 텍스트 범위 설명", ' '"layer": "intro|core|supporting|conclusion", ' '"role": "flow|reference", ' '"section_title": "sidebar에 표시할 섹션 제목 (reference일 때만. 예: 용어 정의, 참고 자료)", ' '"emphasis": true, "direction": "vertical|horizontal|flexible", ' '"content_type": "text|image|table|mixed", ' '"detail_target": false, "page": 1}], ' '"images": [{"topic_id": 1, "role": "key|supporting", "has_text": false, "description": "이미지 설명"}], ' '"tables": [{"topic_id": 2, "rows": 5, "cols": 3, "fits_single_page": true, "description": "표 설명"}]}\n' "```\n\n" "## 콘텐츠:\n" ) async def classify_content(content: str) -> dict[str, Any] | None: """1단계: Kei API를 통해 꼭지를 추출하고 분석한다. Kei API만 사용. fallback 없음. 실패 시 None → pipeline에서 에러. """ result = await _call_kei_api(content) if result: logger.info( f"[Kei API] 꼭지 추출 완료: {result.get('title', '')}, " f"{len(result.get('topics', []))}개 꼭지" ) return result logger.error("[Kei API] 꼭지 추출 실패. Kei API(localhost:8000) 확인 필요.") return None KEI_PROMPT_B = ( "아래는 슬라이드 스토리라인 설계 결과(1단계-A)와 원본 콘텐츠이다.\n" "각 꼭지의 **컨셉을 구체화**해줘. 1회 응답으로 모든 꼭지를 한꺼번에 처리.\n\n" "## 각 꼭지에 대해 판단할 것\n\n" "1. **관계 성격 (relation_type)**:\n" " - sequence: 시간/단계 순서 (A→B→C)\n" " - inclusion: 포함/융합 관계 (A가 B,C를 포함하거나 B,C가 합쳐져 A를 이룸)\n" " - comparison: 대등 비교 (A vs B)\n" " - hierarchy: 상위-하위 (A > B > C)\n" " - definition: 용어 정의 (나열)\n" " - cause_effect: 원인-결과\n" " - none: 관계 표현 불필요 (단순 텍스트)\n\n" "2. **표현 성격 (expression_hint)** — 관계 성격을 설명. 구체 블록 이름은 쓰지 마라:\n" " - 예: '기술 융합/포함 관계. 순서 아님. 구성요소 간 관계 표현 필요.'\n" " - 예: '수단-목적 관계. 대등 비교가 아님. 역할 차이를 보여줘야 함.'\n" " - 예: '용어 3개의 독립적 정의. 나열.'\n\n" "3. **원본 데이터 확인 (source_data)**:\n" " - 원본에 비교표가 있는가? → 행/열 수, 활용 여부\n" " - 원본에 사례/증거가 있는가? → 출처 명시\n" " - 원본에 이미지가 있는가? → 역할\n" " - 놓치면 안 되는 핵심 데이터가 있는가?\n\n" "## 출력 형식 (JSON만)\n" "```json\n" '{"concepts": [' '{"topic_id": 1, ' '"relation_type": "inclusion|sequence|comparison|hierarchy|definition|cause_effect|none", ' '"expression_hint": "관계 성격 설명 (블록 이름 쓰지 말 것)", ' '"source_data": "원본에서 활용해야 할 데이터 설명"}]}\n' "```\n\n" ) async def refine_concepts( content: str, analysis: dict[str, Any], ) -> dict[str, Any]: """1단계-B: 각 꼭지의 컨셉을 구체화한다. 1단계-A 결과(topics)를 받아서, 각 꼭지의 관계 성격/표현 방법/원본 데이터를 판단. Kei API만 사용. fallback 없음. 성공할 때까지 재시도. 1회 호출로 모든 꼭지를 한꺼번에 처리. """ import asyncio topics = analysis.get("topics", []) if not topics: return analysis # 꼭지 요약 텍스트 topics_text = "\n".join( f"- 꼭지 {t.get('id', '?')}: {t.get('title', '?')} " f"[purpose: {t.get('purpose', '?')}, summary: {t.get('summary', '')}]" for t in topics ) prompt = ( KEI_PROMPT_B + f"## 1단계-A 스토리라인 결과\n" f"핵심 메시지: {analysis.get('core_message', '?')}\n" f"꼭지 목록:\n{topics_text}\n\n" f"## 원본 콘텐츠\n{content}\n" ) kei_url = getattr(settings, "kei_api_url", "http://localhost:8000") RETRY_INTERVAL = 10 attempt = 0 while True: attempt += 1 try: async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{kei_url}/api/message", json={ "message": prompt, "session_id": "design-agent-refine", "mode_hint": "chat", }, timeout=None, ) as response: if response.status_code != 200: logger.warning(f"[1단계-B] Kei API HTTP {response.status_code} (시도 {attempt})") await asyncio.sleep(RETRY_INTERVAL) continue full_text = await stream_sse_tokens(response) if not full_text: logger.warning(f"[1단계-B] 응답 텍스트 없음 (시도 {attempt})") await asyncio.sleep(RETRY_INTERVAL) continue result = _parse_json(full_text) if result and "concepts" in result: # topics에 concept 정보 병합 # Kei가 topic_id 또는 id로 응답할 수 있으므로 양쪽 다 체크 concept_map = {} for c in result["concepts"]: tid = c.get("topic_id") or c.get("id") if tid is not None: concept_map[tid] = c for topic in topics: concept = concept_map.get(topic.get("id")) if concept: topic["relation_type"] = concept.get("relation_type", "") topic["expression_hint"] = concept.get("expression_hint", "") topic["source_data"] = concept.get("source_data", "") logger.info(f"[1단계-B] 컨셉 구체화 완료: {len(result['concepts'])}개") return analysis else: logger.warning(f"[1단계-B] JSON 파싱 실패 (시도 {attempt}): {full_text[:200]}") await asyncio.sleep(RETRY_INTERVAL) continue except Exception as e: logger.warning(f"[1단계-B] Kei API 실패 (시도 {attempt}): {e}") await asyncio.sleep(RETRY_INTERVAL) continue async def _call_kei_api(content: str) -> dict[str, Any] | None: """Kei API를 통해 꼭지 추출. SSE 스트리밍으로 실시간 수신.""" kei_url = getattr(settings, "kei_api_url", "http://localhost:8000") try: async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{kei_url}/api/message", json={ "message": KEI_PROMPT + content, "session_id": "design-agent", "mode_hint": "chat", }, timeout=None, ) as response: if response.status_code != 200: logger.warning(f"Kei API HTTP {response.status_code}") return None full_text = await stream_sse_tokens(response) if not full_text: logger.warning("Kei API 응답에서 텍스트 추출 실패") return None result = _parse_json(full_text) if result and "topics" in result: return result logger.warning(f"Kei API JSON 파싱 실패. 텍스트: {full_text[:200]}") return None except Exception as e: logger.warning(f"Kei API 호출 실패: {e}") return None # ────────────────────────────────────── # J-7: Kei 최종 검수 # ────────────────────────────────────── KEI_REVIEW_PROMPT = """당신은 11년 경력의 기획 실장이다. 디자인 팀장이 조립한 슬라이드를 최종 검수한다. ## 검수 관점 1. 핵심 메시지(core_message)가 시각적으로 명확히 전달되는가? 2. 콘텐츠 흐름이 블록 배치와 일치하는가? 3. 각 블록이 해당 꼭지의 purpose에 적합한가? 4. 중요한 내용이 빠지거나 과도하게 축소되지 않았는가? 5. 높이 초과: 각 zone의 블록+텍스트가 예산을 초과하는가? - 텍스트 축약으로 해결 가능 → shrink - 콘텐츠가 본질적으로 큼 → overflow_detected 6. **핵심전달이 body에서 가장 큰 시각적 비중을 차지하는가?** - 핵심전달 블록이 도입부(문제제기+근거사례)보다 작으면 → rewrite로 비중 조정 7. **문제제기가 간결한가? (100자 이내)** - 초과 시 → shrink (target_ratio: 0.5) 8. **용어정의가 sidebar에 있는가?** - body에 있으면 → 구조 문제 지적 (issues에 명시) 9. **핵심전달 블록이 화면 안에 보이는가?** - 잘리면 → overflow_detected ## 조정 action - expand: 텍스트 늘림 (target_ratio, 예: 1.3) - shrink: 텍스트 줄임 (target_ratio, 예: 0.7) - rewrite: 텍스트 재작성 (detail에 방향) - overflow_detected: 높이 초과, 콘텐츠 판단 필요 (zone과 블록 명시) ## 출력 (JSON만. 설명 없이.) {"needs_adjustment": true/false, "issues": ["이슈1"], "adjustments": [{"block_area": "...", "action": "expand|shrink|rewrite|overflow_detected", "target_ratio": 1.3, "detail": "..."}]} """ async def call_kei_final_review( html: str, block_summary: list[str], zone_budget_text: str, overflow_hint_text: str, analysis: dict[str, Any] | None = None, screenshot_b64: str | None = None, ) -> dict[str, Any] | None: """Phase N-4: Kei(Opus)가 스크린샷을 보고 최종 검수한다. 스크린샷이 있으면: Anthropic API 직접 호출 (Opus + 멀티모달) 스크린샷이 없으면: Kei API 경유 (텍스트 기반) 어느 경로든 Kei(Opus)가 판단. Sonnet 절대 금지. """ import anthropic core_message = analysis.get("core_message", "") if analysis else "" topics_summary = "" if analysis: topics_summary = "\n".join( f"- 꼭지 {t.get('id')}: {t.get('title', '')} [{t.get('purpose', '')}]" for t in analysis.get("topics", []) ) review_text = ( f"## 핵심 메시지\n{core_message}\n\n" f"## 꼭지 목록\n{topics_summary}\n\n" f"## 블록별 데이터 양\n" + "\n".join(block_summary) + zone_budget_text + overflow_hint_text + f"\n\n위 슬라이드를 검수하고 조정이 필요한지 판단해. JSON만." ) # 스크린샷이 있으면: Opus 직접 호출 + 이미지 전달 if screenshot_b64: try: client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key) response = await client.messages.create( model="claude-opus-4-0-20250514", max_tokens=4096, system=KEI_REVIEW_PROMPT, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": screenshot_b64, }, }, { "type": "text", "text": review_text, }, ], }], ) result_text = response.content[0].text result = _parse_json(result_text) if result and "needs_adjustment" in result: logger.info( f"[Kei 최종 검수] 스크린샷 기반, needs_adjustment={result['needs_adjustment']}" ) return result logger.warning("[Kei 최종 검수] 스크린샷 기반 JSON 파싱 실패") return None except Exception as e: logger.warning(f"Kei 최종 검수 (스크린샷) 실패: {e}") return None # 스크린샷 없으면: Kei API 경유 (텍스트 기반) kei_url = getattr(settings, "kei_api_url", "http://localhost:8000") prompt = ( KEI_REVIEW_PROMPT + "\n\n" + review_text + f"\n\n## 조립 HTML (요약)\n{html[:3000]}" ) try: async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{kei_url}/api/message", json={ "message": prompt, "session_id": "design-agent-final-review", "mode_hint": "chat", }, timeout=None, ) as response: if response.status_code != 200: logger.warning(f"Kei 최종 검수 HTTP {response.status_code}") return None full_text = await stream_sse_tokens(response) if full_text: result = _parse_json(full_text) if result and "needs_adjustment" in result: logger.info( f"[Kei 최종 검수] 텍스트 기반, needs_adjustment={result['needs_adjustment']}" ) return result logger.warning("[Kei 최종 검수] JSON 파싱 실패") return None logger.warning("Kei 최종 검수 텍스트 추출 실패") return None except Exception as e: logger.warning(f"Kei 최종 검수 실패: {e}") return None # ────────────────────────────────────── # I-9: Kei 넘침 판단 호출 # ────────────────────────────────────── KEI_OVERFLOW_PROMPT = """당신은 슬라이드 콘텐츠 전문가이다. 디자인 팀장이 배치한 블록들이 컨테이너(zone)의 높이 예산을 초과했다. 콘텐츠의 중요도와 전달 메시지를 기준으로 어떻게 처리할지 판단하라. ## 판단 기준 - 텍스트 분량만 줄이면 현재 블록 구조 안에서 해결되는가? → "trim" - 콘텐츠 자체가 컨테이너에 담기엔 본질적으로 많은가? → "restructure" - 중요도가 높은 콘텐츠를 무리하게 축소하면 안 된다 - 부가/상세 정보는 팝업(detail page)으로 분리할 수 있다 ## 출력 (JSON만. 설명 없이.) Option 1 (텍스트 축약): {"decision": "trim", "trim_targets": [{"topic_id": 1, "max_chars": 200, "reason": "부연 설명 축약 가능"}]} Option 2 (핵심 재구성 + 팝업 분리): {"decision": "restructure", "core_topics": [1, 2], "detail_topics": [3], "reason": "12행 비교표는 팝업으로 분리"} """ async def call_kei_overflow_judgment( overflows: list[dict], content: str, analysis: dict[str, Any], ) -> dict[str, Any] | None: """Kei API에 넘침 상황을 전달하고 판단을 받는다. 반드시 Kei API 경유. Anthropic 직접 호출 절대 금지. 실패 시 None → pipeline에서 무한 재시도. """ kei_url = getattr(settings, "kei_api_url", "http://localhost:8000") overflow_desc = json.dumps(overflows, ensure_ascii=False, indent=2) topics_desc = json.dumps( [ { "id": t.get("id"), "title": t.get("title", ""), "purpose": t.get("purpose", ""), "summary": t.get("summary", "")[:100], } for t in analysis.get("topics", []) ], ensure_ascii=False, ) # I-8: 대형 콘텐츠 정보 포함 extra_info = "" tables = analysis.get("tables", []) if tables: extra_info += f"\n\n## 테이블 정보\n{json.dumps(tables, ensure_ascii=False)}" images = analysis.get("images", []) if images: extra_info += f"\n\n## 이미지 정보\n{json.dumps(images, ensure_ascii=False)}" prompt = ( KEI_OVERFLOW_PROMPT + "\n\n" f"## 넘침 현황\n{overflow_desc}\n\n" f"## 꼭지 목록\n{topics_desc}" f"{extra_info}\n\n" f"## 원본 콘텐츠 요약\n{content[:2000]}" ) try: async with httpx.AsyncClient(timeout=None) as client: async with client.stream( "POST", f"{kei_url}/api/message", json={ "message": prompt, "session_id": "design-agent-overflow", "mode_hint": "chat", }, timeout=None, ) as response: if response.status_code != 200: logger.warning(f"Kei API (overflow) HTTP {response.status_code}") return None full_text = await stream_sse_tokens(response) if full_text: result = _parse_json(full_text) if result and "decision" in result: logger.info(f"[Kei 넘침 판단] decision={result['decision']}") return result logger.warning("[Kei 넘침 판단] JSON 파싱 실패 또는 decision 없음") return None logger.warning("Kei API (overflow) 텍스트 추출 실패") return None except Exception as e: logger.warning(f"Kei API (overflow) 호출 실패: {e}") return None def _parse_json(text: str) -> dict[str, Any] | None: """텍스트에서 JSON을 추출한다. Kei API가 마크다운 리스트 접두사(- )를 붙여 응답하는 경우에도 처리. """ # 전처리: 각 줄 앞의 마크다운 리스트 접두사(- ) 제거 # Kei API가 JSON을 마크다운 리스트로 감싸서 응답하는 경우 대응 lines = text.split("\n") cleaned_lines = [] for line in lines: stripped = line.lstrip() if stripped.startswith("- "): cleaned_lines.append(stripped[2:]) elif stripped.startswith("* "): cleaned_lines.append(stripped[2:]) else: cleaned_lines.append(stripped) cleaned = "\n".join(cleaned_lines) # 원본 + 클린 버전 둘 다 시도 for target in [text, cleaned]: patterns = [ r"```json\s*(.*?)```", r"```\s*(.*?)```", r"(\{.*\})", ] for pattern in patterns: match = re.search(pattern, target, re.DOTALL) if match: try: return json.loads(match.group(1).strip()) except json.JSONDecodeError: continue return None # manual_classify 삭제됨. Kei API는 필수. fallback 없음.