基于知识图谱的古诗词问答系统（全网首份 + 包复现 + 实践篇）
中国古典诗词具有独特的艺术表现形式，在人们的日常生活中架起了情感共鸣的桥梁、充当了教育和启蒙的工具，其中很多古诗词蕴含着民族正气和家国情怀，诠释了我们伟大的民族精神。古诗词在教育、艺术以及情感表达等多个方面有着举足轻重的作用，对于传承和弘扬中华优秀传统文化具有重要的意义。
目录
注：
① 基于知识图谱的古诗词问答系统，包复现，建议收藏。
② 若还未查看该项目的综述篇，请跳转查看：基于知识图谱的古诗词问答系统（全网首篇+包复现）
③ 若还未查看该项目的理论篇，请跳转查看：基于知识图谱的古诗词问答系统（全网首篇+包复现+理论篇）
一、前言
中国古典诗词具有独特的艺术表现形式，在人们的日常生活中架起了情感共鸣的桥梁、充当了教育和启蒙的工具，其中很多古诗词蕴含着民族正气和家国情怀，诠释了我们伟大的民族精神。古诗词在教育、艺术以及情感表达等多个方面有着举足轻重的作用，对于传承和弘扬中华优秀传统文化具有重要的意义。
随着大数据、云计算、AI等新兴科学技术的发展，人们的生活方式大大地改变了，在古诗词学习领域体现为，人们以往学习、欣赏、评鉴古诗词的媒介大多是语文教科书、诗词合集、报纸评论等纸质书籍，而目前人们的方式在很大程度上已经从纸质转向电子，诗词信息的来源可以是网页、小程序或者App，例如微信读书、古文岛、西窗烛等。
但不容乐观的是，在如今大数据时代，人们很难在如此快节奏的生活方式里静下心来从纷繁复杂的信息中挑选符合自己意图的答案。无论是传统的通用搜索引擎还是垂直搜索引擎，它们都是基于关键词的搜索，并给用户返回根据相似度、网页评分等技术排序后的网页，用户再从这些网页中挑选答案，这样一来二去，极大地浪费了用户寻找答案的时间。
由于此种局限，问答系统应运而生。基于知识图谱的古诗词问答系统，力求为广大古诗词爱好者提供更加高效、便捷的古诗词领域问答反馈。古诗词问答系统的实现模式是基于端到端的，用户只需输入与古诗词相关问题，系统就会返回对应的“精准”答案。例如，当用户输入问句“小诗，你好，请问著名诗人李白诞生于哪个朝代呢？”，系统则会返回以“唐代”为关键词进行话术包装后的答案。
本篇是项目——基于知识图谱的古诗词问答系统的实践篇，主要从源代码的层面讲解整个系统的实现逻辑。
在本文以下章节中，涉及古诗词原始数据集、所有模型的训练数据集、训练得到的模型、知识图谱的关系和节点以及整个知识图谱，所有提到的相关文件可私信作者获取。本项目在Github上的地址为：基于知识图谱的古诗词问答系统。
二、系统环境配置
2.1 集成开发环境（IDE）
针对本项目的开发，主要在JetBrains PyCharm平台完成，使用的是其专业版本。专业版本的获取方式主要是通过购买或者完成学生认证，当然通过关注微信公众号（火星软件安装🉑🉑🉑）也能够获取到专业版本，本文选取的是PyCharm 2024.1 的版本，安装过程按照其指引完成即可不必赘述。
2.2 Anaconda的安装
Anaconda，是一个开源并专注于数据分析的python发行版本，包含很多科学包及其依赖项。Anaconda的出现为python项目各个包的安装和管理提供了很大的便捷，很多科学计算类的库都包含在里面了，使得安装比常规python安装要容易，具有开源、社区支持、高性能等优点。
安装的版本尽量选取近年稳定的，而不必一定选取最新版本，如果还没有使用过的小伙伴们，赶快点击博文——史上最全最详细的Anaconda安装教程按照其步骤下载安装就好了。
2.3 JDK与Neo4j的安装
本系统依赖的知识图谱会存储在Neo4j中，Neo4j是一款优秀的NoSQL型图数据库，有社区版和企业版，安装社区版就能够满足个人开发中的绝大多数需求了，比如社区版中可存储图的节点数量在亿级，可存储的关系数量同样。安装Neo4j之前还需要安装JDK，因为Neo4j的部分功能依赖Java而实现。
本文使用的JDK版本是17.0.9，使用社区版本的Neo4j的版本是5.15.0；安装教程按照博文——Neo4j安装+安装JDK（Windows超详细）里的步骤一步一步安装就好，若需要本文实现的相关软件安装包，请联系作者获取。
2.4 MySQL 与 Navicat Premium 16 的安装
MySQL数据库在本系统的作用主要在于存储系统注册与登录以及后续用户问答的部分数据，其获取和安装都很简单，这篇博文——mysql数据库安装（详细）非常全面地介绍了其安装的步骤，按照指引安装即可。
使用MySQL常会通过图形化界面操作，这篇博文——MySQL和Navicat下载、安装及使用详细教程介绍了Navicat Premium 的安装，按照步骤安装就好。
三、问答系统源码总览
下图是整个项目的文件树，由于文件太多，只对重点文件夹或文件加了注释。项目中的源文件都有详细的注释，应该很好理解，这里仅作概要总览。
四、 系统处理模块
在本项目理论篇中，阐述了该模块的功能，这里就不再赘述，下面各个模块均如此仅从代码的角度详解如何实现。
针对Web型的系统开发，使用已有的框架可以减少开发部分功能的时间，在完成本系统的用户注册和登录功能时，Flask微框架提供了很多快捷高效的技术支持。在MVC（Model-View-Controller，模型－视图－控制器）框架中，程序被分为三个组件：数据处理（Model）、用户界面（View）、交互逻辑（Controller）。
本系统的开发，基于Flask的MVC架构，在Model中完成系统注册与登录的数据设计和验证，在View中完成系统数据接口以及其他非功能需求项，而在Controller中完成前后端的整体交互。
4.1 系统注册与登录
由于笔者在开发该项目之间，对Web前后端开发了解甚少，所以特地花了点时间在B站学习了Flask的开发，地址——2024版-零基础玩转Python Flask框架-学完可就业，本系统的注册与登录功能就是按照其教学完成的。
用户注册的数据存储在MySQL数据库中，而Flask在与数据库交互时采用的是ORM模型，首先在python中定义用户数据库的各类表，然后直接迁移到MySQL中即可。下面是用户注册时提交数据的用户模型，对应到MySQL中的“user”表单。
# 用户模型
class UserModel(db.Model):
__tablename__ = "user"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(100), nullable=False)
password = db.Column(db.String(300), nullable=False)
email = db.Column(db.String(100), nullable=False, unique=True)
join_time = db.Column(db.DateTime, default=datetime.now) # 需要函数而不是值
模型定义完成后，还需定义用户注册的视图函数，如下所示。
# 3 - 用户注册视图
@bp.route("/register", methods=["GET", "POST"])
def register():
if request.method == 'GET':
return render_template("register.html")
else:
# 验证表单 验证用户提交的邮箱和验证码是否对应且正确
form = RegisterForm(request.form) # 自动调用该类的验证方法
if form.validate():
email = form.email.data
username = form.username.data
password = form.password.data
user = UserModel(email=email, username=username, password=generate_password_hash(password))
db.session.add(user)
db.session.commit()
return redirect(url_for("login"))
else:
error = list(form.errors.values())[0] # 字典 转 列表
flash(error[0]) # 前端显示错误信息
return redirect(url_for("register"))
值得注意的是，在用户注册信息的时候，需要保证用户输入的数据与定义的数据类型一致（邮箱格式是否正确、密码是否符合要求等等），所以这时候就需要使用表单验证技术，Flask已经封装了该技术，只需调用即可，例如下面代码给出了注册时的表单验证。
class RegisterForm(wtforms.Form):
email = wtforms.StringField(validators=[Email(message="邮箱格式错误！")])
captcha = wtforms.StringField(validators=[Length(min=4, max=6, message="验证码格式错误！")])
username = wtforms.StringField(validators=[Length(min=3, max=20, message="用户名格式错误！")])
password = wtforms.StringField(validators=[Length(min=6, max=20, message="密码格式错误！")])
password_confirm = wtforms.StringField(validators=[EqualTo("password", message="两次密码不一致！")])
# 自定义验证 1.邮箱是否已被注册，2.验证码是否正确
def validate_email(self, field):
email = field.data
user = UserModel.query.filter_by(email=email).first()
if user:
raise wtforms.ValidationError(message="该邮箱已经被注册！")
def validate_captcha(self, field):
captcha = field.data
email = self.email.data
captcha_model = EmailCaptchaModel.query.filter_by(email=email, captcha=captcha).first()
if not captcha_model:
raise wtforms.ValidationError(message="邮箱或验证码错误！")
五、古诗词知识图谱模块
在构建古诗词知识图谱时，会在古诗词原始数据获取、数据处理以及古诗词知识抽取和导入花费大量的时间，接下来从这三个方面讲解代码实现。
5.1 数据获取与处理
从系统架构图中可以得知，原始数据的来源一是开源数据集，二是网络爬虫。由于开源数据集的数据都是结构化的，采用字典的形式存储，在处理上较为简单，只是额外需要注意，部分数据是繁体的，在“不必严谨”的情况下可以采用代码将其转换为简体。
从古诗文网的结构分析得知，虽然各个板块的内容有规律可循，但是在爬取数据的时候仍有很多特殊的地方需要通过特殊处理，所以并没有采用Scrapy框架而是手写爬虫脚本完成爬取工作。在提升网络爬虫的速度和效率时，主要使用多进程爬取而没有使用分布式爬虫。
下面以爬取作者、作者的名句、作者的作品为例，详细讲解爬取流程。其代码路径为：/KGQA_Poetry/data_processing/write_spider.py
首先分析网页结构，从下图可以看出，作者栏下对应了网站搜集的所有作者，并按照朝代划分。例如点击先秦，下面就展现了先秦时的所有作者，而每位作者都以锚文本呈现，即每个作者名字都会对应一个超链接。
通过以上分析，爬取的思路已经清晰了——采用广度优先抓取策略，首先从“作者”板块入手，爬取12个朝代对应的链接，然后再针对每一个朝代获取该朝代的所有作者对应的链接。
在爬取的过程中，每完成一项工作就可以将数据存储在磁盘或数据库，防止重复启动爬虫。接下来就会爬取各个诗词作家的作品和名句，下图给出了“诗人李白”的网页布局，点击①和②处的链接可以获取到他的的作品和名句，但由于网页受限，只能够爬取前10页的内容，其余内容可以通过App抓包技术获取，而第③部分的内容，可以作为实体、属性以及关系抽取的语料。
值得注意的是，针对网页里的动态加载的内容，需要采用其它技术爬取，例如使用selenium 自动化测试技术爬取这些内容。
下面是爬取作者的作品和名句的相关代码。
# C - 1 爬取数据
class Spider:
def __init__(self):
self.domain = "http://so.gushiwen.cn"
self.headers = {'user-agent': UserAgent().random}
self.type_name = '' # 使用进程爬虫时，得到相应的名字
self.work_types = []
# 1、解析网页
def parser_url(self, url):
response = requests.get(url, headers=self.headers)
html_text = response.content.decode()
html = etree.HTML(html_text, etree.HTMLParser())
response.close() # 注意关闭response
time.sleep(1) # 自定义
return html # 返回解析树
# 2、更新网站数据 - 各朝代和各个作者对应的链接
def update_data(self):
"""
start_url = "https://so.gushiwen.cn/authors" # 爬取的起始链接
response = requests.get(start_url, headers=self.headers)
html_text = response.content.decode()
# 获取朝代链接
path = '../txts/authors_html.txt'
with open(path, 'w', encoding='utf-8') as f:
f.write(html_text)
f.close()
"""
path = '../txts/authors_html.txt'
with open(path, 'r', encoding='utf-8') as f:
html_text = f.read()
f.close()
html = etree.HTML(html_text, etree.HTMLParser())
dynasty_urls = html.xpath('//div[@class="sright"]/a')
path = '../jsons/dynasty_authors.json' # 解析网页获取各朝代的链接
dynasty_links = {}
with open(path, 'w', encoding='utf-8') as f:
for url in dynasty_urls: # 遍历各个朝代的作者
title = url.xpath('./text()')[0] # xpath 对象 - list
link = url.xpath('./@href')[0]
print(title, ' ', link)
dynasty_links[title] = self.domain + link # 标题加链接
json.dump(dynasty_links, f) # 按 json 格式写入文件
f.close()
# 获取各个朝代 作者 的链接
with open(path, 'r', encoding='utf-8') as f:
dynasty_links = json.load(f)
f.close()
path = '../jsons/authors_name.json'
author_links = {}
with open(path, 'w', encoding='utf-8') as f:
for key, url in dynasty_links.items():
# print(key, url)
html = self.parser_url(url)
author_urls = html.xpath('//div[@class="typecont"]//span/a')
# print(author_urls)
for href in author_urls: # 遍历各个的作者
title = href.xpath('./text()')[0] # xpath 对象 - list
link = href.xpath('./@href')[0]
print(title, ':', link)
author_links[title] = self.domain + link # 标题加链接
json.dump(author_links, f) # 按 json 格式写入文件
f.close()
path = '../txts/authors_name.txt'
with open(path, 'w', encoding='utf-8') as f:
for key, value in author_links.items():
f.write(key + ': ' + value + '\n')
f.close()
# 3、更新网站数据 - 所有诗文和名句的链接
def update_author_works(self):
path = '../jsons/authors_name.json' # 读取文件
with open(path, 'r', encoding='utf-8') as f:
author_links = json.load(f) # 字典
f.close()
authors_works = dict.fromkeys(author_links.keys(), None) # 字典的键为 作者名字
authors_famous_sentences = dict.fromkeys(author_links.keys(), None)
for author, url in author_links.items(): # 处理作者
try:
print(author, ": ", url)
html = self.parser_url(url) # 解析网页
if html is None: # 访问失败
print(author + ': ' + url + '--------注意：链接失效，访问失败! 请更新数据！！！-----\n')
continue
# 解析诗文和名句，得到作者的诗文和名句的链接
div_urls = html.xpath('//div[@class="main3"]/div[@class="left"]'
'/div[@class="sonspic"]/div[@class="cont"]/p/a/@href') # 根据div分组
if not div_urls: # 没有诗文和名句
print(author, ": ", url, '注意：没有诗文和名句...')
continue
elif len(div_urls) == 2: # 既有诗文又有名句
authors_works[author] = self.domain + div_urls[0]
authors_famous_sentences[author] = self.domain + div_urls[1]
else: # 只有诗文或者名句
div_tmp = html.xpath(
'//div[@class="main3"]/div[@class="left"]'
'/div[@class="sonspic"]/div[@class="cont"]/p/a//text()') # 根据div分组
if div_tmp[0][-2:] == '诗文':
authors_works[author] = self.domain + div_urls[0]
if div_tmp[0][-2:] == '名句':
authors_famous_sentences[author] = self.domain + div_urls[0]
print(div_tmp[0][-2:])
except Exception as e:
# 写入数据 - .json
path1 = '../jsons/authors_works.json'
path2 = '../jsons/authors_famous_sentences.json'
with open(path1, 'w', encoding='utf-8') as f:
json.dump(authors_works, f)
f.close()
with open(path2, 'w', encoding='utf-8') as f:
json.dump(authors_famous_sentences, f)
f.close()
# 写入数据 - .txt 便于查看
path1 = '../txts/authors_works.txt'
path2 = '../txts/authors_famous_sentences.txt'
with open(path1, 'w', encoding='utf-8') as f:
for key, value in authors_works.items():
if value is None:
f.write(key + ': null' + '\n')
else:
f.write(key + ': ' + value + '\n')
f.close()
with open(path2, 'w', encoding='utf-8') as f:
for key, value in authors_famous_sentences.items():
if value is None:
f.write(key + ': null' + '\n')
else:
f.write(key + ': ' + value + '\n')
f.close()
print('\n————————出现异常————————\n')
print(e)
continue
# 写入数据 - .json
path1 = '../jsons/authors_works.json'
path2 = '../jsons/authors_famous_sentences.json'
with open(path1, 'w', encoding='utf-8') as f:
json.dump(authors_works, f)
f.close()
with open(path2, 'w', encoding='utf-8') as f:
json.dump(authors_famous_sentences, f)
f.close()
# 写入数据 - .txt 便于查看
path1 = '../txts/authors_works.txt'
path2 = '../txts/authors_famous_sentences.txt'
with open(path1, 'w', encoding='utf-8') as f:
for key, value in authors_works.items():
if value is None:
f.write(key + ': null' + '\n')
else:
f.write(key + ': ' + value + '\n')
f.close()
with open(path2, 'w', encoding='utf-8') as f:
for key, value in authors_famous_sentences.items():
if value is None:
f.write(key + ': null' + '\n')
else:
f.write(key + ': ' + value + '\n')
f.close()
5.2 古诗词知识抽取
完成古诗词领域的原始数据获取后，就可以按照古诗词知识模型对原始数据进行知识抽取了。从知识模型可以看出，抽取工作主要完成实体、属性和关系的抽取，但涉及内容并不复杂，所以不必采用知识抽取模型，而仅将原始数据对应即可。
完成该工作的源代码的路径为：/KGQA_Poetry/data_processing/write_spider.py，而生成的所有节点和关系的文件目录分别为：/KGQA_Poetry/data_nodes/、KGQA_Poetry/data_relationships/。
5.3 古诗词知识存储
有了古诗词知识图谱的节点和关系后，就可以着手完成知识导入工作了。完成古诗词知识存储任务的代码路径为：/KGQA_Poetry/data_import/import_data.py。
导入过程较为简单，py2neo包封装了python与neo4j交互的各类功能，所以只需掌握cypher语句就可以完成节点和关系的导入。但值得注意的是，由于本系统依赖的知识图谱节点和关系数量较大，使用纯python模式采用多进程导入也花费了近10小时之久。
下面代码展示了使用多进程导入作品内容，包括节点和关系的导入。
# 7 - 导入作品内容 对应数据库的label（WorkContent：有属性）
def import_work_content(work_content):
try:
title = work_content['title'] # 作品的标题
content = work_content['content'] # 作品内容
author = work_content['author']
dynasty = work_content['dynasty']
translation = work_content['translation'] # 作品翻译
annotation = work_content['annotation'] # 作品注释
background = work_content['background'] # 作品创作背景
appreciation = work_content['appreciation'] # 作品赏析
# 首先创建实体节点
cypher = "MERGE( :WorkContent{title: '%s', content: '%s', author:'%s', dynasty:'%s'," \
" translation: '%s', annotation:'%s', background:'%s', appreciation:'%s'})"
graph.run(cypher % (title, content, author, dynasty, translation, annotation, background, appreciation))
# 然后创建对应的关系
rel = "create" # 作者创作作品
graph.run(
"MATCH (author: Author), (w_c:WorkContent)"
"WHERE author.name='%s' and w_c.content='%s'"
"MERGE (author)-[r:%s{name: '%s'}]->(w_c) RETURN r"
% (author, content, rel, rel) # 可以考虑为关系添加属性
)
rel = "created_in" # 作品创作于某朝代
graph.run(
"MATCH (w_c: WorkContent), (dynasty:Dynasty) "
"WHERE w_c.content='%s' and dynasty.name='%s'"
"MERGE (w_c)-[r:%s{name: '%s'}]->(dynasty) RETURN r"
% (content, dynasty, rel, rel)
)
rel = "title_is" # 作品的题目
graph.run(
"MATCH (w_c: WorkContent), (title: WorkTitle)"
"WHERE w_c.content='%s' and title.name='%s'"
"MERGE (w_c)-[r:%s{name: '%s'}]->(title) RETURN r"
% (content, title, rel, rel)
)
include_sentences = work_content['include'] # 加入包含的关系，直接将其写入数据库
if not include_sentences: # 不包含名句
return
rel = "include" # 作品包含名句
for sentence in include_sentences:
graph.run(
"MATCH (w_c: WorkContent), (f_s: FamousSentence) "
"WHERE w_c.content='%s' and f_s.content='%s' and f_s.author='%s'"
"MERGE(w_c)-[r:%s{name: '%s'}]->(f_s) RETURN r"
% (content, sentence, author, rel, rel) # 可以考虑为关系添加属性
)
except Exception as e:
print('发生异常的作品：', work_content)
print('注意异常n7：', e)
return
# 7 - 多进程导入作品内容
def pool_import_work_content():
path = '../data_nodes/node_work_content.json'
with open(path, 'r', encoding='utf-8') as f:
work_content = json.load(f)
f.close()
pool = Pool(8) # 创建8个进程对象 print(cpu_count())
pool.map(import_work_content, work_content)
六、问答交互模块
该模块主要完成用户问句到答案的映射任务，在于前端和后端的设计。前端代码较为简单，主要使用三件套操作即可，后端只需针对特定功能响应前端就好，采用的是前后端不分离的开发模式。
下面代码给出了有关古诗词领域问答的后端控制。
# 8 - 知识图谱问答系统页面 - 古诗词
@app.route('/KGQA_Poetry', methods=['GET', 'POST'])
@login_required # 登录验证
def KGQA_Poetry():
return render_template('KGQA_Poetry.html')
# 9 - 基于知识图谱的问答 - 古诗词
@app.route("/KGQA_Poetry_Answer", methods=["POST"])
def KGQA_Poetry_Answer():
# 古诗词领域问答
list_str = request.form.get("prompts", None)
answer_list = json.loads(list_str)
question = answer_list[-1]['content'] # 获取前端输入的问题: 李白写过哪些诗啊？
xiaoshi_robot = XiaoShiRobot(question)
answer = xiaoshi_robot.answer()
answer = answer.replace("\n", "<br>") # 便于前端展示
return markdown.markdown(answer) # 转换成 markdown模式
七、问句解析模块
7.1 意图识别与问句分类
在理论篇已经给出了FastText意图识别与问句分类模型的模型流水线，即如下图所示。接下来会按照该流程，讲解各个步骤的实现。
（1）模型原始数据的获取与清洗
由于目前笔者并没有找到古诗词领域的问句训练数据，所以必须从头开始生成，好在目前AI能够提供一些帮助，从而减少了部分工作量。
第一步，根据理论篇中的43类问句，使用AI生成对应的问句数据。
实现思路即是，针对每一类问句，设计问句“种子”，然后投喂给大模型生成多条相似的问句。例如针对问句类型“author_of_title_1（作品标题对应的作者）”的问句，设计种子问句“小诗，你知道游山西村的作者吗？”，然后在AI大模型（文心一言、通义千问等）里输入提示词——请帮我生成与问句“小诗，你知道游山西村的作者吗？”意思完全相同的40条变体问句，这时模型就会输出对应的问句，但需要注意的是由于不同模型的限制，要求生成的数量不能太大，在20-50条的范围是适合的。
在项目路径 —— /KGQA_Poetry/KGQA/kgqa_icr/dataset/query_type.xlsx 里给出了笔者使用的43类问句的所有种子，如下图所示；小伙伴们在复现的过程中，可以继续完善问句或者扩展问句的类型。
第二步，将所有生成的数据进行简单的数据清洗后存放在磁盘，以待后续使用。在项目目录：/KGQA_Poetry/KGQA/kgqa_icr/query_label/ 里存放了43种类型对应的所有问句数据。
（2）文本分词
由于FastText模型对训练数据格式有特定的要求，所以在得到问句数据后，需要对其进行分词处理。
本文使用的分词工具是 —— Jieba分词，由于该工具的字典主要涵盖通用领域，而针对古诗词垂直领域并不能较好地进行分词，所以必须自定义分词词典。该任务在本项目文件——/KGQA_Poetry/KGQA/kgqa_icr/model_pre_process.py里完成，最后生成的词典路径为 —— /KGQA_Poetry/KGQA/user_dict/dict.txt。
值得注意的是，分词的策略有两种，一种是在分词时采用过滤停用词，另一种是不过滤停用词。从最后的实现效果看，第二种策略效果更佳，因为针对短文本，如果过滤了停用词那么对于FastText模型来说，问句中潜在的特征可能减少，从而导致模型识别效果欠佳。
下面的代码给出了生成用户自定义字典的过程。
# 1 - 根据搜集的数据，生成用户词典，便于分词
def generate_user_dict():
root = '../user_dict/dict_category/'
file_names = ['dict_famous_sentence.txt', 'dict_sorted_title.txt',
'dict_author_sorted_work.txt', 'dict_collective_title.txt',
'dict_dynasty.txt', 'dict_work_type.txt']
dict_path = [root + i for i in file_names]
# 所有类型词典，生成总的用户词典
user_dict = {}
for path in dict_path:
with open(path, 'r', encoding='utf-8') as f:
for line in f.readlines():
line = line.split('\n')[0] + ' ' + str(3) + ' ' + 'n' + '\n' # 按照jieba分词字典的模式来
if line not in user_dict:
user_dict[line] = 1
else:
user_dict[line] += 1
f.close()
# 重复的较少：
# a = sorted(user_dict.items(), key=lambda k: k[1], reverse=True)
# print(a)
path = '../user_dict/user_dict.txt'
with open(path, 'w', encoding='utf-8') as f:
for word, _ in user_dict.items():
f.write(word)
f.close()
（3）生成数据集
完成分词后，就可以按照FastText模型要求的数据格式，生成对应的模型数据集，该任务在项目文件 —— /KGQA_Poetry/KGQA/kgqa_icr/model_pre_process.py 中完成。实现代码如下所示，生成的数据集保存在文件 dataset.xlsx 中。
# 5 - 生成带标签的数据集
def generate_dataset():
path = '../user_dict/user_dict.txt'
jieba.load_userdict(path) # 加载用户词典的时候，尽量放在文件第一次启动的地方
label_name = {}
root = './query_label/'
for folder_path, folder_name, file_names in os.walk(root):
for file in file_names:
label = file.split('.')[0] # 获取 label 名称
label_name[label] = root + file # 该标签下的数据
break
# 过滤停用词 - 不过滤停用词的效果更好
# path = './user_dict/stopwords/cn_stopwords.txt'
# with open(path, 'r', encoding='utf-8') as f:
# stopwords = [line.split('\n')[0] for line in f.readlines()]
# f.close()
# 提取数据，生成数据集
sentences = []
seg_sentences = []
labels = []
for label, path in label_name.items():
with open(path, 'r', encoding='utf-8') as f:
print(path)
for line in f.readlines():
sentence = line.replace('\n', '')
sentences.append(sentence) # 原问句
# seg_sentence = ''
# for word in jieba.lcut(sentence):
# if word not in stopwords:
# seg_sentence += ' ' + word # 切分后的问句
seg_sentence = ' '.join(jieba.cut(sentence))
seg_sentences.append(seg_sentence)
labels.append(label) # 该问句的标签
df = pd.DataFrame({"sentence": sentences, "seg_sentence": seg_sentences, "label": labels})
# path = '../dataset/unfiltered_stopwords/dataset.xlsx' # 未过滤停用词
path = './dataset/dataset.xlsx'
df.to_excel(path, sheet_name='Sheet1', startcol=0, index=False)
（4）模型训练
有了训练集，就可以写代码实现模型训练的工作了，在项目理论篇已经对FastText模型的原理以及训练过程中的参数做了阐述，如果仍有疑惑可以继续查阅相关资料进一步理解。
模型训练和预测在项目文件 —— /KGQA_Poetry/KGQA/kgqa_icr/query_classification.py 中完成，包括拆分训练集、验证集和测试集，训练过程如下所示。
# 5 - 对数据集进行拆分，得到训练集、验证集和测试集
def dataset_split(dataset, path):
df = pd.read_excel(dataset, sheet_name="Sheet1")
X = df['seg_sentence'].tolist() # 分好词的句子
y = df['label'].tolist() # 分类标签
# 拆分训练、验证、测试集
X_train_dev, X_test, y_train_dev, y_test = train_test_split(X, y, test_size=0.1)
X_train, X_dev, y_train, y_dev = train_test_split(X_train_dev, y_train_dev, test_size=0.1)
# 将训练数据与标签组装
train_data = train_data_format(X_train, y_train)
test_data = train_data_format(X_test, y_test)
dev_data = train_data_format(X_dev, y_dev)
# 写入文件
write_list_into_file(train_data, path + 'train_data.txt')
write_list_into_file(test_data, path + 'test_data.txt')
write_list_into_file(dev_data, path + 'dev_data.txt')
# 6 - 训练模型、测试和输出各意图PRF值
def fasttext_train(train_data, test_data, model_path, **kwargs):
# 训练
clf = train_supervised(input=train_data, **kwargs) # **kwargs 训练参数列表
clf.save_model('%s.bin' % model_path)
# 测试
result = clf.test(test_data)
precision = result[1]
recall = result[2]
# print('Precision: {0}, Recall: {1}\n'.format(precision, recall))
logger.info('Precision: {0}, Recall: {1}\n'.format(precision, recall))
# 输出每类PRF值
test_sents, y_true = split_sent_and_label(test_data)
y_pred = [i[0].replace('__label__', '') for i in clf.predict(test_sents)[0]]
logger.info(classification_report(y_true, y_pred, digits=3))
# print(classification_report(y_true, y_pred, digits=3))
（5）模型预测
由前所述，该模块主要完成NLU的任务，得到训练好的模型后，就可以借此解析问句了（包括意图识别和命名实体识别），此任务在项目文件 —— /KGQA_Poetry/KGQA/kgqa_nlu.py 完成，下面是问句理解的实现：
class NaturalLanguageUnderstanding:
__doc__ = "自然语言理解 - 理解用户输入的自然语言问句"
def __init__(self, question, logger):
self.question = question
self.logger = logger
# 都需要使用绝对路径，按需更改
self.fasttext_model_path = r'D:\KGQA_Poetry\KGQA\models\fasttext_model.bin'
self.bert_model_path = r'D:\KGQA_Poetry\KGQA\models\bert_ner.pth'
# 1 - 问句意图分类
def intent_classification(self):
sent_list = [' '.join(jieba.cut(self.question))] # 对分词后的问句进行分类
cls = ft.load_model(self.fasttext_model_path)
# 预测结果 - 二维数组
res = cls.predict(sent_list, k=3) # topK = 3 # 返回最可能的三种情况, threshold=0.3
# 预测标签
topK_labels = [label.replace('__label__', '') for label in res[0][0]]
# 预测概率
topK_probability = [prob for prob in res[1][0]]
# TODO：如果第一意图概率超过0.7，且没有找到答案，那么考虑预测的第二个意图
# print(topK_labels[0], topK_probability[0])
# 对用户的问句解析，从而得到命名实体
intent = topK_labels[0] # 取候选意图的第一个
# probability = topK_probability[0]
entities = self.question_parse()
logger.info("intent: " + intent)
logger.info(entities)
# logger.info(intents)
print(self.question)
return intent, entities # 返回意图和实体，便于NLG生成答案
# 2 - 使用 bert 模型，结合问句意图，进行问句解析（主要是命名实体识别）
def question_parse(self):
result = predict_ner(self.question) # 预测问句的命名实体
# 针对目前模型的局限性，调整输出的结果
entities_dict = {}
for label, entities in result.items():
entities_dict[label] = []
for entity in entities:
entities_dict[label].append(entity[:-1])
return entities_dict
7.2 命名实体识别
同FastText模型一样，在理论篇已经给出BERT模型完成NER任务的流程，即如下图所示。接下来会按照该流程，讲解各个步骤的实现。
（1）数据获取与命名实体确定
从理论篇的系统架构图中可以看出，NER任务的原始数据跟意图识别与问句分类子模块的原始数据相同，所以不必再重复找寻训练数据。目前，笔者完成的系统中的古诗词领域的命名实体一共有八种，如下图所示，小伙伴们在复现完成后，可以继续扩展设计的命名实体，从而可以回答更多的问句类型。
（2）标注训练数据
虽然BERT模型已经经过了预训练，但是NER任务是一个下游任务，需要进行微调训练后才能在NER任务上有更佳的识别效果。有了模型的训练数据，就可以着手开始对数据进行“愉快”地标注了。
笔者选择的数据标注平台是label-studio，目前是开源的，较为稳定，可以完成多种任务的数据标注。这篇文章——命名实体识别（NER）标注神器——Label Studio 简单使用 从安装启动到如何标注作了指引，按照步骤安装就好。
安装完成后，在数据导入环节笔者选择的是将43类问句的所有数据分成了三次标注（标注过程实在是太枯燥了😿😿😿……，不信可以试试，哈哈哈哈），项目文件：/KGQA_Poetry/KGQA/model_label/ft_data_ner_1.txt 是其中的一部分数据。标注完成后，有两种方式导出，一般选择 JSON_MIN 格式就可以了，在项目文件 —— /KGQA_Poetry/KGQA/kgqa_ner/bert_ner_data.py中提供了格式转化的程序，如下所示。最后生成的训练集和测试集在项目 —— /KGQA_Poetry/KGQA/model_label/bert_label/ 中，自行查看，待到后续训练使用。
# 2 - 将打了标签的数据转换成BERT模型训练的格式数据集
def change_label_data():
# path = '../model_label/bert_label/export_0.json' # 从label-studio平台导出的数据
# path = '../model_label/bert_label/export_1.json'
path = '../model_label/bert_label/export_2.json'
with open(path, 'r', encoding='utf-8') as f:
data_list = json.load(f)
f.close()
# 以简单内容（JSON——MIN）导出的 json 文件，以这种方式转换
dataset_labels = []
for label_data in data_list:
if len(label_data) == 8:
text = label_data['text'] # 原文本
labels = label_data['label'] # 注意，不是每一问句都有label，需要特殊处理
entities = [] # 所有实体
for label in labels:
start_idx = label['start'] # 标注开始点和结束点
end_idx = label['end']
entity = label['text']
if len(label['labels']) > 1: # 有多个标签
types = label['labels']
for type in types:
entities.append({
"start_idx": start_idx,
"end_idx": end_idx,
"type": type,
"entity": entity
})
else:
type = label['labels'][0] # 只有一个标签
entities.append({
"start_idx": start_idx,
"end_idx": end_idx,
"type": type,
"entity": entity
})
# 整个标记完的数据集
dataset_labels.append({
"text": text,
"entities": entities
})
else: # 该问句没有实体
text = label_data['text'] # 原文本
# 整个标记完的数据集
dataset_labels.append({
"text": text,
"entities": []
})
# 以详细内容（JSON）导出的 json 文件，以下列这种方式转换
"""
dataset_labels = []
for label_data in data_list:
text = label_data['data']['text'] # 原文本
labels = label_data['annotations'][0]['result']
entities = [] # 所有实体
for label in labels:
value = label['value']
start_idx = value['start'] # 标注开始点和结束点
end_idx = value['end']
entity = value['text']
if len(value['labels']) > 1: # 有多个标签
types = value['labels']
for type in types:
entities.append({
"start_idx": start_idx,
"end_idx": end_idx,
"type": type,
"entity": entity
})
else:
type = value['labels'][0] # 只有一个标签
entities.append({
"start_idx": start_idx,
"end_idx": end_idx,
"type": type,
"entity": entity
})
# 整个标记完的数据集
dataset_labels.append({
"text": text,
"entities": entities
})
"""
# path = '../model_label/bert_label/author_profile.json'
# path = '../model_label/bert_label/ft_data_ner_0.json'
# path = '../model_label/bert_label/ft_data_ner_1.json'
path = '../model_label/bert_label/ft_data_ner_2.json'
with open(path, 'w', encoding='utf-8') as f:
json.dump(dataset_labels, f, indent=2, ensure_ascii=False)
f.close()
（3）模型训练与预测
如前所述，BERT模式的参数量极为庞大，一般的笔记本电脑很难跑完一轮，所以需要租用云服务器，笔者选择的算力平台是AutoDL,该平台机器较多，费用较为合理（对学生有优惠）。文章 —— 新手小白如何租用GPU云服务器跑深度学习 讲解了如何租用实例，以及如何连接Pycharm进行训练，当然还有其它文章讲解，小伙伴们可以自行对照查看。
笔者需要强调两点，一是，为了减少环境配置的时间而加快复现，你们可以先私信我将你们的AutoDL平台的ID给我，我直接将我的镜像分享给你们，然后你们在创建实例的时候使用本地镜像就好。当然除了古诗词领域的数据，其它领域（医疗、音乐、汽车等）的数据完成标注后也可以在该模型上跑。
二是，在云平台跑完模型后，需要将最优的模型保存到本地，Pycharm专业版可以完成此任务，下面的链接提供的是与云平台镜像对应的本地代码，你们放在本地通过Pycharm连接到AutoDL上的机器就可以了。
八、答案生成模块
该模块主要完成NLG的任务，其代码实现在项目文件 —— /KGQA_Poetry/KGQA/kgqa_nlg.py 中，例如以询问诗人的朝代为例，下面展示其代码实现。
elif self.intent == 'dynasty_of_author_1': # 询问作者的朝代 - 请问你知道李白是哪个朝代的吗？小诗
if 'Author' in self.entities: # 是否识别出作者实体
author = list(set(self.entities['Author']))[0] # 获取问句的实体 - 取第一个实体回答
cypher = "match (v: Author{name: '%s'}) return v.dynasty, v.gender limit 1" % author
result = self.graph.run(cypher).data()
if not result: # 知识图谱中找不到匹配，则返回默认的答案
return answer
dynasty, gender = result[0].get('v.dynasty'), result[0].get('v.gender')
if gender == '男':
slot = '他'
else:
slot = '她'
answer = self.ANSWER_PREFIX[
idx] + "当然知道" + author + "的所属朝代啦。" + slot + "是" + dynasty + "的诗词作家呢。"
九、本地项目启动
整个项目的启动较为繁琐，涉及文件补全和相关软件、环境的配置，可联系作者获取项目的相关资料。
参考文献
① 邵浩,张凯,李方圆等. 从零构建知识图谱: 技术、方法与案例[M]. 北京:机械工业出版社,2021.
② 李辉. Flask Web开发实战:入门、进阶与原理解析[M]. 北京:机械工业出版社,2018.
③ 明日科技. Python网络爬虫从入门到精通[M]. 北京:清华大学出版社.2021.
注：由于参考的文献较多，不再逐一列举，若有侵权，联系立删。
其余篇章
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
13
0- 0
已为社区贡献1条内容
所有评论(0)