utils/llm_interface.py

"""
LLM 인터페이스 모듈 - 다양한 LLM을 통합 관리
"""

import os
import logging
from typing import List, Dict, Any, Optional, Union
from dotenv import load_dotenv

# LLM 클라이언트 임포트
from utils.openai_client import OpenAILLM
from utils.deepseek_client import DeepSeekLLM

# 환경 변수 로드
load_dotenv()

# 로거 설정
logger = logging.getLogger("LLMInterface")
if not logger.hasHandlers():
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

class LLMInterface:
    """다양한 LLM API를 통합 관리하는 인터페이스 클래스"""
    
    # 지원되는 LLM 목록 (UI에서 표시될 이름과 내부 식별자)
    SUPPORTED_LLMS = {
        "OpenAI": "openai",
        "DeepSeek": "deepseek"
    }
    
    def __init__(self, default_llm: str = "openai"):
        """LLM 인터페이스 초기화
        
        Args:
            default_llm: 기본 LLM 식별자 ('openai' 또는 'deepseek')
        """
        # LLM 클라이언트 초기화
        self.llm_clients = {
            "openai": OpenAILLM(),
            "deepseek": DeepSeekLLM()
        }
        
        # 기본 LLM 설정 (유효하지 않은 경우 openai로 설정)
        if default_llm not in self.llm_clients:
            logger.warning(f"지정된 기본 LLM '{default_llm}'가 유효하지 않습니다. 'openai'로 설정됩니다.")
            default_llm = "openai"
        
        self.default_llm = default_llm
        self.current_llm = default_llm
        
        logger.info(f"LLM 인터페이스 초기화 완료, 기본 LLM: {default_llm}")
    
    def set_llm(self, llm_id: str) -> bool:
        """현재 LLM을 설정
        
        Args:
            llm_id: LLM 식별자
            
        Returns:
            성공 여부
        """
        if llm_id not in self.llm_clients:
            logger.error(f"지원되지 않는 LLM 식별자: {llm_id}")
            return False
        
        self.current_llm = llm_id
        logger.info(f"현재 LLM이 '{llm_id}'로 설정되었습니다.")
        return True
    
    def get_current_llm_name(self) -> str:
        """현재 LLM의 표시 이름 반환"""
        for name, id in self.SUPPORTED_LLMS.items():
            if id == self.current_llm:
                return name
        return "Unknown"
    
    def get_current_llm_details(self) -> Dict[str, str]:
        """현재 LLM의 세부 정보 반환"""
        name = self.get_current_llm_name()
        model = ""
        
        if self.current_llm == "openai":
            model = self.llm_clients["openai"].model
        elif self.current_llm == "deepseek":
            model = self.llm_clients["deepseek"].model
        
        return {
            "name": name,
            "id": self.current_llm,
            "model": model
        }
    
    def generate(
        self, 
        prompt: str, 
        system_prompt: Optional[str] = None,
        llm_id: Optional[str] = None,
        **kwargs
    ) -> str:
        """텍스트 생성
        
        Args:
            prompt: 사용자 프롬프트
            system_prompt: 시스템 프롬프트 (선택 사항)
            llm_id: 사용할 LLM 식별자 (미지정 시 현재 LLM 사용)
            **kwargs: 추가 인자 (temperature, max_tokens 등)
            
        Returns:
            생성된 텍스트
        """
        # 사용할 LLM 결정
        llm_to_use = llm_id if llm_id and llm_id in self.llm_clients else self.current_llm
        llm_client = self.llm_clients[llm_to_use]
        
        # LLM 정보 로깅
        logger.info(f"텍스트 생성 요청, LLM: {llm_to_use}")
        
        # 생성 요청
        return llm_client.generate(
            prompt=prompt,
            system_prompt=system_prompt,
            **kwargs
        )
    
    def rag_generate(
        self, 
        query: str, 
        context: List[str],
        llm_id: Optional[str] = None,
        **kwargs
    ) -> str:
        """RAG 기반 텍스트 생성
        
        Args:
            query: 사용자 질의
            context: 검색된 문맥 목록
            llm_id: 사용할 LLM 식별자 (미지정 시 현재 LLM 사용)
            **kwargs: 추가 인자 (temperature, max_tokens 등)
            
        Returns:
            생성된 텍스트
        """
        # 사용할 LLM 결정
        llm_to_use = llm_id if llm_id and llm_id in self.llm_clients else self.current_llm
        llm_client = self.llm_clients[llm_to_use]
        
        # LLM 정보 로깅
        logger.info(f"RAG 텍스트 생성 요청, LLM: {llm_to_use}")
        
        # 생성 요청
        return llm_client.rag_generate(
            query=query,
            context=context,
            **kwargs
        )