From fbcb5115d8875dcc05f7744fa68b5d9d42ecf7ab Mon Sep 17 00:00:00 2001 From: catShark <1716967236@qq.com> Date: Sun, 6 Sep 2026 13:57:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20llm=20=E7=BF=BB=E8=AF=91=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E6=89=93=E5=8D=B0=E5=88=86=E6=89=B9=E8=BF=9B=E5=BA=A6?= =?UTF-8?q?=E4=B8=8E=20token=20=E5=A4=84=E7=90=86=E9=80=9F=E5=BA=A6?= =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 翻译节点 translate_lines 分批调用 LLM 时补充可观测日志： - 任务开始打印总行数与总批数；每完成一批打印总进度（已完成/总行数、 第几批/共几批、批耗时、累计耗时与行/s），结束打印汇总（总耗时、 累计 tokens、tok/s 与行/s），便于评估 LLM 处理速度； - _call_llm 解析响应 usage（total_tokens），每次请求打印单批耗时与 token/s（接口不返回 usage 时按 0 处理）； - _translate_batch 改为返回(译文, 本批 tokens)，供累计汇总； - 补充空输入直接返回的单测，保持 100% 行覆盖率。 --- nodes/llm.py | 94 ++++++++++++++++++++++++++++++++++++++------- tests/test_nodes.py | 9 +++++ 2 files changed, 89 insertions(+), 14 deletions(-) diff --git a/nodes/llm.py b/nodes/llm.py index 9bf109c..6c8097c 100755 --- a/nodes/llm.py +++ b/nodes/llm.py @@ -23,10 +23,13 @@ from __future__ import annotations import json import os +import time import urllib.error import urllib.request from pathlib import Path +from wov_app.logging import get_logger + from wov_sdk.models import InvokeRequest, InvokeResponse from nodes.subtitle_cleanup import clean_srt_text from nodes.proper_nouns import build_proper_noun_rule @@ -36,6 +39,9 @@ CHUNK_SIZE = 20 # 批次翻译重试次数（LLM 偶发少行时重发本批，内容缺失无法靠占位恢复）。 MAX_BATCH_RETRIES = 3 +# 节点运行日志：翻译分批进度与处理速度输出到主进程控制台。 +logger = get_logger("llm-translate") + def _system_prompt(target_language: str) -> str: """构造翻译系统提示词（返回单个字符串，不用隐式拼接避免 tuple bug）。 @@ -63,13 +69,20 @@ def _call_llm( system_prompt: str, user_content: str, request_timeout: float, + log_prefix: str = "", **_: object, -) -> str: - """发送一次 OpenAI 兼容的 chat.completions 请求，返回 content 字符串。 +) -> tuple[str, dict | None]: + """发送一次 OpenAI 兼容的 chat.completions 请求，返回 (content, usage)。 支持响应 choices[0].message.content 字段；enable_thinking=False 避免 Qwen3 等模型的 reasoning_content 占满输出导致 content 为空/截断。 + + usage 为响应体里的 usage 对象（含 prompt_tokens/completion_tokens/ + total_tokens），部分兼容接口不返回 usage 时为 None——调用方用其估算 + token 处理速度。log_prefix 为日志行前缀（如"第 2/5 批"），用于打印 + 单批耗时与 token 速度。 """ + started = time.monotonic() body = { "model": model, "messages": [ @@ -91,7 +104,16 @@ def _call_llm( with urllib.request.urlopen(request, timeout=request_timeout) as response: payload = json.loads(response.read().decode("utf-8")) content = payload["choices"][0]["message"]["content"] - return content + usage = payload.get("usage") + # 单批耗时与 token 速度日志：直观反映 LLM 处理速度（wall clock）。 + elapsed = time.monotonic() - started + tokens = int(usage.get("total_tokens", 0)) if isinstance(usage, dict) else 0 + rate = tokens / elapsed if elapsed > 0 and tokens > 0 else 0.0 + logger.info( + "LLM 响应 %s 耗时 %.1fs, tokens=%d (%.1f tok/s)", + log_prefix, elapsed, tokens, rate, + ) + return content, usage def _repair_batch(batch: list[str], expected: int) -> list[str]: @@ -116,6 +138,10 @@ def translate_lines(lines: list[str], params: dict) -> list[str]: 每批输入行数保持一致；若 LLM 返回行数不一致：多行合并、少行重试该批 （最多 MAX_BATCH_RETRIES 次），仍不足则补空串占位。保证每条字幕都有 译文且时间轴与原文逐条对齐，杜绝"内容对错时间"的错位。 + + 日志：每完成一批打印总进度（已完成行数/总行数、第几批/共几批、累计 + 耗时与行处理速度），结束打印汇总（总耗时、累计 tokens 与 tok/s）， + 便于评估 LLM 处理速度。 """ api_base = os.getenv( "LLM_API_BASE", @@ -127,13 +153,45 @@ def translate_lines(lines: list[str], params: dict) -> list[str]: target_language = str(params.get("target_language", "zh-CN")) system_prompt = _system_prompt(target_language) + total_lines = len(lines) + total_batches = (total_lines + CHUNK_SIZE - 1) // CHUNK_SIZE if total_lines else 0 + if total_lines == 0: + return [] + # 任务开始日志：总行数与总批数（批次 = CHUNK_SIZE 行，最后一批可能不足）。 + logger.info("翻译开始: %d 行, 分 %d 批", total_lines, total_batches) + translated: list[str] = [] - for start in range(0, len(lines), CHUNK_SIZE): + total_tokens = 0 + all_started = time.monotonic() + for batch_index in range(1, total_batches + 1): + start = (batch_index - 1) * CHUNK_SIZE chunk = lines[start : start + CHUNK_SIZE] - batch_translated = _translate_batch( - chunk, api_base, api_key, model, system_prompt, request_timeout + # 每批日志前缀（第几批/共几批），供单次 LLM 请求日志与批进度复用。 + log_prefix = f"第 {batch_index}/{total_batches} 批" + batch_started = time.monotonic() + batch_translated, batch_tokens = _translate_batch( + chunk, api_base, api_key, model, system_prompt, request_timeout, log_prefix ) translated.extend(batch_translated) + total_tokens += batch_tokens + # 批进度日志：已完成行数/总行数、当前批耗时、累计耗时与行处理速度。 + done = len(translated) + elapsed_total = time.monotonic() - all_started + logger.info( + "翻译进度 %d/%d 行 (%s完成, 批耗时 %.1fs, 累计 %.1fs, %.1f 行/s)", + done, total_lines, log_prefix, + time.monotonic() - batch_started, elapsed_total, + done / elapsed_total if elapsed_total > 0 else 0.0, + ) + # 任务汇总日志：总耗时、累计 tokens 与 token/行处理速度。 + wall = time.monotonic() - all_started + tok_rate = total_tokens / wall if wall > 0 and total_tokens > 0 else 0.0 + logger.info( + "翻译完成: %d/%d 行, %d 批, 总耗时 %.1fs, 累计 tokens=%d (%.1f tok/s, %.1f 行/s)", + len(translated), total_lines, total_batches, wall, + total_tokens, tok_rate, + len(translated) / wall if wall > 0 else 0.0, + ) return translated @@ -144,39 +202,47 @@ def _translate_batch( model: str, system_prompt: str, request_timeout: float, -) -> list[str]: - """翻译单个批次：行数不一致时多行合并、少行重试，返回与 chunk 等长译文。 + log_prefix: str = "", +) -> tuple[list[str], int]: + """翻译单个批次，返回 (与 chunk 等长译文, 本批 total_tokens)。 - 每批调用前根据本批原文命中情况动态拼接专名/隐语规则（build_proper_noun_rule）， - 注入到系统提示词，让 LLM 正确处理片假名专名与成人语境隐语。""" + 行数不一致时多行合并、少行重试；每批调用前根据本批原文命中情况动态 + 拼接专名/隐语规则（build_proper_noun_rule），注入到系统提示词，让 LLM + 正确处理片假名专名与成人语境隐语。""" # 本批命中的专名/隐语规则（无命中返回 None）。 rule = build_proper_noun_rule(chunk) batch_system = system_prompt if rule: batch_system = system_prompt + "\n\n" + rule attempt = 0 + batch_tokens = 0 while True: - content = _call_llm( + # content 为译文文本；usage 含本批 prompt/completion tokens（接口不 + # 返回时为 None），用于累计任务 token 总量与速度评估。 + content, usage = _call_llm( api_base, api_key, model, batch_system, "\n".join(chunk), request_timeout, + log_prefix, ) + if isinstance(usage, dict): + batch_tokens = int(usage.get("total_tokens", 0) or 0) # 保留所有行：先 rstrip 尾随换行避免多出末尾空行，再 splitlines 保留 # 内容中的空串行（空行可能是合法的空字幕，过滤掉会误判行数）。 batch = content.rstrip("\n").splitlines() if len(batch) == len(chunk): - return batch + return batch, batch_tokens if len(batch) > len(chunk): # 多行：末尾多出的行合并到前一行，直接返回。 - return _repair_batch(batch, len(chunk)) + return _repair_batch(batch, len(chunk)), batch_tokens # 少行：内容缺失，占位补空会丢语义，重试本批。 attempt += 1 if attempt >= MAX_BATCH_RETRIES: # 重试耗尽：补空串占位（宁缺勿错位），避免整条任务失败。 - return _repair_batch(batch, len(chunk)) + return _repair_batch(batch, len(chunk)), batch_tokens def invoke(request: InvokeRequest) -> InvokeResponse: diff --git a/tests/test_nodes.py b/tests/test_nodes.py index 3bf0ad9..23c1b46 100644 --- a/tests/test_nodes.py +++ b/tests/test_nodes.py @@ -520,6 +520,15 @@ def test_llm_translate_lines_via_fake_api(monkeypatch) -> None: thread.join(timeout=5) +def test_llm_translate_lines_empty_input() -> None: + """验证空行列表直接返回空译文，不发起任何 LLM 请求。 + + 翻译开始前对空输入提前返回：没有字幕行时无需计算批数、也无需调用 + 接口（避免无谓请求与除零等边界问题），同时保持返回类型为列表。""" + result = translate_lines([], {}) + assert result == [] + + def test_llm_translate_lines_api_error(monkeypatch) -> None: """验证 LLM 接口不可用时抛出 URLError。""" def fail_open(request, timeout):