hgzero/ai-python/app/services/claude_service.py

"""Claude API 서비스"""
import anthropic
import json
import logging
from typing import List
from datetime import datetime
import uuid
from app.config import get_settings
from app.models import SimpleSuggestion, RealtimeSuggestionsResponse

logger = logging.getLogger(__name__)
settings = get_settings()


class ClaudeService:
    """Claude API 클라이언트"""

    def __init__(self):
        self.client = None
        if settings.claude_api_key:
            self.client = anthropic.Anthropic(api_key=settings.claude_api_key)

    async def analyze_suggestions(self, transcript_text: str) -> RealtimeSuggestionsResponse:
        """
        회의 텍스트를 분석하여 AI 제안사항 생성

        Args:
            transcript_text: 누적된 회의 텍스트

        Returns:
            RealtimeSuggestionsResponse
        """
        if not self.client:
            logger.warning("Claude API 키가 설정되지 않음 - Mock 데이터 반환")
            return self._generate_mock_suggestions()

        logger.info(f"Claude API 호출 - 텍스트 길이: {len(transcript_text)}")

        system_prompt = """당신은 회의록 작성 전문 AI 어시스턴트입니다.

실시간 회의 텍스트를 분석하여 **중요한 제안사항만** 추출하세요.

**추출 기준**:
- 회의 안건과 직접 관련된 내용
- 논의가 필요한 주제
- 결정된 사항
- 액션 아이템

**제외할 내용**:
- 잡담, 농담, 인사말
- 회의와 무관한 대화
- 단순 확인이나 질의응답

**응답 형식**: JSON만 반환 (다른 설명 없이)
{
  "suggestions": [
    {
      "content": "구체적인 제안 내용 (1-2문장으로 명확하게)",
      "confidence": 0.9
    }
  ]
}

**주의**:
- 각 제안은 독립적이고 명확해야 함
- 회의 맥락에서 실제 중요한 내용만 포함
- confidence는 0-1 사이 값 (확신 정도)"""

        try:
            response = self.client.messages.create(
                model=settings.claude_model,
                max_tokens=settings.claude_max_tokens,
                temperature=settings.claude_temperature,
                system=system_prompt,
                messages=[
                    {
                        "role": "user",
                        "content": f"다음 회의 내용을 분석해주세요:\n\n{transcript_text}"
                    }
                ]
            )

            # 응답 파싱
            content_text = response.content[0].text
            suggestions_data = self._parse_claude_response(content_text)

            logger.info(f"Claude API 응답 성공 - 제안사항: {len(suggestions_data.get('suggestions', []))}개")

            return RealtimeSuggestionsResponse(
                suggestions=[
                    SimpleSuggestion(
                        id=str(uuid.uuid4()),
                        content=s["content"],
                        timestamp=self._get_current_timestamp(),
                        confidence=s.get("confidence", 0.8)
                    )
                    for s in suggestions_data.get("suggestions", [])
                ]
            )

        except Exception as e:
            logger.error(f"Claude API 호출 실패: {e}")
            return RealtimeSuggestionsResponse(suggestions=[])

    def _parse_claude_response(self, text: str) -> dict:
        """Claude 응답에서 JSON 추출 및 파싱"""
        # ```json ... ``` 제거
        if "```json" in text:
            start = text.find("```json") + 7
            end = text.rfind("```")
            text = text[start:end].strip()
        elif "```" in text:
            start = text.find("```") + 3
            end = text.rfind("```")
            text = text[start:end].strip()

        try:
            return json.loads(text)
        except json.JSONDecodeError as e:
            logger.error(f"JSON 파싱 실패: {e}, 원문: {text[:200]}")
            return {"suggestions": []}

    def _get_current_timestamp(self) -> str:
        """현재 타임스탬프 (HH:MM:SS)"""
        return datetime.now().strftime("%H:%M:%S")

    def _generate_mock_suggestions(self) -> RealtimeSuggestionsResponse:
        """Mock 제안사항 생성 (테스트용)"""
        mock_suggestions = [
            "신제품의 타겟 고객층을 20-30대로 설정하고, 모바일 우선 전략을 취하기로 논의 중입니다.",
            "개발 일정: 1차 프로토타입은 11월 15일까지 완성, 2차 베타는 12월 1일 론칭",
            "마케팅 예산 배분에 대해 SNS 광고 60%, 인플루언서 마케팅 40%로 의견이 나왔으나 추가 검토 필요"
        ]

        import random
        content = random.choice(mock_suggestions)

        return RealtimeSuggestionsResponse(
            suggestions=[
                SimpleSuggestion(
                    id=str(uuid.uuid4()),
                    content=content,
                    timestamp=self._get_current_timestamp(),
                    confidence=0.85
                )
            ]
        )