File size: 4,676 Bytes
c0e3412 | 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 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Minimal DeepSeek API client using only the Python standard library.
This client targets the Chat Completions endpoint and defaults to model
"DeepSeek-V3.2-Exp" per user request. It reads the API key from the
environment variable `DEEPSEEK_API_KEY` and avoids external dependencies.
"""
from __future__ import annotations
import json
import os
import ssl
from typing import Dict, List, Optional
from urllib.request import Request, urlopen
from .env import load_env
from pathlib import Path
import hashlib
import urllib.error
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
CHAT_COMPLETIONS_PATH = "/chat/completions"
_MODEL_ALIASES = {
"DeepSeek-V3.2-Exp": "deepseek-chat",
"DeepSeek-V3.2": "deepseek-chat",
"DeepSeek-V3": "deepseek-chat",
"DeepSeek-R1": "deepseek-reasoner",
}
def _normalize_model_name(model: str) -> str:
raw = (model or "").strip()
if raw in _MODEL_ALIASES:
return _MODEL_ALIASES[raw]
return raw or "deepseek-chat"
def deepseek_chat(
messages: List[Dict[str, str]],
model: str = "DeepSeek-V3.2-Exp",
stream: bool = False,
base_url: str = DEEPSEEK_BASE_URL,
temperature: float = 0,
max_tokens: int = 2048,
) -> Optional[Dict]:
"""Call DeepSeek Chat Completions API and return the parsed JSON response.
Args:
messages: Conversation messages in OpenAI-compatible format.
model: Model name. Defaults to "DeepSeek-V3.2-Exp" per the request.
stream: When True, asks the API to stream. This client does not handle
streaming responses; the flag is forwarded as-is.
base_url: API base URL. Default points to the official DeepSeek API.
temperature: Sampling temperature (0 = deterministic). Defaults to 0.
max_tokens: Maximum tokens in the response. Defaults to 2048.
Returns:
Parsed JSON dictionary on success, or None if a recoverable error occurs.
"""
# Load .env once before reading variables.
load_env()
api_key = (os.getenv("DEEPSEEK_API_KEY") or "").strip()
if not api_key:
# Fail gracefully if no key is present.
return None
url = f"{base_url}{CHAT_COMPLETIONS_PATH}"
normalized_model = _normalize_model_name(model)
body = {
"model": normalized_model,
"messages": messages,
"stream": stream,
"temperature": temperature,
"max_tokens": max_tokens,
"seed": 0,
}
data = json.dumps(body, sort_keys=True).encode("utf-8")
req = Request(url, data=data, method="POST")
req.add_header("Content-Type", "application/json")
req.add_header("Authorization", f"Bearer {api_key}")
# Create a default SSL context; can be customized if needed.
context = ssl.create_default_context()
use_cache = os.getenv("DEEPSEEK_CACHE", "0") == "1"
cache_dir = Path(".cache")
cache_dir.mkdir(exist_ok=True)
cache_key = hashlib.sha256((url + data.decode("utf-8")).encode("utf-8")).hexdigest()
cache_path = cache_dir / f"deepseek_{cache_key}.json"
import time as _time
max_retries = 3
for attempt in range(max_retries):
try:
if use_cache and cache_path.exists():
return json.loads(cache_path.read_text(encoding="utf-8"))
with urlopen(req, context=context, timeout=30) as resp:
payload = resp.read().decode("utf-8")
parsed = json.loads(payload)
if use_cache:
try:
cache_path.write_text(json.dumps(parsed, ensure_ascii=False), encoding="utf-8")
except (IOError, OSError):
pass
return parsed
except urllib.error.HTTPError as e:
if e.code == 429 and attempt < max_retries - 1:
wait = 2 ** attempt
_time.sleep(wait)
continue
return None
except (urllib.error.URLError, ssl.SSLError, json.JSONDecodeError, TimeoutError):
if attempt < max_retries - 1:
_time.sleep(1)
continue
return None
return None
def extract_text_answer(response: Dict) -> Optional[str]:
"""Extract the assistant text from a chat completions response.
The function safely navigates the typical OpenAI-compatible structure and
returns None if the expected fields are absent.
"""
try:
choices = response.get("choices") or []
if not choices:
return None
message = choices[0].get("message") or {}
return message.get("content")
except (KeyError, IndexError, TypeError) as e:
return None
|