From 5d8706172c00d1c9abd38aa1371eb1a92caa38cf Mon Sep 17 00:00:00 2001 From: kyeongmin Date: Thu, 26 Mar 2026 07:57:39 +0900 Subject: [PATCH] =?UTF-8?q?Phase=20H=20=EB=B3=B4=EC=99=84=20=EA=B5=AC?= =?UTF-8?q?=ED=98=84:=201=EB=8B=A8=EA=B3=84=20A/B=20=EB=B6=84=EB=A6=AC=20+?= =?UTF-8?q?=20=EC=A0=95=EB=B0=80=20=EA=B2=80=ED=86=A0=208=EA=B0=9C=20?= =?UTF-8?q?=EB=B0=98=EC=98=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit H-1a: KEI_PROMPT에 제목 중복 방지 지시 추가 H-1b: KEI_PROMPT_B (컨셉 구체화) + refine_concepts() 신규 함수 - 각 꼭지의 relation_type, expression_hint, source_data 판단 - 1회 호출로 전체 꼭지 처리 - session_id: "design-agent-refine" (별도) - 실패 시 1단계-A 결과 그대로 반환 (pipeline 안 멈춤) H-5: 팀장에게 relation_type + expression_hint + source_data 전달 - 꼭지 요약에 관계/표현/원본데이터 포함 section-title-with-bg body 금지: - STEP_B_PROMPT에 규칙 추가 - BODY_FORBIDDEN_MAP + _validate_height_budget에서 코드 레벨 교체 manual_classify fallback 동기화: - core_message, purpose, source_hint 기본값 추가 Co-Authored-By: Claude Opus 4.6 (1M context) --- src/design_director.py | 24 ++++++++- src/kei_client.py | 112 ++++++++++++++++++++++++++++++++++++++++- src/pipeline.py | 9 +++- 3 files changed, 141 insertions(+), 4 deletions(-) diff --git a/src/design_director.py b/src/design_director.py index 905e809..6d1a87c 100644 --- a/src/design_director.py +++ b/src/design_director.py @@ -258,6 +258,8 @@ header/footer는 고정이므로 건드리지 않는다. - 너비 35% 이하 zone(sidebar)에는 카드 1열, 시각화 블록 금지 - catalog의 when/not_for와 height_cost를 반드시 읽고 선택 - 같은 블록 타입 반복 금지 — 다양한 블록 활용 +- **section-title-with-bg는 body/sidebar/footer zone에서 사용 금지.** 이 블록은 자세히보기 전용 페이지 상단에만 사용. +- 각 꼭지의 relation_type과 expression_hint를 보고 적합한 블록을 선택하라 ## purpose 기반 블록 선택 가이드 (참고, 강제 아님) 각 꼭지의 purpose에 맞는 블록 계열을 선택하라: @@ -472,7 +474,10 @@ async def create_layout_concept( line = ( f"꼭지 {t.get('id', '?')}: {t.get('title', '?')} " f"[{t.get('layer', '?')}, ROLE:{role}, " - f"강조:{t.get('emphasis', False)}]" + f"강조:{t.get('emphasis', False)}, " + f"관계:{t.get('relation_type', '?')}, " + f"표현:{t.get('expression_hint', '?')}, " + f"원본데이터:{t.get('source_data', '?')}]" ) if t.get("detail_target"): line += " → ★detail_target (callout-solution으로 요약 배치 권장)" @@ -665,6 +670,11 @@ HEIGHT_COST_PX = { "xlarge": 400, } +# body/sidebar/footer zone에서 사용 금지인 블록 → 교체 +BODY_FORBIDDEN_MAP = { + "section-title-with-bg": "topic-center", # 500px 블록 → compact 헤더로 +} + # xlarge/large → medium/compact 교체 후보 DOWNGRADE_MAP = { "venn-diagram": "card-icon-desc", @@ -701,6 +711,7 @@ def _load_catalog_map_for_height() -> dict[str, str]: def _validate_height_budget(blocks: list[dict], preset: dict) -> None: """zone별 height_cost 합산을 검증하고, 초과 시 큰 블록을 교체한다. + 또한 body/sidebar/footer zone에서 금지된 블록을 교체한다. 코드 레벨 검증 — Sonnet이 높이 예산을 안 지켜도 강제 교정. """ zones = preset.get("zones", {}) @@ -714,6 +725,17 @@ def _validate_height_budget(blocks: list[dict], preset: dict) -> None: zone_blocks[area] = [] zone_blocks[area].append(block) + # 금지 블록 교체 (body/sidebar/footer에서 사용 불가한 블록) + for block in blocks: + area = block.get("area", "body") + block_type = block.get("type", "") + if area != "header" and block_type in BODY_FORBIDDEN_MAP: + replacement = BODY_FORBIDDEN_MAP[block_type] + logger.warning( + f"[금지 블록 교체] {block_type} → {replacement} (area={area})" + ) + block["type"] = replacement + for area, area_blocks in zone_blocks.items(): zone_info = zones.get(area, {}) budget = zone_info.get("budget_px", 490) diff --git a/src/kei_client.py b/src/kei_client.py index c7ebcca..1d5fed4 100644 --- a/src/kei_client.py +++ b/src/kei_client.py @@ -46,7 +46,8 @@ KEI_PROMPT = ( "- 결론은 layer: 'conclusion' → 하단 배치\n" "- detail_target: true는 정말로 별도로 봐야 하는 상세 데이터에만 사용\n" "- 이미지/표가 있으면 images[], tables[]에 기록\n" - "- 1페이지 적정 꼭지: 5개. 분량 적으면 1페이지로.\n\n" + "- 1페이지 적정 꼭지: 5개. 분량 적으면 1페이지로.\n" + "- **슬라이드 제목(title)과 첫 번째 꼭지 제목은 달라야 한다.** 슬라이드 제목은 전체 주제, 꼭지 제목은 해당 위치의 구체적 내용.\n\n" "## 출력 형식 (JSON만)\n" "```json\n" '{"title": "제목", ' @@ -87,6 +88,112 @@ async def classify_content(content: str) -> dict[str, Any] | None: 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만 사용. 실패 시 1단계-A 결과를 그대로 반환 (pipeline 안 멈춤). + 1회 호출로 모든 꼭지를 한꺼번에 처리. + """ + 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") + + 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}") + return analysis + + full_text = await _stream_sse_tokens(response) + + if not full_text: + logger.warning("[1단계-B] 응답 텍스트 없음. 1단계-A 결과 유지.") + return analysis + + result = _parse_json(full_text) + if result and "concepts" in result: + # topics에 concept 정보 병합 + concept_map = {c.get("topic_id"): c for c in result["concepts"]} + 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'])}개") + else: + logger.warning(f"[1단계-B] JSON 파싱 실패. 1단계-A 결과 유지. 텍스트: {full_text[:200]}") + + except Exception as e: + logger.warning(f"[1단계-B] Kei API 실패: {e}. 1단계-A 결과 유지.") + + return analysis + + async def _call_kei_api(content: str) -> dict[str, Any] | None: """Kei API를 통해 꼭지 추출. SSE 스트리밍으로 실시간 수신.""" kei_url = getattr(settings, "kei_api_url", "http://localhost:8000") @@ -290,6 +397,7 @@ def manual_classify(content: str) -> dict[str, Any]: """분류 실패 시 기본 구조 fallback.""" return { "title": "슬라이드", + "core_message": "", "total_pages": 1, "info_structure": "", "topics": [ @@ -297,6 +405,8 @@ def manual_classify(content: str) -> dict[str, Any]: "id": 1, "title": "핵심 내용", "summary": content[:100], + "purpose": "핵심전달", + "source_hint": "", "layer": "core", "role": "flow", "emphasis": False, diff --git a/src/pipeline.py b/src/pipeline.py index bfd6fd2..9c7b266 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -15,7 +15,7 @@ from typing import Any, AsyncIterator import anthropic -from src.kei_client import classify_content, manual_classify +from src.kei_client import classify_content, manual_classify, refine_concepts from src.design_director import create_layout_concept, LAYOUT_PRESETS, select_preset from src.content_editor import fill_content from src.renderer import render_slide @@ -48,7 +48,12 @@ async def generate_slide( topic_count = len(analysis.get("topics", [])) page_count = analysis.get("total_pages", 1) - logger.info(f"1단계 완료: {topic_count}개 꼭지, {page_count}페이지") + logger.info(f"1단계-A 완료: {topic_count}개 꼭지, {page_count}페이지") + + # 1단계-B: 각 꼭지 컨셉 구체화 + yield {"event": "progress", "data": "1.5/5 Kei 실장이 각 꼭지의 컨셉을 구체화 중..."} + analysis = await refine_concepts(content, analysis) + logger.info("1단계-B 완료: 컨셉 구체화") # 이미지 크기 측정 (base_path 있을 때만) image_sizes = get_image_sizes(content, base_path)