前面两篇博客主要探究了QG（Question Generation）任务的基本策略、评价指标；描述了我的初步探索：
- 博客1：基本策略
- 博客2：评价指标、初步探索
- 博客3：训练数据和预测数据预处理的部分
- 博客4：模型的构建
- 博客5：生成“问答对”
- 博客6：问答对有效性过滤（基于文本分类任务）
- 博客7：用问题库检索服务
一、问题背景和目标
问题生成（Question Generation）是指根据文段内容和期望答案约束自动生成相关问题。本课题旨在针对给定教材文本，生成捕获教材的核心内容的不同难度和不同形式的自然语言问题库。
二、数据预处理
2.1 中医文献阅读理解数据集
模型的基本思想来源于“‘万创杯’中医药天池大数据竞赛”问题生成赛题的第一名的解决方案，我在其基础上进行了扩展，并应用在了自己的训练和测试数据集上。
项目源码：
https://github.com/kangyishuai/CHINESE-MEDICINE-QUESTION-GENERATION
数据来源于“‘万创杯’中医药天池大数据竞赛”在中医药领域的阅读理解数据集。分为训练数据、测试数据。
数据内容包含一段文本、相关的“问题-答案”对，其中答案取自文本截取的一些片段。
篇章文本长度在100以下的数据较少，长度区间400-500的数据占比较大。可以发现，篇章文本最大，其次是答案文本，最后是问题文本。
答案是从篇章中截取的，可以适当截取短一点；篇章在硬件资源允许的范围内，可以尽量截取长一点。
2.2 我的训练数据集：中医文献+CJRC+Squad多数据集结合训练
在模型测试的初期，我采用“‘万创杯’中医药天池大数据竞赛”第一名的解决方案所用的数据集进行训练，将训练的模型放在“经济法教辅语料”上运行的时候就会出现很多不通顺的问题。
经过思考，后期考虑到由于仅用中医文献作为训练数据具有领域的局限性，上述训练过程在“经济法教辅语料”的数据集上表现并不优秀。快速提升模型在多种领域问题上的表现性能的一个方法，是将“多领域的数据”融合，打乱顺序同时放入模型训练。
我将选择采用前面博客提到的中文数据集（这里是：中医文献+CJRC+Squad），融合训练模型。
- 其中“中医文献数据集”上面已经介绍。
- Squad数据集我采取了一个中文的版本。
- CJRC是一个“中文法律阅读理解数据集”，该数据集包含约10,000篇文档，主要涉及民事一审判决书和刑事一审判决书，数据来源于中国裁判文书网。
通过抽取裁判文书的事实描述内容（“经审理查明”或者“原告诉称”部分），针对事实描述内容标注问题，最终形成约50,000个问答对。
该数据集涉及多种问题类型，包括：- 1.片段抽取型问题（Span-Extraction）
- 2.是否类问题（YES/NO）
- 3.拒答类问题（Unanswerable）
数据处理的思想结合了我的目标以及原始项目中的思想，总结如下：
-
首先是各种数据集数据格式不统一，我想将其融合就必须整理成相同的格式，并且需要将整合的数据乱序处理。根据训练的规模，适量减少数据的量，进行基本的筛选。
-
其次要按照原始项目给出的解决方案，做一些清洗和截断操作，主要包含了两个部分：
- 1.数据清洗：剔除空白字符、剔除带括号的英文；处理部分不匹配数据（片段抽取型问题数据应该是匹配的，这样“是否类”问题会被过滤掉）
- 2.文本截断：一般认为答案所在的位置上下问处和问题相关性最强。故篇章取答案所在位置的开头前64个字符和结尾后64个字符；答案取前64个字符；问题取前131个字符。
数据融合的代码：
import json
import random
if __name__ == "__main__":
data = None
data = json.load(open('./data/big_train_data.json', 'r', encoding='utf-8'))['data']
count = 0
res = []
for p in data:
paragraphs = p['paragraphs']
for paragraph in paragraphs:
qas = paragraph['qas']
l = []
for qa in qas:
if qa['is_impossible']:
q = qa['question']
a = qa['answers']
for a_ in a:
a_ = a_['text']
pair = {"Q": q, "A": a_}
l.append(pair)
dl = {"id": count,"text": paragraph['context'], "annotations": l}
res.append(dl)
count += 1
if count > 100000:
break
# 合并
others = json.load(open('./round1_train_0907.json', 'r', encoding='utf-8'))
for i, other in enumerate(others):
other['id'] = count
res.append(other)
count += 1
others = json.load(open('./train-v2.0-zh.json', 'r', encoding='utf-8'))['data']
for other in others:
ps = other['paragraphs']
for p in ps:
text = p['context']
qas = p['qas']
l = []
for qa in qas:
q = qa['question']
a_s = qa['answers']
for a_ in a_s:
a = a_['text']
pair = {"Q": q, "A": a}
l.append(pair)
ll = {"id": count, "text": text, "annotations": l}
count += 1
res.append(ll)
random.shuffle(res)
json.dump(res, open('train.json', 'w'), ensure_ascii=False, indent=2)
清洗和截断操作代码位于utils.py文件中
def preprocess(df):
"""数据预处理。"""
# 剔除空白字符
df = df.applymap(lambda x: re.sub(r'\s', '', x))
df = df.applymap(lambda x: re.sub(r'\\n', '', x))
# 剔除带括号的英文
func = lambda m: '' if len(m.group(0)) > 5 else m.group(0)
df = df.applymap(lambda x: re.sub(r'\([A-Za-z]+\)', func, x))
df = df.applymap(lambda x: re.sub(r'（[A-Za-z]+）', func, x))
# 筛选出答案与篇章不匹配的数据
tmp = list()
for idx, row in df.iterrows():
if row['answer'] not in row['passage']:
tmp.append(idx)
# 处理部分不匹配数据
no_match = df.loc[tmp]
df.drop(index=tmp, inplace=True)
no_match['answer'] = no_match['answer'].map(lambda x: x.replace('.', ''))
df = pd.concat([df, no_match])
df.reset_index(drop=True, inplace=True)
return df
2.3 教材数据集上的NER任务（用于预抽取答案）
我的目标是对一本给定的教材，输出一系列的提问。
我选取“经济法教辅语料”，教材主要描述了相关的经济法学的论述，可以从中找到很多法律相关的描述性的文字，很适合用于生成问题。
其教材内容已经处理成了json格式，如下图所示：
对其分割成3000余篇章，并从中预生成一系列答案。答案主要包含两种类型：
- 1.文本中句子片段；
- 2.文本中抽取的命名实体（NER）。
在完成句子划分时，我先将句子中可能出现的英文符号，替换为对应的中文字符，随后按照句号、逗号、分号等进行短句划分。
我利用NER技术抽取文本中的实体，包括数字、日期、专用名等。
下面代码从“经济法教辅语料”中抽取出相关的片段和答案，组织成字典格式，方便最终用于预测：
第一步
- 输入：
- 文本 : text
- 答案类型:[“all”, “sentences”, “multiple_choice”]
- 其中"sentences"类型表示提取单个子句类型的答案，用逗号等符号隔开
- "multiple_choice"类型表示提取多个命名实体类型的答案，这里会用到NER任务
- "all"表示同时提取"sentences"类型和"multiple_choice"类型的答案。
- 返回：
- answer的list : “”
import re
from typing import Any, List, Mapping, Tuple
import json
import csv
import random
import spacy
from spacy.lang.zh.examples import sentences
def generate_qg_inputs(text: str, answer_style: str, spacy_nlp=None) -> Tuple[List[str], List[str]]:
"""
输入：
文本 : text
答案类型:["all", "sentences", "multiple_choice"]
返回：
answer的list : "<answer text>"
"""
# 抽取的答案的类型
VALID_ANSWER_STYLES = ["all", "sentences", "multiple_choice"]
if answer_style not in VALID_ANSWER_STYLES:
raise ValueError(
"Invalid answer style {}. Please choose from {}".format(
answer_style, VALID_ANSWER_STYLES
)
)
answers = []
if answer_style == "sentences" or answer_style == "all":
segments = text.split("\n")
for segment in segments:
sentences = _split_text(segment)
answers.extend(sentences)
if answer_style == "multiple_choice" or answer_style == "all":
sentences = _split_text(text)
prepped_answers = _prepare_qg_inputs_MC(sentences, spacy_nlp=spacy_nlp)
print(prepped_answers)
answers.extend(prepped_answers)
return list(set(answers))
将文本分割成单个子句，其中只取长度大于5的句子：完成"sentences"类型的答案的提取。
def _split_text(text: str) -> List[str]:
"""
将文本截断成短的句子
"""
MAX_SENTENCE_LEN = 128
sentences = re.findall(".*?[。！？.!\?]", text)
cut_sentences = []
for sentence in sentences:
if len(sentence) > MAX_SENTENCE_LEN:
cut_sentences.extend(re.split("[，；：,;:)]", sentence))
# remove useless post-quote sentence fragments
cut_sentences = [s for s in cut_sentences if len(s) > 5]
sentences = cut_sentences
# sentences = sentences + cut_sentences
return list(set([s.strip(" ") for s in sentences]))
利用NER提取命名实体，完成："multiple_choice"类型的答案的提取。
def _prepare_qg_inputs_MC(sentences: List[str], spacy_nlp=None) -> Tuple[List[str], List[str]]:
"""
在文本上进行 NER 操作
将提取的实体作为 multiple-choice 的候选答案
"""
if spacy_nlp == None:
spacy_nlp = spacy.load("zh_core_web_sm")
docs = list(spacy_nlp.pipe(sentences, disable=["parser"]))
answers_from_text = []
for doc in docs:
entities = doc.ents
if entities:
for entity in entities:
answers = entity.text
answers_from_text.append(answers)
return answers_from_text
从教材json文件中读取数据，并根据上述的函数提供的服务完成“答案-上下文”对的字典输出，用于模型训练结束后，在上面预测答案对应的问题。
def get_datalis(json_dir):
with open(json_dir, 'r') as fp:
dic_R3 = json.load(fp)
data_lis = []
id_1 = 0
csc_count_num = 0
spacy_nlp = spacy.load("zh_core_web_sm")
for k1 in dic_R3:
chap_title = dic_R3[k1]['chap_title']
chap_content = dic_R3[k1]['chap_content']
id_1 += 1
id_2 = 0
for k2 in chap_content:
sec_title = chap_content[k2]['sec_title']
sec_content = chap_content[k2]['sec_content']
if sec_title == '':
sec_title = None
else:
id_2 += 1
id_3 = 0
for k3 in sec_content:
subsec_title = sec_content[k3]['subsec_title']
subsec_content = sec_content[k3]['subsec_content']
if subsec_title == '':
subsec_title = None
else:
id_3 += 1
if sec_title == '' and subsec_title == '' or subsec_content.strip() == '':
continue
csc_list = []
subsec_content.replace(';', '；').replace(':', '：').replace('?', '？').replace('!', '！').replace('.', '。').replace(',', '，')
answer_style = 'all'
answers = generate_qg_inputs(text=subsec_content, answer_style=answer_style, spacy_nlp=spacy_nlp)
for answer in answers:
dataline = {"Q": "", "A": answer}
csc_list.append(dataline)
csc_ph = {"id": csc_count_num, "text": subsec_content.replace('\n', '').replace(' ', '').replace('\r\n', ''), "annotations": csc_list}
csc_count_num += 1
data_lis.append(csc_ph)
return data_lis
使用方法：
if __name__ == "__main__":
data_lis = get_datalis('../textbook/dic_R3.json')
json.dump(data_lis, open('../data/data_sentence.json', 'w'), ensure_ascii=False, indent=2)
print('文件写入完成……')
运行程序，输出的json文件内容如下（局部）：
