deepseek_client.py

import os
from typing import List, Dict, Optional
from config import Config
from custom_openai_client import CustomOpenAI, ModelProvider
import requests
from dataclasses import dataclass
from enum import Enum
import urllib3

class DeepseekClient(CustomOpenAI):
    def __init__(self, api_key=None, **kwargs):
        print(f"[DEBUG] Starting DeepseekClient")
        try:
            api_key = api_key or Config.get_deepseek_key()
            print(f"[DEBUG] Using API key: {api_key[:4]}...")  # 最初の4文字のみ表示
            super().__init__(
                deepseek_api_key=api_key,
                **kwargs
            )
            print(f"[DEBUG] API client initialized successfully")
        except Exception as e:
            print(f"[DEBUG] Initialization error: {str(e)}")
            raise

    def deepseek_chat(
        self,
        messages: List[Dict[str, str]],
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: Optional[int] = None,
        **kwargs
    ) -> Optional[str]:
        """
        Chat completion method for DeepSeek models.
        
        Args:
            messages: List of message dictionaries with 'role' and 'content'
            model: DeepSeek model identifier
            temperature: Sampling temperature (0-2)
            max_tokens: Maximum number of tokens to generate
            **kwargs: Additional parameters to pass to the API
        
        Returns:
            Generated message content or None if an error occurs
        """
        if not self.deepseek_api_key:
            raise ValueError("DeepSeek API key is required for deepseek_chat")

        try:
            headers = {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.deepseek_api_key}",
                "User-Agent": "DeepseekClient/1.0"
            }
            
            data = {
                "model": model,
                "messages": messages,
                "temperature": temperature,
                "stream": False
            }
            
            if max_tokens is not None:
                data["max_tokens"] = max_tokens
            
            # Add any additional kwargs to the request data
            data.update(kwargs)
            
            config = self.API_CONFIGS[ModelProvider.DEEPSEEK]
            response = requests.post(
                config.chat_endpoint,
                headers=headers,
                json=data,
                verify=False,  # SSL検証を無効化（開発環境のみ）
                timeout=(10, 60)  # 接続タイムアウト10秒、読み取りタイムアウト60秒に延長
            )
            # SSL検証を無効にした警告を抑制
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
            
            if response.status_code != 200:
                error_msg = f"DeepSeek API request failed with status {response.status_code}"
                try:
                    error_data = response.json()
                    print(f"[DEBUG] Error response: {error_data}")
                    if "error" in error_data:
                        error_msg += f": {error_data['error']}"
                except Exception as e:
                    print(f"[DEBUG] Failed to parse error response: {str(e)}")
                raise ValueError(error_msg)
            
            response_data = response.json()
            
            if not response_data.get("choices"):
                raise ValueError("No choices in DeepSeek API response")

            return response_data["choices"][0]["message"]["content"]

        except requests.exceptions.RequestException as e:
            print(f"Network error during DeepSeek API call: {str(e)}")
            return None
        except ValueError as e:
            print(f"DeepSeek API Error: {str(e)}")
            return None
        except Exception as e:
            print(f"Unexpected error in DeepSeek chat: {str(e)}")
            return None

    def create(self, messages: List[Dict[str, str]], model: str = None, **kwargs) -> str:
        print(f"[DEBUG] Starting DeepseekClient::create")
        print(f"[DEBUG] Model: {model or 'deepseek-chat'}")
        print(f"[DEBUG] Messages: {[{k: v for k, v in m.items()} for m in messages]}")
        
        try:
            result = self.deepseek_chat(messages=messages, model=model, **kwargs)
            if result:
                print(f"[DEBUG] API request successful")
                print(f"[DEBUG] Response: {result[:100]}...")  # Show first 100 chars of response
                return result
            else:
                raise Exception("No response from DeepSeek API")
        except Exception as e:
            print(f"[DEBUG] Error in create: {str(e)}")
            raise