kmlee1andClaude Opus 4.6
fb67f221f4
1. kei_client.py _parse_json(): Kei가 마크다운 리스트 접두사(- )를 붙여 JSON 응답하는 경우 전처리하여 파싱. 원본 + 클린 버전 둘 다 시도. 2. image_utils.py: base_path와 상대경로가 이중으로 붙는 경우 파일명으로 rglob 재탐색하여 실제 파일 위치 찾기. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""DA-12: 1단계 — Kei 실장 (꼭지 추출 + 분석).
|
|
|
|
1차: Kei API를 통해 Kei persona가 사고하여 꼭지를 추출한다.
|
|
fallback: Kei API 실패 시 Anthropic API 직접 호출.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
import anthropic
|
|
import httpx
|
|
|
|
from src.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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"
|
|
"- reference 꼭지는 페이지 안에 사이드바로 배치한다. popup으로 빼지 않는다.\n"
|
|
"- 본문 흐름을 뒷받침하는 근거/사례도 페이지 안에 배치한다. popup으로 빼지 않는다.\n"
|
|
"- detail_target: true는 정말로 별도로 봐야 하는 상세 데이터(비교표, 상세 스펙)에만 사용한다.\n"
|
|
"- 근거, 사례, 용어 정의는 detail_target이 아니다.\n"
|
|
"- 이미지가 있으면: 몇 개인지, 어떤 꼭지 소속인지, 핵심(도표/차트)인지 보조(참고 사진)인지, 텍스트가 포함된 이미지인지 판단해줘\n"
|
|
"- 표가 있으면: 행/열 규모, 1페이지에 전체 표시 가능한지 판단해줘\n"
|
|
"- 이미지/표 판단 결과를 images[], tables[] 배열에 기록해줘\n"
|
|
"- 1페이지 적정 꼭지: 5개. 초과 시 2페이지 분리.\n"
|
|
"- 2페이지로 분리하는 기준: 꼭지 수가 아니라 콘텐츠 분량이 진짜 많을 때만.\n"
|
|
"- 꼭지가 5~6개라도 각 꼭지의 내용이 적으면 1페이지에 충분히 담을 수 있다.\n"
|
|
"- 억지로 2페이지로 나누지 마라. 1페이지로 담을 수 있으면 1페이지로.\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}], '
|
|
'"images": [{"topic_id": 1, "role": "key|supporting", "has_text": false, "description": "이미지 설명"}], '
|
|
'"tables": [{"topic_id": 2, "rows": 5, "cols": 3, "fits_single_page": true, "description": "표 설명"}]}\n'
|
|
"```\n\n"
|
|
"## 콘텐츠:\n"
|
|
)
|
|
|
|
|
|
async def classify_content(content: str) -> dict[str, Any] | None:
|
|
"""1단계: Kei API를 통해 꼭지를 추출하고 분석한다.
|
|
|
|
1차: Kei API (persona + RAG + 사고)
|
|
fallback: Anthropic API 직접 호출
|
|
"""
|
|
# 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:
|
|
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,
|
|
)
|
|
|
|
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"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"
|
|
"- 참조 정보는 role: 'reference', 본문 흐름은 role: 'flow'로 표시\n"
|
|
"- 1페이지 적정 꼭지 수: 5개\n"
|
|
"- 초과 시 2페이지 분리\n"
|
|
"- 이미지가 있으면 images[]에, 표가 있으면 tables[]에 판단 기록\n\n"
|
|
"## 출력 형식 (JSON만. 설명 없이.)\n"
|
|
'{"title": "제목", "total_pages": 1, '
|
|
'"info_structure": "정보 구조 설명", '
|
|
'"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}], '
|
|
'"images": [], "tables": []}'
|
|
)
|
|
|
|
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
|
|
|
|
|
|
def _parse_json(text: str) -> dict[str, Any] | None:
|
|
"""텍스트에서 JSON을 추출한다.
|
|
|
|
Kei API가 마크다운 리스트 접두사(- )를 붙여 응답하는 경우에도 처리.
|
|
"""
|
|
# 전처리: 각 줄 앞의 마크다운 리스트 접두사(- ) 제거
|
|
# Kei API가 JSON을 마크다운 리스트로 감싸서 응답하는 경우 대응
|
|
lines = text.split("\n")
|
|
cleaned_lines = []
|
|
for line in lines:
|
|
stripped = line.lstrip()
|
|
if stripped.startswith("- "):
|
|
cleaned_lines.append(stripped[2:])
|
|
elif stripped.startswith("* "):
|
|
cleaned_lines.append(stripped[2:])
|
|
else:
|
|
cleaned_lines.append(stripped)
|
|
cleaned = "\n".join(cleaned_lines)
|
|
|
|
# 원본 + 클린 버전 둘 다 시도
|
|
for target in [text, cleaned]:
|
|
patterns = [
|
|
r"```json\s*(.*?)```",
|
|
r"```\s*(.*?)```",
|
|
r"(\{.*\})",
|
|
]
|
|
for pattern in patterns:
|
|
match = re.search(pattern, target, 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]:
|
|
"""분류 실패 시 기본 구조 fallback."""
|
|
return {
|
|
"title": "슬라이드",
|
|
"total_pages": 1,
|
|
"info_structure": "",
|
|
"topics": [
|
|
{
|
|
"id": 1,
|
|
"title": "핵심 내용",
|
|
"summary": content[:100],
|
|
"layer": "core",
|
|
"role": "flow",
|
|
"emphasis": False,
|
|
"direction": "flexible",
|
|
"content_type": "text",
|
|
"detail_target": False,
|
|
"page": 1,
|
|
},
|
|
],
|
|
"images": [],
|
|
"tables": [],
|
|
}
|