Initial commit: Kei Design Agent
콘텐츠를 시각적으로 구조화된 슬라이드 HTML로 변환하는 독립 에이전트. 아키텍처 (4단계 파이프라인): 1. Kei 실장 (Opus) — 콘텐츠 유형 분류 + 블록 배치 2. 디자인 팀장 (Sonnet) — 레이아웃 컨셉 (블록 배치 + 페이지 수) 3. 텍스트 편집자 (Sonnet) — 슬롯 텍스트 정리 (핵심 유지) 4. CSS Grid 렌더러 — HTML 조립 블록 템플릿 7종: comparison, card-grid, relationship, process, quote-block, conclusion-bar, comparison-table 기술 스택: FastAPI + Anthropic API + Jinja2 + CSS Grid Pretendard Variable 한국어 폰트 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
0
src/__init__.py
Normal file
0
src/__init__.py
Normal file
20
src/config.py
Normal file
20
src/config.py
Normal file
@@ -0,0 +1,20 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Design Agent 설정."""
|
||||
|
||||
anthropic_api_key: str = ""
|
||||
kei_api_url: str = "http://localhost:8000"
|
||||
log_level: str = "DEBUG"
|
||||
|
||||
# 슬라이드 크기
|
||||
slide_width: int = 1280
|
||||
slide_height: int = 720
|
||||
|
||||
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
|
||||
|
||||
|
||||
settings = Settings()
|
||||
161
src/content_editor.py
Normal file
161
src/content_editor.py
Normal file
@@ -0,0 +1,161 @@
|
||||
"""DA-13b: 텍스트 편집자 — 슬롯 텍스트 정리 (Kei 역할).
|
||||
|
||||
디자인 팀장의 레이아웃 컨셉 + 원본 콘텐츠를 받아,
|
||||
각 슬롯에 맞는 텍스트를 도메인 전문가로서 정리한다.
|
||||
핵심 내용을 유지하면서 슬롯 분량에 맞게 편집.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import anthropic
|
||||
|
||||
from src.config import settings
|
||||
from src.design_director import BLOCK_SLOTS
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def fill_content(
|
||||
content: str,
|
||||
layout_concept: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""각 페이지의 각 블록 슬롯에 텍스트를 채운다.
|
||||
|
||||
Args:
|
||||
content: 원본 텍스트 콘텐츠
|
||||
layout_concept: 디자인 팀장의 레이아웃 컨셉
|
||||
{"title": "...", "pages": [{"blocks": [...]}]}
|
||||
|
||||
Returns:
|
||||
슬롯이 채워진 layout_concept (pages[n].blocks[m].data에 텍스트 추가)
|
||||
"""
|
||||
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
|
||||
|
||||
for page_idx, page in enumerate(layout_concept.get("pages", [])):
|
||||
blocks = page.get("blocks", [])
|
||||
if not blocks:
|
||||
continue
|
||||
|
||||
# 슬롯 요구사항 생성
|
||||
slot_requirements = []
|
||||
for i, block in enumerate(blocks):
|
||||
block_type = block["type"]
|
||||
slots = BLOCK_SLOTS.get(block_type, {})
|
||||
slot_requirements.append(
|
||||
f"블록 {i+1} ({block_type}, 영역: {block['area']}):\n"
|
||||
f" 필수 슬롯: {slots.get('required', [])}\n"
|
||||
f" 선택 슬롯: {slots.get('optional', [])}\n"
|
||||
f" 용도: {block.get('reason', '미지정')}"
|
||||
)
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
page_label = f"(페이지 {page_idx + 1}/{len(layout_concept['pages'])})" if len(layout_concept['pages']) > 1 else ""
|
||||
|
||||
user_prompt = (
|
||||
f"## 원본 콘텐츠\n{content}\n\n"
|
||||
f"## 블록 배치 {page_label}\n"
|
||||
+ "\n".join(slot_requirements)
|
||||
+ "\n\n## 요청\n"
|
||||
"위 블록별로 슬롯에 들어갈 텍스트를 정리하여 JSON으로 반환해줘.\n"
|
||||
"원본의 핵심 내용을 충실하게 반영하되, 각 슬롯 분량에 맞게 편집해.\n"
|
||||
"형식:\n"
|
||||
'{"blocks": [{"area": "...", "type": "...", "data": {슬롯 키-값}}]}'
|
||||
)
|
||||
|
||||
try:
|
||||
response = await client.messages.create(
|
||||
model="claude-sonnet-4-20250514",
|
||||
max_tokens=4096,
|
||||
system=system_prompt,
|
||||
messages=[{"role": "user", "content": user_prompt}],
|
||||
)
|
||||
|
||||
result_text = response.content[0].text
|
||||
filled = _parse_json(result_text)
|
||||
|
||||
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"):
|
||||
orig_block["data"] = filled_block.get("data", {})
|
||||
break
|
||||
|
||||
logger.info(
|
||||
f"텍스트 정리 완료 (페이지 {page_idx + 1}): "
|
||||
f"{len(filled['blocks'])}개 블록"
|
||||
)
|
||||
else:
|
||||
logger.warning(f"텍스트 정리 파싱 실패 (페이지 {page_idx + 1}). 기본값 사용.")
|
||||
_apply_defaults(blocks)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"텍스트 편집자 호출 실패: {e}", exc_info=True)
|
||||
_apply_defaults(blocks)
|
||||
|
||||
return layout_concept
|
||||
|
||||
|
||||
def _apply_defaults(blocks: list[dict[str, Any]]) -> None:
|
||||
"""실패 시 기본 데이터 적용."""
|
||||
defaults = {
|
||||
"quote-block": {"quote_text": "(텍스트 정리 실패)"},
|
||||
"card-grid": {"cards": []},
|
||||
"conclusion-bar": {"conclusion_text": "(결론 생성 실패)"},
|
||||
"comparison": {
|
||||
"left_title": "항목 A", "left_content": "-",
|
||||
"right_title": "항목 B", "right_content": "-",
|
||||
},
|
||||
"relationship": {
|
||||
"center_label": "관계도", "center_sub": "",
|
||||
"items": [], "description": "",
|
||||
},
|
||||
"process": {"steps": []},
|
||||
"comparison-table": {"headers": [], "rows": []},
|
||||
}
|
||||
for block in blocks:
|
||||
if "data" not in block:
|
||||
block["data"] = defaults.get(block["type"], {})
|
||||
|
||||
|
||||
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
|
||||
172
src/design_director.py
Normal file
172
src/design_director.py
Normal file
@@ -0,0 +1,172 @@
|
||||
"""DA-13: 디자인 팀장 — 레이아웃 컨셉만 (Sonnet).
|
||||
|
||||
Opus의 분류 결과 + 원본 콘텐츠를 받아,
|
||||
레이아웃 컨셉(블록 배치 + 페이지 수 + 슬롯 목록)만 결정한다.
|
||||
텍스트 정리는 하지 않는다 — content_editor가 담당.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import anthropic
|
||||
|
||||
from src.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 블록별 슬롯 정의 (content_editor에서도 참조)
|
||||
BLOCK_SLOTS = {
|
||||
"comparison": {
|
||||
"required": ["left_title", "left_content", "right_title", "right_content"],
|
||||
"optional": ["left_subtitle", "right_subtitle"],
|
||||
},
|
||||
"card-grid": {
|
||||
"required": ["cards"],
|
||||
"optional": [],
|
||||
},
|
||||
"relationship": {
|
||||
"required": ["center_label", "items"],
|
||||
"optional": ["center_sub", "description"],
|
||||
},
|
||||
"process": {
|
||||
"required": ["steps"],
|
||||
"optional": [],
|
||||
},
|
||||
"quote-block": {
|
||||
"required": ["quote_text"],
|
||||
"optional": ["source"],
|
||||
},
|
||||
"conclusion-bar": {
|
||||
"required": ["conclusion_text"],
|
||||
"optional": ["label"],
|
||||
},
|
||||
"comparison-table": {
|
||||
"required": ["headers", "rows"],
|
||||
"optional": [],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
async def create_layout_concept(
|
||||
content: str,
|
||||
classification: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""디자인 팀장이 레이아웃 컨셉을 결정한다.
|
||||
|
||||
텍스트는 채우지 않는다. 블록 배치, 페이지 수, 슬롯 목록만 반환.
|
||||
|
||||
Args:
|
||||
content: 원본 텍스트 콘텐츠
|
||||
classification: Opus의 분류 결과
|
||||
|
||||
Returns:
|
||||
레이아웃 컨셉:
|
||||
{
|
||||
"pages": [
|
||||
{
|
||||
"grid_areas": "...",
|
||||
"grid_columns": "...",
|
||||
"grid_rows": "...",
|
||||
"blocks": [{"area": "header", "type": "quote-block", "reason": "..."}]
|
||||
}
|
||||
],
|
||||
"title": "슬라이드 제목"
|
||||
}
|
||||
"""
|
||||
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', '')}"
|
||||
)
|
||||
|
||||
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": "..."}]}]}'
|
||||
)
|
||||
|
||||
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"위 블록을 몇 페이지에 어떻게 배치할지 결정해줘. "
|
||||
f"텍스트는 채우지 마. 배치 구조만 JSON으로 반환해."
|
||||
)
|
||||
|
||||
try:
|
||||
response = await client.messages.create(
|
||||
model="claude-sonnet-4-20250514",
|
||||
max_tokens=2048,
|
||||
system=system_prompt,
|
||||
messages=[{"role": "user", "content": user_prompt}],
|
||||
)
|
||||
|
||||
result_text = response.content[0].text
|
||||
concept = _parse_json(result_text)
|
||||
|
||||
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"{total_blocks}개 블록"
|
||||
)
|
||||
return concept
|
||||
else:
|
||||
logger.warning("레이아웃 컨셉 파싱 실패. 기존 분류를 1페이지로 사용.")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"디자인 팀장 호출 실패: {e}", exc_info=True)
|
||||
|
||||
# fallback: 기존 분류를 1페이지로 감싸기
|
||||
return _fallback_single_page(classification)
|
||||
|
||||
|
||||
def _fallback_single_page(classification: dict[str, Any]) -> dict[str, Any]:
|
||||
"""분류 결과를 1페이지 컨셉으로 변환 (fallback)."""
|
||||
return {
|
||||
"title": classification.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", []),
|
||||
}],
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
141
src/kei_client.py
Normal file
141
src/kei_client.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""DA-12: Kei API 연동 + Opus 직접 분류.
|
||||
|
||||
1차: Opus API를 직접 호출하여 콘텐츠 유형을 분류한다 (안정적).
|
||||
2차: Kei API 연동은 향후 RAG 통합 시 활용.
|
||||
Opus 실패 시: 수동 분류 fallback.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
import anthropic
|
||||
|
||||
from src.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
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
|
||||
|
||||
## 규칙
|
||||
- 콘텐츠를 분석하여 각 덩어리의 유형을 판단한다
|
||||
- 한 슬라이드에 블록 4~6개가 적절하다
|
||||
- 정보 계층: 위→아래 (문제 제기 → 분석 → 결론)
|
||||
- 반드시 JSON으로만 응답한다. 설명 없이 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": "핵심 결론"}
|
||||
]
|
||||
}
|
||||
```"""
|
||||
|
||||
|
||||
async def classify_content(content: str) -> dict[str, Any] | None:
|
||||
"""Opus API를 직접 호출하여 콘텐츠를 분류한다.
|
||||
|
||||
Args:
|
||||
content: 원본 텍스트 콘텐츠
|
||||
|
||||
Returns:
|
||||
분류 결과 JSON. 실패 시 None.
|
||||
"""
|
||||
if not settings.anthropic_api_key:
|
||||
logger.warning("ANTHROPIC_API_KEY 미설정. 수동 분류 모드.")
|
||||
return None
|
||||
|
||||
try:
|
||||
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
|
||||
|
||||
response = await client.messages.create(
|
||||
model="claude-sonnet-4-20250514",
|
||||
max_tokens=2048,
|
||||
system=CLASSIFICATION_PROMPT,
|
||||
messages=[
|
||||
{"role": "user", "content": f"다음 콘텐츠의 레이아웃을 결정해줘:\n\n{content}"}
|
||||
],
|
||||
)
|
||||
|
||||
result_text = response.content[0].text
|
||||
layout = _parse_layout_json(result_text)
|
||||
|
||||
if layout and "blocks" in layout:
|
||||
logger.info(
|
||||
f"콘텐츠 분류 완료: {layout.get('title', 'untitled')}, "
|
||||
f"{len(layout['blocks'])}개 블록"
|
||||
)
|
||||
return layout
|
||||
else:
|
||||
logger.warning(f"분류 JSON 파싱 실패. 응답: {result_text[:200]}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Opus 분류 호출 실패: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _parse_layout_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
|
||||
|
||||
|
||||
def manual_classify(content: str) -> dict[str, Any]:
|
||||
"""Opus 실패 시 기본 레이아웃을 반환하는 fallback."""
|
||||
return {
|
||||
"title": "슬라이드",
|
||||
"grid_areas": "'header' 'main' 'footer'",
|
||||
"grid_columns": "1fr",
|
||||
"grid_rows": "auto 1fr auto",
|
||||
"blocks": [
|
||||
{
|
||||
"area": "header",
|
||||
"type": "quote-block",
|
||||
"reason": "기본 인용 블록",
|
||||
},
|
||||
{
|
||||
"area": "main",
|
||||
"type": "card-grid",
|
||||
"reason": "기본 카드 그리드",
|
||||
},
|
||||
{
|
||||
"area": "footer",
|
||||
"type": "conclusion-bar",
|
||||
"reason": "기본 결론",
|
||||
},
|
||||
],
|
||||
}
|
||||
60
src/main.py
Normal file
60
src/main.py
Normal file
@@ -0,0 +1,60 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pathlib import Path
|
||||
from pydantic import BaseModel
|
||||
from sse_starlette.sse import EventSourceResponse
|
||||
|
||||
from src.config import settings
|
||||
from src.pipeline import generate_slide
|
||||
|
||||
logging.basicConfig(level=getattr(logging, settings.log_level, logging.DEBUG))
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
app = FastAPI(title="Design Agent", version="0.1.0")
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["http://localhost:5174", "http://localhost:5173"],
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# 정적 파일 서빙 (CSS, 폰트 등)
|
||||
static_dir = Path(__file__).parent.parent / "static"
|
||||
if static_dir.exists():
|
||||
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
|
||||
|
||||
|
||||
class SlideRequest(BaseModel):
|
||||
content: str
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
async def health():
|
||||
return {"status": "ok", "service": "design-agent"}
|
||||
|
||||
|
||||
@app.post("/api/generate")
|
||||
async def generate(req: SlideRequest):
|
||||
"""콘텐츠 → 슬라이드 생성 (SSE 스트리밍)."""
|
||||
async def event_stream():
|
||||
async for event in generate_slide(req.content):
|
||||
yield {
|
||||
"event": event["event"],
|
||||
"data": json.dumps(event["data"], ensure_ascii=False),
|
||||
}
|
||||
|
||||
return EventSourceResponse(event_stream())
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def frontend():
|
||||
"""프론트엔드 UI — static/index.html 서빙."""
|
||||
return FileResponse(str(static_dir / "index.html"), media_type="text/html")
|
||||
71
src/pipeline.py
Normal file
71
src/pipeline.py
Normal file
@@ -0,0 +1,71 @@
|
||||
"""DA-14: 전체 파이프라인 (3단계).
|
||||
|
||||
콘텐츠 입력 → Opus 분류 → 디자인 팀장 컨셉 → 텍스트 편집자 정리 → 렌더러 조립 → HTML 출력.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, AsyncIterator
|
||||
|
||||
from src.kei_client import classify_content, manual_classify
|
||||
from src.design_director import create_layout_concept, _fallback_single_page
|
||||
from src.content_editor import fill_content
|
||||
from src.renderer import render_slide
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def generate_slide(
|
||||
content: str,
|
||||
manual_layout: dict[str, Any] | None = None,
|
||||
) -> AsyncIterator[dict[str, str]]:
|
||||
"""콘텐츠를 슬라이드 HTML로 변환하는 전체 파이프라인.
|
||||
|
||||
Args:
|
||||
content: 원본 텍스트 콘텐츠
|
||||
manual_layout: 수동 레이아웃 명세 (Opus 대신 사용)
|
||||
|
||||
Yields:
|
||||
SSE 이벤트:
|
||||
{"event": "progress", "data": "단계 설명"}
|
||||
{"event": "result", "data": "완성 HTML"}
|
||||
{"event": "error", "data": "에러 메시지"}
|
||||
"""
|
||||
try:
|
||||
# 1단계: Kei 실장 (Opus) — 콘텐츠 분류
|
||||
yield {"event": "progress", "data": "1/4 Kei 실장이 콘텐츠를 분석 중..."}
|
||||
|
||||
if manual_layout:
|
||||
classification = manual_layout
|
||||
else:
|
||||
classification = await classify_content(content)
|
||||
if classification is None:
|
||||
classification = manual_classify(content)
|
||||
|
||||
logger.info(f"분류 완료: {len(classification.get('blocks', []))}개 블록")
|
||||
|
||||
# 2단계: 디자인 팀장 — 레이아웃 컨셉
|
||||
yield {"event": "progress", "data": "2/4 디자인 팀장이 레이아웃을 설계 중..."}
|
||||
|
||||
layout_concept = await create_layout_concept(content, classification)
|
||||
|
||||
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}개 블록")
|
||||
|
||||
# 3단계: 텍스트 편집자 (Kei 역할) — 슬롯 텍스트 정리
|
||||
yield {"event": "progress", "data": "3/4 텍스트 편집자가 핵심을 정리 중..."}
|
||||
|
||||
layout_concept = await fill_content(content, layout_concept)
|
||||
|
||||
# 4단계: 실무자 — HTML 렌더링
|
||||
yield {"event": "progress", "data": "4/4 슬라이드를 조립 중..."}
|
||||
|
||||
html = render_slide(layout_concept)
|
||||
|
||||
yield {"event": "result", "data": html}
|
||||
logger.info(f"슬라이드 생성 완료: {total_pages}페이지")
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"파이프라인 오류: {e}")
|
||||
yield {"event": "error", "data": str(e)}
|
||||
171
src/renderer.py
Normal file
171
src/renderer.py
Normal file
@@ -0,0 +1,171 @@
|
||||
"""DA-11: 슬라이드 조합 렌더러.
|
||||
|
||||
블록 배치 명세(JSON)를 받아 Jinja2로 HTML을 생성한다.
|
||||
다중 페이지 지원: pages 배열의 각 페이지를 .slide div로 렌더링.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
|
||||
STATIC_DIR = Path(__file__).parent.parent / "static"
|
||||
|
||||
|
||||
def create_jinja_env() -> Environment:
|
||||
"""Jinja2 환경 생성. templates/ 폴더를 로더로 사용."""
|
||||
return Environment(
|
||||
loader=FileSystemLoader(str(TEMPLATES_DIR)),
|
||||
autoescape=False,
|
||||
)
|
||||
|
||||
|
||||
def render_multi_page(layout_concept: dict[str, Any]) -> str:
|
||||
"""다중 페이지 레이아웃 컨셉으로 완성 HTML을 생성한다.
|
||||
|
||||
Args:
|
||||
layout_concept: 디자인 팀장 + 텍스트 편집자가 완성한 구조:
|
||||
{
|
||||
"title": "슬라이드 제목",
|
||||
"pages": [
|
||||
{
|
||||
"grid_areas": "...",
|
||||
"grid_columns": "...",
|
||||
"grid_rows": "...",
|
||||
"blocks": [{"area": "...", "type": "...", "data": {...}}]
|
||||
},
|
||||
...
|
||||
]
|
||||
}
|
||||
|
||||
Returns:
|
||||
완성된 HTML 문자열 (다중 페이지 시 .slide div 여러 개).
|
||||
"""
|
||||
env = create_jinja_env()
|
||||
title = layout_concept.get("title", "슬라이드")
|
||||
pages = layout_concept.get("pages", [])
|
||||
|
||||
if not pages:
|
||||
logger.warning("페이지가 없습니다. 빈 HTML 반환.")
|
||||
return "<html><body><p>페이지가 없습니다.</p></body></html>"
|
||||
|
||||
# 각 페이지의 블록을 개별 렌더링
|
||||
pages_rendered = []
|
||||
for page_idx, page in enumerate(pages):
|
||||
blocks_rendered = []
|
||||
for block in page.get("blocks", []):
|
||||
block_type = block.get("type", "")
|
||||
block_data = block.get("data", {})
|
||||
template_path = f"blocks/{block_type}.html"
|
||||
|
||||
try:
|
||||
block_template = env.get_template(template_path)
|
||||
rendered_html = block_template.render(**block_data)
|
||||
except Exception as e:
|
||||
logger.warning(f"블록 렌더링 실패 ({block_type}): {e}")
|
||||
rendered_html = f'<div class="body-text">블록 렌더링 실패: {block_type}</div>'
|
||||
|
||||
blocks_rendered.append({
|
||||
"area": block.get("area", "main"),
|
||||
"html": rendered_html,
|
||||
})
|
||||
|
||||
pages_rendered.append({
|
||||
"grid_areas": page.get("grid_areas", "'main'"),
|
||||
"grid_columns": page.get("grid_columns", "1fr"),
|
||||
"grid_rows": page.get("grid_rows", "auto"),
|
||||
"blocks": blocks_rendered,
|
||||
"page_number": page_idx + 1,
|
||||
})
|
||||
|
||||
# base 템플릿 렌더링
|
||||
base_template = env.get_template("slide-base.html")
|
||||
html = base_template.render(
|
||||
slide_title=title,
|
||||
pages=pages_rendered,
|
||||
total_pages=len(pages_rendered),
|
||||
)
|
||||
|
||||
# CSS를 인라인으로 삽입
|
||||
tokens_css = (STATIC_DIR / "tokens.css").read_text(encoding="utf-8")
|
||||
base_css = (STATIC_DIR / "base.css").read_text(encoding="utf-8")
|
||||
base_css = base_css.replace("@import url('./tokens.css');", "")
|
||||
|
||||
inline_css = f"<style>\n{tokens_css}\n{base_css}\n</style>"
|
||||
html = html.replace(
|
||||
'<link rel="stylesheet" href="/static/base.css">',
|
||||
inline_css,
|
||||
)
|
||||
|
||||
logger.info(f"슬라이드 렌더링 완료: {title}, {len(pages_rendered)}페이지")
|
||||
return html
|
||||
|
||||
|
||||
# 하위 호환: 기존 render_slide도 유지
|
||||
def render_slide(layout: dict[str, Any]) -> str:
|
||||
"""기존 단일 페이지 렌더링 (하위 호환).
|
||||
|
||||
pages 구조가 있으면 render_multi_page로 위임.
|
||||
없으면 기존 방식으로 단일 페이지 렌더링.
|
||||
"""
|
||||
if "pages" in layout:
|
||||
return render_multi_page(layout)
|
||||
|
||||
# 기존 단일 페이지 로직
|
||||
env = create_jinja_env()
|
||||
|
||||
blocks_rendered = []
|
||||
for block in layout.get("blocks", []):
|
||||
block_type = block["type"]
|
||||
block_data = block.get("data", {})
|
||||
template_path = f"blocks/{block_type}.html"
|
||||
|
||||
try:
|
||||
block_template = env.get_template(template_path)
|
||||
rendered_html = block_template.render(**block_data)
|
||||
except Exception as e:
|
||||
logger.warning(f"블록 렌더링 실패 ({block_type}): {e}")
|
||||
rendered_html = f'<div class="body-text">블록 렌더링 실패: {block_type}</div>'
|
||||
|
||||
blocks_rendered.append({
|
||||
"area": block["area"],
|
||||
"html": rendered_html,
|
||||
})
|
||||
|
||||
base_template = env.get_template("slide-base.html")
|
||||
html = base_template.render(
|
||||
slide_title=layout.get("title", ""),
|
||||
pages=[{
|
||||
"grid_areas": layout.get("grid_areas", "'header' 'main' 'footer'"),
|
||||
"grid_columns": layout.get("grid_columns", "1fr"),
|
||||
"grid_rows": layout.get("grid_rows", "auto 1fr auto"),
|
||||
"blocks": blocks_rendered,
|
||||
"page_number": 1,
|
||||
}],
|
||||
total_pages=1,
|
||||
)
|
||||
|
||||
tokens_css = (STATIC_DIR / "tokens.css").read_text(encoding="utf-8")
|
||||
base_css = (STATIC_DIR / "base.css").read_text(encoding="utf-8")
|
||||
base_css = base_css.replace("@import url('./tokens.css');", "")
|
||||
|
||||
inline_css = f"<style>\n{tokens_css}\n{base_css}\n</style>"
|
||||
html = html.replace(
|
||||
'<link rel="stylesheet" href="/static/base.css">',
|
||||
inline_css,
|
||||
)
|
||||
|
||||
logger.info(f"슬라이드 렌더링 완료: {layout.get('title', 'untitled')}")
|
||||
return html
|
||||
|
||||
|
||||
def render_standalone_block(block_type: str, data: dict[str, Any]) -> str:
|
||||
"""단일 블록을 독립 HTML로 렌더링 (테스트/미리보기용)."""
|
||||
env = create_jinja_env()
|
||||
template = env.get_template(f"blocks/{block_type}.html")
|
||||
return template.render(**data)
|
||||
Reference in New Issue
Block a user