基于Python的PDF知识图谱构建与检索源码实战

简介这是一套面向计算机、数学、电子信息等专业学生的Python项目源码围绕PDF识别与分析、信息抽取、知识图谱构建及基于图谱的信息检索展开适合用作课程设计、期末大作业或高分毕设的参考方案。资源包共119个文件以47个Python源码文件为核心辅以28个pyc编译文件、15个JavaScript前端脚本、9个JSON配置、5张PNG图片及Vue组件、SQLite数据库、SQL脚本、xlsx数据表等压缩包约4.79MB前后端与数据层结构相对完整。目前已有962人学习下载具备一定参考热度。读者可从中了解PDF文本解析与结构化抽取的实现思路学习实体关系抽取并落地为知识图谱再基于图谱完成检索查询的完整链路同时可借鉴项目目录组织、前后端交互与数据存储方式便于在此基础上调试扩展或迁移到自己的课题中。1. 从一堆 PDF 到可查询知识图谱这套 Python 源码到底能跑出什么手里攒了几百份行业研报、技术白皮书、项目验收文档想快速查“某型号设备的额定参数在哪份文件里提过”“A 公司和 B 公司之间有过哪些合作记录”靠 CtrlF 一份份翻翻到第三十份就开始怀疑人生。这套基于 Python 的 PDF 识别与分析源码解决的正是这个场景把非结构化 PDF 文本抽出来做实体与关系抽取落成知识图谱再基于图谱做信息检索。它适合计算机、电子信息、数学等专业做课程设计或毕设的同学也适合需要快速验证“文档→图谱→检索”链路的工程师。整个项目是前后端分离结构前端用 Webpack 打包后端 Python 负责解析、抽取和检索源码完整下载后按依赖装好就能跑通主流程。2. PDF 解析与文本抽取先把脏数据洗干净再谈算法PDF 解析是整个链路的第一道关也是最容易翻车的地方。很多人以为pdfplumber或PyPDF2一行extract_text()就能拿到干净文本实际跑起来会发现扫描件返回空字符串、双栏排版文字顺序错乱、表格被压成一行、页眉页脚混进正文。这套源码在解析层做了分路处理先判断 PDF 是文本型还是扫描型再决定走直接抽取还是 OCR 兜底。2.1 文本型与扫描型 PDF 的分流判断文本型 PDF 内部存的是字符编码和坐标抽取快、准确率高扫描型 PDF 本质是图片必须走 OCR。判断方法很直接用pdfplumber打开后取第一页看extract_text()返回的字符数如果少于阈值常见做法是 50 个字符基本可以判定为扫描件或加密文档。import pdfplumber def is_scanned_pdf(pdf_path, sample_pages3, char_threshold50): 判断 PDF 是否为扫描型抽样前几页字符数低于阈值即视为扫描件 with pdfplumber.open(pdf_path) as pdf: total_chars 0 for page in pdf.pages[:sample_pages]: text page.extract_text() or total_chars len(text.strip()) # 平均每页字符数低于阈值判定为扫描型 avg_chars total_chars / min(sample_pages, len(pdf.pages)) return avg_chars char_threshold这里sample_pages取 3 是平衡速度和准确率页数多的文档不必全量扫描char_threshold设 50 是经验值纯封面页可能字符少所以要看平均值而不是单页。如果判定为扫描型就转给 OCR 模块处理常见做法是用pytesseract配合pdf2image或者用RapidOCR这类轻量方案。注意 OCR 对 CPU 占用较高批量处理时建议加并发限制别一次性把几百页全塞进内存。2.2 版面分析与表格抽取的取舍双栏论文和带表格的研报是解析的重灾区。pdfplumber提供了extract_tables()方法能按坐标把表格还原成二维列表但前提是 PDF 里有明确的表格线。没有边框的表格抽取结果往往是一团乱麻。源码里的处理策略是先尝试extract_tables()如果返回的表格行列数异常比如只有一列或超过 20 列就降级为按文本块坐标聚类。def extract_page_content(page): 单页内容抽取优先表格失败则按文本块坐标排序 tables page.extract_tables() valid_tables [] for table in tables: if table and len(table[0]) 1 and len(table) 1: valid_tables.append(table) # 按 top 坐标排序文本块缓解双栏顺序错乱 words page.extract_words(use_text_flowFalse, keep_blank_charsFalse) words_sorted sorted(words, keylambda w: (round(w[top] / 10), w[x0])) text .join(w[text] for w in words_sorted) return {tables: valid_tables, text: text}extract_words返回每个词的坐标按top坐标做粗粒度分组除以 10 再取整再按x0从左到右排能在一定程度上还原阅读顺序。这个方案不是万能的遇到复杂三栏排版仍会串行但比直接extract_text()稳定得多。表格抽取结果单独存成 JSON后续做信息抽取时可以针对表格内容走规则匹配比纯文本更可靠。2.3 清洗规则与分块策略抽出来的文本不能直接喂给下游模型得先过一遍清洗去掉页眉页脚通常是每页重复出现的短行、合并断行行尾没有标点且下一行首字母小写的大概率是同一句、统一全角半角。清洗完还要分块因为后续做实体抽取时整篇文档太长会导致模型截断。常见做法是按段落分块每块控制在 300500 字块之间保留 50 字重叠避免实体被切断。import re def clean_text(raw_text): 基础清洗去页眉页脚、合并断行、统一标点 lines raw_text.split(\n) # 去掉长度小于 8 且不含标点的行大概率是页眉页脚 lines [ln for ln in lines if len(ln.strip()) 8 or re.search(r[。], ln)] text \n.join(lines) # 合并被硬换行切断的句子 text re.sub(r([^\n。])\n([a-zA-Z\u4e00-\u9fa5]), r\1\2, text) # 全角转半角数字和字母 text text.replace(, ().replace(, )).replace(, :) return text.strip() def chunk_text(text, chunk_size400, overlap50): 按字符数分块保留重叠避免实体截断 chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) start end - overlap return chunkschunk_size设 400 是兼顾实体完整性和模型输入限制overlap设 50 能覆盖大多数实体长度。如果下游用的是大模型 API块可以再大一些如果用本地小模型做 NER块要更小。清洗规则里最容易误杀的是短行有些文档的标题就是短行所以判断条件里加了“不含标点”这个约束标题通常不带句号。3. 信息抽取与知识图谱构建实体、关系怎么落成图文本洗干净之后下一步是把里面的实体和关系抽出来组织成图谱。这一步的核心问题是用什么方法抽规则、模型还是混合这套源码走的是“规则打底 模型补充”的路线对课程设计和毕设来说纯模型方案依赖标注数据成本太高纯规则方案覆盖面有限但胜在可控、可解释。图谱存储用 Neo4j 比较常见也可以用 NetworkX 先做原型验证。3.1 实体识别规则与模型的分工实体识别的目标是找出人名、机构名、产品名、时间、金额等。规则方案用正则和词典匹配适合格式固定的实体比如日期、金额、编号。模型方案用spaCy或HanLP做 NER适合人名、地名这类没有固定格式的实体。import re from collections import defaultdict # 规则实体日期、金额、编号 PATTERNS { DATE: r\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?, MONEY: r[¥$]?\d(?:\.\d)?\s*(?:万元|亿元|美元|元), CODE: r[A-Z]{2,}-\d{3,}, } def rule_extract(text): 基于正则的实体抽取返回 {实体类型: [实体列表]} result defaultdict(list) for label, pattern in PATTERNS.items(): for match in re.finditer(pattern, text): result[label].append(match.group()) return dict(result)PATTERNS里的正则按实际文档调整比如合同文档要加合同编号规则技术文档要加版本号规则。规则抽取的优点是零训练成本、结果可解释缺点是遇到变体就失效。模型抽取用spaCy的话中文需要加载对应语言包实体类型也要根据业务定义重新映射。实际项目里规则先跑一遍模型再补一遍两者结果合并去重。3.2 关系抽取与三元组生成有了实体还要知道实体之间的关系。关系抽取比实体识别更难因为关系类型多、表达方式灵活。源码里的做法是先定义关系模式比如“公司-合作-公司”“产品-属于-类别”“文档-提及-实体”然后用依存句法分析或模板匹配来抽。def extract_triples(sentence, entities): 基于模板的三元组抽取找实体之间的动词短语作为关系 triples [] for i, e1 in enumerate(entities): for e2 in entities[i1:]: if e1 e2: continue # 找两个实体之间的文本片段 start sentence.find(e1) end sentence.find(e2) if start -1 or end -1 or start end: continue between sentence[startlen(e1):end] # 关系词通常是 2-6 个字的动词短语 if 2 len(between) 6: triples.append((e1, between, e2)) return triples这个模板法的逻辑是两个实体之间的短文本片段大概率是关系词。between长度限制在 2 到 6 个字是为了过滤掉过长的从句。这个方法召回率一般但准确率尚可适合做原型。如果要提升效果可以引入HanLP的依存句法分析找nsubj和dobj关系或者用预训练模型做关系分类。三元组生成后统一存成(头实体, 关系, 尾实体)的格式方便导入图数据库。3.3 图谱入库Neo4j 与 NetworkX 的选择图谱存储有两个选择Neo4j 适合做持久化和复杂查询NetworkX 适合做原型验证和小规模分析。源码里两者都支持先用 NetworkX 建图验证逻辑再导出到 Neo4j。import networkx as nx from neo4j import GraphDatabase def build_graph(triples): 用 NetworkX 建图节点是实体边是关系 G nx.DiGraph() for head, rel, tail in triples: G.add_node(head) G.add_node(tail) G.add_edge(head, tail, relationrel) return G def export_to_neo4j(G, uri, user, password): 把 NetworkX 图导入 Neo4j driver GraphDatabase.driver(uri, auth(user, password)) with driver.session() as session: for node in G.nodes: session.run(MERGE (n:Entity {name: $name}), namenode) for u, v, data in G.edges(dataTrue): session.run( MATCH (a:Entity {name: $u}), (b:Entity {name: $v}) MERGE (a)-[:REL {type: $rel}]-(b), uu, vv, reldata[relation] ) driver.close()MERGE而不是CREATE是为了避免重复节点。Neo4j 的连接参数uri通常是bolt://localhost:7687默认用户neo4j密码在首次启动时设置。导入大量节点时逐条session.run效率低常见做法是用UNWIND批量导入或者先导出 CSV 再用LOAD CSV。NetworkX 的图对象可以直接用nx.draw可视化方便调试阶段看图谱结构对不对。4. 基于知识图谱的信息检索查询怎么落到图结构上图谱建好之后检索层要做的是把用户的自然语言查询翻译成图查询。这一步的难点在于用户输入的是“A 公司和 B 公司有什么合作”而图里存的是(A, 合作, B)这样的三元组中间需要一个映射层。源码里的检索模块支持两种模式关键词匹配和路径查询。4.1 查询意图解析与实体链接用户输入查询后先做实体识别把查询里的实体找出来再链接到图谱里的节点。比如查询“某型号设备的额定功率”先识别出“某型号设备”这个实体然后在图里找对应节点再沿边找“额定功率”相关的属性或关系。def parse_query(query, graph): 解析查询抽实体匹配图谱节点 # 复用前面的规则抽取 entities rule_extract(query) matched_nodes [] for label, ents in entities.items(): for ent in ents: if ent in graph.nodes: matched_nodes.append(ent) # 如果规则没抽到退化为子串匹配 if not matched_nodes: for node in graph.nodes: if node in query: matched_nodes.append(node) return matched_nodes实体链接的准确率直接影响检索效果。规则抽取覆盖不到的实体用子串匹配兜底但子串匹配容易误召回比如“北京”会匹配到“北京大学”。常见做法是加一层消歧用上下文或实体类型过滤。源码里对课程设计场景做了简化优先保证能跑通。4.2 路径查询与结果排序找到实体后沿图谱的边做路径查询。比如找两个实体之间的最短路径或者找某个实体的所有邻居。Neo4j 的 Cypher 查询写起来很直观NetworkX 也有对应的shortest_path方法。def query_graph(graph, start_node, end_nodeNone, max_depth3): 图查询找路径或邻居 if start_node not in graph.nodes: return [] if end_node and end_node in graph.nodes: try: path nx.shortest_path(graph, start_node, end_node) return [path] except nx.NetworkXNoPath: return [] # 没指定终点返回指定深度内的邻居 neighbors nx.single_source_shortest_path_length(graph, start_node, cutoffmax_depth) return list(neighbors.keys())max_depth限制在 3 是为了避免路径爆炸图谱大了之后深度查询很吃性能。结果排序可以按路径长度、边的权重或节点重要性来排。源码里用的是路径长度优先短路径排前面。如果图谱规模大建议在 Neo4j 里建索引CREATE INDEX ON :Entity(name)查询速度会明显提升。4.3 检索结果的可视化与前端对接检索结果最终要展示给用户前端用 Webpack 打包图谱可视化常见做法是用ECharts的 graph 类型或D3.js的力导向图。源码的前端部分在index.html和webpack.*.conf.js里构建流程是标准的 Webpack 配置webpack.dev.conf.js用于开发环境webpack.prod.conf.js用于生产打包。// 前端调用检索接口的示例 fetch(/api/search, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({query: 某型号设备 额定功率}) }) .then(res res.json()) .then(data { // data.nodes 和 data.edges 传给 ECharts 渲染 renderGraph(data.nodes, data.edges); });接口返回的nodes和edges结构要和前端渲染库对齐ECharts的 graph 需要nodes里有name和valueedges里有source和target。前后端联调时最常见的坑是跨域开发环境在webpack.dev.conf.js里配devServer.proxy转发到后端端口生产环境用 Nginx 反代。前端构建命令是npm run dev和npm run build依赖装好后直接跑。5. 避坑与排查这套源码跑起来最容易卡在哪这套源码结构不算复杂但环境依赖和数据处理环节有几个高频翻车点。下面几条是我实际跑的时候踩过的按“现象→原因→解决”整理照着排查能省不少时间。现象pdfplumber打开 PDF 报PDFSyntaxError。原因PDF 文件损坏或加密pdfplumber无法解析。有些文档设了权限密码虽然能打开看但程序读不了。 解决先用PyPDF2的is_encrypted判断是否加密加密的用decrypt()尝试空密码解密解不开就跳过并在日志里记录文件名。损坏的文件用qpdf --check检查能修复的修复不能修复的单独放一个目录。现象OCR 跑起来 CPU 占满批量处理卡死。原因pytesseract或RapidOCR默认用满所有核心几百页并发处理直接把机器拖垮。 解决限制并发数用concurrent.futures.ThreadPoolExecutor(max_workers2)控制同时处理的页数。另外 OCR 前先把 PDF 转成图片时降低 DPI150 到 200 足够300 以上对识别率提升有限但耗时翻倍。现象Neo4j 导入节点时报ConstraintValidationFailed。原因重复节点或关系违反唯一性约束MERGE在并发写入时可能触发。 解决导入前先去重用set对三元组去重。批量导入时加UNWIND减少事务次数或者先导入节点再导入关系分两步走。如果还是报错检查是不是有节点名为null或空字符串。现象前端npm run dev报Module not found。原因node_modules没装全或者 Node 版本和 Webpack 配置不兼容。老项目的 Webpack 配置在新版 Node 上经常出问题。 解决先npm install如果报错看具体缺哪个包。Node 版本建议用 14 或 16新版 Node 对老 Webpack 支持不好。实在不行用nvm切版本别硬扛。现象检索结果为空但图谱里明明有数据。原因实体链接没匹配上查询里的实体名和图谱里的节点名不完全一致比如多了空格、大小写不同、全半角差异。 解决在实体链接前做归一化统一转小写、去空格、全角转半角。图谱入库时也做同样的归一化保证两边一致。调试时把查询解析的中间结果打出来看实体到底匹配到了什么。6. 进阶玩法把抽取流水线做成可配置的跑通主流程之后如果想把这套源码用得更顺手一个实用的进阶方向是把抽取规则和模型参数做成配置文件而不是硬编码在 Python 里。这样换一批文档时不用改代码改配置就行。具体做法是建一个config.yaml把正则模式、实体类型映射、分块大小、Neo4j 连接信息都放进去代码里用yaml.safe_load读。import yaml def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) # config.yaml 示例结构 # patterns: # DATE: \d{4}[-/年]\d{1,2}[-/月]\d{1,2}日? # MONEY: [¥$]?\d(?:\.\d)?\s*(?:万元|亿元|美元|元) # chunk: # size: 400 # overlap: 50 # neo4j: # uri: bolt://localhost:7687 # user: neo4j配置化之后换文档类型只需要改patterns里的正则调分块参数也只改chunk下的数字。另一个进阶点是加一层抽取结果的质量校验统计每次抽取的实体数量、三元组数量、空结果比例如果某份文档抽出来实体数为 0大概率是解析环节出了问题自动标记出来人工复查。验证抽取质量有个简单办法随机抽 20 个三元组人工判断准确率。如果准确率低于 70%说明规则或模型需要调。这个数字不是绝对的课程设计场景下 70% 够用生产场景要 90% 以上。我一般会在流水线最后加一个validate.py跑完自动输出统计报告省得每次手动数。从那以后我每次跑新的文档集都强制先跑一遍小样本验证确认解析和抽取没问题再全量跑。这套源码的价值在于把整条链路串通了你可以在此基础上换更好的 NER 模型、加更细的关系类型、接更顺手的可视化但底层的“解析→抽取→建图→检索”框架不用动。希望帮到你。本文还有配套的精品资源点击获取