336 lines
13 KiB
Python
336 lines
13 KiB
Python
# 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)
|
|
|
|
# =============================================================================
|
|
# 헬스체크 및 상태 확인 메서드들
|
|
# =============================================================================
|
|
|
|
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() |