4.3 问答系统构建

问答系统（Question Answering, QA）是自然语言处理领域最具应用价值的技术之一。从搜索引擎到智能客服，从教育辅助到法律咨询，问答系统无处不在。本章将重点讲解基于 Transformer 的抽取式问答系统，从经典 SQuAD 数据集出发，覆盖模型训练、评估、部署的完整流程，并深入探讨中文问答的适配方案。

4.3.1 问答系统分类

问答系统根据回答方式可以分为以下几类：

抽取式问答（Extractive QA）：从给定文本段落中直接抽取一个连续片段作为答案。代表模型有 BERT-QA、RoBERTa-QA。这是最成熟的方案，本章重点讲解。

生成式问答（Generative QA）：基于上下文自由生成答案文本，答案不一定原文中出现。代表模型有 T5、BART、GPT 系列。适合答案需要总结、推理的场景。

知识库问答（KBQA）：结合知识图谱，将自然语言问题转换为结构化查询（SPARQL），从知识库中检索答案。适合精确事实查询。

对话式问答（Conversational QA）：支持多轮对话上下文，能追问和澄清。代表系统如 Google 的 LaMDA、Meta 的 BlenderBot。

在 HuggingFace 生态中，抽取式问答系统是最成熟的落地场景，仅需几行代码即可完成部署。

4.3.2 SQuAD 数据集

数据集介绍

SQuAD（Stanford Question Answering Dataset）是问答领域最权威的评测基准，由斯坦福大学于 2016 年发布：

• SQuAD 1.1：10万对问答对，答案一定存在于段落中（不可回答问题不存在）
• SQuAD 2.0：15万对问答对，新增5万个"不可回答"问题，要求模型判断段落中是否包含答案

数据样例：

from datasets import load_dataset

dataset = load_dataset("squad")

# 查看数据结构
sample = dataset["train"][0]
print("上下文:", sample["context"][:100])
# "Architecturally, the school has a Catholic character..."
print("问题:", sample["question"])
# "To whom did the Virgin Mary allegedly appear in 1858?"
print("答案:", sample["answers"])
# {'text': ['Saint Bernadette Soubirous'], 'answer_start': [515]}
​

每条数据包含三个关键字段：context（参考段落）、question（问题）、answers（答案列表，包含文本和起始位置）。

数据统计与挑战

SQuAD 数据集的典型挑战包括：

多答案标注：同一个问题可能有多个合理的答案文本（不同标注者标注），模型需要处理这种模糊性。

长文档推理：段落长度通常在 100-500 词之间，模型需要理解长距离依赖关系来定位答案。

不可回答问题（SQuAD 2.0）：模型需要学会判断段落中是否包含答案，输出空答案而不是强行抽取。

4.3.3 抽取式问答建模

问题建模方式

抽取式问答被转化为两个预测任务：

1. 答案起始位置预测：在输入序列中，预测答案开始的 token 位置
2. 答案结束位置预测：在输入序列中，预测答案结束的 token 位置

模型输入的构造方式是将问题和段落拼接为一个序列：

[CLS] 问题文本 [SEP] 段落文本 [SEP]
​

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")

question = "Who founded Microsoft?"
context = "Microsoft was founded by Bill Gates and Paul Allen in 1975."

inputs = tokenizer(
    question,
    context,
    return_tensors="pt",
    max_length=384,
    truncation="only_second",  # 只截断段落部分
)
​

数据预处理

def preprocess_squad(examples):
    # Tokenize
    tokenized_examples = tokenizer(
        examples["question"],
        examples["context"],
        truncation="only_second",
        max_length=384,
        stride=128,       # 滑动窗口步长
        return_overflowing_tokens=True,   # 返回截断片段
        return_offsets_mapping=True,       # 返回字符到 token 的映射
    )
    
    # 处理答案位置映射
    sample_mapping = tokenized_examples.pop("overflow_to_sample_mapping")
    offset_mapping = tokenized_examples.pop("offset_mapping")
    
    tokenized_examples["start_positions"] = []
    tokenized_examples["end_positions"] = []
    
    for i, offsets in enumerate(offset_mapping):
        input_ids = tokenized_examples["input_ids"][i]
        cls_index = input_ids.index(tokenizer.cls_token_id)
        
        # 获取对应的样本
        sample_index = sample_mapping[i]
        answers = examples["answers"][sample_index]
        
        # 如果没有答案，标记为 [CLS]
        if len(answers["answer_start"]) == 0:
            tokenized_examples["start_positions"].append(cls_index)
            tokenized_examples["end_positions"].append(cls_index)
            continue
        
        # 字符级别的起始位置
        start_char = answers["answer_start"][0]
        end_char = start_char + len(answers["text"][0])
        
        # 映射到 token 位置
        start_token = None
        end_token = None
        
        for j, (start, end) in enumerate(offsets):
            if start <= start_char and end >= start_char:
                start_token = j
            if end >= end_char and start < end_char:
                end_token = j
                break
        
        # 如果答案在截断部分，设为不可回答
        if start_token is None or end_token is None:
            start_token = cls_index
            end_token = cls_index
        
        tokenized_examples["start_positions"].append(start_token)
        tokenized_examples["end_positions"].append(end_token)
    
    return tokenized_examples
​

长文档滑动窗口

当段落长度超过模型最大输入长度时，使用滑动窗口策略：

# stride=128, max_length=384 的效果：
# 窗口1: tokens[0:384]
# 窗口2: tokens[256:640]   ( stride=128，重叠256个token)
# 窗口3: tokens[512:896]
# ...
​

重叠区域保证了即使答案跨越窗口边界，也能在某个窗口中被完整包含。

4.3.4 模型训练

from transformers import (
    AutoModelForQuestionAnswering,
    TrainingArguments,
    Trainer,
)

model = AutoModelForQuestionAnswering.from_pretrained("bert-base-uncased")

# 训练配置
training_args = TrainingArguments(
    output_dir="./qa_model",
    num_train_epochs=3,
    per_device_train_batch_size=16,
    learning_rate=3e-5,
    warmup_ratio=0.1,
    weight_decay=0.01,
    eval_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    fp16=True,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_train,
    eval_dataset=tokenized_validation,
    tokenizer=tokenizer,
)

trainer.train()
​

后处理：选择最佳答案

当使用滑动窗口产生多个候选答案时，需要后处理选择最佳答案：

def postprocess_qa_predictions(examples, features, predictions, 
                                 n_best_size=20, max_answer_length=30):
    """从多个候选答案中选择最佳答案"""
    
    all_start_logits = predictions[0]
    all_end_logits = predictions[1]
    
    # 构建示例到特征的映射
    example_id_to_index = {k: i for i, k in enumerate(examples["id"])}
    features_per_example = collections.defaultdict(list)
    for i, feature in enumerate(features):
        features_per_example[example_id_to_index[feature["example_id"]]].append(i)
    
    predictions = []
    
    for example_index in range(len(examples)):
        feature_indices = features_per_example[example_index]
        
        # 从所有窗口中收集候选答案
        valid_answers = []
        
        for feature_index in feature_indices:
            start_logits = all_start_logits[feature_index]
            end_logits = all_end_logits[feature_index]
            
            # 取 top-k 起始和结束位置组合
            start_indexes = sorted(range(len(start_logits)), 
                                     key=lambda i: start_logits[i], 
                                     reverse=True)[:n_best_size]
            end_indexes = sorted(range(len(end_logits)), 
                                   key=lambda i: end_logits[i], 
                                   reverse=True)[:n_best_size]
            
            for start_index in start_indexes:
                for end_index in end_indexes:
                    # 起始必须在结束之前
                    if start_index >= end_index:
                        continue
                    # 答案长度限制
                    if end_index - start_index + 1 > max_answer_length:
                        continue
                    
                    score = start_logits[start_index] + end_logits[end_index]
                    valid_answers.append({
                        "score": score,
                        "text": decode_answer(feature, start_index, end_index),
                    })
        
        # 按分数排序，返回最佳答案
        if valid_answers:
            best_answer = sorted(valid_answers, 
                                  key=lambda x: x["score"], 
                                  reverse=True)[0]
            predictions.append({
                "id": examples["id"][example_index],
                "prediction_text": best_answer["text"],
                "no_answer_probability": 1.0 - best_answer["score"],
            })
        else:
            predictions.append({
                "id": examples["id"][example_index],
                "prediction_text": "",
                "no_answer_probability": 1.0,
            })
    
    return predictions
​

4.3.5 评估指标

Exact Match 和 F1

SQuAD 使用两个核心评估指标：

• Exact Match（EM）：预测答案与标准答案完全一致（忽略大小写和标点），精确匹配率
• F1：预测答案与标准答案的 token 级别 F1 值，容忍部分正确

import collections
import string

def compute_exact_match(prediction, truth):
    return prediction.strip() == truth.strip()

def compute_f1(prediction, truth):
    pred_tokens = prediction.strip().split()
    truth_tokens = truth.strip().split()
    
    common = collections.Counter(pred_tokens) & collections.Counter(truth_tokens)
    num_common = sum(common.values())
    
    if num_common == 0:
        return 0
    
    precision = num_common / len(pred_tokens)
    recall = num_common / len(truth_tokens)
    return 2 * precision * recall / (precision + recall)
​

当前 SQuAD 2.0 排行榜上，人类水平约为 EM=86.8、F1=89.5，顶尖模型已接近甚至超越人类水平。

4.3.6 中文问答适配

中文问答的特殊挑战

中文问答面临以下独特挑战：

分词问题：答案起始和结束位置的粒度取决于分词器。中文 BERT 以字为单位，答案边界是字符级别的。

答案长度差异：中文答案通常比英文更短更简洁，如"北京"、"5G"。

语义理解深度：中文的省略、回指、修辞等语言现象更复杂，对模型的理解能力要求更高。

中文问答数据集

# 常用中文问答数据集
# CMRC 2018：中文机器阅读理解数据集
# DuReader：百度大规模中文阅读理解数据集
# DRCD：繁体中文阅读理解数据集

cmrc_dataset = load_dataset("cmrc2018")
​

中文问答训练示例

from transformers import AutoTokenizer, AutoModelForQuestionAnswering

# 使用中文预训练模型
model_name = "hfl/chinese-roberta-wwm-ext-large"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForQuestionAnswering.from_pretrained(model_name)

# 中文问答推理
question = "中国的首都是哪个城市？"
context = "北京是中国的首都，也是中国的政治中心和文化中心。"

inputs = tokenizer(question, context, return_tensors="pt")
outputs = model(**inputs)

start_idx = outputs.start_logits.argmax()
end_idx = outputs.end_logits.argmax()
answer = tokenizer.decode(inputs["input_ids"][0][start_idx:end_idx+1])

print(f"问题: {question}")
print(f"答案: {answer}")  # "北京"
​

中文问答性能优化技巧

领域适配预训练：在目标领域（法律、医疗、金融）的文本上继续预训练 BERT，再微调 QA 任务，通常能提升 2-5 个 F1 点。

数据增强：使用回译、同义词替换、问题改写等方式扩充训练数据。

模型融合：将多个不同预训练模型（RoBERTa、MacBERT、ERNIE）的预测结果集成，取分数最高的答案。

4.3.7 生产级 QA 系统

Pipeline 部署

from transformers import pipeline

qa_pipeline = pipeline(
    "question-answering",
    model="./qa_model/best",
    tokenizer="./qa_model/best",
    device=0,
    max_answer_len=50,
)

# 单次问答
result = qa_pipeline(
    question="Transformers架构由哪篇论文提出？",
    context="Attention Is All You Need由Vaswani等人在2017年提出，"
            "引入了Transformer架构，使用自注意力机制替代了传统的"
            "循环和卷积结构。",
)
print(result)
# {'score': 0.892, 'start': 0, 'end': 22, 'answer': 'Attention Is All You Need'}

# 带置信度阈值
if result["score"] < 0.5:
    print("未找到高置信度答案")
​

多段落问答

实际应用中，通常需要先通过检索系统召回相关段落，再用 QA 模型精确抽取答案：

def multi_passage_qa(question, passages, top_k=3):
    """多段落问答：先检索相关段落，再逐段抽取答案"""
    
    results = []
    for passage in passages:
        result = qa_pipeline(question=question, context=passage)
        results.append({
            "answer": result["answer"],
            "score": result["score"],
            "context": passage,
        })
    
    # 按置信度排序，返回 top-k 答案
    results.sort(key=lambda x: x["score"], reverse=True)
    return results[:top_k]
​

这种"检索+阅读理解"的架构（Retrieval-Augmented Generation 的前身）是现代问答系统的标准范式，也是 RAG（Retrieval-Augmented Generation）技术的基础。

通过本章的学习，你已经掌握了用 Transformers 构建完整 QA 系统的全部技能，从数据处理到模型训练，从评估优化到生产部署，形成了一条完整的实战路径。