简介:这份资源面向具备一定Python基础、希望入门自然语言处理与知识图谱构建的开发者与学习者,围绕「文本转知识图谱」这一典型AI应用场景,提供可运行的完整项目代码。包内共63个文件,以41个JavaScript前端脚本、5个XML配置、4个Python源码及3个编译缓存文件为主,另含CSS、HTML与说明文本,压缩包约605.62MB,其中包含分词与图数据等依赖资源。项目覆盖文本预处理、命名实体识别、关系抽取、图结构构建与可视化等关键环节,并配有测试样例用于验证各阶段输出。目前已有597人学习下载。读者可借助其中的Python脚本与前端展示页面,理解从原始文本到实体关系网络的完整链路,并参考其模块划分与封装思路,快速搭建自己的知识图谱实验环境。
1. 从一段合同文本到可查询图谱:Python 文本转知识图谱到底在做什么
手里有一批合同、工单、论文摘要或者设备手册,老板说「做成知识图谱」,很多人第一反应是打开 Neo4j 建几个节点,结果发现连实体都抽不准,图谱建出来是一张废网。文本转化知识图谱这件事,本质是把非结构化文本里的实体、关系、属性抽出来,按本体约束组织成图结构,再写进图数据库供查询和推理。它解决的是「文本查不动、关系看不见」的问题,适合手上有垂直领域语料、需要做语义检索或关系推理的工程师。Python 在这条链路里承担的是胶水角色:调 NLP 模型抽实体、写规则做关系对齐、用驱动把三元组灌进 Neo4j。下面按「先立本体、再抽三元组、后入库验证」的顺序,把这条链路拆成能直接抄的步骤。
2. 先定本体再动手:文本转知识图谱的 schema 设计与选型
2.1 为什么不能跳过本体直接抽实体
本体建模是知识图谱的骨架,它规定了「有哪些类型的实体、哪些类型的关系、关系的头尾类型是什么」。跳过本体直接让模型抽实体,抽出来的东西类型混乱,同一个「华为」可能被标成公司、组织、品牌三种标签,后面做关系推理时对不上。工业场景下的知识图谱设计尤其吃这一套,设备、部件、故障、工单之间的关系必须提前约束死,否则图谱规模一上来就是一团乱麻。
本体不用一上来就搞得很重,先用一张表把核心概念列清楚即可。下面是我做垂直领域图谱时常用的最小本体表,字段含义:实体类型、中文名、典型属性、可参与的关系。
| 实体类型 | 中文名 | 典型属性 | 可参与关系 |
|---|---|---|---|
| Company | 公司 | 名称、统一社会信用代码 | 签订合同、供应产品 |
| Contract | 合同 | 合同编号、金额、签署日期 | 由公司签订、涉及产品 |
| Product | 产品 | 型号、规格 | 被合同涉及、由公司供应 |
| Person | 人员 | 姓名、职务 | 代表公司签署 |
这张表就是后续所有抽取和校验的依据。实体类型控制在 5 到 10 个,关系类型控制在 10 到 20 个,超过这个量级说明领域切得太宽,先拆子领域。
2.2 用 Python 把本体写成可校验的配置
本体不要只写在文档里,要落成代码里的常量,抽取阶段直接引用,避免字符串硬编码写错。下面这段配置用 dataclass 定义实体和关系,附带一个校验函数,抽取结果入库前先过一遍。
from dataclasses import dataclass, field from typing import List, Dict @dataclass class EntityType: name: str # 英文类型名,入库用 cn_name: str # 中文名,展示用 attributes: List[str] = field(default_factory=list) @dataclass class RelationType: name: str # 关系英文名 head: str # 头实体类型 tail: str # 尾实体类型 ENTITY_TYPES = { "Company": EntityType("Company", "公司", ["name", "credit_code"]), "Contract": EntityType("Contract", "合同", ["contract_no", "amount", "sign_date"]), "Product": EntityType("Product", "产品", ["model", "spec"]), "Person": EntityType("Person", "人员", ["name", "title"]), } RELATION_TYPES = { "SIGNS": RelationType("SIGNS", "Company", "Contract"), "SUPPLIES": RelationType("SUPPLIES", "Company", "Product"), "INVOLVES": RelationType("INVOLVES", "Contract", "Product"), "REPRESENTS": RelationType("REPRESENTS", "Person", "Company"), } def validate_triple(head_type: str, rel: str, tail_type: str) -> bool: """校验三元组是否符合本体约束,不符合的直接丢弃""" if rel not in RELATION_TYPES: return False rt = RELATION_TYPES[rel] return rt.head == head_type and rt.tail == tail_type逻辑说明:ENTITY_TYPES和RELATION_TYPES是全局唯一事实来源,抽取模块、入库模块都从这里取类型名。validate_triple在写库前调用,头尾类型对不上就丢弃,这是防止图谱被脏数据污染的第一道闸门。参数上,head和tail必须和ENTITY_TYPES的 key 完全一致,大小写敏感,建议统一用大驼峰。
提示:本体一旦定下来,中途改类型名会导致已有图谱数据对不上,改之前先想清楚。宁可前期多花半天讨论,也别入库后再返工。
3. 三元组抽取:从规则到模型,Python 怎么把句子拆成关系
3.1 规则抽取打底:正则和依存句法能覆盖多少
垂直领域文本句式相对固定,规则抽取的准确率往往比通用模型还高。合同类文本里「甲方:XX公司」这种模式,一条正则就能拿到实体。下面这段代码用正则抽合同编号和金额,再用 jieba 分词加词性标注辅助定位公司名。
import re import jieba.posseg as pseg CONTRACT_NO_PATTERN = re.compile(r"合同编号[::]\s*([A-Z0-9\-]+)") AMOUNT_PATTERN = re.compile(r"金额[::]\s*([0-9,]+\.?\d*)\s*元") def extract_by_rule(text: str) -> dict: result = {"contract_no": None, "amount": None, "companies": []} m = CONTRACT_NO_PATTERN.search(text) if m: result["contract_no"] = m.group(1) m = AMOUNT_PATTERN.search(text) if m: # 去掉千分位逗号,转成浮点 result["amount"] = float(m.group(1).replace(",", "")) # 用词性标注找机构名,nt 是 jieba 的机构名标记 for word, flag in pseg.cut(text): if flag == "nt" and len(word) >= 4: result["companies"].append(word) return result逻辑说明:CONTRACT_NO_PATTERN和AMOUNT_PATTERN用非贪婪匹配抓冒号后的值,\s*兼容中英文冒号后的空格。pseg.cut返回词和词性,nt标记机构名,长度过滤掉「公司」这种泛称。参数上,正则里的[A-Z0-9\-]按你实际合同编号格式调整,如果编号含中文要加\u4e00-\u9fa5。
规则抽取的边界很清楚:句式一变就失效。所以规则只用来抽高置信度的结构化字段,实体和关系的召回靠模型补。
3.2 模型抽取:用 spaCy 或 LLM 做关系分类
通用关系抽取现在两条路:一是用 spaCy 训练一个关系分类器,二是调大模型做 few-shot 抽取。前者可控、离线、成本低,后者上手快但依赖接口。下面给一个 spaCy 关系分类的训练数据格式和推理代码,适合有标注数据的团队。
import spacy from spacy.training import Example # 关系分类的训练样本格式:文本 + 实体对 + 关系标签 TRAIN_DATA = [ ("华为技术有限公司签订了合同HT2023001", { "entities": [(0, 8, "Company"), (12, 21, "Contract")], "relations": [(0, 8, 12, 21, "SIGNS")] }), ] nlp = spacy.blank("zh") ner = nlp.add_pipe("ner") rel = nlp.add_pipe("relation_extractor") # 需安装 spacy-relation-extractor for text, annot in TRAIN_DATA: doc = nlp.make_doc(text) example = Example.from_dict(doc, annot) nlp.update([example])逻辑说明:entities里是实体在文本中的字符起止位置和类型,relations是头实体起止、尾实体起止加关系名。spacy-relation-extractor是社区组件,装之前确认版本和 spaCy 主版本匹配。训练数据至少准备 200 条以上,关系类型均衡,否则分类器会偏向多数类。
如果没有标注数据,用大模型做 few-shot 是更现实的选择。把本体里的关系类型和几个示例拼进 prompt,让模型输出 JSON 格式的三元组,再用validate_triple过滤。这条路的问题是输出不稳定,需要加重试和格式校验。
3.3 实体对齐:同一个公司三种写法怎么合并
文本里「华为」「华为公司」「华为技术有限公司」指的是同一个实体,不合并的话图谱里会出现三个节点,关系全散开。实体对齐常用做法是归一化加相似度:先做规则归一(去后缀、统一全半角),再用编辑距离或向量相似度兜底。
import re from difflib import SequenceMatcher SUFFIXES = ["有限公司", "股份有限公司", "公司", "集团"] def normalize_name(name: str) -> str: name = name.strip().replace("(", "(").replace(")", ")") for suf in SUFFIXES: if name.endswith(suf) and len(name) > len(suf): name = name[: -len(suf)] break return name def is_same_entity(a: str, b: str, threshold: float = 0.85) -> bool: na, nb = normalize_name(a), normalize_name(b) if na == nb: return True return SequenceMatcher(None, na, nb).ratio() >= threshold逻辑说明:normalize_name去掉公司后缀,让「华为技术有限公司」和「华为公司」都变成「华为技术」。is_same_entity先比归一化后的字符串,不等再用SequenceMatcher算相似度,阈值 0.85 是经验值,低于这个值误合并风险明显上升。参数上,SUFFIXES按你的领域补充,医疗领域要加「医院」「诊所」,工业领域加「厂」「车间」。
注意:实体对齐阈值调低会误合并,调高会漏合并。建议先在一个小样本上人工核对,找到误合并和漏合并的平衡点再全量跑。
4. 入库 Neo4j:Python 驱动批量写入与索引调优
4.1 用 neo4j 驱动建约束和批量写入
三元组抽完,下一步是写进 Neo4j。不要一条一条CREATE,几千条以上必须用UNWIND批量写,否则性能差一个数量级。先建唯一约束,保证同一实体不会重复创建。
from neo4j import GraphDatabase driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password")) def create_constraints(tx): # 给每类实体的唯一标识建约束,name 或 contract_no tx.run("CREATE CONSTRAINT company_name IF NOT EXISTS " "FOR (c:Company) REQUIRE c.name IS UNIQUE") tx.run("CREATE CONSTRAINT contract_no IF NOT EXISTS " "FOR (c:Contract) REQUIRE c.contract_no IS UNIQUE") def batch_write(tx, triples): # triples: [{"head": "华为", "head_type": "Company", # "rel": "SIGNS", "tail": "HT2023001", "tail_type": "Contract"}] query = """ UNWIND $rows AS row MERGE (h:Company {name: row.head}) MERGE (t:Contract {contract_no: row.tail}) MERGE (h)-[r:SIGNS]->(t) SET r.source = row.source """ tx.run(query, rows=triples) with driver.session() as session: session.execute_write(create_constraints) session.execute_write(batch_write, triples)逻辑说明:CREATE CONSTRAINT建唯一约束,MERGE依赖约束才能高效判断节点是否存在,没约束的MERGE会全表扫描。UNWIND $rows把列表展开成多行,一次网络往返写一批,建议每批 1000 到 5000 条。参数上,auth里的密码换成你实际设置的,bolt://默认端口 7687。
不同实体类型要写不同标签,上面示例只处理了 Company 和 Contract。实际项目里按head_type和tail_type分组,每组一条 Cypher,或者用 APOC 的动态标签。动态标签写法:
query = """ UNWIND $rows AS row CALL apoc.merge.node([row.head_type], {name: row.head}) YIELD node AS h CALL apoc.merge.node([row.tail_type], {contract_no: row.tail}) YIELD node AS t CALL apoc.merge.relationship(h, row.rel, {}, {}, t) YIELD rel RETURN count(rel) """apoc.merge.node第一个参数是标签列表,第二个是属性 map。用 APOC 前确认插件已装,Neo4j 5.x 之后 APOC 要单独下载对应版本。
4.2 索引和查询性能:哪些属性必须建索引
图谱查询慢,九成是没建索引。除了唯一约束自带的索引,经常用来做查询入口的属性都要单独建。下面这张表列出常见实体和必须建索引的属性。
| 实体类型 | 必建索引属性 | 原因 |
|---|---|---|
| Company | name | 按公司名查合同、产品 |
| Contract | contract_no, sign_date | 按编号精确查、按日期范围查 |
| Product | model | 按型号查供应关系 |
| Person | name | 按人名查代表关系 |
建索引语句:
tx.run("CREATE INDEX contract_date IF NOT EXISTS FOR (c:Contract) ON (c.sign_date)")范围查询(比如查某时间段签的合同)走sign_date索引,等值查询走唯一约束。索引不是越多越好,每个索引都会拖慢写入,只给高频查询属性建。
提示:批量写入前先建约束和索引,写入过程中不要频繁改 schema,否则会触发索引重建,大批量数据下很慢。
5. 避坑与排查:文本转知识图谱最常见的 5 个翻车点
5.1 实体类型标错导致关系全丢
现象:入库后发现图谱里节点不少,但关系边几乎为零。原因:抽取阶段实体类型标错,比如把公司标成ORG,而本体里定义的是Company,validate_triple校验时头类型对不上,三元组全被丢弃。解决:在抽取和校验之间加一层类型映射,把模型输出的标签统一映射到本体类型名,映射表单独维护,映射不上的记录日志人工确认。
5.2 批量写入时内存溢出
现象:跑几万条三元组时 Python 进程被 OOM kill。原因:一次性把所有三元组读进内存再传给UNWIND,数据量大时内存扛不住。解决:分批读、分批写,每批 1000 到 5000 条,用生成器逐批产出。下面是一个分批写入的骨架:
def chunked(iterable, size=2000): batch = [] for item in iterable: batch.append(item) if len(batch) >= size: yield batch batch = [] if batch: yield batch for batch in chunked(all_triples): session.execute_write(batch_write, batch)5.3 实体对齐把不同公司合并成一个
现象:图谱里「华为技术」和「华为投资」被合并成同一个节点。原因:归一化去后缀后两者都变成「华为」,相似度直接判等。解决:归一化只去通用后缀,保留能区分主体的关键词;相似度阈值不要低于 0.85;对高频实体名建白名单,白名单内不走模糊匹配。
5.4 Neo4j 连接超时或认证失败
现象:neo4j.exceptions.ServiceUnavailable或AuthError。原因:bolt 端口没开、密码错、或者 Neo4j 服务没启动。解决:先telnet localhost 7687确认端口通,再确认neo4j.conf里dbms.default_listen_address配置,最后核对密码。Docker 部署的话确认端口映射-p 7687:7687。
5.5 中文分词把实体切碎
现象:公司名被 jieba 切成「华为」「技术」「有限」「公司」四个词,实体抽不出来。原因:通用词典不含领域专有名词。解决:加载自定义词典,把领域实体词提前加进去。
import jieba jieba.load_userdict("domain_dict.txt") # 每行一个词,可带词频和词性domain_dict.txt格式:华为技术有限公司 100 nt,词频给高一点保证不被切碎,词性标nt让后续词性过滤能命中。
6. 让图谱可验证:用 Cypher 反查和抽样评估抽取质量
图谱建完不是终点,得能验证抽得对不对。最直接的办法是用 Cypher 反查,看关系是否符合预期。比如查所有公司签订的合同金额总和:
query = """ MATCH (c:Company)-[:SIGNS]->(ct:Contract) RETURN c.name AS company, count(ct) AS contract_count, sum(ct.amount) AS total ORDER BY total DESC LIMIT 10 """ with driver.session() as session: for record in session.run(query): print(record["company"], record["contract_count"], record["total"])如果结果里出现明显不该有的公司,或者金额为 null 的比例很高,说明抽取或入库有问题。抽样评估更严谨的做法是人工标注 100 条文本的三元组,和系统输出比对,算准确率和召回率。准确率低于 0.8 先别上生产,回去调抽取规则或模型。
进阶一点,可以用图谱做一致性校验:本体里定义「合同必须有签订方」,那就查有没有孤立合同节点。
query = """ MATCH (ct:Contract) WHERE NOT (ct)<-[:SIGNS]-() RETURN ct.contract_no """查出来的就是缺签订方的合同,要么是抽取漏了,要么是原文本身没写。这个查询我一般做成定时任务,每天跑一次,结果推到告警群。
我自己踩过最深的坑是本体没定就开抽,抽了两万条三元组才发现类型对不上,全部返工。后来养成习惯:任何文本转图谱的项目,第一周只做本体和 100 条样本的抽取验证,验证通过再全量跑。这个节奏慢,但省下的返工时间远超前期投入。希望帮到你。
本文还有配套的精品资源,点击获取