本文目录导读:

基础重试逻辑(Python示例)
import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
# 方法1:手动重试
def translate_with_retry(text, max_retries=3):
for attempt in range(max_retries):
try:
# 调用翻译API
result = translate_api("Hello World")
return result
except Exception as e:
if attempt == max_retries - 1: # 最后一次尝试失败
raise e
wait_time = 2 ** attempt # 指数退避
time.sleep(wait_time)
# 方法2:使用重试库(推荐)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def translate_with_tenacity(text):
return translate_api(text)
完整重试策略
import random
from typing import Optional
class TranslationService:
def __init__(self):
self.retry_count = 0
self.max_retries = 3
self.backoff_factor = 0.5 # 指数退避因子
def translate_with_retry(self, text: str, target_lang: str = "zh-CN") -> Optional[str]:
"""带重试的翻译函数"""
for attempt in range(self.max_retries):
try:
# 1. 尝试主翻译服务
result = self._translate_primary(text, target_lang)
if result:
return result
except ConnectionError:
# 2. 网络错误 - 指数退避
wait_time = self.backoff_factor * (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait_time)
continue
except Exception as e:
# 3. 其他错误 - 记录日志
self.log_error(f"翻译失败: {str(e)}")
if attempt < self.max_retries - 1:
# 尝试备用翻译服务
result = self._translate_fallback(text, target_lang)
if result:
return result
return None # 所有重试失败
def _translate_primary(self, text: str, lang: str) -> str:
"""主翻译服务"""
# 实现实际的API调用
pass
def _translate_fallback(self, text: str, lang: str) -> str:
"""备用翻译服务"""
# 使用备用API
pass
失败处理策略
A. 指数退避 + 抖动
def exponential_backoff_with_jitter(attempt):
base_delay = 1.0 # 基础延迟
max_delay = 30.0 # 最大延迟
jitter = random.uniform(0, 0.1) # 随机抖动
delay = min(base_delay * (2 ** attempt) + jitter, max_delay)
return delay
B. 备用服务切换
def get_translation_service():
"""获取可用的翻译服务"""
services = [
{"name": "google", "url": "..."},
{"name": "deepl", "url": "..."},
{"name": "azure", "url": "..."},
]
for service in services:
if self.check_service_health(service):
return service
return None # 所有服务都不可用
C. 缓存机制
from functools import lru_cache
import hashlib
@lru_cache(maxsize=1000)
def get_cached_translation(text: str, target_lang: str) -> str:
"""缓存翻译结果避免重复请求"""
key = hashlib.md5(f"{text}_{target_lang}".encode()).hexdigest()
# 先查缓存
cached = cache.get(key)
if cached:
return cached
# 缓存未命中则翻译
result = translate_with_retry(text, target_lang)
cache.set(key, result, expire=3600) # 缓存1小时
return result
监控和日志
import logging
from datetime import datetime
class TranslationMonitor:
def __init__(self):
self.success_count = 0
self.failure_count = 0
self.last_failure_time = None
def record_success(self):
self.success_count += 1
def record_failure(self, error: Exception):
self.failure_count += 1
self.last_failure_time = datetime.now()
# 记录详细日志
logging.error(f"翻译失败: {str(error)}")
# 超过阈值发送告警
if self.failure_count > 10:
self.send_alert("翻译服务频繁失败")
配置建议
# config.py
TRANSLATION_CONFIG = {
"retry_policy": {
"max_attempts": 3,
"backoff_factor": 1.5,
"jitter": True,
"retry_on": ["ConnectionError", "Timeout", "5xx_status"]
},
"fallback_services": ["google", "azure", "local"],
"cache_enabled": True,
"cache_ttl": 3600, # 秒
"timeout": 10, # API超时时间
}
最佳实践建议
-
分级重试策略
- 第1次:立即重试(可能是瞬时问题)
- 第2次:等待2秒后重试
- 第3次:等待5秒后重试
-
失败降级方案
- 主服务失败 → 切备用服务
- 所有服务失败 → 返回原文或本地字典
- 完全失败 → 提示用户稍后重试
-
健康检查
def check_service_health(service_url): try: response = requests.get(f"{service_url}/health", timeout=5) return response.status_code == 200 except: return False
选择哪种方案取决于具体需求,简单的应用可以使用基础重试,生产环境建议使用完整的重试策略。
标签: 翻译重试
版权声明:除非特别标注,否则均为本站原创文章,转载时请以链接形式注明文章出处。