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

Phase S: Claude HTML 직접 생성 + 독립 검증 시스템 도입

Browse Source 
블록 선택 방식(Phase P/Q/R) 폐기 → Claude Sonnet이 영역별 HTML 직접 생성.
생성-검증 분리: content_verifier.py로 텍스트 보존/금지 콘텐츠/구조를 코드 검증.

주요 변경:
- src/html_generator.py: 4개 프롬프트 템플릿(BG/CORE/SIDEBAR/FOOTER) + 영역별 Claude 호출
- src/content_verifier.py: L1 텍스트 보존, L2 금지 콘텐츠, L3 구조 검증 + 재시도 루프
- src/html_validator.py: 보안 검증(script/iframe 제거)
- src/renderer.py: render_slide_from_html() 추가, area div overflow:hidden
- scripts/test_phase_s.py: generate_with_retry() 통합, step2b_verification 결과 저장
- 배경 라이트 디자인(#f8fafc), 개조식 어미 변환, 축약 금지 규칙

다음 과제: 폰트 위계(핵심14>본문12>배경10-12>첨부9-11) + 동적 컨테이너 계산

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
이경민
2026-03-31 08:37:05 +09:00
parent 9410576e60
commit 0e4b8c091c
14 changed files with 3875 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

343
src/pipeline.py
View File

@@ -17,42 +17,53 @@ from typing import Any, AsyncIterator
import anthropic
from src.kei_client import classify_content, refine_concepts, call_kei_overflow_judgment, call_kei_final_review
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
from src.kei_client import classify_content, refine_concepts
from src.design_director import LAYOUT_PRESETS, select_preset
from src.image_utils import get_image_sizes, embed_images
from src.space_allocator import calculate_container_specs, finalize_block_specs, find_container_for_topic, calculate_trim_chars
from src.slide_measurer import measure_rendered_heights, format_measurement_for_kei, capture_slide_screenshot
from src.space_allocator import calculate_container_specs
from src.slide_measurer import measure_rendered_heights, capture_slide_screenshot
from src.config import settings
logger = logging.getLogger(__name__)
# Kei API 재시도 간격(초). 제한 없음 — 성공할 때까지 무한 재시도.
# Kei API 재시도 설정 (P0 수정: 무한 루프 방지)
KEI_RETRY_INTERVAL = 10
KEI_MAX_RETRY_ATTEMPTS = 30 # 최대 30회 (5분)
KEI_MAX_RETRY_DURATION = 300 # 절대 제한 300초
async def _retry_kei(fn, *args, **kwargs):
"""Kei API 호출을 성공할 때까지 무한 재시도한다.
"""Kei API 호출을 성공할 때까지 재시도한다.
Kei API는 필수 인프라. fallback 없음. 제한 없음.
10분이든 20분이든 Kei가 응답할 때까지 기다린다.
Kei API는 필수 인프라. fallback 없음.
최대 30회 또는 300초까지 재시도 후 TimeoutError.
"""
import asyncio
attempt = 0
while True:
start_time = time.time()
while attempt < KEI_MAX_RETRY_ATTEMPTS:
attempt += 1
elapsed = time.time() - start_time
if elapsed > KEI_MAX_RETRY_DURATION:
raise TimeoutError(
f"Kei API 타임아웃: {fn.__name__} — "
f"{elapsed:.0f}초 경과 (제한 {KEI_MAX_RETRY_DURATION}초)"
)
result = await fn(*args, **kwargs)
if result is not None:
if attempt > 1:
logger.info(f"[Kei 재시도] {fn.__name__} 성공 ({attempt}번째 시도)")
return result
logger.warning(
f"[Kei 재시도] {fn.__name__} 실패 (시도 {attempt}). "
f"[Kei 재시도] {fn.__name__} 실패 (시도 {attempt}/{KEI_MAX_RETRY_ATTEMPTS}). "
f"{KEI_RETRY_INTERVAL}초 후 재시도..."
)
await asyncio.sleep(KEI_RETRY_INTERVAL)
raise RuntimeError(
f"Kei API 최대 재시도 초과: {fn.__name__} — {attempt}회 시도"
)
def _save_step(run_dir: Path, filename: str, data: Any) -> None:
"""스텝 결과를 JSON 또는 HTML로 저장한다. (K-1)"""
@@ -150,238 +161,110 @@ async def generate_slide(
for role, spec in container_specs.items()
})
# 2단계: 디자인 팀장 — Step A(프리셋) + Step A-2(Kei 블록 확정) + Step B(zone 배치)
yield {"event": "progress", "data": "2/5 디자인 팀장이 레이아웃을 설계 중..."}
# ★ Phase S: Claude Sonnet이 HTML 직접 생성
# 블록 선택 없음. 슬롯 채우기 없음. AI가 콘텐츠에 맞는 HTML 구조를 직접 만든다.
yield {"event": "progress", "data": "2/4 슬라이드 HTML 생성 중..."}
layout_concept = await create_layout_concept(content, analysis, container_specs=container_specs)
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}개 블록"
)
_save_step(run_dir, "step2_layout.json", {
"preset": layout_concept.get("pages", [{}])[0].get("grid_areas", ""),
"blocks": [
{
"area": b.get("area"), "type": b.get("type"),
"topic_id": b.get("topic_id"), "purpose": b.get("purpose"),
"reason": b.get("reason", ""), "size": b.get("size", ""),
}
for p in layout_concept.get("pages", [])
for b in p.get("blocks", [])
],
"overflow": layout_concept.get("overflow", []),
})
# ★ Phase O-3: 블록 스펙 확정 (컨테이너 크기 → 항목수/글자수/폰트)
for page in layout_concept.get("pages", []):
finalize_block_specs(page.get("blocks", []), container_specs)
# 컨테이너 스펙을 layout_concept에 저장 (렌더러에서 사용)
layout_concept["_container_specs"] = container_specs
_save_step(run_dir, "step2c_block_specs.json", {
"blocks": [
{
"type": b.get("type"), "topic_id": b.get("topic_id"),
"area": b.get("area"),
"_container_height_px": b.get("_container_height_px"),
"_max_items": b.get("_max_items"),
"_max_chars_per_item": b.get("_max_chars_per_item"),
"_max_chars_total": b.get("_max_chars_total"),
"_font_size_px": b.get("_font_size_px"),
}
for p in layout_concept.get("pages", [])
for b in p.get("blocks", [])
]
})
# 3단계: 텍스트 편집자 — 텍스트 정리
yield {"event": "progress", "data": "3/5 텍스트 편집자가 핵심을 정리 중..."}
layout_concept = await fill_content(content, layout_concept, analysis)
logger.info("3단계 완료: 텍스트 정리")
_save_step(run_dir, "step3_filled_blocks.json", {
"blocks": [
{
"area": b.get("area"), "type": b.get("type"),
"topic_id": b.get("topic_id"), "purpose": b.get("purpose"),
"data": b.get("data", {}),
"char_count": len(json.dumps(b.get("data", {}), ensure_ascii=False)),
}
for p in layout_concept.get("pages", [])
for b in p.get("blocks", [])
]
})
# 4단계: 디자인 실무자 — 디자인 조정 + HTML 조립
yield {"event": "progress", "data": "4/5 디자인 실무자가 슬라이드를 조립 중..."}
layout_concept = await _adjust_design(layout_concept, analysis)
html = render_slide(layout_concept)
logger.info("4단계 완료: HTML 조립")
_save_step(run_dir, "step4_css_adjustment.json", {
"area_styles": layout_concept.get("pages", [{}])[0].get("area_styles", {})
})
_save_step(run_dir, "step4_rendered.html", html)
# Phase L: 렌더링 측정 + 피드백 루프 (최대 3회)
from src.html_generator import generate_slide_html
from src.html_validator import validate_and_clean_html
from src.renderer import render_slide_from_html
from src.kei_client import vision_quality_gate
import asyncio
MAX_MEASURE_ROUNDS = 3
measurement = None
for measure_round in range(MAX_MEASURE_ROUNDS):
measurement = await asyncio.to_thread(measure_rendered_heights, html)
_save_step(run_dir, f"step4_measurement_round{measure_round + 1}.json", measurement)
topics = analysis.get("topics", [])
# overflow 감지 — zone + container 양쪽 체크
has_overflow = False
for zone_name, zone_data in measurement.get("zones", {}).items():
if zone_data.get("overflowed"):
has_overflow = True
break
# Phase O: container 레벨 overflow도 체크
for cont_name, cont_data in measurement.get("containers", {}).items():
if cont_data.get("overflowed"):
has_overflow = True
logger.warning(
f"[측정] container-{cont_name}: "
f"scroll={cont_data.get('scrollHeight')}px > "
f"allocated={cont_data.get('allocatedHeight')}px "
f"(+{cont_data.get('excess_px')}px)"
)
break
# 이미지 정보 구성 (base64 포함)
slide_images = []
if image_sizes:
import base64 as b64_mod
for img_key, img_info in image_sizes.items():
img_path = Path(base_path) / img_key if base_path else Path(img_key)
img_b64 = ""
if img_path.exists():
img_b64 = b64_mod.b64encode(img_path.read_bytes()).decode()
slide_images.append({
"path": str(img_path),
"width": img_info.get("width", 0),
"height": img_info.get("height", 0),
"ratio": round(img_info.get("width", 1) / max(1, img_info.get("height", 1)), 2),
"topic_id": img_info.get("topic_id"),
"b64": img_b64,
})
if not has_overflow:
logger.info(f"[측정] 모든 zone/container 정상 (round {measure_round + 1})")
break
# Claude Sonnet이 HTML 생성
generated = await generate_slide_html(
content=content,
analysis=analysis,
container_specs=container_specs,
preset=preset,
images=slide_images,
)
logger.warning(f"[측정] overflow 감지 (round {measure_round + 1})")
# HTML 정화 + 검증
generated = validate_and_clean_html(generated)
# 수학적 축약량 계산 → 편집자 재호출
adjusted = False
for zone_name, zone_data in measurement.get("zones", {}).items():
if not zone_data.get("overflowed"):
continue
excess = zone_data.get("excess_px", 0)
zone_info = preset.get("zones", {}).get(zone_name, {})
width_px = int(settings.slide_width * zone_info.get("width_pct", 100) / 100 * 0.85)
_save_step(run_dir, "step2_generated.json", {
"body_html_length": len(generated.get("body_html", "")),
"sidebar_html_length": len(generated.get("sidebar_html", "")),
"footer_html_length": len(generated.get("footer_html", "")),
"reasoning": generated.get("reasoning", ""),
})
logger.info(
f"[Phase S] HTML 생성 완료: body={len(generated.get('body_html', ''))}자, "
f"sidebar={len(generated.get('sidebar_html', ''))}자, "
f"footer={len(generated.get('footer_html', ''))}자"
)
# Phase O: overflow 블록의 _max_chars_total 축소
for block_m in zone_data.get("blocks", []):
if block_m.get("overflowed"):
trim_chars = calculate_trim_chars(
block_m.get("excess_px", excess),
width_px,
)
for page in layout_concept.get("pages", []):
for block in page.get("blocks", []):
if block.get("area") == zone_name:
current_max = block.get("_max_chars_total", 400)
block["_max_chars_total"] = max(20, current_max - trim_chars)
if "data" in block:
del block["data"]
adjusted = True
logger.info(
f"[측정 조정] {zone_name}/{block_m.get('block_type')}: "
f"{block_m.get('excess_px')}px 초과 → "
f"_max_chars_total {current_max}→{block['_max_chars_total']} ({trim_chars}자 축약)"
)
break
# 3단계: 렌더링 — AI 생성 HTML을 슬라이드 프레임에 삽입
yield {"event": "progress", "data": "3/4 슬라이드 조립 중..."}
if not adjusted:
logger.info("[측정] 조정 대상 없음, 현재 결과 확정")
break
html = render_slide_from_html(generated, analysis, preset)
logger.info("[Phase S] 슬라이드 조립 완료")
_save_step(run_dir, "step3_rendered.html", html)
# 편집자 재호출 → 재렌더링
layout_concept = await fill_content(content, layout_concept, analysis)
layout_concept = await _adjust_design(layout_concept, analysis)
html = render_slide(layout_concept)
logger.info(f"[측정] round {measure_round + 1} 재렌더링 완료")
# ★ Phase Q: 검증 렌더링 + 수학적 조정 + 비전 품질 게이트
measurement = await asyncio.to_thread(measure_rendered_heights, html)
_save_step(run_dir, "step4_measurement.json", measurement)
# 측정 결과 텍스트 (Kei 검수에 전달)
measurement_text = format_measurement_for_kei(measurement) if measurement else ""
# Phase S: overflow 감지
has_overflow = False
for zone_name, zone_data in measurement.get("zones", {}).items():
if zone_data.get("overflowed"):
has_overflow = True
logger.warning(f"[Phase S] {zone_name}: overflow +{zone_data.get('excess_px', 0)}px")
# Phase N-4: 5단계 — Kei 실장 최종 검수 (스크린샷 기반, 최대 1회)
# overflow 없으면 skip (시간 절약)
has_any_overflow = False
if measurement:
for zone_data in measurement.get("zones", {}).values():
if zone_data.get("overflowed"):
has_any_overflow = True
break
if measurement.get("slide", {}).get("overflowed"):
has_any_overflow = True
MAX_REVIEW_ROUNDS = 1
screenshot_b64 = None
if not has_any_overflow:
logger.info("5단계 skip: overflow 없음. 검수 불필요.")
if has_overflow:
logger.warning("[Phase S] overflow 감지 — 결과물에 반영 (후속 품질 게이트에서 평가)")
else:
yield {"event": "progress", "data": "5/5 Kei 실장이 최종 검수 중..."}
logger.info("[Phase S] overflow 없음")
# 스크린샷 캡처 (Selenium)
screenshot_b64 = await asyncio.to_thread(capture_slide_screenshot, html)
if screenshot_b64:
_save_step(run_dir, "step5_screenshot.txt", f"base64 PNG, {len(screenshot_b64)} chars")
logger.info("[5단계] 스크린샷 캡처 완료 → Kei에게 전달")
# Phase S: 비전 모델 품질 게이트
yield {"event": "progress", "data": "4/4 품질 검증 중..."}
for review_round in range(MAX_REVIEW_ROUNDS if has_any_overflow else 0):
review_result = await _review_balance(
html, layout_concept, content, analysis, measurement_text,
screenshot_b64=screenshot_b64,
)
screenshot_b64 = await asyncio.to_thread(capture_slide_screenshot, html)
quality_result = None
if not review_result or not review_result.get("needs_adjustment"):
if review_round == 0:
logger.info("5단계 완료: 조정 불필요")
if screenshot_b64:
_save_step(run_dir, "step5_screenshot.txt", f"base64 PNG, {len(screenshot_b64)} chars")
quality_result = await vision_quality_gate(screenshot_b64, analysis)
if quality_result:
_save_step(run_dir, "step5_quality_gate.json", quality_result)
if not quality_result.get("passed", True):
score = quality_result.get("score", 0)
issues = quality_result.get("issues", [])
logger.warning(f"[Q-6] 품질 게이트 FAIL: {score}/100 — {issues}")
# Q-8: 심각한 품질 문제 시 출력 차단
if score < 30:
logger.error(f"[Q-8] 출력 차단: 품질 {score}/100 < 30 최소 기준")
yield {"event": "error", "data": f"슬라이드 품질 미달 ({score}/100). 재시도해 주세요."}
return
else:
logger.info(f"5단계 완료: {review_round}차 조정 후 균형 확인")
break
issues = review_result.get("issues", [])
logger.info(
f"5단계 ({review_round + 1}/{MAX_REVIEW_ROUNDS}): "
f"조정 필요 — {issues}"
)
_save_step(run_dir, f"step5_review_round{review_round + 1}.json", review_result)
# overflow_detected가 있으면 Kei에게 판단 요청 (Sonnet은 감지만, 판단은 Kei)
overflow_adjs = [
adj for adj in review_result.get("adjustments", [])
if adj.get("action") == "overflow_detected"
]
if overflow_adjs:
overflow_context = _build_overflow_context(
layout_concept, overflow_adjs
)
kei_judgment = await call_kei_overflow_judgment(
overflow_context, content, analysis
)
if kei_judgment is None:
# 넘침 판단도 Kei 필수 — 성공할 때까지 무한 재시도
kei_judgment = await _retry_kei(
call_kei_overflow_judgment, overflow_context, content, analysis
)
_convert_kei_judgment(review_result, kei_judgment)
logger.info(
f"[Kei 넘침 판단] decision={kei_judgment.get('decision')}"
)
layout_concept = await _apply_adjustments(
layout_concept, review_result, content
)
html = render_slide(layout_concept)
logger.info(f"5단계: {review_round + 1}차 조정 반영, 재검토 진행")
logger.info(f"[Q-6] 품질 게이트 PASS: {quality_result.get('score', 0)}/100")
else:
# MAX_REVIEW_ROUNDS 초과
logger.warning(
f"5단계: 최대 재조정 횟수({MAX_REVIEW_ROUNDS}) 도달. 현재 결과로 확정."
)
logger.warning("[Q-6] 스크린샷 캡처 실패 — 품질 게이트 스킵")
# D-5: 이미지를 base64로 삽입 (다운로드 HTML에서도 보이도록)
if base_path:
@@ -390,7 +273,7 @@ async def generate_slide(
_save_step(run_dir, "final.html", html)
yield {"event": "result", "data": html}
logger.info(f"슬라이드 생성 완료: {len(layout_concept.get('pages', []))}페이지, run={run_id}")
logger.info(f"슬라이드 생성 완료: run={run_id}")
except Exception as e:
logger.exception(f"파이프라인 오류: {e}")
@@ -558,7 +441,8 @@ async def _review_balance(
if measurement_text:
overflow_hint_text += f"\n\n{measurement_text}"
# Kei로 최종 검수 (Sonnet 절대 금지, 스크린샷 있으면 이미지 기반)
# Kei로 최종 검수 (레거시 — Phase S에서는 메인 흐름에서 미사용)
from src.kei_client import call_kei_final_review
return await call_kei_final_review(
html, block_summary, zone_budget_text, overflow_hint_text, analysis,
screenshot_b64=screenshot_b64,
@@ -635,7 +519,8 @@ async def _apply_adjustments(
f"조정: {area} → kei_restructure (detail_target)"
)
# 조정된 가이드로 재편집
# 조정된 가이드로 재편집 (레거시 — Phase S에서는 미사용)
from src.content_editor import fill_content
layout_concept = await fill_content(content, layout_concept)
return layout_concept