KingFlow 前往官网 →

用 Claude 中转批量翻译与处理文档：省钱又省心

KingFlow · 国内直连 AI API 中转

KingFlow

去年接了个活儿，帮一家做跨境电商的朋友把几千篇商品描述、说明书、FAQ 从中文批量翻成英文和日文。文件多、术语杂、还得保留原来的格式，人工翻根本翻不完。我第一反应是上 Claude——它翻译的语感和对上下文的把控确实比一般机翻强一大截。但真跑起来才发现，批量场景和平时聊天完全是两码事。这篇就把我踩过的坑和后来跑顺的方案讲清楚，重点是怎么又省钱又稳。

一、批量翻译和文档处理，难在哪

平时开个对话框翻一段话，谁都会。可一旦上了量，几个问题立刻冒出来：

第一是量真的大。 几千篇文档，每篇几百到几千字，一趟跑下来是几百万甚至上千万 token。这时候模型选贵了、缓存没用上，成本能差出好几倍。

第二是官方直连贵而且不好接。 直接走官方，一是按量计费单价不便宜，批量一跑账单很吓人；二是国内接入本身就麻烦，美区信用卡的 BIN 经常被拒，账号风控一严就可能冻结，跑批跑到一半被封那才叫欲哭无泪。

第三是并发要稳。 批量任务本质是"同一段脚本循环调用几千次"，一旦并发一上来，延迟飙高、动不动 429 限流、连接三五分钟被掐断，整个任务就卡死在那儿。批处理最怕的不是慢，是跑到 80% 突然全挂，前功尽弃。

所以批量这事儿，光有个好模型不够，接入通道、成本结构、并发容错三样得一起解决。

二、中转方案：改一行 base_url，脚本照跑

我最后走的是中转的路子，用 KingFlow 做通道。它的好处是国内节点直连，不用自己挂代理，首字延迟（TTFT）我这边实测通常一两秒就出来了，跑批的时候不会因为网络抖动动不动超时。而且它走的是官方 /v1/messages 协议，不是那种逆向反代，Anthropic 那边更新了也不容易突然挂掉——这对要跑好几天的批量任务很关键。

接入基本零改动。OpenAI 兼容的写法，把 base_url 指到 https://www.kingflow.ai/v1，Key 换成中转的就行：

from openai import OpenAI

client = OpenAI(
    api_key="你的_KingFlow_Key",
    base_url="https://www.kingflow.ai/v1",
)

def translate(text, target_lang="英文"):
    resp = client.chat.completions.create(
        model="claude-haiku-4-5",
        messages=[
            {"role": "system", "content": f"你是专业翻译，把用户文本翻成{target_lang}，保留原始 Markdown 格式，只输出译文。"},
            {"role": "user", "content": text},
        ],
    )
    return resp.choices[0].message.content

批量的话就是套个循环，把文档一篇篇喂进去：

import os, glob

os.makedirs("out", exist_ok=True)
for path in glob.glob("docs/*.md"):
    with open(path, encoding="utf-8") as f:
        content = f.read()
    result = translate(content, "英文")
    name = os.path.basename(path)
    with open(f"out/{name}", "w", encoding="utf-8") as f:
        f.write(result)
    print(f"完成 {name}")

如果你原来就是 Anthropic 原生 SDK 写的，也不用重写，把 ANTHROPIC_BASE_URL 指到 https://www.kingflow.ai、ANTHROPIC_AUTH_TOKEN 换成中转 Key 即可，一个 Key 多模型，改 model 参数就能切换，不用维护好几套凭证。

三、省钱：高频用 haiku，难段落再上 sonnet/opus

批量场景省钱的核心思路就一句话——别用大炮打蚊子。

绝大多数翻译内容其实没那么难，商品描述、常规说明这类，claude-haiku-4-5 完全够用，它就是为高频、低成本设计的，跑几千篇下来成本压得很低。真正需要拔高质量的是那些拗口的长句、专业术语密集的段落、或者要保留微妙语气的营销文案，这些再单独挑出来用 claude-sonnet-4-6 或者 claude-opus-4-8 重跑一遍。我一般是先用 haiku 全量跑一遍，再按字数、术语命中之类的规则筛出一小撮"疑难段落"升档处理，这样大头走便宜模型，只有少数走贵的，整体成本能省下一大截。

另一个省钱大杀器是 Prompt Cache。批量翻译时，那段系统指令（"你是专业翻译，保留格式，只输出译文……"）每一次调用都一模一样，是纯纯的重复输入。KingFlow 对 Prompt Cache 是完整透传的，给固定指令加上 cache_control，第二次之后这部分就走缓存命中，不用重复计费。翻译这种"输入远大于输出"的活儿，命中缓存后能省下相当可观的一块成本——按我实测大概能砍掉输入侧一大半开销，具体幅度看你固定指令占比多大。

resp = client.messages.create(  # Anthropic 原生写法演示 cache_control
    model="claude-haiku-4-5",
    max_tokens=4096,
    system=[{
        "type": "text",
        "text": "你是专业翻译，把文本翻成英文，保留 Markdown 格式，只输出译文。",
        "cache_control": {"type": "ephemeral"},
    }],
    messages=[{"role": "user", "content": text}],
)

四、并发与重试：控速、重试、别单点

批量任务最忌讳"一把梭"，几千个请求瞬间全甩出去，再稳的通道也顶不住，限流、超时哗哗来。我的做法是控制并发数，配上失败重试和指数退避，让任务能自己扛过偶发抖动：

import time
from concurrent.futures import ThreadPoolExecutor

def safe_translate(text, retries=4):
    for i in range(retries):
        try:
            return translate(text)
        except Exception as e:
            wait = 2 ** i
            print(f"第 {i+1} 次失败，{wait}s 后重试：{e}")
            time.sleep(wait)
    return None  # 记录失败，最后单独补跑

# 并发数别拉太满，一般开个几路到十几路，看实测调
with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(safe_translate, texts))

几个经验：并发数不用一开始就拉满，从小往上试，观察成功率和延迟找到甜点区；一定要把失败的条目记下来，跑完主流程再统一补跑，别让个别失败拖垮整批；关键任务别单点押注，中转通道选那种容错兜底做得好的——某个模型临时不可用能自动切换的，跑长任务时心里踏实。KingFlow 这块因为是国内节点直连、高峰期限速情况我遇到得少，并发成功率整体还行。

五、对账：后台看用量，花在哪清清楚楚

批量任务跑完，最怕的是账对不上——扣了多少、每个模型花了多少、缓存到底命中没有，心里没数。这也是我不太敢用那种野中转的原因，倍率不透明，扣费和预期对不上还没处查。

KingFlow 后台能查调用日志、余额、token 用量和每一笔调用明细。我一般跑完批量就去后台拉一下用量：看看 haiku 和 sonnet 各占多少、缓存读取的 token 有没有起来（缓存命中的话这个数会很明显）、总消耗和预估差多少。这样下次跑批就能反过来优化——比如发现某类文档 haiku 质量够，就把它从升档名单里剔掉，进一步省钱。人民币小额充值、先充点测一测再决定跑不跑全量，也不用一上来就压一大笔进去。

六、FAQ

Q1：批量翻译到底该选哪个模型？ 大头用 claude-haiku-4-5，它就是给高频低成本场景准备的，常规文档质量足够。挑出来的难段落、营销文案、术语密集的再用 claude-sonnet-4-6 或 claude-opus-4-8 重跑。先全量 haiku、再筛少数升档，是我觉得性价比最高的组合。

Q2：Prompt Cache 怎么确认真的命中了？ 给固定的系统指令加 cache_control，连着发两次，看返回的 usage 里 cache_read_input_tokens 这个字段第二次是不是非零。非零就说明命中了，KingFlow 对缓存是完整透传的，不用担心中间被吞掉。

Q3：跑批中途报 429 或超时怎么办？ 先把并发数降下来，别一次性甩太多请求；加上指数退避重试，偶发失败能自己扛过去；失败的条目记下来最后补跑。国内直连的通道抖动会少一些，但重试逻辑该写还得写，这是批处理的基本功。

Q4：从官方或原生 SDK 迁过来要改多少代码？ 基本就改两处：base_url 指到 https://www.kingflow.ai/v1（OpenAI 兼容），或者把 ANTHROPIC_BASE_URL 指到 https://www.kingflow.ai（Claude 原生），再把 Key 换成中转的。业务逻辑、调用方式一行都不用动，一个 Key 还能多模型切换，迁移成本非常低。

批量翻译和文档处理，说白了就是拿工程化的方式去用大模型：模型分档控成本、缓存复用省重复、并发重试保稳定、后台对账做优化。通道选对了，剩下的就是把脚本跑顺。这套跑下来，几千篇文档我基本能放着让它自己跑，省钱也省心。