企业官网建设流程全解析

简介：本资源是一套基于开源保险产品数据构建的保险知识图谱及简易问答系统完整实现，面向计算机、人工智能、信息管理等专业在校学生、教师及初学者，适用于课程设计、毕业设计、项目实训与知识图谱入门实践。包内含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 定义三类核心节点及关键属性

节点设计遵循“可查询、可解释、可扩展”原则：

• :Product：product_id（主键，索引）、name、coverage_type（枚举：重疾/医疗/意外）、premium_range（JSON字符串存区间）、effective_date
• :Disease：code（ICD-10标准码，唯一索引）、name、severity（1-5级）、is_covered_by_default（布尔，是否默认承保）
• :Clause：clause_id（自增ID）、content（文本，全文索引）、type（枚举：exclusion/waiting_period/deductible）、source_product（关联product_id）

提示：Disease节点必须用ICD-10编码而非中文名作主键，避免“甲状腺癌”“甲状腺恶性肿瘤”等同义词歧义；Clause节点不直接存长文本，而是将条款切分为原子单元（如每条免责独立成节点），便于精准匹配。

2.2 构建四类业务强相关关系

关系设计体现保险逻辑链条，而非简单“包含”：

• (:Product)-[:COVERS]->(:Disease)：带属性coverage_status（full/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_coverage"）

2.3 用Py2neo批量导入数据的最小可行脚本

from py2neo import Graph, Node, Relationship, NodeMatcher import pandas as pd # 连接本地Neo4j（默认http://localhost:7474，auth=("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_id=row["product_id"], name=row["product_name"], coverage_type=row["coverage_type"], premium_range=str({"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", code=row["icd10_code"], name=row["disease_name"], severity=int(row["severity_level"]), is_covered_by_default=bool(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", code=disease_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批量导入（UNWIND+MERGE），避免Python层循环性能瓶颈。

3. 实现简易问答系统：用规则+关键词匹配替代复杂NLU，精准响应保险术语查询

保险问答场景高度结构化：用户提问集中于“XX病是否赔？”“A产品和B产品在YY责任上有什么区别？”“等待期多久？”。这类问题无需BERT微调或大模型生成，用确定性规则+关键词映射+图查询即可达到90%+准确率，且响应快（<200ms）、可解释（返回具体条款ID）。本系统采用三层处理架构：意图识别 → 实体链接 → 图查询生成，全部基于Python标准库与Neo4j驱动实现，不依赖外部API。

3.1 意图识别：用正则模板匹配三类高频问题

定义三个正则模式覆盖80%以上业务提问，避免引入jieba等分词库带来的歧义（如“甲状腺”可能被切为“甲状”+“腺”，而保险中必须整体识别）：

• 赔付类：r"(?P<disease>[^\s，。？！]+)(?:是否|能不能|可不可以|赔不赔|给不给)赔"→ 提取disease组
• 对比类：r"(?P<prod_a>[^\s，。？！]+)和(?P<prod_b>[^\s，。？！]+)(?:在|关于|对于)(?P<dimension>[^\s，。？！]+)(?:有|存在|的)什么区别"→ 提取prod_a,prod_b,dimension
• 条款类：r"(?P<product>[^\s，。？！]+)(?:的|关于|针对)(?P<clause_type>等待期|免赔额|除外责任|轻症定义)(?:是|为|多少|怎么规定)"→ 提取product,clause_type

import re INTENT_PATTERNS = { "coverage": re.compile(r"(?P<disease>[^\s，。？！]+)(?:是否|能不能|可不可以|赔不赔|给不给)赔"), "comparison": re.compile(r"(?P<prod_a>[^\s，。？！]+)和(?P<prod_b>[^\s，。？！]+)(?:在|关于|对于)(?P<dimension>[^\s，。？！]+)(?:有|存在|的)什么区别"), "clause": re.compile(r"(?P<product>[^\s，。？！]+)(?:的|关于|针对)(?P<clause_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 “甲状腺功能亢进症”）。我们构建两级映射：

1. 同义词词典：synonym_dict.json存手动维护的映射，如{"甲亢": "甲状腺功能亢进症", "e生保": "e生保长期医疗险"}；
2. 编辑距离回退：当词典未命中时，计算输入词与所有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", encoding="utf-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, key=lambda x: x[1])[0] # 返回最匹配的ICD-10码 return None

3.2.1 同义词词典的维护策略

• 初始版本由核保岗提供200个高频简称/俗称（如“达尔文”→“达尔文3号重疾险”）；
• 上线后记录用户提问中未匹配成功的Top 10疾病名/产品名，每周人工确认后加入词典；
• 禁止自动聚类生成同义词，避免“心梗”和“心肌梗死”被错误合并（临床定义不同，保险责任可能不同）。

3.3 图查询生成：将自然语言转为Cypher，直连Neo4j返回结构化答案

意图与实体确定后，生成针对性Cypher查询。以“甲状腺癌是否赔？”为例：

1. 意图coverage+ 疾病实体C73→ 查询所有COVERS该疾病的Product节点；
2. 过滤coverage_status != "excluded"；
3. 关联HAS_CLAUSE关系获取对应条款内容；
4. 返回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 # 明确指定neo4j==5.20.0, pandas==2.0.3等版本，避免依赖冲突 ├── config/ │ ├── synonym_dict.json # 同义词映射表，UTF-8编码，支持中文键 │ └── neo4j_config.py # 数据库连接参数，含默认本地地址与测试账号 ├── data/ │ ├── raw/ # 原始CSV（products.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服务端需neo4j==5.20.0），否则graph.run()报ProtocolError；
• pandas锁定==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_list：ICD-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 reason

5.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_answer

5.2.1 归因结果的呈现规范

• 归因段落以【归因】开头，与基础答案用空行分隔，避免混淆；
• 仅对excluded状态返回归因，partial状态需人工判断条件，不自动归因；
• 限制最多返回3个产品的归因（excluded_products[:3]），防止长尾产品拖慢响应；
• 若归因查询超时（设置graph.run(..., timeout=3)），静默忽略，不中断基础答案返回。

本文还有配套的精品资源，点击获取