""" DeepSeek API 클라이언트 모듈 """ import os import json import logging import traceback 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" # 디버깅: API 요청 데이터 로깅 (민감 정보 제외) debug_data = data.copy() debug_data["messages"] = f"[{len(data['messages'])}개 메시지]" logger.debug(f"DeepSeek API 요청 데이터: {json.dumps(debug_data)}") response = requests.post( endpoint, headers=headers, json=data, timeout=30 # 30초 타임아웃 설정 ) # 응답 상태 코드 확인 if not response.ok: logger.error(f"DeepSeek API 오류: 상태 코드 {response.status_code}") logger.error(f"응답 내용: {response.text}") return {"error": f"API 오류: 상태 코드 {response.status_code}", "detail": response.text} # 응답 파싱 try: result = response.json() logger.debug(f"API 응답 구조: {list(result.keys())}") return result except json.JSONDecodeError as e: logger.error(f"DeepSeek API JSON 파싱 실패: {e}") logger.error(f"원본 응답: {response.text[:500]}...") return {"error": "API 응답을 파싱할 수 없습니다", "detail": str(e)} except requests.exceptions.RequestException as e: logger.error(f"DeepSeek API 요청 실패: {e}") return {"error": f"API 요청 실패: {str(e)}"} except Exception as e: logger.error(f"DeepSeek API 호출 중 예상치 못한 오류: {e}") logger.error(traceback.format_exc()) return {"error": f"예상치 못한 오류: {str(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 "error" in response: logger.error(f"텍스트 생성 중 API 오류: {response['error']}") error_detail = response.get("detail", "") return f"API 오류: {response['error']} {error_detail}" # 응답 형식 검증 if 'choices' not in response or not response['choices']: logger.error(f"API 응답에 'choices' 필드가 없습니다: {response}") return "응답 형식 오류: 생성된 텍스트를 찾을 수 없습니다." # 메시지 컨텐츠 확인 choice = response['choices'][0] if 'message' not in choice or 'content' not in choice['message']: logger.error(f"API 응답에 예상 필드가 없습니다: {choice}") return "응답 형식 오류: 메시지 내용을 찾을 수 없습니다." generated_text = choice['message']['content'].strip() logger.info(f"텍스트 생성 완료 (길이: {len(generated_text)})") return generated_text except Exception as e: logger.error(f"텍스트 생성 중 예외 발생: {e}") logger.error(traceback.format_exc()) 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 = """당신은 검색 결과를 기반으로 질문에 답변하는 도우미입니다. - 검색 결과는 태그 안에 제공됩니다. - 검색 결과에 답변이 있으면 해당 정보를 사용하여 명확하게 답변하세요. - 검색 결과에 답변이 없으면 "검색 결과에 관련 정보가 없습니다"라고 말하세요. - 검색 내용을 그대로 복사하지 말고, 자연스러운 한국어로 답변을 작성하세요. - 답변은 간결하고 정확하게 제공하세요.""" try: # 중요: 컨텍스트 길이 제한 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_text} 위 검색 결과를 참고하여 질문에 답변해 주세요.""" logger.info(f"RAG 프롬프트 생성 완료 (길이: {len(prompt)})") result = self.generate( prompt=prompt, system_prompt=system_prompt, temperature=temperature, max_tokens=max_tokens, **kwargs ) # 결과가 오류 메시지인지 확인 if result.startswith("오류") or result.startswith("API 오류") or result.startswith("응답 형식 오류"): logger.error(f"RAG 생성 결과가 오류를 포함합니다: {result}") # 좀 더 사용자 친화적인 오류 메시지 반환 return "죄송합니다. 현재 응답을 생성하는데 문제가 발생했습니다. 잠시 후 다시 시도해주세요." return result except Exception as e: logger.error(f"RAG 텍스트 생성 중 예외 발생: {str(e)}") logger.error(traceback.format_exc()) return "죄송합니다. 응답 생성 중 오류가 발생했습니다. 잠시 후 다시 시도해주세요."