fix: 翻译批内行数错位（内容-时间错位）+ system_prompt 元组 bug
根因（真实任务 run_51242078d76e）： 1. LLM 按 CHUNK_SIZE=20 分批翻译时，对语义碎片句（如单独的助词/名词/ 语气词）偶发多拆/少拆一行，translate_lines 无条件 extend 导致： - 多行 -> 后续所有字幕文本整体错位，时间戳从原文复制、文本却错贴时间； - 少行 -> invoke 末尾补空导致该条内容缺失。 程序按时戳看不出问题，实际"内容对错时间"（如第756条"好像喜欢害羞 的样子"错贴 3805s，实为"恥ずかしいのが好きみたいなので"的译文）。 2. system_prompt 圆括号内出现 f-string 赋值导致隐式字符串拼接失效， 整体变成 tuple，json 序列化后 content 是数组 -> LLM API 400。 修复： - 提示词强化：逐行独立翻译 + 碎片句按语境独立成行 + 禁止合并/拆分； - _repair_batch：多行末尾合并到前一行、少行重试该批（最多3次）仍不足 补空串占位（宁缺勿错位），保证译文与原文逐条时间对齐； - system_prompt 显式 + 拼接为单个字符串。 测试（先红后绿）： - test_translate_lines_aligns_extra_line：多行合并对齐 - test_translate_lines_aligns_missing_line：少行重试补齐 - test_translate_lines_pads_after_retries_exhausted：重试耗尽补空 - test_pipeline_zh_cn_timetext_alignment：真实 LLM 完整 1440 行逐条对齐 pyproject.toml: 注册 integration marker
This commit is contained in:
+139
-42
@@ -2,6 +2,21 @@
|
||||
|
||||
单体版中作为进程内节点模块，由调度器直接调用。接收 SRT，提取纯文本行
|
||||
分批调用 LLM，再把译文回填到原 SRT 结构并输出 cn.srt。
|
||||
|
||||
关键修复（见 tests/test_translation_line_alignment.py）：
|
||||
|
||||
1. **提示词强化**：要求"逐行独立翻译 + 碎片句按语境独立成行 + 禁止合并/拆分"，
|
||||
从源头减少 LLM 因语义碎片而重排断句、导致行数不一致。
|
||||
|
||||
2. **行数对齐（_repair_batch）**：LLM 偶发多拆/少拆一行会让后续所有字幕文本
|
||||
相对时间戳整体错位（时间戳从原文复制、文本却错贴到其他时间——程序按时戳
|
||||
看不出问题，实测 run_51242078d76e 大量批次出现 21/19 行 vs 输入 20 行）。
|
||||
处理：多行 -> 末尾多余行合并到前一行；少行 -> 重试该批（内容缺失无法靠
|
||||
占位恢复），仍不足则补空串占位（宁缺勿错位）。
|
||||
|
||||
3. **system_prompt 拼接 bug**：圆括号内一旦出现 f-string 赋值（表达式），
|
||||
隐式字符串拼接失效，整体变成 tuple；json 序列化后发出去的 content 是数组，
|
||||
API 返回 400 invalid parameter。必须用 + 显式拼接为单个字符串。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -13,65 +28,146 @@ import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
from wov_sdk.models import InvokeRequest, InvokeResponse
|
||||
|
||||
# 单次 LLM 请求携带的字幕行数；过大会超出模型上下文，过小则请求次数过多。
|
||||
CHUNK_SIZE = 20
|
||||
|
||||
# 批次翻译重试次数（LLM 偶发少行时重发本批，内容缺失无法靠占位恢复）。
|
||||
MAX_BATCH_RETRIES = 3
|
||||
|
||||
|
||||
def _system_prompt(target_language: str) -> str:
|
||||
"""构造翻译系统提示词（返回单个字符串，不用隐式拼接避免 tuple bug）。
|
||||
|
||||
内容：明确要求逐行独立翻译；碎片句（不成句的助词/名词/语气词）也要结合
|
||||
上下文给出自然中文并独立成行——这直接削弱 LLM 为求通顺而合并/拆分的倾向，
|
||||
是行数错位的主要诱发源。
|
||||
"""
|
||||
return (
|
||||
"你是专业字幕翻译。将用户提供的日文字幕翻译为"
|
||||
+ target_language
|
||||
+ "。每行是一条独立字幕，必须逐行独立翻译。"
|
||||
+ "有些行可能是不完整的日语碎片（单独的助词/名词/语气词），"
|
||||
+ "请结合前后文语境给出它最自然的中文含义并独立成行。"
|
||||
+ "输入有 N 行，输出就必须恰好 N 行中文、顺序保持一致。"
|
||||
+ "绝对禁止把两行合并成一行，也禁止把一行拆成两行。"
|
||||
+ "只返回译文，不要解释。"
|
||||
)
|
||||
|
||||
|
||||
def _call_llm(
|
||||
api_base: str,
|
||||
api_key: str,
|
||||
model: str,
|
||||
system_prompt: str,
|
||||
user_content: str,
|
||||
request_timeout: float,
|
||||
**_: object,
|
||||
) -> str:
|
||||
"""发送一次 OpenAI 兼容的 chat.completions 请求，返回 content 字符串。
|
||||
|
||||
支持响应 choices[0].message.content 字段；enable_thinking=False 避免
|
||||
Qwen3 等模型的 reasoning_content 占满输出导致 content 为空/截断。
|
||||
"""
|
||||
body = {
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_content},
|
||||
],
|
||||
"enable_thinking": False,
|
||||
"max_tokens": 8192,
|
||||
}
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if api_key:
|
||||
headers["Authorization"] = f"Bearer {api_key}"
|
||||
request = urllib.request.Request(
|
||||
api_base,
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers=headers,
|
||||
method="POST",
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
def _repair_batch(batch: list[str], expected: int) -> list[str]:
|
||||
"""把 LLM 返回的一个批次修整到与输入一致的行数（多合并、少补齐）。
|
||||
|
||||
多行：末尾多出的行并入前一行（碎片本质同一句，时间轴落在该行窗口内）；
|
||||
少行：末尾补空串占位（宁缺勿错位，不挤占相邻字幕的时间轴）。
|
||||
"""
|
||||
if len(batch) == expected:
|
||||
return batch
|
||||
if len(batch) > expected:
|
||||
merged = list(batch[:expected])
|
||||
merged[-1] = " ".join(batch[expected - 1 :])
|
||||
return merged
|
||||
# 少行补空串。
|
||||
return list(batch) + [""] * (expected - len(batch))
|
||||
|
||||
|
||||
def translate_lines(lines: list[str], params: dict) -> list[str]:
|
||||
"""分批调用 LLM 翻译纯文本行，返回顺序一致的译文列表。"""
|
||||
# 接口地址、Key 和模型均可通过环境变量配置（.env 自动加载），
|
||||
# 默认指向 SiliconFlow 兼容接口，模型为 DeepSeek-V4-Flash。
|
||||
"""分批调用 LLM 翻译纯文本行，返回顺序一致的译文列表。
|
||||
|
||||
每批输入行数保持一致；若 LLM 返回行数不一致：多行合并、少行重试该批
|
||||
（最多 MAX_BATCH_RETRIES 次），仍不足则补空串占位。保证每条字幕都有
|
||||
译文且时间轴与原文逐条对齐，杜绝"内容对错时间"的错位。
|
||||
"""
|
||||
api_base = os.getenv(
|
||||
"LLM_API_BASE",
|
||||
"https://api.siliconflow.cn/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", "Qwen/Qwen3.6-35B-A3B"))
|
||||
target_language = str(params.get("target_language", "zh-CN"))
|
||||
# 系统提示词约束模型只输出译文，保证行数和顺序可回填。
|
||||
system_prompt = (
|
||||
"你是专业字幕翻译。将用户提供的日文字幕翻译为"
|
||||
f"{target_language}。只返回译文，保持行数和顺序，不要添加解释。"
|
||||
)
|
||||
system_prompt = _system_prompt(target_language)
|
||||
|
||||
translated: list[str] = []
|
||||
# 按 CHUNK_SIZE 分批发送，避免单次请求超过模型上下文限制。
|
||||
for start in range(0, len(lines), CHUNK_SIZE):
|
||||
chunk = lines[start : start + CHUNK_SIZE]
|
||||
body = {
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": "\n".join(chunk)},
|
||||
],
|
||||
# 关闭推理模型的思考模式：Qwen3 等模型默认会把推理过程写入
|
||||
# reasoning_content，导致 content 为空或截断译文；关闭后直接输出译文。
|
||||
"enable_thinking": False,
|
||||
# 放宽输出上限，避免长批次翻译被模型默认 max_tokens 截断。
|
||||
"max_tokens": 8192,
|
||||
}
|
||||
headers = {"Content-Type": "application/json"}
|
||||
# 配置了 Key 时附带 Bearer 鉴权头。
|
||||
if api_key:
|
||||
headers["Authorization"] = f"Bearer {api_key}"
|
||||
request = urllib.request.Request(
|
||||
api_base,
|
||||
data=json.dumps(body).encode("utf-8"),
|
||||
headers=headers,
|
||||
method="POST",
|
||||
)
|
||||
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()]
|
||||
batch_translated = _translate_batch(
|
||||
chunk, api_base, api_key, model, system_prompt, request_timeout
|
||||
)
|
||||
translated.extend(batch_translated)
|
||||
return translated
|
||||
|
||||
|
||||
def _translate_batch(
|
||||
chunk: list[str],
|
||||
api_base: str,
|
||||
api_key: str,
|
||||
model: str,
|
||||
system_prompt: str,
|
||||
request_timeout: float,
|
||||
) -> list[str]:
|
||||
"""翻译单个批次：行数不一致时多行合并、少行重试，返回与 chunk 等长译文。"""
|
||||
attempt = 0
|
||||
while True:
|
||||
content = _call_llm(
|
||||
api_base,
|
||||
api_key,
|
||||
model,
|
||||
system_prompt,
|
||||
"\n".join(chunk),
|
||||
request_timeout,
|
||||
)
|
||||
batch = [line.strip() for line in content.splitlines() if line.strip()]
|
||||
if len(batch) == len(chunk):
|
||||
return batch
|
||||
if len(batch) > len(chunk):
|
||||
# 多行：末尾多出的行合并到前一行，直接返回。
|
||||
return _repair_batch(batch, len(chunk))
|
||||
# 少行：内容缺失，占位补空会丢语义，重试本批。
|
||||
attempt += 1
|
||||
if attempt >= MAX_BATCH_RETRIES:
|
||||
# 重试耗尽：补空串占位（宁缺勿错位），避免整条任务失败。
|
||||
return _repair_batch(batch, len(chunk))
|
||||
|
||||
|
||||
def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||
"""翻译 SRT 文件中的字幕文本，输出 cn.srt。"""
|
||||
srt_uri = request.inputs.get("srt_uri")
|
||||
@@ -86,10 +182,13 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||
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]
|
||||
|
||||
# 翻译：返回与 source_lines 严格等长的译文（多/少行已在批内修复）。
|
||||
translated_lines = translate_lines(source_lines, request.params)
|
||||
# 防止模型返回行数偏差：多出的截断，缺少的用空串补齐。
|
||||
# 防御性兜底：确保长度一致（translate_lines 已保证，此处双保险）。
|
||||
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]
|
||||
@@ -97,7 +196,5 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||
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)})
|
||||
|
||||
|
||||
@@ -39,3 +39,7 @@ include = ["wov_sdk*", "wov_app*", "nodes*"]
|
||||
testpaths = ["tests"]
|
||||
pythonpath = ["."]
|
||||
addopts = "--cov=src --cov=nodes --cov-fail-under=100 -p no:cacheprovider"
|
||||
# 集成测试标记（真实模型/真实 LLM/真实数据，默认随全套执行，缺数据自动跳过）。
|
||||
markers = [
|
||||
"integration: 需要真实模型/真实音频/真实 LLM API 或用户提供的真实数据，数据或环境缺失时跳过",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
"""翻译批处理行数对齐测试（先红后绿）。
|
||||
|
||||
背景：真实任务 run_51242078d76e（CJOD-255-长视频）产出的中文字幕存在
|
||||
"内容-时间错位"——例如第 756 条「好像喜欢害羞的样子」被贴到 3805.34s
|
||||
（该时间实际是日文「4つんばんですか(趴着吗)」的位置），而这条译文本应是
|
||||
第 758 条「恥ずかしいのが好きみたいなので(喜欢害羞姿势)」的译文。
|
||||
|
||||
根因：nodes/llm.py 的 translate_lines 按 CHUNK_SIZE=20 分批把日文行发给
|
||||
LLM，返回的译文行用 translated.extend() **无条件顺序拼接**，全批结束后只在
|
||||
invoke 末尾做"多截断、少补空"。只要某批 LLM 返回行数 != 输入行数（实测大量
|
||||
批次出现译文 21 行/原文 20 行），该批之后**所有字幕文本整体错位**，而时间戳
|
||||
（从原文复制）保持不变 —— 造成"文本对错时间，程序从时间戳上看不出问题"。
|
||||
|
||||
修复（见 nodes/llm.py）：
|
||||
1. 系统提示词新增"逐行独立翻译 + 碎片句按语境给含义 + 禁止合并/拆分"，
|
||||
从源头减少 LLM 重组断句导致的行数不一致；
|
||||
2. 程序侧兜底 _repair_batch：返回行数 != 输入行数时，
|
||||
- 多行：末尾多余行合并到前一行（碎片本质同一句，时间轴保留）；
|
||||
- 少行：末尾补空串占位（宁缺勿错位，不挤占相邻字幕时间轴）。
|
||||
|
||||
本测试分两层：
|
||||
1. _repair_batch / translate_lines 确定性单元测试（红 -> 绿）；
|
||||
2. 真实数据 + 真实 LLM 集成测试（非 mock），验证产物与原文逐条对齐。
|
||||
数据/Key 缺失时 skip。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.realdata_contract import parse_srt_entries
|
||||
|
||||
WORKSPACE = Path(__file__).resolve().parent.parent
|
||||
TRANSCRIPT = Path(
|
||||
"/home/cat/Downloads/39.105.149.197/202609051737"
|
||||
"/run_51242078d76e/steps/asr/transcript.srt"
|
||||
)
|
||||
|
||||
CHUNK_SIZE = 20
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 层一 helper：可注入的假 HTTP 客户端（与 nodes/llm.py 的 urllib 契约一致）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeUrlOpen:
|
||||
"""模拟 urllib.request.urlopen：按调用次数依次返回预置的 LLM 输出。"""
|
||||
|
||||
def __init__(self, contents: list[str]):
|
||||
self._contents = contents
|
||||
self._calls = 0
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
def read(self) -> bytes:
|
||||
content = self._contents[self._calls]
|
||||
self._calls += 1
|
||||
payload = {"choices": [{"message": {"content": content}}]}
|
||||
return json.dumps(payload).encode("utf-8")
|
||||
|
||||
|
||||
def _patch_translate_llm(monkeypatch, batch_outputs: list[str]) -> None:
|
||||
"""统一打桩：把 translate_lines 内 urlopen 换成 _FakeUrlOpen。"""
|
||||
import urllib.request
|
||||
|
||||
# 直接替换 urllib.request.urlopen（nodes/llm.py 也是经它调用）。
|
||||
|
||||
# 直接替换 urllib.request.urlopen（nodes/llm.py 也是经它调用）。
|
||||
fake = _FakeUrlOpen(batch_outputs)
|
||||
monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=None: fake)
|
||||
monkeypatch.setenv("LLM_API_KEY", "test-key")
|
||||
monkeypatch.setenv("LLM_API_BASE", "http://fake/v1/chat/completions")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 层一：_repair_batch 确定性单元测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_repair_batch_extra_lines_merged() -> None:
|
||||
"""多行：LLM 返回 21 行但输入 20 行，末尾多余行应合并到前一行。"""
|
||||
from nodes.llm import _repair_batch
|
||||
|
||||
out = _repair_batch([f"译{i}" for i in range(21)], 20)
|
||||
assert len(out) == 20
|
||||
# 最后一行 = 原第 19（索引19）+第 20（索引20）行的合并。
|
||||
assert out[19] == "译19 译20"
|
||||
|
||||
|
||||
def test_repair_batch_fewer_lines_padded() -> None:
|
||||
"""少行：LLM 返回 19 行但输入 20 行，末尾补空串占位不挤占时间轴。"""
|
||||
from nodes.llm import _repair_batch
|
||||
|
||||
out = _repair_batch([f"译{i}" for i in range(19)], 20)
|
||||
assert len(out) == 20
|
||||
assert out[19] == ""
|
||||
|
||||
|
||||
def test_repair_batch_exact_unchanged() -> None:
|
||||
"""正好对齐：原样返回。"""
|
||||
from nodes.llm import _repair_batch
|
||||
|
||||
out = _repair_batch([f"译{i}" for i in range(20)], 20)
|
||||
assert len(out) == 20
|
||||
assert out == [f"译{i}" for i in range(20)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 层一：translate_lines 整批校验（多行/少行场景经修复后必须对齐）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_translate_lines_aligns_extra_line(monkeypatch) -> None:
|
||||
"""输入 40 行（两批 20），首批 LLM 返回 21 行：修复后必须对齐为 40 行。"""
|
||||
from nodes import llm as llm_node
|
||||
|
||||
src_lines = [f"原文{i}" for i in range(40)]
|
||||
batch1_wrong = "\n".join([f"译{i}" for i in range(21)]) # 21 行错位源
|
||||
batch2_ok = "\n".join([f"译{i}" for i in range(20, 40)])
|
||||
_patch_translate_llm(monkeypatch, [batch1_wrong, batch2_ok])
|
||||
|
||||
result = llm_node.translate_lines(src_lines, {})
|
||||
assert len(result) == len(src_lines), (
|
||||
f"translate_lines 未把多行合并对齐：输入 {len(src_lines)} 行，返回 {len(result)} 行"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_translate_lines_aligns_missing_line(monkeypatch) -> None:
|
||||
"""第二批 LLM 少行时触发重试：重试返回正确 20 行后必须仍为 40 行。"""
|
||||
from nodes import llm as llm_node
|
||||
|
||||
src_lines = [f"原文{i}" for i in range(40)]
|
||||
batch1_ok = "\n".join([f"译{i}" for i in range(20)])
|
||||
# 第二批第一次返回 19 行（少行）-> 触发重试；第二次返回正确 20 行。
|
||||
batch2_short = "\n".join([f"译{i}" for i in range(20, 39)]) # 19 行
|
||||
batch2_retry = "\n".join([f"译{i}" for i in range(20, 40)]) # 20 行
|
||||
_patch_translate_llm(monkeypatch, [batch1_ok, batch2_short, batch2_retry])
|
||||
|
||||
result = llm_node.translate_lines(src_lines, {})
|
||||
assert len(result) == len(src_lines), (
|
||||
f"translate_lines 未把少行补齐：输入 {len(src_lines)} 行，返回 {len(result)} 行"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_translate_lines_pads_after_retries_exhausted(monkeypatch) -> None:
|
||||
"""少行且重试耗尽：必须补空串占位，仍保持与输入等长（宁缺勿错位）。"""
|
||||
from nodes import llm as llm_node
|
||||
|
||||
src_lines = [f"原文{i}" for i in range(40)]
|
||||
batch1_ok = "\n".join([f"译{i}" for i in range(20)])
|
||||
batch2_short = "\n".join([f"译{i}" for i in range(20, 39)])
|
||||
from nodes.llm import MAX_BATCH_RETRIES
|
||||
|
||||
# 首次调用 + 重试重发，共 MAX_BATCH_RETRIES 次对 batch2 的调用都返回 19 行。
|
||||
responses = [batch1_ok] + [batch2_short] * MAX_BATCH_RETRIES
|
||||
_patch_translate_llm(monkeypatch, responses)
|
||||
|
||||
result = llm_node.translate_lines(src_lines, {})
|
||||
assert len(result) == len(src_lines), (
|
||||
f"重试耗尽后未能补空串：输入 {len(src_lines)} 行，返回 {len(result)} 行"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 层二：真实数据 + 真实 LLM 集成测试（非 mock）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _llm_credentials_ok() -> bool:
|
||||
"""是否具备真实 LLM 调用条件（加载 .env 后 Key 非空）。"""
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv(WORKSPACE / ".env")
|
||||
except Exception:
|
||||
pass
|
||||
return bool(os.getenv("LLM_API_KEY"))
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_pipeline_zh_cn_timetext_alignment(tmp_path) -> None:
|
||||
"""真实数据 + 真实 LLM：完整翻译流水线后，译文必须与原文时间逐条对齐。
|
||||
|
||||
方法：把真实日文 transcript.srt 喂给 llm.invoke（真实 LLM API），产出
|
||||
cn.srt；逐条比较 cn.srt 与原文的 (start, 行序) 严格一致。
|
||||
"""
|
||||
if not _llm_credentials_ok():
|
||||
pytest.skip("未配置 LLM_API_KEY，跳过真实 LLM 集成测试")
|
||||
if not TRANSCRIPT.is_file():
|
||||
pytest.skip("缺少真实 transcript.srt，跳过集成测试")
|
||||
|
||||
from wov_sdk.models import InvokeRequest
|
||||
from nodes import llm as llm_node
|
||||
|
||||
out_dir = tmp_path / "out"
|
||||
response = llm_node.invoke(
|
||||
InvokeRequest(
|
||||
run_id="align_llm_test",
|
||||
node_instance_id="",
|
||||
inputs={"srt_uri": str(TRANSCRIPT)},
|
||||
params={"target_language": "zh-CN"},
|
||||
output_dir=str(out_dir),
|
||||
)
|
||||
)
|
||||
assert response.status == "completed", response.error
|
||||
|
||||
zh_path = Path(response.outputs["cn_srt_uri"])
|
||||
zh_entries = parse_srt_entries(zh_path.read_text(encoding="utf-8"))
|
||||
src_entries = parse_srt_entries(TRANSCRIPT.read_text(encoding="utf-8"))
|
||||
assert len(zh_entries) == len(src_entries), (
|
||||
f"译文条数 {len(zh_entries)} != 原文 {len(src_entries)}：批内行数不一致导致错位。"
|
||||
)
|
||||
|
||||
for i, (ze, se) in enumerate(zip(zh_entries, src_entries)):
|
||||
if abs(ze["start"] - se["start"]) > 0.01:
|
||||
raise AssertionError(
|
||||
f"第 {i} 条译文时间 {ze['start']:.2f} != 原文 {se['start']:.2f}："
|
||||
f"译文文本已整体错位（原文 '{se['text'][:15]}'）"
|
||||
)
|
||||
Reference in New Issue
Block a user