简介一份基于 OneKE 模型构建知识图谱并搭建问答系统的 Python 毕设项目资源包主要面向计算机、通信、人工智能、自动化等专业学生也适合课程设计、大作业或毕业设计。项目围绕实体关系抽取、SPO 三元组处理、图谱构建与 RAG 问答链路展开结构清晰可用于入门学习或二次改造。资源压缩包共 27 个文件约 2.87MB含 Python 脚本、JSON/CSV 结果数据、Cypher 图谱导入查询脚本、PNG 流程示意图、Shell 辅助脚本及 README 文档覆盖从数据处理到问答验证的完整环节数据样例、输出结果与流程说明相互对应便于核验。已有 450 人学习下载。项目为答辩 98 分的高分毕设代码经过调试测试运行有保障文档说明和可视化流程便于理解 OneKE 的知识抽取逻辑与问答系统搭建步骤适合作为实战参考。1. 用 Python 和 OneKE 做知识图谱问答先想清楚这三件事知识图谱的价值不在于那张漂亮的节点关系图而在于它能回答“A 和 B 之间是什么关系”“C 的属性有哪些”这类结构化问题。OneKE 是一个面向中文的大模型知识抽取框架它把“从非结构化文本里抽实体和关系”这件事的门槛压得很低——不需要为每种关系训练一个模型只要给定 schema它就能按指令输出结构化结果。把 OneKE 抽出来的三元组灌进 Neo4j再包一层问答服务一个基于知识图谱的问答系统就成型了。这套组合适合两类人一类是手头有大量业务文档、想做领域问答但不想人工整理知识库的工程师另一类是课程设计或毕业设计需要“从数据到可演示 Demo”完整链路的同学。整个链路里“模型怎么跑起来”和“知识怎么存进图数据库”是两个最大的坑所以下面从环境开始一条条拆。2. 搭好 OneKE 运行环境Python 版本、CUDA 与模型加载方式2.1 用 conda 建一个干净的 Python 3.10 环境OneKE 基于 Qwen 系列底座微调依赖 PyTorch 和 TransformersPython 版本建议直接用 3.10。实操上最省事的方式是 conda 独立环境避免把系统 Python 搞乱conda create -n oneke python3.10 -y conda activate oneke pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 pip install transformers accelerate bitsandbytes pandas openpyxl pip install neo4j py2neo这里把torch的 CUDA 版本显式指定为 cu118是为了和显卡驱动解耦。如果你机器上的驱动版本较新也可以直接pip install torch装默认的 cu121 版本。判断标准很简单装完后跑一句python -c import torch; print(torch.cuda.is_available())输出True就说明 GPU 可用如果是False后面用 CPU 跑也能出结果只是速度慢一个量级。提示bitsandbytes在 Windows 上的兼容性比较差如果你是 Windows 环境优先用 WSL2Mac 用户直接用 CPU 推理即可OneKE 7B 量化的模型在 M 系列芯片上也能跑只是需要把max_new_tokens调小。2.2 OneKE 模型加载Transformers 直接推理与量化参数OneKE 的模型权重从 ModelScope 或 HuggingFace 拉下来后加载方式跟普通 Qwen 模型完全一致。核心代码里做两件事一是用AutoModelForCausalLM加载二是用BitsAndBytesConfig做 4bit 量化让 7B 模型能在 12GB 显存的卡上跑起来import torch from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig quant_config BitsAndBytesConfig( load_in_4bitTrue, bnb_4bit_compute_dtypetorch.float16, bnb_4bit_quant_typenf4, bnb_4bit_use_double_quantTrue ) tokenizer AutoTokenizer.from_pretrained( oneke_model_path, trust_remote_codeTrue ) model AutoModelForCausalLM.from_pretrained( oneke_model_path, quantization_configquant_config, device_mapauto, trust_remote_codeTrue )这段配置里load_in_4bitTrue是显存不够时的关键开关nf4是较新的量化类型比原来的 fp4 损失更小device_mapauto让模型自动分布到可用的 GPU 和 CPU 上。如果你的显存大于 24GB可以去掉quant_config直接用torch.float16加载抽取质量会稍微好一点尤其在关系类别多、抽取边界模糊的领域数据上。2.3 验证模型是否就绪跑一个最小的抽取请求环境有没有问题最好用一个最小请求验证。OneKE 的输入格式遵循对话模板用户消息里放抽取指令和文本模型返回 JSON 格式的结果messages [ {role: user, content: 请从下面文本中抽取实体和关系并以JSON格式输出。\n文本张三毕业于北京大学现任华为技术有限公司的算法工程师。} ] text tokenizer.apply_chat_template( messages, tokenizeFalse, add_generation_promptTrue ) inputs tokenizer([text], return_tensorspt).to(cuda) outputs model.generate( **inputs, max_new_tokens512, temperature0.1, top_p0.9, do_sampleFalse ) response tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:], skip_special_tokensTrue) print(response)这里temperature0.1和do_sampleFalse很关键知识抽取是确定性任务采样温度太高会导致同一段文本多次抽取结果不一致。max_new_tokens512是输出长度上限文本长、实体多的时候需要调大到 1024否则 JSON 会被截断后面解析就会失败。如果这一步能稳定输出{entities: [...], relations: [...]}结构的数据环境就算全部打通了。3. 用 OneKE 做实体与关系抽取Schema 定义和批量抽取代码3.1 先定义知识图谱的 Schema再写抽取逻辑很多人在 OneKE 上翻车不是模型跑不起来而是没定义清楚“要抽什么”。OneKE 虽然支持零样本抽取但不给 schema 时它会把文本里所有名词都当实体把所有动词都当关系结果就是图谱里全是噪声。Schema 要回答三个问题实体有哪些类型、关系有哪些类型、每种关系连接哪两类实体。以“人物履历”场景为例实体类型关系类型头实体尾实体人物毕业院校人物组织机构组织机构就职公司人物组织机构职位担任职位人物职位地点出生地人物地点这个表格就是抽取的约束。OneKE 支持这种“体系结构”式的定制把 schema 写进 system prompt模型就会严格按给定类型输出而不是自由发挥。3.2 封装一个批量抽取类并发与失败重试单条文本调用模型太慢实际工程中至少要对齐到“批量处理”。写一个简单的抽取器内部维护 schema外部只暴露extract(text)接口import json from concurrent.futures import ThreadPoolExecutor, as_completed class OneKEExtractor: def __init__(self, model, tokenizer, schema_prompt, max_tokens800): self.model model self.tokenizer tokenizer self.schema_prompt schema_prompt self.max_tokens max_tokens def build_messages(self, text): return [ {role: system, content: self.schema_prompt}, {role: user, content: f抽取以下文本中的知识\n{text}} ] def extract(self, text): messages self.build_messages(text) prompt self.tokenizer.apply_chat_template( messages, tokenizeFalse, add_generation_promptTrue ) inputs self.tokenizer([prompt], return_tensorspt).to(cuda) outputs self.model.generate( **inputs, max_new_tokensself.max_tokens, temperature0.1, do_sampleFalse ) response self.tokenizer.decode( outputs[0][inputs[input_ids].shape[1]:], skip_special_tokensTrue ) try: return json.loads(response) except json.JSONDecodeError: return self._repair_json(response) def batch_extract(self, texts, max_workers4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(self.extract, t): t for t in texts} for future in as_completed(future_map): try: results.append(future.result()) except Exception as e: results.append({error: str(e), text: future_map[future]}) return resultsThreadPoolExecutor在这里的作用不是加速单卡推理而是实现“边推理边解析”的流水线。模型generate是同步阻塞的多线程能让 CPU 侧的 JSON 解析和 GPU 侧的生成重叠起来吞吐大概能提升 20% 到 30%。max_workers不要超过 GPU 显存能同时承载的 batch 数7B 量化模型在 24GB 显存上开 4 个比较安全。注意batch_extract的结果顺序和输入顺序不一定一致如果你后续要按原文本顺序对齐数据记得用 dict 保存text - result的映射而不是直接 append。3.3 抽取结果的清洗与实体对齐OneKE 返回的结果里最典型的问题是同一个实体在不同文本里写法不同比如“北大”和“北京大学”是同一个学校“华为技术有限公司”和“华为”是同一个公司。这一步不做实体对齐入库后图谱里就会出现大量冗余节点问答时也会因为名字不匹配而查不到。常见做法是用字符串归一化加别名映射import re def normalize_entity_name(name): # 去掉全角空格、括号内容归一化 name re.sub(r.*?|\(.*?\), , name) name name.replace( , ).replace( , ) return name.strip() alias_map { 北大: 北京大学, 华为: 华为技术有限公司, 北邮: 北京邮电大学 } def resolve_entity(name): norm normalize_entity_name(name) return alias_map.get(norm, norm)这段代码的逻辑很简单先做规则清洗再用词典消歧。对生产级场景alias_map 应该从数据里自动挖掘比如把两个实体的字符串相似度和共现文档数组合起来做聚类对课程设计和中小规模 Demo手写别名表完全够用而且可解释性更强。清洗后的结果再进入 Neo4j能省掉大量后续维护成本。4. 把知识抽取结果写入 Neo4j节点设计、批量入库与索引4.1 图谱模型设计用标签区分实体类型Neo4j 里节点用 Label 区分类型关系用 Type 区分语义。把第 3 章的 Schema 直接映射过来即可不需要额外设计。一个容易忽略的点是给实体节点加上name属性作为业务主键同时保留一个entity_id作为内部唯一标识。原因在于后续问答系统要根据用户提到的实体名去查询如果只用内部 ID还得再做一次名字映射。4.2 用 py2neo 批量写入MERGE 而不是 CREATECREATE每次都会新建节点重复跑一次脚本就会产生重复数据MERGE则按给定属性去匹配存在就不重复创建。这里选择用MERGE配合UNIQUE约束保证同一名字的同类型节点只有一个。from py2neo import Graph, Node, Relationship graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) def build_and_write(extracted_doc): tx graph.begin() for ent in extracted_doc.get(entities, []): node Node( ent[type], # 实体类型作为 Label nameent[name], entity_ident.get(id, ent[name]) ) tx.merge(node, name, name) for rel in extracted_doc.get(relations, []): head Node(rel[head_type], namerel[head]) tail Node(rel[tail_type], namerel[tail]) tx.merge(head, name, name) tx.merge(tail, name, name) relationship Relationship.type(rel[relation])( head, tail, source_textrel.get(source_text, ) ) tx.merge(relationship, source_text, source_text) tx.commit()这里的tx.merge是关键它要求节点上有唯一性约束才能高效运行。否则每次 MERGE 都会扫描全表判断是否已存在数据量到几万节点后速度会急剧下降。另一个细节是关系上也挂了source_text属性这是为了追溯“这条关系是从哪句话里抽出来的”排查错误时非常好用。4.3 Cypher 建索引与约束在 Python 里执行 Cypher 也可以但更清晰的做法是直接在 Neo4j Browser 里执行一次CREATE CONSTRAINT entity_name_unique IF NOT EXISTS FOR (n) REQUIRE n.name IS UNIQUE; CREATE INDEX entity_type_index IF NOT EXISTS FOR (n) ON (n.type);第一条要求所有节点的name属性全局唯一这是 MERGE 能高效工作的前提。第二条索引为后续按类型过滤的查询加速。要注意的是如果一个 Label 下已经有重复数据约束是建不起来的得先清理数据再用MERGE去重。这也是“先约束后入库”的原因——一旦脏数据进去了清理成本远高于开始时多花的那几分钟。5. 搭建知识图谱问答系统实体链接、Cypher 生成与兜底策略5.1 问答系统的整体流程问答系统的输入是自然语言问题输出是答案。图谱问答的典型流程分成三段先从问题里识别出实体再把问题意图映射为查询模板最后把查询结果拼装成自然语言答案。这个流程和基于向量检索的 RAG 有本质区别——RAG 查的是文档片段图谱问答查的是结构化关系所以它可以回答“张三在哪家公司任职”这种精确问题而文档检索只能返回相关段落。5.2 基于规则模板生成 Cypher以“人物-公司-职位”查询为例OneKE 在问答阶段的作用是实体识别和意图分类但实际工程里最简单的做法是先用规则模板覆盖 80% 的查询模板覆盖不了的再交给大模型生成 Cypher。规则模板的好处是稳定、可调试不会出现模型生成的 Cypher 语法错误。import re from py2neo import Graph graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) QUESTION_PATTERNS [ { intent: person_company, template: MATCH (p:人物 {name: $entity})-[r:就职公司]-(c:组织机构) RETURN c.name AS answer, regex: re.compile(r(.?)(?:在哪|在什么|就职于|任职于)(?:公司|企业)?(?:工作|上班)?) } ] def extract_entity(question): # 简单策略问句中出现的已知实体名 candidates [] for node in graph.run(MATCH (n) RETURN n.name AS name LIMIT 5000): cand node[name] if cand and cand in question: candidates.append(cand) if not candidates: return None return max(candidates, keylen) def answer(question): entity extract_entity(question) if not entity: return 知识库中没有找到相关实体请换一种问法试试。 for pattern in QUESTION_PATTERNS: if pattern[regex].search(question): result graph.run( pattern[template], entityentity ).data() if result: values [row[answer] for row in result] return f{entity}的相关结果有 、.join(values) return 暂时无法回答这个问题。extract_entity是最朴素的主语识别策略——拿图谱里已有的实体名去问句里做最长匹配。这个策略对“张三在哪家公司”这种主谓宾完整的问题有效因为实体名完整出现在句子里。如果用户说“他的公司是哪家”就得先做指代消解把“他”替换成上文的实体。更稳的方式是直接用 OneKE 再跑一次问题文本抽取其中的实体类型然后和库内实体做匹配。5.3 结果拼装与兜底策略查询结果可能是空、可能有多条、也可能用户问的问题不在预设模板里。兜底策略分三层处理。def answer_with_fallback(question): entity extract_entity(question) if not entity: # 兜底1实体都没识别出来转给 OneKE 做开放抽取 extracted extractor.extract(question) if extracted.get(entities): entity extracted[entities][0][name] else: return 未识别到知识库中的实体。 # 兜底2模板没覆盖到的问题生成通用属性查询 result graph.run( MATCH (n {name: $entity}) RETURN labels(n)[0] AS label, keys(n) AS props, entityentity ).data() if result: return f在知识库中找到实体“{entity}”类型是{result[0][label]}属性包括{, .join(result[0][props])}。 return 知识库中没有这个实体的关联信息。第一层兜底解决“实体识别失败”手段是回退到 OneKE 再做一次抽取利用模型的泛化能力从问句里抽出实体名第二层兜底解决“模板未覆盖”手段是动态查询实体的属性和类型把图谱里已有信息以半结构化形式返回给用户。两层兜底配合后系统的可用性会明显提升——至少不会出现用户问了个不在模板里的问题就直接返回空。问答系统的难点从来不是单条回答有多准而是系统边界在哪里、边界外的请求怎么优雅地落到下一个策略上。6. 问答质量提升的三个细节置信度过滤、实体消歧与缓存更新6.1 给 OneKE 的输出加置信度过滤OneKE 返回的 JSON 里没有置信度字段但可以从解码概率里近似估计。Transformers 的generate接口可以带回scores对每个输出 token 取概率均值低置信度的抽取结果直接丢弃或打回人工审核。实操上更省事的方式是让模型在输出时附带一句“不确定”比如在 system prompt 里加“如果文本中没有明确提及关系返回 relation 为 null”。这一步能显著减少幻觉型三元组入库。6.2 实体消歧的两个代价最低的做法第一个做法是“图谱内引用计数”——同一个 name 被多条文本同时提及这个实体就更可信反之只出现过一次的实体在问答时权重降低。第二个做法是“关系类型约束校验”——比如“毕业院校”的关系头实体必须是人物尾实体必须是组织机构类型不匹配的三元组直接过滤。这两个规则都不用额外模型Python 里几十行就能实现但能把图谱的噪声降一个量级。6.3 图谱增量更新与问答缓存的联动新文档进来时只需要对新增文本跑 OneKE然后用第 4 章的MERGE逻辑增量写入MERGE天然幂等重复跑不会产生重复数据。问答侧的缓存则要建立“实体名 - 答案”的映射当某个实体的关联关系被更新时主动失效该实体的缓存条目避免用户拿到旧答案。用 Redis 做缓存时key 直接设计成实体名即可。验证项预期结果排查方向单条文本抽取JSON 可解析schema 字段完整prompt 是否给了 schemamax_new_tokens是否过短批量入库MERGE 后节点数稳定唯一约束是否建立事务是否提交问答命中返回实体关联结果实体链接是否命中模板正则是否覆盖无实体问题走兜底不报错extract_entity返回值是否为None把这三个细节加进去这个系统就不再是“能跑的 Demo”而是一个可以被持续投喂数据的知识图谱问答服务。每一条新文本进来图谱长一分问答能覆盖的边界就外扩一分。这种循环跑通之后你会发现 OneKE 最值钱的能力不是它抽得有多准而是它让你从“手工整理知识”这种脏活里彻底解放出来把精力放到数据清洗、图谱设计和问答策略这些真正决定上限的地方。本文还有配套的精品资源点击获取
