feat: llm 翻译节点打印分批进度与 token 处理速度日志

翻译节点 translate_lines 分批调用 LLM 时补充可观测日志:
- 任务开始打印总行数与总批数;每完成一批打印总进度(已完成/总行数、
  第几批/共几批、批耗时、累计耗时与行/s),结束打印汇总(总耗时、
  累计 tokens、tok/s 与行/s),便于评估 LLM 处理速度;
- _call_llm 解析响应 usage(total_tokens),每次请求打印单批耗时与
  token/s(接口不返回 usage 时按 0 处理);
- _translate_batch 改为返回(译文, 本批 tokens),供累计汇总;
- 补充空输入直接返回的单测,保持 100% 行覆盖率。
This commit is contained in:
2026-09-06 13:57:44 +08:00
parent 6a265a1bc8
commit fbcb5115d8
2 changed files with 89 additions and 14 deletions
+80 -14
View File
@@ -23,10 +23,13 @@ from __future__ import annotations
import json import json
import os import os
import time
import urllib.error import urllib.error
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
from wov_app.logging import get_logger
from wov_sdk.models import InvokeRequest, InvokeResponse from wov_sdk.models import InvokeRequest, InvokeResponse
from nodes.subtitle_cleanup import clean_srt_text from nodes.subtitle_cleanup import clean_srt_text
from nodes.proper_nouns import build_proper_noun_rule from nodes.proper_nouns import build_proper_noun_rule
@@ -36,6 +39,9 @@ CHUNK_SIZE = 20
# 批次翻译重试次数(LLM 偶发少行时重发本批,内容缺失无法靠占位恢复)。 # 批次翻译重试次数(LLM 偶发少行时重发本批,内容缺失无法靠占位恢复)。
MAX_BATCH_RETRIES = 3 MAX_BATCH_RETRIES = 3
# 节点运行日志:翻译分批进度与处理速度输出到主进程控制台。
logger = get_logger("llm-translate")
def _system_prompt(target_language: str) -> str: def _system_prompt(target_language: str) -> str:
"""构造翻译系统提示词(返回单个字符串,不用隐式拼接避免 tuple bug)。 """构造翻译系统提示词(返回单个字符串,不用隐式拼接避免 tuple bug)。
@@ -63,13 +69,20 @@ def _call_llm(
system_prompt: str, system_prompt: str,
user_content: str, user_content: str,
request_timeout: float, request_timeout: float,
log_prefix: str = "",
**_: object, **_: object,
) -> str: ) -> tuple[str, dict | None]:
"""发送一次 OpenAI 兼容的 chat.completions 请求,返回 content 字符串 """发送一次 OpenAI 兼容的 chat.completions 请求,返回 (content, usage)
支持响应 choices[0].message.content 字段;enable_thinking=False 避免 支持响应 choices[0].message.content 字段;enable_thinking=False 避免
Qwen3 等模型的 reasoning_content 占满输出导致 content 为空/截断。 Qwen3 等模型的 reasoning_content 占满输出导致 content 为空/截断。
usage 为响应体里的 usage 对象(含 prompt_tokens/completion_tokens/
total_tokens),部分兼容接口不返回 usage 时为 None——调用方用其估算
token 处理速度。log_prefix 为日志行前缀(如"第 2/5 批"),用于打印
单批耗时与 token 速度。
""" """
started = time.monotonic()
body = { body = {
"model": model, "model": model,
"messages": [ "messages": [
@@ -91,7 +104,16 @@ def _call_llm(
with urllib.request.urlopen(request, timeout=request_timeout) as response: with urllib.request.urlopen(request, timeout=request_timeout) as response:
payload = json.loads(response.read().decode("utf-8")) payload = json.loads(response.read().decode("utf-8"))
content = payload["choices"][0]["message"]["content"] 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]: 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 返回行数不一致:多行合并、少行重试该批 每批输入行数保持一致;若 LLM 返回行数不一致:多行合并、少行重试该批
(最多 MAX_BATCH_RETRIES 次),仍不足则补空串占位。保证每条字幕都有 (最多 MAX_BATCH_RETRIES 次),仍不足则补空串占位。保证每条字幕都有
译文且时间轴与原文逐条对齐,杜绝"内容对错时间"的错位。 译文且时间轴与原文逐条对齐,杜绝"内容对错时间"的错位。
日志:每完成一批打印总进度(已完成行数/总行数、第几批/共几批、累计
耗时与行处理速度),结束打印汇总(总耗时、累计 tokens 与 tok/s),
便于评估 LLM 处理速度。
""" """
api_base = os.getenv( api_base = os.getenv(
"LLM_API_BASE", "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")) target_language = str(params.get("target_language", "zh-CN"))
system_prompt = _system_prompt(target_language) 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] = [] 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] chunk = lines[start : start + CHUNK_SIZE]
batch_translated = _translate_batch( # 每批日志前缀(第几批/共几批),供单次 LLM 请求日志与批进度复用。
chunk, api_base, api_key, model, system_prompt, request_timeout 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) 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 return translated
@@ -144,39 +202,47 @@ def _translate_batch(
model: str, model: str,
system_prompt: str, system_prompt: str,
request_timeout: float, request_timeout: float,
) -> list[str]: log_prefix: str = "",
"""翻译单个批次:行数不一致时多行合并、少行重试,返回与 chunk 等长译文。 ) -> tuple[list[str], int]:
"""翻译单个批次,返回 (与 chunk 等长译文, 本批 total_tokens)。
每批调用前根据本批原文命中情况动态拼接专名/隐语规则(build_proper_noun_rule), 行数不一致时多行合并、少行重试;每批调用前根据本批原文命中情况动态
注入到系统提示词,让 LLM 正确处理片假名专名与成人语境隐语。""" 拼接专名/隐语规则(build_proper_noun_rule),注入到系统提示词,让 LLM
正确处理片假名专名与成人语境隐语。"""
# 本批命中的专名/隐语规则(无命中返回 None)。 # 本批命中的专名/隐语规则(无命中返回 None)。
rule = build_proper_noun_rule(chunk) rule = build_proper_noun_rule(chunk)
batch_system = system_prompt batch_system = system_prompt
if rule: if rule:
batch_system = system_prompt + "\n\n" + rule batch_system = system_prompt + "\n\n" + rule
attempt = 0 attempt = 0
batch_tokens = 0
while True: while True:
content = _call_llm( # content 为译文文本;usage 含本批 prompt/completion tokens(接口不
# 返回时为 None),用于累计任务 token 总量与速度评估。
content, usage = _call_llm(
api_base, api_base,
api_key, api_key,
model, model,
batch_system, batch_system,
"\n".join(chunk), "\n".join(chunk),
request_timeout, request_timeout,
log_prefix,
) )
if isinstance(usage, dict):
batch_tokens = int(usage.get("total_tokens", 0) or 0)
# 保留所有行:先 rstrip 尾随换行避免多出末尾空行,再 splitlines 保留 # 保留所有行:先 rstrip 尾随换行避免多出末尾空行,再 splitlines 保留
# 内容中的空串行(空行可能是合法的空字幕,过滤掉会误判行数)。 # 内容中的空串行(空行可能是合法的空字幕,过滤掉会误判行数)。
batch = content.rstrip("\n").splitlines() batch = content.rstrip("\n").splitlines()
if len(batch) == len(chunk): if len(batch) == len(chunk):
return batch return batch, batch_tokens
if len(batch) > len(chunk): if len(batch) > len(chunk):
# 多行:末尾多出的行合并到前一行,直接返回。 # 多行:末尾多出的行合并到前一行,直接返回。
return _repair_batch(batch, len(chunk)) return _repair_batch(batch, len(chunk)), batch_tokens
# 少行:内容缺失,占位补空会丢语义,重试本批。 # 少行:内容缺失,占位补空会丢语义,重试本批。
attempt += 1 attempt += 1
if attempt >= MAX_BATCH_RETRIES: if attempt >= MAX_BATCH_RETRIES:
# 重试耗尽:补空串占位(宁缺勿错位),避免整条任务失败。 # 重试耗尽:补空串占位(宁缺勿错位),避免整条任务失败。
return _repair_batch(batch, len(chunk)) return _repair_batch(batch, len(chunk)), batch_tokens
def invoke(request: InvokeRequest) -> InvokeResponse: def invoke(request: InvokeRequest) -> InvokeResponse:
+9
View File
@@ -520,6 +520,15 @@ def test_llm_translate_lines_via_fake_api(monkeypatch) -> None:
thread.join(timeout=5) 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: def test_llm_translate_lines_api_error(monkeypatch) -> None:
"""验证 LLM 接口不可用时抛出 URLError。""" """验证 LLM 接口不可用时抛出 URLError。"""
def fail_open(request, timeout): def fail_open(request, timeout):