基于农业知识图谱的苹果问答系统实现方法与步骤

2024年12月5日 134点热度

环境部署

环境安装

jieba 是高效的中文分词工具,支持词性标注和关键词提取;neo4j 是用于与 Neo4j 图数据库交互的 Python 驱动,支持复杂图形数据查询;numpy 提供多维数组操作和数值计算功能,是数据处理的基础工具;torch 是 PyTorch 深度学习框架,支持动态计算图和 GPU 加速,用于深度学习模型的构建与训练;transformers = 4.46.3 是 Hugging Face 提供的 NLP 库,支持预训练模型的使用与微调。

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

neo4j 图数据库安装

知识图谱数据需要存储于图数据库之中,由于 Neo4j 是基于 Java 的图数据库,运行 Neo4j 需要启动 JVM 进程,因此必须安装 JAVASE 的 JDK。本次实验使用 jgk17.0.8 版本,与 neo4j5.11.0 版本。

JDK 下载地址:https://www.oracle.com/java/technologies/downloads/?er = 221886#java21

下载后直接点击安装,新版本会自动设置环境变量

验证安装:

java -version

image-20241125203211778

neo4j 图数据库下载地址:https://neo4j.com/deployment-center/

解压到自定义英文文件夹,新建环境变量 NEO4J_HOME,变量值就是刚才解压下载文件的地址。在系统变量区的 Path 中新建变量:%NEO4J_HOME%\bin

image-20241125203838730

img

启动服务验证安装:

neo4j console

image-20241125204242845

按住 Ctrl 键点击 http://localhost: 7474/,重置密码(默认密码 neo4j),进入数据库页面,安装成功。

image-20241125204623998

数据说明

apple_entities_relations.csv 中存储数据有两种。一种结构为 <实体,关系,实体>,例如 <'苹果枝干虫害' , '包括' , '梨小食心虫'>。知识图谱总体结构如图所示。另一种结构为 <实体,属性,属性值>,例如 <'苹果白纹羽病' , '病症' , '地上部叶片变小、黄化,叶柄和中脉发红,枝条节间短;'>。所有数据来源为李林光《苹果实用栽培技术》,在构建知识图谱时可以根据书籍自行定义,也可以训练模型进行实体和关系的抽取。知识图谱结构如图所示。

image-20241123173022411

templates 文件夹下有问题模板 question_templates.json 和回答模板 answer_templates.json,均以 json 形式存储。在意图识别过程中与用户输入文本进行对比匹配,提取用户问题中的意图信息。所有模板需要根据实体关系自行定义。在问题模板中,键“发病原因”表示意图,值为字符串的列表,每个字符串都是一个可能的问题,用于询问某种病害的发病原因。在回答模板中,键“发病原因”与问题模板相对应,值为一个字符串,其中 {} 是占位符,用于在生成答案时被替换为具体的病害名称和发病原因。
$$
a+b =\hat{c}
$$

//question_templates
{
    "发病原因": [
        "病害的发病原因是什么?",
        "病害发病的因素有什么"
    ],
}
//answer_templates.json
{
    "发病原因": "{}的发病原因是:{}"
}

知识图谱构建

代码通过调用 build_knowledge_graph_from_csv 函数,将 CSV 文件中的实体和关系数据导入 Neo4j 数据库,完成知识图谱的构建;随后,调用 process_entity_frequencies 函数,统计 CSV 文件中实体的出现频次,并将结果写入 apple.txt 中。

# 示例调用
if __name__ == "__main__":
    # 配置数据库连接信息和文件路径
    NEO4J_URI = "bolt://localhost:7687"
    USERNAME = "neo4j"
    PASSWORD = "123456"
    CSV_FILE_PATH = "apple_entities_relations.csv"
    OUTPUT_FILE_PATH = "apple.txt"
    # 调用主函数构建知识图谱
    build_knowledge_graph_from_csv(CSV_FILE_PATH, NEO4J_URI, USERNAME, PASSWORD)
    # 统计实体频次并写入文件
    process_entity_frequencies(CSV_FILE_PATH, OUTPUT_FILE_PATH)

KnowledgeGraph 类用于与 Neo4j 数据库交互,提供构建知识图谱的功能,包括创建实体、关系和属性。通过 init 方法初始化数据库连接,接受 URI、用户名和密码作为参数,并通过 GraphDatabase.driver 建立驱动实例;close 方法用于关闭数据库连接,释放资源。create_node_relationship 方法实现两个实体及其关系的创建,接收实体名称和关系类型作为参数,利用 Cypher 语句确保实体和关系的存在。create_property_relationship 方法为实体添加属性及属性关系,将属性名和属性值作为独立节点,与实体通过关系相连。

class KnowledgeGraph:
    def __init__(self, uri, user, password):
        """初始化 Neo4j 数据库连接"""
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        """关闭数据库连接"""
        self.driver.close()

    def create_node_relationship(self, entity1, relation, entity2):
        """创建实体和关系"""
        with self.driver.session() as session:
            session.run("""
                MERGE (e1:Entity {name: $entity1})
                MERGE (e2:Entity {name: $entity2})
                MERGE (e1)-[:RELATION {type: $relation}]->(e2)
            """, entity1=entity1, relation=relation, entity2=entity2)

    def create_property_relationship(self, entity, property_name, property_value):
        """
        为实体添加属性关系
        将属性名和属性值转化为独立节点,并通过关系与实体相连
        """
        with self.driver.session() as session:
            session.run("""
                MERGE (e:Entity {name: $entity})
                MERGE (p:Property {name: $property_name, value: $property_value})
                MERGE (e)-[:HAS_PROPERTY]->(p)
            """, entity=entity, property_name=property_name, property_value=property_value)

build_knowledge_graph_from_csv 函数用于从 CSV 文件构建知识图谱,连接到指定的 Neo4j 数据库,并根据 CSV 中的数据创建实体、关系和属性。首先,函数通过创建 KnowledgeGraph 类的实例来初始化与 Neo4j 的连接。接着,函数尝试打开 CSV 文件并读取内容,使用 csv.reader 按制表符(\t)分割每行数据。在每行数据中,如果有三个元素,分别表示两个实体和它们之间的关系,函数根据关系类型决定是创建普通的实体关系(通过 create_node_relationship 方法)还是将关系转化为实体属性关系(通过 create_property_relationship 方法)。完成图谱构建后,函数会关闭数据库连接并输出构建完成的消息。如果在处理过程中出现任何异常,错误信息会被捕捉并输出。

def build_knowledge_graph_from_csv(csv_file, uri, user, password):
    """
    从 CSV 文件构建知识图谱的主函数

    参数:
        csv_file (str): CSV 文件路径
        uri (str): Neo4j 数据库 URI
        user (str): 数据库用户名
        password (str): 数据库密码
    """
    graph = KnowledgeGraph(uri, user, password)
    try:
        with open(csv_file, "r", encoding="utf-8") as file:
            reader = csv.reader(file, delimiter="\t")
            for row in reader:
                if len(row) == 3:
                    entity1, relation, entity2 = row
                    if relation == "包括":
                        graph.create_node_relationship(entity1, relation, entity2)
                    else:
                        # 替换为使用关系存储属性
                        graph.create_property_relationship(entity1, relation, entity2)
                else:
                    print(len(row))
                    print(f"数据格式不正确: {row}")
        print("知识图谱构建完成!")
    except Exception as e:
        print(f"构建知识图谱时出错: {e}")
    finally:
        graph.close()

process_entity_frequencies 函数用于从 CSV 文件中对实体去重存储,统计频次,作为词典用于后续 jieba 分词。

def process_entity_frequencies(csv_file, output_file):
    """
    从 CSV 文件中统计实体频次,并将结果写入文本文件

    参数:
        csv_file (str): CSV 文件路径
        output_file (str): 输出文件路径
    """
    entity_counter = Counter()
    # 从 CSV 文件中读取数据
    with open(csv_file, 'r', encoding='utf-8') as csvfile:
        reader = csv.reader(csvfile, delimiter='\t')
        for row in reader:
            if len(row) >= 3:
                entity1 = row[0].strip()  # 实体1
                # 更新实体计数
                entity_counter[entity1] += 1

    # 将统计结果写入文本文件
    with open(output_file, 'w', encoding='utf-8') as f:
        for entity, frequency in entity_counter.items():
            f.write(f"{entity} {frequency} n\n")  # 添加自动计算的词频和固定的词性
        # f.write(f"{'苹果'} {100} n\n")
    print(f"实体及其词频已成功从 {csv_file} 写入 {output_file} 文件。")
知识图谱构建完成!
实体及其词频已成功从 apple_entities_relations.csv 写入 apple.txt 文件。

image-20241121213803420

image-20241121213835419

苹果叶部病害 6 n
苹果病害 4 n
主要品种 11 n
苹果白纹羽病 5 n
苹果裂果病 4 n
果锈 4 n
锈果病 4 n
苹小卷叶蛾 3 n
秦阳 5 n
泰山嘎啦 5 n

意图识别

代码首先通过 load_pretrained_model_and_tokenizer 函数加载预训练的模型和分词器,为后续的自然语言处理任务提供支持。接着调用 load_entities 加载实体集合,调用 load_templates 加载问题模板和查询模板,用于匹配和处理用户输入。用户通过控制台输入问题后,代码使用 process_user_question 函数结合模型、模板和实体集合,对问题进行解析,提取出用户意图和相关实体。

if __name__ == "__main__":
    # 加载预训练模型和分词器
    tokenizer, model = load_pretrained_model_and_tokenizer()
    # 加载模板和实体
    entities = load_entities()
    question_templates, query_templates = load_templates()
    # entities = load_entities()
    # 用户输入
    user_input = input("请输入你的问题:")
    intent,category = process_user_question(user_input, tokenizer, model, question_templates, entities)
    if category and intent:
        print(f"用户提问分类结果:\n  实体: {category}, 意图: {intent}")
def load_pretrained_model_and_tokenizer(model_path="./bert-base-chinese"):
    tokenizer = BertTokenizer.from_pretrained(model_path)
    model = BertModel.from_pretrained(model_path)
    return tokenizer, model

def load_entities(entity_file='apple.txt'):
    entities = set()
    with open(entity_file, 'r', encoding='utf-8') as f:
        for line in f:
            entity = line.split()[0]  # 获取每行的实体名称
            entities.add(entity)
    return entities

def load_templates(question_template_path='templates/question_templates.json',
                   answer_template_path='templates/answer_templates.json'):
    with open(question_template_path, 'r', encoding='utf-8') as f:
        question_templates = json.load(f)
    with open(answer_template_path, 'r', encoding='utf-8') as f:
        answer_templates = json.load(f)
    return question_templates, answer_templates

get_sentence_vector 函数接收输入句子、分词器(tokenizer)和预训练模型(model),将输入文本转化为张量格式,通过模型前向传播计算隐状态,并对所有 Token 的隐藏状态进行平均池化,生成句子向量(句子的固定维度表示),最终返回为 NumPy 数组格式。extract_entity 函数用于从一组单词(words)中提取指定实体(例如已知的苹果品种名称)。通过逐一检查单词是否属于指定的实体集合(entities),如果匹配则返回该实体,否则返回 None,表示未识别到实体。cosine_similarity 函数:用于计算两个向量之间的余弦相似度,首先将输入向量展平为一维,然后计算点积和向量的范数(模长),利用公式计算余弦相似度(向量夹角的余弦值)。如果输入向量中有零向量,返回 0.0,表示无相似度。

def get_sentence_vector(sentence, tokenizer, model):
    inputs = tokenizer(sentence, return_tensors="pt")  # 将输入文本转为 tensor 格式
    with torch.no_grad():
        outputs = model(**inputs)
        hidden_states = outputs.last_hidden_state
    # 对所有 token 进行平均池化,得到句子向量
    return hidden_states.mean(dim=1).numpy()

def extract_entity(words, entities):
    for word in words:
        if word in entities:  # 检查已知的苹果品种名称
            return word
    return None  # 未识别实体

def cosine_similarity(vec1, vec2):
    # 确保输入向量是1D
    vec1 = vec1.flatten()
    vec2 = vec2.flatten()
    dot_product = np.dot(vec1, vec2)
    norm_vec1 = np.linalg.norm(vec1)
    norm_vec2 = np.linalg.norm(vec2)
    return dot_product / (norm_vec1 * norm_vec2) if norm_vec1 and norm_vec2 else 0.0

process_user_question 首先使用 get_sentence_vector 函数将用户输入转化为句向量表示,并通过 jieba 加载自定义词典对用户输入进行分词,同时动态提取分词结果中的实体。如果未识别到实体,函数会终止并返回 None。接着,通过遍历问题模板库,对每个模板生成句向量,与用户输入向量进行余弦相似度计算,找到与用户输入最匹配的意图类别(最高相似度对应的类别)。如果成功匹配到意图,函数返回最佳意图类别和提取到的实体;否则返回 None 并提示无法识别意图。

def process_user_question(user_input, tokenizer, model, question_templates, entities):
    user_vector = get_sentence_vector(user_input, tokenizer, model)
    jieba.load_userdict('apple.txt')
    words = jieba.lcut(user_input)  # 分词
    print(f"分词结果: {words}")

    # 动态提取实体
    entity = extract_entity(words, entities)
    if not entity:
        print("未能识别实体。")
        return None, None

    # 意图识别
    best_intent = None
    highest_similarity = 0.0

    # 遍历模板库匹配最佳意图
    for category, questions in question_templates.items():
        for question in questions:
            # 获取模板的词向量
            template_vector = get_sentence_vector(question, tokenizer, model)
            similarity = cosine_similarity(user_vector, template_vector)

            # 更新最优意图
            if similarity > highest_similarity:
                highest_similarity = similarity
                best_intent = category

    # 输出匹配结果
    if best_intent:
        # print(f"识别到的意图:{best_intent}")
        return best_intent, entity
    else:
        print("未能识别意图。")
        return None, None

输出

请输入你的问题:花叶病怎么治疗?
Building prefix dict from the default dictionary ...
Loading model from cache C:\Users\Lenovo\AppData\Local\Temp\jieba.cache
分词结果: ['花叶病', '怎么', '治疗', '?']
Loading model cost 0.347 seconds.
Prefix dict has been built successfully.
用户提问分类结果:
  实体: 花叶病, 意图: 防治方法

进程已结束,退出代码为 0

图数据库搜索

answer = fetch_disease_info(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="123456",
    disease_name="烟嘎2号",
    attribute_type="成熟时间",
    templates_path="templates/answer_templates.json"
)
print(answer)

fetch_disease_info 函数用于查询病害信息并基于模板生成格式化答案。函数接受数据库连接参数(uri、user、password)、病害名称(disease_name)、查询的属性类型(attribute_type)以及模板文件路径(templates_path)。首先,通过读取指定路径的模板文件加载模板,确保根据属性类型生成对应的格式化回答。接着,利用 GraphDatabaseQuery 类与 Neo4j 数据库交互,通过 query_disease_info 方法查询指定病害及其属性类型的相关信息。查询结果若存在,则将结果列表格式化为逗号分隔的字符串,并根据模板生成答案;若无结果,则返回模板中的默认提示信息。最后,无论查询是否成功,都通过 db_query.close()关闭数据库连接。

def fetch_disease_info(uri, user, password, disease_name, attribute_type, templates_path):
    """
    查询病害信息并基于模板生成答案。
    参数:
        uri (str): 数据库连接 URI
        user (str): 用户名
        password (str): 密码
        disease_name (str): 病害名称
        attribute_type (str): 查询的属性类型
        templates_path (str): 模板文件路径
    返回:
        str: 格式化的最终答案
    """
    # 加载模板
    with open(templates_path, "r", encoding="utf-8") as f:
        templates = json.load(f)

    db_query = GraphDatabaseQuery(uri, user, password)
    try:
        # 获取查询结果
        result = db_query.query_disease_info(disease_name, attribute_type)
        print(result)

        # 根据模板生成答案
        template = templates.get(attribute_type, "{}没有相关的{}信息。")

        # 如果有结果,进行格式化
        if result:
            result_list = "、".join(result)
            return template.format(disease_name, result_list)
        else:
            return template.format(disease_name, attribute_type)
    finally:
        db_query.close()

GraphDatabaseQuery 类用于与 Neo4j 数据库交互,实现针对病害信息的查询功能。通过 __init__ 方法初始化数据库连接,接收 URI、用户名和密码作为参数,并创建一个驱动实例以与数据库通信;close 方法用于关闭数据库连接,释放相关资源。核心方法 query_disease_info 通过接收病害名称(disease_name)和属性类型(attribute_type),执行针对性查询并返回结果。如果属性类型为“包括”,方法执行的 Cypher 查询会匹配病害与其关联实体的关系并返回实体名称;否则,方法将匹配病害节点与属性节点的关系,基于属性类型返回属性值。查询结果通过会话执行后,提取为一个值列表返回。

class GraphDatabaseQuery:
    def __init__(self, uri, user, password):
        # 初始化 Neo4j 图数据库连接
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        # 关闭数据库连接
        self.driver.close()

    def query_disease_info(self, disease_name, attribute_type):
        """
        根据病害名称和属性类型查询相关信息。
        参数:
            disease_name (str): 病害的名称
            attribute_type (str): 需要查询的属性类型(如:防治方法、发病原因等)
        返回:
            list: 查询结果(如防治方法、发病原因等的值)
        """
        with self.driver.session() as session:
            # 根据 attribute_type 判断选择的查询模板
            if attribute_type == "包括":
                query = """
                    MATCH (d {name: $disease_name})-[:RELATION]->(e)
                    RETURN e.name AS attribute_value
                """
            else:
                query = """
                    MATCH (d {name: $disease_name})-[:HAS_PROPERTY]->(p:Property)
                    WHERE p.name = $attribute_type
                    RETURN p.value AS attribute_value
                """

            # 执行查询并返回结果
            result = session.run(query, disease_name=disease_name, attribute_type=attribute_type)
            return [record["attribute_value"] for record in result]

输出

['9月上旬']
烟嘎2号的果实成熟时间为:9月上旬

问答系统实现

代码实现了从知识图谱构建到用户问题处理与回答生成的完整流程。首先,代码配置了 Neo4j 数据库连接信息(NEO4J_URIUSERNAMEPASSWORD)以及文件路径(CSV_FILE_PATHOUTPUT_FILE_PATH),并调用 build_knowledge_graph_from_csv 函数将 CSV 文件中的实体和关系数据导入 Neo4j 数据库,构建知识图谱;随后通过 process_entity_frequencies 函数统计实体频次并保存到指定文件中。接着,代码加载了预训练的 BERT 模型和分词器(tokenizermodel),以及实体和问题模板数据,用于后续的自然语言处理任务。用户输入问题后,通过 process_user_question 函数解析问题,提取出用户意图和目标实体分类结果(如实体类别和意图类型),并将这些信息传递给 fetch_disease_info 函数。该函数通过查询 Neo4j 数据库,结合回答模板生成最终的答案并打印出来。

import json
from collections import Counter
import jieba
import numpy as np
import torch
from neo4j import GraphDatabase
import csv
from transformers import BertTokenizer, BertModel

if __name__ == "__main__":
    # 配置数据库连接信息和文件路径
    NEO4J_URI = "bolt://localhost:7687"
    USERNAME = "neo4j"
    PASSWORD = "123456"
    CSV_FILE_PATH = "apple_entities_relations.csv"
    OUTPUT_FILE_PATH = "apple.txt"

    # 调用主函数构建知识图谱
    build_knowledge_graph_from_csv(CSV_FILE_PATH, NEO4J_URI, USERNAME, PASSWORD)
    # 统计实体频次并写入文件
    process_entity_frequencies(CSV_FILE_PATH, OUTPUT_FILE_PATH)
    # 加载预训练模型和分词器
    tokenizer, model = load_pretrained_model_and_tokenizer()
    # 加载模板和实体
    entities = load_entities()
    question_templates, _ = load_templates()

    # 用户输入
    user_input = input("请输入你的问题:")
    intent,category = process_user_question(user_input, tokenizer, model, question_templates, entities)
    if category and intent:
        print(f"用户提问分类结果:\n  实体: {category}, 意图: {intent}")
    answer = fetch_disease_info(uri=NEO4J_URI,user=USERNAME,password=PASSWORD,disease_name=category,
                                           attribute_type=intent,templates_path="templates/answer_templates.json"
    )
    print(answer)
知识图谱构建完成!
实体及其词频已成功从 apple_entities_relations.csv 写入 apple.txt 文件。
请输入你的问题:苹果的主要品种有哪些?
Building prefix dict from the default dictionary ...
Loading model from cache C:\Users\Lenovo\AppData\Local\Temp\jieba.cache
Loading model cost 0.369 seconds.
Prefix dict has been built successfully.
分词结果: ['苹果', '的', '主要品种', '有', '哪些', '?']
用户提问分类结果:
  实体: 主要品种, 意图: 包括
['秦阳', '泰山嘎啦', '华硕', '华玉', '富红早嘎', '红凤', '齐早红', '华瑞', '华美', '烟嘎1号', '烟嘎2号']
主要品种包括:秦阳、泰山嘎啦、华硕、华玉、富红早嘎、红凤、齐早红、华瑞、华美、烟嘎1号、烟嘎2号
请输入你的问题:褐斑病的症状有什么?
Building prefix dict from the default dictionary ...
Loading model from cache C:\Users\Lenovo\AppData\Local\Temp\jieba.cache
分词结果: ['褐斑病', '的', '症状', '有', '什么', '?']
Loading model cost 0.375 seconds.
Prefix dict has been built successfully.
用户提问分类结果:
  实体: 褐斑病, 意图: 病症
['叶片发病初期,正面出现黄褐色小点,逐渐扩大成褐色不规则病斑,边缘有绿色晕圈;后期病斑中央呈灰白色,边缘褐色,常出现同心轮纹状排列的小黑点。']
褐斑病病症有:叶片发病初期,正面出现黄褐色小点,逐渐扩大成褐色不规则病斑,边缘有绿色晕圈;后期病斑中央呈灰白色,边缘褐色,常出现同心轮纹状排列的小黑点。

xxs9331

这个人很懒,什么都没留下

文章评论