用 LangGraph 构建四步翻译法：从状态图到生产级翻译流水线
本文以"O小译"智能翻译系统为蓝本，深入剖析如何用 LangGraph 状态图编排一个术语识别 → 直译 → 自审 → 意译的四步翻译流水线，包含条件边回退、Chunk 并发、多层兜底策略等生产级实践。
一、为什么选 LangGraph？
传统 LLM 应用大多是"一次 Prompt 出结果"的单轮模式。但高质量翻译是一个多阶段、可回退、需要全局状态协调的复杂流程——这恰好是 LangGraph 的核心设计目标：
| 需求 | LangGraph 能力 |
|---|---|
| 多步骤串行 | StateGraph + add_edge |
| 条件分支/回退 | add_conditional_edges |
| 全局状态共享 | TypedDict State |
| 并发处理 Chunk | 节点内 asyncio.gather |
| 异步执行 | graph.ainvoke() |
二、状态定义：翻译流程的"数据总线"
LangGraph 的核心理念是所有节点共享同一个 State。我们定义了两层状态结构：
class ChunkState(TypedDict):chunk_id: strsource_text: strterminology: Dict # Step1 识别的术语映射literal_translation: str # Step2 直译结果self_review: Dict # Step3 自审报告final_translation: str # Step4 意译定稿status: str # pending | running | translated | reviewed | polished | failedretry_count: intclass TranslateState(TypedDict):session_id: strsource_lang: strtarget_lang: strdomain: Optional[str] # legal / medical / tech / literaturemodel: str # 用户指定模型source_text: strchunks: List[ChunkState]global_terminology: Dict # 跨 Chunk 统一术语表final_result: stroverall_status: strsse_queue: List[Dict]error_message: Optional[str]token_usage: Dictstart_time: float
设计要点：
- Chunk 粒度状态机：每个 Chunk 有独立的
status字段，允许不同 Chunk 处于不同阶段（如 Chunk A 已在审校，Chunk B 还在直译）。 - 全局术语表 + 局部术语表：
global_terminology由全文提取得到，每个 Chunk 的terminology是全局+局部的合并（局部优先），保证跨段术语一致性。 - SSE 事件队列：
sse_queue用于前端实时推送，每个节点在处理过程中主动推送进度事件。
三、图构建：六节点 + 条件回退
[Start] → [chunk_split] → [step1_terminology] → [step2_literal] → [step3_review]↑ ↓| [需要修订?]| ↓ ↓└──[step2_literal] [step4_polish]↓[aggregate] → [END]
核心构建代码：
def build_translation_graph():builder = StateGraph(TranslateState)builder.add_node("chunk_split", split_into_chunks)builder.add_node("step1_terminology", step1_terminology)builder.add_node("step2_literal", step2_literal)builder.add_node("step3_review", step3_review)builder.add_node("step4_polish", step4_polish)builder.add_node("aggregate", aggregate_node)builder.set_entry_point("chunk_split")builder.add_edge("chunk_split", "step1_terminology")builder.add_edge("step1_terminology", "step2_literal")builder.add_edge("step2_literal", "step3_review")# 条件边：自审不通过则回退到直译builder.add_conditional_edges("step3_review",needs_revision,{"step2_literal": "step2_literal", "step4_polish": "step4_polish"})builder.add_edge("step4_polish", "aggregate")builder.add_edge("aggregate", END)return builder.compile()
条件边的核心逻辑
def needs_revision(state: TranslateState) -> str:for chunk in state.get("chunks", []):self_review = chunk.get("self_review", {})if self_review.get("needs_revision", False):if chunk.get("retry_count", 0) < 2: # 最多回退 2 次chunk["retry_count"] = chunk.get("retry_count", 0) + 1chunk["status"] = "pending"return "step2_literal" # 回退到直译return "step4_polish" # 审校通过，进入意译
关键决策：
- 回退粒度：只要任一 Chunk 需要修订，就整体回退到直译。这看似粗暴，但保证了修订时所有 Chunk 都能拿到最新术语表。
- 重试上限 = 2：防止无限循环。LLM 的输出有随机性，2 次重试已足够让模型"自我修正"。
- 回退时重置状态：
chunk["status"] = "pending"确保直译节点能重新处理该 Chunk。
四、节点深度解析
4.1 文本切分（chunk_split）
这不是简单的 split("\n\n")，而是一个三级切分策略：
原文 → 段落边界切分 → 超长段落按句子切分 → 相邻 Chunk 重叠窗口（50 tokens）
Token 估算公式（针对中英文混合场景）：
def estimate_tokens(text: str) -> int:chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')english_chars = sum(1 for c in text if c.isascii() and c.isalnum())return int(chinese_chars * 1.5 + english_chars * 0.25)
为什么中文字符 × 1.5？ 因为主流 Tokenizer（如 tiktoken）中，一个中文字符通常消耗 1~2 个 token，取 1.5 是一个偏保守的均值估计，避免 Chunk 超过模型上下文窗口。
4.2 术语识别（step1_terminology）
采用全局 + 局部双层提取：
- 全局提取：取前 2000 字符做全文术语扫描，生成
global_terminology - 局部补充：对每个 Chunk 并发提取局部术语，与全局合并
术语验证逻辑（防止 LLM 幻觉）：
# 验证1：原文和译文不能相同
if source_term.strip().lower() == translation.strip().lower():continue# 验证2：译文不应包含原文（避免 "prompt（提示词）" 这种格式）
if source_term.lower() in translation.lower() and len(source_term) > 3:continue# 验证3：括号注释格式过滤
if re.search(re.escape(source_term) + r'\s*[（(]', translation):continue
这三层验证在生产环境中极其重要——LLM 在术语提取时经常偷懒，直接输出 "token（token）" 或原封不动的术语。
4.3 直译（step2_literal）
直译节点的核心是严格遵循术语表：
- 每个 Chunk 的 Prompt 中注入了该 Chunk 的术语表（JSON 格式）
- 修订模式下，还会注入上一轮的审校意见，让模型针对性修正
兜底策略：直译失败时，使用原文作为 literal_translation，避免后续流程中断。这是生产系统的核心原则——降级优于崩溃。
4.4 自审（step3_review）
让 LLM 扮演"审校者"，从 5 个维度打分：
| 维度 | 说明 |
|---|---|
| 准确性 | 信息是否遗漏或错误 |
| 术语一致性 | 术语翻译是否统一 |
| 语法流畅度 | 目标语言语法是否正确 |
| 格式保留 | 原文格式是否完整保留 |
| 文化适应性 | 是否存在文化冲突 |
修订触发条件：overall_score < 6.0 或存在 critical 级别问题。
审校结果是一个结构化 JSON，包含 issues 列表，每个 issue 有 severity、category、description、suggestion 四个字段——这些会被原封不动地传给直译节点作为修订依据。
4.5 意译定稿（step4_polish）
意译是整个流程中 temperature 最高 的步骤（0.5 vs 直译的 0.3 vs 术语的 0.2），因为润色需要更多"创造性"。
输入同时包含：直译结果 + 审校意见 + 术语表。模型需要在准确性约束下最大化自然度。
4.6 聚合（aggregate）
聚合不只是 "\n\n".join()，还有一步术语一致性二次校验：
def apply_terminology_consistency(text, terminology):for source_term, target_term in terminology.items():if source_term in text and target_term not in text:pattern = re.compile(re.escape(source_term))text = pattern.sub(target_term, text)return text
如果某个 Chunk 的 LLM 翻译"漏掉"了术语表中的翻译，聚合阶段会用正则替换做一次强制修正——这是最后一道防线。
五、并发模型：Chunk 级别的 asyncio.gather
每个节点内部都采用了Chunk 并发模式：
tasks = []
for chunk in pending_chunks:tasks.append(process_chunk(state, chunk))
await asyncio.gather(*tasks)
这意味着：
- 术语识别阶段：所有 Chunk 的术语提取是并发的
- 直译阶段：所有 Chunk 的翻译是并发的
- 审校阶段：所有 Chunk 的审校是并发的
对于一篇被切成 5 个 Chunk 的文章，理论并发度是 5x（相比串行处理，端到端延迟降低约 80%）。
六、多层兜底策略总结
| 层级 | 失败场景 | 兜底方案 |
|---|---|---|
| 术语提取 | LLM 调用失败 | 使用空术语表，直译靠模型自身能力 |
| 直译 | LLM 返回空/异常 | 使用原文作为"直译结果" |
| 自审 | JSON 解析失败 | 给默认低分(3.0)，触发修订 |
| 意译 | LLM 调用失败 | 使用直译结果作为最终译文 |
| 聚合 | Chunk 状态异常 | 使用原文兜底 |
| 修订循环 | 超过 2 次重试 | 强制进入意译，不再回退 |
设计哲学：每一步都有兜底，保证用户至少能看到一个结果（即使不完美），而不是一个 500 错误。
七、单例图 + 全局复用
translation_graph = Nonedef get_translation_graph():global translation_graphif translation_graph is None:translation_graph = build_translation_graph()return translation_graph
StateGraph.compile() 是一个有开销的操作（构建执行计划、验证图结构），因此我们在进程级别做单例缓存。所有翻译任务共享同一个编译后的图实例，通过传入不同的 state 来区分。
八、分治视角：状态图背后的架构哲学
如果只从"流程编排"角度理解 LangGraph，就错过了它更深层的设计意义。这张状态图实际上是分治原则在翻译领域的工程化表达。
流程分治：将矛盾目标分解为独立步骤
一个 Prompt 同时要求"准确"和"自然"，等于让模型在两个矛盾目标之间做妥协。四步法的分治策略是：
术语识别 ──→ 只追求"领域准确性"，温度 0.2（最确定性）
直译 ──→ 只追求"信息忠实度"，温度 0.3
自审 ──→ 只追求"批判性检查"，温度 0.2
意译 ──→ 只追求"表达自然度"，温度 0.5（最创造性）
每一步的 Prompt 只需要优化一个维度，模型不再需要在多个矛盾目标间做权衡。这就是分治的核心收益——降低单次 LLM 调用的认知复杂度。
数据分治 × 流程分治的交叉
当数据分治（Chunk 切分）和流程分治（四步法）交叉时，就形成了一个二维并行矩阵：
Chunk 1 Chunk 2 Chunk 3 Chunk 4
术语 [done] [done] [done] [done] ← 并发
直译 [done] [running] [running] [pending] ← 并发
自审 [pending] [pending] [pending] [pending] ← 等直译全完成
意译 [pending] [pending] [pending] [pending] ← 并发
这种Chunk 级并发 + 步骤级串行的模式，是分治在性能维度的体现——数据维度的子问题互相独立可以并发，流程维度的子问题有依赖必须串行。
条件边：分治中的反馈环
传统分治是"分→治→合"的单向流程，但翻译系统增加了一个反馈环（条件边回退）。这是对经典分治的工程化扩展：
经典分治：Divide → Conquer → Merge
翻译分治：Divide → Conquer → Verify → (不满意?) → Re-Conquer → Merge
Verify 步骤（自审）的存在，让系统具备了自我纠错能力——这在传统分治算法中是不存在的，但在 LLM 应用中是必要的，因为 LLM 的输出具有随机性。
九、总结
LangGraph 在这个翻译系统中扮演的角色远不止"流程编排"：
- 状态管理：TypedDict 提供了类型安全的全局状态总线
- 流程控制：条件边实现了"审校-回退"闭环
- 并发原语：节点内 asyncio.gather 实现了 Chunk 级并发
- 可观测性：每个节点的 State 变更天然构成了执行轨迹
四步翻译法的核心洞察是：翻译不是一个 Prompt 能解决的问题，而是一个需要多角色协作、多轮迭代的工程问题。LangGraph 让我们能用声明式的方式表达这种复杂性，同时保持代码的可读性和可维护性。