Kei API 연동 복구 + 실장 정보구조 분석 + 팀장 role 기반 배치
1단계 (실장): - Kei API 연동 복구 (타임아웃 무제한, Kei persona 사고) - 정보 구조 파악 단계 추가 (본문 흐름 vs 참조 분리) - 각 꼭지에 role(flow/reference) 부여 - fallback: Anthropic 직접 호출 (info_structure + role 포함) 2단계 (팀장): - info_structure + role 기반 배치 규칙 추가 - flow → 좌측/메인, reference → 우측/사이드 - detail_target → 본문 제외 - 중복 방지 규칙 파이프라인: - pipeline.py import re 추가 Figma 관련 (다른 Claude Code 작업분): - catalog.yaml, figma-screenshots, figma-analysis, 테스트 HTML Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
"""DA-12: 1단계 — Kei 실장 (꼭지 추출 + 분석).
|
||||
|
||||
본문에서 핵심 꼭지를 추출하고, 각 꼭지의 레이어/강조/배치 방향을 분석한다.
|
||||
이미지/표/상세 콘텐츠도 판단한다.
|
||||
1차: Kei API를 통해 Kei persona가 사고하여 꼭지를 추출한다.
|
||||
fallback: Kei API 실패 시 Anthropic API 직접 호출.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -11,103 +11,191 @@ import re
|
||||
from typing import Any
|
||||
|
||||
import anthropic
|
||||
import httpx
|
||||
|
||||
from src.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CLASSIFICATION_PROMPT = """당신은 콘텐츠를 분석하여 슬라이드 구조를 설계하는 실장이다.
|
||||
|
||||
## 역할
|
||||
본문에서 핵심 꼭지를 추출하고, 각 꼭지의 성격을 분석하여 슬라이드 구조를 설계한다.
|
||||
|
||||
## 꼭지 추출 규칙
|
||||
- 본문에서 2~5개의 핵심 꼭지(파트)를 추출한다
|
||||
- 1페이지 적정 꼭지 수: 5개
|
||||
- 꼭지가 5개를 넘고 중요도가 동등하면 → 2페이지로 분리 (의미 기반 분할)
|
||||
- 5개인데 내용이 많으면 → 세부 내용은 "자세히보기" 대상으로 표시
|
||||
|
||||
## 각 꼭지 분석 항목
|
||||
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": "슬라이드 제목",
|
||||
"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
|
||||
}
|
||||
]
|
||||
}
|
||||
```"""
|
||||
KEI_PROMPT = (
|
||||
"다음 콘텐츠를 슬라이드로 정리하려고 해.\n\n"
|
||||
"## 1단계: 정보 구조 파악 (꼭지 추출 전에 먼저 수행)\n"
|
||||
"- 이 콘텐츠가 하나의 흐름으로 읽히는가, 아니면 '본문 흐름'과 '참조 정보'로 분리되는 구조인가?\n"
|
||||
"- 독립적으로 참조되는 정보(용어 정의, 부록, 별도 설명)가 있는가?\n"
|
||||
"- 상세히 다뤄야 하지만 본문 흐름을 끊는 내용(비교표, 상세 데이터)이 있는가?\n"
|
||||
"- 정보 구조를 info_structure 필드에 기술해줘.\n\n"
|
||||
"## 2단계: 꼭지 추출\n"
|
||||
"- 1단계에서 파악한 정보 구조를 바탕으로 꼭지를 나눠줘\n"
|
||||
"- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n"
|
||||
"- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n"
|
||||
"- 슬라이드에 맞게 정리하되, 원본이 말하려는 흐름은 유지\n"
|
||||
"- 각 꼭지의 레이어(도입/핵심/보조/결론), 강조 여부, 배치 방향을 판단해줘\n"
|
||||
"- 참조 정보는 role: 'reference'로, 본문 흐름은 role: 'flow'로 표시\n"
|
||||
"- 본문 흐름을 끊는 상세 내용은 detail_target: true로 표시\n"
|
||||
"- 이미지/표가 있으면 그것도 판단해줘\n"
|
||||
"- 1페이지 적정 꼭지: 5개. 초과 시 2페이지 분리.\n\n"
|
||||
"## 출력 형식 (JSON만)\n"
|
||||
"```json\n"
|
||||
'{"title": "제목", "total_pages": 1, '
|
||||
'"info_structure": "이 콘텐츠의 정보 구조 설명 (본문 흐름 vs 참조 분리 등)", '
|
||||
'"topics": ['
|
||||
'{"id": 1, "title": "꼭지 제목", "summary": "요약", '
|
||||
'"layer": "intro|core|supporting|conclusion", '
|
||||
'"role": "flow|reference", '
|
||||
'"emphasis": true, "direction": "vertical|horizontal|flexible", '
|
||||
'"content_type": "text|image|table|mixed", '
|
||||
'"detail_target": false, "page": 1}]}\n'
|
||||
"```\n\n"
|
||||
"## 콘텐츠:\n"
|
||||
)
|
||||
|
||||
|
||||
async def classify_content(content: str) -> dict[str, Any] | None:
|
||||
"""1단계: 본문에서 꼭지를 추출하고 분석한다.
|
||||
"""1단계: Kei API를 통해 꼭지를 추출하고 분석한다.
|
||||
|
||||
Args:
|
||||
content: 원본 텍스트 콘텐츠
|
||||
|
||||
Returns:
|
||||
분류 결과 JSON. 실패 시 None.
|
||||
1차: Kei API (persona + RAG + 사고)
|
||||
fallback: Anthropic API 직접 호출
|
||||
"""
|
||||
if not settings.anthropic_api_key:
|
||||
logger.warning("ANTHROPIC_API_KEY 미설정. 수동 분류 모드.")
|
||||
return None
|
||||
# 1차: Kei API
|
||||
result = await _call_kei_api(content)
|
||||
if result:
|
||||
logger.info(
|
||||
f"[Kei API] 꼭지 추출 완료: {result.get('title', '')}, "
|
||||
f"{len(result.get('topics', []))}개 꼭지"
|
||||
)
|
||||
return result
|
||||
|
||||
# fallback: Anthropic 직접
|
||||
logger.warning("Kei API 실패. Anthropic 직접 호출로 fallback.")
|
||||
result = await _call_anthropic_direct(content)
|
||||
if result:
|
||||
logger.info(
|
||||
f"[Anthropic] 꼭지 추출 완료: {result.get('title', '')}, "
|
||||
f"{len(result.get('topics', []))}개 꼭지"
|
||||
)
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
|
||||
async def _call_kei_api(content: str) -> dict[str, Any] | None:
|
||||
"""Kei API를 통해 꼭지 추출. SSE 스트리밍 응답을 파싱."""
|
||||
kei_url = getattr(settings, "kei_api_url", "http://localhost:8000")
|
||||
|
||||
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
|
||||
analysis = _parse_json(result_text)
|
||||
|
||||
if analysis and "topics" in analysis:
|
||||
logger.info(
|
||||
f"꼭지 추출 완료: {analysis.get('title', 'untitled')}, "
|
||||
f"{len(analysis['topics'])}개 꼭지, "
|
||||
f"{analysis.get('total_pages', 1)}페이지"
|
||||
async with httpx.AsyncClient(timeout=None) as client:
|
||||
response = await client.post(
|
||||
f"{kei_url}/api/message",
|
||||
json={
|
||||
"message": KEI_PROMPT + content,
|
||||
"session_id": "design-agent",
|
||||
"mode": "chat",
|
||||
},
|
||||
timeout=None,
|
||||
)
|
||||
return analysis
|
||||
else:
|
||||
logger.warning(f"분류 JSON 파싱 실패. 응답: {result_text[:200]}")
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.warning(f"Kei API HTTP {response.status_code}")
|
||||
return None
|
||||
|
||||
# SSE 응답에서 토큰 수집
|
||||
full_text = _extract_sse_text(response.text)
|
||||
|
||||
if not full_text:
|
||||
logger.warning("Kei API 응답에서 텍스트 추출 실패")
|
||||
return None
|
||||
|
||||
# JSON 추출
|
||||
result = _parse_json(full_text)
|
||||
if result and "topics" in result:
|
||||
return result
|
||||
|
||||
logger.warning(f"Kei API JSON 파싱 실패. 텍스트: {full_text[:200]}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"실장 분류 호출 실패: {e}")
|
||||
logger.warning(f"Kei API 호출 실패: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _extract_sse_text(raw: str) -> str:
|
||||
"""SSE 응답에서 토큰 텍스트를 수집한다. CRLF/LF 모두 처리."""
|
||||
tokens = []
|
||||
# CRLF 또는 LF로 이벤트 분리
|
||||
events = re.split(r'\r?\n\r?\n', raw)
|
||||
|
||||
for event in events:
|
||||
if not event.strip():
|
||||
continue
|
||||
|
||||
event_type = ""
|
||||
event_data = ""
|
||||
|
||||
for line in event.split('\n'):
|
||||
line = line.strip('\r')
|
||||
if line.startswith('event:'):
|
||||
event_type = line[6:].strip()
|
||||
elif line.startswith('data:'):
|
||||
event_data = line[5:].strip()
|
||||
|
||||
if not event_data:
|
||||
continue
|
||||
|
||||
if event_type == 'token':
|
||||
try:
|
||||
token = json.loads(event_data)
|
||||
if isinstance(token, str):
|
||||
tokens.append(token)
|
||||
except json.JSONDecodeError:
|
||||
tokens.append(event_data)
|
||||
elif event_type == 'done':
|
||||
break
|
||||
|
||||
return "".join(tokens)
|
||||
|
||||
|
||||
async def _call_anthropic_direct(content: str) -> dict[str, Any] | None:
|
||||
"""Anthropic API 직접 호출 (Kei API fallback)."""
|
||||
if not settings.anthropic_api_key:
|
||||
return None
|
||||
|
||||
system_prompt = (
|
||||
"당신은 콘텐츠를 분석하여 슬라이드 구조를 설계하는 실장이다.\n\n"
|
||||
"## 핵심 원칙\n"
|
||||
"- 원본의 논리 흐름과 정보를 빠뜨리지 마라\n"
|
||||
"- 원본에 있는 내용을 임의로 제거하거나 다른 의미로 바꾸지 마라\n"
|
||||
"- 슬라이드에 맞게 정리하되, 원본이 말하려는 흐름은 유지\n\n"
|
||||
"## 꼭지 추출 규칙\n"
|
||||
"- 본문에서 2~5개의 핵심 꼭지를 추출한다\n"
|
||||
"- 1페이지 적정 꼭지 수: 5개\n"
|
||||
"- 초과 시 2페이지 분리\n\n"
|
||||
"## 출력 형식 (JSON만. 설명 없이.)\n"
|
||||
'{"title": "제목", "total_pages": 1, "topics": ['
|
||||
'{"id": 1, "title": "꼭지 제목", "summary": "요약", '
|
||||
'"layer": "intro|core|supporting|conclusion", '
|
||||
'"emphasis": true, "direction": "vertical|horizontal|flexible", '
|
||||
'"content_type": "text", "detail_target": false, "page": 1}]}'
|
||||
)
|
||||
|
||||
try:
|
||||
client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
|
||||
response = await client.messages.create(
|
||||
model="claude-sonnet-4-20250514",
|
||||
max_tokens=2048,
|
||||
system=system_prompt,
|
||||
messages=[{"role": "user", "content": f"다음 콘텐츠의 꼭지를 추출해줘:\n\n{content}"}],
|
||||
)
|
||||
|
||||
result_text = response.content[0].text
|
||||
result = _parse_json(result_text)
|
||||
|
||||
if result and "topics" in result:
|
||||
return result
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Anthropic 직접 호출 실패: {e}")
|
||||
return None
|
||||
|
||||
|
||||
@@ -129,7 +217,7 @@ def _parse_json(text: str) -> dict[str, Any] | None:
|
||||
|
||||
|
||||
def manual_classify(content: str) -> dict[str, Any]:
|
||||
"""실장 분류 실패 시 기본 구조를 반환하는 fallback."""
|
||||
"""분류 실패 시 기본 구조 fallback."""
|
||||
return {
|
||||
"title": "슬라이드",
|
||||
"total_pages": 1,
|
||||
|
||||
Reference in New Issue
Block a user