Phase H 보완 구현: 1단계 A/B 분리 + 정밀 검토 8개 반영
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user