Files
C.E.L_Slide_test2/src/design_director.py
kyeongmin 265d70ed91 refactor(#28): IMP-28 L4 _parse_json dedup (4 modules -> src/json_utils)
Consolidate duplicate _parse_json helpers from content_editor.py /
design_director.py / kei_client.py (fuller form) and pipeline.py (simple form)
into shared src/json_utils.parse_json (strict superset). All 18 call-sites
preserved via `parse_json as _parse_json` alias import; no behavior change.

- src/json_utils.py (new): shared helper, fenced/plain-fence/bare-brace patterns
  + list-prefix cleanup fallback.
- tests/test_json_utils.py (new): 9 unit tests pinning parser semantics.
- src/content_editor.py / design_director.py: remove local helper +
  unused `import json` / `import re`.
- src/kei_client.py / pipeline.py: remove local helper; `json` / `re` retained
  (used elsewhere).

Targeted tests 9 passed; full pytest 374 passed (3 pre-existing scripts/
collection errors reproduce on baseline 909bf75, IMP-28 unrelated).
2026-05-20 20:44:19 +09:00

1068 lines
42 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""DA-13a + DA-13b: 2단계 — 디자인 팀장.
Step A: 레이아웃 프리셋 선택 (규칙 기반, LLM 불필요)
Step B: 프리셋 안에서 블록 매핑 + 글자 수 가이드 (Sonnet)
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
import httpx
import yaml
from src.config import settings
from src.json_utils import parse_json as _parse_json
from src.sse_utils import stream_sse_tokens
logger = logging.getLogger(__name__)
# ──────────────────────────────────────
# 블록별 슬롯 정의
# ──────────────────────────────────────
BLOCK_SLOTS = {
# headers/ (5개)
"section-title-with-bg": {
"required": ["title_ko"], "optional": ["title_en", "breadcrumb", "bg_image"],
"slot_desc": {
"title_ko": "한글 메인 타이틀",
"title_en": "영문 서브 타이틀 (없으면 생략)",
# [legacy Phase R'/Q example — INTEGRATION-AUDIT-01 §10.4]
"breadcrumb": "상위 카테고리 경로 (예: 디지털전환 > BIM)",
"bg_image": "배경 이미지 경로",
},
},
"section-header-bar": {
"required": ["title"], "optional": ["subtitle"],
"slot_desc": {
"title": "섹션 제목 (짧고 굵게)",
"subtitle": "보조 설명 (한 줄)",
},
},
"topic-left-right": {
"required": ["title", "description"], "optional": [],
"slot_desc": {
"title": "꼭지 제목 (좌측, 굵게)",
"description": "꼭지 설명 (우측, 2~3줄)",
},
},
"topic-center": {
"required": ["title"], "optional": ["subtitle", "description"],
"slot_desc": {
"title": "중앙 정렬 대제목",
"subtitle": "부제목 (작은 글씨)",
"description": "추가 설명 (1~2줄)",
},
},
"topic-numbered": {
"required": ["number", "title"], "optional": ["description", "color"],
"slot_desc": {
"number": "순번 (1, 2, 3 등)",
"title": "단계/항목 제목",
"description": "설명 텍스트",
"color": "원형 번호 색상 (CSS 색상값)",
},
},
# cards/ (9개)
"card-image-3col": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "카드 배열. 각 카드: {image: '이미지 경로', title: '제목', title_en: '영문', bullets: ['항목1', '항목2']}. 3장.",
},
},
"card-dark-overlay": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "카드 배열. 각 카드: {image: '배경 이미지', title: '키워드', description: '짧은 설명'}. 3~5장.",
},
},
"card-tag-image": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "카드 배열. 각 카드: {tag: '카테고리 라벨', tag_color: '색상', image: '이미지', title: '제목', description: '설명'}. 3장.",
},
},
"card-icon-desc": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "카드 배열. 각 카드: {icon: '이모지', title: '제목', description: '설명 (2~3줄)'}. 2~4장.",
},
},
"card-compare-3col": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "비교 카드 배열. 각 카드: {header: '카테고리명', header_color: '색상', bullets: ['항목1', '항목2']}. 3장.",
},
},
"card-step-vertical": {
"required": ["steps"], "optional": [],
"slot_desc": {
"steps": "단계 배열. 각 단계: {number: '01', title: '단계명', description: '설명', image: '이미지(선택)'}. 3~5단계.",
},
},
"card-image-round": {
"required": ["cards"], "optional": [],
"slot_desc": {
"cards": "카드 배열. 각 카드: {image: '원형 이미지', title: '제목', description: '설명'}. 2~3장.",
},
},
"card-stat-number": {
"required": ["stats"], "optional": [],
"slot_desc": {
"stats": "통계 배열. 각 항목: {number: '85', unit: '%', label: '비용 절감율'}. 2~4개. 숫자는 출처 있는 것만!",
},
},
"card-numbered": {
"required": ["items"], "optional": [],
"slot_desc": {
"items": "항목 배열. 각 항목: {title: '항목 제목', description: '설명'}. 3~5개.",
},
},
# tables/ (3개)
"compare-3col-badge": {
"required": ["headers", "rows"], "optional": [],
"slot_desc": {
"headers": "3개 열 헤더 배열: ['항목', 'A 대상', 'B 대상']",
"rows": "비교 행 배열. 각 행: {criteria: '비교 기준', left: 'A 내용', right: 'B 내용'}. 최소 3행.",
},
},
"compare-2col-split": {
"required": ["left_title", "right_title", "rows"], "optional": [],
"slot_desc": {
"left_title": "왼쪽 열 헤더",
"right_title": "오른쪽 열 헤더",
"rows": "비교 행 배열. 각 행: {criteria: '비교 기준', left: '왼쪽 내용', right: '오른쪽 내용'}. 최소 3행.",
},
},
"table-simple-striped": {
"required": ["headers", "rows"], "optional": [],
"slot_desc": {
"headers": "열 헤더 배열: ['열1', '열2', '열3']",
"rows": "데이터 행 배열. 각 행: ['셀1', '셀2', '셀3']. 행 수 자유.",
},
},
# visuals/ (6개)
"venn-diagram": {
"required": ["center_label", "items"], "optional": ["center_sub", "description"],
"slot_desc": {
"center_label": "중앙 교집합 라벨 (핵심 키워드)",
"items": "원 배열. 각 원: {label: '영역명', sub: '설명'}. 2~5개.",
"center_sub": "중앙 부가 설명",
"description": "다이어그램 하단 설명",
},
},
"circle-gradient": {
"required": ["label"], "optional": ["sub_label"],
"slot_desc": {
"label": "원 중앙 메인 텍스트 (키워드, 1~2단어)",
"sub_label": "원 아래 보조 텍스트",
},
},
"compare-pill-pair": {
"required": ["left_label", "right_label"], "optional": ["left_sub", "right_sub"],
"slot_desc": {
"left_label": "왼쪽 개념명 (1~2단어)",
"right_label": "오른쪽 개념명 (1~2단어)",
"left_sub": "왼쪽 보조 설명",
"right_sub": "오른쪽 보조 설명",
},
},
"process-horizontal": {
"required": ["steps"], "optional": [],
"slot_desc": {
"steps": "단계 배열. 각 단계: {number: '01', title: '단계명', description: '설명'}. 3~5단계.",
},
},
"flow-arrow-horizontal": {
"required": ["steps"], "optional": [],
"slot_desc": {
"steps": "흐름 배열. 각 항목: {label: '단계명'}. 3~5개. 화살표로 연결됨.",
},
},
"keyword-circle-row": {
"required": ["keywords"], "optional": [],
"slot_desc": {
"keywords": "키워드 배열. 각 항목: {letter: '약어 (G)', label: '풀네임', description: '설명'}. 3~5개.",
},
},
# emphasis/ (10개)
"quote-big-mark": {
"required": ["quote_text"], "optional": ["source"],
"slot_desc": {
"quote_text": "인용할 본문 텍스트 (핵심 발언, 1~3문장)",
"source": "출처 (예: 국토교통부, 2024). 꼭지 제목이 아님!",
},
},
"quote-question": {
"required": ["question"], "optional": ["description"],
"slot_desc": {
"question": "독자에게 던지는 질문 (1문장, 물음표로 끝)",
"description": "질문에 대한 부연 (1~2줄)",
},
},
"comparison-2col": {
"required": ["left_title", "left_content", "right_title", "right_content"],
"optional": ["left_subtitle", "right_subtitle"],
"slot_desc": {
"left_title": "왼쪽 개념 제목 (파란색)",
"left_content": "왼쪽 본문 (불릿 또는 문장)",
"right_title": "오른쪽 개념 제목 (빨간색)",
"right_content": "오른쪽 본문 (불릿 또는 문장)",
"left_subtitle": "왼쪽 보조 제목",
"right_subtitle": "오른쪽 보조 제목",
},
},
"banner-gradient": {
"required": ["text"], "optional": ["sub_text"],
"slot_desc": {
"text": "핵심 결론 한 줄 (굵은 대형 텍스트. 가장 중요한 메시지)",
"sub_text": "부연 설명 (작은 보조 텍스트. text보다 덜 중요)",
},
},
"dark-bullet-list": {
"required": ["bullets"], "optional": ["title"],
"slot_desc": {
"title": "리스트 상단 제목 (파란색, 선택)",
"bullets": "불릿 항목 배열: ['핵심 포인트 1', '핵심 포인트 2']. 3~5개.",
},
},
"highlight-strip": {
"required": ["segments"], "optional": [],
"slot_desc": {
"segments": "색상 구간 배열. 각 구간: {label: '카테고리명', color: '색상'}. 3~5개.",
},
},
"callout-solution": {
"required": ["title", "description"], "optional": ["icon", "source"],
"slot_desc": {
"title": "솔루션/방향성 제목",
"description": "상세 설명 (2~3줄)",
"icon": "아이콘 이모지 (예: 💡)",
"source": "출처 (있으면)",
},
},
"callout-warning": {
"required": ["title", "description"], "optional": ["icon"],
"slot_desc": {
"title": "문제점/경고 제목",
"description": "상세 설명 (2~3줄)",
"icon": "아이콘 이모지 (예: ⚠️)",
},
},
"tab-label-row": {
"required": ["tabs"], "optional": [],
"slot_desc": {
"tabs": "탭 배열. 각 탭: {label: '탭 이름', active: true/false}. 3~5개. 하나만 active.",
},
},
"divider-text": {
"required": ["text"], "optional": [],
"slot_desc": {
"text": "구분선 중앙 텍스트 (짧은 전환 문구, 1~5단어)",
},
},
# media/ (5개)
"image-row-2col": {
"required": ["images"], "optional": [],
"slot_desc": {
"images": "이미지 배열. 각 항목: {src: '이미지 경로', alt: '설명', caption: '캡션'}. 2장.",
},
},
"image-grid-2x2": {
"required": ["images"], "optional": [],
"slot_desc": {
"images": "이미지 배열. 각 항목: {src: '이미지 경로', alt: '설명'}. 4장 (2x2).",
},
},
"image-side-text": {
"required": ["image_src"], "optional": ["image_alt", "title", "description", "bullets"],
"slot_desc": {
"image_src": "좌측 이미지 경로",
"image_alt": "이미지 대체 텍스트",
"title": "우측 제목",
"description": "우측 설명 텍스트",
"bullets": "우측 불릿 항목 배열: ['항목1', '항목2']",
},
},
"image-full-caption": {
"required": ["src"], "optional": ["alt", "caption"],
"slot_desc": {
"src": "전체 너비 이미지 경로",
"alt": "이미지 대체 텍스트",
"caption": "이미지 하단 캡션",
},
},
"image-before-after": {
"required": ["before_src", "after_src"], "optional": ["before_label", "after_label", "caption"],
"slot_desc": {
"before_src": "Before 이미지 경로",
"after_src": "After 이미지 경로",
"before_label": "Before 라벨 (기본: Before)",
"after_label": "After 라벨 (기본: After)",
"caption": "비교 설명 캡션",
},
},
}
# ──────────────────────────────────────
# 슬라이드 물리적 제약
# ──────────────────────────────────────
# 프레임: 1280×720px, 패딩 40px×4 → 가용 1200×640px
# grid gap: 20px, header ~50px, footer ~60px
# → 본문 zone 가용 높이 ≈ 490px (640 - 50 - 20 - 60 - 20)
FRAME_AVAILABLE_HEIGHT = 490
# ──────────────────────────────────────
# 레이아웃 프리셋 정의
# zone별 budget_px = 해당 zone에 넣을 수 있는 최대 높이
# width_pct = zone의 가로 비율 (블록 선택 시 참고)
# ──────────────────────────────────────
LAYOUT_PRESETS = {
"sidebar-right": {
"description": "좌측 본문 흐름 + 우측 참조 사이드바",
"grid_areas": "'header header' 'body sidebar' 'footer footer'",
"grid_columns": "65fr 35fr",
"grid_rows": "auto 1fr auto",
"zones": {
"header": {"desc": "슬라이드 제목. 자동 크기.", "budget_px": 50, "width_pct": 100},
"body": {"desc": "flow 꼭지 배치 (위→아래 순서).", "budget_px": 490, "width_pct": 65},
"sidebar": {"desc": "reference 꼭지. 좁으므로 card-grid 1열, 시각화 블록 금지.", "budget_px": 490, "width_pct": 35},
"footer": {"desc": "결론 꼭지. 전체 너비.", "budget_px": 60, "width_pct": 100},
},
},
"two-column": {
"description": "대등한 2단 비교",
"grid_areas": "'header header' 'left right' 'footer footer'",
"grid_columns": "1fr 1fr",
"grid_rows": "auto 1fr auto",
"zones": {
"header": {"desc": "슬라이드 제목. 자동 크기.", "budget_px": 50, "width_pct": 100},
"left": {"desc": "첫 번째 비교 대상.", "budget_px": 490, "width_pct": 50},
"right": {"desc": "두 번째 비교 대상.", "budget_px": 490, "width_pct": 50},
"footer": {"desc": "결론 꼭지.", "budget_px": 60, "width_pct": 100},
},
},
"hero-detail": {
"description": "고강조 1개 + 보조 상세",
"grid_areas": "'header header' 'hero hero' 'detail detail' 'footer footer'",
"grid_columns": "1fr 1fr",
"grid_rows": "auto 2fr 1fr auto",
"zones": {
"header": {"desc": "슬라이드 제목. 자동 크기.", "budget_px": 50, "width_pct": 100},
"hero": {"desc": "고강조 꼭지 (크게).", "budget_px": 310, "width_pct": 100},
"detail": {"desc": "나머지 보조 꼭지.", "budget_px": 155, "width_pct": 100},
"footer": {"desc": "결론 꼭지.", "budget_px": 60, "width_pct": 100},
},
},
"single-column": {
"description": "단일 컬럼 순차 배치",
"grid_areas": "'header' 'body' 'footer'",
"grid_columns": "1fr",
"grid_rows": "auto 1fr auto",
"zones": {
"header": {"desc": "슬라이드 제목. 자동 크기.", "budget_px": 50, "width_pct": 100},
"body": {"desc": "모든 꼭지 위→아래 순서.", "budget_px": 490, "width_pct": 100},
"footer": {"desc": "결론 꼭지.", "budget_px": 60, "width_pct": 100},
},
},
}
# ──────────────────────────────────────
# Step A: 프리셋 선택 (규칙 기반)
# ──────────────────────────────────────
def select_preset(analysis: dict[str, Any]) -> str:
"""실장의 role 분석을 보고 레이아웃 프리셋을 자동 선택한다.
LLM 호출 불필요. 규칙 기반.
"""
topics = analysis.get("topics", [])
has_reference = any(
t.get("role") == "reference" for t in topics
)
flow_topics = [t for t in topics if t.get("role", "flow") == "flow"]
high_emphasis = [t for t in flow_topics if t.get("emphasis")]
# reference 꼭지가 있으면 sidebar
if has_reference:
preset = "sidebar-right"
# flow 꼭지가 정확히 2개이고 대등 비교이면 two-column
elif (
len(flow_topics) == 2
and all(t.get("layer") == "core" for t in flow_topics)
):
preset = "two-column"
# 고강조 1개 + 나머지가 보조이면 hero
elif (
len(high_emphasis) == 1
and len(flow_topics) >= 3
):
preset = "hero-detail"
# 기본: single-column
else:
preset = "single-column"
logger.info(f"[Step A] 프리셋 선택: {preset}")
return preset
# ──────────────────────────────────────
# Step B: 프리셋 내 블록 매핑 (Sonnet)
# ──────────────────────────────────────
def _get_registered_block_ids() -> set[str]:
"""catalog.yaml에 등록된 블록 ID 집합을 반환한다."""
catalog_path = Path(__file__).parent.parent / "templates" / "catalog.yaml"
if not catalog_path.exists():
return set(BLOCK_SLOTS.keys())
try:
with open(catalog_path, encoding="utf-8") as f:
data = yaml.safe_load(f)
return {
b["id"] for b in data.get("blocks", [])
if b.get("id") and not b.get("id", "").replace("-", "").isdigit()
}
except Exception:
return set(BLOCK_SLOTS.keys())
def _load_catalog() -> str:
"""catalog.yaml 로드."""
catalog_path = Path(__file__).parent.parent / "templates" / "catalog.yaml"
if catalog_path.exists():
return catalog_path.read_text(encoding="utf-8")
return """사용 가능한 블록:
- quote-question: 질문형 강조. 문제 제기, 전환점.
- compare-pill-pair: 2개 키워드 시각 대비.
- comparison-2col: 2항목 비교.
- card-icon-desc: 아이콘+설명 카드.
- card-dark-overlay: 다크 배경 키워드 카드.
- venn-diagram: 벤 다이어그램. 포함/상위-하위 관계.
- process-horizontal: 단계 흐름. 절차.
- topic-left-right: 꼭지 제목+설명.
- banner-gradient: 섹션 강조 배너."""
# Step B(Sonnet) 제거됨 — Phase O에서 Kei 확정 + 코드 검증으로 대체.
# STEP_B_PROMPT, _fallback_layout, PURPOSE_FALLBACK, DOWNGRADE_MAP, _downgrade_fallback 삭제.
# Step B(Sonnet) 제거됨 — Phase O에서 Kei 확정 + 코드 검증으로 대체.
async def _opus_batch_recommend(
analysis: dict[str, Any],
faiss_candidates: dict[int, list[dict]],
container_specs: dict | None = None,
) -> dict[int, str]:
"""Phase P: 전체 topic을 한꺼번에 보여주고 topic별 Opus 추천 1개씩 받는다.
FAISS 후보 2개를 함께 보여주고, Opus가 도메인 지식으로 다른 1개를 추천.
1회 Kei API 호출로 전체 topic 처리.
Returns:
{topic_id: block_type} — 각 topic별 Opus 추천 블록
"""
import httpx
from src.sse_utils import stream_sse_tokens
kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")
# 각 topic의 정보 + FAISS 후보 정리
topic_sections = []
for topic in analysis.get("topics", []):
tid = topic.get("id")
faiss_blocks = faiss_candidates.get(tid, [])
faiss_ids = [b["id"] for b in faiss_blocks]
# 컨테이너 제약 정보
container_info = ""
if container_specs:
from src.space_allocator import find_container_for_topic
spec = find_container_for_topic(tid, container_specs)
if spec:
per_topic = spec.height_px // max(1, len(spec.topic_ids))
container_info = f"컨테이너: {per_topic}px, 허용 height_cost: {spec.max_height_cost} 이하"
topic_sections.append(
f"- 꼭지 {tid}: {topic.get('title', '')}\n"
f" purpose: {topic.get('purpose', '')}\n"
f" relation_type: {topic.get('relation_type', '')}\n"
f" expression_hint: {topic.get('expression_hint', '')}\n"
f" FAISS 후보: {faiss_ids}\n"
f" {container_info}"
)
prompt = (
"아래 각 꼭지에 대해 FAISS가 추천한 블록 2개를 참고하되,\n"
"도메인 지식을 활용하여 **FAISS 후보에 없는 다른 블록 1개**를 추천해줘.\n"
"FAISS 후보와 중복되면 안 된다.\n"
"각 꼭지의 purpose, relation_type, expression_hint를 보고\n"
"**콘텐츠의 의미와 목적에 가장 적합한** 블록을 추천하라.\n"
"컨테이너 크기 제약도 반드시 고려하라.\n\n"
f"## 꼭지 목록\n" + "\n".join(topic_sections) +
"\n\n## 출력 (JSON만)\n"
'{"recommendations": [{"topic_id": 1, "block_type": "...", "reason": "..."}]}'
)
try:
async with httpx.AsyncClient(timeout=None) as client:
async with client.stream(
"POST",
f"{kei_url}/api/direct",
json={
"message": prompt,
},
timeout=None,
) as response:
if response.status_code != 200:
logger.warning(f"[Phase P] Opus 배치 추천 HTTP {response.status_code}")
return {}
full_text = await stream_sse_tokens(response)
if not full_text:
return {}
result = _parse_json(full_text)
if result and "recommendations" in result:
mapping = {}
for rec in result["recommendations"]:
tid = rec.get("topic_id") or rec.get("id")
if tid is not None:
mapping[tid] = rec.get("block_type", "")
logger.info(f"[Phase P] Opus 배치 추천: {mapping}")
return mapping
logger.warning(f"[Phase P] Opus 배치 추천 JSON 파싱 실패: {full_text[:200]}")
return {}
except Exception as e:
logger.warning(f"[Phase P] Opus 배치 추천 실패: {e}")
return {}
async def _opus_block_recommendation(
analysis: dict[str, Any],
block_candidates: str,
preset_name: str,
preset: dict[str, Any],
container_specs: dict | None = None,
) -> dict[str, Any] | None:
"""Phase O: Kei(Opus)가 컨테이너 제약을 보고 블록을 확정한다.
Kei API를 통해 Opus가 사고하여:
- 컨테이너 크기(px)에 맞는 블록 선정
- height_cost가 컨테이너보다 큰 블록은 선택 금지
- 도메인 지식 기반 판단
반드시 Kei API 경유. Anthropic 직접 호출 절대 금지.
"""
import httpx
kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")
zone_desc = "\n".join(
f"- {name}: {z['desc']} [높이: ~{z['budget_px']}px, 너비: {z['width_pct']}%]"
for name, z in preset["zones"].items()
)
topics_text = "\n".join(
f"- 꼭지 {t.get('id', '?')}: {t.get('title', '')} "
f"[{t.get('layer', '?')}, {t.get('role', 'flow')}, 강조:{t.get('emphasis', False)}]"
for t in analysis.get("topics", [])
)
# Phase O: 컨테이너 제약 텍스트
container_text = ""
if container_specs:
from src.space_allocator import ContainerSpec
lines = ["## 컨테이너 제약 (반드시 준수)\n각 꼭지는 아래 컨테이너 안에 들어가야 한다. height_cost가 허용 범위를 초과하면 선택 금지.\n"]
for role, spec in container_specs.items():
for tid in spec.topic_ids:
lines.append(
f"- 꼭지 {tid}: 컨테이너 {spec.height_px}px × {spec.width_px}px, "
f"허용 height_cost: **{spec.max_height_cost} 이하**, "
f"최대 항목 수: {spec.block_constraints.get('max_items', '?')}"
)
container_text = "\n".join(lines) + "\n\n"
prompt = (
f"슬라이드 디자인 블록 추천을 해줘.\n\n"
f"## 프리셋: {preset_name}\n{preset['description']}\n\n"
f"## Zone 구조 (반드시 이 zone에 배정하라)\n{zone_desc}\n\n"
f"## Zone 배정 규칙 (절대 규칙)\n"
f"- flow 꼭지 → body / left / hero zone\n"
f"- reference 꼭지 → sidebar zone\n"
f"- conclusion 꼭지 → **반드시 footer zone** (banner-gradient 권장)\n"
f"- sidebar(35%)에는 시각화 블록 금지\n\n"
f"{container_text}"
f"## 꼭지 목록\n{topics_text}\n\n"
f"## 블록 후보 (FAISS 검색 결과)\n{block_candidates}\n\n"
f"## 요청\n"
f"각 꼭지에 가장 적합한 블록을 추천해줘.\n"
f"컨테이너 높이(px)와 허용 height_cost를 반드시 확인하고,\n"
f"도메인 지식을 활용하여 콘텐츠 성격에 맞는 블록을 선택해.\n\n"
f"## 출력 형식 (JSON만)\n"
f'{{"recommendations": ['
f'{{"topic_id": 1, "block_type": "...", "area": "...", '
f'"reason": "도메인 관점에서 이 블록이 적합한 이유", '
f'"size_guide": "compact|medium|large"}}]}}'
)
try:
async with httpx.AsyncClient(timeout=None) as client:
async with client.stream(
"POST",
f"{kei_url}/api/direct",
json={
"message": prompt,
},
timeout=None,
) as response:
if response.status_code != 200:
logger.warning(f"[Step A-2] Kei API HTTP {response.status_code}")
return None
full_text = await stream_sse_tokens(response)
if not full_text:
logger.warning("[Step A-2] Kei API 응답 텍스트 없음")
return None
result = _parse_json(full_text)
if result and "recommendations" in result:
logger.info(
f"[Step A-2] Opus 블록 추천 완료: "
f"{len(result['recommendations'])}"
)
return result
logger.warning(f"[Step A-2] JSON 파싱 실패: {full_text[:200]}")
return None
except Exception as e:
logger.warning(f"[Step A-2] Kei API 호출 실패: {e}")
return None
async def create_layout_concept(
content: str,
analysis: dict[str, Any],
container_specs: dict | None = None,
) -> dict[str, Any]:
"""2단계: Step A(프리셋) + Step B(블록 매핑).
Args:
content: 원본 텍스트
analysis: 1단계 실장의 꼭지 분석 결과
Returns:
레이아웃 컨셉 JSON
"""
# Step A: 프리셋 선택 (규칙 기반)
preset_name = select_preset(analysis)
preset = LAYOUT_PRESETS[preset_name]
# P2-A: FAISS 검색으로 관련 블록만 추출
from src.block_search import search_blocks_for_topics
topics = analysis.get("topics", [])
catalog_text = search_blocks_for_topics(topics, top_k_per_topic=3, total_max=10)
logger.info(f"[Step A] 블록 후보 검색 완료 (FAISS)")
# Phase N-1: Step A-2 — Kei(Opus)가 블록 확정. Sonnet은 zone + char_guide만.
opus_recommendation = await _opus_block_recommendation(
analysis, catalog_text, preset_name, preset,
container_specs=container_specs,
)
# Kei 확정 블록 매핑 (topic_id → block_type)
kei_confirmed_blocks: dict[int, str] = {}
kei_confirmed_areas: dict[int, str] = {}
if opus_recommendation and opus_recommendation.get("recommendations"):
recs = opus_recommendation["recommendations"]
for rec in recs:
# Kei가 topic_id 또는 id로 응답할 수 있으므로 양쪽 체크
tid = rec.get("topic_id") or rec.get("id")
if tid is not None:
kei_confirmed_blocks[tid] = rec.get("block_type", "")
kei_confirmed_areas[tid] = rec.get("area", "")
logger.info(f"[Step A-2] Kei 블록 확정: {kei_confirmed_blocks}")
else:
# Kei API 필수. 응답 없으면 성공할 때까지 무한 재시도.
import asyncio
RETRY_INTERVAL = 10
attempt = 0
while not opus_recommendation or not opus_recommendation.get("recommendations"):
attempt += 1
logger.warning(f"[Step A-2] Kei API 응답 없음 (시도 {attempt}). {RETRY_INTERVAL}초 후 재시도...")
await asyncio.sleep(RETRY_INTERVAL)
opus_recommendation = await _opus_block_recommendation(
analysis, catalog_text, preset_name, preset
)
# 재시도 성공 → 확정 블록 매핑
for rec in opus_recommendation["recommendations"]:
tid = rec.get("topic_id") or rec.get("id")
if tid is not None:
kei_confirmed_blocks[tid] = rec.get("block_type", "")
kei_confirmed_areas[tid] = rec.get("area", "")
logger.info(f"[Step A-2] Kei 블록 확정 (재시도 후): {kei_confirmed_blocks}")
# Phase O: Kei 확정 블록 + 코드 검증으로 직접 layout_concept 생성
# Step B(Sonnet) 제거됨 — Kei가 블록/zone을 확정, 코드가 스펙 계산
blocks = []
registered_ids = _get_registered_block_ids()
valid_zones = {z for z in preset["zones"] if z != "header"}
default_zone = "body" if "body" in valid_zones else next(iter(valid_zones))
for topic in topics:
tid = topic.get("id")
role = topic.get("role", "flow")
# 블록 타입: Kei 확정값
block_type = kei_confirmed_blocks.get(tid, "topic-left-right")
# 블록 ID 검증: catalog에 없으면 에러 로그 (fallback 없음)
if block_type not in registered_ids:
logger.error(f"[블록 검증] Kei 확정 블록 '{block_type}'이 catalog에 없음. topic {tid}")
block_type = "topic-left-right" # 최소 안전 블록
# zone 배치: Kei 확정값 → 검증
area = kei_confirmed_areas.get(tid, "")
if not area or area not in valid_zones:
# Kei가 area를 안 줬으면 role에서 결정
if role == "reference" and "sidebar" in valid_zones:
area = "sidebar"
elif topic.get("layer") == "conclusion" and "footer" in valid_zones:
area = "footer"
else:
area = default_zone
# conclusion 꼭지 → footer 강제
if topic.get("layer") == "conclusion" and "footer" in valid_zones:
area = "footer"
# body/sidebar 금지 블록 검증
if area in ("body", "left", "right", "hero", "detail") and block_type in BODY_FORBIDDEN_MAP:
replacement = BODY_FORBIDDEN_MAP[block_type]
if replacement:
logger.warning(f"[블록 검증] body 금지 '{block_type}''{replacement}'")
block_type = replacement
else:
continue # None이면 삭제
if area == "sidebar" and block_type in SIDEBAR_FORBIDDEN_BLOCKS:
replacement = SIDEBAR_FORBIDDEN_BLOCKS[block_type]
if replacement:
logger.warning(f"[블록 검증] sidebar 금지 '{block_type}''{replacement}'")
block_type = replacement
else:
continue
blocks.append({
"area": area,
"type": block_type,
"topic_id": tid,
"purpose": topic.get("purpose", ""),
"reason": kei_confirmed_blocks.get(tid, ""),
"size": "medium",
})
# Phase N-2: sidebar에 reference 블록이 있으면 section label 자동 삽입
sidebar_blocks = [b for b in blocks if b.get("area") == "sidebar"]
if sidebar_blocks:
first_sidebar = sidebar_blocks[0]
sidebar_topic = next(
(t for t in topics if t.get("id") == first_sidebar.get("topic_id")),
None,
)
section_title = ""
if sidebar_topic:
section_title = sidebar_topic.get("section_title", "")
if not section_title:
purpose = first_sidebar.get("purpose", "")
section_title = {
"용어정의": "용어 정의",
"근거사례": "참고 자료",
}.get(purpose, "")
if section_title:
first_sidebar_idx = next(
i for i, b in enumerate(blocks) if b.get("area") == "sidebar"
)
blocks.insert(first_sidebar_idx, {
"area": "sidebar",
"type": "divider-text",
"topic_id": None,
"purpose": "_label",
"data": {"text": section_title},
"size": "compact",
"_is_label": True,
})
logger.info(f"[N-2] sidebar 섹션 제목 삽입: '{section_title}'")
# zone별 height_cost 합산 검증
overflows = _validate_height_budget(blocks, preset)
logger.info(
f"[레이아웃] 블록 배치 완료: {preset_name}, {len(blocks)}개 블록"
+ (f", overflow {len(overflows)}" if overflows else "")
)
result = {
"title": analysis.get("title", "슬라이드"),
"pages": [{
"grid_areas": preset["grid_areas"],
"grid_columns": preset["grid_columns"],
"grid_rows": preset["grid_rows"],
"blocks": blocks,
}],
}
if overflows:
result["overflow"] = overflows
return result
# height_cost → px 변환 (결정론적)
HEIGHT_COST_PX = {
"compact": 70,
"medium": 150,
"large": 250,
"xlarge": 400,
}
# body/sidebar/footer zone에서 사용 금지인 블록 → 교체
BODY_FORBIDDEN_MAP = {
"section-title-with-bg": "topic-center", # 500px 블록 → compact 헤더로
"section-header-bar": None, # body에서 제거 — header에 이미 slide-title 있음 (J-2)
}
# Phase M: 블록-zone 적합성 맵
# sidebar(35% 너비)에서 사용 불가한 블록 → 대체 블록
SIDEBAR_FORBIDDEN_BLOCKS = {
"card-compare-3col": "card-numbered",
"card-dark-overlay": "card-numbered",
"card-icon-desc": "card-numbered",
"card-image-3col": "card-numbered",
"card-image-round": "card-numbered",
"card-stat-number": "card-numbered",
"card-tag-image": "card-numbered",
"comparison-2col": "dark-bullet-list",
"compare-2col-split": "dark-bullet-list",
"compare-pill-pair": "dark-bullet-list",
"section-title-with-bg": None,
"section-header-bar": None,
"topic-center": "topic-left-right",
"quote-big-mark": "quote-question",
"image-full-caption": "image-row-2col",
}
def _get_block_height(block_type: str) -> int:
"""블록 타입의 height_cost를 px로 반환."""
catalog_map = _load_catalog_map_for_height()
cost_label = catalog_map.get(block_type, "medium")
return HEIGHT_COST_PX.get(cost_label, 150)
def _load_catalog_map_for_height() -> dict[str, str]:
"""catalog.yaml에서 id → height_cost 매핑을 로드."""
catalog_path = Path(__file__).parent.parent / "templates" / "catalog.yaml"
if not catalog_path.exists():
return {}
try:
with open(catalog_path, encoding="utf-8") as f:
data = yaml.safe_load(f)
return {b["id"]: b.get("height_cost", "medium") for b in data.get("blocks", [])}
except Exception:
return {}
def _load_catalog_purpose_fit() -> dict[str, list[str]]:
"""catalog.yaml에서 id → purpose_fit 매핑을 로드."""
catalog_path = Path(__file__).parent.parent / "templates" / "catalog.yaml"
if not catalog_path.exists():
return {}
try:
with open(catalog_path, encoding="utf-8") as f:
data = yaml.safe_load(f)
return {
b["id"]: b.get("purpose_fit", [])
for b in data.get("blocks", [])
}
except Exception:
return {}
def _validate_purpose_fit(blocks: list[dict]) -> int:
"""각 블록의 purpose_fit을 검증하고, 불일치 시 대체한다.
Returns:
교체된 블록 수.
"""
purpose_fit_map = _load_catalog_purpose_fit()
replaced = 0
for block in blocks:
block_type = block.get("type", "")
purpose = block.get("purpose", "")
if not block_type or not purpose:
continue
allowed_purposes = purpose_fit_map.get(block_type, [])
# purpose_fit이 빈 리스트면 범용 블록 → 검증 스킵
if not allowed_purposes:
continue
if purpose not in allowed_purposes:
# Kei가 확정한 블록이므로 경고만 출력. 강제 교체 안 함.
logger.warning(
f"[purpose_fit 검증] '{block_type}'의 purpose_fit={allowed_purposes}"
f"'{purpose}' 없음 — Kei 확정이므로 유지"
)
return replaced
def _validate_height_budget(blocks: list[dict], preset: dict) -> list[dict]:
"""zone별 height_cost 합산을 검증한다. (I-9 개정)
금지 블록 교체, pill-pair 단독 검증은 수행하되,
높이 초과 시 블록을 자동 교체하지 않는다.
대신 overflow 정보를 수집하여 반환 → pipeline에서 Kei에게 판단 요청.
Returns:
overflow 정보 리스트. 초과 없으면 빈 리스트.
"""
zones = preset.get("zones", {})
gap_px = 20 # --spacing-block
# zone별 블록 그룹핑
zone_blocks: dict[str, list[dict]] = {}
for block in blocks:
area = block.get("area", "body")
if area not in zone_blocks:
zone_blocks[area] = []
zone_blocks[area].append(block)
# 금지 블록 처리: 교체 또는 삭제 (J-2: None이면 삭제)
blocks_to_remove = []
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]
if replacement is None:
blocks_to_remove.append(block)
logger.warning(
f"[금지 블록 삭제] {block_type} (area={area})"
)
else:
block["type"] = replacement
logger.warning(
f"[금지 블록 교체] {block_type}{replacement} (area={area})"
)
for block in blocks_to_remove:
blocks.remove(block)
# [legacy Phase R'/Q example — INTEGRATION-AUDIT-01 §10.4]
# 삭제 후 zone_blocks 재구성 (후속 pill-pair/높이 체크에 반영)
zone_blocks.clear()
for block in blocks:
area = block.get("area", "body")
if area not in zone_blocks:
zone_blocks[area] = []
zone_blocks[area].append(block)
# Phase M: sidebar 블록-zone 적합성 검증 (P-6)
for block in blocks:
if block.get("area") == "sidebar" and block.get("type") in SIDEBAR_FORBIDDEN_BLOCKS:
replacement = SIDEBAR_FORBIDDEN_BLOCKS[block["type"]]
if replacement is None:
logger.warning(f"[zone 적합성] sidebar에서 {block['type']} 삭제")
else:
logger.warning(f"[zone 적합성] sidebar: {block['type']}{replacement}")
block["type"] = replacement
# sidebar 카드 블록 1열 강제 (J-6)
CARD_BLOCKS = {
"card-tag-image", "card-icon-desc", "card-image-3col",
"card-dark-overlay", "card-compare-3col", "card-image-round",
"card-stat-number",
}
for block in blocks:
if block.get("area") == "sidebar" and block.get("type") in CARD_BLOCKS:
if "data" not in block:
block["data"] = {}
block["data"]["column_override"] = 1
# compare-pill-pair 단독 사용 금지 (I-7)
COMPARISON_BLOCKS = {"compare-2col-split", "compare-3col-badge", "comparison-2col"}
for area, area_blocks in zone_blocks.items():
types = {b.get("type") for b in area_blocks}
if "compare-pill-pair" in types and not types & COMPARISON_BLOCKS:
for block in area_blocks:
if block.get("type") == "compare-pill-pair":
block["type"] = "comparison-2col"
logger.warning(
"[pill-pair 단독 금지] compare-pill-pair → comparison-2col"
)
# 높이 예산 검증 — 초과 시 자동 조치 + overflow 정보 수집
overflows: list[dict] = []
for area, area_blocks in zone_blocks.items():
zone_info = zones.get(area, {})
budget = zone_info.get("budget_px", 490)
total = sum(_get_block_height(b.get("type", "")) for b in area_blocks)
total += gap_px * max(0, len(area_blocks) - 1)
if total <= budget:
continue
overflow_px = total - budget
# footer 초과 자동 조치: banner-gradient의 sub_text 제거로 높이 축소
if area == "footer" and overflow_px <= 30:
for block in area_blocks:
if block.get("type") == "banner-gradient":
if "data" not in block:
block["data"] = {}
block["data"]["_strip_sub_text"] = True
logger.info(
f"[높이 자동 조치] footer 초과 {overflow_px}px → "
f"banner-gradient sub_text 제거"
)
# sub_text 제거 시 compact(50px)로 줄어들므로 재계산
total_after = sum(
50 if (b.get("type") == "banner-gradient" and b.get("data", {}).get("_strip_sub_text"))
else _get_block_height(b.get("type", ""))
for b in area_blocks
)
total_after += gap_px * max(0, len(area_blocks) - 1)
if total_after <= budget:
continue # 조치 후 예산 이내 → overflow 아님
logger.warning(
f"[높이 예산 초과] {area}: {total}px > {budget}px. "
f"블록: {[b.get('type') for b in area_blocks]}"
)
overflows.append({
"area": area,
"overflow_px": total - budget,
"budget_px": budget,
"total_px": total,
"blocks": [
{
"type": b.get("type", ""),
"purpose": b.get("purpose", ""),
"topic_id": b.get("topic_id"),
"height_px": _get_block_height(b.get("type", "")),
}
for b in area_blocks
],
})
return overflows