简介这份资源面向计算机、人工智能、自动化等专业的学生与从业者提供一套基于 OneKE 模型构建知识图谱并搭建问答系统的完整 Python 项目源码可作为毕业设计、课程大作业或进阶练手参考。项目围绕知识图谱的 schema 设计、SPO 三元组抽取与转换、图谱导入及 RAG 问答流程展开代码经过调试测试可直接运行基础较好的读者还能在此基础上修改扩展功能。压缩包共 27 个文件约 2.87MB以 json 数据与配置、png 流程示意图、py 脚本为主另含 csv 结果、cypher 导入语句、sh 运行脚本及 md 说明文档覆盖从数据处理到图谱落地的关键环节。目前已有 450 人学习下载适合希望系统理解知识图谱构建与问答系统实现路径的读者参考借鉴。1. 从一堆PDF到能对话的知识库OneKE 到底解决了什么手里有几十份设备手册、工艺规范、故障案例老板一句“做个问答系统”多数人第一反应是上 RAG切块、向量化、塞进向量库、接个大模型。真跑起来才发现问“3号泵上次大修换了哪些密封件”这种需要跨三份文档、还要把设备-部件-工单串起来的问题纯向量检索经常召回一堆语义相近但答非所问的段落。rag瓶颈就在这——它擅长找“像”的文本不擅长回答“关系”的问题。OneKE 这类大模型知识抽取框架的价值是把非结构化文本先变成结构化的三元组存进知识图谱再让问答走图查询而不是纯向量匹配。Python 基于 OneKE 模型构建知识图谱并搭建问答系统本质是搭一条“抽取→入库→检索→生成”的流水线OneKE 负责从文本里抽实体和关系Neo4j 存图问答层用图查询加 RAG 兜底。这套方案适合手头有领域文档、又需要处理实体关系类问题的开发者新手能按步骤跑通最小闭环熟手能看清抽取精度和查询性能的边界。2. OneKE 抽取与图谱建模先想清楚 schema 再动手2.1 OneKE 是什么为什么不用正则和通用 NEROneKE 是面向知识抽取的大模型框架核心能力是给定一段文本和一套 schema输出符合 schema 的实体与关系三元组。和传统做法比差异在三点。第一传统 NER 模型要标注大量数据再训练换一个领域就得重标OneKE 走的是 schema 驱动你告诉它要抽“设备、部件、故障现象、维修动作”这几类它就能按这个结构输出冷启动成本低。第二正则表达式在格式固定的日志里好用但设备手册的表述千变万化“更换了密封圈”和“密封件已替换”要归一到同一个关系正则维护到后面就是血泪经验。第三通用 NER 模型识别的实体类型是人物、地点、组织这类通用类别工业场景里的“轴承型号”“工单编号”它根本不认。选 OneKE 的核心理由是 schema 可控。你定义什么它就抽什么抽取结果直接对应图谱的节点和边省掉一层映射转换。代价是抽取质量依赖 schema 设计得是否清晰以及文本里信息是否完整。schema 太粗抽出来的图没区分度schema 太细模型容易漏抽。2.2 定义 schema实体类型和关系类型怎么定schema 是整个系统的地基。常见做法是先拿五六份典型文档人工标一遍你希望问答系统能回答的问题涉及哪些实体和关系再归纳成类型。不要一上来就设计几十个类型先跑通再扩展。下面是一个设备运维场景的 schema 定义用 Python 字典描述后续直接喂给抽取流程# schema.py # 实体类型定义key 是类型名value 是该类型的说明和示例 ENTITY_TYPES { 设备: { desc: 具体的设备或机组, examples: [3号循环泵, 2号空压机, 1号锅炉] }, 部件: { desc: 设备的组成部件或易损件, examples: [机械密封, 轴承, 叶轮, O型圈] }, 故障现象: { desc: 设备运行中出现的异常表现, examples: [振动超标, 温度过高, 泄漏] }, 维修动作: { desc: 针对故障执行的维修操作, examples: [更换, 紧固, 清洗, 校准] }, 工单: { desc: 维修工单编号, examples: [WO-2024-0312, WO-2024-0455] } } # 关系类型定义头实体类型 - 尾实体类型关系名 RELATION_TYPES [ {head: 设备, relation: 包含部件, tail: 部件}, {head: 设备, relation: 出现故障, tail: 故障现象}, {head: 故障现象, relation: 维修方式, tail: 维修动作}, {head: 维修动作, relation: 涉及部件, tail: 部件}, {head: 工单, relation: 对应设备, tail: 设备}, ]这段定义里实体类型控制在 5 个关系类型 5 条覆盖了“哪台设备出了什么故障、怎么修的、换了什么件、对应哪张工单”这条主线。参数上要注意examples不是给模型看的训练数据而是帮你自己校验类型边界——如果两个类型的示例看起来能互相替换说明类型划分有问题。关系类型里head和tail必须都在实体类型里存在否则抽取时会产出悬空边。提示schema 第一版不要超过 8 个实体类型和 10 条关系跑通一轮抽取后看哪些关系大量漏抽再针对性补充。2.3 用 OneKE 跑通一次抽取输入、输出与参数抽取的输入是一段文本加 schema输出是三元组列表。实际工程里不会把整份 PDF 直接丢进去而是先按段落或章节切分每段控制在 500 到 1000 字太长了模型注意力会散太短了关系抽不全。# extract.py from oneke import OneKExtractor # 假设的调用方式按实际框架 API 调整 extractor OneKExtractor( model_nameoneke-base, # 模型规格按显存选 schemaENTITY_TYPES, # 实体类型 relation_schemaRELATION_TYPES, max_length1024, # 单次输入最大 token 数 temperature0.1 # 抽取任务要确定性温度调低 ) text 2024年3月12日3号循环泵出现机械密封泄漏 检修工单WO-2024-0312记录更换机械密封检查轴承磨损情况。 triples extractor.extract(text) for t in triples: print(t) # 期望输出类似 # {head: 3号循环泵, head_type: 设备, relation: 出现故障, tail: 机械密封泄漏, tail_type: 故障现象} # {head: 机械密封泄漏, head_type: 故障现象, relation: 维修方式, tail: 更换, tail_type: 维修动作} # {head: 更换, head_type: 维修动作, relation: 涉及部件, tail: 机械密封, tail_type: 部件}关键参数说明temperature设 0.1 是为了让同一段文本多次抽取结果稳定抽取任务不需要创造性max_length根据你的文本块长度和显存调整1024 是常见起点model_name按实际可用的模型规格填显存不够就换小规格。抽取结果里如果出现head_type不在 schema 里的情况说明模型在自由发挥需要在后处理里过滤掉。2.4 抽取结果清洗去重、归一和置信度过滤模型抽出来的三元组不能直接入库至少要做三件事。第一实体归一“3号循环泵”和“3号泵”要合并成同一个节点常见做法是维护一个别名词典或者用编辑距离加规则匹配。第二去重同一段文本里“更换机械密封”可能被抽两次按(head, relation, tail)三元组去重。第三置信度过滤如果框架返回置信度分数低于阈值的丢掉宁可少抽也不要脏数据入库。# clean.py def normalize_entity(name, alias_map): 实体归一查别名词典没有就返回原名 return alias_map.get(name, name) def dedup_triples(triples): 按三元组去重 seen set() result [] for t in triples: key (t[head], t[relation], t[tail]) if key not in seen: seen.add(key) result.append(t) return result def filter_by_confidence(triples, threshold0.7): 置信度过滤没有分数字段的默认保留 return [t for t in triples if t.get(confidence, 1.0) threshold]别名词典初期可以手工维护几十条高频词后面从抽取结果里统计低频实体再补充。置信度阈值 0.7 是经验值抽得太少就降到 0.5抽得太脏就提到 0.8按实际效果调。3. 从三元组到 Neo4j入库、索引与查询性能3.1 Neo4j 环境准备与连接配置图数据库选 Neo4j 是常见做法社区版够用Cypher 查询语言上手快。本地跑通最小环境用 Docker 最省事# 启动 Neo4j 社区版映射 7474 浏览器端口和 7687 Bolt 协议端口 docker run -d \ --name neo4j-kg \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTHneo4j/your_password \ -v $HOME/neo4j/data:/data \ neo4j:5-community启动后浏览器打开http://localhost:7474用neo4j/your_password登录。Python 侧用官方驱动连接# db.py from neo4j import GraphDatabase driver GraphDatabase.driver( bolt://localhost:7687, auth(neo4j, your_password) ) def verify_connection(): with driver.session() as session: result session.run(RETURN 1 AS ok) return result.single()[ok] 1NEO4J_AUTH里的密码要换成你自己的别用默认密码直接上生产。-v挂载数据目录是为了容器重启后数据不丢这个后悔药一定要提前吃。3.2 批量写入三元组MERGE 而不是 CREATE入库的核心是“有则复用无则创建”所以用MERGE而不是CREATE。CREATE每次都会新建节点跑两遍就出现重复节点图谱直接废掉。# ingest.py def insert_triples(tx, triples): 批量写入三元组按类型创建节点和关系 for t in triples: cypher MERGE (h:%s {name: $head}) MERGE (t:%s {name: $tail}) MERGE (h)-[:%s]-(t) % (t[head_type], t[tail_type], t[relation]) tx.run(cypher, headt[head], tailt[tail]) def batch_ingest(triples, batch_size500): with driver.session() as session: for i in range(0, len(triples), batch_size): batch triples[i:ibatch_size] session.execute_write(insert_triples, batch)这里用字符串拼接把类型名和关系名嵌进 Cypher是因为 Neo4j 不支持把标签和关系类型参数化。拼接前必须确保类型名来自你自己的 schema 白名单不能直接拼用户输入否则有注入风险。batch_size设 500 是平衡内存和速度的经验值数据量大就分批跑。3.3 建索引让图查询从秒级降到毫秒级不建索引的图查询节点一多就是全图扫描。至少给每个实体类型的name属性建索引# index.py def create_indexes(): with driver.session() as session: for label in [设备, 部件, 故障现象, 维修动作, 工单]: session.run(fCREATE INDEX IF NOT EXISTS FOR (n:{label}) ON (n.name)) create_indexes()建完索引后按名称查节点的查询会走索引。可以用EXPLAIN前缀看查询计划确认走的是NodeIndexSeek而不是AllNodesScan。数据量到十万节点级别索引的有无就是秒级和毫秒级的差别。3.4 图查询示例从问题到 Cypher问答系统的图查询层本质是把自然语言问题翻译成 Cypher。常见问题类型和对应查询问题类型示例问题Cypher 思路实体属性查询3号泵有哪些部件从设备节点出发走“包含部件”关系故障关联查询振动超标一般怎么修从故障现象出发走“维修方式”关系路径查询哪张工单换了机械密封从部件反向走到工单多跳查询3号泵的故障涉及哪些部件设备→故障→维修动作→部件三跳# query.py def query_parts_of_equipment(equipment_name): 查某设备包含的部件 cypher MATCH (e:设备 {name: $name})-[:包含部件]-(p:部件) RETURN p.name AS part with driver.session() as session: result session.run(cypher, nameequipment_name) return [r[part] for r in result]多跳查询要注意路径长度超过三跳的查询在稠密图上可能爆炸实际用的时候加LIMIT限制返回条数。4. 问答系统搭建图查询、RAG 与意图路由4.1 意图识别什么问题走图什么问题走 RAG不是所有问题都适合走图查询。“3号泵换了什么密封件”走图“机械密封的安装扭矩标准是多少”这种需要大段说明文字的走 RAG 更合适。所以问答层第一步是意图分类常见做法是用大模型做 few-shot 分类或者用关键词规则先兜底。# router.py GRAPH_KEYWORDS [哪个, 哪些, 换了, 对应, 关联, 哪张工单] def route_question(question): 简单规则路由命中图查询关键词走图否则走 RAG for kw in GRAPH_KEYWORDS: if kw in question: return graph return rag规则路由的好处是可控、可解释坏处是覆盖不全。上线初期用规则积累一批 badcase 后再换成模型分类。别一上来就上复杂路由先把两条链路各自跑通。4.2 图查询链路自然语言转 Cypher 的两种做法第一种是模板匹配预先定义好问题模板和对应 Cypher抽取问题里的实体填进去。适合问题类型固定的场景准确率高但不灵活。第二种是大模型生成 Cypher把 schema 和几个示例查询喂给大模型让它生成 Cypher。灵活但可能生成语法错误或危险查询。# text2cypher.py PROMPT_TEMPLATE 你是一个 Neo4j Cypher 专家。已知图中有以下节点和关系 节点设备(name), 部件(name), 故障现象(name), 维修动作(name), 工单(name) 关系设备-[:包含部件]-部件, 设备-[:出现故障]-故障现象, 故障现象-[:维修方式]-维修动作, 维修动作-[:涉及部件]-部件, 工单-[:对应设备]-设备 请把下面的问题转成 Cypher 查询只输出 Cypher不要解释 问题{question} def generate_cypher(question, llm_client): prompt PROMPT_TEMPLATE.format(questionquestion) cypher llm_client.generate(prompt) # 安全校验只允许 MATCH/RETURN禁止写操作 if any(kw in cypher.upper() for kw in [DELETE, CREATE, SET, DROP]): raise ValueError(生成的 Cypher 包含写操作已拦截) return cypher安全校验这步不能省。大模型生成的 Cypher 如果带DELETE一条查询就能把图谱清空。白名单只放MATCH、RETURN、WHERE、LIMIT这些读操作关键词。4.3 RAG 兜底链路向量检索加图谱上下文RAG 链路负责回答需要文字说明的问题。和纯 RAG 的区别是检索到的文本块可以附带图谱里的相关实体信息让生成的回答更有结构。# rag_pipeline.py def rag_answer(question, vector_store, llm_client, driver): # 1. 向量检索拿相关文本块 docs vector_store.similarity_search(question, k5) context \n.join([d.page_content for d in docs]) # 2. 从问题里抽实体查图谱补充上下文 entities extract_entities_from_question(question) graph_context for ent in entities: related query_related_triples(driver, ent) graph_context \n.join(related) # 3. 拼 prompt 生成回答 prompt f参考资料\n{context}\n\n图谱信息\n{graph_context}\n\n问题{question}\n回答 return llm_client.generate(prompt)k5是检索返回的文本块数量太多会超上下文长度太少可能漏信息。图谱上下文作为补充不是替代向量检索两者结合能缓解纯 RAG 答不准关系类问题的情况。4.4 把两条链路拼成完整问答接口# qa_system.py def answer(question): route route_question(question) if route graph: try: cypher generate_cypher(question, llm_client) result run_cypher(cypher) if result: return format_graph_result(result) except Exception as e: # 图查询失败降级到 RAG pass return rag_answer(question, vector_store, llm_client, driver)图查询失败降级到 RAG 是必要的容错。Cypher 生成可能出错图里可能没数据这时候不能让用户看到报错降级到 RAG 至少能返回一个基于文本的回答。5. 避坑与排查抽取和入库阶段最容易翻车的五件事5.1 抽取结果实体类型对不上 schema现象模型输出的head_type是“设备类型”而不是 schema 里定义的“设备”或者冒出 schema 里没有的类型。原因schema 描述不够清晰模型按自己的理解归类。解决在 schema 的desc里写清楚类型边界抽取后加一层类型白名单过滤不在白名单里的类型直接丢弃或映射到最近的类型。5.2 同一实体多个名称导致图谱碎片化现象图里同时存在“3号循环泵”“3号泵”“#3循环泵”三个节点查一个查不全。原因没有做实体归一。解决维护别名词典入库前统一名称或者入库后用 Cypher 做合并把多个节点的关系转移到主节点再删除冗余节点。别名词典要持续维护从查询日志里发现新的别名。5.3 批量写入时内存溢出现象一次性把几万条三元组传给session.execute_write程序卡死或 OOM。原因单次事务太大。解决分批写入batch_size控制在 500 到 1000每批一个事务。数据量特别大就用LOAD CSV方式导入比逐条 MERGE 快一个数量级。5.4 图查询返回空结果但数据明明存在现象Cypher 查询返回空但浏览器里能看到节点。原因常见的是标签或关系名拼写不一致比如 schema 里是“包含部件”入库时写成了“包含零件”或者节点名称有空格差异。解决先用MATCH (n) RETURN DISTINCT labels(n)和MATCH ()-[r]-() RETURN DISTINCT type(r)确认实际的标签和关系名再对照查询语句。5.5 大模型生成的 Cypher 语法错误或性能差现象生成的 Cypher 报语法错误或者能跑但特别慢。原因模型对 Cypher 语法掌握不稳定或者生成的查询没走索引。解决prompt 里给两三个正确示例生成后用EXPLAIN检查查询计划加超时限制超过 5 秒的查询直接中断降级到 RAG。别让一个慢查询拖垮整个问答接口。6. 抽取质量怎么验证一套可复用的评估脚本系统跑起来只是开始真正决定能不能用的是抽取质量。我一般会留一份人工标注的测试集哪怕只有五十条文本每轮改完 schema 或换模型都跑一遍评估。评估指标看三个实体抽取的准确率和召回率、关系抽取的准确率和召回率、以及端到端问答的命中率。# evaluate.py def evaluate_extraction(predicted_triples, gold_triples): 对比预测三元组和人工标注三元组 pred_set set((t[head], t[relation], t[tail]) for t in predicted_triples) gold_set set((t[head], t[relation], t[tail]) for t in gold_triples) tp len(pred_set gold_set) precision tp / len(pred_set) if pred_set else 0 recall tp / len(gold_set) if gold_set else 0 f1 2 * precision * recall / (precision recall) if (precision recall) else 0 return {precision: precision, recall: recall, f1: f1}这个脚本的关键在于gold_triples必须人工标注不能拿模型输出当标准答案。标注五十条文本大概花两小时但能帮你判断每次调整是变好还是变坏。precision 低说明抽了太多错的去查 schema 是不是太宽recall 低说明漏抽多去查文本切分是不是把关系切断了或者 schema 里关系定义不够明确。端到端问答的评估更直接准备三十个问题每个问题有标准答案跑一遍看答对多少。图查询链路和 RAG 链路分开统计这样能定位是抽取的问题还是查询生成的问题。我自己的习惯是每次改完 schema 先跑抽取评估F1 没提升就不动查询层避免两个变量一起改导致说不清哪里出了问题。这套流程跑顺了后面扩实体类型、换模型、加数据源都有据可依不会越调越乱。希望帮到你。本文还有配套的精品资源点击获取