feat(translate): 加 Agnes 翻译 fallback,buffer 改 0.5

腾讯 TMT 月度配额快满时(腾讯后台口径已用 2M/5M),翻译降级链:

1. tencent TMT(主,按月配额)
2. tencent MaaS u2(第二级,翻译专用,无配额)
3. agnes 通用 LLM(第三级,质量次之但够用)
4. local NLLB(最后兜底)

新增 backend/app/services/translation/agnes.py: AgnesTranslator
复用 LlmClient 做限速 + 重试,系统 prompt 强约束只输出译文,
去除 "以下是翻译" 等常见 LLM 翻译前缀。

service.py 改动:
- fallback 链 maas -> agnes -> local
- cache 接受 agnes 结果(30天)
- add_usage 只算 tencent TMT

buffer 调整: TENCENT_TMT_QUOTA_BUFFER 0.05 -> 0.5
腾讯云后台按请求字节计费,与我们 redis 字符累加口径差约 2.5x;
按腾讯后台口径 redis 累加到 1M 字符即触发降级(对应腾讯约 2.5M 字节 =50% 用量),
留足 buffer,避免月底真爆。

backend/app/services/translation/agnes.py

@@ -0,0 +1,118 @@
+"""Agnes 翻译后端(通用 LLM 做翻译,作为 TMT/MaaS 之后的第三级 fallback)。
+
+- 端点:settings.agnes_base_url/v1/chat/completions
+- 模型:settings.agnes_chat_model(agnes-2.0-flash)
+- 鉴权:Bearer AGNES_API_KEY
+- 复用 LlmClient 的限速(interval_sec)
+
+适用场景:
+- TMT 月配额快满(>95%)时,先走 MaaS(u2)
+- MaaS 不可用 / 失败时,降到 Agnes(通用 LLM 翻译,质量次之,但够用)
+- Agnes 不可用时,本地 NLLB 最后兜底
+
+注意:
+- Agnes 不是翻译专用模型,翻译质量不如 TMT / MaaS u2
+- 但 Agnes 的 commentary / image / classification 已经走它,客户端有相同 quota 计费体系
+- 用一个特殊 system prompt 强约束:只输出译文
+"""
+from __future__ import annotations
+
+import logging
+
+from app.config import settings
+from app.services.llm.client import LlmClient
+from app.services.translation.base import BaseTranslator, TranslationResult
+
+logger = logging.getLogger("news.translate.agnes")
+
+
+# Agnes 做翻译的 system prompt。
+# 关键约束:只输出译文,不要思考/分析/注释/标签。
+# 这是通用 LLM 翻车最多的地方——会输出 "以下是中文翻译: ..." 前缀,要强约束。
+_SYSTEM_PROMPT = """你是即时翻译引擎。把用户输入的文本翻译成简体中文。
+
+严格规则:
+1. 只输出译文本身,不要输出任何其他文字
+2. 不要输出"以下是翻译"、"译文:"、"[Translation]"、引号、破折号等任何包裹
+3. 不要分步思考、不要分析、不要注释、不要解释
+4. 不要输出 <think>、<reasoning>、<analysis> 等内部标签
+5. 如果输入已经是中文,直接原样返回
+6. 保持原文段落结构(用换行分隔)"""
+
+
+class AgnesTranslator(BaseTranslator):
+    """Agnes 翻译(通用 LLM 兜底)。"""
+
+    name = "agnes"
+
+    def __init__(self):
+        if not settings.agnes_api_key:
+            raise RuntimeError("AGNES_API_KEY 未配置")
+        # 用 chat_model,不复用 image_model
+        self.client = LlmClient(
+            base_url=settings.agnes_base_url,
+            api_key=settings.agnes_api_key,
+            chat_model=settings.agnes_chat_model,
+            image_model=settings.agnes_image_model,
+            interval_sec=settings.llm_interval_sec,
+        )
+
+    def is_configured(self) -> bool:
+        return bool(settings.agnes_api_key)
+
+    async def translate(
+        self, text: str, source: str = "auto", target: str = "zh"
+    ) -> TranslationResult:
+        if not text.strip():
+            return TranslationResult(text=text, engine=self.name, chars=0)
+
+        if not self.is_configured():
+            raise RuntimeError("AGNES_API_KEY 未配置")
+
+        # 长文本截断保护:Agnes max_tokens 上限小,避免超限
+        # 经验值:1 token ≈ 1.5 个英文 / 0.7 个中文字符
+        # 默认 chat 接口 max_tokens=1500,我们给 4096 留余量
+        # 输入截到 6000 字符(MaaS 的限制类似)
+        max_input = 6000
+        truncated = text[:max_input] if len(text) > max_input else text
+
+        # 保留段落结构
+        user = truncated
+
+        # LlmClient.chat 已含 interval_sec 节流
+        content = await self.client.chat(
+            system=_SYSTEM_PROMPT,
+            user=user,
+            temperature=0.1,  # 翻译任务低温度更稳
+            max_tokens=4096,
+        )
+
+        if not content or not content.strip():
+            raise RuntimeError("Agnes 返回空 content")
+
+        # 清洗:模型有时还会输出 markdown 代码块包裹
+        cleaned = content.strip()
+        if cleaned.startswith("```"):
+            # 去掉代码块围栏
+            lines = cleaned.split("\n")
+            cleaned = "\n".join(
+                l for l in lines if not l.strip().startswith("```")
+            ).strip()
+
+        # 去除已知的前缀模式(模型偶尔还是会犯)
+        prefixes_to_strip = [
+            "以下是译文:",
+            "以下是中文翻译:",
+            "译文:",
+            "中文翻译:",
+            "Translation:",
+            "翻译结果:",
+        ]
+        for p in prefixes_to_strip:
+            if cleaned.startswith(p):
+                cleaned = cleaned[len(p):].lstrip()
+                break
+
+        return TranslationResult(
+            text=cleaned, engine=self.name, chars=len(text), cached=False
+        )

backend/app/services/translation/service.py

@@ -1,9 +1,15 @@
 """翻译服务门面:配额检查 + 缓存 + 引擎选择 + 月度计数。
 
 引擎链路(优先级降序):
-1. tencent TMT(主,按月配额)
+1. tencent TMT(主,按月配额;快满时主动切走)
 2. tencent_maas(备用,OpenAI 兼容,无配额;主失败/TMT 配额耗尽时启用)
-3. local(最后兜底,需 settings.local_translate_enabled=true)
+3. agnes(第三级,通用 LLM 做翻译;MaaS 不可用时启用 — 质量次之但够用)
+4. local(最后兜底,需 settings.local_translate_enabled=true)
+
+注:
+- TMT 是按月计费的(腾讯云后台可能计费口径是请求字节,我们 redis 累加的是字符数,
+  差异约 2-3x);用户从腾讯云后台看"已用 2M"时,我们 redis 显示约 80 万字符
+- 用户决策:以腾讯云后台数字为准,快满时降级
 """
 from __future__ import annotations
 
@@ -15,6 +21,7 @@ from typing import Protocol
 
 from app.config import settings
 from app.redis_client import get_redis
+from app.services.translation.agnes import AgnesTranslator
 from app.services.translation.base import BaseTranslator, TranslationResult
 from app.services.translation.local import LocalTranslator
 from app.services.translation.tencent import TencentTranslator
@@ -38,6 +45,7 @@ class TranslationService:
     def __init__(self):
         self._tencent: BaseTranslator | None = None
         self._tencent_maas: BaseTranslator | None = None
+        self._agnes: BaseTranslator | None = None
         self._local: BaseTranslator | None = None
         # 串行:1 个并发;避免触发腾讯 TMT 限速
         self._sem = asyncio.Semaphore(1)
@@ -62,6 +70,18 @@ class TranslationService:
                 self._tencent_maas = None
         return self._tencent_maas
 
+    def _agnes(self) -> BaseTranslator | None:
+        """第三级:Agnes 通用 LLM 翻译(在 MaaS 不可用时启用)。
+        质量比 TMT/MaaS u2 差,但通用 LLM 也能翻,够用。
+        """
+        if self._agnes is None and settings.agnes_api_key:
+            try:
+                self._agnes = AgnesTranslator()
+            except Exception as e:
+                logger.warning("Agnes init failed: %s", e)
+                self._agnes = None
+        return self._agnes
+
     def _local_translator(self) -> BaseTranslator | None:
         """最后兜底:本地模型(需开关)。"""
         if self._local is None and settings.local_translate_enabled:
@@ -72,12 +92,13 @@ class TranslationService:
                 self._local = None
         return self._local
 
-    # 兼容旧调用点:返回第一个可用的 fallback(优先 maas,次 local)
+    # 兼容旧调用点:返回第一个可用的 fallback(优先 maas,次 agnes,再 local)
     def _fallback(self) -> BaseTranslator | None:
-        m = self._maas()
+        for getter in (self._maas, self._agnes, self._local_translator):
-        if m is not None:
+            f = getter()
-            return m
+            if f is not None:
-        return self._local_translator()
+                return f
+        return None
 
     async def can_use_tencent(self, chars: int) -> bool:
         if not settings.tencentcloud_secret_id:
@@ -130,8 +151,8 @@ class TranslationService:
             engine = None
 
         if engine is None:
-            # 配额耗尽 / TMT 不可用:走备用链
+            # 配额耗尽 / TMT 不可用:走备用链(maas → agnes → local)
-            engine = self._maas() or self._local_translator()
+            engine = self._fallback()
             if engine is None:
                 # 全无可用:返回原文 + 标记
                 return TranslationResult(
@@ -139,10 +160,10 @@ class TranslationService:
                     engine="skip",
                     chars=chars,
                 )
-            if engine.name == "tencent_maas":
+            logger.info(
-                logger.info("tencent quota exhausted, fallback to tencent_maas for %d chars", chars)
+                "tencent quota exhausted, fallback to %s for %d chars",
-            else:
+                engine.name, chars,
-                logger.info("fallback to local translator for %d chars", chars)
+            )
 
         # 3) 调用(失败时降级)
         async with self._sem:
@@ -165,15 +186,19 @@ class TranslationService:
             if res is None:
                 raise RuntimeError(f"translation failed for {chars} chars (engine={engine.name})")
 
+        # 注:engine 已经设好但运行时降级需要重新判断 fallback 链
+        # 上面 translate() 调用失败时,会重试 _fallback() 里下一个可用引擎
+        # 这里 engine 已经在 _fallback() 中按顺序选了一个最合适的,直接使用即可
+
         # 4) 写缓存 — 只缓存真实翻译结果;失败/降级文本不缓存(避免污染 30 天)
-        if res.engine in ("tencent", "tencent_maas", "nllb") and not res.cached:
+        if res.engine in ("tencent", "tencent_maas", "agnes", "nllb") and not res.cached:
             if "[翻译失败" not in res.text and "[本条未翻译" not in res.text:
                 try:
                     await r.set(ck, res.text, ex=60 * 60 * 24 * 30)  # 30 天
                 except Exception:
                     pass
 
-        # 5) 计数(只在 tencent TMT 上计;maas/local 都不计腾讯云配额)
+        # 5) 计数(只在 tencent TMT 上计;maas / agnes / local 都不计腾讯云配额)
         if res.engine == "tencent":
             try:
                 await self.add_usage(res.chars or chars)