# Phase K-1: 파이프라인 스텝별 중간 산출물 로컬 저장 > 각 스텝에서 뭘 결정했고 왜 그렇게 했는지를 파일로 저장하여, > 사용자가 확인하고 피드백할 수 있도록 한다. > 당초부터 있어야 했던 기능. --- ## 문제 - 현재 파이프라인 중간 결과는 메모리에만 존재, 파이프라인 끝나면 사라짐 - 사용자가 "어디서 잘못됐는지" 확인할 방법이 없음 - 로그에 WARNING/INFO 한 줄만 남아서 판단 근거 부족 --- ## 저장 구조 ``` data/runs/{timestamp}/ ├── step1_analysis.json # Kei 꼭지 추출 (topics, purpose, core_message) ├── step1b_concepts.json # Kei 컨셉 구체화 (relation_type, expression_hint) ├── step2_opus_recommendation.json # Opus 블록 추천 ├── step2_sonnet_mapping.json # Sonnet 최종 블록 매핑 ├── step2_validation.json # 높이 검증, 금지 블록 삭제, overflow 내역 ├── step3_filled_blocks.json # 편집자가 채운 텍스트 (블록별 data + 글자 수) ├── step4_css_adjustment.json # CSS 변수 override 내역 ├── step4_rendered.html # 렌더링된 HTML ├── step5_review_round1.json # Kei 1차 검수 결과 (issues + adjustments) ├── step5_review_round2.json # Kei 2차 검수 결과 (있으면) └── final.html # 최종 HTML ``` --- ## 각 파일 내용 상세 ### step1_analysis.json ```json { "title": "건설산업 DX의 올바른 이해", "core_message": "BIM은 DX의 기초적 일부분이다", "total_pages": 1, "info_structure": "...", "topics": [ { "id": 1, "title": "DX와 BIM의 개념 혼용 현실", "purpose": "문제제기", "layer": "intro", "role": "flow", "emphasis": true, "summary": "...", "source_hint": "..." } ], "images": [], "tables": [] } ``` ### step1b_concepts.json ```json { "concepts": [ { "topic_id": 1, "relation_type": "cause_effect", "expression_hint": "현상-문제 인과관계", "source_data": "용어 혼용 현상..." } ] } ``` ### step2_opus_recommendation.json ```json { "recommendations": [ { "topic_id": 1, "block_type": "quote-big-mark", "area": "body", "reason": "문제 제기를 임팩트 있게 강조" } ] } ``` ### step2_sonnet_mapping.json ```json { "preset": "sidebar-right", "blocks": [ { "area": "body", "type": "quote-big-mark", "topic_id": 1, "purpose": "문제제기", "reason": "Opus 추천 유지", "size": "medium", "char_guide": {"quote_text": 150} } ], "opus_diff": [ "Opus 추천과 동일" 또는 "topic_id 4: card-tag-image → card-numbered (사유: ...)" ] } ``` ### step2_validation.json ```json { "forbidden_blocks_removed": ["section-header-bar (body)"], "pill_pair_replaced": [], "sidebar_column_override": [{"topic_id": 4, "column_override": 1}], "overflow": [ {"area": "body", "total_px": 510, "budget_px": 490, "overflow_px": 20} ] } ``` ### step3_filled_blocks.json ```json { "blocks": [ { "area": "body", "type": "quote-big-mark", "topic_id": 1, "purpose": "문제제기", "data": {"quote_text": "건설산업의 디지털 전환...", "source": ""}, "char_count": 95 } ] } ``` ### step4_css_adjustment.json ```json { "area_styles": { "body": "--font-body: 0.85rem; --spacing-inner: 12px;", "sidebar": "--font-body: 0.8rem;", "footer": "" } } ``` ### step5_review_round1.json ```json { "needs_adjustment": true, "issues": ["body zone 높이 초과 (+20px)"], "adjustments": [ {"block_area": "body", "action": "shrink", "target_ratio": 0.8, "detail": "..."} ], "kei_overflow_judgment": null } ``` --- ## 구현 방안 ### 반영 위치 `src/pipeline.py` — `generate_slide()` 함수에서 각 스텝 완료 시 저장 ### 유틸 함수 ```python # pipeline.py 상단에 추가 import time from pathlib import Path def _save_step(run_dir: Path, filename: str, data: Any) -> None: """스텝 결과를 JSON 또는 HTML로 저장한다.""" run_dir.mkdir(parents=True, exist_ok=True) filepath = run_dir / filename if filename.endswith(".html"): filepath.write_text(data, encoding="utf-8") else: with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) logger.info(f"[중간 산출물] {filename} 저장") ``` ### 각 스텝 저장 시점 ```python async def generate_slide(content, manual_layout=None, base_path=""): run_id = str(int(time.time() * 1000)) run_dir = Path("data/runs") / run_id # Step 1-A analysis = await classify_content(content) _save_step(run_dir, "step1_analysis.json", analysis) # Step 1-B analysis = await refine_concepts(content, analysis) _save_step(run_dir, "step1b_concepts.json", { "concepts": [ {k: t.get(k) for k in ("id", "relation_type", "expression_hint", "source_data")} for t in analysis.get("topics", []) if t.get("relation_type") ] }) # Step 2 (Opus + Sonnet + validation) layout_concept = await create_layout_concept(content, analysis) _save_step(run_dir, "step2_sonnet_mapping.json", layout_concept) # Step 3 layout_concept = await fill_content(content, layout_concept, analysis) _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", []) ] }) # Step 4 html = render_slide(layout_concept) _save_step(run_dir, "step4_rendered.html", html) # Step 5 (검수 결과는 루프 안에서) # review_result 저장 # 최종 _save_step(run_dir, "final.html", html) ``` ### Opus 추천 저장 현재 Opus 추천 결과가 `create_layout_concept()` 내부에서 소비되고 사라짐. 추천 결과를 반환값에 포함하거나, 별도로 저장하는 로직 필요. **방법:** `create_layout_concept()` 반환값에 `"opus_recommendation"` 키 추가 --- ## 충돌/회귀 검토 | 항목 | 영향 | |------|------| | pipeline.py | `_save_step()` 함수 추가 + 각 스텝 후 호출 | | design_director.py | `create_layout_concept()` 반환값에 opus 추천 포함 (선택적) | | 기존 기능 | 변경 없음 — 저장은 추가 기능이므로 기존 흐름에 영향 없음 | | Phase I/J/K | 회귀 없음 | | 성능 | JSON 저장은 ms 수준, HTML 저장도 ms 수준 — 영향 미미 | --- ## 실행 순서 1. `_save_step()` 유틸 함수 추가 (pipeline.py) 2. `data/runs/` 디렉토리 구조 설정 3. `generate_slide()` 각 스텝 완료 시점에 저장 호출 추가 4. Opus 추천 결과 반환값 포함 (design_director.py, 선택적) 5. 검증: 파이프라인 실행 후 `data/runs/{timestamp}/` 파일 확인 --- ## 이력 | 날짜 | 내용 | |------|------| | 2026-03-26 | Phase K 완료 후. 사용자 피드백 확인을 위한 중간 산출물 저장 기능 계획. |