"""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 ) ] )