Evolution of Machine Translation

Machine Translation (MT) is a core NLP discipline focused on transforming text from a source language to a target language automatically. As global communication accelerates, the demand for robust MT systems has grown significantly. The field has evolved through three primary paradigms:

Rule-Based MT (1950s–1980s)

Early systems relied on hand-crafted linguistic rules and dictionaries. This approach treated translation as a direct substitution of words based on syntactic rules.

Pros:

• Straightforward logic and easy to interpret.

Cons:

• Lacks flexibility and struggles with context or ambiguous words.
• Hard to scale due to the manual effort required for rule creation.
• Poor robustness when facing novel sentence structures.

Statistical MT (1990s–2000s)

With the rise of digital corpora, statistical methods began analyzing bilingual text pairs to learn probabilistic translation patterns.

Pros:

• Data-driven, allowing better handling of lexical variations.
• Generally outperformed rule-based systems.

Cons:

• Requires massive parallel datasets, failing with low-resource languages.
• Performance degrades on long, complex sentences.

Neural MT (2010s–Present)

Deep learning introduced sequence-to-sequence (Seq2Seq) architectures and atttention mechanisms, revolutionizing translation quality.

Pros:

• Excels at capturing long-range dependencies and complex syntax.
• Benefited greatly from large pre-trained models (e.g., BERT, GPT).
• Supports few-shot learning scenarios better than previous methods.

Cons:

• Computationally expensive and data-hungry.
• Still faces challenges with low-resource languages.

Competition Workflow

The practical task involves training a model on the ModelScope platform and submitting predictions to the iFLYTEK competition portal.

1. Data Partitioning: Split data into training, validation, and test sets to ensure unbiased evaluation.
2. Model Training: Select an architecture (e.g., Transformer or GRU-based Seq2Seq) and monitor for overfitting.
3. Submission: Generate translations via the platform and upload results in the specific competition format.

Core Code Implemantation

Training Loop

def run_epoch(model, dataloader, optim, loss_fn, clip_value, device):
    model.train()
    total_loss = 0.0
    
    for src_batch, tgt_batch in dataloader:
        src_batch, tgt_batch = src_batch.to(device), tgt_batch.to(device)
        optim.zero_grad()
        
        predictions = model(src_batch, tgt_batch)
        
        # Reshape for loss calculation (ignore first token)
        preds_flat = predictions[:, 1:].reshape(-1, predictions.size(-1))
        targets_flat = tgt_batch[:, 1:].reshape(-1)
        
        batch_loss = loss_fn(preds_flat, targets_flat)
        batch_loss.backward()
        
        torch.nn.utils.clip_grad_norm_(model.parameters(), clip_value)
        optim.step()
        
        total_loss += batch_loss.item()
        
    return total_loss / len(dataloader)

Source Encoder

class SourceEncoder(nn.Module):
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers, dropout_rate):
        super().__init__()
        self.embed_layer = nn.Embedding(vocab_size, embed_size)
        self.gru_layer = nn.GRU(embed_size, hidden_size, num_layers, dropout=dropout_rate, batch_first=True)
        self.dropout_layer = nn.Dropout(dropout_rate)

    def forward(self, input_seq):
        embedded = self.dropout_layer(self.embed_layer(input_seq))
        outputs, hidden_state = self.gru_layer(embedded)
        return outputs, hidden_state

Target Decoder

class TargetDecoder(nn.Module):
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers, dropout_rate):
        super().__init__()
        self.vocab_size = vocab_size
        self.embed_layer = nn.Embedding(vocab_size, embed_size)
        self.gru_layer = nn.GRU(embed_size, hidden_size, num_layers, dropout=dropout_rate, batch_first=True)
        self.output_projection = nn.Linear(hidden_size, vocab_size)
        self.dropout_layer = nn.Dropout(dropout_rate)

    def forward(self, input_token, prev_hidden):
        embedded = self.dropout_layer(self.embed_layer(input_token))
        output, new_hidden = self.gru_layer(embedded, prev_hidden)
        prediction = self.output_projection(output.squeeze(1))
        return prediction, new_hidden

Seq2Seq Wrapper

class NeuralTranslator(nn.Module):
    def __init__(self, src_encoder, tgt_decoder, device):
        super().__init__()
        self.encoder = src_encoder
        self.decoder = tgt_decoder
        self.device = device

    def forward(self, src_seq, tgt_seq, teacher_forcing_prob=0.5):
        batch_sz = src_seq.size(0)
        seq_len = tgt_seq.size(1)
        vocab_sz = self.decoder.vocab_size
        
        outputs = torch.zeros(batch_sz, seq_len, vocab_sz).to(self.device)
        
        _, hidden_state = self.encoder(src_seq)
        
        current_input = tgt_seq[:, 0].unsqueeze(1)
        
        for t in range(1, seq_len):
            step_output, hidden_state = self.decoder(current_input, hidden_state)
            outputs[:, t, :] = step_output
            
            use_teacher = random.random() < teacher_forcing_prob
            top_pred = step_output.argmax(1)
            
            current_input = tgt_seq[:, t].unsqueeze(1) if use_teacher else top_pred.unsqueeze(1)
            
        return outputs

Execution Script

import time

start = time.time()

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# Load dictionaries and datasets
term_dict = load_terminology('./dataset/en-zh.dic')
lang_dataset = TranslationDataset('./dataset/train.txt', terminology=term_dict)

loader = DataLoader(lang_dataset, batch_size=32, shuffle=True, collate_fn=collate_fn)

# Model configuration
SRC_VOCAB = len(lang_dataset.en_vocab)
TGT_VOCAB = len(lang_dataset.zh_vocab)
EMB_SIZE = 512
HIDDEN = 1024
LAYERS = 3
DROP = 0.5

enc = SourceEncoder(SRC_VOCAB, EMB_SIZE, HIDDEN, LAYERS, DROP)
dec = TargetDecoder(TGT_VOCAB, EMB_SIZE, HIDDEN, LAYERS, DROP)
translator = NeuralTranslator(enc, dec, device).to(device)

optimizer = torch.optim.Adam(translator.parameters(), lr=0.0001)
loss_function = nn.CrossEntropyLoss(ignore_index=lang_dataset.zh_word2idx['<pad>'])

EPOCHS = 10
GRAD_CLIP = 1
best_loss = float('inf')

for epoch in range(EPOCHS):
    epoch_loss = run_epoch(translator, loader, optimizer, loss_function, GRAD_CLIP, device)
    print(f'Epoch {epoch+1} | Loss: {epoch_loss:.3f}')
    
    if epoch_loss < best_loss:
        best_loss = epoch_loss
        torch.save(translator.state_dict(), 'best_gru_model.pth')

print(f"Training completed in {(time.time() - start)/60:.2f} mins")

Evaluation Pipeline

Sentence Translation

def convert_sentence(text, model, data_handler, term_map, device, max_len=50):
    model.eval()
    with torch.no_grad():
        # Tokenize and map to indices
        words = data_handler.en_tokenizer(text)
        indices = [data_handler.en_word2idx.get(w, data_handler.en_word2idx['<sos>']) for w in words]
        src_tensor = torch.LongTensor(indices).unsqueeze(0).to(device)
        
        # Encode
        _, hidden = model.encoder(src_tensor)
        
        # Start decoding
        result = []
        input_tok = torch.LongTensor([[data_handler.zh_word2idx['<sos>']]]).to(device)
        
        for _ in range(max_len):
            out, hidden = model.decoder(input_tok, hidden)
            pred_idx = out.argmax(1).item()
            word = data_handler.zh_vocab[pred_idx]
            
            if word == '<eos>': break
            
            # Check terminology
            if word in term_map.values():
                word = [k for k, v in term_map.items() if v == word][0]
                
            result.append(word)
            input_tok = torch.LongTensor([[pred_idx]]).to(device)
            
    return ''.join(result)

BLEU Calculation

def calculate_bleu(model, data_handler, src_path, ref_path, term_map, device):
    model.eval()
    
    sources = [line.strip() for line in open(src_path, 'r', encoding='utf-8')]
    references = [line.strip() for line in open(ref_path, 'r', encoding='utf-8')]
    
    hypotheses = [convert_sentence(s, model, data_handler, term_map, device) for s in sources]
    
    bleu_metric = BLEU()
    return bleu_metric.corpus_score(hypotheses, [references])

Evaluation Execution

device = torch.device('cuda')
term_dict = load_terminology('./dataset/en-zh.dic')
data = TranslationDataset('./dataset/train.txt', terminology=term_dict)

# Initialize model
enc = SourceEncoder(len(data.en_vocab), 256, 512, 2, 0.5)
dec = TargetDecoder(len(data.zh_vocab), 256, 512, 2, 0.5)
model = NeuralTranslator(enc, dec, device).to(device)

# Load weights
model.load_state_dict(torch.load('best_gru_model.pth'))

# Run evaluation
score = calculate_bleu(model, data, './dataset/dev_en.txt', './dataset/dev_zh.txt', term_dict, device)
print(f'BLEU-4: {score.score:.2f}')

Inference Pipeline

Batch Translation

def batch_translate(model, data_handler, source_path, output_path, term_map, device):
    model.eval()
    src_lines = [line.strip() for line in open(source_path, 'r', encoding='utf-8')]
    
    translated = []
    for line in src_lines:
        translated.append(convert_sentence(line, model, data_handler, term_map, device))
    
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('\n'.join(translated))

Run Inference

device = torch.device('cuda')
term_dict = load_terminology('./dataset/en-zh.dic')
data = TranslationDataset('./dataset/train.txt', terminology=term_dict)

enc = SourceEncoder(len(data.en_vocab), 256, 512, 2, 0.5)
dec = TargetDecoder(len(data.zh_vocab), 256, 512, 2, 0.5)
model = NeuralTranslator(enc, dec, device).to(device)
model.load_state_dict(torch.load('best_gru_model.pth'))

batch_translate(model, data, './dataset/test_en.txt', './dataset/submit.txt', term_dict, device)
print('Inference complete.')