一、接口概览
1.1 接口简介
t_text_all是全能型翻译接口，支持多引擎、多格式、多场景的文本翻译，集成了智能路由、质量评估、术语定制、批量处理等高级功能，是企业级翻译解决方案的核心接口。
1.2 核心特性
✅ 多引擎支持：集成10+主流翻译引擎
✅ 智能路由：自动选择最佳翻译引擎
✅ 质量评估：实时评估翻译质量
✅ 术语定制：支持专业领域术语库
✅ 批量处理：高效处理大规模文本
✅ 格式保持：HTML/Markdown/JSON等格式翻译
✅ 实时监控：翻译质量与性能监控
✅ 成本优化：智能成本控制策略
二、准备工作
2.1 环境配置
requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0
pydantic>=2.0.0
aiohttp>=3.8.0
redis>=4.5.0
openai>=1.0.0
google-cloud-translate>=3.0.0
tencentcloud-sdk-python>=3.0.0
alibabacloud_tea_openapi>=0.3.0
numpy>=1.21.0
sentencepiece>=0.1.0
transformers>=4.0.0
2.2 认证配置
config.py
import os
from dotenv import load_dotenv
from typing import Dict, Any, List
load_dotenv()
class TranslationConfig:
# 翻译服务配置
TRANSLATION_SERVICES = {
'baidu': {
'app_id': os.getenv('BAIDU_TRANSLATE_APP_ID'),
'app_key': os.getenv('BAIDU_TRANSLATE_APP_KEY'),
'api_base': 'https://fanyi-api.baidu.com/api/trans/vip/translate',
'enabled': True,
'priority': 3
},
'tencent': {
'secret_id': os.getenv('TENCENT_TRANSLATE_SECRET_ID'),
'secret_key': os.getenv('TENCENT_TRANSLATE_SECRET_KEY'),
'api_base': 'https://tmt.tencentcloudapi.com',
'enabled': True,
'priority': 2
},
'aliyun': {
'access_key_id': os.getenv('ALIYUN_TRANSLATE_ACCESS_KEY_ID'),
'access_key_secret': os.getenv('ALIYUN_TRANSLATE_ACCESS_KEY_SECRET'),
'api_base': 'https://mt.cn-hangzhou.aliyuncs.com',
'enabled': True,
'priority': 3
},
'youdao': {
'app_key': os.getenv('YOUDAO_TRANSLATE_APP_KEY'),
'app_secret': os.getenv('YOUDAO_TRANSLATE_APP_SECRET'),
'api_base': 'https://openapi.youdao.com/api',
'enabled': True,
'priority': 4
},
'google': {
'api_key': os.getenv('GOOGLE_TRANSLATE_API_KEY'),
'api_base': 'https://translation.googleapis.com/language/translate/v2',
'enabled': True,
'priority': 1
},
'openai': {
'api_key': os.getenv('OPENAI_API_KEY'),
'api_base': 'https://api.openai.com/v1/chat/completions',
'enabled': True,
'priority': 1
},
'azure': {
'api_key': os.getenv('AZURE_TRANSLATE_API_KEY'),
'endpoint': os.getenv('AZURE_TRANSLATE_ENDPOINT'),
'location': os.getenv('AZURE_TRANSLATE_LOCATION'),
'enabled': True,
'priority': 2
},
'deepl': {
'api_key': os.getenv('DEEPL_API_KEY'),
'api_base': 'https://api-free.deepl.com/v2/translate',
'enabled': True,
'priority': 1
},
'volcengine': {
'access_key': os.getenv('VOLCENGINE_ACCESS_KEY'),
'secret_key': os.getenv('VOLCENGINE_SECRET_KEY'),
'api_base': 'https://open.volcengineapi.com',
'enabled': True,
'priority': 4
},
'local': {
'model_path': os.getenv('LOCAL_TRANSLATION_MODEL_PATH'),
'enabled': False,
'priority': 5
}
}
# 请求配置
REQUEST_TIMEOUT = 30
MAX_TEXT_LENGTH = 5000
BATCH_SIZE = 50
MAX_CONCURRENT_REQUESTS = 10
# 缓存配置
CACHE_TTL = 86400
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
# 质量配置
MIN_QUALITY_SCORE = 0.7
MAX_RETRIES = 3
FALLBACK_STRATEGY = 'quality' # quality/cost/speed
# 成本配置
COST_LIMITS = {
'daily': 100.0, # 每日成本限制
'monthly': 1000.0
}
三、接口详解
3.1 接口地址
POST /v2/translate/all
3.2 请求参数详解
基础参数
参数名
类型
必填
说明
示例
text
string/list
是
待翻译文本
"Hello World"
source_lang
string
是
源语言
"en"
target_lang
string
是
目标语言
"zh"
format
string
否
文本格式
"text/html/markdown/json"
高级参数
参数名
类型
必填
说明
示例
engine
string
否
指定引擎
"auto/baidu/google/openai"
strategy
string
否
选择策略
"quality/cost/speed/auto"
glossary_id
string
否
术语库ID
"glossary_001"
domain
string
否
专业领域
"finance/medical/legal"
quality_threshold
float
否
质量阈值
0.8
preserve_format
bool
否
保持格式
true
batch_mode
bool
否
批量模式
true
fallback
bool
否
降级处理
true
retry_count
int
否
重试次数
3
3.3 返回参数详解
参数名
类型
说明
示例
success
bool
是否成功
true
code
int
状态码
200
message
string
消息
"成功"
data
object
翻译数据
-
translated_text
string/list
译文
"你好世界"
source_text
string/list
原文
"Hello World"
engine_used
string
使用引擎
"google"
quality_score
float
质量评分
0.92
confidence
float
置信度
0.95
translation_time
float
翻译耗时
0.345
cost
float
翻译成本
0.002
glossary_matches
list
术语匹配
["API", "接口"]
四、完整代码实现
4.1 核心类实现
import requests
import time
import hashlib
import hmac
import json
import random
import asyncio
import aiohttp
import numpy as np
from typing import Dict, Any, List, Optional, Union, Tuple
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from urllib.parse import urlencode
import redis
from openai import OpenAI
from google.cloud import translate_v2 as google_translate
from tencentcloud.common import credential
from tencentcloud.tmt.v20180321 import tmt_client, models
from alibabacloud_tea_openapi import models as aliyun_models
from alibabacloud_mt20180807.client import Client as AliyunClient
from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM
import sentencepiece as spm
@dataclass
class TranslationRequest:
"""翻译请求"""
text: Union[str, List[str]]
source_lang: str
target_lang: str
format: str = "text"
domain: Optional[str] = None
glossary_id: Optional[str] = None
engine: str = "auto"
strategy: str = "auto"
quality_threshold: float = 0.7
preserve_format: bool = True
batch_mode: bool = True
fallback: bool = True
retry_count: int = 3
custom_params: Dict[str, Any] = field(default_factory=dict)
@dataclass
class TranslationResult:
"""翻译结果"""
success: bool
code: int
message: str
data: Dict[str, Any]
source_text: Union[str, List[str]]
translated_text: Union[str, List[str]]
source_lang: str
target_lang: str
engine_used: str
quality_score: Optional[float] = None
confidence: Optional[float] = None
translation_time: float = 0.0
cost: float = 0.0
glossary_matches: Optional[List[str]] = None
engine_ranking: Optional[List[Dict[str, Any]]] = None
warnings: Optional[List[str]] = None
@dataclass
class EnginePerformance:
"""引擎性能指标"""
engine: str
total_requests: int = 0
success_requests: int = 0
total_time: float = 0.0
avg_time: float = 0.0
success_rate: float = 0.0
avg_quality: float = 0.0
total_cost: float = 0.0
last_used: Optional[datetime] = None
class UniversalTranslationAPI:
"""全能型翻译API"""
def __init__(self, config: Dict[str, Any], redis_client=None):
self.config = config
self.redis = redis_client
self.session = requests.Session()
self.async_session = None
self.engine_performance = {}
self.quality_evaluator = TranslationQualityEvaluator()
self.glossary_manager = GlossaryManager(redis_client)
self.cost_tracker = CostTracker()
self.engine_router = EngineRouter(config)
# 初始化各引擎客户端
self._init_engines()
def _init_engines(self):
"""初始化各翻译引擎"""
self.engine_clients = {}
service_config = self.config.get('TRANSLATION_SERVICES', {})
# 百度翻译
if service_config.get('baidu', {}).get('enabled'):
self.engine_clients['baidu'] = BaiduTranslationClient(service_config['baidu'])
# 腾讯翻译
if service_config.get('tencent', {}).get('enabled'):
self.engine_clients['tencent'] = TencentTranslationClient(service_config['tencent'])
# 阿里云翻译
if service_config.get('aliyun', {}).get('enabled'):
self.engine_clients['aliyun'] = AliyunTranslationClient(service_config['aliyun'])
# Google翻译
if service_config.get('google', {}).get('enabled'):
self.engine_clients['google'] = GoogleTranslationClient(service_config['google'])
# OpenAI翻译
if service_config.get('openai', {}).get('enabled'):
self.engine_clients['openai'] = OpenAITranslationClient(service_config['openai'])
# Azure翻译
if service_config.get('azure', {}).get('enabled'):
self.engine_clients['azure'] = AzureTranslationClient(service_config['azure'])
# DeepL翻译
if service_config.get('deepl', {}).get('enabled'):
self.engine_clients['deepl'] = DeepLTranslationClient(service_config['deepl'])
# 火山翻译
if service_config.get('volcengine', {}).get('enabled'):
self.engine_clients['volcengine'] = VolcengineTranslationClient(service_config['volcengine'])
# 本地模型
if service_config.get('local', {}).get('enabled'):
self.engine_clients['local'] = LocalTranslationClient(service_config['local'])
def translate(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str,
**kwargs
) -> TranslationResult:
"""
全能翻译接口
Args:
text: 待翻译文本
source_lang: 源语言
target_lang: 目标语言
**kwargs: 其他参数
Returns:
翻译结果
"""
start_time = time.time()
# 构建请求
request = TranslationRequest(
text=text,
source_lang=source_lang,
target_lang=target_lang,
**kwargs
)
# 检查缓存
cache_key = self._get_cache_key(request)
if self.redis:
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
return TranslationResult(**data)
try:
# 智能路由选择引擎
selected_engines = self.engine_router.select_engines(request)
# 执行翻译
result = self._execute_translation(request, selected_engines)
result.translation_time = time.time() - start_time
# 计算成本
result.cost = self.cost_tracker.calculate_cost(result.engine_used, len(str(text)))
# 缓存结果
if self.redis and result.success:
self.redis.setex(
cache_key,
self.config.get('CACHE_TTL', 86400),
json.dumps(result.__dict__)
)
# 更新引擎性能指标
self._update_engine_performance(result.engine_used, result)
return result
except Exception as e:
translation_time = time.time() - start_time
return TranslationResult(
success=False,
code=500,
message=f"翻译失败: {str(e)}",
data={},
source_text=text,
translated_text=[] if isinstance(text, list) else "",
source_lang=source_lang,
target_lang=target_lang,
engine_used="unknown",
translation_time=translation_time,
cost=0.0
)
def _execute_translation(
self,
request: TranslationRequest,
selected_engines: List[str]
) -> TranslationResult:
"""执行翻译"""
# 应用术语库
if request.glossary_id:
if isinstance(request.text, str):
request.text = self.glossary_manager.apply_glossary(request.text, request.glossary_id)
else:
request.text = [self.glossary_manager.apply_glossary(t, request.glossary_id) for t in request.text]
# 尝试选择的引擎
best_result = None
engine_ranking = []
for engine in selected_engines:
if engine not in self.engine_clients:
continue
try:
# 调用引擎
engine_result = self.engine_clients[engine].translate(request)
# 质量评估
if engine_result.success:
quality_score = self.quality_evaluator.evaluate(
request.text,
engine_result.translated_text,
request.source_lang,
request.target_lang
)
engine_result.quality_score = quality_score
engine_result.confidence = quality_score
# 记录引擎排名
engine_ranking.append({
'engine': engine,
'quality_score': quality_score,
'response_time': engine_result.translation_time
})
# 检查质量阈值
if quality_score >= request.quality_threshold:
best_result = engine_result
best_result.engine_ranking = engine_ranking
break
elif best_result is None or quality_score > best_result.quality_score:
best_result = engine_result
except Exception as e:
print(f"引擎 {engine} 翻译失败: {e}")
continue
# 处理结果
if best_result is not None:
best_result.engine_ranking = engine_ranking
return best_result
else:
# 所有引擎都失败
return TranslationResult(
success=False,
code=500,
message="所有翻译引擎都失败了",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="none",
translation_time=0.0,
cost=0.0
)
def batch_translate(
self,
texts: List[str],
source_lang: str,
target_lang: str,
**kwargs
) -> List[TranslationResult]:
"""批量翻译"""
results = []
batch_size = self.config.get('BATCH_SIZE', 50)
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
result = self.translate(batch_texts, source_lang, target_lang, **kwargs)
results.append(result)
# 避免频率限制
time.sleep(0.1)
return results
async def translate_async(
self,
text: Union[str, List[str]],
source_lang: str,
target_lang: str,
**kwargs
) -> TranslationResult:
"""异步翻译"""
if self.async_session is None:
self.async_session = aiohttp.ClientSession()
# 异步实现逻辑
# 这里可以并行调用多个引擎进行质量对比
pass
def _get_cache_key(self, request: TranslationRequest) -> str:
"""生成缓存键"""
import hashlib
text_str = request.text if isinstance(request.text, str) else '|'.join(request.text)
key_data = f"{text_str}:{request.source_lang}:{request.target_lang}:{request.engine}:{request.strategy}:{request.glossary_id or ''}"
return f"translation:{hashlib.md5(key_data.encode()).hexdigest()}"
def _update_engine_performance(self, engine: str, result: TranslationResult):
"""更新引擎性能指标"""
if engine not in self.engine_performance:
self.engine_performance[engine] = EnginePerformance(engine=engine)
perf = self.engine_performance[engine]
perf.total_requests += 1
perf.last_used = datetime.now()
if result.success:
perf.success_requests += 1
perf.total_time += result.translation_time
perf.avg_time = perf.total_time / perf.success_requests
perf.success_rate = perf.success_requests / perf.total_requests
if result.quality_score:
perf.avg_quality = (
perf.avg_quality * (perf.success_requests - 1) + result.quality_score
) / perf.success_requests
perf.total_cost += result.cost
各引擎客户端实现
class BaiduTranslationClient:
"""百度翻译客户端"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.session = requests.Session()
def translate(self, request: TranslationRequest) -> TranslationResult:
"""百度翻译"""
try:
# 构建请求参数
params = {
'q': request.text if isinstance(request.text, str) else '\n'.join(request.text),
'from': request.source_lang,
'to': request.target_lang,
'appid': self.config['app_id']
}
# 生成签名
salt = str(random.randint(32768, 65536))
sign_str = self.config['app_id'] + params['q'] + salt + self.config['app_key']
sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
params['salt'] = salt
params['sign'] = sign
# 发送请求
start_time = time.time()
response = self.session.post(self.config['api_base'], data=params, timeout=30)
translation_time = time.time() - start_time
if response.status_code == 200:
data = response.json()
if 'trans_result' in data:
trans_result = data['trans_result']
if isinstance(request.text, str):
translated_text = trans_result[0]['dst']
else:
translated_text = [item['dst'] for item in trans_result]
return TranslationResult(
success=True,
code=200,
message="成功",
data=data,
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="baidu",
translation_time=translation_time
)
else:
error_code = data.get('error_code', 0)
error_msg = data.get('error_msg', '未知错误')
return TranslationResult(
success=False,
code=error_code,
message=error_msg,
data=data,
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="baidu",
translation_time=translation_time
)
else:
return TranslationResult(
success=False,
code=response.status_code,
message=f"HTTP {response.status_code}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="baidu",
translation_time=translation_time
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"百度翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="baidu",
translation_time=0.0
)
class GoogleTranslationClient:
"""Google翻译客户端"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.client = google_translate.Client()
def translate(self, request: TranslationRequest) -> TranslationResult:
"""Google翻译"""
try:
start_time = time.time()
if isinstance(request.text, str):
result = self.client.translate(
request.text,
source_language=request.source_lang,
target_language=request.target_lang
)
translated_text = result['translatedText']
else:
results = self.client.translate(
request.text,
source_language=request.source_lang,
target_language=request.target_lang
)
translated_text = [result['translatedText'] for result in results]
translation_time = time.time() - start_time
return TranslationResult(
success=True,
code=200,
message="成功",
data={},
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="google",
translation_time=translation_time
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"Google翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="google",
translation_time=0.0
)
class OpenAITranslationClient:
"""OpenAI翻译客户端"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.client = OpenAI(api_key=config['api_key'])
def translate(self, request: TranslationRequest) -> TranslationResult:
"""OpenAI翻译"""
try:
# 构建提示词
system_prompt = f"""
你是一位专业的翻译专家，请将{request.source_lang}翻译成{request.target_lang}。
保持原文格式和语义准确，专业术语要翻译正确。
"""
if isinstance(request.text, str):
user_content = f"请翻译以下文本：\n\n{request.text}"
else:
user_content = f"请翻译以下文本列表：\n\n" + '\n'.join([f"{i+1}. {text}" for i, text in enumerate(request.text)])
start_time = time.time()
# 调用OpenAI API
response = self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_content}
],
temperature=0.1
)
translation_time = time.time() - start_time
translated_content = response.choices[0].message.content
if isinstance(request.text, str):
translated_text = translated_content.strip()
else:
# 解析批量翻译结果
translated_text = []
lines = translated_content.strip().split('\n')
for line in lines:
if '. ' in line:
text = line.split('. ', 1)[1]
translated_text.append(text)
else:
translated_text.append(line)
return TranslationResult(
success=True,
code=200,
message="成功",
data=response.to_dict(),
source_text=request.text,
translated_text=translated_text,
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="openai",
translation_time=translation_time
)
except Exception as e:
return TranslationResult(
success=False,
code=500,
message=f"OpenAI翻译失败: {str(e)}",
data={},
source_text=request.text,
translated_text=[] if isinstance(request.text, list) else "",
source_lang=request.source_lang,
target_lang=request.target_lang,
engine_used="openai",
translation_time=0.0
)
其他引擎客户端实现类似...
4.2 智能路由系统
class EngineRouter:
"""引擎智能路由器"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.engine_capabilities = self._build_engine_capabilities()
def _build_engine_capabilities(self) -> Dict[str, Dict[str, Any]]:
"""构建引擎能力矩阵"""
return {
'baidu': {
'languages': ['zh', 'en', 'ja', 'ko', 'fr', 'de', 'ru', 'es', 'ar', 'pt'],
'quality': 0.85,
'speed': 0.8,
'cost': 0.001,
'max_length': 5000,
'batch_support': True
},
'google': {
'languages': ['zh', 'en', 'ja', 'ko', 'fr', 'de', 'ru', 'es', 'ar', 'pt', 'it', 'nl'],
'quality': 0.92,
'speed': 0.9,
'cost': 0.003,
'max_length': 30000,
'batch_support': True
},
'openai': {
'languages': ['zh', 'en', 'ja', 'ko', 'fr', 'de', 'ru', 'es', 'ar', 'pt', 'it', 'nl'],
'quality': 0.95,
'speed': 0.7,
'cost': 0.01,
'max_length': 4000,
'batch_support': True
},
'tencent': {
'languages': ['zh', 'en', 'ja', 'ko', 'fr', 'de', 'ru', 'es'],
'quality': 0.82,
'speed': 0.85,
'cost': 0.002,
'max_length': 2000,
'batch_support': True
},
'deepl': {
'languages': ['zh', 'en', 'ja', 'ko', 'fr', 'de', 'ru', 'es', 'it', 'nl', 'pl'],
'quality': 0.94,
'speed': 0.88,
'cost': 0.004,
'max_length': 5000,
'batch_support': True
}
}
def select_engines(self, request: TranslationRequest) -> List[str]:
"""选择适合的翻译引擎"""
available_engines = self._get_available_engines(request)
if request.engine != 'auto':
if request.engine in available_engines:
return [request.engine]
else:
return available_engines
# 根据策略选择引擎
if request.strategy == 'quality':
return self._select_by_quality(available_engines)
elif request.strategy == 'cost':
return self._select_by_cost(available_engines)
elif request.strategy == 'speed':
return self._select_by_speed(available_engines)
else: # auto
return self._select_by_auto(available_engines, request)
def _get_available_engines(self, request: TranslationRequest) -> List[str]:
"""获取可用的引擎列表"""
available = []
for engine, capabilities in self.engine_capabilities.items():
# 检查语言支持
if (request.source_lang in capabilities['languages'] and
request.target_lang in capabilities['languages']):
# 检查文本长度
text_length = len(str(request.text))
if text_length <= capabilities['max_length']:
available.append(engine)
return available
def _select_by_quality(self, engines: List[str]) -> List[str]:
"""按质量选择引擎"""
scored_engines = []
for engine in engines:
score = self.engine_capabilities[engine]['quality']
scored_engines.append((engine, score))
scored_engines.sort(key=lambda x: x[1], reverse=True)
return [engine for engine, score in scored_engines]
def _select_by_cost(self, engines: List[str]) -> List[str]:
"""按成本选择引擎"""
scored_engines = []
for engine in engines:
score = 1.0 / self.engine_capabilities[engine]['cost'] # 成本越低分数越高
scored_engines.append((engine, score))
scored_engines.sort(key=lambda x: x[1], reverse=True)
return [engine for engine, score in scored_engines]
def _select_by_auto(self, engines: List[str], request: TranslationRequest) -> List[str]:
"""自动选择引擎"""
# 综合考虑质量、成本、速度
scored_engines = []
for engine in engines:
capabilities = self.engine_capabilities[engine]
# 计算综合得分
quality_score = capabilities['quality'] * 0.5
cost_score = (1.0 / capabilities['cost']) * 0.3
speed_score = capabilities['speed'] * 0.2
total_score = quality_score + cost_score + speed_score
scored_engines.append((engine, total_score))
scored_engines.sort(key=lambda x: x[1], reverse=True)
return [engine for engine, score in scored_engines]
4.3 质量评估系统
class TranslationQualityEvaluator:
"""翻译质量评估器"""
def __init__(self):
self.metrics = ['accuracy', 'fluency', 'terminology', 'style']
self.models = self._load_models()
def _load_models(self):
"""加载质量评估模型"""
try:
# 加载预训练模型进行质量评估
# 这里使用简化实现，实际可以使用BERTScore等先进方法
return {}
except:
return {}
def evaluate(
self,
source_text: Union[str, List[str]],
translated_text: Union[str, List[str]],
source_lang: str,
target_lang: str
) -> float:
"""评估翻译质量"""
if isinstance(source_text, list) and isinstance(translated_text, list):
# 批量评估
scores = []
for src, tgt in zip(source_text, translated_text):
score = self._evaluate_single(src, tgt, source_lang, target_lang)
scores.append(score)
return sum(scores) / len(scores) if scores else 0.0
else:
# 单文本评估
return self._evaluate_single(source_text, translated_text, source_lang, target_lang)
def _evaluate_single(
self,
source_text: str,
translated_text: str,
source_lang: str,
target_lang: str
) -> float:
"""评估单条翻译质量"""
if len(source_text.strip()) == 0 or len(translated_text.strip()) == 0:
return 0.0
# 计算多个维度的评分
accuracy_score = self._calculate_accuracy_score(source_text, translated_text)
fluency_score = self._calculate_fluency_score(translated_text, target_lang)
terminology_score = self._calculate_terminology_score(source_text, translated_text)
style_score = self._calculate_style_score(source_text, translated_text)
# 加权平均
weights = [0.4, 0.3, 0.2, 0.1] # 准确性权重最高
scores = [accuracy_score, fluency_score, terminology_score, style_score]
return sum(score * weight for score, weight in zip(scores, weights))
def _calculate_accuracy_score(self, source: str, target: str) -> float:
"""计算准确性评分"""
# 基于BLEU评分原理的简化实现
import re
# 分词
source_words = re.findall(r'\w+', source.lower())
target_words = re.findall(r'\w+', target.lower())
if len(source_words) == 0 or len(target_words) == 0:
return 0.0
# 计算词重叠度
common_words = set(source_words) & set(target_words)
precision = len(common_words) / len(target_words) if len(target_words) > 0 else 0.0
recall = len(common_words) / len(source_words) if len(source_words) > 0 else 0.0
# F1分数
if precision + recall > 0:
f1_score = 2 * precision * recall / (precision + recall)
else:
f1_score = 0.0
return min(f1_score * 2, 1.0) # 归一化到0-1
def _calculate_fluency_score(self, text: str, lang: str) -> float:
"""计算流畅性评分"""
if len(text.strip()) == 0:
return 0.0
# 基于句子结构和标点的简单评估
import re
# 检查句子完整性
sentence_endings = ['.', '!', '?', '。', '！', '？']
has_ending = any(text.endswith(char) for char in sentence_endings)
# 检查句子长度合理性
word_count = len(re.findall(r'\w+', text))
if word_count < 3:
length_score = 0.6
elif word_count < 20:
length_score = 0.8
else:
length_score = 0.9
# 检查标点符号使用
punctuation_count = len(re.findall(r'[.!?,;:，。！？；：]', text))
if word_count > 0:
punctuation_score = min(punctuation_count / (word_count / 10), 1.0)
else:
punctuation_score = 0.0
# 综合评分
ending_score = 0.8 if has_ending else 0.6
fluency_score = (ending_score + length_score + punctuation_score) / 3
return min(fluency_score, 1.0)
def _calculate_terminology_score(self, source: str, target: str) -> float:
"""计算术语一致性评分"""
# 这里可以集成术语库进行专业评估
# 简化实现：检查数字、专有名词的一致性
import re
source_numbers = re.findall(r'\d+', source)
target_numbers = re.findall(r'\d+', target)
if len(source_numbers) == 0:
return 0.8 # 没有数字，默认分数
# 检查数字是否匹配
matched_numbers = 0
for num in source_numbers:
if num in target_numbers:
matched_numbers += 1
terminology_score = matched_numbers / len(source_numbers) if len(source_numbers) > 0 else 1.0
return terminology_score
def _calculate_style_score(self, source: str, target: str) -> float:
"""计算风格匹配度评分"""
# 简化实现：检查语气和格式一致性
source_uppercase = sum(1 for c in source if c.isupper())
target_uppercase = sum(1 for c in target if c.isupper())
source_length = len(source)
target_length = len(target)
if source_length == 0 or target_length == 0:
return 0.0
# 大写比例一致性
uppercase_ratio_source = source_uppercase / source_length
uppercase_ratio_target = target_uppercase / target_length
uppercase_score = 1.0 - abs(uppercase_ratio_source - uppercase_ratio_target)
# 长度比例一致性
length_ratio = min(source_length, target_length) / max(source_length, target_length)
return (uppercase_score + length_ratio) / 2
4.4 术语库管理系统
class GlossaryManager:
"""术语库管理器"""
def __init__(self, redis_client):
self.redis = redis_client
self.glossary_prefix = "glossary:"
def create_glossary(self, glossary_id: str, terms: Dict[str, str]) -> bool:
"""创建术语库"""
try:
glossary_key = f"{self.glossary_prefix}{glossary_id}"
self.redis.hset(glossary_key, mapping=terms)
return True
except Exception as e:
print(f"创建术语库失败: {e}")
return False
def get_glossary_terms(self, glossary_id: str) -> Dict[str, str]:
"""获取术语库"""
glossary_key = f"{self.glossary_prefix}{glossary_id}"
terms = self.redis.hgetall(glossary_key)
return {k.decode(): v.decode() for k, v in terms.items()} if terms else {}
def apply_glossary(self, text: str, glossary_id: str) -> str:
"""应用术语库"""
terms = self.get_glossary_terms(glossary_id)
for source, target in terms.items():
# 使用正则表达式进行精确匹配
import re
text = re.sub(rf'\b{re.escape(source)}\b', target, text, flags=re.IGNORECASE)
return text
def batch_apply_glossary(self, texts: List[str], glossary_id: str) -> List[str]:
"""批量应用术语库"""
terms = self.get_glossary_terms(glossary_id)
results = []
for text in texts:
processed_text = self.apply_glossary(text, glossary_id)
results.append(processed_text)
return results
def detect_glossary_matches(self, text: str, glossary_id: str) -> List[str]:
"""检测术语匹配"""
terms = self.get_glossary_terms(glossary_id)
matches = []
for term in terms.keys():
if term.lower() in text.lower():
matches.append(term)
return matches
4.5 成本跟踪系统
class CostTracker:
"""成本跟踪系统"""
def __init__(self):
self.cost_records = {}
self.engine_costs = {
'baidu': 0.001, # 元/字符
'tencent': 0.002, # 元/字符
'aliyun': 0.0015, # 元/字符
'youdao': 0.002, # 元/字符
'google': 0.003, # 元/字符
'openai': 0.01, # 元/字符
'azure': 0.0025, # 元/字符
'deepl': 0.004, # 元/字符
'volcengine': 0.002, # 元/字符
'local': 0.0001 # 元/字符
}
def calculate_cost(self, engine: str, text_length: int) -> float:
"""计算翻译成本"""
unit_cost = self.engine_costs.get(engine, 0.001)
return unit_cost * text_length
def record_cost(self, engine: str, cost: float, timestamp: datetime = None):
"""记录成本"""
if timestamp is None:
timestamp = datetime.now()
date_key = timestamp.strftime('%Y-%m-%d')
month_key = timestamp.strftime('%Y-%m')
# 记录每日成本
if date_key not in self.cost_records:
self.cost_records[date_key] = {}
if engine not in self.cost_records[date_key]:
self.cost_records[date_key][engine] = 0.0
self.cost_records[date_key][engine] += cost
# 记录月度成本
if month_key not in self.cost_records:
self.cost_records[month_key] = {}
if engine not in self.cost_records[month_key]:
self.cost_records[month_key][engine] = 0.0
self.cost_records[month_key][engine] += cost
def get_daily_cost(self, date: str = None) -> float:
"""获取指定日期的总成本"""
if date is None:
date = datetime.now().strftime('%Y-%m-%d')
if date in self.cost_records:
return sum(self.cost_records[date].values())
return 0.0
def get_monthly_cost(self, month: str = None) -> float:
"""获取指定月份的总成本"""
if month is None:
month = datetime.now().strftime('%Y-%m')
if month in self.cost_records:
return sum(self.cost_records[month].values())
return 0.0
def check_cost_limit(self, limit_type: str = 'daily', limit_amount: float = 100.0) -> bool:
"""检查成本限制"""
if limit_type == 'daily':
current_cost = self.get_daily_cost()
elif limit_type == 'monthly':
current_cost = self.get_monthly_cost()
else:
return True
return current_cost < limit_amount
五、使用示例
5.1 基础使用
初始化API
api = UniversalTranslationAPI(TranslationConfig, redis_client)
简单翻译
result = api.translate(
text="Hello, world! This is a translation example.",
source_lang="en",
target_lang="zh"
)
if result.success:
print(f"原文: {result.source_text}")
print(f"译文: {result.translated_text}")
print(f"使用引擎: {result.engine_used}")
print(f"质量评分: {result.quality_score:.2f}")
print(f"翻译耗时: {result.translation_time:.3f}s")
print(f"翻译成本: ¥{result.cost:.4f}")
批量翻译
texts = [
"Good morning!",
"How are you today?",
"Have a nice day!",
"Thank you very much!"
]
results = api.batch_translate(texts, "en", "zh")
for i, result in enumerate(results):
if result.success:
print(f"文本{i+1}: {result.translated_text}")
指定引擎
result = api.translate(
text="The quick brown fox jumps over the lazy dog.",
source_lang="en",
target_lang="zh",
engine="openai"
)
5.2 高级功能
质量优先翻译
result = api.translate(
text="This is a technical document about artificial intelligence.",
source_lang="en",
target_lang="zh",
strategy="quality",
quality_threshold=0.9
)
成本优先翻译
result = api.translate(
text="A simple greeting message for our application.",
source_lang="en",
target_lang="zh",
strategy="cost"
)
使用术语库
api.glossary_manager.create_glossary("tech_glossary", {
"API": "应用程序接口",
"AI": "人工智能",
"Machine Learning": "机器学习",
"Cloud Computing": "云计算"
})
result = api.translate(
text="We use AI and Machine Learning to build better APIs.",
source_lang="en",
target_lang="zh",
glossary_id="tech_glossary"
)
获取引擎性能报告
performance = api.engine_performance
for engine, perf in performance.items():
print(f"{engine}: 成功率{perf.success_rate:.1%}, 平均质量{perf.avg_quality:.2f}, 总成本¥{perf.total_cost:.2f}")
检查成本限制
if api.cost_tracker.check_cost_limit('daily', 50.0):
print("今日成本在限制范围内")
else:
print("今日成本已超限")
六、性能优化
6.1 缓存优化
class SmartTranslationCache:
"""智能翻译缓存"""
def __init__(self, redis_client):
self.redis = redis_client
self.memory_cache = {}
self.memory_ttl = 300 # 5分钟
self.redis_ttl = 86400 # 24小时
def get_translation(self, cache_key: str):
"""获取翻译缓存"""
# 检查内存缓存
if cache_key in self.memory_cache:
data, expire_time = self.memory_cache[cache_key]
if time.time() < expire_time:
return data
# 检查Redis缓存
if self.redis:
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
# 更新内存缓存
self.memory_cache[cache_key] = (data, time.time() + self.memory_ttl)
return data
return None
def set_translation(self, cache_key: str, data: Dict[str, Any], ttl: int = None):
"""设置翻译缓存"""
if ttl is None:
ttl = self.redis_ttl
# 设置内存缓存
self.memory_cache[cache_key] = (data, time.time() + self.memory_ttl)
# 设置Redis缓存
if self.redis:
self.redis.setex(cache_key, ttl, json.dumps(data))
6.2 异步优化
async def async_batch_translate(api, texts: List[str], source_lang: str, target_lang: str) -> List[TranslationResult]:
"""异步批量翻译"""
tasks = []
for text in texts:
task = api.translate_async(text, source_lang, target_lang)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
processed_results = []
for result in results:
if isinstance(result, Exception):
processed_results.append(TranslationResult(
success=False,
code=500,
message=str(result),
data={},
source_text="",
translated_text="",
source_lang=source_lang,
target_lang=target_lang,
engine_used="unknown",
translation_time=0.0
))
else:
processed_results.append(result)
return processed_results
七、最佳实践总结
7.1 核心优势
智能路由：自动选择最佳翻译引擎
质量保障：实时评估翻译质量
成本控制：精确跟踪翻译成本
术语管理：支持专业领域术语库
性能优化：多级缓存和异步处理
7.2 使用建议
质量优先：技术文档使用OpenAI/Google引擎
成本优先：日常文本使用百度/腾讯引擎
术语定制：专业领域必须使用术语库
批量处理：大规模翻译使用批量接口
成本监控：定期检查翻译成本
7.3 部署建议
Redis缓存：使用Redis存储缓存和术语库
连接池：配置HTTP连接池提高性能
监控告警：设置成本和质量告警
负载均衡：多引擎部署确保高可用
附录：快速开始模板
quick_start.py
from universal_translation import UniversalTranslationAPI, TranslationConfig
import redis
1. 初始化
redis_client = redis.Redis.from_url(TranslationConfig.REDIS_URL)
api = UniversalTranslationAPI(TranslationConfig, redis_client)
2. 基础翻译
result = api.translate(
text="Hello, world! This is a translation example.",
source_lang="en",
target_lang="zh"
)
if result.success:
print(f"译文: {result.translated_text}")
print(f"引擎: {result.engine_used}")
print(f"质量: {result.quality_score:.2f}")
print(f"成本: ¥{result.cost:.4f}")
3. 批量翻译
texts = ["Text 1", "Text 2", "Text 3"]
results = api.batch_translate(texts, "en", "zh")
4. 高级功能
result = api.translate(
text="AI and Machine Learning",
source_lang="en",
target_lang="zh",
strategy="quality",
glossary_id="tech_glossary"
)
通过本攻略，您应该能够：
理解全能型翻译接口的完整功能架构
实现智能路由和多引擎融合翻译
构建高质量的企业级翻译解决方案
有效控制翻译成本和质量
在实际业务中灵活应用翻译服务
建议根据实际业务需求配置合适的翻译引擎组合，并遵循最佳实践确保系统的稳定性和可维护性。
