机器翻译
机器翻译(Machine Translation,MT)是自然语言处理中最具挑战性和实用价值的任务之一,其目标是将文本从一种语言自动翻译成另一种语言。从早期的基于规则的方法,到统计机器翻译,再到如今的神经机器翻译,机器翻译技术经历了巨大的变革。
什么是机器翻译
机器翻译是利用计算机自动将一种自然语言(源语言)翻译成另一种自然语言(目标语言)的技术。这不仅仅是简单的词对词替换,而是需要理解源语言的语义,并用目标语言重新表达。
机器翻译的核心挑战在于:语言之间的结构差异、一词多义的歧义性、文化背景的差异以及上下文的理解。比如英语中的 "apple" 在不同上下文中可能指水果或公司,翻译时需要根据语境判断。
机器翻译的发展历程
基于规则的方法(1950s-1980s)
早期的机器翻译系统依赖语言学家手工编写的翻译规则。这种方法需要大量的语言学知识,规则编写成本高,且难以覆盖语言的多样性和例外情况。当遇到规则未覆盖的情况时,翻译质量会急剧下降。
统计机器翻译(1990s-2010s)
统计机器翻译(Statistical Machine Translation,SMT)利用大规模平行语料库(源语言和目标语言的对照文本),通过统计方法学习翻译概率。
SMT 的核心思想是:给定源语言句子 ,找到最可能的目标语言句子 :
其中 是翻译模型(给定目标语言,源语言出现的概率), 是语言模型(目标语言句子的流畅程度)。
SMT 的代表性系统包括 IBM 模型、短语翻译模型和层级短语翻译模型。虽然 SMT 取得了很大成功,但它存在难以处理长距离依赖、需要大量特征工程等问题。
神经机器翻译(2014-至今)
神经机器翻译(Neural Machine Translation,NMT)使用端到端的神经网络模型,直接学习从源语言到目标语言的映射,无需人工设计特征。
2014 年,Seq2Seq(Sequence-to-Sequence)模型的提出标志着神经机器翻译时代的开始。2016 年,Google 宣布其翻译系统全面转向神经机器翻译,翻译质量大幅提升。
NMT 的优势在于:端到端学习、更好的泛化能力、能够捕捉长距离依赖、生成的译文更加流畅。
Seq2Seq 架构
Seq2Seq 是神经机器翻译的核心架构,由编码器(Encoder)和解码器(Decoder)两部分组成。
编码器-解码器框架
编码器将源语言句子编码为一个固定长度的向量表示,解码器根据这个向量生成目标语言句子。
import torch
import torch.nn as nn
class Encoder(nn.Module):
"""编码器:将输入序列编码为隐藏状态"""
def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=2):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.lstm = nn.LSTM(embed_dim, hidden_dim, num_layers, batch_first=True)
def forward(self, x):
# x: (batch_size, seq_len)
embedded = self.embedding(x) # (batch_size, seq_len, embed_dim)
outputs, (hidden, cell) = self.lstm(embedded)
# outputs: 所有时间步的输出
# hidden, cell: 最后时间步的隐藏状态
return outputs, (hidden, cell)
class Decoder(nn.Module):
"""解码器:根据编码器状态生成目标序列"""
def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=2):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.lstm = nn.LSTM(embed_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim, vocab_size)
def forward(self, x, hidden, cell):
# x: (batch_size, 1) - 当前时间步的输入
embedded = self.embedding(x) # (batch_size, 1, embed_dim)
output, (hidden, cell) = self.lstm(embedded, (hidden, cell))
prediction = self.fc(output) # (batch_size, 1, vocab_size)
return prediction, hidden, cell
class Seq2Seq(nn.Module):
"""完整的 Seq2Seq 模型"""
def __init__(self, encoder, decoder, device):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.device = device
def forward(self, src, trg, teacher_forcing_ratio=0.5):
"""
src: 源语言序列 (batch_size, src_len)
trg: 目标语言序列 (batch_size, trg_len)
teacher_forcing_ratio: 使用教师强制的概率
"""
batch_size = src.shape[0]
trg_len = trg.shape[1]
trg_vocab_size = self.decoder.fc.out_features
# 存储输出
outputs = torch.zeros(batch_size, trg_len, trg_vocab_size).to(self.device)
# 编码
_, (hidden, cell) = self.encoder(src)
# 解码器的第一个输入是 <sos> 标记
input = trg[:, 0].unsqueeze(1)
for t in range(1, trg_len):
# 解码
output, hidden, cell = self.decoder(input, hidden, cell)
outputs[:, t] = output.squeeze(1)
# 决定是否使用教师强制
teacher_force = torch.random(1).item() < teacher_forcing_ratio
# 获取预测的词
top1 = output.argmax(2)
# 下一个输入:使用真实标签或预测值
input = trg[:, t].unsqueeze(1) if teacher_force else top1
return outputs
# 创建模型
vocab_size_src = 10000
vocab_size_trg = 8000
embed_dim = 256
hidden_dim = 512
encoder = Encoder(vocab_size_src, embed_dim, hidden_dim)
decoder = Decoder(vocab_size_trg, embed_dim, hidden_dim)
model = Seq2Seq(encoder, decoder, device='cpu')
print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")
教师强制
教师强制(Teacher Forcing)是训练 Seq2Seq 模型的重要技术。在训练时,解码器的输入可以使用真实的目标词(教师强制)或模型自己的预测值。
教师强制的优势在于训练更稳定、收敛更快。但如果一直使用教师强制,模型在推理时可能会出现偏差,因为推理时只能使用模型自己的预测。因此在实际训练中,通常会随机混合使用两种方式。
注意力机制
Seq2Seq 模型的一个问题是:编码器将整个源句子压缩为一个固定长度的向量,这会导致信息丢失,尤其是对于长句子。注意力机制的提出解决了这个问题。
注意力机制的原理
注意力机制允许解码器在生成每个目标词时,"关注"源句子的不同部分。具体来说,解码器在每一步都会计算源句子所有位置的权重,然后加权求和得到一个上下文向量。
import torch.nn.functional as F
class Attention(nn.Module):
"""注意力机制"""
def __init__(self, hidden_dim):
super().__init__()
self.attn = nn.Linear(hidden_dim * 2, hidden_dim)
self.v = nn.Linear(hidden_dim, 1, bias=False)
def forward(self, hidden, encoder_outputs):
"""
hidden: 解码器当前隐藏状态 (batch_size, hidden_dim)
encoder_outputs: 编码器所有输出 (batch_size, src_len, hidden_dim)
"""
src_len = encoder_outputs.shape[1]
# 重复 hidden 以便与 encoder_outputs 计算
hidden = hidden.unsqueeze(1).repeat(1, src_len, 1)
# 计算能量分数
energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim=2)))
attention = self.v(energy).squeeze(2) # (batch_size, src_len)
# 归一化得到注意力权重
return F.softmax(attention, dim=1)
class AttnDecoder(nn.Module):
"""带注意力的解码器"""
def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=2):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.attention = Attention(hidden_dim)
self.lstm = nn.LSTM(hidden_dim + embed_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim * 2 + embed_dim, vocab_size)
def forward(self, x, hidden, cell, encoder_outputs):
# x: (batch_size, 1)
embedded = self.embedding(x) # (batch_size, 1, embed_dim)
# 计算注意力权重
attn_weights = self.attention(hidden[-1], encoder_outputs) # (batch_size, src_len)
# 计算上下文向量
attn_weights = attn_weights.unsqueeze(1) # (batch_size, 1, src_len)
context = torch.bmm(attn_weights, encoder_outputs) # (batch_size, 1, hidden_dim)
# 拼接嵌入和上下文
lstm_input = torch.cat((embedded, context), dim=2)
# LSTM
output, (hidden, cell) = self.lstm(lstm_input, (hidden, cell))
# 预测
output = torch.cat((output.squeeze(1), context.squeeze(1), embedded.squeeze(1)), dim=1)
prediction = self.fc(output)
return prediction, hidden, cell, attn_weights
注意力机制的核心在于:它让模型能够动态地选择源句子的哪些部分与当前生成相关,从而更好地处理长句子和复杂结构。
多头注意力
Transformer 模型使用了多头注意力(Multi-Head Attention),通过并行运行多个注意力函数,让模型能够同时关注不同位置的不同表示子空间。
import math
class MultiHeadAttention(nn.Module):
"""多头注意力机制"""
def __init__(self, embed_dim, num_heads):
super().__init__()
self.embed_dim = embed_dim
self.num_heads = num_heads
self.head_dim = embed_dim // num_heads
assert embed_dim % num_heads == 0, "embed_dim 必须能被 num_heads 整除"
self.q_proj = nn.Linear(embed_dim, embed_dim)
self.k_proj = nn.Linear(embed_dim, embed_dim)
self.v_proj = nn.Linear(embed_dim, embed_dim)
self.out_proj = nn.Linear(embed_dim, embed_dim)
def forward(self, query, key, value, mask=None):
batch_size = query.shape[0]
# 线性投影
Q = self.q_proj(query) # (batch_size, seq_len, embed_dim)
K = self.k_proj(key)
V = self.v_proj(value)
# 分割成多个头
Q = Q.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
K = K.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
V = V.view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
# 计算注意力分数
scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.head_dim)
# 应用掩码
if mask is not None:
scores = scores.masked_fill(mask == 0, float('-inf'))
# Softmax
attn_weights = F.softmax(scores, dim=-1)
# 加权求和
output = torch.matmul(attn_weights, V)
# 合并多头
output = output.transpose(1, 2).contiguous().view(batch_size, -1, self.embed_dim)
# 输出投影
output = self.out_proj(output)
return output, attn_weights
# 使用示例
embed_dim = 512
num_heads = 8
mha = MultiHeadAttention(embed_dim, num_heads)
# 模拟输入
batch_size = 2
seq_len = 10
x = torch.randn(batch_size, seq_len, embed_dim)
output, weights = mha(x, x, x)
print(f"输出形状: {output.shape}") # (2, 10, 512)
print(f"注意力权重形状: {weights.shape}") # (2, 8, 10, 10)
使用 Hugging Face 进行机器翻译
Hugging Face 提供了丰富的预训练翻译模型,可以直接使用或进行微调。
使用 Pipeline 进行翻译
最简单的方式是使用 pipeline API:
from transformers import pipeline
# 创建翻译管道
# 支持的语言对包括:en-zh(英译中)、zh-en(中译英)、en-fr(英译法)等
translator = pipeline("translation_en_to_zh", model="Helsinki-NLP/opus-mt-en-zh")
# 翻译文本
text = "Natural language processing is a fascinating field of artificial intelligence."
result = translator(text)
print(f"原文: {text}")
print(f"译文: {result[0]['translation_text']}")
# 输出: 自然语言处理是人工智能的一个迷人领域。
使用多语言模型
对于多语言翻译,可以使用支持多语言的模型:
from transformers import pipeline
# 使用支持多语言的模型
translator = pipeline("translation", model="facebook/nllb-200-distilled-600M")
# 翻译时指定源语言和目标语言
text = "Hello, how are you?"
# 英语到中文
result = translator(text, src_lang="eng_Latn", tgt_lang="zho_Hans")
print(f"英译中: {result[0]['translation_text']}")
# 英语到日语
result = translator(text, src_lang="eng_Latn", tgt_lang="jpn_Jpan")
print(f"英译日: {result[0]['translation_text']}")
# 英语到法语
result = translator(text, src_lang="eng_Latn", tgt_lang="fra_Latn")
print(f"英译法: {result[0]['translation_text']}")
NLLB(No Language Left Behind)是 Meta 开发的多语言翻译模型,支持 200 多种语言,适合需要翻译多种语言的场景。
使用 MarianMT 模型
MarianMT 是另一个流行的翻译模型系列:
from transformers import MarianMTModel, MarianTokenizer
# 加载模型和分词器
model_name = "Helsinki-NLP/opus-mt-zh-en"
tokenizer = MarianTokenizer.from_pretrained(model_name)
model = MarianMTModel.from_pretrained(model_name)
# 准备文本
texts = [
"自然语言处理是人工智能的重要分支",
"深度学习改变了机器翻译的格局"
]
# 分词
inputs = tokenizer(texts, return_tensors="pt", padding=True, truncation=True)
# 翻译
translated = model.generate(**inputs)
# 解码
translations = [tokenizer.decode(t, skip_special_tokens=True) for t in translated]
for src, tgt in zip(texts, translations):
print(f"原文: {src}")
print(f"译文: {tgt}")
print()
使用 T5 进行翻译
T5(Text-to-Text Transfer Transformer)是一个通用的文本到文本模型,也支持翻译任务:
from transformers import T5ForConditionalGeneration, T5Tokenizer
# 加载模型
model_name = "google-t5/t5-base"
tokenizer = T5Tokenizer.from_pretrained(model_name)
model = T5ForConditionalGeneration.from_pretrained(model_name)
# T5 需要添加任务前缀
text = "translate English to French: The weather is nice today."
# 分词
inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True)
# 生成翻译
outputs = model.generate(
inputs["input_ids"],
max_length=150,
num_beams=4,
early_stopping=True
)
# 解码
translation = tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"译文: {translation}")
微调翻译模型
当预训练模型的翻译效果不能满足特定领域需求时,可以对模型进行微调。
准备数据
微调需要平行语料数据,即源语言和目标语言的对照文本:
from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, DataCollatorForSeq2Seq
from transformers import Seq2SeqTrainingArguments, Seq2SeqTrainer
# 加载数据集
# 这里使用 OPUS Books 数据集作为示例
dataset = load_dataset("opus_books", "en-zh")
# 查看数据结构
print(dataset)
print(dataset["train"][0])
# 加载分词器和模型
model_name = "Helsinki-NLP/opus-mt-en-zh"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
# 预处理函数
def preprocess_function(examples):
inputs = [ex["en"] for ex in examples["translation"]]
targets = [ex["zh"] for ex in examples["translation"]]
model_inputs = tokenizer(
inputs,
max_length=128,
truncation=True,
padding="max_length"
)
# 分词目标文本
with tokenizer.as_target_tokenizer():
labels = tokenizer(
targets,
max_length=128,
truncation=True,
padding="max_length"
)
model_inputs["labels"] = labels["input_ids"]
return model_inputs
# 预处理数据集
tokenized_dataset = dataset.map(preprocess_function, batched=True)
# 数据整理器
data_collator = DataCollatorForSeq2Seq(tokenizer, model=model)
训练配置和微调
import numpy as np
import evaluate
# 加载评估指标
metric = evaluate.load("sacrebleu")
def compute_metrics(eval_preds):
preds, labels = eval_preds
# 解码预测
if isinstance(preds, tuple):
preds = preds[0]
decoded_preds = tokenizer.batch_decode(preds, skip_special_tokens=True)
# 处理标签
labels = np.where(labels != -100, labels, tokenizer.pad_token_id)
decoded_labels = tokenizer.batch_decode(labels, skip_special_tokens=True)
# 计算 BLEU 分数
decoded_preds = [pred.strip() for pred in decoded_preds]
decoded_labels = [[label.strip()] for label in decoded_labels]
result = metric.compute(predictions=decoded_preds, references=decoded_labels)
return {"bleu": result["score"]}
# 训练参数
training_args = Seq2SeqTrainingArguments(
output_dir="./translation_model",
evaluation_strategy="epoch",
learning_rate=2e-5,
per_device_train_batch_size=16,
per_device_eval_batch_size=16,
weight_decay=0.01,
save_total_limit=3,
num_train_epochs=3,
predict_with_generate=True,
fp16=True, # 使用混合精度加速训练
push_to_hub=False
)
# 创建训练器
trainer = Seq2SeqTrainer(
model=model,
args=training_args,
train_dataset=tokenized_dataset["train"],
eval_dataset=tokenized_dataset["train"].select(range(100)), # 使用部分数据评估
tokenizer=tokenizer,
data_collator=data_collator,
compute_metrics=compute_metrics
)
# 开始训练
trainer.train()
使用微调后的模型
# 加载微调后的模型
model_path = "./translation_model/checkpoint-best"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForSeq2SeqLM.from_pretrained(model_path)
# 翻译函数
def translate(text, model, tokenizer, max_length=128):
inputs = tokenizer(text, return_tensors="pt", max_length=max_length, truncation=True)
outputs = model.generate(
inputs["input_ids"],
max_length=max_length,
num_beams=5,
early_stopping=True
)
return tokenizer.decode(outputs[0], skip_special_tokens=True)
# 测试
text = "Machine learning is transforming the way we process natural language."
translation = translate(text, model, tokenizer)
print(f"译文: {translation}")
翻译质量评估
BLEU 分数
BLEU(Bilingual Evaluation Understudy)是最常用的机器翻译评估指标。它通过计算机器翻译结果与参考译文之间的 n-gram 重叠度来衡量翻译质量。
from nltk.translate.bleu_score import sentence_bleu, corpus_bleu
from nltk.translate.bleu_score import SmoothingFunction
# 单句 BLEU 计算
reference = [["这", "是", "一个", "测试"]]
candidate = ["这", "是", "一个", "测试"]
score = sentence_bleu(reference, candidate)
print(f"BLEU 分数: {score:.4f}")
# 使用平滑处理
smoothie = SmoothingFunction().method1
score = sentence_bleu(reference, candidate, smoothing_function=smoothie)
print(f"平滑后 BLEU 分数: {score:.4f}")
# 多个参考译文
references = [
[["这", "是", "一个", "测试"], ["这", "是", "测试"]],
[["我", "喜欢", "学习"], ["我", "爱", "学习"]]
]
candidates = [
["这", "是", "一个", "测试"],
["我", "喜欢", "学习"]
]
# 语料级别 BLEU
score = corpus_bleu(references, candidates)
print(f"语料 BLEU 分数: {score:.4f}")
BLEU 的优点是计算简单、与人工评估相关性较高。但它也有局限性:只关注精确匹配、不考虑语义等价性、对参考译文的数量和质量敏感。
其他评估指标
# 使用 sacrebleu 库
import evaluate
# 加载指标
bleu = evaluate.load("sacrebleu")
meteor = evaluate.load("meteor")
bertscore = evaluate.load("bertscore")
predictions = ["这是一个测试句子"]
references = [["这是一个测试句子"]]
# 计算 BLEU
bleu_result = bleu.compute(predictions=predictions, references=references)
print(f"BLEU: {bleu_result['score']}")
# 计算 METEOR(考虑同义词和词形变化)
meteor_result = meteor.compute(predictions=predictions, references=references)
print(f"METEOR: {meteor_result['meteor']}")
# 计算 BERTScore(基于语义相似度)
bertscore_result = bertscore.compute(
predictions=predictions,
references=references,
lang="zh"
)
print(f"BERTScore F1: {bertscore_result['f1'][0]:.4f}")
高级技术
束搜索
在推理时,我们希望找到最可能的翻译序列。贪心搜索每步选择概率最高的词,但可能错过全局最优解。束搜索(Beam Search)保留多个候选序列,提高找到更好翻译的可能性。
def beam_search_decode(model, tokenizer, src_text, beam_size=5, max_length=50):
"""束搜索解码"""
# 编码输入
inputs = tokenizer(src_text, return_tensors="pt")
# 使用模型的 generate 方法,它内置了束搜索
outputs = model.generate(
inputs["input_ids"],
max_length=max_length,
num_beams=beam_size,
num_return_sequences=3, # 返回 top-3 结果
early_stopping=True
)
# 解码所有候选
translations = []
for output in outputs:
text = tokenizer.decode(output, skip_special_tokens=True)
translations.append(text)
return translations
# 使用示例
translations = beam_search_decode(model, tokenizer, "Hello world", beam_size=5)
for i, trans in enumerate(translations):
print(f"候选 {i+1}: {trans}")
处理长文本
对于超过模型最大长度的长文本,需要分段处理:
def translate_long_text(text, model, tokenizer, max_length=512):
"""翻译长文本"""
# 按句子分割
sentences = text.split('. ')
translations = []
current_chunk = ""
for sentence in sentences:
# 如果当前块加上新句子不超过长度限制
if len(current_chunk) + len(sentence) < max_length:
current_chunk += sentence + ". "
else:
# 翻译当前块
if current_chunk:
trans = translate(current_chunk, model, tokenizer, max_length)
translations.append(trans)
current_chunk = sentence + ". "
# 翻译最后一块
if current_chunk:
trans = translate(current_chunk, model, tokenizer, max_length)
translations.append(trans)
return " ".join(translations)
实际应用案例
实时翻译 API
from fastapi import FastAPI
from pydantic import BaseModel
from transformers import pipeline
app = FastAPI()
# 加载翻译模型
translator = pipeline("translation", model="Helsinki-NLP/opus-mt-en-zh")
class TranslationRequest(BaseModel):
text: str
source_lang: str = "en"
target_lang: str = "zh"
@app.post("/translate")
async def translate_text(request: TranslationRequest):
"""翻译 API"""
result = translator(request.text)
return {
"original": request.text,
"translation": result[0]["translation_text"]
}
# 运行: uvicorn app:app --reload
批量文档翻译
import pandas as pd
from tqdm import tqdm
def batch_translate(documents, model, tokenizer, batch_size=8):
"""批量翻译文档"""
translations = []
for i in tqdm(range(0, len(documents), batch_size)):
batch = documents[i:i+batch_size]
# 分词
inputs = tokenizer(
batch,
return_tensors="pt",
padding=True,
truncation=True,
max_length=512
)
# 翻译
outputs = model.generate(**inputs, max_length=512)
# 解码
batch_translations = [
tokenizer.decode(output, skip_special_tokens=True)
for output in outputs
]
translations.extend(batch_translations)
return translations
# 使用示例
# documents = ["Document 1...", "Document 2...", ...]
# translations = batch_translate(documents, model, tokenizer)
常见问题与解决方案
翻译质量不稳定
翻译质量受多种因素影响:源文本质量、领域差异、模型训练数据等。可以通过以下方式改善:
- 文本预处理:清洗噪声、统一格式
- 领域微调:在特定领域数据上微调模型
- 后处理:修正常见错误、调整格式
处理低资源语言
对于训练数据少的语言对:
- 使用多语言模型:如 mBART、NLLB
- 迁移学习:从高资源语言对迁移知识
- 数据增强:回译、合成数据
翻译速度优化
# 使用 ONNX 加速推理
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
from optimum.onnxruntime import ORTModelForSeq2SeqLM
# 加载 ONNX 模型
onnx_model = ORTModelForSeq2SeqLM.from_pretrained(
"Helsinki-NLP/opus-mt-en-zh",
export=True
)
tokenizer = AutoTokenizer.from_pretrained("Helsinki-NLP/opus-mt-en-zh")
# ONNX 模型推理更快
总结
机器翻译是 NLP 的核心应用之一,本章介绍了:
- 发展历程:从规则方法到统计方法再到神经方法
- Seq2Seq 架构:编码器-解码器框架
- 注意力机制:让模型动态关注源句子的不同部分
- 预训练模型:使用 Hugging Face 进行翻译
- 模型微调:针对特定领域优化翻译质量
- 评估指标:BLEU 等常用指标
- 实际应用:API 开发和批量处理
机器翻译技术仍在快速发展,大语言模型的出现为翻译带来了新的可能性。掌握这些技术,可以帮助你在实际项目中构建高质量的翻译系统。