简介基于开源保险产品数据构建的保险知识图谱与简易问答系统是一套完整的毕设源码包采用Python实现面向计算机、人工智能等相关专业的在校学生与开发者适用于毕业设计、课程设计或知识图谱方向入门实践。压缩包共20个文件核心为6个Python源码覆盖保险知识图谱构建、问题意图分类、语义解析、图谱查询及WebSocket问答服务等完整流程形成从数据到图谱再到问答的端到端链路另附保险产品数据表、产品文档、Web交互页面与运行配置文件项目结构清晰整体1.84MB轻量且易于部署。已有102人学习下载代码经真实测试运行通过配套说明文档与产品文档可辅助快速理解项目结构。下载后可直接复现保险产品实体关系图谱并通过浏览器界面体验智能问答若遇到环境配置或运行问题作者提供远程教学支持特别适合需要完整可运行方案作为毕设参考或初学知识图谱与NLP问答的读者。1. 把保险产品表变成图谱数据到问答的闭环能解决什么拿到一份开源保险产品数据你能直接回答「哪款百万医疗险的免赔额最低」吗如果数据是几十张 Excel 表这个问题往往要人肉翻半天。把产品、公司、险种、保障责任拆成实体和关系再叠一层简易问答这个问题就变成一句图查询。这就是保险知识图谱在问答场景里的价值——给产品数据建一个能查询的语义层再让 python 程序替用户问出来。这类项目适合有 Python 和数据库基础、又想把知识图谱落地的人。它常被拿来当课程设计、毕业设计选题也是保险行业数据从业者验证图技术的第一步。它能解决的是「产品对比」和「责任关系」类问题哪款产品包含某责任、某责任被哪些产品覆盖、某公司的产品线有哪些。下面我按数据清洗、图谱构建、问答实现三条线把完整做法和踩过的坑讲透。2. 保险产品数据清洗与本体设计把表格拆成实体和关系2.1 开源保险产品数据的常见形态标题里说的“开源保险产品数据”落到实操中通常指公开可获取的产品条款、保险产品信息表或者保险平台的产品快照。很多开发者用 python 爬虫从公开页面抓产品信息也有人直接整理行业协会、保险机构披露的信息转成 CSV 或 JSON。数据来源不同但关键字段高度一致产品名称、险种类别百万医疗险、重疾险、定期寿险等、保险公司、投保年龄、保险期间、缴费方式、等待期、保障责任列表、保额/免赔额/赔付比例、保费费率。字段常见来源适合直接入图谱吗处理建议产品名称产品列表页适合作为节点标识清洗空格、全角字符保留版本号险种类别保险分类适合作为分类节点统一叫法医疗险 / 百万医疗 / 重疾险保险公司公司信息栏适合作为公司节点去掉“股份有限公司”等后缀再统一保障责任产品详情页适合作为责任节点或属性责任名带括号说明时需要剥离保费费率表不适合直接建节点提数值或区间粒度到投保年龄等待期产品规则适合作为属性统一单位为天这些字段里最麻烦的是“保障责任”。一个产品可能包含十几条责任责任名长得不一样语义却差不多。比如“住院医疗费用保险金责任一”和“一般医疗保险金”在不同产品里往往都指同一个核心责任。如果直接在表格里当字符串处理后面进图时会出现几十个意思重复的责任节点问答系统也没办法聚合查询。2.2 本体建模把产品表拆成节点和关系为什么非要做本体设计因为表格里一行一个产品字段是平铺的但问答系统要回答“某责任覆盖哪些产品”“某公司有哪些产品”这类问题需要把表和表之间的逆询路径提前显式建模。把产品名、公司名、责任名各建一个节点再挂上关系查询才能沿着关系走。我一般会定义四个实体类型不额外堆复杂语义Company 公司核心属性是统一后的公司名。Category 险种类别核心属性是分类名比如“百万医疗险”。Product 产品核心属性是 product_code 和 product_name。product_code 用内部生成的稳定编号因为产品名后来可能会改版本。Coverage 保障责任核心属性是责任名比如“住院医疗”。关系只有三条Company -[:PUBLISHED]- ProductProduct -[:BELONGS_TO]- CategoryProduct -[:HAS_COVERAGE]- Coverage。至于“免赔额”“保额”“赔付比例”这类数值属性我建议放在产品节点上不单独建节点。原因有两个一是问答时“某产品免赔额多少”已经知道产品名直接取属性最快二是“哪款产品免赔额最低”这种比较查询只需要在产品节点上做排序。如果把这些属性也做成实体图谱会爆炸式膨胀而简易问答完全用不上那么细的粒度。这个本体设计不是拍脑袋。公司、产品、险种、责任这四个维度正好覆盖保险产品数据的核心检索场景按公司查产品、按险种查产品、按责任反查产品、按产品查详情。等以后真要扩展到健康险理赔等场景再往实体上加属性或者挂新关系就行不需要推倒重来。2.3 数据处理代码清洗、去重、对齐的 Python 实现下面是一段可复用的 Python 清洗代码。假设输入是一个 CSV每行一个产品coverage_list字段用中文逗号分隔多个责任名。import pandas as pd import re import unicodedata raw pd.read_csv(insurance_products.csv, encodingutf-8-sig) def clean_cell(value): if pd.isna(value): return value str(value) value unicodedata.normalize(NFKC, value) value value.replace(\u3000, ).replace(\u00a0, ) value re.sub(r[\u200b\u200c\u200d], , value) # 零宽字符 return re.sub(r\s, , value).strip() raw[product_name] raw[product_name].apply(clean_cell) raw[company_name] raw[company_name].apply(clean_cell) raw[category_name] raw[category_name].apply(clean_cell) raw[product_code] raw.apply( lambda r: re.sub(r\W, -, r[company_name] - r[product_name]), axis1 ) raw raw.drop_duplicates(subset[product_code], keeplast)这里做了四件事。先用unicodedata.normalize(NFKC)把全角数字、字母统一成半角避免“202”和“2024”被当成两个产品再把全角空格和不可见字符清掉接着用“公司名产品名”生成稳定的 product_code最后按 product_code 去重保留最后一条记录。utf-8-sig这个编码参数很实用处理从 Excel 导出的 CSV 时能自动去掉 BOM不然第一个字段名会多个不可见字符。下面是保障责任的拆解和实体对齐COVERAGE_ALIAS { 住院医疗费用保险金责任一: 住院医疗, 住院医疗费用保险金责任二: 住院医疗, 一般医疗保险金: 住院医疗, 重大疾病医疗保险金: 重疾医疗, 恶性肿瘤医疗保险金: 癌症医疗, } def split_coverage_list(cell): if not cell: return [] parts re.split(r[;,], cell) parts [clean_cell(x) for x in parts if clean_cell(x)] seen set() result [] for p in parts: aliased COVERAGE_ALIAS.get(p, re.sub(r.*?, , p)) if aliased not in seen: seen.add(aliased) result.append(aliased) return result raw[coverage_list] raw[coverage_list].apply(split_coverage_list) print(raw[[product_code, product_name, coverage_list]].head())拆解逻辑不复杂先按标点分隔责任名再按别名表合并没有别名时用正则去括号里面的“责任一”这类版本说明。COVERAGE_ALIAS是整个清洗过程的核心它决定了实体对齐的上限。保险责任在行业内叫法非常多强烈建议在项目初期把数据里所有责任名列出来人工归一次类不要边跑边补。如果你是第一次用 Python 做这类数据处理先确认环境没装 Python 的可以去 python官网下载安装包安装时勾选 Add Python to PATH要在服务器上跑就查 linux系统安装python包管理方式不一样基础环境不通后面每一步都会被放大。3. 用 Python 和 Neo4j 建图谱约束、MERGE 与批量导入3.1 为什么选 Neo4j图数据库的选型理由知识图谱建完以后需要一个能高效做多跳查询的存储。关系型数据库做两跳 JOIN 还能接受三跳以上就难维护。保险问答里最常见的“哪家公司的哪款产品包含住院医疗责任”关系型写法要 join 公司表、产品表、产品责任关联表、责任表代码长且没法一眼看出语义。用 Neo4j 的 Cypher 写同样查询只要三行所以它成为知识图谱项目最常用的开源图数据库。Neo4j 不是唯一选择内存图库、nebula 等也有优点。但对“保险知识图谱 简易问答系统”这个规模社区版完全够用官方 Python driver 维护也稳定。想从头学 neo4j构建知识图谱用它练手阻力最小。还有一个现实考虑课程设计和毕设都需要“能被老师一眼看懂的技术”Cypher 的可读性比一堆前端可视化代码更直观。提示如果你用的是 Neo4j Desktop默认用户名是 neo4j密码是第一次启动时设置的用 Docker 启动时注意把 7474 和 7687 端口都映射出来。3.2 建约束和索引避免重复节点的第一步数据导入前必须先建唯一约束。不建约束MERGE 在并发场景下可能写出重复节点另一种常见错误是多人往同一个库导入数据几十个重复产品名瞬间出现后面问答全乱。Neo4j 5.x 的约束语法使用REQUIRE4.x 使用ON。我一般先建一个约束脚本CREATE CONSTRAINT product_code IF NOT EXISTS FOR (p:Product) REQUIRE p.code IS UNIQUE; CREATE CONSTRAINT company_name IF NOT EXISTS FOR (c:Company) REQUIRE c.name IS UNIQUE; CREATE CONSTRAINT category_name IF NOT EXISTS FOR (c:Category) REQUIRE c.name IS UNIQUE; CREATE INDEX product_name_idx IF NOT EXISTS FOR (p:Product) ON (p.name);第一行给 Product 的 code 加唯一约束这是整个图谱不重复的基石。Company 和 Category 的 name 也要求唯一确保“某公司”不会因为多一个空格被建两次。最后那个索引不是唯一约束而是给产品名字段加速精确匹配因为问答系统经常用“产品名”反查节点。IF NOT EXISTS是 Neo4j 4.4 以后支持的如果你的版本较旧重复执行建约束语句会报错去掉这段再手动确认即可。3.3 批量导入从 pandas 到 Cypher 的 Python 代码清洗后的 DataFrame 要转成图谱。常见做法是用官方neo4j驱动而不是老旧的 py2neo。py2neo 在 Neo4j 4.4 之后上游维护变慢和 5.x 服务端握手经常出现认证报错。现在更建议直接用官方 driver。from neo4j import GraphDatabase BATCH_SIZE 500 class InsuranceGraphImporter: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def close(self): self.driver.close() staticmethod def _clean_value(value): return if value is None else str(value).strip() def create_graph(self, batch): cypher UNWIND $batch AS row MERGE (c:Company {name: row.company_name}) MERGE (cat:Category {name: row.category_name}) MERGE (p:Product {code: row.product_code, name: row.product_name}) MERGE (c)-[:PUBLISHED]-(p) MERGE (p)-[:BELONGS_TO]-(cat) FOREACH (cov_name IN row.coverage_list | MERGE (cov:Coverage {name: cov_name}) MERGE (p)-[:HAS_COVERAGE]-(cov) ) with self.driver.session() as session: session.run(cypher, batchbatch) def import_dataframe(self, df): records [] for _, row in df.iterrows(): records.append({ company_name: self._clean_value(row[company_name]), category_name: self._clean_value(row[category_name]), product_code: self._clean_value(row[product_code]), product_name: self._clean_value(row[product_name]), coverage_list: row[coverage_list] or [], }) if len(records) BATCH_SIZE: self.create_graph(records) records.clear() if records: self.create_graph(records)这段代码的核心在 Cypher先把每一条记录展开成row然后按 key 分别 MERGE 公司、产品、险种。注意 Product 的 MERGE 只用了code所以 code 必须稳定唯一name作为属性同步写进去查询时既能按 name 匹配又能返回给用户。Coverage 放在FOREACH里逐个 MERGE并创建HAS_COVERAGE关系。这样同一个责任被多个产品复用时节点会被复用而不是重复创建。BATCH_SIZE我建议设在 200 到 1000。太小会频繁开启事务太大请求体过大本地机器内存不够时会卡死。数据量上万一定要用这种方式批量提交不要 for 循环逐条写。导入完成后用一句 Cypher 检查节点总量MATCH (p:Product) RETURN count(p) AS product_cnt, count(DISTINCT p.code) AS distinct_code;当product_cnt和distinct_code不一致说明还有重复 code要回清洗阶段处理。这是图构建的质检关口很多初学者跳过结果后面问答查出的产品数虚高。与其去免费python源码大全里翻一个现成的图谱脚本不如先按这个小结构自己跑一遍你会更清楚生产数据里每个字段去了哪儿。4. 简易问答系统实现从中文问题到 Cypher 查询的模板路由4.1 问答流程设计从自然语言到图查询的路径简易问答系统通常不引入机器学习靠模板路由就能覆盖相当比例的产品查询类问题。流程分四步分词词表加载、实体识别、意图判断、生成 Cypher 查询。分词用 jieba并把产品名、公司名、险种名、责任名加入自定义词典保证“百万医疗险”不被拆开。实体识别根据词表在原句中做匹配识别出问题里的公司、险种、产品或责任。意图判断根据关键词把问题分类比如“有哪些产品”“某产品某责任是什么”“某产品的某属性是多少”。生成 Cypher把识别出的实体填到模板里执行查询返回格式化结果。这类“智能问答系统”的实现方式没有想象中复杂。核心就一句话把“用户问什么”映射成“图里怎么查”。你不需要一次做完所有意图先覆盖三类高频问题整个系统的可用性就出来了。网上 python教程 经常跳过这一步的设计直接贴模型但模板路由的难度恰恰不在模型而在实体边界怎么划。4.2 Python 代码最小的模板问答系统下面是可运行的 QAEngine 骨架。假设你已经导入了图谱并且 Neo4j 在本地 7687 端口。先做词表加载和实体识别import jieba from neo4j import GraphDatabase jieba.load_userdict(insurance_words.txt) # insurance_words.txt 每行一个词例如 # 百万医疗险 # 住院医疗费用保险金 # 免赔额 product_names [平安e生保2024, 超越保2020, 好医保长期医疗] # 可从图谱中导出 category_names [百万医疗险, 重疾险, 定期寿险] attribute_kws {免赔额: deductible, 保额: amount, 赔付比例: ratio} class QAEngine: def __init__(self, uri, user, password): self.driver GraphDatabase.driver(uri, auth(user, password)) def _match_entity(self, question): hit {} for name in product_names: if name in question: hit[product] name for cat in category_names: if cat in question: hit[category] cat for alias, prop in attribute_kws.items(): if alias in question: hit[attribute] prop return hit然后执行查询def answer(self, question): ent self._match_entity(question) if product in ent and ent.get(attribute): cypher ( MATCH (p:Product {name: $name}) RETURN p.name AS name, p. ent[attribute] AS value LIMIT 1 ) params {name: ent[product]} with self.driver.session() as session: record session.run(cypher, params).single() return f{record[name]} : {record[value]} if record else 没有查到结果 if category in ent and product not in ent: cypher ( MATCH (p:Product)-[:BELONGS_TO]-(c:Category {name: $name}) RETURN p.name AS name LIMIT 20 ) params {name: ent[category]} with self.driver.session() as session: result session.run(cypher, params) return [record[name] for record in result] or 没有查到结果 if product in ent and 责任 in question: cypher ( MATCH (p:Product {name: $name})-[:HAS_COVERAGE]-(cov:Coverage) RETURN cov.name AS name LIMIT 20 ) params {name: ent[product]} with self.driver.session() as session: result session.run(cypher, params) return [record[name] for record in result] or 没有查到结果 return 抱歉这个问题还没覆盖。代码逻辑很直白先匹配实体再看问题里是否带属性关键词。如果问“平安e生保2024免赔额多少”命中 product attribute就查产品节点上的 deductible 属性如果问“百万医疗险有哪些产品”命中 category沿BELONGS_TO关系查产品列表如果问“好医保长期医疗包含哪些责任”命中 product 和“责任”关键词走HAS_COVERAGE关系返回责任列表。匹配失败就回一句“还没覆盖”不抛异常。这里有两个参数容易踩坑。一是attribute_kws的 key 必须和图谱里的属性名对齐数据清洗时把“免赔额”存成deductible这里要同一套命名二是LIMIT 20限制返回条数。如果问题带“最低”“最高”这类比较词模板还要再拆一层先查所有产品或责任节点的属性排序后取第一个。实现上只是把 Cypher 加一句ORDER BY p.attr ASC LIMIT 1。4.3 回答边界哪些问题能答哪些答不了模板问答的边界一定要心里有数。它能稳定回答“某产品的属性是什么”“某险种有哪些产品”“某公司发布过什么产品”“某个责任被哪些产品覆盖”这类结构清晰的事实问题。它不能回答“我妈今年 50 岁买什么保险最合适”这种需要年龄计算、预算判断和产品比较的推理问题也不能回答“这款产品比那款产品好在哪里”这种综合评分问题。遇到这种问题时简易问答系统要有一个兜底话术而不是强行拼接不靠谱的答案。这也是这个方向值得做的教训把“不知道”明确说出来比给错误答案更让用户信任。后续升级方向也很清楚——把图谱查询结果拼成上下文交给 LLM 生成回答这就是现在主流的 RAG 路线。但要不要上 RAG先让模板问答跑通再说因为实体识别和查询路径在 RAG 里一样需要。5. 保险知识图谱问答系统5 个常见坑与排查方法5.1 数据清洗与实体对齐的坑坑一产品名看起来一样Cypher 却查不到。现象图谱可视化里产品节点存在但问答系统按产品名查询返回“没有查到结果”。我排查了很久最后在 Python 里用repr(product_name)才发现字符串中间藏了零宽空格和全角冒号。原因很简单某保险平台导出数据时把中文冒号和英文冒号混用产品名又夹了不可见字符导致“平安e生保 2024”和“平安e生保2024”被当成两个节点。解决在清洗阶段统一调用规范化函数unicodedata.normalize(NFKC, value)处理全角字符再用正则剔除\u200b-\u200d零宽字符。如果已经建好图可以用一句 Cypher 把异常节点找出来MATCH (p:Product) WHERE p.name CONTAINS \u200b RETURN p.name;坑二同一责任在不同产品里叫法不同责任节点满天飞。现象导入职责后责任名多达几十条实际语义只有几个。原因只做了字符级清洗没做语义别名合并。解决维护一个责任别名表在清洗阶段把“住院医疗费用保险金责任一”和“一般医疗保险金”映射到同一个“住院医疗”。这个表只能靠人工迭代建立没有捷径。我通常先跑一句MATCH (cov:Coverage) RETURN cov.name, count(*)把责任节点全部列出按出现频次人工合并高频名。5.2 图数据库写入与查询的坑坑三MERGE 之后仍然出现重复产品节点。现象count(p)明显大于产品实际数量按产品名查询时返回多行。原因一是没有建唯一约束MERGE 只在同一个事务里检查重复跨事务场景下不能保证全局唯一原因二是导入时 product_code 在不同批次里格式不一致比如一批带空格一批不带MERGE 认为不是同一个节点。解决先执行第 3.2 节里的约束语句再检查 code 生成规则统一strip()不要混用product_code和product_name当唯一键。顺带提一句批量导入时batch_size控制在 200 到 1000也别一次性把整个 DataFrame 全转成 records否则本地 Neo4j 会被内存撑爆。5.3 问答系统匹配的坑坑四jieba 把“百万医疗险”切碎实体识别失败。现象输入“百万医疗险有哪些产品”_match_entity没有命中 category因为分词结果是“百万/医疗险”。原因自定义词典没加载或者词典文件编码不对。解决确保insurance_words.txt是 UTF-8 无 BOM每行一个词更保险是在脚本里显式加词jieba.add_word(百万医疗险, freq10000) jieba.add_word(免赔额, freq10000)坑五问题命中意图但生成的 Cypher 返回空。现象输入“平安e生保2024的免赔额是多少”命中 attribute但返回空。原因图谱模型里免赔额没有存在 Product 节点属性上而是存在 Coverage 节点的属性里或者根本就没存。这个坑的本质是本体设计和问答系统的字段命名没有对齐。解决先手动执行一句 Cypher 看数据真实结构MATCH (p:Product {name: 平安e生保2024}) RETURN p;确认节点上有哪些属性再回到attribute_kws把映射补齐。这五个坑多数不是技术复杂度问题而是“数据和代码各自为政”。图数据库建模的每个字段都会在问答层暴露出来建图前多花一小时理字段问答时就能少熬夜。6. 给简易问答加一层验证用测试问题和向量检索打通 RAG模板问答上线后第一件事不是继续加模板而是先做一个“测试集体检脚本”把系统的覆盖度摊开看。建议准备 20 到 30 个实际会问的问题跑一遍自动化检查。test_questions [ 百万医疗险有哪些产品, 平安e生保2024免赔额是多少, 超越保2020包含哪些责任, 哪种产品包含住院医疗责任, ] for q in test_questions: ans engine.answer(q) if isinstance(ans, str): print(q, , ans) else: print(q, , len(ans), 条结果)这个脚本的价值不是算准确率而是让你一次性看出三类情况返回空、返回多个异常实体、回答不符合预期。如果是实体没对上优先补词典和别名表如果是模板没覆盖再加意图分支。把这个测试集当成图谱质量的体检单问答系统的演进就变成了持续维护问题集。如果想把简易问答升级成能自然语言对话的 RAG不用急着换架构。常见做法是保留当前图谱查询结果把它作为上下文拼给 LLM。必要时可以加一层向量检索用sentence-transformers的paraphrase-multilingual-MiniLM-L12-v2把问题和候选实体名编码成向量算余弦相似度取 topk再把这些候选传给模板问答。这个小改动只加一个依赖却能明显改善实体别名导致的漏召回。我自己做这个方向时最大的教训是数据清洗和实体对齐花的时间远远超过写问答逻辑。别把根因甩给图数据库或 NLP模板问答能跑通RAG 才有底气图谱里没有的正确答案模型再大也答不出来。希望这些参数、代码和踩坑记录能帮你在保险知识图谱方向少走几步弯路。本文还有配套的精品资源点击获取