"""Claude API Service""" import anthropic import json import logging from typing import Dict, Any from app.config import get_settings logger = logging.getLogger(__name__) settings = get_settings() class ClaudeService: """Claude API 호출 서비스""" def __init__(self): self.client = anthropic.Anthropic(api_key=settings.claude_api_key) self.model = settings.claude_model self.max_tokens = settings.claude_max_tokens self.temperature = settings.claude_temperature async def generate_completion( self, prompt: str, system_prompt: str = None ) -> Dict[str, Any]: """ Claude API 호출하여 응답 생성 Args: prompt: 사용자 프롬프트 system_prompt: 시스템 프롬프트 (선택) Returns: Claude API 응답 (JSON 파싱) """ try: # 메시지 구성 messages = [ { "role": "user", "content": prompt } ] # API 호출 logger.info(f"Claude API 호출 시작 - Model: {self.model}") if system_prompt: response = self.client.messages.create( model=self.model, max_tokens=self.max_tokens, temperature=self.temperature, system=system_prompt, messages=messages ) else: response = self.client.messages.create( model=self.model, max_tokens=self.max_tokens, temperature=self.temperature, messages=messages ) # 응답 텍스트 추출 response_text = response.content[0].text logger.info(f"Claude API 응답 수신 완료 - Tokens: {response.usage.input_tokens + response.usage.output_tokens}") # JSON 파싱 # ```json ... ``` 블록 제거 if "```json" in response_text: response_text = response_text.split("```json")[1].split("```")[0].strip() elif "```" in response_text: response_text = response_text.split("```")[1].split("```")[0].strip() result = json.loads(response_text) return result except json.JSONDecodeError as e: logger.error(f"JSON 파싱 실패: {e}") logger.error(f"응답 텍스트: {response_text[:500]}...") raise ValueError(f"Claude API 응답을 JSON으로 파싱할 수 없습니다: {str(e)}") except Exception as e: logger.error(f"Claude API 호출 실패: {e}") raise # 싱글톤 인스턴스 claude_service = ClaudeService()