简介:面向计算机相关专业学生、职场新人及知识图谱初学者的保险知识图谱与简易问答系统毕设源码。项目基于开源保险产品数据完成知识抽取、实体关系建模与图谱存储,并实现问句意图识别、语义解析和答案查询的完整问答链路,配套产品文档、原始保险产品数据及说明文件,解压后可直接运行。压缩包共20个文件,以Python脚本、XML配置、xls数据表、markdown说明与docx文档为主,整体仅1.84MB,轻量易部署。已有102人下载学习,适合毕业设计、课程设计或人工智能方向的项目初探。资源内包含图谱构建与修订脚本、问题分类与解析模块、Web交互页面及数据文件,并附有README与环境说明,可帮助读者清晰理解KGQA从数据准备到系统实现的工程路径,便于在此基础上扩展功能或迁移到其他领域。
1. 保险知识图谱问答系统:一份 Excel 就能跑起来的开源项目
把一张几百行的保险产品 Excel 变成一台能回答“这款重疾险多大年龄能买”的问答系统——这个开源保险知识图谱项目给人的第一印象就是这种反差。它的核心是一个完整的 KGQA 流程:ins_product_data.xls是输入,graph_build.py负责把表格数据写进 Neo4j 图谱,question_classifier.py做意图识别,question_parse.py抽实体,question_query.py查图谱,最后通过server_websocket.py把答案送回浏览器界面。适合正在做知识图谱毕设、课设,或者想快速理解“从关系型表格到图数据库再到问答”这条链路的人。它不依赖训练好的大模型,全部基于规则和词典匹配,吃透它之后再去看工业级的问答系统,很多思路是相通的。
2. 从保险产品 Excel 到 Neo4j 图谱:schema 设计、导入脚本与索引约束
2.1 先看数据再定 schema:保险产品字段如何映射成节点和关系
我拆这个项目的第一件事不是跑代码,而是打开ins_product_data.xls看字段。保险产品数据通常包含产品名称、保险公司、险种类型、投保年龄范围、保障期限、缴费方式、保额、犹豫期这些列。这些字段天然适合映射成知识图谱里的两类东西:属性和关系。产品名称、保额、犹豫期属于产品节点自身的属性;保险公司应该独立成一个节点,因为多个产品会归属同一家公司,做成关系可以避免大量重复的公司名称字符串。
当时我对这份数据的处理是:先按“实体类型候选”把字段分成三组。第一组是产品本身,节点标签用Product;第二组是保险公司和险种类别,分别用Company和Category;第三组是保障责任、缴费方式这类多值字段,拆成独立的Coverage节点,用关系连回产品。这样设计的好处是:用户问“重疾险有哪些产品”时,可以直接从Category节点反查Product,不需要在 Excel 里做模糊匹配。
| 数据字段 | 图谱映射 | 节点/关系 | 说明 |
|---|---|---|---|
| 产品名称 | Product.name | 节点属性 | 主标识,建议加唯一约束 |
| 保险公司 | Company | 节点 | 一对多关系Company-[:HAS_PRODUCT]->Product |
| 险种类型 | Category | 节点 | 如重疾险、医疗险、意外险,Product-[:BELONGS_TO]->Category |
| 投保年龄 | Product.age_range | 节点属性 | 字符串存“28天-60周岁”,查询时再解析 |
| 保障期限 | Product.term | 节点属性 | 字符串存“保至70岁/终身” |
| 缴费方式 | Coverage | 节点 | 按缴费类型拆成独立节点更利于问答匹配 |
| 保额 | Product.sum_insured | 节点属性 | 数值型,注意 Excel 读取后的类型转换 |
上面这张表就是我当时做完的数据字典。如果你是拿这份资源做自己的课设,建议先照着这份数据字典把 Excel 打开核对一遍,因为你拿到的数据字段可能跟我上面列的有细微差异——比如有些产品表里没有“犹豫期”列,或者“投保年龄”写成了“出生满28天至60周岁”。字段不一致不影响建图逻辑,但会影响你后续问答模板里的正则表达式。
2.2 graph_build.py 做了什么:读 Excel、建节点、建关系的完整链路
这个项目的建图脚本graph_build.py是整个资源里第一个要跑的文件,它的核心逻辑并不复杂:先用 pandas 读 Excel,然后逐行创建节点和关系。源码里没有用LOAD CSV的批量导入方式,而是直接用 Neo4j 的 Python 驱动逐条执行 Cypher。我一般会先建索引和约束,再导数据,避免半路出现重复节点。
from neo4j import GraphDatabase import pandas as pd driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "123456")) def create_indexes(tx): tx.run("CREATE CONSTRAINT product_name IF NOT EXISTS FOR (p:Product) REQUIRE p.name IS UNIQUE") tx.run("CREATE CONSTRAINT company_name IF NOT EXISTS FOR (c:Company) REQUIRE c.name IS UNIQUE") tx.run("CREATE CONSTRAINT category_name IF NOT EXISTS FOR (c:Category) REQUIRE c.name IS UNIQUE") def build_graph(tx, row): tx.run( """ MERGE (c:Company {name: $company}) MERGE (cat:Category {name: $category}) MERGE (p:Product { name: $name, age_range: $age_range, term: $term, sum_insured: $sum_insured, pay_period: $pay_period }) MERGE (c)-[:HAS_PRODUCT]->(p) MERGE (p)-[:BELONGS_TO]->(cat) """, company=row["保险公司"], category=row["险种类型"], name=row["产品名称"], age_range=row["投保年龄"], term=row["保障期限"], sum_insured=row["保额"], pay_period=row["缴费方式"] ) df = pd.read_excel("data/ins_product_data.xls") with driver.session() as session: session.execute_write(create_indexes) for _, row in df.iterrows(): session.execute_write(build_graph, row) driver.close()这段代码里有几个点我需要单独解释。第一,MERGE而不是CREATE,这是刻意为之:产品数据里同一个保险公司会出现几十次,用MERGE保证公司节点只创建一次,后续只是添加关系。第二,execute_write是 Neo4j Python 驱动 4.x 之后推荐的事务写法,比旧的session.run更安全,出错会自动回滚。第三,索引和约束放在导入前执行,这算是我自己的习惯,先卡住唯一性再灌数据,否则导到一半发现重复节点,整张图都要重来。
这里有一个很容易翻车的地方:如果graph_build.py里连的是没有认证的本地 Neo4j,驱动初始化时传auth=None或者删掉auth参数,但项目中server_websocket.py和question_query.py还是按auth=("neo4j", "123456")去连,两边的连接配置不一致就会导致建图成功、查询失败。我建议你拿到源码后,先全局搜一下GraphDatabase.driver,把三处连接参数对齐。
2.3 索引、约束与 graph_revise.py:为什么批量导入后要单独刷一遍
graph_revise.py是容易被忽略的一个脚本,它的作用是在图谱构建完成后做一轮修正。常见场景是这样:Excel 里“缴费方式”一列同时存在“年交”和“20年交”,前者是缴费频率,后者是缴费期限,如果不拆分开,问答系统后续做槽位匹配时会把“年交”和“20年交”混为一谈。graph_revise.py做的事情就是扫描一遍已有的节点属性,按规则把混合字段拆成Coverage节点并重新建立关系。
这段逻辑在代码里通常表现为:查询所有Product节点,读pay_period属性,用正则判断是“年交/月交/趸交”还是数字加“年交”,然后分别挂到不同的Coverage节点下。这种修正脚本单独拎出来而不是写进graph_build.py,好处是你改完修正规则后不需要重新全量导数据,只跑graph_revise.py就能增量刷新图谱。我自己的经验是,产品数据每次更新后,先跑graph_build再跑一次graph_revise,整个过程控制在两分钟以内。
跑通建图这一步后,你可以在 Neo4j Browser 里执行MATCH (p:Product) RETURN p LIMIT 25快速检查节点是否正常生成。如果看到某个产品的name是nan或者数字带上小数点,说明 Excel 读取时列类型没处理干净,下一步问答系统肯定跟着出问题。这个坑我在第 4 章会展开讲。
3. 简易问答系统的三个 Python 模块:意图分类、槽位解析与查询组装
3.1 意图分类:规则模板比训练模型更适合小样本
这个项目把问答系统拆成了三个文件:question_classifier.py、question_parse.py、question_query.py,对应经典问答系统的意图识别、槽位提取、查询生成三个阶段。question_classifier.py是入口,它负责判断用户问的是“保障内容”“投保年龄”“保费价格”还是“产品推荐”。这一步用规则模板而不是训练深度学习模型,本质原因是:保险产品问句的句式高度集中,几百条典型问法就能覆盖绝大多数用户输入,规则模板在这种封闭域场景下准确率能到 90% 以上,而且不需要标注数据、不需要调参。
import re # 意图模板:关键词组合 -> 意图标签 intent_rules = { "age": ["年龄", "几岁", "多大", "能买吗", "投保年龄"], "coverage": ["保障", "保什么", "责任", "包括"], "price": ["保费", "多少钱", "价格", "费用"], "term": ["期限", "保多久", "保障期", "终身", "定期"], "recommend": ["推荐", "有哪些", "什么产品", "哪个好"], } def classify(question: str) -> str: for intent, keywords in intent_rules.items(): for kw in keywords: if re.search(kw, question): return intent return "unknown"这段代码的精髓是intent_rules的结构设计:意图名作为键,关键词列表作为值。为什么用re.search而不是简单的in判断?因为保险问句里经常出现“我今年 45 岁还能买重疾险吗”这种句子,“岁”和“能买吗”不是连续出现的,in只能做子串匹配,如果问句里写的是“年龄 45”,in "几岁"就会误判,而re.search至少能匹配到“年龄”这个词。我后来在实际项目中把这个模板换成了更细粒度的“词表 + 正则组合”,比如(年龄|几岁|多大).*(能买|投保|可以)这种交叉正则,准确率还能再提一截。
这个文件里还有一段值得细看的地方:意图优先级。我给的示例里recommend放在最后,因为“推荐”这个词很少和其他意图词冲突,而“有哪些”如果放到coverage之后,用户问“有哪些重疾险”会被coverage的“保障”命中吗?不会,但用户问“重疾险保哪些疾病”时,如果不控制优先级,“有哪些”就会先被recommend匹配,导致意图跑偏。源码里对这类交叉情况做了顺序调整,我强烈建议你别改这个顺序。
3.2 槽位解析:词典 + 多模匹配抽取“重疾险”“某公司”这种实体
意图确定了,接下来要从问句里抽出实体。question_parse.py主要做槽位提取,它不依赖 NLP 工具包,而是维护了一个产品名词典和公司词典,然后用多模匹配做最大匹配。为什么用最大匹配?因为产品名存在包含关系,比如“平安福”和“平安福尊享版”,如果只做最短匹配,用户问“平安福尊享版有什么保障”会先命中“平安福”,抽出来的实体就是错的。
from ahocorasick import Automaton # 构建关键词自动机 def build_automaton(words: list) -> Automaton: automaton = Automaton() for idx, word in enumerate(words): automaton.add_word(word, (idx, word)) automaton.make_automaton() return automaton # 用自动机做最长匹配 def extract_entities(question: str, automaton: Automaton) -> list: entities = [] for end_index, (_, word) in automaton.iter(question): start_index = end_index - len(word) + 1 entities.append((start_index, word)) # 合并重叠实体,取最长 entities.sort(key=lambda x: (x[0], -len(x[1]))) merged = [] for pos, word in entities: if not merged or pos >= merged[-1][0] + len(merged[-1][1]): merged.append((pos, word)) return [word for _, word in merged]这里的ahocorasick就是 Python 生态里做多模式匹配最常用的库,底层是 AC 自动机,一次性把所有词典词条编译成状态机,然后线性扫描问句。拿保险产品问句实测,几百个产品名和公司名的词典规模下,单句匹配耗时在毫秒级,比用 for 循环逐词in判断快了一个量级。项目源码里没有完全照抄这个写法,但核心思路一致:词典匹配 + 长词优先。
槽位解析这块有一个我在实际项目里反复踩的坑:产品词典不完整。ins_product_data.xls里的产品名是有限的,但用户问话里可能出现“平安福”“少儿平安福”等变体,词典里没有的词直接抽不出来。这个项目的兜底方案是:在question_query.py里做属性模糊匹配,用CONTAINS替代精确匹配,这是最后一道防线。
3.3 查询组装与 WebSocket 接入:从问句到 Cypher 到聊天框
question_query.py是整个系统最务实的一个文件,它接收意图和实体,拼 Cypher 语句,返回查询结果。比如意图是age,问句抽出来的实体是“平安福”,它就会查询这个产品的投保年龄范围,然后原样返回字符串。我给出一段常见玩法的示意代码,你会发现翻译过程非常直接:
from neo4j import GraphDatabase driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "123456")) cypher_map = { "age": "MATCH (p:Product {name: $name}) RETURN p.age_range AS answer", "price": "MATCH (p:Product {name: $name}) RETURN p.premium AS answer", "coverage": """ MATCH (p:Product {name: $name})-[:HAS_COVERAGE]->(c:Coverage) RETURN collect(c.name) AS answer """, } def query_answer(intent: str, entity: str): cypher = cypher_map.get(intent) if not cypher: return "这个问题我暂时还不会回答" with driver.session() as session: result = session.run(cypher, name=entity) record = result.single() return record["answer"] if record else "没有找到相关产品"从这段代码能看到整个项目最关键的约定:question_classifier.py输出的意图字符串,必须和cypher_map的键一一对应。我拆项目时发现很多人改了intent_rules里的意图名,比如把age改成age_limit,却忘了同步改question_query.py,结果问答系统所有关于年龄的问题全部返回“不会回答”。源码的健壮性不够,它不会对这种不匹配做主动报错,只会静默失败。你改的时候务必两边对齐。
项目里的server_websocket.py和web_socket.html组成一个简易的浏览器问答界面。server_websocket.py用 Python 的 websocket 库监听端口,收到浏览器发来的问句后,调用classify -> extract_entities -> query_answer三步,再把结果通过同一个 WebSocket 连接推回前端。整个链路不涉及 HTTP 请求、不需要 RESTful 接口,对课设答辩来说足够直观。你在本地验证时,先启动server_websocket.py,再用浏览器打开web_socket.html,就可以直接测试。
4. 避坑记录:Python 缓存、Neo4j 连接、Excel 类型与 WebSocket 端口
4.1 Python 3.6 的 pyc 缓存让你改了代码却不生效
现象:我修改了question_classifier.py里的关键词表,重新运行服务器,可无论怎么问,分类结果还是旧关键词的行为。
原因:这个项目在__pycache__目录里带着三个.pyc文件,文件名带cpython-36标识,说明是 Python 3.6 环境编译的缓存。如果你本机是 Python 3.6 以上版本,Python 解释器发现源文件修改时间没有变化时,会优先加载.pyc缓存;而我当时遇到的情况是解压后的文件时间戳异常,导致修改后的源码没有被重新编译。
解决:删掉整个__pycache__目录,然后重启server_websocket.py。从那以后我每次改完源码都会顺手执行find . -name "__pycache__" -exec rm -rf {} +,避免这种低级干扰。如果你用的是 PyCharm,记得把.idea目录里的项目解释器版本和本地 Python 版本对齐,否则缓存问题会反复出现。
4.2 Neo4j 连接被拒:要么没启动,要么认证没对上
现象:运行graph_build.py报Failed to establish connection,但 Neo4j Desktop 明明打开了。
原因:Neo4j 的连接方式分两种,Desktop 版默认bolt://localhost:7687需要认证,而早期版本默认neo4j/neo4j,首次登录会强制改密码;项目代码里写死的是123456。如果你安装的 Neo4j 版本比较新,默认强制开启认证,密码对不上就直接拒连。
解决:先在 Neo4j Browser 里确认当前数据库的实际密码,改了graph_build.py、question_query.py、server_websocket.py三处auth参数。更稳妥的做法是新建一个只读账号专门给问答系统用,这样建图和查询用不同权限的账号,避免误操作删数据。我用的是 Neo4j 5.x,驱动版本要对应neo4jPython 库 5.x,否则语法兼容性也可能出问题。
4.3 Excel 里的 ID 被读成浮点型,图谱里多出一堆“1.0”
现象:图谱里产品节点的name不是中文而是1.0、2.0这类数字,或者公司节点数量比预期多了好几倍。
原因:pd.read_excel读取时,如果某一列本来是数字 ID,但表头下方有过空行或格式不统一,pandas 会把它读成float64类型。比如“产品代码”列是 104 和 201,Excel 里显示为整数,但 pandas 读出来后是104.0,直接拼接进 Cypher 就变成了带小数点的节点名。
解决:读入数据后先执行df["产品名称"] = df["产品名称"].astype(str),然后把.0后缀清洗掉。我一般会加一行df["产品名称"] = df["产品名称"].str.replace(r"\.0$", "", regex=True),跑完再打印df.dtypes确认。这种问题不会让程序报错,但会让图谱数据整个变脏,是最隐蔽的坑。
4.4 中文拼接 Cypher 报错:双引号、单引号与转义
现象:手工在 Neo4j Browser 里执行MATCH (p:Product {name:"平安福"}) RETURN p没问题,但从 Python 传参执行就报语法错误,或者查出一条空记录。
原因:问题基本都出在参数化没有做全。有人喜欢把实体直接拼进 Cypher 字符串,比如f"MATCH (p:Product {{name: '{entity}'}})",一旦实体里含有英文单引号,或者本身就是个长句子,Cypher 串就断了。中文本身不背锅,但中文句子里的标点符号经常是全角,跟查询条件对不上,表现成“中文参数报错”。
解决:所有动态内容一律用$param参数传递,不要拼字符串。我给的示例代码里session.run(cypher, name=entity)就是标准写法,让驱动帮你做转义。如果你改过源码发现还是查不到,把entity打印出来,和 Neo4j 里的name逐字符比对,八成是多了一个空格或括号。
4.5 WebSocket 端口被占用与跨域问题
现象:启动server_websocket.py后提示Address already in use,或者浏览器页面打开后输入问句没反应。
原因:server_websocket.py默认监听8765端口,如果你之前启动过没关掉,或者另一个程序占了这个端口,服务就起不来。还有一个常见原因是浏览器的安全策略拦截了 WebSocket 跨域请求,如果你直接用file://协议打开web_socket.html,有些浏览器默认不允许页面发起 WebSocket 连接。
解决:启动前先查看端口占用,Linux 上执行lsof -i:8765,Windows 上执行netstat -ano | findstr 8765,找到占用进程 PID 后结束它。跨域问题最简单的方式是用python -m http.server 8080起一个本地静态服务,然后访问http://localhost:8080/web_socket.html,服务器再开一个 CORS 中间件放行所有来源。这个项目源码里server_websocket.py末尾带了跨域放行头,但我发现不同 Python 版本的 websocket 库对跨域头支持差异挺大,项目里那台机器能跑,挪到你的环境未必能行。
5. 把 Neo4j Browser 当调试台:问答准确率的验证闭环
5.1 先手工调 Cypher 再回填代码的节奏
拆完源码后,我形成了一个固定的验证习惯:先用 Neo4j Browser 手工把 Cypher 跑通,再回填到question_query.py。以“推荐适合 60 岁老人的医疗险”为例,先在 Browser 里执行MATCH (c:Category {name:"医疗险"})<-[:BELONGS_TO]-(p:Product) WHERE p.age_range CONTAINS "60" RETURN p.name,确认返回结果符合预期,再把这个 Cypher 原样填到cypher_map里。这样做能帮你把错误边界卡在两层:第一层排除 Cypher 语法问题,第二层只剩 Python 侧的参数传递和意图分类问题。
5.2 一个自检脚本:三个典型问句跑一遍
我后来给这种项目加了一个自检脚本,专门用来回归测试。核心思路是准备一组问句和期望答案,然后依次调用分类、解析、查询三个模块,最后对比输出。你不用一次性做很多条,挑三类就够了:问年龄限制的、问保障内容的、问产品推荐的。
# check_questions.py from question_classifier import classify from question_parse import extract_entities from question_query import query_answer test_cases = [ ("平安福多少岁能买", "age", "平安福"), ("守卫者1号保哪些疾病", "coverage", "守卫者1号"), ("推荐一款儿童重疾险", "recommend", ""), ] for question, expected_intent, expected_entity in test_cases: intent = classify(question) entity_list = extract_entities(question) answer = query_answer(intent, entity_list[0] if entity_list else "") print(f"问句: {question}") print(f"意图: {intent} | 实体: {entity_list} | 答案: {answer}") assert intent == expected_intent, f"意图不匹配: {intent}" if expected_entity: assert expected_entity in entity_list, f"实体抽取失败: {entity_list}"这个脚本的价值在于:当你改了关键词表或者词典之后,跑一遍就能知道哪些问句行为变了。我第一次跑的时候,第三句“推荐一款儿童重疾险”的意图被分类成了coverage,因为“重疾险”命中了“保障”模板里的词,后来我在recommend规则里加了“推荐”作为强信号词,并把它的优先级提到coverage之前,才解决了冲突。这类问题肉眼很难发现,脚本一跑就露馅。
5.3 验证意图分类的边界:问法变形
还有一个值得单独说的经验:意图分类的准确率验证,要看问法变形而不是标准问句。“年龄”意图如果只测“这款产品多大年龄能买”这种标准句,准确率肯定好看,但用户实际会问“50 岁还能投吗”“超过 60 岁可以买吗”“投保年龄上限是多少”,这些变体才是最考验规则模板的地方。我的做法是把每个意图的类型模板扩充到至少十种问法,包括否定式“多大就不能买了”、疑问式“有年龄限制吗”,然后全部丢进上一节的自检脚本里跑。这个项目本身的规则模板覆盖了常见问法,但你扩充词典时一定要同步扩问法。
从那以后我每次改完graph_build.py或者新增产品数据,都会强制清空 Neo4j 库重新全量导入,然后跑一遍上面的自检脚本,确认三类问句的答案没有变化才继续改下一个模块。这套“先手工查 Cypher、再脚本回归问句、最后全量重建图谱”的习惯看着笨,但确实帮我少走了很多冤枉路。希望帮到你。
本文还有配套的精品资源,点击获取