""" DeepSeek LLM API 클라이언트 모듈 """ import os import json import logging import requests from typing import List, Dict, Any, Optional, Union from dotenv import load_dotenv # 환경 변수 로드 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 LLM API 래퍼 클래스""" def __init__(self): """DeepSeek LLM 클래스 초기화""" self.api_key = os.getenv("DEEPSEEK_API_KEY") self.endpoint = os.getenv("DEEPSEEK_ENDPOINT", "https://api.deepseek.com/v1/chat/completions") 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 LLM API 키 로드 완료.") def chat_completion( self, messages: List[Dict[str, str]], temperature: float = 0.7, max_tokens: int = 1000, stream: bool = False, **kwargs ) -> Dict[str, Any]: """ DeepSeek 채팅 완성 API 호출 Args: messages: 채팅 메시지 목록 temperature: 생성 온도 (낮을수록 결정적) max_tokens: 생성할 최대 토큰 수 stream: 스트리밍 응답 활성화 여부 **kwargs: 추가 API 매개변수 Returns: API 응답 (딕셔너리) """ if not self.api_key: logger.error("API 키가 설정되지 않아 DeepSeek API를 호출할 수 없습니다.") raise ValueError("DeepSeek API 키가 설정되지 않았습니다.") headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "model": self.model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": stream, **kwargs } try: logger.info(f"DeepSeek API 요청 전송 중: {self.endpoint}") response = requests.post( self.endpoint, headers=headers, json=payload, timeout=60 # 타임아웃 설정 ) response.raise_for_status() if stream: return response # 스트리밍 응답은 원시 응답 객체 반환 else: return response.json() except requests.exceptions.Timeout: logger.error("DeepSeek API 요청 시간 초과") raise TimeoutError("DeepSeek API 요청 시간 초과") except requests.exceptions.RequestException as e: logger.error(f"DeepSeek API 요청 실패: {e}") if hasattr(e, 'response') and e.response is not None: logger.error(f"응답 코드: {e.response.status_code}, 내용: {e.response.text}") raise ConnectionError(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_text = "\n\n".join([f"문서 {i+1}: {doc}" for i, doc in enumerate(context)]) prompt = f"""질문: {query} {context_text} 위 검색 결과를 참고하여 질문에 답변해 주세요.""" 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)}"