AGI 之 【Hugging Face】 的【问答系统】的 [评估并改进问答Pipeline] / [ 生成式问答 ] 的简单整理
目录
AGI 之 【Hugging Face】 的【问答系统】的 [评估并改进问答Pipeline] / [ 生成式问答 ] 的简单整理
一、简单介绍
二、构建问答系统
三、评估并改进问答pipeline
1、评估检索器
2、评估阅读器
3、领域自适应
4、评估整体问答pipeline
四、生成式问答
五、整理
一、简单介绍
AGI，即通用人工智能（Artificial General Intelligence），是一种具备人类智能水平的人工智能系统。它不仅能够执行特定的任务，而且能够理解、学习和应用知识于广泛的问题解决中，具有较高的自主性和适应性。AGI的能力包括但不限于自我学习、自我改进、自我调整，并能在没有人为干预的情况下解决各种复杂问题。
- AGI能做的事情非常广泛：
跨领域任务执行：AGI能够处理多领域的任务，不受限于特定应用场景。
自主学习与适应：AGI能够从经验中学习，并适应新环境和新情境。
创造性思考：AGI能够进行创新思维，提出新的解决方案。
社会交互：AGI能够与人类进行复杂的社会交互，理解情感和社会信号。
- 关于AGI的未来发展前景，它被认为是人工智能研究的最终目标之一，具有巨大的变革潜力：
技术创新：随着机器学习、神经网络等技术的进步，AGI的实现可能会越来越接近。
跨学科整合：实现AGI需要整合计算机科学、神经科学、心理学等多个学科的知识。
伦理和社会考量：AGI的发展需要考虑隐私、安全和就业等伦理和社会问题。
增强学习和自适应能力：未来的AGI系统可能利用先进的算法，从环境中学习并优化行为。
多模态交互：AGI将具备多种感知和交互方式，与人类和其他系统交互。
Hugging Face作为当前全球最受欢迎的开源机器学习社区和平台之一，在AGI时代扮演着重要角色。它提供了丰富的预训练模型和数据集资源，推动了机器学习领域的发展。Hugging Face的特点在于易用性和开放性，通过其Transformers库，为用户提供了方便的模型处理文本的方式。随着AI技术的发展，Hugging Face社区将继续发挥重要作用，推动AI技术的发展和应用，尤其是在多模态AI技术发展方面，Hugging Face社区将扩展其模型和数据集的多样性，包括图像、音频和视频等多模态数据。
- 在AGI时代，Hugging Face可能会通过以下方式发挥作用：
模型共享：作为模型共享的平台，Hugging Face将继续促进先进的AGI模型的共享和协作。
开源生态：Hugging Face的开源生态将有助于加速AGI技术的发展和创新。
工具和服务：提供丰富的工具和服务，支持开发者和研究者在AGI领域的研究和应用。
伦理和社会责任：Hugging Face注重AI伦理，将推动负责任的AGI模型开发和应用，确保技术进步同时符合伦理标准。
AGI作为未来人工智能的高级形态，具有广泛的应用前景，而Hugging Face作为开源社区，将在推动AGI的发展和应用中扮演关键角色。
（注意：以下代码运行，可能需要科学上网）
二、构建问答系统
无论你是研究人员、分析师还是数据科学家，都很有可能需要在浩如烟海的文档中跋山涉水才能找到你所需要的信息。最让人感到崩溃的是，在使用Google或者Bing搜索引擎的时候，它们还不断提醒你，还有更好的搜索方法。例如，使用Google搜索：玛丽·居里什么时候获得她的第一个诺贝尔奖？可以立即得到正确答案：1903，如图下图所示
虽然在这种情况下，每个人都知道DropC是最好的吉他调音方式。
在这个例子中，Google首先检索出了大约319 000个与查询信息相关的文档，然后执行了另一个处理步骤，即从这些文档中提取出带有相应段落和网页的答案片段。所以，搜索引擎的每条搜索结果看起来似乎对你都很有用。再比如，使用Google搜索一个稍微棘手的问题：“哪种吉他调音最好？”这次并没有直接显示出答案片段，而是需要点击搜索引擎推荐的网页链接，跳转到各个网站中才能找到我们想要的答案。
这项技术背后的方法被称为问答系统，问答系统有各种各样的类型，但最为常见的是提取式问答系统，它将所涉问题的答案识别为文档中的一小段文本，这里的文档可以是网页、法律合同或新闻文章。这种先查到相关文档，然后再从中提取答案的两阶段处理方式，是许多现代问答系统的理论基础，像语义搜索引擎、智能助手或者自动信息提取器，都是基于这种理论来构建的。在本节中，我们将使用这种处理方式来解决电商网站所面临的一个常见问题：帮助消费者解答特定问题，帮助其了解一个商品。在这个场景中，把用户的评论当作问答系统的文本数据源，在此过程中，我们将了解到Transformer模型如何作为强大的阅读理解工具，来从文本中提取有价值的信息。
- 什么是问答系统？
问答系统（Question Answering, QA）是一种自然语言处理（NLP）任务，旨在从给定的文本或知识库中找到问题的答案。这些系统可以分为以下几种类型：
- 开放域问答系统：能够回答关于广泛主题的问题，通常需要从大量未结构化数据中提取答案。
- 封闭域问答系统：专注于特定领域内的问题，通常基于领域特定的知识库。
- 基于文本的问答系统：直接从给定的文本（如一段文章）中提取答案。
- 知识库问答系统：从结构化的知识库（如数据库）中检索答案。
- 问答系统的实现方式
构建问答系统有多种方法，每种方法都有其优缺点。以下是一些常见的实现方式：
- 1. 基于规则的方法
这种方法依赖于预定义的规则和模板来匹配问题并从文本中提取答案。这种方法适用于简单且格式固定的问答任务，但在处理复杂和多变的问题时效果不佳。
- 2. 基于信息检索的方法
这类系统首先使用信息检索技术找到与问题相关的文档，然后从这些文档中提取答案。典型的步骤包括：
- 文档检索：从大型文档库中找到最相关的文档。
- 段落检索：从相关文档中找到最相关的段落。
- 答案提取：从相关段落中提取答案。
- 3. 基于深度学习的方法
深度学习方法已经显著提高了问答系统的性能。常见的模型包括：
- BERT（Bidirectional Encoder Representations from Transformers）：一个双向Transformer模型，可以理解上下文并从中提取答案。
- RoBERTa（Robustly optimized BERT approach）：BERT的改进版，性能更强。
- DistilBERT：BERT的轻量级版本，计算效率更高。
这些模型通常在大型问答数据集（如SQuAD）上进行预训练，然后在特定任务上进行微调。
基于深度学习的问答系统构建步骤
以下是使用Hugging Face的Transformers库构建基于深度学习的问答系统的步骤：
- 1. 安装库
pip install transformers torch2. 加载预训练模型和Tokenizer
from transformers import AutoTokenizer, AutoModelForQuestionAnswering # 加载预训练的BERT模型和tokenizer tokenizer = AutoTokenizer.from_pretrained("bert-large-uncased-whole-word-masking-finetuned-squad") model = AutoModelForQuestionAnswering.from_pretrained("bert-large-uncased-whole-word-masking-finetuned-squad")3. 编写问答函数
import torch def answer_question(question, context): inputs = tokenizer.encode_plus(question, context, return_tensors="pt") input_ids = inputs["input_ids"].tolist()[0] outputs = model(**inputs) answer_start_scores = outputs.start_logits answer_end_scores = outputs.end_logits # 获取答案的开始和结束位置 answer_start = torch.argmax(answer_start_scores) answer_end = torch.argmax(answer_end_scores) + 1 # 将答案token转换为字符串 answer = tokenizer.convert_tokens_to_string(tokenizer.convert_ids_to_tokens(input_ids[answer_start:answer_end])) return answer4. 测试问答系统
context = "Hugging Face 是一个开源社区，专注于自然语言处理技术。" question = "Hugging Face 是什么？" answer = answer_question(question, context) print(f"Question: {question}") print(f"Answer: {answer}")
本节着重介绍提取式问答，不同场景适用的问答形式也不同。比如，技术社区的问答系统收集的是用户在Stack Overflow（https://stackoverflow.com）等技术论坛上的问答对，然后使用语义相似度搜索来找到与新问题最匹配的答案。还有开放域长格式（long-form）问答，旨在为诸如“为什么天空是蓝色的？”之类的开放性问题生成复杂的长文本答案。另外还可以针对表格，像TAPAS（https://oreil.ly/vVPWO）这种Transformer模型能为表格生成聚合操作语句，来获取相关信息。
三、评估并改进问答pipeline
尽管业界对于问答系统的研究大部分都聚焦在改进阅读理解模型上，但是在实际应用场景中，如果检索器一开始就检索不到相关文档，那么即使拥有再好的阅读器也无济于事。实质上，检索器的性能决定了整个问答系统的性能上限，因此检索器使用是否得当就显得很重要。考虑到这一点，我们来看看用以评估检索器的一些常用指标，对比稀疏检索器和密集检索器的性能差异。
1、评估检索器
评估检索器的一个常用指标是召回率，它表示检索出的相关文档数与所有的相关文档数的比率，在这种情况下，“相关”仅仅表示答案是否存在于一段文本中。所以如果给定一组问题，可以通过计算一个答案出现在检索器返回的前k个文档中的次数来计算召回率。
在Haystack中，有两种评估检索器的方法：
- 使用检索器内置的eval()方法。该方法可用于开放域和封闭域的问答，但不能用于像SubjQA这样的数据集，因为每个文档都与单个商品相关联，每个问题都需按商品维度进行过滤。
- 构建一个将检索器和EvalRetriever类相结合的自定义Pipeline，这种方式支持定制评估函数和检索流程。
召回率的一个补充指标是平均查准率（mean Average Precision，mAP），它可以让那种检索文档更加精准的检索器获得正向收益。
由于需要评估每个商品的召回率，然后进行聚合，所以这里更倾向于使用第二种方法。Pipeline是以图的形式组织的，图中的每个节点表示一个类，该类有输入和输出，通过run()函数接收前一个节点的输出作为其输入：
以上代码段的kwargs参数接收图中前一个节点的输出，在run()方法中做一些处理以后将元组连同传出边（outgoing edge）名称输出给下一个节点，另外，PipelineNode还包含一个outgoing_edgs参数，表示该节点输出的边数（在大多数情况下outgoing_edgs=1，只有在pipeline中还存在其他分支节点才需要），给它正确赋值即可。
在这个案例中，需要用一个节点来评估检索器，这里会使用到EvalRetriever类，它的run()函数会追踪那些与ground truth集合匹配的文档。使用该类，就可以在检索器节点之后添加评估节点来构建Pipeline：
# 从 Haystack 的 pipeline 模块中导入 Pipeline
from haystack.pipeline import Pipeline
# 从 Haystack 的 eval 模块中导入 EvalDocuments
from haystack.eval import EvalDocuments
# 定义一个名为 EvalRetrieverPipeline 的类
class EvalRetrieverPipeline:
def __init__(self, retriever):
# 初始化 EvalRetrieverPipeline 类，并接收一个检索器 (retriever) 作为参数
self.retriever = retriever
# 创建一个 EvalDocuments 实例，用于评估检索结果
self.eval_retriever = EvalDocuments()
# 创建一个 Pipeline 实例
pipe = Pipeline()
# 将检索器组件添加到管道中，命名为 "ESRetriever"，输入为 "Query"
pipe.add_node(component=self.retriever, name="ESRetriever", inputs=["Query"])
# 将评估组件添加到管道中，命名为 "EvalRetriever"，输入为 "ESRetriever"
pipe.add_node(component=self.eval_retriever, name="EvalRetriever", inputs=["ESRetriever"])
# 将配置好的管道赋值给实例变量 self.pipeline
self.pipeline = pipe
# 使用之前创建的 es_retriever 实例，创建 EvalRetrieverPipeline 实例
pipe = EvalRetrieverPipeline(es_retriever)
以上代码段中会为每个节点设置名称和输入信息inputs。在大多数情况下，每个节点都有一个出边，所以只需要在inputs中设置前一个节点的名称即可。
至此，我们制作了包含评估节点的pipeline，现在只需要将问题和对应的答案传给该pipeline用于评估。为此，我们需要将答案添加到文档存储的label索引中。Haystack提供了一个Label对象，它将答案片段和元数据以规范化的方式进行封装。我们通过迭代测试集中的每个问题，同时提取匹配的答案和其他元数据来创建Label对象数组：
# 从 Haystack 导入 Label
from haystack import Label
# 创建一个空列表，用于存储标签
labels = []
# 遍历 "test" 数据集中的每一行
for i, row in dfs["test"].iterrows():
# 创建元数据字典，用于在检索器中进行过滤
meta = {"item_id": row["title"], "question_id": row["id"]}
# 为有答案的问题填充标签
if len(row["answers.text"]):
for answer in row["answers.text"]:
# 创建一个 Label 实例，并设置相应的属性
label = Label(
question=row["question"], # 设置问题
answer=answer, # 设置答案
id=i, # 设置标签的唯一标识符
origin=row["id"], # 设置标签的来源标识符
meta=meta, # 设置元数据
is_correct_answer=True, # 标记为正确答案
is_correct_document=True, # 标记为正确文档
no_answer=False # 标记该标签有答案
)
# 将标签添加到列表中
labels.append(label)
# 为没有答案的问题填充标签
else:
# 创建一个 Label 实例，并设置相应的属性
label = Label(
question=row["question"], # 设置问题
answer="", # 设置答案为空
id=i, # 设置标签的唯一标识符
origin=row["id"], # 设置标签的来源标识符
meta=meta, # 设置元数据
is_correct_answer=True, # 标记为正确答案
is_correct_document=True, # 标记为正确文档
no_answer=True # 标记该标签没有答案
)
# 将标签添加到列表中
labels.append(label)
我们看看封装后的Label对象中都有什么内容：
# 打印标签列表中的第一个标签
print(labels[0])
运行结果：
{'id': 'e28f5e62-85e8-41b2-8a34-fbff63b7a466', 'created_at': None, 'updated_at':
None, 'question': 'What is the tonal balance of these headphones?', 'answer': 'I
have been a headphone fanatic for thirty years', 'is_correct_answer': True,
'is_correct_document': True, 'origin': 'd0781d13200014aa25860e44da9d5ea7',
'document_id': None, 'offset_start_in_doc': None, 'no_answer': False,
'model_id': None, 'meta': {'item_id': 'B00001WRSJ', 'question_id':
'd0781d13200014aa25860e44da9d5ea7'}}
从中我们可以看到问答对，以及包含问题的唯一ID的origin字段，这样就能按问题过滤文档存储内容。此外，我们还将item_id添加到了meta字段中，以便可以按商品过滤label，得到label之后，再将它们添加到Elasticsearch的label索引，如下所示：
# 将标签写入到 Elasticsearch 的 label 索引中
document_store.write_labels(labels, index="label")
# 打印写入到 label 索引中的问题-答案对的数量
print(f"""Loaded {document_store.get_label_count(index="label")} \
question-answer pairs""")
运行结果：
Loaded 358 question-answer pairs
接下来，我们需要将问题ID和相应的答案之间建立一个映射关系，之后传给pipeline。为了获取所有label，我们可以使用文档存储中的get_all_labels_aggregated()方法，该方法会聚合与唯一ID关联的所有问答对，这个方法会返回MultiLabel对象数组，但是在我们的例子中只返回了一个元素，那是因为我们是按问题ID来进行过滤的。下面是建立聚合label的示例：
# 从 Elasticsearch 的 label 索引中获取聚合的标签
labels_agg = document_store.get_all_labels_aggregated(
index="label", # 指定索引为 label
open_domain=True, # 设置为开放域
aggregate_by_meta=["item_id"] # 按 item_id 聚合标签
)
# 打印聚合后的标签数量
print(len(labels_agg))
运行结果：
330
通过查看其中一个label，我们可以看到与给定问题相关的所有答案都被聚合在multiple_answers字段中：
# 打印聚合的标签中的第 109 个标签
print(labels_agg[109])
运行结果：
{'question': 'How does the fan work?', 'multiple_answers': ['the fan is really
really good', "the fan itself isn't super loud. There is an adjustable dial to
change fan speed"], 'is_correct_answer': True, 'is_correct_document': True,
'origin': '5a9b7616541f700f103d21f8ad41bc4b', 'multiple_document_ids': [None,
None], 'multiple_offset_start_in_docs': [None, None], 'no_answer': False,
'model_id': None, 'meta': {'item_id': 'B002MU1ZRS'}}
现在我们已经了解了评估检索器的所有关键步骤，下面我们就定义一个函数，将与每个商品关联的问答对都发送到评估pipeline，并在pipe对象中追踪正确的检索结果：
# 定义运行管道的函数
def run_pipeline(pipeline, top_k_retriever=10, top_k_reader=4):
# 遍历所有聚合的标签
for l in labels_agg:
# 运行管道，传入问题、检索器的 top_k 参数、阅读器的 top_k 参数、评估文档的 top_k 参数、标签以及过滤器
_ = pipeline.pipeline.run(
query=l.question, # 查询问题
top_k_retriever=top_k_retriever, # 检索器返回的文档数量
top_k_reader=top_k_reader, # 阅读器返回的答案数量
top_k_eval_documents=top_k_retriever, # 评估文档的数量
labels=l, # 标签
filters={"item_id": [l.meta["item_id"]], "split": ["test"]} # 过滤器，根据 item_id 和 split 过滤文档
)
# 运行管道，并指定检索器返回的文档数量为 3
run_pipeline(pipe, top_k_retriever=3)
# 打印召回率，表示在前 3 个返回的文档中有多少是相关的
print(f"Recall@3: {pipe.eval_retriever.recall:.2f}")
运行结果：
Recall@3: 0.95
以上代码执行后成功计算出了召回率，这里为top_k_retriever设置了一个比较特殊的值来指定需要检索的文档数量。一般来说，增大这个值会提高召回率，但代价是会向阅读器输出更多文档，并且会降低端到端pipeline的处理速度。为了设置一个恰当的k值，我们编写了一个函数，依次计算数组中每个k值在测试集的召回率：
# 定义一个函数，用于评估检索器的性能
def evaluate_retriever(retriever, topk_values=[1, 3, 5, 10, 20]):
topk_results = {}
# 遍历不同的 top-k 值
for topk in topk_values:
# 创建评估管道
p = EvalRetrieverPipeline(retriever)
# 运行管道，评估检索器性能
run_pipeline(p, top_k_retriever=topk)
# 获取评估指标（这里只获取召回率）
topk_results[topk] = {"recall": p.eval_retriever.recall}
# 将结果转换为 DataFrame 格式并返回
return pd.DataFrame.from_dict(topk_results, orient="index")
# 使用 Elasticsearch 检索器进行评估
es_topk_df = evaluate_retriever(es_retriever)
为了能够直观地看到随着k值的增加，召回率如何变化，下面来绘制曲线图：
import matplotlib.pyplot as plt
# 定义函数用于绘制检索器评估结果的折线图并保存
def plot_retriever_eval(dfs, retriever_names, save_path=None):
fig, ax = plt.subplots()
# 遍历每个数据框和检索器名称，绘制折线图
for df, retriever_name in zip(dfs, retriever_names):
df.plot(y="recall", ax=ax, label=retriever_name)
# 设置图表的标签和标题
plt.xticks(df.index)
plt.ylabel("Top-k Recall")
plt.xlabel("k")
plt.legend() # 显示图例
# 如果指定了保存路径，则保存图片
if save_path:
plt.savefig(save_path)
plt.show() # 显示图表
# 绘制 Elasticsearch 检索器的评估结果并保存为图片
save_path = "images/retriever_eval.png"
plot_retriever_eval([es_topk_df], ["BM25"], save_path=save_path)
print(f"Saved evaluation plot as '{save_path}'")
运行结果：
从图中可以看出，在k=5附近有一个明显的拐点，到k=10的时候，召回率基本不再增加。是否存在方法，能在更小的k值上取得相似的效果呢？目前使用的检索器是BM25，它使用稀疏向量，稀疏向量的问题在于如果文档和问题没有匹配的项就不会被召回。这时密集向量就有了用武之地，我们可以为问题和文档分别生成密集向量，来获取语义层面的表示。下面让我们看看如何使用密集向量技术来检索文档。
1.1 密集文档检索
V. Karpukhin et al., “Dense Passage Retrieval for Open-Domain Question Answering”（https://arxiv.org/abs/2004.04906），（2020）.
密集文档检索（Dense Passage Retrieve，DPR）主要解决开放域问答中的检索问题。从上一个例子可以看出，当k=10时，召回率已经非常接近于理想的状态，我们不禁会产生遐想，能否使用较小的k值得到近似的效果呢？因为这样做的一个最直观的收益是可以把更少的文档传给阅读器，减少处理步骤，从而让问答pipeline整体的时延降低。类似BM25这样的稀疏检索器有一个众所周知的局限性，如果查询内容中有评论内容不完全匹配的术语，则可能无法提取相关的文档。一种有望解决此问题的方案是使用密集嵌入来表示问题和文档，这种当前最前沿的技术叫作密集文档检索。DPR的主要思想是采用两个BERT模型作为问题和文本段的编码器。如图7-10所示，这些编码器将输入的文本映射为[CLS]词元的d维向量表示。
Haystack为DPR初始化检索器的方式与BM25相似。除了设置文档存储之外，我们还需要为问题和段落挑选BERT编码器。这些编码器是通过负样本（问题与不相关的文本段）和正样本（问题与相关的文本段）来进行训练的，目的是学习具有更高相似度的问题-段落对。这里的示例使用已经在NQ语料库上微调过的编码器：
from haystack.retriever.dense import DensePassageRetriever
# 导入密集检索器模块，并配置参数初始化一个 Dense Passage Retriever 实例
dpr_retriever = DensePassageRetriever(
document_store=document_store, # 指定文档存储
query_embedding_model="facebook/dpr-question_encoder-single-nq-base", # 查询编码模型
passage_embedding_model="facebook/dpr-ctx_encoder-single-nq-base", # 文档编码模型
embed_title=False # 是否嵌入标题
)
运行结果：
代码中设置了embed_title=False，因为在过滤了商品之后，简单连接文档的标题（即item id）并不会提供更多的信息。一旦将密集检索器初始化，下一个步骤就是遍历Elasticsearch所有索引文档，并使用编码器来更新嵌入，可以通过以下方式完成：
# 更新文档存储中的嵌入向量，使用指定的检索器
document_store.update_embeddings(retriever=dpr_retriever)
准备好上述步骤后，下面来用与BM25相同的方式来评估密集检索器，比较它们的查全率：
# 使用 Dense Passage Retriever (DPR) 进行检索器评估，并获取评估结果
dpr_topk_df = evaluate_retriever(dpr_retriever)
# 绘制检索器评估结果的图表，比较 BM25 和 DPR 的表现
plot_retriever_eval([es_topk_df, dpr_topk_df], ["BM25", "DPR"])
运行结果：
从图中可以看出，使用DPR并没有获得比BM25更高的召回率，在k=3的时候接近饱和。
使用Facebook的FAISS库（https://oreil.ly/1E8Z0）作为文档存储可以提升嵌入的相似度搜索性能。同样地，提升DPR检索器的性能也可以通过在目标域做微调操作来完成。如果想要了解怎样对DPR做微调操作，请查阅Haystack文档（https://oreil.ly/eXyro）。
到这里我们已经完成了对评估检索器的探究，下面来看看如何评估阅读器。
2、评估阅读器
在提取式问答场景中，主要有两个指标用于评估阅读器：
- 精准匹配（Exact Match，EM）
如果预测结果与事实答案中的字符完全匹配，则用EM=1表示，否则EM=0，如果不存在预期的答案，模型预测任何文本都用EM=0来表示。
- F1 分数（F1-score）
F1分数是统计学中用来度量二分类模型查准率的一种指标，同时兼顾了分类模型的查准率和召回率，是查准率和召回率的调和平均数，最大值是1，最小值是0。
下面我们从FARM中导入一些辅助函数来建立一个简单示例，观察这两个指标：
# 导入用于评估的函数
from farm.evaluation.squad_evaluation import compute_f1, compute_exact
# 设置预测和标签
pred = "about 6000 hours"
label = "6000 hours"
# 计算并打印精确匹配度（Exact Match）和 F1 分数
print(f"EM: {compute_exact(label, pred)}")
print(f"F1: {compute_f1(label, pred)}")
运行结果：
EM: 0 F1: 0.8
在函数底层实现中，我们首先通过删除标点符号、修复空格和将字母转化为小写等操作来将答案进行规范化，然后将规范化的字符串词元化处理成词袋（bag-of-word），最后在词元层面来计算指标。从上面例子可以看出，EM是一个比F1分数要严格很多的指标：多一个或少一个词元都会使EM=0，而F1分数可能无法捕捉到真正错误的答案。例如预测的答案片段是“about 6000 dollars”，将会得到以下结果：
# 设置预测和标签
pred = "about 6000 dollars"
# 计算并打印精确匹配度（Exact Match）和 F1 分数
print(f"EM: {compute_exact(label, pred)}")
print(f"F1: {compute_f1(label, pred)}")
运行结果：
EM: 0 F1: 0.4
所以，仅仅依靠F1分数来评估阅读器可能会产生一些误差，需要结合EM来做权衡。
一般来说，每个问题都可能会有多个合理的答案，因此会为数据集中每个用于评估的问答对分别计算这两个指标，然后选择最佳分数的答案。最后对每个问答对的EM和F1分数求平均数，得到的就是模型总体的EM和F1分数。
为了评估阅读器，下面我们将创建一个具有两个节点的新pipeline：一个阅读器节点，一个用于评估阅读器的节点。我们将会使用到EvalReader类，该类从阅读器获取预测结果并计算相应的EM和F1分数，为了与SQuAD评估结果进行比较，我们将使用存储在EvalAnswers中的top_1_em和top_1_f1指标为每个查询获取最佳答案：
# 导入评估模块
from haystack.eval import EvalAnswers
def evaluate_reader(reader):
score_keys = ['top_1_em', 'top_1_f1']
# 创建 EvalAnswers 实例
eval_reader = EvalAnswers(skip_incorrect_retrieval=False)
# 创建管道
pipe = Pipeline()
pipe.add_node(component=reader, name="QAReader", inputs=["Query"])
pipe.add_node(component=eval_reader, name="EvalReader", inputs=["QAReader"])
# 遍历所有标签
for l in labels_agg:
# 查询文档存储中的文档
doc = document_store.query(l.question, filters={"question_id":[l.origin]})
# 运行管道，评估阅读器性能
_ = pipe.run(query=l.question, documents=doc, labels=l)
# 返回评估指标字典
return {k:v for k,v in eval_reader.__dict__.items() if k in score_keys}
# 评估阅读器性能并存储结果
reader_eval = {}
reader_eval["Fine-tune on SQuAD"] = evaluate_reader(reader)
请注意，以上代码设置了skip_incorrect_retrieval=False，作用是可以确保检索器始终会将上下文传给阅读器（在SQuAD评估中也是一样）。现在我们已经通过阅读器运行了每个问题，下面来输出分数：
# 导入必要的库
import matplotlib.pyplot as plt
import pandas as pd
# 定义函数，用于绘制阅读器评估结果的条形图，并保存为指定文件名的图像文件
def plot_reader_eval(reader_eval, save_path=None):
# 创建图形和轴对象
fig, ax = plt.subplots()
# 将评估结果转换为数据框
df = pd.DataFrame.from_dict(reader_eval)
# 绘制条形图
df.plot(kind="bar", ylabel="Score", rot=0, ax=ax)
# 设置 x 轴标签
ax.set_xticklabels(["EM", "F1"])
# 显示图例
plt.legend(loc='upper left')
# 如果提供了保存路径，则保存图像
if save_path:
plt.savefig(save_path)
# 显示图形
plt.show()
# 绘制阅读器评估结果条形图，并保存为指定文件名的图像文件
plot_reader_eval(reader_eval, save_path="images/reader_eval_scores.png")
运行结果：
经过微调的模型在SubjQA上的效果似乎要比在SQuAD 2.0上的效果要差得多，其中MiniLM的EM和F1分数分别是76.1和79.5。造成性能下降的一个原因是用户对产品的真实评论和SQuAD的维基百科文章有很大的差异，而且用户评论的语法是偏口语化的。另一个原因可能是数据集语料具备固有的主观性，这些问答对与维基百科中的科普信息相比显得不着边际。下面我们来看看如何在数据集上对模型进行微调操作，来获得更好的领域自适应结果。
3、领域自适应
D. Yogatama et al., “Learning and Evaluating General Linguistic Intelligence”（https://arXiv.org/abs/1901.11373），（2019）.
尽管在SQuAD上微调的模型可以很好地泛化到其他领域，但对于SubjQA来说，模型的EM和F1分数要比SQuAD差得多。在其他提取式问答数据集中也存在这种泛化失效的情况，这有可能是因为Transformer模型在SQuAD上容易产生过拟合 。改善阅读器的最直接方法是在SubjQA训练集上进一步微调MiniLM模型。FARMReader中有一个专门为此设计的train()方法，输入数据采用SQuAD JSON格式，其中所有问答对按照商品维度分别组合在一起，如图所示。
SQuAD JSON是一个比较复杂的数据格式，需要借助一些方法和Pandas库来辅助操作。首先来创建一个函数，该函数可以创建与每个商品ID相关联的paragraphs数组，数组的每个元素都包含一个上下文（即评论）和一个问答对的qas数组，如下所示：
def create_paragraphs(df):
paragraphs = [] # 初始化一个空列表，用于存储段落
id2context = dict(zip(df["review_id"], df["context"])) # 创建一个字典，将评论ID映射到对应的上下文
# 遍历每个评论ID及其对应的上下文
for review_id, review in id2context.items():
qas = [] # 初始化一个空列表，用于存储问题-答案对
# 筛选特定上下文的所有问题-答案对
review_df = df.query(f"review_id == '{review_id}'")
id2question = dict(zip(review_df["id"], review_df["question"])) # 创建一个字典，将问题ID映射到问题
# 遍历每个问题ID及其对应的问题
for qid, question in id2question.items():
# 筛选单个问题ID
question_df = df.query(f"id == '{qid}'").to_dict(orient="list")
# 提取答案起始索引和答案文本
ans_start_idxs = question_df["answers.answer_start"][0].tolist()
ans_text = question_df["answers.text"][0].tolist()
# 确定问题是否可回答
if len(ans_start_idxs):
# 构造符合要求的答案格式
answers = [
{"text": text, "answer_start": answer_start}
for text, answer_start in zip(ans_text, ans_start_idxs)
]
is_impossible = False # 可回答的问题
else:
answers = [] # 没有找到答案
is_impossible = True # 问题不可回答
# 将问题-答案对添加到qas列表中
qas.append({
"question": question,
"id": qid,
"is_impossible": is_impossible,
"answers": answers
})
# 将上下文和问题-答案对添加到paragraphs列表中
paragraphs.append({
"qas": qas,
"context": review
})
return paragraphs # 返回包含问题-答案对和上下文的段落列表
有了paragraphs数组的创建方法之后，传入任意单个商品ID关联的DataFrame就可以获得SQuAD格式的数据：
# 从训练数据集中筛选出标题为'B00001P4ZH'的产品数据
product = dfs["train"].query("title == 'B00001P4ZH'")
# 调用create_paragraphs函数，处理选定产品的数据
create_paragraphs(product)
运行结果：
[{'qas': [{'question': 'How is the bass?',
'id': '2543d296da9766d8d17d040ecc781699',
'is_impossible': True,
'answers': []}],
'context': 'I have had Koss headphones ...',
'id': 'd476830bf9282e2b9033e2bb44bbb995',
'is_impossible': False,
'answers': [{'text': 'Bass is weak as expected', 'answer_start': 1302},
{'text': 'Bass is weak as expected, even with EQ adjusted up',
'answer_start': 1302}]}],
'context': 'To anyone who hasn\'t tried all ...'},
{'qas': [{'question': 'How is the bass?',
'id': '455575557886d6dfeea5aa19577e5de4',
'is_impossible': False,
'answers': [{'text': 'The only fault in the sound is the bass',
'answer_start': 650}]}],
'context': "I have had many sub-$100 headphones ..."}]
最后将此函数应用于每个被拆分的DataFrame中的每个商品ID，下面的convert_to_squad()函数将执行此操作，并将结果保存在electronics-{split}.json文件中：
import json
def convert_to_squad(dfs):
for split, df in dfs.items():
subjqa_data = {} # 初始化一个空字典，用于存储处理后的数据
# 对每个产品ID创建段落
groups = (df.groupby("title").apply(create_paragraphs)
.to_frame(name="paragraphs").reset_index())
# 将处理后的数据存储为字典列表，并存入subjqa_data中的"data"键
subjqa_data["data"] = groups.to_dict(orient="records")
# 将结果保存到磁盘上
with open(f"electronics-{split}.json", "w+", encoding="utf-8") as f:
json.dump(subjqa_data, f)
# 调用convert_to_squad函数，将数据集dfs转换为SQuAD格式并保存为JSON文件
convert_to_squad(dfs)
现在我们知道了如何正确地拆分这种格式的数据，下面通过指定训练和拆分的位置以及保存微调之后模型的位置，来对阅读器进行微调操作：
# 定义训练数据集和验证数据集的文件名
train_filename = "electronics-train.json"
dev_filename = "electronics-validation.json"
# 使用阅读器对象进行训练
reader.train(data_dir=".", # 数据目录，当前目录 "."
use_gpu=True, # 使用GPU加速（如果可用）
n_epochs=1, # 训练的轮数
batch_size=16, # 批量大小
train_filename=train_filename, # 训练数据集的JSON文件名
dev_filename=dev_filename) # 验证数据集的JSON文件名
对阅读器经过微调操作之后，下面来比较它与基线模型的性能差异：
# 对阅读器进行评估，并将评估结果存储在reader_eval字典中的"Fine-tune on SQuAD + SubjQA"键下
reader_eval["Fine-tune on SQuAD + SubjQA"] = evaluate_reader(reader)
# 绘制阅读器评估结果的图表
plot_reader_eval(reader_eval)
运行结果：
从结果可以看出，领域自适应使EM分数提升到原来的6倍！F1分数也提升到原来的1倍左右！可能有人会问，为什么不直接在SubjQA训练集上微调预训练的语言模型，原因之一是在SubjQA中只有1295个训练语料，而SQuAD有超过100 000个，我们很可能会遇到过拟合的问题。不过，可以看看如果这样做会产生什么结果，为了能进行公平的比较，我们将使用用于微调SQuAD基线相同的语言模型。和之前一样，使用FARMReader加载模型：
# 定义MiniLM模型的检查点路径
minilm_ckpt = "microsoft/MiniLM-L12-H384-uncased"
# 使用MiniLM模型初始化阅读器对象
minilm_reader = FARMReader(
model_name_or_path=minilm_ckpt, # MiniLM模型的检查点路径或模型名称
progress_bar=False, # 是否显示进度条
max_seq_len=max_seq_length, # 最大序列长度
doc_stride=doc_stride, # 文档跨度
return_no_answer=True # 是否返回无答案的情况
)
然后使用一轮（n epochs=1）来进行微调操作：
# 使用MiniLM阅读器对象进行训练
minilm_reader.train(
data_dir=".", # 数据目录，当前目录 "."
use_gpu=True, # 使用GPU加速（如果可用）
n_epochs=1, # 训练的轮数
batch_size=16, # 批量大小
train_filename=train_filename, # 训练数据集的JSON文件名
dev_filename=dev_filename # 验证数据集的JSON文件名
)
结合对测试集的评估结果，可以得到：
# 对MiniLM阅读器进行评估，并将评估结果存储在reader_eval字典中的"Fine-tune on SubjQA"键下
reader_eval["Fine-tune on SubjQA"] = evaluate_reader(minilm_reader)
# 绘制阅读器评估结果的图表
plot_reader_eval(reader_eval)
运行结果：
可以看出，直接在SubjQA上微调语言模型比在SQuAD+SubjQA上性能要差得多。
在处理小型数据集时，在评估Transformer模型时最好使用交叉校验，因为在小型数据集上非常容易过拟合。可以在FARM仓库（https://oreil.ly/K3nK8）中找到如何使用SQuAD格式的数据集执行交叉验证的例子。
4、评估整体问答pipeline
目前我们已经知道了如何评估检索器和阅读器，本节将它们整合到一起，评估pipeline的整体性能，要完成这一工作，只需要在检索器pipeline中添加阅读器节点。此前我们验证了检索器在k=10的时候能够获取不错的召回率，在加入阅读器后，最终还需要再微调这个值，以评估它对阅读器性能的影响（与此前SQuAD类似的评估方式相比，现在的情况是每次查询操作，阅读器都会接收多个上下文）：
# 初始化检索器流水线，使用ES检索器
pipe = EvalRetrieverPipeline(es_retriever)
# 添加阅读器节点
eval_reader = EvalAnswers()
pipe.pipeline.add_node(component=reader, name="QAReader", inputs=["EvalRetriever"])
# 添加评估阅读器节点
pipe.pipeline.add_node(component=eval_reader, name="EvalReader", inputs=["QAReader"])
# 执行流水线
run_pipeline(pipe)
# 从阅读器中提取指标
reader_eval["QA Pipeline (top-1)"] = {
k: v for k, v in eval_reader.__dict__.items()
if k in ["top_1_em", "top_1_f1"]
}
我们通过比较模型的top 1的EM和F1分数，来预测图7-12中检索器返回的文档中的答案。
# 绘制阅读器评估结果的图表，比较Fine-tune on SQuAD + SubjQA和QA Pipeline (top-1)的指标
plot_reader_eval({
"Reader": reader_eval["Fine-tune on SQuAD + SubjQA"], # 使用Fine-tune on SQuAD + SubjQA的阅读器的评估结果
"QA pipeline (top-1)": reader_eval["QA Pipeline (top-1)"] # 使用QA Pipeline (top-1)的评估结果
})
运行结果：
从图中可以看出检索器对整个pipeline性能的影响，因为多了检索器，整体问答pipeline性能比单独评估阅读器要低一些，这主要是因为检索器会限制查到的文档数，这个问题可以通过调整检索器查到的文档数来解决。
到目前为止，我们只从文档中提取了答案片段，在实际情况中，答案很可能零散地分布在整个文档中，我们当然希望模型能将这些片段综合成一个连贯的答案。下面我们看看生成式问答如何解决这个问题。
四、生成式问答
P. Lewis et al., “Retrieval-Augmented Generation for Knowledge-IntensiveNLPTasks”（https://arxiv.org/abs/2005.11401），（2020）.
有一种十分有趣的方案可以替代将答案提取为文本片段这种方案。该方案使用预训练语言模型来直接生成答案，通常被称为抽象式问答（Abstractive QA）或生成式问答（Generative QA），并且有能力生成具备更好措辞的答案，从而解决答案分散的问题。虽然目前这种技术方案还不如提取式问答成熟，但它发展的速度很快，很可能目前已经有工业级应用案例了。本节将会介绍目前最前沿的技术：检索增强生成（Retrieval-Augmented Generation，RAG）。
RAG扩展了本章中介绍的经典检索器-阅读器架构，将阅读器替换为生成器，并使用DPR作为检索器来搭配使用。生成器是一个预训练的序列到序列（常用Seq2Seq称呼）的Transformer模型，类似T5和BART模型，首先从DPR获取文档的隐式向量，然后根据查询内容和文档内容迭代生成答案。由于DPR和生成器是可区分的，因此整个过程也可以进行端到端的微调处理，如下图所示。
下面我们来看看RAG的一个简单示例，需要用到之前使用的DPRetriever，另外再实例化一个生成器，RAG有两种类型可供选择：
- RAG-Sequence
使用被检索出的文档来生成完整的答案，检索器中前k个被传到生成器的文档，会被当成额外的文本信息，通过边缘化的方式融合到生成器，最后为每个文档输出序列，以获取最佳答案。简单来说，就是使用单一文档生成整个序列，然后使用检索器对该文档的检索概率加权求和。
- RAG-Token
使用不同文档来生成答案中的每个词元，每个词元最终的概率分布是检索器对该文档的检索概率乘基于该文档生成词元的概率，最后把所有词元的概率相乘得到最终序列的概率。
由于RAG-Token模型往往比RAG-Sequence模型的效果更好，我们将使用在NQ上微调的模型作为生成器。在Haystack中，实例化生成器类似于实例化阅读器，但不会涉及滑动窗口的设置，比如不用设置max_seq_length和doc_stride参数，而是指定控制文本生成的超参数：
# 导入 RAGenerator 类从 Haystack 的 transformers 模块中
from haystack.generator.transformers import RAGenerator
# 初始化 RAGenerator 对象，使用 "facebook/rag-token-nq" 模型
generator = RAGenerator(
model_name_or_path="facebook/rag-token-nq", # RAGenerator 使用的模型的名称或路径
embed_title=False, # 是否嵌入标题（通常用于生成模型）
num_beams=5 # 每个输入的束搜索数量
)
我们使用num_beams参数指定了在束搜索中的束数（，就像在DPR检索器所做的那样，不需要引入文档标题，语料库是按商品ID过滤的。
接下来要做的是使用Haystack的GenerativeQAPipeline将检索器和生成器组合在一起：
# 导入 GenerativeQAPipeline 类从 Haystack 的 pipeline 模块中
from haystack.pipeline import GenerativeQAPipeline
# 初始化 GenerativeQAPipeline 对象，使用指定的生成器和检索器
pipe = GenerativeQAPipeline(
generator=generator, # 生成器对象，用于生成答案
retriever=dpr_retriever # 检索器对象，用于检索相关文档
)
在RAG中，查询编码器和生成器都是经过端到端训练的，而上下文编码器并不会启用。在Haystack中，Generative QAPipeline使用RAGenerator的查询编码器和DensePassageRetriever的上下文编码器。
下面试试给RAG输入一些之前的Amazon Fire平板电脑的查询信息，来看看效果。为了简化查询，这里编写了一个简单的函数来获取查询结果并输出最佳答案：
def generate_answers(query, top_k_generator=3):
# 运行生成问答流水线，使用指定的查询、生成器数量和检索器数量，并过滤特定的项目ID
preds = pipe.run(
query=query, # 查询问题文本
top_k_generator=top_k_generator, # 生成器返回的前k个答案
top_k_retriever=5, # 检索器返回的前k个相关文档
filters={"item_id": ["B0074BW614"]} # 过滤器，筛选具有特定item_id的文档
)
# 打印问题
print(f"Question: {preds['query']} \n")
# 打印生成的答案
for idx in range(top_k_generator):
print(f"Answer {idx+1}: {preds['answers'][idx]['answer']}")
我们来使用此函数做个测试：
# 调用生成答案函数，传入指定的查询文本
generate_answers(query)
运行结果：
Question: Is it good for reading? Answer 1: the screen is absolutely beautiful Answer 2: the Screen is absolutely beautiful Answer 3: Kindle fire
这些结果看起来并不是我们想要的，但也不是那么糟糕，这种主观性问题确实会使生成器“感到困惑”，我们换一种事实陈述性问题来试试：
# 调用生成答案函数，传入特定的查询文本
generate_answers("What is the main drawback?")
运行结果：
Question: What is the main drawback? Answer 1: the price Answer 2: no flash support Answer 3: the cost
可以看出，事实陈述性的问题获得了不错的答案。为了获取更好的结果，我们可以在SubjQA上对RAG进行端到端的微调，这可以作为本书的一个实践练习题，如果你有兴趣，Hugging Face Transformers库（https://oreil.ly/oZz4S）中的脚本可以帮助到你。
五、整理
以上讲解了问答系统的两种技术（提取式和生成式），研究了两种不同的检索算法（BM25和DPR）。在这里，我们看到，虽然领域自适应是一种简单的技术，但它却可以显著提升问答系统的性能，随后我们研究了一些用于评估此类系统的常见指标。尽管这里关注的是封闭域问答（电子产品的单个域），但提到的技术可以很容易地推广到开放域问答案例。建议读者阅读Cloudera的Fast Forward问答系列（https://oreil.ly/Fd6lc）文章。
从头开始构建问答系统是一项有挑战的工作，有个可取的经验是首先为终端用户提供搜索功能，然后才是其他组件的部署，比如提取组件。除了按需为用户响应查询之外，阅读器还可以提供一些比较新颖的功能。例如，Grid Dynamics（https://oreil.ly/CGLh1）的研究人员能够使用他们的阅读器自动提取用户目录中每种商品的一组优缺点。另外，还可以通过构造诸如“这是什么类型的相机？”之类的查询问题，以零样本的方式提取命名实体。考虑到其还在学术初期，我们建议在其他两种方式不能很好地解决问题之后才开始研究生成式问答。借用马斯洛需求层次理论，问答领域的需求层次可以用下图表示。
展望一下未来，一个令人振奋的研究领域是多模态问答（Multimodal QA），它涉及文本、表格和图像等多种模态。如MultimodalQA的基准测试所述，这样的问答系统可以回答更加复杂的问题，整合不同模态的信息，比如“When was the famous painting with two touching fingers completed?”（那幅两只手指触碰的名画是什么时候完成的？）这样的问题。另一个具有实际业务价值的应用场景是知识图谱的问答，图谱节点对应于现实世界的实体，他们之间的关系由边（edge）来定义。通过将仿真的描述编码为（主语、谓语、宾语）三元组，可以使用图谱来回答有关缺失元素的问题。关于将Transformer和知识图谱结合的示例，请参阅Haystack教程（https://oreil.ly/n7lZb）。一个有前景的研究方向是自动问题生成（Automatic Question Generation），它是一种使用未标注数据或使用数据增强的无监督/弱监督训练方法，最近有两篇关于它的论文，一篇关于PAQ的基准测试（Probably Answered Questions benchmark），另一篇关于跨语言环境的合成数据增强。
