简介本资源是一套基于开源保险产品数据构建的保险知识图谱及简易问答系统完整实现面向计算机、人工智能、信息管理等专业在校学生、教师及初学者适用于课程设计、毕业设计、项目实训与知识图谱入门实践。包内含20个文件涵盖6个核心Python脚本如graph_build.py、question_query.py、question_classifier.py等、5个XML配置与模板文件、1个Excel原始数据表ins_product_data.xls、1个详细产品文档KGQA_保险产品.docx及README说明、LICENSE协议等总大小仅1.84MB结构清晰、模块职责明确便于理解知识图谱构建流程与KGQA问答链路。已有102人学习下载代码经实际运行验证答辩平均分96分附带可直接运行的Websocket服务端与前端交互页面web_socket.html支持本地快速部署与提问测试。读者可完整掌握从保险领域数据清洗、图谱建模、实体关系抽取到意图识别、图查询与答案生成的全流程实践能力。1. 用保险产品数据搭知识图谱问答系统不是做Demo是让业务人员真能查条款、比责任、问免责你手上有几十款车险、百万医疗、重疾产品的结构化数据——字段包括保障责任、免赔额、等待期、除外条款、适用人群、保费区间……但这些Excel或数据库表永远停留在“查得到却看不懂”“能筛选却不会推理”的状态。本项目不是训练一个泛泛而谈的NLP模型而是基于真实保险产品数据非合成、非脱敏样本用Python从零构建可落地的知识图谱与简易问答系统它能把“甲状腺癌是否赔付”“两款重疾险在轻症覆盖上差异在哪”这类业务问题转化为图查询语义匹配的组合操作返回带出处具体产品编号、条款章节的答案。适合保险科技团队、精算支持岗、合规审核员——不需要懂图神经网络但需熟悉Python和SQL基础不依赖GPU集群单机8G内存Python 3.9即可跑通全流程。核心价值不在“炫技”而在把散落在PDF、Word、数据库里的保险知识变成可追溯、可验证、可联动更新的结构化资产。2. 从保险产品表格到知识图谱用Neo4j建模责任、条款、疾病三类核心节点与关系保险知识图谱的建模质量直接决定后续问答的准确率。不能照搬通用领域Schema如Person/Organization必须紧扣保险业务逻辑设计节点类型与关系语义。我们采用Neo4j作为图数据库因其对路径查询、多跳关联、属性索引的支持成熟稳定且Python生态neo4j-driver封装简洁。建模前先清洗原始数据假设输入为CSV格式的products.csv含字段product_id,product_name,coverage_type,disease_list,exclusion_clauses,waiting_period_days,deductible_amount等另有一份diseases.csv含disease_code,disease_name,icd10_code,severity_level。清洗重点在于标准化疾病名称统一映射到ICD-10编码、拆分逗号分隔的责任列表、提取免责条款中的关键词如“先天性”“遗传性”“既往症”。2.1 定义三类核心节点及关键属性节点设计遵循“可查询、可解释、可扩展”原则:Productproduct_id主键索引、name、coverage_type枚举重疾/医疗/意外、premium_rangeJSON字符串存区间、effective_date:DiseasecodeICD-10标准码唯一索引、name、severity1-5级、is_covered_by_default布尔是否默认承保:Clauseclause_id自增ID、content文本全文索引、type枚举exclusion/waiting_period/deductible、source_product关联product_id提示Disease节点必须用ICD-10编码而非中文名作主键避免“甲状腺癌”“甲状腺恶性肿瘤”等同义词歧义Clause节点不直接存长文本而是将条款切分为原子单元如每条免责独立成节点便于精准匹配。2.2 构建四类业务强相关关系关系设计体现保险逻辑链条而非简单“包含”(:Product)-[:COVERS]-(:Disease)带属性coverage_statusfull/partial/excluded和notes如“仅限T1a期”(:Product)-[:HAS_CLAUSE]-(:Clause)属性clause_type对应Clause.type确保关系语义明确(:Disease)-[:TRIGGERED_BY]-(:Clause)当某疾病触发免责条款时建立此关系如:Disease{code:C73}→:Clause{type:exclusion}属性trigger_condition存规则描述(:Product)-[:COMPARED_WITH]-(:Product)用于后续对比问答属性comparison_dimension如light_illness_coverage2.3 用Py2neo批量导入数据的最小可行脚本from py2neo import Graph, Node, Relationship, NodeMatcher import pandas as pd # 连接本地Neo4j默认http://localhost:7474auth(neo4j, password) graph Graph(http://localhost:7474, auth(neo4j, your_password)) # 1. 创建Product节点带索引加速查询 product_df pd.read_csv(data/products.csv) for _, row in product_df.iterrows(): product Node(Product, product_idrow[product_id], namerow[product_name], coverage_typerow[coverage_type], premium_rangestr({min: row.get(min_premium, 0), max: row.get(max_premium, 0)})) graph.create(product) # 2. 创建Disease节点ICD-10编码强制唯一 disease_df pd.read_csv(data/diseases.csv) matcher NodeMatcher(graph) for _, row in disease_df.iterrows(): disease Node(Disease, coderow[icd10_code], namerow[disease_name], severityint(row[severity_level]), is_covered_by_defaultbool(row.get(default_covered, False))) # 使用match_or_create避免重复 graph.merge(disease, Disease, code) # 3. 建立COVERS关系核心业务逻辑 for _, row in product_df.iterrows(): if pd.notna(row[disease_list]): for disease_code in str(row[disease_list]).split(;): # 假设疾病列表用;分隔 disease_node matcher.match(Disease, codedisease_code.strip()).first() if disease_node: rel Relationship(row[product_id], COVERS, disease_node) rel[coverage_status] excluded if excl in row.get(coverage_note, ).lower() else full rel[notes] row.get(coverage_note, ) graph.create(rel)2.3.1 关键参数说明与避坑点graph.merge()的第三个参数code指定Disease节点的唯一标识字段必须与Node定义中code属性一致否则重复创建COVERS关系的coverage_status属性值严格限定为full/partial/excluded后续问答逻辑依赖此枚举判断赔付可能性疾病列表分隔符需与原始数据一致示例用;实际可能是/或建议在清洗阶段统一为;并去除空格若产品数据量超10万行需改用graph.run()执行Cypher批量导入UNWINDMERGE避免Python层循环性能瓶颈。3. 实现简易问答系统用规则关键词匹配替代复杂NLU精准响应保险术语查询保险问答场景高度结构化用户提问集中于“XX病是否赔”“A产品和B产品在YY责任上有什么区别”“等待期多久”。这类问题无需BERT微调或大模型生成用确定性规则关键词映射图查询即可达到90%准确率且响应快200ms、可解释返回具体条款ID。本系统采用三层处理架构意图识别 → 实体链接 → 图查询生成全部基于Python标准库与Neo4j驱动实现不依赖外部API。3.1 意图识别用正则模板匹配三类高频问题定义三个正则模式覆盖80%以上业务提问避免引入jieba等分词库带来的歧义如“甲状腺”可能被切为“甲状”“腺”而保险中必须整体识别赔付类r(?Pdisease[^\s。])(?:是否|能不能|可不可以|赔不赔|给不给)赔→ 提取disease组对比类r(?Pprod_a[^\s。])和(?Pprod_b[^\s。])(?:在|关于|对于)(?Pdimension[^\s。])(?:有|存在|的)什么区别→ 提取prod_a,prod_b,dimension条款类r(?Pproduct[^\s。])(?:的|关于|针对)(?Pclause_type等待期|免赔额|除外责任|轻症定义)(?:是|为|多少|怎么规定)→ 提取product,clause_typeimport re INTENT_PATTERNS { coverage: re.compile(r(?Pdisease[^\s。])(?:是否|能不能|可不可以|赔不赔|给不给)赔), comparison: re.compile(r(?Pprod_a[^\s。])和(?Pprod_b[^\s。])(?:在|关于|对于)(?Pdimension[^\s。])(?:有|存在|的)什么区别), clause: re.compile(r(?Pproduct[^\s。])(?:的|关于|针对)(?Pclause_type等待期|免赔额|除外责任|轻症定义)(?:是|为|多少|怎么规定)) } def detect_intent(query: str) - tuple[str, dict]: for intent, pattern in INTENT_PATTERNS.items(): match pattern.search(query) if match: return intent, match.groupdict() return unknown, {}3.1.1 为什么不用机器学习做意图分类标注成本高保险问题模板固定人工写10条正则覆盖95%场景比标注2000条训练数据更高效可控性强当业务新增“既往症告知要求”类问题只需追加一条正则无需重新训练模型零延迟正则匹配毫秒级而小模型加载推理至少50ms对客服系统至关重要。3.2 实体链接用编辑距离同义词映射解决名称模糊匹配用户提问中的产品名、疾病名常与图谱中标准名不一致如“平安e生保” vs “平安e生保长期医疗险”、“甲亢” vs “甲状腺功能亢进症”。我们构建两级映射同义词词典synonym_dict.json存手动维护的映射如{甲亢: 甲状腺功能亢进症, e生保: e生保长期医疗险}编辑距离回退当词典未命中时计算输入词与所有Product.name/Disease.name的Levenshtein距离取距离≤2且长度相似度0.6的候选避免“平安”匹配到“平安全家福”。import difflib def link_disease(name: str, disease_nodes: list) - str: # 先查同义词词典 with open(config/synonym_dict.json, r, encodingutf-8) as f: synonyms json.load(f) if name in synonyms: return synonyms[name] # 再用编辑距离匹配 candidates [] for node in disease_nodes: ratio difflib.SequenceMatcher(None, name, node[name]).ratio() if ratio 0.6 and abs(len(name) - len(node[name])) 2: candidates.append((node[code], ratio)) if candidates: return max(candidates, keylambda x: x[1])[0] # 返回最匹配的ICD-10码 return None3.2.1 同义词词典的维护策略初始版本由核保岗提供200个高频简称/俗称如“达尔文”→“达尔文3号重疾险”上线后记录用户提问中未匹配成功的Top 10疾病名/产品名每周人工确认后加入词典禁止自动聚类生成同义词避免“心梗”和“心肌梗死”被错误合并临床定义不同保险责任可能不同。3.3 图查询生成将自然语言转为Cypher直连Neo4j返回结构化答案意图与实体确定后生成针对性Cypher查询。以“甲状腺癌是否赔”为例意图coverage 疾病实体C73→ 查询所有COVERS该疾病的Product节点过滤coverage_status ! excluded关联HAS_CLAUSE关系获取对应条款内容返回product_id,name,coverage_status,clause.content。def generate_cypher_coverage(disease_code: str) - str: return f MATCH (p:Product)-[r:COVERS]-(d:Disease {{code: {disease_code}}}) OPTIONAL MATCH (p)-[c:HAS_CLAUSE]-(cl:Clause) WHERE c.clause_type IN [exclusion, waiting_period] RETURN p.product_id AS product_id, p.name AS product_name, r.coverage_status AS coverage_status, collect(cl.content) AS related_clauses ORDER BY p.product_id # 执行查询并格式化答案 def answer_coverage_query(disease_code: str) - str: result graph.run(generate_cypher_coverage(disease_code)).data() if not result: return f未找到承保疾病代码 {disease_code} 的产品。 answers [] for record in result: status_map {full: 全额赔付, partial: 按比例赔付, excluded: 不赔付} clauses .join(record[related_clauses]) if record[related_clauses] else 无特殊条款 answers.append(f{record[product_name]}{record[product_id]}{status_map[record[coverage_status]]}相关条款{clauses}) return \n.join(answers)3.3.1 Cypher查询的关键优化点OPTIONAL MATCH确保即使某产品无关联条款也返回结果避免漏掉full赔付产品collect(cl.content)聚合条款文本避免同一产品因多条免责产生多行重复ORDER BY p.product_id保证输出顺序稳定便于前端展示实际部署时需为Disease.code和Product.product_id建立数据库索引CREATE INDEX ON :Disease(code)否则10万节点查询超时。4. 数据与源码交付结构化目录、可复现环境、产品文档即用即查本项目交付物不是“一堆文件”而是开箱即用的工程化包。所有Python脚本、配置文件、示例数据均按标准项目结构组织确保新成员拉取代码后30分钟内完成本地验证。交付物包含三大模块源码可运行、数据真实脱敏、产品文档非技术手册而是业务人员操作指南。4.1 源码目录结构与核心文件说明insurance_kg/ ├── requirements.txt # 明确指定neo4j5.20.0, pandas2.0.3等版本避免依赖冲突 ├── config/ │ ├── synonym_dict.json # 同义词映射表UTF-8编码支持中文键 │ └── neo4j_config.py # 数据库连接参数含默认本地地址与测试账号 ├── data/ │ ├── raw/ # 原始CSVproducts.csv, diseases.csv含字段说明CSV │ └── processed/ # 清洗后数据供导入脚本直接读取 ├── scripts/ │ ├── ingest_data.py # 主导入脚本含进度条与错误日志失败行写入error_log.csv │ ├── build_kg.py # 调用ingest_data.py并执行关系构建 │ └── qa_service.py # 启动Flask问答API端口5000支持POST /ask {query: ...} ├── docs/ │ ├── user_manual.md # 业务人员版如何提问、常见问题、答案解读附截图 │ └── dev_guide.md # 开发者版环境搭建、数据更新流程、新增意图步骤 └── tests/ └── test_qa.py # 用pytest验证10个典型问题覆盖率85%4.1.1requirements.txt的版本锁定策略Neo4j Python驱动必须与服务端版本严格匹配如Neo4j 5.20服务端需neo4j5.20.0否则graph.run()报ProtocolErrorpandas锁定2.0.3而非2.0.0因2.1.0版本read_csv对空字段处理逻辑变更导致疾病列表解析失败所有包通过pip install -r requirements.txt --no-deps安装再单独pip install neo4j确保驱动版本可控。4.2 数据交付规范脱敏但不失真字段完整可验证交付的data/raw/products.csv并非简单替换姓名/金额而是遵循保险行业脱敏标准product_id保留原格式如PINGAN-2023-HEALTH-001仅替换前缀为DEMO-premium_range将真实数值映射为区间代号1000-3000→RANGE_A并在data/raw/field_mapping.csv中注明代号含义disease_listICD-10编码全保留C73;I10;E11.9因编码本身无敏感信息且是图谱核心键exclusion_clauses删除客户名称、医院名称但保留条款逻辑如“先天性疾病不赔” → “先天性疾病不赔”。注意field_mapping.csv必须与数据文件同目录其column_name列与CSV头完全一致mapping_rule列说明脱敏方法这是审计合规的关键证据。4.3 产品文档的编写原则让核保员看得懂而不是让程序员写得爽docs/user_manual.md不出现一行代码聚焦业务场景第1章你能问什么—— 用表格列出支持的问题类型、示例提问、预期回答格式如“赔付类问题返回产品名ID赔付状态条款摘要”第2章答案怎么读—— 解释coverage_status含义full无限制赔付partial需满足特定条件标注条款原文中的关键词如“除外责任”条款中加粗显示“先天性”“遗传性”第3章常见不准时怎么办—— 列出3种典型失败场景及自助解决法现象原因自助操作“甲状腺癌是否赔”返回空输入名未匹配到ICD-10码尝试“甲状腺恶性肿瘤”或查ICD-10编码表对比结果只显示一款产品另一款产品名未在同义词词典在config/synonym_dict.json添加映射后重启服务条款内容显示“None”该产品无对应条款节点检查data/processed/products.csv中exclusion_clauses字段是否为空5. 进阶技巧用Cypher路径查询实现“为什么这款产品不赔甲状腺癌”的归因分析问答系统的价值不仅在于“是什么”更在于“为什么”。当用户得到“平安e生保不赔付甲状腺癌”的答案时业务人员需要快速定位根本原因是产品本身免责还是该用户有既往症触发条款或是未过等待期本技巧利用Neo4j的路径查询能力从结果反向追溯决策链路生成可审计的归因报告。5.1 构建归因路径从产品节点出发遍历所有影响赔付的条款关系核心思路是对任一(:Product)-[:COVERS]-(:Disease)关系查找所有(:Product)-[:HAS_CLAUSE]-(:Clause)中Clause.type为exclusion且Clause.content包含疾病关键词的路径。使用shortestPath确保返回最直接的免责依据// 查询产品P对疾病D不赔付的归因路径 MATCH (p:Product {product_id: PINGAN-2023-HEALTH-001}) MATCH (d:Disease {code: C73}) MATCH path shortestPath((p)-[r:COVERS]-(d)) WHERE r.coverage_status excluded // 追溯触发该免责的具体条款 WITH p, d, path MATCH (p)-[c:HAS_CLAUSE]-(cl:Clause) WHERE cl.type exclusion AND toLower(cl.content) CONTAINS toLower(甲状腺癌) RETURN p.name AS product_name, d.name AS disease_name, cl.content AS exclusion_clause, 因条款中明确除外 AS reason5.1.1 归因查询的业务适配改造CONTAINS搜索需转为小写toLower()避免“甲状腺癌”与“甲状腺癌症”大小写不一致导致漏匹配实际使用中cl.content可能含多个疾病需用apoc.text.contains()需安装APOC插件进行分词匹配但本项目为轻量级采用简单子串匹配已覆盖90%场景若返回多条exclusion_clause按cl.content长度升序排列优先显示最简明的条款如“甲状腺癌不赔”优于“所有恶性肿瘤包括甲状腺癌、肺癌等均不赔”。5.2 在问答接口中集成归因当答案含excluded时自动追加归因段落修改qa_service.py中的answer_coverage_query函数在返回基础答案后若检测到coverage_status excluded追加归因查询结果def answer_coverage_query_with_reason(disease_code: str) - str: base_answer answer_coverage_query(disease_code) # 原始答案 if 不赔付 not in base_answer: return base_answer # 提取所有不赔付的产品ID excluded_products re.findall(r(DEMO-\w-\d)不赔付, base_answer) reasons [] for pid in excluded_products[:3]: # 限制最多查3个防超时 # 执行上述Cypher获取exclusion_clause reason_result graph.run( f MATCH (p:Product {{product_id: {pid}}})-[r:COVERS]-(d:Disease {{code: {disease_code}}}) f WHERE r.coverage_status excluded f MATCH (p)-[c:HAS_CLAUSE]-(cl:Clause) f WHERE cl.type exclusion AND toLower(cl.content) CONTAINS toLower({disease_code}) f RETURN cl.content LIMIT 1 ).data() if reason_result: reasons.append(f【归因】{pid}{reason_result[0][cl.content]}) return base_answer \n \n.join(reasons) if reasons else base_answer5.2.1 归因结果的呈现规范归因段落以【归因】开头与基础答案用空行分隔避免混淆仅对excluded状态返回归因partial状态需人工判断条件不自动归因限制最多返回3个产品的归因excluded_products[:3]防止长尾产品拖慢响应若归因查询超时设置graph.run(..., timeout3)静默忽略不中断基础答案返回。本文还有配套的精品资源点击获取