Spaces:
Sleeping
Sleeping
File size: 4,825 Bytes
6c4292b d0d1766 6c4292b d0d1766 6c4292b 2cf6258 af01c91 46e66e9 2cf6258 46e66e9 d0d1766 46e66e9 2cf6258 46e66e9 2cf6258 46e66e9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
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
|