From 664957b5ef9a31252bc2e881b483961fe9b8bb0e Mon Sep 17 00:00:00 2001 From: cat-shark Date: Thu, 13 Aug 2026 22:09:55 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E4=B8=BA=E5=85=A8=E9=83=A8=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E8=A1=A5=E5=85=85=E4=B8=AD=E6=96=87=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E5=B9=B6=E5=8A=A0=E5=85=A5=20AGENTS=20=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E8=A7=84=E8=8C=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 7 +++++++ pyproject.toml | 5 +++++ tests/test_llm_node.py | 18 ++++++++++++++++++ wov_node_llm/__init__.py | 6 +++++- wov_node_llm/__main__.py | 21 +++++++++++++++++++++ 5 files changed, 56 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 62c73fa..65fa13d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,3 +17,10 @@ - `LLM_API_KEY`：可选。 - `LLM_MODEL`：默认模型，默认值 `default`。 - `LLM_TIMEOUT_SECONDS`：单次请求超时，默认 `600`。 + +## 代码注释规范 + +- 本仓库所有源码（Python、TOML 等支持注释的文件）必须配有详细中文注释，说明模块职责、LLM 分批调用与 SRT 回填逻辑，确保后续维护人员可以快速理解代码工作原理。 +- 新增或修改代码时，必须同步补充或更新对应注释；不得删除已有注释。 +- 测试代码同样必须配有中文注释，说明每条测试验证的行为。 +- JSON 数据文件（`node.manifest.json`）不支持注释，字段语义以 `wov-sdk` 的 `NodeManifest` 模型注释和本文档输入/输出说明为准。 diff --git a/pyproject.toml b/pyproject.toml index 49ff106..8c2641c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,4 @@ +# WOV LLM 节点配置：使用 uv 管理环境与依赖。 [project] name = "wov-node-llm" version = "0.1.0" @@ -5,16 +6,20 @@ description = "WOV LLM translation node" requires-python = ">=3.11" dependencies = ["wov-sdk"] +# 本地路径依赖 wov-sdk。 [tool.uv.sources] wov-sdk = { path = "../wov-sdk" } +# 开发依赖：pytest 与覆盖率工具。 [dependency-groups] dev = ["pytest", "pytest-cov"] +# pytest 配置：强制 100% 行覆盖率。 [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] addopts = "--cov=wov_node_llm --cov-report=term-missing --cov-fail-under=100" +# 仅打包节点包本身。 [tool.setuptools] packages = ["wov_node_llm"] diff --git a/tests/test_llm_node.py b/tests/test_llm_node.py index 63b62d4..27901ec 100644 --- a/tests/test_llm_node.py +++ b/tests/test_llm_node.py @@ -1,3 +1,8 @@ +"""LLM 翻译节点测试。 + +覆盖真实 HTTP 服务调用、超时配置、SRT 回填、异常与入口点启动等路径。 +""" + import json import runpy import threading @@ -11,7 +16,10 @@ from wov_sdk.models import InvokeRequest class FakeUrlOpenResponse: + """模拟 urllib 响应对象，提供固定 LLM 译文内容。""" + def __init__(self, content: str) -> None: + # 预编码为 Chat Completions 风格的 JSON 响应体。 self._payload = json.dumps( {"choices": [{"message": {"content": content}}]} ).encode("utf-8") @@ -27,6 +35,7 @@ class FakeUrlOpenResponse: def _make_srt(tmp_path, count=5) -> Path: + """生成标准 SRT 测试文件，文本行为“原文字幕N”。""" lines = [] for index in range(count): lines.extend( @@ -43,6 +52,7 @@ def _make_srt(tmp_path, count=5) -> Path: def test_translate_lines_via_fake_api(monkeypatch) -> None: + """验证通过真实 HTTP 服务器调用 LLM 接口并保持行顺序。""" class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: length = int(self.headers.get("Content-Length", "0")) @@ -88,6 +98,7 @@ def test_translate_lines_via_fake_api(monkeypatch) -> None: def test_translate_lines_api_error(monkeypatch) -> None: + """验证 LLM 接口不可用时抛出 URLError。""" def fail_open(request, timeout): raise urllib.error.URLError("api down") @@ -100,6 +111,7 @@ def test_translate_lines_api_error(monkeypatch) -> None: def test_translate_lines_default_timeout(monkeypatch) -> None: + """验证未配置超时时使用默认 600 秒。""" captured = {} def fake_open(request, timeout): @@ -116,6 +128,7 @@ def test_translate_lines_default_timeout(monkeypatch) -> None: def test_translate_lines_env_timeout(monkeypatch) -> None: + """验证 LLM_TIMEOUT_SECONDS 环境变量可覆盖超时。""" captured = {} def fake_open(request, timeout): @@ -132,6 +145,7 @@ def test_translate_lines_env_timeout(monkeypatch) -> None: def test_invoke_success(tmp_path, monkeypatch) -> None: + """验证成功调用会把译文回填到 SRT 并输出 cn.srt。""" source = _make_srt(tmp_path) def fake_translate(lines, params): @@ -153,6 +167,7 @@ def test_invoke_success(tmp_path, monkeypatch) -> None: def test_invoke_pads_short_translation(tmp_path, monkeypatch) -> None: + """验证译文行数不足时用空行补齐，保持 SRT 结构完整。""" source = _make_srt(tmp_path, count=3) monkeypatch.setattr( "wov_node_llm.__main__.translate_lines", @@ -170,6 +185,7 @@ def test_invoke_pads_short_translation(tmp_path, monkeypatch) -> None: def test_invoke_missing_input(tmp_path) -> None: + """验证缺少 srt_uri 时返回失败。""" response = invoke( InvokeRequest( run_id="run_3", @@ -182,6 +198,7 @@ def test_invoke_missing_input(tmp_path) -> None: def test_invoke_missing_file(tmp_path) -> None: + """验证 SRT 文件不存在时返回失败。""" response = invoke( InvokeRequest( run_id="run_4", @@ -194,6 +211,7 @@ def test_invoke_missing_file(tmp_path) -> None: def test_entrypoint(monkeypatch) -> None: + """验证 python -m wov_node_llm 会加载 llm-translate manifest。""" module_path = Path(__file__).resolve().parent.parent / "wov_node_llm" / "__main__.py" captured = {} diff --git a/wov_node_llm/__init__.py b/wov_node_llm/__init__.py index 3c608c8..3552ba1 100644 --- a/wov_node_llm/__init__.py +++ b/wov_node_llm/__init__.py @@ -1 +1,5 @@ -"""WOV LLM translation node.""" +"""WOV LLM 字幕翻译节点。 + +调用 OpenAI 兼容接口，把 ASR 产出的日文 SRT 字幕逐批翻译为目标语言， +同时保持 SRT 的序号与时间轴结构不变。 +""" diff --git a/wov_node_llm/__main__.py b/wov_node_llm/__main__.py index cccd6d8..97dc77d 100644 --- a/wov_node_llm/__main__.py +++ b/wov_node_llm/__main__.py @@ -1,3 +1,9 @@ +"""LLM 翻译节点入口。 + +通过标准节点 HTTP 服务接收 SRT，提取纯文本行分批调用 LLM，再把译文回填到 +原 SRT 结构并输出 cn.srt。 +""" + from __future__ import annotations import json @@ -9,23 +15,29 @@ from pathlib import Path from wov_sdk.models import InvokeRequest, InvokeResponse, NodeManifest from wov_sdk.server import run_node +# 单次 LLM 请求携带的字幕行数；过大会超出模型上下文，过小则请求次数过多。 CHUNK_SIZE = 20 def translate_lines(lines: list[str], params: dict) -> list[str]: + """分批调用 LLM 翻译纯文本行，返回顺序一致的译文列表。""" + # 接口地址、Key 和模型均可通过环境变量配置，默认指向内网兼容接口。 api_base = os.getenv( "LLM_API_BASE", "http://192.168.123.70:8080/v1/chat/completions", ) api_key = os.getenv("LLM_API_KEY", "") + # 单次请求超时可配置，长文本翻译场景下需要放宽。 request_timeout = float(os.getenv("LLM_TIMEOUT_SECONDS", "600")) model = str(params.get("model") or os.getenv("LLM_MODEL", "default")) target_language = str(params.get("target_language", "zh-CN")) + # 系统提示词约束模型只输出译文，保证行数和顺序可回填。 system_prompt = ( "你是专业字幕翻译。将用户提供的日文字幕翻译为" f"{target_language}。只返回译文，保持行数和顺序，不要添加解释。" ) translated: list[str] = [] + # 按 CHUNK_SIZE 分批发送，避免单次请求超过模型上下文限制。 for start in range(0, len(lines), CHUNK_SIZE): chunk = lines[start : start + CHUNK_SIZE] body = { @@ -36,6 +48,7 @@ def translate_lines(lines: list[str], params: dict) -> list[str]: ], } headers = {"Content-Type": "application/json"} + # 配置了 Key 时附带 Bearer 鉴权头。 if api_key: headers["Authorization"] = f"Bearer {api_key}" request = urllib.request.Request( @@ -46,7 +59,9 @@ def translate_lines(lines: list[str], params: dict) -> list[str]: ) with urllib.request.urlopen(request, timeout=request_timeout) as response: payload = json.loads(response.read().decode("utf-8")) + # 兼容 OpenAI Chat Completions 响应格式，取第一条消息内容。 content = payload["choices"][0]["message"]["content"] + # 忽略空行，保证译文列表与输入行一一对应。 translated.extend( [line.strip() for line in content.splitlines() if line.strip()] ) @@ -54,6 +69,7 @@ def translate_lines(lines: list[str], params: dict) -> list[str]: def invoke(request: InvokeRequest) -> InvokeResponse: + """翻译 SRT 文件中的字幕文本，输出 cn.srt。""" srt_uri = request.inputs.get("srt_uri") if not srt_uri: return InvokeResponse(status="failed", error="srt_uri is required") @@ -62,23 +78,28 @@ def invoke(request: InvokeRequest) -> InvokeResponse: if not srt_path.is_file(): return InvokeResponse(status="failed", error="srt file not found") + # 标准 SRT 每 4 行一组：序号、时间轴、文本、空行；文本位于第 3 行。 lines = srt_path.read_text(encoding="utf-8").splitlines() text_indices = list(range(2, len(lines), 4)) source_lines = [lines[index] for index in text_indices] translated_lines = translate_lines(source_lines, request.params) + # 防止模型返回行数偏差：多出的截断，缺少的用空串补齐。 translated_lines = translated_lines[: len(source_lines)] translated_lines += [""] * max(0, len(source_lines) - len(translated_lines)) + # 只替换文本行，序号、时间轴和空行保持不变。 for index, text_index in enumerate(text_indices): lines[text_index] = translated_lines[index] output_dir = Path(request.output_dir) output_dir.mkdir(parents=True, exist_ok=True) output_path = output_dir / "cn.srt" + # 末尾补一个换行，让文件满足常见文本工具习惯。 output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") return InvokeResponse(status="completed", outputs={"cn_srt_uri": str(output_path)}) def main() -> None: + """加载节点清单并以本模块的 invoke 处理器启动服务。""" manifest_path = Path(__file__).resolve().parent.parent / "node.manifest.json" with open(manifest_path, "r", encoding="utf-8") as f: manifest = NodeManifest.from_dict(json.load(f))