# app/services/claude_service.py import json import logging from typing import Optional, Dict, Any, Tuple import anthropic from ..config.settings import settings logger = logging.getLogger(__name__) class ClaudeService: """Claude API 연동 서비스""" def __init__(self): try: # API 키 유효성 검사 if not settings.CLAUDE_API_KEY or settings.CLAUDE_API_KEY.strip() == "": raise ValueError("CLAUDE_API_KEY가 설정되지 않았습니다") # Claude 클라이언트 초기화 self.client = anthropic.Anthropic(api_key=settings.CLAUDE_API_KEY) self.model = settings.CLAUDE_MODEL self.initialization_error = None logger.info(f"✅ ClaudeService 초기화 완료 (모델: {self.model})") except Exception as e: self.initialization_error = str(e) self.client = None logger.error(f"❌ ClaudeService 초기화 실패: {e}") def is_ready(self) -> bool: """서비스 준비 상태 확인""" return self.client is not None and self.initialization_error is None def get_initialization_error(self) -> Optional[str]: """초기화 에러 메시지 반환""" return self.initialization_error async def test_api_connection(self) -> Tuple[bool, Optional[str]]: """Claude API 연결을 테스트합니다.""" if not self.is_ready(): return False, self.initialization_error try: logger.info("🔍 Claude API 연결 테스트 시작...") response = self.client.messages.create( model=self.model, max_tokens=50, messages=[ { "role": "user", "content": "안녕하세요. 연결 테스트입니다. '연결 성공'이라고 답변해주세요." } ] ) if response.content and len(response.content) > 0: logger.info("✅ Claude API 연결 테스트 성공") return True, None else: error_msg = "Claude API 응답이 비어있음" logger.warning(f"⚠️ {error_msg}") return False, error_msg except Exception as e: error_msg = f"Claude API 연결 테스트 실패: {str(e)}" logger.error(f"❌ {error_msg}") return False, error_msg async def generate_action_recommendations(self, context: str, additional_context: Optional[str] = None) -> Tuple[Optional[str], Optional[Dict[str, Any]]]: """점주를 위한 액션 추천을 생성합니다.""" if not self.is_ready(): logger.error("ClaudeService가 준비되지 않음") return None, None try: logger.info("🤖 Claude API를 통한 액션 추천 생성 시작") prompt = self._build_action_prompt(context, additional_context) response = self.client.messages.create( model=self.model, max_tokens=4000, temperature=0.7, messages=[ { "role": "user", "content": prompt } ] ) if response.content and len(response.content) > 0: raw_response = response.content[0].text logger.info(f"✅ 액션 추천 생성 완료: {len(raw_response)} 문자") parsed_response = self._parse_json_response(raw_response) return raw_response, parsed_response else: logger.warning("⚠️ Claude API 응답이 비어있음") return None, None except Exception as e: logger.error(f"❌ 액션 추천 생성 중 오류: {e}") return None, None def _parse_json_response(self, raw_response: str) -> Optional[Dict[str, Any]]: """Claude의 원본 응답에서 JSON을 추출하고 파싱합니다.""" try: import re # JSON 블록 찾기 json_match = re.search(r'```json\s*\n(.*?)\n```', raw_response, re.DOTALL) if json_match: json_str = json_match.group(1).strip() else: brace_start = raw_response.find('{') brace_end = raw_response.rfind('}') if brace_start != -1 and brace_end != -1 and brace_end > brace_start: json_str = raw_response[brace_start:brace_end + 1] else: logger.warning("⚠️ JSON 패턴을 찾을 수 없음") return None parsed_json = json.loads(json_str) logger.info("✅ JSON 파싱 성공") return parsed_json except json.JSONDecodeError as e: logger.warning(f"⚠️ JSON 파싱 실패: {e}") return None except Exception as e: logger.warning(f"⚠️ JSON 추출 실패: {e}") return None def _build_action_prompt(self, context: str, additional_context: Optional[str] = None) -> str: """액션 추천을 위한 프롬프트를 구성합니다.""" base_prompt = f"""당신은 소상공인을 위한 경영 컨설턴트입니다. 아래 정보를 바탕으로 실질적이고 구체적인 액션 추천을 해주세요. **분석 데이터:** {context} **추천 요구사항:** 1. 단기 계획 (1-3개월): 즉시 실행 가능한 개선사항 2. 중기 계획 (3-6개월): 점진적 개선 방안 3. 장기 계획 (6개월-1년): 전략적 발전 방향 **응답 형식:** 반드시 아래 JSON 형식으로만 응답해주세요: ```json {{ "summary": {{ "current_situation": "현재 상황 요약", "key_insights": ["핵심 인사이트 1", "핵심 인사이트 2", "핵심 인사이트 3"], "priority_areas": ["우선 개선 영역 1", "우선 개선 영역 2"] }}, "action_plans": {{ "short_term": [ {{ "title": "액션 제목", "description": "구체적인 실행 방법", "expected_impact": "예상 효과", "timeline": "실행 기간", "cost": "예상 비용" }} ], "mid_term": [ {{ "title": "액션 제목", "description": "구체적인 실행 방법", "expected_impact": "예상 효과", "timeline": "실행 기간", "cost": "예상 비용" }} ], "long_term": [ {{ "title": "액션 제목", "description": "구체적인 실행 방법", "expected_impact": "예상 효과", "timeline": "실행 기간", "cost": "예상 비용" }} ] }}, "implementation_tips": [ "실행 팁 1", "실행 팁 2", "실행 팁 3" ] }} ``` **응답은 반드시 유효한 JSON 형식으로만 작성하고, JSON 앞뒤에 다른 텍스트는 포함하지 마세요.**""" if additional_context: base_prompt += f"\n\n**점주 추가 요청사항:**\n{additional_context}\n" return base_prompt # ============================================================================= # 호환성을 위한 메서드들 (기존 코드가 사용할 수 있도록) # ============================================================================= async def get_recommendation(self, prompt: str) -> Optional[str]: """Claude API를 호출하여 추천을 받습니다. (호환성용)""" if not self.is_ready(): logger.error("ClaudeService가 준비되지 않음") return None try: logger.info("🤖 Claude API 호출 시작") response = self.client.messages.create( model=self.model, max_tokens=4000, temperature=0.7, messages=[ { "role": "user", "content": prompt } ] ) if response.content and len(response.content) > 0: raw_response = response.content[0].text logger.info(f"✅ Claude API 응답 성공: {len(raw_response)} 문자") return raw_response else: logger.warning("⚠️ Claude API 응답이 비어있음") return None except Exception as e: logger.error(f"❌ Claude API 호출 실패: {e}") return None def parse_recommendation_response(self, raw_response: str) -> Optional[Dict[str, Any]]: """Claude 응답에서 JSON을 추출하고 파싱합니다. (호환성용)""" return self._parse_json_response(raw_response) def build_recommendation_prompt(self, store_id: str, context: str, vector_context: Optional[str] = None) -> str: """액션 추천용 프롬프트를 구성합니다. (호환성용)""" prompt_parts = [ "당신은 소상공인을 위한 경영 컨설턴트입니다.", f"가게 ID: {store_id}", f"점주 요청: {context}" ] if vector_context: prompt_parts.extend([ "\n--- 동종 업체 분석 데이터 ---", vector_context, "--- 분석 데이터 끝 ---\n" ]) prompt_parts.extend([ "\n위 정보를 바탕으로 실질적이고 구체적인 액션 추천을 해주세요.", "응답은 반드시 아래 JSON 형식으로만 작성해주세요:", "", "```json", "{", ' "summary": {', ' "current_situation": "현재 상황 요약",', ' "key_insights": ["핵심 인사이트 1", "핵심 인사이트 2"],', ' "priority_areas": ["우선 개선 영역 1", "우선 개선 영역 2"]', ' },', ' "action_plans": {', ' "short_term": [', ' {', ' "title": "즉시 실행 가능한 액션",', ' "description": "구체적인 실행 방법",', ' "expected_impact": "예상 효과",', ' "timeline": "1-2주",', ' "cost": "예상 비용"', ' }', ' ],', ' "mid_term": [', ' {', ' "title": "중기 개선 방안",', ' "description": "구체적인 실행 방법",', ' "expected_impact": "예상 효과",', ' "timeline": "1-3개월",', ' "cost": "예상 비용"', ' }', ' ]', ' },', ' "implementation_tips": ["실행 팁 1", "실행 팁 2"]', "}", "```" ]) return "\n".join(prompt_parts) def create_prompt_for_api_response(self, context: str, additional_context: Optional[str] = None) -> str: """API 응답용 프롬프트를 생성합니다. (호환성용)""" return self._build_action_prompt(context, additional_context) async def generate_action_recommendations_optimized( self, context: str, additional_context: Optional[str] = None ) -> Tuple[Optional[str], Optional[Dict[str, Any]]]: """ 최적화된 액션 추천 생성 - 더 명확한 JSON 지시사항 - 토큰 효율성 개선 - 파싱 안정성 향상 """ if not self.is_ready(): return None, None try: # 최적화된 프롬프트 구성 prompt = self._build_optimized_prompt(context, additional_context) response = self.client.messages.create( model=self.model, max_tokens=3000, # 토큰 수 최적화 temperature=0.3, # 일관성 향상 messages=[{"role": "user", "content": prompt}] ) if response.content and len(response.content) > 0: raw_response = response.content[0].text # 즉시 JSON 파싱 시도 parsed_json = self._parse_json_response_enhanced(raw_response) return raw_response, parsed_json return None, None except Exception as e: logger.error(f"Claude AI 호출 실패: {e}") return None, None def _build_optimized_prompt(self, context: str, additional_context: Optional[str] = None) -> str: """최적화된 프롬프트 구성""" prompt_parts = [ "당신은 소상공인 경영 컨설턴트입니다.", f"분석 요청: {context}", ] if additional_context: prompt_parts.extend([ "\n=== 참고 데이터 ===", additional_context, "==================\n" ]) prompt_parts.extend([ "위 정보를 바탕으로 실행 가능한 액션을 추천해주세요.", "", "⚠️ 응답은 반드시 아래 JSON 형식으로만 작성하세요:", "다른 텍스트는 포함하지 마세요.", "", "```json", "{", ' "summary": {', ' "current_situation": "현재 상황 요약 (50자 이내)",', ' "key_insights": ["핵심 포인트1", "핵심 포인트2"],', ' "priority": "high|medium|low"', ' },', ' "actions": [', ' {', ' "title": "액션명",', ' "description": "구체적 실행방법",', ' "timeline": "실행기간",', ' "cost": "예상비용",', ' "impact": "예상효과"', ' }', ' ],', ' "quick_tips": ["즉시 실행 팁1", "즉시 실행 팁2"]', "}", "```" ]) return "\n".join(prompt_parts) def _parse_json_response_enhanced(self, raw_response: str) -> Optional[Dict[str, Any]]: """향상된 JSON 파싱""" try: # 1. JSON 블록 추출 json_match = re.search(r'```json\s*(\{.*?\})\s*```', raw_response, re.DOTALL) if json_match: json_str = json_match.group(1) else: # 2. 직접 JSON 찾기 json_match = re.search(r'(\{.*\})', raw_response, re.DOTALL) if json_match: json_str = json_match.group(1) else: return None # 3. JSON 파싱 return json.loads(json_str) except json.JSONDecodeError as e: logger.error(f"JSON 파싱 오류: {e}") return None except Exception as e: logger.error(f"JSON 추출 실패: {e}") return None # ============================================================================= # 헬스체크 및 상태 확인 메서드들 # ============================================================================= def get_health_status(self) -> Dict[str, Any]: """ClaudeService 상태 확인""" try: status = { "service": "claude_api", "status": "healthy" if self.is_ready() else "unhealthy", "model": self.model, "api_key_configured": bool(settings.CLAUDE_API_KEY and settings.CLAUDE_API_KEY.strip()), "timestamp": self._get_timestamp() } if self.initialization_error: status["initialization_error"] = self.initialization_error status["status"] = "error" return status except Exception as e: return { "service": "claude_api", "status": "error", "error": str(e), "timestamp": self._get_timestamp() } def _get_timestamp(self) -> str: """현재 시간 문자열 반환""" from datetime import datetime return datetime.now().isoformat()