deepseek_utils.py

"""
DeepSeek API 테스트 및 유틸리티 기능
"""
import os
import logging
import requests
import json
from typing import Dict, Any, Optional

# 로깅 설정
logger = logging.getLogger("DeepSeekUtils")


class DeepSeekError(Exception):
    """DeepSeek API 관련 오류"""
    pass


def test_deepseek_api(api_key: str, endpoint: str, model: str) -> Dict[str, Any]:
    """
    DeepSeek API 연결 테스트

    Args:
        api_key: DeepSeek API 키
        endpoint: DeepSeek API 엔드포인트
        model: 사용할 모델명

    Returns:
        테스트 결과 딕셔너리 (성공 여부 및 메시지)
    """
    if not api_key:
        logger.error("DeepSeek API 키가 제공되지 않았습니다.")
        return {
            "success": False,
            "message": "API 키가 제공되지 않았습니다.",
            "status_code": None,
            "response": None
        }

    try:
        logger.info(f"DeepSeek API 연결 테스트 시작: {endpoint}, 모델: {model}")

        # 테스트용 간단한 프롬프트
        test_prompt = "Hello, please respond with a short greeting."

        # API 요청 헤더 및 데이터
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        }

        payload = {
            "model": model,
            "messages": [{"role": "user", "content": test_prompt}],
            "temperature": 0.7,
            "max_tokens": 50
        }

        # API 요청 전송
        response = requests.post(
            endpoint,
            headers=headers,
            data=json.dumps(payload),
            timeout=10  # 10초 타임아웃
        )

        # 응답 확인
        if response.status_code == 200:
            logger.info("DeepSeek API 연결 성공")
            response_data = response.json()

            # 응답 내용 확인
            if "choices" in response_data and len(response_data["choices"]) > 0:
                message_content = response_data["choices"][0].get("message", {}).get("content", "")
                return {
                    "success": True,
                    "message": "API 연결 성공",
                    "status_code": response.status_code,
                    "response": message_content[:100] + "..." if len(message_content) > 100 else message_content
                }
            else:
                return {
                    "success": True,
                    "message": "API 연결 성공했으나 응답 형식이 예상과 다릅니다.",
                    "status_code": response.status_code,
                    "response": response_data
                }
        else:
            logger.error(f"DeepSeek API 오류: 상태 코드 {response.status_code}")
            error_message = ""
            try:
                error_data = response.json()
                error_message = error_data.get("error", {}).get("message", str(error_data))
            except:
                error_message = response.text

            return {
                "success": False,
                "message": f"API 오류: {error_message}",
                "status_code": response.status_code,
                "response": error_message
            }

    except requests.exceptions.Timeout:
        logger.error("DeepSeek API 요청 시간 초과")
        return {
            "success": False,
            "message": "API 요청 시간 초과",
            "status_code": None,
            "response": None
        }
    except requests.exceptions.ConnectionError:
        logger.error("DeepSeek API 연결 실패")
        return {
            "success": False,
            "message": "API 서버 연결 실패",
            "status_code": None,
            "response": None
        }
    except Exception as e:
        logger.error(f"DeepSeek API 테스트 중 예상치 못한 오류: {e}", exc_info=True)
        return {
            "success": False,
            "message": f"예상치 못한 오류: {str(e)}",
            "status_code": None,
            "response": None
        }


def create_deepseek_client(api_key: str, endpoint: str, model: str):
    """
    DeepSeek 클라이언트 생성 (LangChain 호환)

    Args:
        api_key: DeepSeek API 키
        endpoint: DeepSeek API 엔드포인트
        model: 사용할 모델명

    Returns:
        DeepSeek 클라이언트 객체 또는 None
    """
    # LangChain과 DeepSeek 통합 시도
    try:
        from langchain_openai import ChatOpenAI

        # API 연결 테스트 먼저 수행
        test_result = test_deepseek_api(api_key, endpoint, model)

        if not test_result["success"]:
            logger.error(f"DeepSeek API 연결 테스트 실패: {test_result['message']}")
            return None

        # 정상 연결 시 클라이언트 생성
        # DeepSeek는 OpenAI 호환 API를 제공하므로 ChatOpenAI를 사용
        client = ChatOpenAI(
            model=model,
            temperature=0.2,
            api_key=api_key,
            base_url=endpoint.rstrip("/v1/chat/completions"),  # OpenAI 호환 베이스 URL
        )

        logger.info(f"DeepSeek 클라이언트 생성 성공: {model}")
        return client

    except ImportError as e:
        logger.error(f"필요한 라이브러리 임포트 실패: {e}")
        return None
    except Exception as e:
        logger.error(f"DeepSeek 클라이언트 생성 중 오류: {e}", exc_info=True)
        return None