feat: add DeepSeek and Minimax (m2.7) translation providers
Some checks failed
Deploy to Homelab / Deploy Wordly to 192.168.1.151 (push) Has been cancelled
Deploy to Homelab / Deploy Monitoring (if configured) (push) Has been cancelled

New providers:
- DeepSeek: direct API with deepseek-chat model, very cost-effective
- Minimax: MiniMax-M1 model via OpenAI-compatible API, supports m2.7

Changes:
- Full provider implementations with retry, health check, batch support
- Provider config with env vars (DEEPSEEK_*, MINIMAX_*)
- Auto-registration in provider registry
- Updated fallback chain to include new providers
- Updated setup-env.sh wizard with options 6 (deepseek) and 7 (minimax)
- Updated manage-keys.sh with new menu entries and provider switching
- Updated docker-compose.yml with new env vars

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-10 12:30:36 +02:00
parent d6d19eaf0c
commit e6e1678b1d
7 changed files with 603 additions and 11 deletions

View File

@@ -81,6 +81,14 @@ services:
- DEEPL_API_KEY=${DEEPL_API_KEY:-}
- OPENAI_API_KEY=${OPENAI_API_KEY:-}
- OPENROUTER_API_KEY=${OPENROUTER_API_KEY:-}
- DEEPSEEK_ENABLED=${DEEPSEEK_ENABLED:-false}
- DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY:-}
- DEEPSEEK_MODEL=${DEEPSEEK_MODEL:-deepseek-chat}
- DEEPSEEK_BASE_URL=${DEEPSEEK_BASE_URL:-https://api.deepseek.com/v1}
- MINIMAX_ENABLED=${MINIMAX_ENABLED:-false}
- MINIMAX_API_KEY=${MINIMAX_API_KEY:-}
- MINIMAX_MODEL=${MINIMAX_MODEL:-MiniMax-M1}
- MINIMAX_BASE_URL=${MINIMAX_BASE_URL:-https://api.minimax.chat/v1}
- MAX_FILE_SIZE_MB=${MAX_FILE_SIZE_MB:-50}
- RATE_LIMIT_REQUESTS_PER_MINUTE=${RATE_LIMIT_REQUESTS_PER_MINUTE:-60}
- RATE_LIMIT_TRANSLATIONS_PER_MINUTE=${RATE_LIMIT_TRANSLATIONS_PER_MINUTE:-10}

View File

@@ -101,6 +101,8 @@ while true; do
show_key_status "OpenAI" "OPENAI_API_KEY"
show_key_status "DeepL" "DEEPL_API_KEY"
show_key_status "OpenRouter" "OPENROUTER_API_KEY"
show_key_status "DeepSeek" "DEEPSEEK_API_KEY"
show_key_status "Minimax" "MINIMAX_API_KEY"
show_key_status "Google" "GOOGLE_API_KEY"
echo ""
@@ -121,7 +123,9 @@ while true; do
echo " 4) Configurer Stripe (toutes les cles)"
echo " 5) Changer le mot de passe admin"
echo " 6) Changer le service de traduction"
echo " 7) Tout afficher (attention: secrets visibles)"
echo " 7) Configurer DeepSeek"
echo " 8) Configurer Minimax (m2.7)"
echo " 9) Tout afficher (attention: secrets visibles)"
echo " 0) Quitter"
echo ""
echo -ne "${YELLOW}Choix:${NC} "
@@ -192,10 +196,12 @@ while true; do
6)
echo -e "\n${BOLD}--- Service de traduction ---${NC}"
echo " Actuel: $(get_env TRANSLATION_SERVICE)"
echo " 1) ollama (local, gratuit)"
echo " 2) deepl (haute qualite)"
echo " 3) openai (GPT)"
echo " 1) ollama (local, gratuit)"
echo " 2) deepl (haute qualite)"
echo " 3) openai (GPT)"
echo " 4) openrouter (multi-modeles)"
echo " 5) deepseek (bon rapport qualite/prix)"
echo " 6) minimax (MiniMax-M1 / m2.7)"
echo ""
echo -ne " Choix: "
read -r svc
@@ -204,12 +210,34 @@ while true; do
2) set_env "TRANSLATION_SERVICE" "deepl" ;;
3) set_env "TRANSLATION_SERVICE" "openai" ;;
4) set_env "TRANSLATION_SERVICE" "openrouter" ;;
5) set_env "TRANSLATION_SERVICE" "deepseek" ;;
6) set_env "TRANSLATION_SERVICE" "minimax" ;;
*) echo -e " ${RED}Choix invalide${NC}" ;;
esac
echo -e " ${GREEN}Service mis a jour${NC}"
echo -e " ${YELLOW}Redemarre: docker compose restart backend${NC}"
;;
7)
echo -e "\n${BOLD}--- DeepSeek ---${NC}"
echo " Ou: https://platform.deepseek.com/api_keys"
echo " Modele par defaut: deepseek-chat"
ask_key "DeepSeek" "DEEPSEEK_API_KEY"
if [ -n "$(get_env DEEPSEEK_API_KEY)" ]; then
set_env "DEEPSEEK_ENABLED" "true"
echo -e " ${GREEN}Provider DeepSeek active${NC}"
fi
;;
8)
echo -e "\n${BOLD}--- Minimax (m2.7 / MiniMax-M1) ---${NC}"
echo " Ou: https://platform.minimaxi.com/"
echo " Modele par defaut: MiniMax-M1"
ask_key "Minimax" "MINIMAX_API_KEY"
if [ -n "$(get_env MINIMAX_API_KEY)" ]; then
set_env "MINIMAX_ENABLED" "true"
echo -e " ${GREEN}Provider Minimax active${NC}"
fi
;;
9)
echo ""
echo -e "${RED}${BOLD}ATTENTION: Secrets visibles!${NC}"
echo -ne "${YELLOW}Taper 'oui' pour continuer:${NC} "

View File

@@ -157,12 +157,14 @@ echo ""
echo -e "${BOLD}--- Services de traduction ---${NC}"
echo ""
echo "Quel service de traduction par defaut ?"
echo " 1) ollama (local, gratuit)"
echo " 2) google (API payante)"
echo " 3) deepl (API payante, haute qualite)"
echo " 4) openai (GPT, payant)"
echo " 1) ollama (local, gratuit)"
echo " 2) google (API gratuite via deep_translator)"
echo " 3) deepl (haute qualite, 500k car/mois gratuit)"
echo " 4) openai (GPT, payant)"
echo " 5) openrouter (multi-modeles, payant)"
ask "Choix (1-5)" "1" TRANSLATION_CHOICE
echo " 6) deepseek (tres bon rapport qualite/prix)"
echo " 7) minimax (MiniMax-M1 / m2.7, payant)"
ask "Choix (1-7)" "1" TRANSLATION_CHOICE
case "$TRANSLATION_CHOICE" in
1) TRANSLATION_SERVICE="ollama" ;;
@@ -170,12 +172,16 @@ case "$TRANSLATION_CHOICE" in
3) TRANSLATION_SERVICE="deepl" ;;
4) TRANSLATION_SERVICE="openai" ;;
5) TRANSLATION_SERVICE="openrouter" ;;
6) TRANSLATION_SERVICE="deepseek" ;;
7) TRANSLATION_SERVICE="minimax" ;;
*) TRANSLATION_SERVICE="ollama" ;;
esac
DEEPL_API_KEY=""
OPENAI_API_KEY=""
OPENROUTER_API_KEY=""
DEEPSEEK_API_KEY=""
MINIMAX_API_KEY=""
if [ "$TRANSLATION_SERVICE" = "ollama" ]; then
ask "URL Ollama" "http://ollama:11434" OLLAMA_BASE_URL
@@ -194,6 +200,14 @@ if [ "$TRANSLATION_SERVICE" = "openrouter" ] || [ "$TRANSLATION_SERVICE" = "all"
ask "Cle API OpenRouter (laisser vide si pas maintenant)" "" OPENROUTER_API_KEY
fi
if [ "$TRANSLATION_SERVICE" = "deepseek" ] || [ "$TRANSLATION_SERVICE" = "all" ]; then
ask "Cle API DeepSeek (laisser vide si pas maintenant)" "" DEEPSEEK_API_KEY
fi
if [ "$TRANSLATION_SERVICE" = "minimax" ] || [ "$TRANSLATION_SERVICE" = "all" ]; then
ask "Cle API Minimax (laisser vide si pas maintenant)" "" MINIMAX_API_KEY
fi
# ===========================================
# ETAPE 6 : Monitoring
# ===========================================
@@ -291,6 +305,16 @@ OPENAI_API_KEY=${OPENAI_API_KEY}
OPENAI_MODEL=gpt-4o-mini
OPENROUTER_API_KEY=${OPENROUTER_API_KEY}
# DeepSeek
DEEPSEEK_ENABLED=${DEEPSEEK_ENABLED:-false}
DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
DEEPSEEK_MODEL=deepseek-chat
# Minimax (m2.7 / MiniMax-M1)
MINIMAX_ENABLED=${MINIMAX_ENABLED:-false}
MINIMAX_API_KEY=${MINIMAX_API_KEY}
MINIMAX_MODEL=MiniMax-M1
# Upload
MAX_FILE_SIZE_MB=50
ALLOWED_EXTENSIONS=.docx,.xlsx,.pptx

View File

@@ -69,6 +69,16 @@ def _auto_register_providers() -> None:
register_openai_provider()
if ProvidersConfig.DEEPSEEK_ENABLED and ProvidersConfig.DEEPSEEK_API_KEY:
from .deepseek_provider import register_deepseek_provider
register_deepseek_provider()
if ProvidersConfig.MINIMAX_ENABLED and ProvidersConfig.MINIMAX_API_KEY:
from .minimax_provider import register_minimax_provider
register_minimax_provider()
_auto_register_providers()

View File

@@ -96,12 +96,31 @@ class ProvidersConfig:
OPENROUTER_API_KEY: str = os.getenv("OPENROUTER_API_KEY", "")
OPENROUTER_MODEL: str = os.getenv("OPENROUTER_MODEL", "deepseek/deepseek-chat")
# DeepSeek (direct API)
DEEPSEEK_ENABLED: bool = os.getenv("DEEPSEEK_ENABLED", "false").lower() == "true"
DEEPSEEK_API_KEY: str = os.getenv("DEEPSEEK_API_KEY", "")
DEEPSEEK_MODEL: str = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
DEEPSEEK_BASE_URL: str = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1")
DEEPSEEK_TIMEOUT: int = int(os.getenv("DEEPSEEK_TIMEOUT", "60"))
DEEPSEEK_MAX_RETRIES: int = int(os.getenv("DEEPSEEK_MAX_RETRIES", "3"))
DEEPSEEK_RETRY_DELAY: float = float(os.getenv("DEEPSEEK_RETRY_DELAY", "1.0"))
# Minimax (direct API - m2.7, MiniMax-M1)
MINIMAX_ENABLED: bool = os.getenv("MINIMAX_ENABLED", "false").lower() == "true"
MINIMAX_API_KEY: str = os.getenv("MINIMAX_API_KEY", "")
MINIMAX_MODEL: str = os.getenv("MINIMAX_MODEL", "MiniMax-M1")
MINIMAX_BASE_URL: str = os.getenv("MINIMAX_BASE_URL", "https://api.minimax.chat/v1")
MINIMAX_GROUP_ID: str = os.getenv("MINIMAX_GROUP_ID", "")
MINIMAX_TIMEOUT: int = int(os.getenv("MINIMAX_TIMEOUT", "60"))
MINIMAX_MAX_RETRIES: int = int(os.getenv("MINIMAX_MAX_RETRIES", "3"))
MINIMAX_RETRY_DELAY: float = float(os.getenv("MINIMAX_RETRY_DELAY", "1.0"))
# Fallback chain configuration
# General fallback chain (backward compatibility)
FALLBACK_CHAIN: List[str] = [
name.strip()
for name in os.getenv(
"PROVIDER_FALLBACK_CHAIN", "google,deepl,openai,ollama,openrouter"
"PROVIDER_FALLBACK_CHAIN", "google,deepl,deepseek,minimax,openai,ollama,openrouter"
).split(",")
if name.strip()
]
@@ -186,6 +205,18 @@ class ProvidersConfig:
base_url="https://openrouter.ai/api/v1",
model=cls.OPENROUTER_MODEL,
),
"deepseek": ProviderSettings(
enabled=cls.DEEPSEEK_ENABLED,
api_key=cls.DEEPSEEK_API_KEY if cls.DEEPSEEK_API_KEY else None,
base_url=cls.DEEPSEEK_BASE_URL,
model=cls.DEEPSEEK_MODEL,
),
"minimax": ProviderSettings(
enabled=cls.MINIMAX_ENABLED,
api_key=cls.MINIMAX_API_KEY if cls.MINIMAX_API_KEY else None,
base_url=cls.MINIMAX_BASE_URL,
model=cls.MINIMAX_MODEL,
),
}
return settings_map.get(provider_name.lower(), ProviderSettings())
@@ -206,7 +237,7 @@ class ProvidersConfig:
return False
# Providers requiring API keys
providers_requiring_key = {"deepl", "openai", "openrouter", "google_cloud"}
providers_requiring_key = {"deepl", "openai", "openrouter", "google_cloud", "deepseek", "minimax"}
if provider_name.lower() in providers_requiring_key:
return bool(settings.api_key)

View File

@@ -0,0 +1,244 @@
"""
DeepSeek Provider - Cloud LLM translation via DeepSeek API.
DeepSeek uses an OpenAI-compatible Chat Completions API.
"""
import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from requests.exceptions import Timeout, ConnectionError as RequestsConnectionError
from core.logging import get_logger
from .base import TranslationProvider
from .schemas import ProviderHealthStatus, TranslationRequest, TranslationResponse
logger = get_logger(__name__)
DEEPSEEK_RATE_LIMITED = "DEEPSEEK_RATE_LIMITED"
DEEPSEEK_INVALID_KEY = "DEEPSEEK_INVALID_KEY"
DEEPSEEK_TIMEOUT = "DEEPSEEK_TIMEOUT"
DEEPSEEK_SERVICE_ERROR = "DEEPSEEK_SERVICE_ERROR"
_RETRYABLE_ERRORS = {DEEPSEEK_RATE_LIMITED, DEEPSEEK_TIMEOUT, DEEPSEEK_SERVICE_ERROR}
DEFAULT_TRANSLATION_PROMPT = """You are a professional translator. Translate the following text from {source_lang} to {target_lang}.
Rules:
- Translate ONLY the text, do not add explanations or notes
- Preserve the original formatting, line breaks, and structure
- Maintain the original tone and style
- For technical terms, use the standard translation in the target language
- If the text contains proper nouns or brand names, keep them unchanged unless there's a well-known translation"""
def _get_language_name(code: str) -> str:
language_names = {
"en": "English", "fr": "French", "es": "Spanish", "de": "German",
"it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian",
"zh": "Chinese", "ja": "Japanese", "ko": "Korean", "ar": "Arabic",
"hi": "Hindi", "tr": "Turkish", "pl": "Polish", "vi": "Vietnamese",
"th": "Thai", "uk": "Ukrainian", "cs": "Czech", "sv": "Swedish",
"ro": "Romanian", "hu": "Hungarian", "el": "Greek", "he": "Hebrew",
}
return language_names.get(code.split("-")[0].lower(), code)
class DeepSeekProviderError(Exception):
def __init__(self, code: str, message: str, details: Optional[Dict[str, Any]] = None):
self.code = code
self.message = message
self.details = details or {}
super().__init__(message)
class DeepSeekTranslationProvider(TranslationProvider):
"""
DeepSeek translation provider using OpenAI-compatible API.
Supports DeepSeek Chat and DeepSeek Reasoner models.
"""
def __init__(
self,
api_key: str,
model: str = "deepseek-chat",
timeout: int = 60,
max_retries: int = 3,
retry_delay: float = 1.0,
base_url: str = "https://api.deepseek.com/v1",
):
if not api_key or not api_key.strip():
raise ValueError("DeepSeek API key cannot be empty")
self._api_key = api_key
self._model = model
self._base_url = base_url.rstrip("/")
self._provider_name = "deepseek"
self._timeout = timeout
self._max_retries = max_retries
self._retry_delay = retry_delay
self._health_cache: Dict[str, Any] = {}
self._health_cache_ttl = 60
self._health_cache_lock = threading.Lock()
def _make_api_request(self, text: str, system_prompt: str) -> tuple:
if not text or not text.strip():
return text, {}
url = f"{self._base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self._model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text},
],
"temperature": 0.3,
"max_tokens": 4096,
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=self._timeout)
if response.status_code == 401:
raise DeepSeekProviderError(DEEPSEEK_INVALID_KEY, "Cle API DeepSeek invalide.")
if response.status_code == 429:
raise DeepSeekProviderError(DEEPSEEK_RATE_LIMITED, "Limite de requetes DeepSeek atteinte.")
if response.status_code >= 500:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, "Service DeepSeek temporairement indisponible.")
if response.status_code != 200:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, f"Erreur DeepSeek: {response.text[:200]}")
data = response.json()
choices = data.get("choices", [])
if not choices:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, "Reponse DeepSeek vide")
content = choices[0].get("message", {}).get("content", "")
if not content:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, "Reponse DeepSeek vide")
return content.strip(), data.get("usage", {})
except Timeout:
raise DeepSeekProviderError(DEEPSEEK_TIMEOUT, "Delai d'attente DeepSeek depasse.")
except RequestsConnectionError:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, "Service DeepSeek indisponible.")
except DeepSeekProviderError:
raise
except Exception as e:
raise DeepSeekProviderError(DEEPSEEK_SERVICE_ERROR, f"Erreur DeepSeek: {str(e)[:100]}")
def get_name(self) -> str:
return self._provider_name
def is_available(self) -> bool:
try:
headers = {"Authorization": f"Bearer {self._api_key}"}
response = requests.get(f"{self._base_url}/models", headers=headers, timeout=5)
return response.status_code == 200
except Exception:
return False
def translate_text(self, request: TranslationRequest) -> TranslationResponse:
text = request.text
target_language = request.target_language
source_language = request.source_language or "auto"
if not text or not text.strip():
return TranslationResponse(translated_text=text, provider_name=self._provider_name, from_cache=False)
source_lang_name = _get_language_name(source_language)
target_lang_name = _get_language_name(target_language)
custom_prompt = request.metadata.get("custom_prompt") if request.metadata else None
system_prompt = custom_prompt or DEFAULT_TRANSLATION_PROMPT.format(
source_lang=source_lang_name, target_lang=target_lang_name
)
last_error = None
for attempt in range(self._max_retries + 1):
try:
start_time = time.time()
result, usage = self._make_api_request(text, system_prompt)
latency = time.time() - start_time
logger.info("deepseek_translation_success",
chars=len(text), source_lang=source_language, target_lang=target_language,
model=self._model, latency_ms=round(latency * 1000, 2), retries=attempt)
return TranslationResponse(
translated_text=result, provider_name=self._provider_name,
from_cache=False, source_language=source_language)
except DeepSeekProviderError as e:
last_error = e
if e.code not in _RETRYABLE_ERRORS or attempt >= self._max_retries:
break
delay = self._retry_delay * (2 ** attempt)
time.sleep(delay)
return TranslationResponse(
translated_text=text, provider_name=self._provider_name, from_cache=False,
error=last_error.message if last_error else "Unknown error",
error_code=last_error.code if last_error else DEEPSEEK_SERVICE_ERROR)
def translate_batch(self, requests: List[TranslationRequest]) -> List[TranslationResponse]:
return [self.translate_text(req) for req in requests]
def health_check(self) -> ProviderHealthStatus:
start_time = time.time()
try:
headers = {"Authorization": f"Bearer {self._api_key}"}
response = requests.get(f"{self._base_url}/models", headers=headers, timeout=5)
latency_ms = (time.time() - start_time) * 1000
return ProviderHealthStatus(
name=self._provider_name, available=response.status_code == 200,
latency_ms=round(latency_ms, 2), last_check=datetime.now(timezone.utc).isoformat(),
model=self._model)
except Exception as e:
return ProviderHealthStatus(
name=self._provider_name, available=False,
latency_ms=round((time.time() - start_time) * 1000, 2),
error=str(e)[:100], last_check=datetime.now(timezone.utc).isoformat(),
model=self._model)
_provider_instance: Optional[DeepSeekTranslationProvider] = None
_provider_lock = threading.Lock()
def get_deepseek_provider() -> DeepSeekTranslationProvider:
global _provider_instance
if _provider_instance is None:
with _provider_lock:
if _provider_instance is None:
from .config import ProvidersConfig
_provider_instance = DeepSeekTranslationProvider(
api_key=ProvidersConfig.DEEPSEEK_API_KEY,
model=ProvidersConfig.DEEPSEEK_MODEL,
timeout=ProvidersConfig.DEEPSEEK_TIMEOUT,
max_retries=ProvidersConfig.DEEPSEEK_MAX_RETRIES,
retry_delay=ProvidersConfig.DEEPSEEK_RETRY_DELAY,
base_url=ProvidersConfig.DEEPSEEK_BASE_URL,
)
return _provider_instance
def register_deepseek_provider():
from .registry import registry
provider = get_deepseek_provider()
registry.register("deepseek", provider)
return provider
def reset_deepseek_provider():
global _provider_instance
with _provider_lock:
_provider_instance = None

View File

@@ -0,0 +1,247 @@
"""
Minimax Provider - Cloud LLM translation via Minimax API (m2.7).
Minimax uses an OpenAI-compatible Chat Completions API.
"""
import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from requests.exceptions import Timeout, ConnectionError as RequestsConnectionError
from core.logging import get_logger
from .base import TranslationProvider
from .schemas import ProviderHealthStatus, TranslationRequest, TranslationResponse
logger = get_logger(__name__)
MINIMAX_RATE_LIMITED = "MINIMAX_RATE_LIMITED"
MINIMAX_INVALID_KEY = "MINIMAX_INVALID_KEY"
MINIMAX_TIMEOUT = "MINIMAX_TIMEOUT"
MINIMAX_SERVICE_ERROR = "MINIMAX_SERVICE_ERROR"
_RETRYABLE_ERRORS = {MINIMAX_RATE_LIMITED, MINIMAX_TIMEOUT, MINIMAX_SERVICE_ERROR}
DEFAULT_TRANSLATION_PROMPT = """You are a professional translator. Translate the following text from {source_lang} to {target_lang}.
Rules:
- Translate ONLY the text, do not add explanations or notes
- Preserve the original formatting, line breaks, and structure
- Maintain the original tone and style
- For technical terms, use the standard translation in the target language
- If the text contains proper nouns or brand names, keep them unchanged unless there's a well-known translation"""
def _get_language_name(code: str) -> str:
language_names = {
"en": "English", "fr": "French", "es": "Spanish", "de": "German",
"it": "Italian", "pt": "Portuguese", "nl": "Dutch", "ru": "Russian",
"zh": "Chinese", "ja": "Japanese", "ko": "Korean", "ar": "Arabic",
"hi": "Hindi", "tr": "Turkish", "pl": "Polish", "vi": "Vietnamese",
"th": "Thai", "uk": "Ukrainian", "cs": "Czech", "sv": "Swedish",
"ro": "Romanian", "hu": "Hungarian", "el": "Greek", "he": "Hebrew",
}
return language_names.get(code.split("-")[0].lower(), code)
class MinimaxProviderError(Exception):
def __init__(self, code: str, message: str, details: Optional[Dict[str, Any]] = None):
self.code = code
self.message = message
self.details = details or {}
super().__init__(message)
class MinimaxTranslationProvider(TranslationProvider):
"""
Minimax translation provider using OpenAI-compatible API.
Default model: MiniMax-M1 (latest). Also supports m2.7 via env config.
"""
def __init__(
self,
api_key: str,
model: str = "MiniMax-M1",
timeout: int = 60,
max_retries: int = 3,
retry_delay: float = 1.0,
base_url: str = "https://api.minimax.chat/v1",
group_id: str = "",
):
if not api_key or not api_key.strip():
raise ValueError("Minimax API key cannot be empty")
self._api_key = api_key
self._model = model
self._base_url = base_url.rstrip("/")
self._group_id = group_id
self._provider_name = "minimax"
self._timeout = timeout
self._max_retries = max_retries
self._retry_delay = retry_delay
self._health_cache: Dict[str, Any] = {}
self._health_cache_ttl = 60
self._health_cache_lock = threading.Lock()
def _make_api_request(self, text: str, system_prompt: str) -> tuple:
if not text or not text.strip():
return text, {}
url = f"{self._base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self._model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text},
],
"temperature": 0.3,
"max_tokens": 4096,
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=self._timeout)
if response.status_code == 401:
raise MinimaxProviderError(MINIMAX_INVALID_KEY, "Cle API Minimax invalide.")
if response.status_code == 429:
raise MinimaxProviderError(MINIMAX_RATE_LIMITED, "Limite de requetes Minimax atteinte.")
if response.status_code >= 500:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Service Minimax temporairement indisponible.")
if response.status_code != 200:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, f"Erreur Minimax: {response.text[:200]}")
data = response.json()
choices = data.get("choices", [])
if not choices:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Reponse Minimax vide")
content = choices[0].get("message", {}).get("content", "")
if not content:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Reponse Minimax vide")
return content.strip(), data.get("usage", {})
except Timeout:
raise MinimaxProviderError(MINIMAX_TIMEOUT, "Delai d'attente Minimax depasse.")
except RequestsConnectionError:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, "Service Minimax indisponible.")
except MinimaxProviderError:
raise
except Exception as e:
raise MinimaxProviderError(MINIMAX_SERVICE_ERROR, f"Erreur Minimax: {str(e)[:100]}")
def get_name(self) -> str:
return self._provider_name
def is_available(self) -> bool:
try:
headers = {"Authorization": f"Bearer {self._api_key}"}
response = requests.get(f"{self._base_url}/models", headers=headers, timeout=5)
return response.status_code == 200
except Exception:
return False
def translate_text(self, request: TranslationRequest) -> TranslationResponse:
text = request.text
target_language = request.target_language
source_language = request.source_language or "auto"
if not text or not text.strip():
return TranslationResponse(translated_text=text, provider_name=self._provider_name, from_cache=False)
source_lang_name = _get_language_name(source_language)
target_lang_name = _get_language_name(target_language)
custom_prompt = request.metadata.get("custom_prompt") if request.metadata else None
system_prompt = custom_prompt or DEFAULT_TRANSLATION_PROMPT.format(
source_lang=source_lang_name, target_lang=target_lang_name
)
last_error = None
for attempt in range(self._max_retries + 1):
try:
start_time = time.time()
result, usage = self._make_api_request(text, system_prompt)
latency = time.time() - start_time
logger.info("minimax_translation_success",
chars=len(text), source_lang=source_language, target_lang=target_language,
model=self._model, latency_ms=round(latency * 1000, 2), retries=attempt)
return TranslationResponse(
translated_text=result, provider_name=self._provider_name,
from_cache=False, source_language=source_language)
except MinimaxProviderError as e:
last_error = e
if e.code not in _RETRYABLE_ERRORS or attempt >= self._max_retries:
break
delay = self._retry_delay * (2 ** attempt)
time.sleep(delay)
return TranslationResponse(
translated_text=text, provider_name=self._provider_name, from_cache=False,
error=last_error.message if last_error else "Unknown error",
error_code=last_error.code if last_error else MINIMAX_SERVICE_ERROR)
def translate_batch(self, requests: List[TranslationRequest]) -> List[TranslationResponse]:
return [self.translate_text(req) for req in requests]
def health_check(self) -> ProviderHealthStatus:
start_time = time.time()
try:
headers = {"Authorization": f"Bearer {self._api_key}"}
response = requests.get(f"{self._base_url}/models", headers=headers, timeout=5)
latency_ms = (time.time() - start_time) * 1000
return ProviderHealthStatus(
name=self._provider_name, available=response.status_code == 200,
latency_ms=round(latency_ms, 2), last_check=datetime.now(timezone.utc).isoformat(),
model=self._model)
except Exception as e:
return ProviderHealthStatus(
name=self._provider_name, available=False,
latency_ms=round((time.time() - start_time) * 1000, 2),
error=str(e)[:100], last_check=datetime.now(timezone.utc).isoformat(),
model=self._model)
_provider_instance: Optional[MinimaxTranslationProvider] = None
_provider_lock = threading.Lock()
def get_minimax_provider() -> MinimaxTranslationProvider:
global _provider_instance
if _provider_instance is None:
with _provider_lock:
if _provider_instance is None:
from .config import ProvidersConfig
_provider_instance = MinimaxTranslationProvider(
api_key=ProvidersConfig.MINIMAX_API_KEY,
model=ProvidersConfig.MINIMAX_MODEL,
timeout=ProvidersConfig.MINIMAX_TIMEOUT,
max_retries=ProvidersConfig.MINIMAX_MAX_RETRIES,
retry_delay=ProvidersConfig.MINIMAX_RETRY_DELAY,
base_url=ProvidersConfig.MINIMAX_BASE_URL,
group_id=ProvidersConfig.MINIMAX_GROUP_ID,
)
return _provider_instance
def register_minimax_provider():
from .registry import registry
provider = get_minimax_provider()
registry.register("minimax", provider)
return provider
def reset_minimax_provider():
global _provider_instance
with _provider_lock:
_provider_instance = None