"""
将 locomo 风格 JSON（如 locomo10.json）中的英文内容批量翻译为简体中文。
使用 OpenAI 兼容的 /v1/chat/completions 接口，通过 httpx 调用（不使用官方 SDK）。
每个可翻译字段单独请求一次 API。
用法示例：
export OPENAI_API_KEY=sk-...
python3 translate_locomo_zh.py locomo10.json -o locomo10_zh.json \
--base-url https://api.openai.com/v1 --model gpt-4o-mini
火山引擎Coding Plan：
python3 translate_locomo_zh.py locomo10.json -o locomo10_zh.json \
--base-url https://ark.cn-beijing.volces.com/api/coding/v3 \
--model doubao-seed-2.0-lite
阿里云：
python3 translate_locomo_zh.py locomo10.json -o locomo10_zh.json \
--base-url https://dashscope.aliyuncs.com/compatible-mode/v1 \
--model qwen-plus
minimax：
python3 translate_locomo_zh.py locomo10.json -o locomo10_zh.json \
--base-url https://api.minimaxi.com/v1 \
--model minimax-m2.7
也可使用 --api-url 直接指定完整的 chat completions 地址。
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
import time
from typing import Any
import httpx
try:
from tqdm import tqdm
except ImportError:
tqdm = None # type: ignore[misc, assignment]
SESSION_LIST_KEY = re.compile(r"^session_\d+$")
EVENTS_SESSION_KEY = re.compile(r"^events_session_\d+$")
OBS_SESSION_KEY = re.compile(r"^session_\d+_observation$")
SUMMARY_KEY = re.compile(r"^session_\d+_summary$")
SYSTEM_PROMPT = """You are a professional translator. Translate the following English text into Simplified Chinese.
Strict rules:
- Keep all person names, place names, and time/date expressions exactly as in the source (do not translate them or rewrite them into Chinese-style forms).
- Keep dialogue/evidence reference tags matching the pattern like D1:3, D12:45 (letter D, digits, colon, digits) unchanged wherever they appear.
- Output only the translated Chinese text, with no quotes or explanation."""
def build_chat_url(base_url: str | None, api_url: str | None) -> str:
if api_url:
return api_url.rstrip("/")
if not base_url:
base_url = "https://api.openai.com/v1"
return base_url.rstrip("/") + "/chat/completions"
def translate_one(
client: httpx.Client,
url: str,
api_key: str,
model: str,
text: str,
timeout: float,
max_retries: int,
) -> str:
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
payload: dict[str, Any] = {
"model": model,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": text},
],
"temperature": 0.2,
}
last_err: Exception | None = None
for attempt in range(max_retries + 1):
try:
r = client.post(url, json=payload, headers=headers, timeout=timeout)
r.raise_for_status()
data = r.json()
choice = data["choices"][0]["message"]["content"]
if isinstance(choice, str):
return choice.strip()
return str(choice).strip()
except (httpx.HTTPStatusError, httpx.RequestError, KeyError, json.JSONDecodeError) as e:
last_err = e
if isinstance(e, httpx.HTTPStatusError) and e.response.status_code in (429, 502, 503, 504):
wait = min(2**attempt, 60)
time.sleep(wait)
continue
if attempt < max_retries:
time.sleep(min(2**attempt, 10))
continue
raise
assert last_err is not None
raise last_err
def should_skip_text(s: str) -> bool:
if not isinstance(s, str):
return True
if not s.strip():
return True
return False
class Translator:
def __init__(
self,
client: httpx.Client,
url: str,
api_key: str,
model: str,
timeout: float,
max_retries: int,
delay_s: float,
progress: Any,
) -> None:
self._client = client
self._url = url
self._api_key = api_key
self._model = model
self._timeout = timeout
self._max_retries = max_retries
self._delay_s = delay_s
self._progress = progress
def __call__(self, text: str) -> str:
if should_skip_text(text):
return text
out = translate_one(
self._client,
self._url,
self._api_key,
self._model,
text,
self._timeout,
self._max_retries,
)
if self._progress is not None:
self._progress.update(1)
if self._delay_s > 0:
time.sleep(self._delay_s)
return out
def process_qa(qa: list[dict[str, Any]], tr: Translator) -> None:
for item in qa:
if "question" in item:
item["question"] = tr(str(item["question"]))
if "answer" in item and item["answer"] is not None:
a = item["answer"]
item["answer"] = tr(str(a))
def process_conversation(conv: dict[str, Any], tr: Translator) -> None:
for key, val in conv.items():
if not SESSION_LIST_KEY.match(key):
continue
if not isinstance(val, list):
continue
for msg in val:
if not isinstance(msg, dict):
continue
if "text" in msg and msg["text"] is not None:
msg["text"] = tr(str(msg["text"]))
if "blip_caption" in msg and msg["blip_caption"] is not None:
msg["blip_caption"] = tr(str(msg["blip_caption"]))
if "query" in msg and msg["query"] is not None:
msg["query"] = tr(str(msg["query"]))
def process_event_summary(event_summary: dict[str, Any], tr: Translator) -> None:
for ek, ev in event_summary.items():
if not EVENTS_SESSION_KEY.match(ek):
continue
if not isinstance(ev, dict):
continue
for subk, subv in ev.items():
if subk == "date":
continue
if not isinstance(subv, list):
continue
for i, s in enumerate(subv):
if isinstance(s, str) and not should_skip_text(s):
subv[i] = tr(s)
def process_observation(observation: dict[str, Any], tr: Translator) -> None:
for ok, ov in observation.items():
if not OBS_SESSION_KEY.match(ok):
continue
if not isinstance(ov, dict):
continue
for _person, obs_list in ov.items():
if not isinstance(obs_list, list):
continue
for inner in obs_list:
if not isinstance(inner, list) or len(inner) < 2:
continue
first = inner[0]
# 第二项通常为 D数字:数字 引用，不翻译；只翻译首段自然语言
if isinstance(first, str) and not should_skip_text(first):
inner[0] = tr(first)
def process_session_summary(session_summary: dict[str, Any], tr: Translator) -> None:
for sk, sv in session_summary.items():
if not SUMMARY_KEY.match(sk):
continue
if isinstance(sv, str) and not should_skip_text(sv):
session_summary[sk] = tr(sv)
def count_translatable_strings(obj: Any) -> int:
"""用于进度条：预估将要调用翻译的次数。"""
n = 0
if isinstance(obj, list):
for item in obj:
n += count_translatable_strings(item)
return n
if not isinstance(obj, dict):
return n
if "qa" in obj and isinstance(obj["qa"], list):
for q in obj["qa"]:
if isinstance(q, dict):
if "question" in q and q["question"] is not None and str(q["question"]).strip():
n += 1
if "answer" in q and q["answer"] is not None and str(q["answer"]).strip():
n += 1
conv = obj.get("conversation")
if isinstance(conv, dict):
for key, val in conv.items():
if not SESSION_LIST_KEY.match(key) or not isinstance(val, list):
continue
for msg in val:
if not isinstance(msg, dict):
continue
for field in ("text", "blip_caption", "query"):
if field in msg and msg[field] is not None and str(msg[field]).strip():
n += 1
es = obj.get("event_summary")
if isinstance(es, dict):
for ek, ev in es.items():
if not EVENTS_SESSION_KEY.match(ek) or not isinstance(ev, dict):
continue
for subk, subv in ev.items():
if subk == "date" or not isinstance(subv, list):
continue
for s in subv:
if isinstance(s, str) and s.strip():
n += 1
obs = obj.get("observation")
if isinstance(obs, dict):
for ok, ov in obs.items():
if not OBS_SESSION_KEY.match(ok) or not isinstance(ov, dict):
continue
for _p, obs_list in ov.items():
if not isinstance(obs_list, list):
continue
for inner in obs_list:
if isinstance(inner, list) and inner and isinstance(inner[0], str) and inner[0].strip():
n += 1
ss = obj.get("session_summary")
if isinstance(ss, dict):
for sk, sv in ss.items():
if SUMMARY_KEY.match(sk) and isinstance(sv, str) and sv.strip():
n += 1
return n
def process_sample(sample: dict[str, Any], tr: Translator) -> None:
if "qa" in sample and isinstance(sample["qa"], list):
process_qa(sample["qa"], tr)
if "conversation" in sample and isinstance(sample["conversation"], dict):
process_conversation(sample["conversation"], tr)
if "event_summary" in sample and isinstance(sample["event_summary"], dict):
process_event_summary(sample["event_summary"], tr)
if "observation" in sample and isinstance(sample["observation"], dict):
process_observation(sample["observation"], tr)
if "session_summary" in sample and isinstance(sample["session_summary"], dict):
process_session_summary(sample["session_summary"], tr)
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="将 locomo JSON 翻译为简体中文（OpenAI 兼容 API + httpx）")
p.add_argument("input", type=str, help="输入 JSON 路径，例如 locomo10.json")
p.add_argument(
"-o",
"--output",
type=str,
default=None,
help="输出路径（默认：在输入文件名后加 _zh，如 locomo10_zh.json）",
)
p.add_argument(
"--base-url",
type=str,
default=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
help="API 根地址，默认 https://api.openai.com/v1；最终请求 {base-url}/chat/completions",
)
p.add_argument(
"--api-url",
type=str,
default=os.environ.get("OPENAI_COMPAT_CHAT_URL"),
help="若设置则直接使用该完整 chat completions URL，忽略 --base-url",
)
p.add_argument(
"--api-key",
type=str,
default=os.environ.get("OPENAI_API_KEY", ""),
help="Bearer Token（也可环境变量 OPENAI_API_KEY）",
)
p.add_argument(
"--model",
type=str,
default=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
help="模型名称",
)
p.add_argument("--timeout", type=float, default=120.0, help="单次请求超时（秒）")
p.add_argument("--max-retries", type=int, default=3, help="失败重试次数")
p.add_argument(
"--delay",
type=float,
default=0.0,
metavar="SEC",
help="每次翻译请求之后休眠秒数（限流用，默认 0）",
)
p.add_argument("--no-progress", action="store_true", help="不显示 tqdm 进度条")
return p.parse_args()
def main() -> None:
args = parse_args()
if not args.api_key:
print("错误：请通过 --api-key 或环境变量 OPENAI_API_KEY 提供密钥。", file=sys.stderr)
sys.exit(1)
in_path = os.path.abspath(args.input)
if args.output:
out_path = os.path.abspath(args.output)
else:
root, ext = os.path.splitext(in_path)
out_path = f"{root}_zh{ext or '.json'}"
with open(in_path, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, list):
print("错误：顶层 JSON 应为数组（与 locomo10.json 一致）。", file=sys.stderr)
sys.exit(1)
total_calls = sum(count_translatable_strings(s) for s in data)
print(f"预计翻译 API 调用次数: {total_calls}", file=sys.stderr)
url = build_chat_url(args.base_url, args.api_url)
progress = None
if tqdm and not args.no_progress and total_calls > 0:
progress = tqdm(total=total_calls, unit="call", desc="Translating")
with httpx.Client() as client:
tr = Translator(
client=client,
url=url,
api_key=args.api_key,
model=args.model,
timeout=args.timeout,
max_retries=args.max_retries,
delay_s=args.delay,
progress=progress,
)
for sample in data:
if isinstance(sample, dict):
process_sample(sample, tr)
if progress is not None:
progress.close()
with open(out_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
print(f"已写入: {out_path}", file=sys.stderr)
if __name__ == "__main__":
main()
背景
LoCoMo 评测数据（如
locomo10.json）为英文对话、摘要、问答等。我提供脚本translate_locomo_zh.py，通过 OpenAI 兼容的/v1/chat/completions接口（httpx调用，无需官方 SDK），将数据中的可翻译字段批量译为简体中文，并写出新的 JSON 文件。注意：脚本会在原内存中的结构上就地改写各字段，最后整文件写入
-o指定路径；请始终对副本运行或指定新输出路径，避免覆盖原始数据。1. 依赖
httpx（必需）tqdm（可选，用于进度条；未安装时仍可运行，加--no-progress或自动无进度条）2. 认证与环境变量
OPENAI_API_KEY：Bearer Token（也可用--api-key）OPENAI_BASE_URL：可选，作为--base-url的默认值OPENAI_COMPAT_CHAT_URL：可选，若设置则作为--api-url的默认值（完整 chat completions URL，优先级高于--base-url）OPENAI_MODEL：可选，作为--model的默认值3. 基本用法
export OPENAI_API_KEY=sk-... python3 translate_locomo_zh.py path/to/locomo10.json -o path/to/locomo10_zh.json \ --base-url https://api.openai.com/v1 \ --model gpt-4o-mini未指定
-o时，默认在输入文件名后加_zh（例如locomo10.json→locomo10_zh.json）。输入格式要求：顶层 JSON 必须是数组（与官方
locomo10.json结构一致），否则脚本会报错退出。4. 各云厂商 / 兼容网关示例（与脚本 docstring 一致）
火山引擎 Coding Plan：
阿里云（兼容模式）：
MiniMax：
若网关的 chat 地址不是
{base-url}/chat/completions，可使用--api-url指向完整 URL。5. 常用参数说明
input-o/--output--base-url{base-url}/chat/completions--api-url--base-url）--api-keyOPENAI_API_KEY--model--timeout--max-retries--delay--no-progress运行开始会在 stderr 打印 预计 API 调用次数（每个可翻译字符串单独一次请求），便于估算费用与时间。
6. 会翻译哪些字段？
对每个样本大致包括：
qa：question、answer（非空字符串）conversation：匹配session_N的列表中，每条消息的text、blip_caption、queryevent_summary：events_session_N下除date外的列表中的英文字符串observation：session_N_observation下，每条观测为[自然语言, 引用]时，只翻译第一项；第二项常为D数字:数字引用，保持不译session_summary：session_N_summary字符串系统提示词约束：人名、地名、时间/日期保持原文；类似
D1:3、D12:45的对话/证据引用标签保持不变；模型只输出译文，无解释。7. 实践建议
locomo10.json调用量很大：先用小文件跑通再全量；必要时加大--delay或换限额更高的密钥。temperature=0.2）下译文会有差异，评测对比时请固定模型与版本。qa与conversation中的引用标签、日期字段是否与英文版对齐。