utils/deepseek_client.py

"""
DeepSeek API 클라이언트 모듈
"""

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

# 환경 변수 로드
load_dotenv()

# 로거 설정
logger = logging.getLogger("DeepSeekLLM")
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 DeepSeekLLM:
    """DeepSeek API 래퍼 클래스"""
    
    def __init__(self):
        """DeepSeek LLM 클래스 초기화"""
        self.api_key = os.getenv("DEEPSEEK_API_KEY")
        self.api_base = os.getenv("DEEPSEEK_API_BASE", "https://api.deepseek.com")
        self.model = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
        
        if not self.api_key:
            logger.warning("DeepSeek API 키가 .env 파일에 설정되지 않았습니다.")
            logger.warning("DEEPSEEK_API_KEY를 확인하세요.")
        else:
            logger.info("DeepSeek API 키 로드 완료.")
    
    def chat_completion(
        self, 
        messages: List[Dict[str, str]], 
        temperature: float = 0.7, 
        max_tokens: int = 1000,
        **kwargs
    ) -> Dict[str, Any]:
        """
        DeepSeek 채팅 완성 API 호출
        
        Args:
            messages: 채팅 메시지 목록
            temperature: 생성 온도 (낮을수록 결정적)
            max_tokens: 생성할 최대 토큰 수
            **kwargs: 추가 API 매개변수
        
        Returns:
            API 응답 (딕셔너리)
        """
        if not self.api_key:
            logger.error("API 키가 설정되지 않아 DeepSeek API를 호출할 수 없습니다.")
            raise ValueError("DeepSeek API 키가 설정되지 않았습니다.")
        
        try:
            logger.info(f"DeepSeek API 요청 전송 중 (모델: {self.model})")
            
            # API 요청 헤더 및 데이터 준비
            headers = {
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
            
            data = {
                "model": self.model,
                "messages": messages,
                "temperature": temperature,
                "max_tokens": max_tokens
            }
            
            # 추가 매개변수 병합
            for key, value in kwargs.items():
                if key not in data:
                    data[key] = value
            
            # API 요청 보내기
            endpoint = f"{self.api_base}/v1/chat/completions"
            response = requests.post(
                endpoint,
                headers=headers,
                json=data
            )
            
            # 응답 검증
            response.raise_for_status()
            return response.json()
                
        except requests.exceptions.RequestException as e:
            logger.error(f"DeepSeek API 요청 실패: {e}")
            raise Exception(f"DeepSeek API 요청 실패: {e}")
        except json.JSONDecodeError as e:
            logger.error(f"DeepSeek API 응답 파싱 실패: {e}")
            raise Exception(f"DeepSeek API 응답 파싱 실패: {e}")
    
    def generate(
        self, 
        prompt: str, 
        system_prompt: Optional[str] = None,
        temperature: float = 0.7,
        max_tokens: int = 1000,
        **kwargs
    ) -> str:
        """
        간단한 텍스트 생성 인터페이스
        
        Args:
            prompt: 사용자 프롬프트
            system_prompt: 시스템 프롬프트 (선택 사항)
            temperature: 생성 온도
            max_tokens: 생성할 최대 토큰 수
            **kwargs: 추가 API 매개변수
        
        Returns:
            생성된 텍스트
        """
        messages = []
        
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        
        messages.append({"role": "user", "content": prompt})
        
        try:
            response = self.chat_completion(
                messages=messages,
                temperature=temperature,
                max_tokens=max_tokens,
                **kwargs
            )
            
            # 응답 검증 및 처리
            if not response or 'choices' not in response or not response['choices']:
                logger.error("DeepSeek API 응답에서 생성된 텍스트를 찾을 수 없습니다.")
                return ""
            
            return response['choices'][0]['message']['content'].strip()
            
        except Exception as e:
            logger.error(f"텍스트 생성 중 오류 발생: {e}")
            return f"오류: {str(e)}"
    
    def rag_generate(
        self, 
        query: str, 
        context: List[str],
        system_prompt: Optional[str] = None,
        temperature: float = 0.3,
        max_tokens: int = 1000,
        **kwargs
    ) -> str:
        """
        RAG 검색 결과를 활용한 텍스트 생성
        
        Args:
            query: 사용자 질의
            context: 검색된 문맥 목록
            system_prompt: 시스템 프롬프트 (선택 사항)
            temperature: 생성 온도
            max_tokens: 생성할 최대 토큰 수
            **kwargs: 추가 API 매개변수
        
        Returns:
            생성된 텍스트
        """
        if not system_prompt:
            system_prompt = """당신은 검색 결과를 기반으로 질문에 답변하는 도우미입니다.
- 검색 결과는 <context> 태그 안에 제공됩니다.
- 검색 결과에 답변이 있으면 해당 정보를 사용하여 명확하게 답변하세요.
- 검색 결과에 답변이 없으면 "검색 결과에 관련 정보가 없습니다"라고 말하세요.
- 검색 내용을 그대로 복사하지 말고, 자연스러운 한국어로 답변을 작성하세요.
- 답변은 간결하고 정확하게 제공하세요."""
        
        # 중요: 컨텍스트 길이 제한
        max_context = 10
        if len(context) > max_context:
            logger.warning(f"컨텍스트가 너무 길어 처음 {max_context}개만 사용합니다.")
            context = context[:max_context]
        
        # 각 컨텍스트 액세스
        limited_context = []
        for i, doc in enumerate(context):
            # 각 문서를 1000자로 제한
            if len(doc) > 1000:
                logger.warning(f"문서 {i+1}의 길이가 제한되었습니다 ({len(doc)} -> 1000)")
                doc = doc[:1000] + "...(생략)"
            limited_context.append(doc)
        
        context_text = "\n\n".join([f"문서 {i+1}: {doc}" for i, doc in enumerate(limited_context)])
        
        prompt = f"""질문: {query}

<context>
{context_text}
</context>

위 검색 결과를 참고하여 질문에 답변해 주세요."""
        
        try:
            return self.generate(
                prompt=prompt,
                system_prompt=system_prompt,
                temperature=temperature,
                max_tokens=max_tokens,
                **kwargs
            )
        except Exception as e:
            logger.error(f"RAG 텍스트 생성 중 오류 발생: {e}")
            return f"오류: {str(e)}"