Kyeongmin/C.E.L_Slide_test2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

5단계 파이프라인 전면 재작성 + Figma 추출 계획 업데이트

Browse Source 
DA-12: 1단계 Kei 실장 — 꼭지 2~5개 추출 + 레이어/강조/배치/이미지/표/자세히보기 판단
DA-13: 2단계 디자인 팀장 — catalog 연동 + 블록 매핑 + 공간 배분 + 글자 수 가이드
DA-13b: 3단계 텍스트 편집자 — 글자 수 가이드 참고, 의미 우선 편집 + 자세히보기(요약+상세)
DA-14: 4단계 실무자(AI+코드) + 5단계 팀장 재검토 (균형 점검 → 2차 조정)

문서:
- CLAUDE.md: 5단계 프로세스 + 이미지/표/자세히보기 처리 원칙
- PLAN.md: DA-12~14 태스크 전면 재작성
- PROGRESS.md: 동기화
- FIGMA-COMPONENT-EXTRACTION-PLAN.md: 모드 독립 블록, 변환 규칙, image-block/details-block, MCP, 토큰 매핑

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
이경민
2026-03-25 08:44:10 +09:00
parent 9a780828df
commit 33bd3a56c6
9 changed files with 1015 additions and 397 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
src/content_editor.py
View File

@@ -1,8 +1,8 @@
"""DA-13b: 텍스트 편집자 — 슬롯 텍스트 정리 (Kei 역할).
"""DA-13b: 3단계 — Kei 텍스트 편집자 (텍스트 정리).
디자인 팀장의 레이아웃 컨셉 + 원본 콘텐츠를 받아,
각 슬롯에 맞는 텍스트를 도메인 전문가로서 정리한다.
핵심 내용을 유지하면서 슬롯 분량에 맞게 편집.
팀장의 글자 수 가이드를 참고하되 내용 의미가 우선.
"""
from __future__ import annotations
@@ -18,20 +18,47 @@ from src.design_director import BLOCK_SLOTS
logger = logging.getLogger(__name__)
EDITOR_PROMPT = """당신은 도메인 전문가이자 콘텐츠 편집자이다.
원본 콘텐츠의 핵심 내용을 유지하면서 각 블록의 슬롯에 맞게 텍스트를 정리한다.
## 핵심 원칙
- **내용의 의미와 정확성이 글자 수보다 우선한다**
- 팀장이 제시한 글자 수 가이드는 참고. 의미를 살리려면 가이드를 초과해도 된다.
- 디자인 실무자가 텍스트에 맞게 디자인을 조정할 것이므로, 텍스트를 억지로 자르지 않는다.
## 편집 규칙
- 전체 컨텍스트와 핵심 용어를 보존한다
- 세련된 표현으로 편집한다 (원본 그대로가 아님)
- 개조식(불릿, 번호)으로 작성한다. 줄글 금지.
- 출처가 있는 내용은 출처를 반드시 보존한다
- 출처가 없는 수치나 통계를 만들지 않는다
## 표 편집 규칙
- 표는 표로 유지한다 (다른 형태로 전환하지 않음)
- 팀장이 요약 요청하면 핵심 행/열만 선택하고 "...외 N건" 표기
## 자세히보기 편집 규칙
- detail_target인 꼭지는 두 버전을 작성:
- summary: 슬라이드 표면에 보일 요약 (3줄 이내)
- detail: 펼치면 보일 전체 내용
## JSON 형식으로만 응답한다. 설명 없이 JSON만."""
async def fill_content(
content: str,
layout_concept: dict[str, Any],
analysis: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""각 페이지의 각 블록 슬롯에 텍스트를 채운다.
"""3단계: 각 페이지의 각 블록 슬롯에 텍스트를 채운다.
Args:
content: 원본 텍스트 콘텐츠
layout_concept: 디자인 팀장의 레이아웃 컨셉
{"title": "...", "pages": [{"blocks": [...]}]}
analysis: 1단계 실장의 꼭지 분석 결과 (참고용)
Returns:
슬롯이 채워진 layout_concept (pages[n].blocks[m].data에 텍스트 추가)
슬롯이 채워진 layout_concept
"""
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
@@ -40,51 +67,39 @@ async def fill_content(
if not blocks:
continue
# 슬롯 요구사항 생성
# 블록별 슬롯 + 글자 수 가이드 생성
slot_requirements = []
for i, block in enumerate(blocks):
block_type = block["type"]
block_type = block.get("type", "")
slots = BLOCK_SLOTS.get(block_type, {})
slot_requirements.append(
f"블록 {i+1} ({block_type}, 영역: {block['area']}):\n"
char_guide = block.get("char_guide", {})
req_text = (
f"블록 {i+1} ({block_type}, 영역: {block.get('area', '?')}):\n"
f" 용도: {block.get('reason', '미지정')}\n"
f" 크기: {block.get('size', 'medium')}\n"
f" 필수 슬롯: {slots.get('required', [])}\n"
f" 선택 슬롯: {slots.get('optional', [])}\n"
f" 용도: {block.get('reason', '미지정')}"
f" 선택 슬롯: {slots.get('optional', [])}"
)
system_prompt = (
"당신은 도메인 전문가이자 콘텐츠 편집자이다.\n"
"원본 콘텐츠의 핵심 내용을 유지하면서 각 블록의 슬롯에 맞게 텍스트를 정리한다.\n\n"
"## 규칙\n"
"- 핵심 내용과 맥락을 보존한다. 과도한 요약 금지.\n"
"- 개조식(불릿, 번호)으로 작성한다. 줄글 금지.\n"
"- 출처가 있는 내용은 출처를 보존한다.\n"
"- 출처가 없는 수치나 통계를 만들지 않는다.\n"
"- 각 슬롯의 분량을 지킨다:\n"
" - 제목(title): 최대 30자\n"
" - 본문(content/description): 최대 200자\n"
" - 설명(subtitle/source): 최대 80자\n"
" - 카드 설명: 카드당 최대 150자\n"
"- JSON 형식으로만 응답한다. 설명 없이 JSON만.\n\n"
"## 슬롯 구조 참고\n"
"- comparison: {left_title, left_content, right_title, right_content}\n"
"- card-grid: {cards: [{title, description, category?, source?}]}\n"
"- relationship: {center_label, center_sub?, items: [{label, color?}], description?}\n"
"- process: {steps: [{title, description?, number?}]}\n"
"- quote-block: {quote_text, source?}\n"
"- conclusion-bar: {conclusion_text, label?}\n"
"- comparison-table: {headers: [...], rows: [[...], ...]}\n"
)
if char_guide:
guide_lines = [f" {k}: ~{v}자" for k, v in char_guide.items()]
req_text += "\n 글자 수 가이드 (참고, 의미 우선):\n" + "\n".join(guide_lines)
page_label = f"(페이지 {page_idx + 1}/{len(layout_concept['pages'])})" if len(layout_concept['pages']) > 1 else ""
slot_requirements.append(req_text)
page_label = ""
if len(layout_concept.get("pages", [])) > 1:
page_label = f" (페이지 {page_idx + 1}/{len(layout_concept['pages'])})"
user_prompt = (
f"## 원본 콘텐츠\n{content}\n\n"
f"## 블록 배치 {page_label}\n"
f"## 블록 배치{page_label}\n"
+ "\n".join(slot_requirements)
+ "\n\n## 요청\n"
"위 블록별로 슬롯에 들어갈 텍스트를 정리하여 JSON으로 반환해줘.\n"
"원본의 핵심 내용을 충실하게 반영하되, 각 슬롯 분량에 맞게 편집해.\n"
"내용의 의미를 살려서 편집해. 글자 수 가이드는 참고만.\n"
"자세히보기 대상 블록은 summary + detail 두 버전을 작성해.\n"
"형식:\n"
'{"blocks": [{"area": "...", "type": "...", "data": {슬롯 키-값}}]}'
)
@@ -93,7 +108,7 @@ async def fill_content(
response = await client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
system=system_prompt,
system=EDITOR_PROMPT,
messages=[{"role": "user", "content": user_prompt}],
)
@@ -103,7 +118,7 @@ async def fill_content(
if filled and "blocks" in filled:
for filled_block in filled["blocks"]:
for orig_block in blocks:
if orig_block["area"] == filled_block.get("area"):
if orig_block.get("area") == filled_block.get("area"):
orig_block["data"] = filled_block.get("data", {})
break
@@ -112,7 +127,7 @@ async def fill_content(
f"{len(filled['blocks'])}개 블록"
)
else:
logger.warning(f"텍스트 정리 파싱 실패 (페이지 {page_idx + 1}). 기본값 사용.")
logger.warning(f"텍스트 정리 파싱 실패 (페이지 {page_idx + 1}). 기본값.")
_apply_defaults(blocks)
except Exception as e:
@@ -138,18 +153,20 @@ def _apply_defaults(blocks: list[dict[str, Any]]) -> None:
},
"process": {"steps": []},
"comparison-table": {"headers": [], "rows": []},
"image-block": {"src": "", "alt": "이미지"},
"details-block": {"summary_text": "(상세 내용)", "detail_content": ""},
}
for block in blocks:
if "data" not in block:
block["data"] = defaults.get(block["type"], {})
block["data"] = defaults.get(block.get("type", ""), {})
def _parse_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 JSON을 추출한다."""
patterns = [
r'```json\s*(.*?)```',
r'```\s*(.*?)```',
r'(\{.*\})',
r"```json\s*(.*?)```",
r"```\s*(.*?)```",
r"(\{.*\})",
]
for pattern in patterns:
match = re.search(pattern, text, re.DOTALL)

222
src/design_director.py
View File

@@ -1,14 +1,15 @@
"""DA-13: 디자인 팀장 — 레이아웃 컨셉만 (Sonnet).
"""DA-13: 2단계 — 디자인 팀장 (레이아웃 설계).
Opus의 분류 결과 + 원본 콘텐츠를 받아,
레이아웃 컨셉(블록 배치 + 페이지 수 + 슬롯 목록)만 결정한다.
텍스트 정리는 하지 않는다 — content_editor가 담당.
실장의 꼭지 분석 결과를 받아,
각 꼭지에 적합한 블록을 매핑하고 공간 배분 + 글자 수 가이드를 결정한다.
텍스트 정리는 하지 않는다.
"""
from __future__ import annotations
import json
import logging
import re
from pathlib import Path
from typing import Any
import anthropic
@@ -17,7 +18,7 @@ from src.config import settings
logger = logging.getLogger(__name__)
# 블록별 슬롯 정의 (content_editor에서도 참조)
# 블록별 슬롯 정의 (content_editor, renderer에서도 참조)
BLOCK_SLOTS = {
"comparison": {
"required": ["left_title", "left_content", "right_title", "right_content"],
@@ -47,78 +48,147 @@ BLOCK_SLOTS = {
"required": ["headers", "rows"],
"optional": [],
},
"image-block": {
"required": ["src", "alt"],
"optional": ["caption", "layout"],
},
"details-block": {
"required": ["summary_text", "detail_content"],
"optional": ["label"],
},
}
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")
# fallback: 기본 블록 목록
return """사용 가능한 블록:
- quote-block: 좌측 컬러 라인 + 인용 텍스트. 문제 제기, 핵심 주장할 때.
- card-grid: 2~4열 카드. 용어 정의, 개념 나열할 때.
- comparison: 2단 병렬. A vs B 비교할 때.
- comparison-table: 다항목 비교 테이블. 행/열 많을 때.
- relationship: 벤 다이어그램. 포함/상위-하위 관계할 때.
- process: 단계 흐름. 절차, 워크플로우할 때.
- conclusion-bar: 하단 결론 바. 핵심 한 줄.
- image-block: 이미지 + 캡션. full(전체너비)/side(텍스트옆)/thumb(썸네일) 3변형.
- details-block: 자세히보기. 요약 표면 + 펼치면 상세."""
DIRECTOR_PROMPT = """당신은 디자인 팀장이다. 실장이 분석한 꼭지 목록을 받아 레이아웃을 설계한다.
## 역할
- 각 꼭지에 적합한 블록을 매핑한다
- 전체 공간을 배분하고 겹침을 방지한다
- 각 블록의 글자 수 가이드를 결정한다
- **텍스트는 절대 정리하지 않는다** (텍스트 편집자가 별도로 한다)
## {catalog}
## 이미지 처리 규칙
- 원본 이미지를 그대로 사용한다 (crop 안 함, 크기만 조절)
- 가로형 이미지(비율 > 1.2) → 전체 너비(image-full)
- 세로형 이미지(비율 < 0.8) → 텍스트 옆(image-side)
- 텍스트 포함 도표 → 너무 작게 축소하면 안 됨
## 표 처리 규칙
- 표는 표로 유지한다 (다른 형태로 전환하지 않음)
- 공간에 안 들어가면 → 요약 요청 또는 페이지 분리
## 자세히보기 규칙
- 너무 구체적/세부적인 내용은 details-block으로 설계
- 슬라이드 표면: 요약만, 펼치면: 전체 상세
## 공간 배분 규칙
- CSS grid-template-areas 형식으로 배치
- 영역명: header, left, right, center, main, footer 등
- 꼭지끼리 겹치지 않도록 설계
- 각 블록에 대략적 크기 감(small/medium/large) 제시
## 글자 수 가이드 규칙
- 블록의 공간에 따라 대략적 글자 수 가이드를 제시
- 이것은 하드코딩 기준이 아니라 참고 가이드
- 텍스트 편집자가 의미를 우선하여 가이드와 다를 수 있음
## 출력 형식 (반드시 JSON만. 설명 없이.)
```json
{{
"pages": [
{{
"grid_areas": "'header header' 'left right' 'footer footer'",
"grid_columns": "1fr 1fr",
"grid_rows": "auto 1fr auto",
"blocks": [
{{
"area": "header",
"type": "quote-block",
"topic_id": 1,
"reason": "문제 제기 꼭지",
"size": "small",
"char_guide": {{"quote_text": 80, "source": 30}}
}}
]
}}
]
}}
```"""
async def create_layout_concept(
content: str,
classification: dict[str, Any],
analysis: dict[str, Any],
) -> dict[str, Any]:
"""디자인 팀장이 레이아웃 컨셉을 결정한다.
텍스트는 채우지 않는다. 블록 배치, 페이지 수, 슬롯 목록만 반환.
"""2단계: 디자인 팀장이 레이아웃 컨셉을 설계한다.
Args:
content: 원본 텍스트 콘텐츠
classification: Opus의 분류 결과
content: 원본 텍스트 (분량 참고용)
analysis: 1단계 실장의 꼭지 분석 결과
Returns:
레이아웃 컨셉:
{
"pages": [
{
"grid_areas": "...",
"grid_columns": "...",
"grid_rows": "...",
"blocks": [{"area": "header", "type": "quote-block", "reason": "..."}]
}
],
"title": "슬라이드 제목"
}
{"title": "...", "pages": [{"grid_areas": "...", "blocks": [...]}]}
"""
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
# 기존 분류에서 블록 목록 추출
blocks = classification.get("blocks", [])
block_summary = []
for i, block in enumerate(blocks):
block_summary.append(
f"{i+1}. {block['type']} (영역: {block['area']}) — {block.get('reason', '')}"
)
catalog_text = _load_catalog()
system_prompt = (
"당신은 디자인 팀장이다. 콘텐츠의 구조를 보고 레이아웃 컨셉을 결정한다.\n\n"
"## 역할\n"
"- 블록 배치와 페이지 수만 결정한다\n"
"- 텍스트 내용은 절대 정리하지 않는다 (텍스트 편집자가 별도로 한다)\n\n"
"## 규칙\n"
"- 1페이지에 4~5파트가 적절하다\n"
"- 6파트 이상이면 2페이지로 나눈다\n"
"- 핵심 파트를 억지로 줄이지 않는다\n"
"- CSS grid-template-areas 형식으로 배치를 지정한다\n"
"- JSON 형식으로만 응답한다\n\n"
"## 사용 가능한 블록 타입\n"
"comparison, card-grid, relationship, process, quote-block, conclusion-bar, comparison-table\n\n"
"## 출력 형식\n"
'{"title": "제목", "pages": [{"grid_areas": "...", "grid_columns": "...", "grid_rows": "...", '
'"blocks": [{"area": "...", "type": "...", "reason": "..."}]}]}'
)
# 꼭지 요약
topics_summary = []
for t in analysis.get("topics", []):
line = (
f"꼭지 {t['id']}: {t['title']} "
f"[{t.get('layer', '?')}, 강조:{t.get('emphasis', False)}, "
f"방향:{t.get('direction', '?')}, 유형:{t.get('content_type', 'text')}]"
)
if t.get("image_info"):
line += f" 이미지:{t['image_info']}"
if t.get("table_info"):
line += f" 표:{t['table_info']}"
if t.get("detail_target"):
line += " → 자세히보기 대상"
topics_summary.append(line)
system = DIRECTOR_PROMPT.replace("{catalog}", catalog_text)
user_prompt = (
f"## Opus 실장의 분류 결과\n"
f"제목: {classification.get('title', '')}\n"
f"블록 목록:\n" + "\n".join(block_summary) +
f"\n\n## 원본 콘텐츠 (분량 참고용)\n{content[:2000]}\n\n"
f"## 실장 분석 결과\n"
f"제목: {analysis.get('title', '')}\n"
f"페이지 수: {analysis.get('total_pages', 1)}\n"
f"꼭지 목록:\n" + "\n".join(topics_summary) +
f"\n\n## 원본 콘텐츠 (분량 참고)\n{content[:2000]}\n\n"
f"## 요청\n"
f"위 블록을 몇 페이지에 어떻게 배치할지 결정해줘. "
f"텍스트는 채우지 마. 배치 구조만 JSON으로 반환해."
f"위 꼭지를 어떤 블록으로, 어디에, 몇 페이지로 배치할지 설계해줘.\n"
f"텍스트는 채우지 마. 구조만 JSON으로."
)
try:
response = await client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
system=system_prompt,
system=system,
messages=[{"role": "user", "content": user_prompt}],
)
@@ -128,29 +198,45 @@ async def create_layout_concept(
if concept and "pages" in concept:
total_blocks = sum(len(p.get("blocks", [])) for p in concept["pages"])
logger.info(
f"레이아웃 컨셉 완료: {len(concept['pages'])}페이지, "
f"레이아웃 설계 완료: {len(concept['pages'])}페이지, "
f"{total_blocks}개 블록"
)
return concept
return {
"title": analysis.get("title", "슬라이드"),
**concept,
}
else:
logger.warning("레이아웃 컨셉 파싱 실패. 기존 분류를 1페이지로 사용.")
logger.warning("레이아웃 설계 파싱 실패. fallback 사용.")
except Exception as e:
logger.error(f"디자인 팀장 호출 실패: {e}", exc_info=True)
# fallback: 기존 분류를 1페이지로 감싸기
return _fallback_single_page(classification)
# fallback
return _fallback_layout(analysis)
def _fallback_single_page(classification: dict[str, Any]) -> dict[str, Any]:
"""분류 결과를 1페이지 컨셉으로 변환 (fallback)."""
def _fallback_layout(analysis: dict[str, Any]) -> dict[str, Any]:
"""팀장 실패 시 기본 레이아웃."""
blocks = []
areas = ["header", "main", "footer"]
for i, topic in enumerate(analysis.get("topics", [])[:3]):
area = areas[min(i, len(areas) - 1)]
blocks.append({
"area": area,
"type": "card-grid",
"topic_id": topic.get("id", i + 1),
"reason": topic.get("title", ""),
"size": "medium",
"char_guide": {"title": 20, "description": 100},
})
return {
"title": classification.get("title", "슬라이드"),
"title": analysis.get("title", "슬라이드"),
"pages": [{
"grid_areas": classification.get("grid_areas", "'header' 'main' 'footer'"),
"grid_columns": classification.get("grid_columns", "1fr"),
"grid_rows": classification.get("grid_rows", "auto 1fr auto"),
"blocks": classification.get("blocks", []),
"grid_areas": "'header' 'main' 'footer'",
"grid_columns": "1fr",
"grid_rows": "auto 1fr auto",
"blocks": blocks,
}],
}
@@ -158,9 +244,9 @@ def _fallback_single_page(classification: dict[str, Any]) -> dict[str, Any]:
def _parse_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 JSON을 추출한다."""
patterns = [
r'```json\s*(.*?)```',
r'```\s*(.*?)```',
r'(\{.*\})',
r"```json\s*(.*?)```",
r"```\s*(.*?)```",
r"(\{.*\})",
]
for pattern in patterns:
match = re.search(pattern, text, re.DOTALL)

128
src/kei_client.py
View File

@@ -1,8 +1,7 @@
"""DA-12: Kei API 연동 + Opus 직접 분류.
"""DA-12: 1단계 — Kei 실장 (꼭지 추출 + 분석).
1차: Opus API를 직접 호출하여 콘텐츠 유형을 분류한다 (안정적).
2차: Kei API 연동은 향후 RAG 통합 시 활용.
Opus 실패 시: 수동 분류 fallback.
본문에서 핵심 꼭지를 추출하고, 각 꼭지의 레이어/강조/배치 방향을 분석한다.
이미지/표/상세 콘텐츠도 판단한다.
"""
from __future__ import annotations
@@ -17,45 +16,56 @@ from src.config import settings
logger = logging.getLogger(__name__)
CLASSIFICATION_PROMPT = """당신은 콘텐츠를 분석하여 슬라이드 레이아웃을 결정하는 실장이다.
CLASSIFICATION_PROMPT = """당신은 콘텐츠를 분석하여 슬라이드 구조를 설계하는 실장이다.
## 사용 가능한 블록 타입
- comparison: 2단 병렬 비교 (A vs B)
- card-grid: 카드 배열 (용어 정의, 개념 설명)
- relationship: 벤 다이어그램 (상위/하위, 포함 관계)
- process: 단계 흐름 (절차, 워크플로우)
- quote-block: 강조 인용 (문제 제기, 핵심 메시지)
- conclusion-bar: 결론 바 (핵심 한 줄)
- comparison-table: 다항목 비교 테이블
## 역할
본문에서 핵심 꼭지를 추출하고, 각 꼭지의 성격을 분석하여 슬라이드 구조를 설계한다.
## 배치 영역
grid-template-areas로 정의. 사용 가능한 영역명: header, left, right, center, main, footer
## 꼭지 추출 규칙
- 본문에서 2~5개의 핵심 꼭지(파트)를 추출한다
- 1페이지 적정 꼭지 수: 5개
- 꼭지가 5개를 넘고 중요도가 동등하면 → 2페이지로 분리 (의미 기반 분할)
- 5개인데 내용이 많으면 → 세부 내용은 "자세히보기" 대상으로 표시
## 규칙
- 콘텐츠를 분석하여 각 덩어리의 유형을 판단한다
- 한 슬라이드에 블록 4~6개가 적절하다
- 정보 계층: 위→아래 (문제 제기 → 분석 → 결론)
- 반드시 JSON으로만 응답한다. 설명 없이 JSON만.
## 각 꼭지 분석 항목
1. **레이어 수준**: 도입(문제 제기, 배경) / 핵심(핵심 내용, 정의) / 보조(사례, 근거) / 결론(요약, 핵심 메시지)
2. **강조**: 눈에 띄게 해야 하는 꼭지 표시 (true/false)
3. **배치 방향**: 세로로 긴 내용(vertical) / 가로로 나열(horizontal) / 유연(flexible)
4. **콘텐츠 유형**: text(텍스트) / image(이미지) / table(표) / mixed(혼합)
5. **이미지 정보** (이미지가 있는 경우):
- 핵심인지 보조인지 (core/supplementary)
- 텍스트 포함 여부 (도표/차트는 true)
6. **표 정보** (표가 있는 경우):
- 대략적 행/열 수
- 전체 표시 가능한지 판단
7. **자세히보기 대상**: 너무 구체적/세부적인 내용은 detail_target: true
## 출력 형식
## 출력 형식 (반드시 JSON만. 설명 없이.)
```json
{
"title": "슬라이드 제목",
"grid_areas": "'header header' 'left right' 'footer footer'",
"grid_columns": "1fr 1fr",
"grid_rows": "auto 1fr auto",
"blocks": [
{"area": "header", "type": "quote-block", "reason": "문제 제기"},
{"area": "left", "type": "comparison", "reason": "정책 비교"},
{"area": "right", "type": "card-grid", "reason": "용어 정의 3개"},
{"area": "footer", "type": "conclusion-bar", "reason": "핵심 결론"}
"total_pages": 1,
"topics": [
{
"id": 1,
"title": "꼭지 제목",
"summary": "꼭지 내용 요약 (1~2줄)",
"layer": "intro|core|supporting|conclusion",
"emphasis": true,
"direction": "vertical|horizontal|flexible",
"content_type": "text|image|table|mixed",
"image_info": {"role": "core|supplementary", "has_text": true},
"table_info": {"rows": 5, "cols": 3, "fits_page": true},
"detail_target": false,
"page": 1
}
]
}
```"""
async def classify_content(content: str) -> dict[str, Any] | None:
"""Opus API를 직접 호출하여 콘텐츠를 분류한다.
"""1단계: 본문에서 꼭지를 추출하고 분석한다.
Args:
content: 원본 텍스트 콘텐츠
@@ -75,34 +85,38 @@ async def classify_content(content: str) -> dict[str, Any] | None:
max_tokens=2048,
system=CLASSIFICATION_PROMPT,
messages=[
{"role": "user", "content": f"다음 콘텐츠의 레이아웃을 결정해줘:\n\n{content}"}
{
"role": "user",
"content": f"다음 콘텐츠를 분석하여 꼭지를 추출하고 구조를 설계해줘:\n\n{content}",
}
],
)
result_text = response.content[0].text
layout = _parse_layout_json(result_text)
analysis = _parse_json(result_text)
if layout and "blocks" in layout:
if analysis and "topics" in analysis:
logger.info(
f"콘텐츠 분류 완료: {layout.get('title', 'untitled')}, "
f"{len(layout['blocks'])}개 블록"
f"꼭지 추출 완료: {analysis.get('title', 'untitled')}, "
f"{len(analysis['topics'])}개 꼭지, "
f"{analysis.get('total_pages', 1)}페이지"
)
return layout
return analysis
else:
logger.warning(f"분류 JSON 파싱 실패. 응답: {result_text[:200]}")
return None
except Exception as e:
logger.warning(f"Opus 분류 호출 실패: {e}")
logger.warning(f"실장 분류 호출 실패: {e}")
return None
def _parse_layout_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 레이아웃 JSON을 추출한다."""
def _parse_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 JSON을 추출한다."""
patterns = [
r'```json\s*(.*?)```',
r'```\s*(.*?)```',
r'(\{.*\})',
r"```json\s*(.*?)```",
r"```\s*(.*?)```",
r"(\{.*\})",
]
for pattern in patterns:
match = re.search(pattern, text, re.DOTALL)
@@ -115,27 +129,21 @@ def _parse_layout_json(text: str) -> dict[str, Any] | None:
def manual_classify(content: str) -> dict[str, Any]:
"""Opus 실패 시 기본 레이아웃을 반환하는 fallback."""
"""실장 분류 실패 시 기본 구조를 반환하는 fallback."""
return {
"title": "슬라이드",
"grid_areas": "'header' 'main' 'footer'",
"grid_columns": "1fr",
"grid_rows": "auto 1fr auto",
"blocks": [
"total_pages": 1,
"topics": [
{
"area": "header",
"type": "quote-block",
"reason": "기본 인용 블록",
},
{
"area": "main",
"type": "card-grid",
"reason": "기본 카드 그리드",
},
{
"area": "footer",
"type": "conclusion-bar",
"reason": "기본 결론",
"id": 1,
"title": "핵심 내용",
"summary": content[:100],
"layer": "core",
"emphasis": False,
"direction": "flexible",
"content_type": "text",
"detail_target": False,
"page": 1,
},
],
}

200
src/pipeline.py
View File

@@ -1,16 +1,24 @@
"""DA-14: 전체 파이프라인 (3단계).
"""DA-14: 전체 파이프라인 (5단계).
콘텐츠 입력 → Opus 분류 → 디자인 팀장 컨셉 → 텍스트 편집자 정리 → 렌더러 조립 → HTML 출력.
1. Kei 실장: 꼭지 추출 + 분석
2. 디자인 팀장: 레이아웃 설계
3. 텍스트 편집자: 텍스트 정리
4. 디자인 실무자: HTML 조립
5. 디자인 팀장: 전체 재검토
"""
from __future__ import annotations
import json
import logging
from typing import Any, AsyncIterator
import anthropic
from src.kei_client import classify_content, manual_classify
from src.design_director import create_layout_concept, _fallback_single_page
from src.design_director import create_layout_concept
from src.content_editor import fill_content
from src.renderer import render_slide
from src.config import settings
logger = logging.getLogger(__name__)
@@ -19,53 +27,183 @@ async def generate_slide(
content: str,
manual_layout: dict[str, Any] | None = None,
) -> AsyncIterator[dict[str, str]]:
"""콘텐츠를 슬라이드 HTML로 변환하는 전체 파이프라인.
Args:
content: 원본 텍스트 콘텐츠
manual_layout: 수동 레이아웃 명세 (Opus 대신 사용)
"""콘텐츠를 슬라이드 HTML로 변환하는 5단계 파이프라인.
Yields:
SSE 이벤트:
{"event": "progress", "data": "단계 설명"}
{"event": "result", "data": "완성 HTML"}
{"event": "error", "data": "에러 메시지"}
SSE 이벤트: progress / result / error
"""
try:
# 1단계: Kei 실장 (Opus) — 콘텐츠 분류
yield {"event": "progress", "data": "1/4 Kei 실장이 콘텐츠를 분석 중..."}
# 1단계: Kei 실장 — 꼭지 추출 + 분석
yield {"event": "progress", "data": "1/5 Kei 실장이 꼭지를 추출 중..."}
if manual_layout:
classification = manual_layout
analysis = manual_layout
else:
classification = await classify_content(content)
if classification is None:
classification = manual_classify(content)
analysis = await classify_content(content)
if analysis is None:
analysis = manual_classify(content)
logger.info(f"분류 완료: {len(classification.get('blocks', []))}개 블록")
topic_count = len(analysis.get("topics", []))
page_count = analysis.get("total_pages", 1)
logger.info(f"1단계 완료: {topic_count}개 꼭지, {page_count}페이지")
# 2단계: 디자인 팀장 — 레이아웃 컨셉
yield {"event": "progress", "data": "2/4 디자인 팀장이 레이아웃을 설계 중..."}
# 2단계: 디자인 팀장 — 레이아웃 설계
yield {"event": "progress", "data": "2/5 디자인 팀장이 레이아웃을 설계 중..."}
layout_concept = await create_layout_concept(content, classification)
layout_concept = await create_layout_concept(content, analysis)
total_pages = len(layout_concept.get("pages", []))
total_blocks = sum(len(p.get("blocks", [])) for p in layout_concept.get("pages", []))
logger.info(f"레이아웃 컨셉: {total_pages}페이지, {total_blocks}개 블록")
total_blocks = sum(
len(p.get("blocks", [])) for p in layout_concept.get("pages", [])
)
logger.info(
f"2단계 완료: {len(layout_concept.get('pages', []))}페이지, "
f"{total_blocks}개 블록"
)
# 3단계: 텍스트 편집자 (Kei 역할) — 슬롯 텍스트 정리
yield {"event": "progress", "data": "3/4 텍스트 편집자가 핵심을 정리 중..."}
# 3단계: 텍스트 편집자 — 텍스트 정리
yield {"event": "progress", "data": "3/5 텍스트 편집자가 핵심을 정리 중..."}
layout_concept = await fill_content(content, layout_concept)
layout_concept = await fill_content(content, layout_concept, analysis)
logger.info("3단계 완료: 텍스트 정리")
# 4단계: 실무자 — HTML 렌더링
yield {"event": "progress", "data": "4/4 슬라이드를 조립 중..."}
# 4단계: 디자인 실무자 — HTML 조립
yield {"event": "progress", "data": "4/5 디자인 실무자가 슬라이드를 조립 중..."}
html = render_slide(layout_concept)
logger.info("4단계 완료: HTML 조립")
# 5단계: 디자인 팀장 — 전체 재검토
yield {"event": "progress", "data": "5/5 디자인 팀장이 전체 균형을 검토 중..."}
review_result = await _review_balance(html, layout_concept, content)
if review_result and review_result.get("needs_adjustment"):
logger.info(
f"5단계: 조정 필요 — {review_result.get('issues', [])}"
)
# 조정 지시에 따라 텍스트 재편집 또는 레이아웃 재조정
layout_concept = await _apply_adjustments(
layout_concept, review_result, content
)
html = render_slide(layout_concept)
logger.info("5단계 완료: 2차 조정 반영")
else:
logger.info("5단계 완료: 조정 불필요")
yield {"event": "result", "data": html}
logger.info(f"슬라이드 생성 완료: {total_pages}페이지")
logger.info(f"슬라이드 생성 완료: {len(layout_concept.get('pages', []))}페이지")
except Exception as e:
logger.exception(f"파이프라인 오류: {e}")
yield {"event": "error", "data": str(e)}
async def _review_balance(
html: str,
layout_concept: dict[str, Any],
content: str,
) -> dict[str, Any] | None:
"""5단계: 디자인 팀장이 1차 조립 결과를 재검토한다.
HTML 코드 기반으로 구조적 점검:
- 빈 블록 감지
- 블록 간 채움 비율 불균형
- 이미지/표 크기 적절성
"""
try:
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
# 블록별 텍스트 양 요약
block_summary = []
for page in layout_concept.get("pages", []):
for block in page.get("blocks", []):
data = block.get("data", {})
text_len = len(json.dumps(data, ensure_ascii=False))
block_summary.append(
f" {block.get('area')}/{block.get('type')}: "
f"데이터 {text_len}자"
)
system = (
"당신은 디자인 팀장이다. 1차 조립 결과를 검토하여 균형을 점검한다.\n\n"
"## 점검 항목\n"
"1. 빈 블록: 데이터가 없거나 극히 적은 블록\n"
"2. 채움 불균형: 한 블록은 빽빽하고 다른 블록은 비어있음\n"
"3. 이미지/표: 너무 작거나 큰 것은 없는지\n"
"4. 전체 정보량: 한 페이지에 너무 많거나 적은지\n\n"
"## 출력 형식 (JSON만)\n"
'{"needs_adjustment": true/false, '
'"issues": ["이슈1", "이슈2"], '
'"adjustments": [{"block_area": "...", "action": "expand|shrink|rewrite", "detail": "..."}]}'
)
user_prompt = (
f"## 블록별 데이터 양\n" + "\n".join(block_summary) +
f"\n\n## 레이아웃 구조\n"
f"페이지 수: {len(layout_concept.get('pages', []))}\n"
f"총 블록 수: {sum(len(p.get('blocks', [])) for p in layout_concept.get('pages', []))}\n\n"
f"조정이 필요한가? JSON으로 답해."
)
response = await client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system=system,
messages=[{"role": "user", "content": user_prompt}],
)
result_text = response.content[0].text
return _parse_json(result_text)
except Exception as e:
logger.warning(f"재검토 실패: {e}")
return None
async def _apply_adjustments(
layout_concept: dict[str, Any],
review: dict[str, Any],
content: str,
) -> dict[str, Any]:
"""재검토 결과에 따라 텍스트를 재편집한다."""
adjustments = review.get("adjustments", [])
if not adjustments:
return layout_concept
# 조정이 필요한 블록만 재편집
for adj in adjustments:
area = adj.get("block_area", "")
action = adj.get("action", "")
detail = adj.get("detail", "")
for page in layout_concept.get("pages", []):
for block in page.get("blocks", []):
if block.get("area") == area and action in ("expand", "rewrite"):
# 해당 블록의 char_guide를 조정하여 재편집 유도
if action == "expand":
for key in block.get("char_guide", {}):
block["char_guide"][key] = int(
block["char_guide"][key] * 1.5
)
logger.info(f"조정: {area} → {action} ({detail})")
# 조정된 가이드로 재편집
layout_concept = await fill_content(content, layout_concept)
return layout_concept
def _parse_json(text: str) -> dict[str, Any] | None:
"""텍스트에서 JSON을 추출한다."""
patterns = [
r"```json\s*(.*?)```",
r"```\s*(.*?)```",
r"(\{.*\})",
]
for pattern in patterns:
match = re.search(pattern, text, re.DOTALL)
if match:
try:
return json.loads(match.group(1).strip())
except json.JSONDecodeError:
continue
return None