3e56fed541
新通道:腾讯 MaaS u2 模型(云知声),OpenAI 兼容协议 - 端点:https://maas-api.hivoice.cn/v1 - 模型:u2(翻译专用,实测 + 锁定 prompt 后译文质量稳定) - 备用链路:TMT 配额耗尽 / TMT 失败时自动降级到 MaaS 关键 prompt 工程(锁定): - 必须用 user 提供的固定中文 prompt,否则 u2 会把译文放进 reasoning_content 而 content 返乱码 - 限定只接 EN/JA → ZH - 中文输入固定返回拒绝文案 新增/改动: - backend/app/services/translation/tencent_maas.py: 新建 - backend/app/services/translation/service.py: 备用链 maas → local,初始化失败友好降级 - backend/app/config.py: 加 tencent_maas_* 4 个配置 - .env.example: 文档化
143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
"""腾讯 MaaS 翻译(OpenAI 兼容协议)。
|
|
|
|
- 端点:https://maas-api.hivoice.cn/v1
|
|
- 模型:u2(翻译专用)
|
|
- 鉴权:Bearer token(api_key 直接当 Bearer)
|
|
- 请求:POST /chat/completions,system prompt 告诉模型做翻译
|
|
|
|
设计上独立于 LlmClient(不走 agnes_* 配置),专门走 tencent_maas_* 配置,
|
|
避免和 LLM 智能增强共用 client 的节流。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import random
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
from app.services.translation.base import BaseTranslator, TranslationResult
|
|
|
|
logger = logging.getLogger("news.translate.tencent_maas")
|
|
|
|
|
|
# 简单的源/目标语言映射(MaaS 模型期望 ISO 639-1 代码)
|
|
_LANG_MAP = {
|
|
"en": "English",
|
|
"zh": "Chinese",
|
|
"ja": "Japanese",
|
|
"ko": "Korean",
|
|
"fr": "French",
|
|
"de": "German",
|
|
"es": "Spanish",
|
|
"ru": "Russian",
|
|
"ar": "Arabic",
|
|
}
|
|
|
|
|
|
def _lang_label(code: str) -> str:
|
|
"""把 ISO 639-1 转成自然语言名(给模型做 prompt 用)。"""
|
|
if not code or code == "auto":
|
|
return "the source language"
|
|
c = code.split("-")[0].lower()
|
|
return _LANG_MAP.get(c, c)
|
|
|
|
|
|
# 经过反复测试,这个 prompt 是云知声 u2 模型的关键:
|
|
# - 明确禁止 reasoning / 分析 / 注释(否则模型会把译文放进 reasoning_content)
|
|
# - 限定只接 EN/JA → ZH(对应用户场景)
|
|
# - 非英日输入时返回固定拒绝文案(便于上层识别)
|
|
_SYSTEM_PROMPT = """你是一个即时翻译助手。对于用户输入的英文或日文文章,请直接输出对应的简体中文译文。严格遵守:
|
|
|
|
不要进行任何分步思考、不要输出分析、不要添加注释或说明。
|
|
|
|
只输出中文译文本身,不包含任何额外文字(如"翻译结果:"、"以下是中文:"等)。
|
|
|
|
如果输入内容既非英文也非日文,仅回复:"仅支持英文或日文翻译为中文。" """
|
|
|
|
|
|
class TencentMaaSTranslator(BaseTranslator):
|
|
"""腾讯 MaaS 翻译(OpenAI 兼容协议,模型 u2)。"""
|
|
|
|
name = "tencent_maas"
|
|
|
|
def __init__(self):
|
|
if not settings.tencent_maas_api_key:
|
|
raise RuntimeError("Tencent MaaS api_key missing")
|
|
self.api_key = settings.tencent_maas_api_key
|
|
self.base_url = settings.tencent_maas_base_url.rstrip("/")
|
|
self.model = settings.tencent_maas_model
|
|
self.interval_sec = settings.tencent_maas_interval_sec
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self.api_key)
|
|
|
|
async def translate(
|
|
self, text: str, source: str = "auto", target: str = "zh"
|
|
) -> TranslationResult:
|
|
"""翻译接口。
|
|
|
|
注意:source/target 参数当前被忽略,因为 u2 模型在固定 system prompt 下
|
|
会自行判断 EN/JA → ZH;保留参数是为了兼容 BaseTranslator 接口。
|
|
"""
|
|
if not text.strip():
|
|
return TranslationResult(text=text, engine=self.name, chars=0)
|
|
|
|
if not self.is_configured():
|
|
raise RuntimeError("Tencent MaaS api_key missing")
|
|
|
|
# 固定 system prompt(经过反复测试,这套 prompt 是云知声 u2 模型唯一能稳定输出
|
|
# 译文到 content 字段的写法;改 prompt 格式会导致模型把译文放进 reasoning_content)
|
|
system = _SYSTEM_PROMPT
|
|
user = text
|
|
|
|
url = f"{self.base_url}/chat/completions"
|
|
payload = {
|
|
"model": self.model,
|
|
"messages": [
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": user},
|
|
],
|
|
"temperature": 0.0,
|
|
"max_tokens": max(256, len(text) * 3),
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
# 简单串行 + 重试 1 次
|
|
last_exc: Exception | None = None
|
|
for attempt in range(2):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
r = await client.post(url, json=payload, headers=headers)
|
|
if r.status_code >= 500:
|
|
raise RuntimeError(f"TencentMaas 5xx: {r.status_code} {r.text[:200]}")
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"TencentMaas {r.status_code}: {r.text[:300]}")
|
|
data = r.json()
|
|
content = (
|
|
data.get("choices", [{}])[0]
|
|
.get("message", {})
|
|
.get("content", "")
|
|
.strip()
|
|
)
|
|
if not content:
|
|
raise RuntimeError(f"TencentMaas empty content: {r.text[:300]}")
|
|
# 节流(避免被 MaaS 限流)
|
|
await asyncio.sleep(self.interval_sec)
|
|
return TranslationResult(
|
|
text=content, engine=self.name, chars=len(text), cached=False
|
|
)
|
|
except Exception as e:
|
|
last_exc = e
|
|
logger.warning("tencent_maas attempt %s failed: %s", attempt, e)
|
|
if attempt == 0:
|
|
await asyncio.sleep(0.5 + random.random())
|
|
else:
|
|
raise
|
|
# 不可达
|
|
assert last_exc is not None
|
|
raise last_exc |