341 lines
10 KiB
Python
341 lines
10 KiB
Python
"""
|
|
AI Client - AI 프로바이더 관리 및 통합 인터페이스
|
|
|
|
사용 예시:
|
|
from app.ai.ai_client import AIClient, AIProviderType
|
|
|
|
# 클라이언트 생성
|
|
client = AIClient()
|
|
|
|
# 프로바이더 설정
|
|
client.set_provider(AIProviderType.OPENAI, api_key="sk-...")
|
|
|
|
# 또는 설정에서 자동 로드
|
|
client.load_from_settings()
|
|
|
|
# 채팅 요청
|
|
response = client.chat("안녕하세요!")
|
|
"""
|
|
|
|
from typing import Optional, Dict, List, Type
|
|
from .base import BaseAIProvider, AIProviderType, AIMessage, AIResponse
|
|
from .providers import OpenAIProvider, OpenRouterProvider, GeminiProvider, XAIProvider
|
|
|
|
|
|
class AIClient:
|
|
"""
|
|
AI 클라이언트 매니저
|
|
여러 AI 프로바이더를 관리하고 통합 인터페이스를 제공합니다.
|
|
"""
|
|
|
|
# 프로바이더 타입별 클래스 매핑
|
|
PROVIDER_CLASSES: Dict[AIProviderType, Type[BaseAIProvider]] = {
|
|
AIProviderType.OPENAI: OpenAIProvider,
|
|
AIProviderType.OPENROUTER: OpenRouterProvider,
|
|
AIProviderType.GEMINI: GeminiProvider,
|
|
AIProviderType.XAI: XAIProvider,
|
|
}
|
|
|
|
def __init__(self):
|
|
self._current_provider: Optional[BaseAIProvider] = None
|
|
self._providers: Dict[AIProviderType, BaseAIProvider] = {}
|
|
self._settings = None
|
|
|
|
@property
|
|
def current_provider(self) -> Optional[BaseAIProvider]:
|
|
"""현재 활성화된 프로바이더"""
|
|
return self._current_provider
|
|
|
|
@property
|
|
def current_provider_type(self) -> Optional[AIProviderType]:
|
|
"""현재 프로바이더 타입"""
|
|
if self._current_provider:
|
|
return self._current_provider.provider_type
|
|
return None
|
|
|
|
@property
|
|
def available_provider_types(self) -> List[AIProviderType]:
|
|
"""사용 가능한 프로바이더 타입 목록"""
|
|
return list(self.PROVIDER_CLASSES.keys())
|
|
|
|
def get_provider_name(self, provider_type: AIProviderType) -> str:
|
|
"""프로바이더 타입에 해당하는 표시 이름 반환"""
|
|
names = {
|
|
AIProviderType.OPENAI: "OpenAI",
|
|
AIProviderType.OPENROUTER: "OpenRouter",
|
|
AIProviderType.GEMINI: "Google Gemini",
|
|
AIProviderType.XAI: "xAI (Grok)",
|
|
}
|
|
return names.get(provider_type, provider_type.value)
|
|
|
|
def set_provider(
|
|
self,
|
|
provider_type: AIProviderType,
|
|
api_key: str,
|
|
model: Optional[str] = None
|
|
) -> bool:
|
|
"""
|
|
프로바이더 설정 및 활성화
|
|
|
|
Args:
|
|
provider_type: 프로바이더 타입
|
|
api_key: API 키
|
|
model: 사용할 모델 (None이면 기본 모델)
|
|
|
|
Returns:
|
|
초기화 성공 여부
|
|
"""
|
|
provider_class = self.PROVIDER_CLASSES.get(provider_type)
|
|
if not provider_class:
|
|
raise ValueError(f"지원하지 않는 프로바이더: {provider_type}")
|
|
|
|
# 프로바이더 인스턴스 생성
|
|
provider = provider_class(api_key=api_key, model=model)
|
|
|
|
# 초기화 시도
|
|
if provider.initialize():
|
|
self._providers[provider_type] = provider
|
|
self._current_provider = provider
|
|
return True
|
|
|
|
return False
|
|
|
|
def switch_provider(self, provider_type: AIProviderType) -> bool:
|
|
"""
|
|
이미 등록된 프로바이더로 전환
|
|
|
|
Args:
|
|
provider_type: 전환할 프로바이더 타입
|
|
|
|
Returns:
|
|
전환 성공 여부
|
|
"""
|
|
if provider_type in self._providers:
|
|
self._current_provider = self._providers[provider_type]
|
|
return True
|
|
return False
|
|
|
|
def update_api_key(self, provider_type: AIProviderType, api_key: str) -> bool:
|
|
"""
|
|
프로바이더의 API 키 업데이트
|
|
|
|
Args:
|
|
provider_type: 프로바이더 타입
|
|
api_key: 새 API 키
|
|
|
|
Returns:
|
|
업데이트 및 재초기화 성공 여부
|
|
"""
|
|
if provider_type in self._providers:
|
|
provider = self._providers[provider_type]
|
|
provider.update_api_key(api_key)
|
|
return provider.initialize()
|
|
return False
|
|
|
|
def update_model(self, model: str) -> bool:
|
|
"""
|
|
현재 프로바이더의 모델 변경
|
|
|
|
Args:
|
|
model: 새 모델명
|
|
|
|
Returns:
|
|
변경 성공 여부
|
|
"""
|
|
if self._current_provider:
|
|
self._current_provider.model = model
|
|
return True
|
|
return False
|
|
|
|
def get_available_models(self) -> List[str]:
|
|
"""현재 프로바이더의 사용 가능한 모델 목록"""
|
|
if self._current_provider:
|
|
return self._current_provider.available_models
|
|
return []
|
|
|
|
def load_from_settings(self, settings: 'AISettings') -> bool:
|
|
"""
|
|
설정에서 프로바이더 정보 로드
|
|
|
|
Args:
|
|
settings: AISettings 인스턴스
|
|
|
|
Returns:
|
|
로드 성공 여부
|
|
"""
|
|
self._settings = settings
|
|
|
|
# 현재 선택된 프로바이더 가져오기
|
|
current_type = settings.get_current_provider()
|
|
if not current_type:
|
|
return False
|
|
|
|
# API 키 가져오기
|
|
api_key = settings.get_api_key(current_type)
|
|
if not api_key:
|
|
return False
|
|
|
|
# 모델 가져오기
|
|
model = settings.get_model(current_type)
|
|
|
|
return self.set_provider(current_type, api_key, model)
|
|
|
|
def chat(
|
|
self,
|
|
user_message: str,
|
|
system_prompt: Optional[str] = None,
|
|
temperature: float = 0.7,
|
|
max_tokens: Optional[int] = None,
|
|
**kwargs
|
|
) -> AIResponse:
|
|
"""
|
|
간편 채팅 메서드 (동기)
|
|
|
|
Args:
|
|
user_message: 사용자 메시지
|
|
system_prompt: 시스템 프롬프트 (선택)
|
|
temperature: 응답 다양성
|
|
max_tokens: 최대 토큰
|
|
|
|
Returns:
|
|
AI 응답
|
|
"""
|
|
if not self._current_provider:
|
|
raise RuntimeError("프로바이더가 설정되지 않았습니다.")
|
|
|
|
messages = []
|
|
if system_prompt:
|
|
messages.append(AIMessage(role="system", content=system_prompt))
|
|
messages.append(AIMessage(role="user", content=user_message))
|
|
|
|
return self._current_provider.chat_sync(
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
**kwargs
|
|
)
|
|
|
|
async def chat_async(
|
|
self,
|
|
user_message: str,
|
|
system_prompt: Optional[str] = None,
|
|
temperature: float = 0.7,
|
|
max_tokens: Optional[int] = None,
|
|
**kwargs
|
|
) -> AIResponse:
|
|
"""
|
|
간편 채팅 메서드 (비동기)
|
|
|
|
Args:
|
|
user_message: 사용자 메시지
|
|
system_prompt: 시스템 프롬프트 (선택)
|
|
temperature: 응답 다양성
|
|
max_tokens: 최대 토큰
|
|
|
|
Returns:
|
|
AI 응답
|
|
"""
|
|
if not self._current_provider:
|
|
raise RuntimeError("프로바이더가 설정되지 않았습니다.")
|
|
|
|
messages = []
|
|
if system_prompt:
|
|
messages.append(AIMessage(role="system", content=system_prompt))
|
|
messages.append(AIMessage(role="user", content=user_message))
|
|
|
|
return await self._current_provider.chat(
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
**kwargs
|
|
)
|
|
|
|
def chat_with_history(
|
|
self,
|
|
messages: List[AIMessage],
|
|
temperature: float = 0.7,
|
|
max_tokens: Optional[int] = None,
|
|
**kwargs
|
|
) -> AIResponse:
|
|
"""
|
|
대화 히스토리와 함께 채팅 (동기)
|
|
|
|
Args:
|
|
messages: 메시지 리스트
|
|
temperature: 응답 다양성
|
|
max_tokens: 최대 토큰
|
|
|
|
Returns:
|
|
AI 응답
|
|
"""
|
|
if not self._current_provider:
|
|
raise RuntimeError("프로바이더가 설정되지 않았습니다.")
|
|
|
|
return self._current_provider.chat_sync(
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
**kwargs
|
|
)
|
|
|
|
async def chat_with_history_async(
|
|
self,
|
|
messages: List[AIMessage],
|
|
temperature: float = 0.7,
|
|
max_tokens: Optional[int] = None,
|
|
**kwargs
|
|
) -> AIResponse:
|
|
"""
|
|
대화 히스토리와 함께 채팅 (비동기)
|
|
|
|
Args:
|
|
messages: 메시지 리스트
|
|
temperature: 응답 다양성
|
|
max_tokens: 최대 토큰
|
|
|
|
Returns:
|
|
AI 응답
|
|
"""
|
|
if not self._current_provider:
|
|
raise RuntimeError("프로바이더가 설정되지 않았습니다.")
|
|
|
|
return await self._current_provider.chat(
|
|
messages=messages,
|
|
temperature=temperature,
|
|
max_tokens=max_tokens,
|
|
**kwargs
|
|
)
|
|
|
|
def is_ready(self) -> bool:
|
|
"""프로바이더가 사용 준비되었는지 확인"""
|
|
return self._current_provider is not None and self._current_provider._is_initialized
|
|
|
|
def get_status(self) -> Dict:
|
|
"""현재 상태 정보 반환"""
|
|
if self._current_provider:
|
|
return {
|
|
"ready": self.is_ready(),
|
|
"provider": self._current_provider.provider_name,
|
|
"model": self._current_provider.model,
|
|
"api_key": self._current_provider.api_key, # 마스킹됨
|
|
}
|
|
return {
|
|
"ready": False,
|
|
"provider": None,
|
|
"model": None,
|
|
"api_key": None,
|
|
}
|
|
|
|
|
|
# 싱글톤 인스턴스
|
|
_ai_client_instance: Optional[AIClient] = None
|
|
|
|
|
|
def get_ai_client() -> AIClient:
|
|
"""AI 클라이언트 싱글톤 인스턴스 반환"""
|
|
global _ai_client_instance
|
|
if _ai_client_instance is None:
|
|
_ai_client_instance = AIClient()
|
|
return _ai_client_instance
|
|
|