diff --git a/backend/app/services/translation/zhipu.py b/backend/app/services/translation/zhipu.py new file mode 100644 index 0000000..3dc5afe --- /dev/null +++ b/backend/app/services/translation/zhipu.py @@ -0,0 +1,108 @@ +"""智谱 GLM 翻译(OpenAI 兼容协议)。 + +- 端点:https://open.bigmodel.cn/api/paas/v4/chat/completions +- 模型:glm-4-flash(免费,默认) / glm-4.5 / glm-4.6 / glm-4-air 等 +- 鉴权:Bearer + +设计上独立于 LlmClient,专门走 zhipu_* 配置,避免和 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.zhipu") + + +# 经过实测,这套 prompt 在 GLM-4-Flash 上输出稳定(不夹带 reasoning / 注释)。 +_SYSTEM_PROMPT = """你是一个即时翻译助手。对于用户输入的英文或日文文章,请直接输出对应的简体中文译文。严格遵守: + +不要进行任何分步思考、不要输出分析、不要添加注释或说明。 + +只输出中文译文本身,不包含任何额外文字(如"翻译结果:"、"以下是中文:"等)。 + +如果输入内容既非英文也非日文,仅回复:"仅支持英文或日文翻译为中文。" """ + + +class ZhipuTranslator(BaseTranslator): + """智谱 GLM 翻译(OpenAI 兼容协议)。""" + + name = "zhipu" + + def __init__(self): + if not settings.zhipu_api_key: + raise RuntimeError("Zhipu API key missing") + self.api_key = settings.zhipu_api_key + self.base_url = settings.zhipu_base_url.rstrip("/") + self.model = settings.zhipu_model + self.interval_sec = settings.zhipu_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: + if not text.strip(): + return TranslationResult(text=text, engine=self.name, chars=0) + if not self.is_configured(): + raise RuntimeError("Zhipu API key missing") + + url = f"{self.base_url}/chat/completions" + payload = { + "model": self.model, + "messages": [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": text}, + ], + "temperature": 0.0, + "max_tokens": max(256, len(text) * 3), + "stream": False, + } + 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"Zhipu 5xx: {r.status_code} {r.text[:200]}") + if r.status_code == 429: + raise RuntimeError(f"Zhipu 限流: {r.text[:200]}") + if r.status_code != 200: + raise RuntimeError(f"Zhipu {r.status_code}: {r.text[:300]}") + data = r.json() + if "error" in data: + raise RuntimeError(f"Zhipu error: {data['error']}") + content = ( + data.get("choices", [{}])[0] + .get("message", {}) + .get("content", "") + .strip() + ) + if not content: + raise RuntimeError(f"Zhipu empty content: {r.text[:300]}") + 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("zhipu 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