简介：本资源为基于Python与Neo4j的知识图谱上传与处理设计源码，面向希望掌握图数据库应用、数据上传与图查询分析的开发者与研究人员，可作为课程设计、毕业项目或工程实践的参考方案。压缩包共25个文件，约27.84MB，以12个XML配置文件和3个IML项目文件为主，用于数据库连接、运行参数与IDE工程结构管理；另含TXT说明、JSON数据、CSV数据源、Git忽略文件、DOCX需求文档及核心PY源文件，覆盖从数据解析、映射、清洗到批量上传Neo4j的完整流程。资源借助Py2neo等库实现图数据的增删改查与复杂图查询分析，目录结构清晰，便于按模块理解工程组织。目前已有473人学习下载，适合需要快速搭建知识图谱上传处理原型、参考数据转换与图数据库交互实现的读者。
1. 从一堆 CSV 到能查的图谱：这套 Python + Neo4j 源码到底解决什么
手里有一批结构化数据，想把它变成能跑 Cypher 查询、能做多跳关系推理的知识图谱，多数人卡在同一个地方：Neo4j 装好了，浏览器也打开了，但数据怎么进去、节点和关系怎么设计、批量导入为什么慢得离谱，全靠现搜。这套基于 Python 的 Neo4j 知识图谱上传与处理设计源码，针对的就是这个环节——它把「读数据 → 建节点 → 建关系 → 批量写入 → 校验」整条链路用 Python 串起来，配合 Neo4j 的官方驱动完成上传与处理。
它适合两类人：一类是刚接触知识图谱构建、想找一个能直接跑通的 Python 工程骨架的开发者；另一类是已经会用 Neo4j 手工敲 Cypher，但需要把上传流程脚本化、可重复执行的从业者。源码本身不绑定具体业务领域，工业场景下的知识图谱设计也好，通用实体关系建模也好，改的是数据映射那几行，骨架不用动。下面按「先跑通、再讲透、最后避坑」的顺序拆。
2. 环境与依赖：Python 驱动、Neo4j 版本和连接参数怎么定
2.1 为什么用官方 neo4j 驱动而不是 py2neo
Python 操作 Neo4j 常见两条路：官方neo4j驱动和第三方py2neo。这套源码走的是官方驱动，理由很实际。官方驱动由 Neo4j 团队维护，和数据库版本的兼容节奏一致，session.execute_write这类事务封装是原生的，批量写入时对连接池的控制更直接。py2neo的 OGM 写法确实顺手，但它在复杂批量场景下容易把事务粒度写粗，一旦数据量上去，内存和超时问题会集中暴露。
选型上还有一层：官方驱动的Driver是线程安全的，连接池由它自己管，你不需要为每个线程 new 一个 driver。这一点在批量上传脚本里很关键，后面第 4 章的并发写入会用到。
安装依赖就一行，但版本要对齐：
# 官方驱动，建议 5.x，与 Neo4j 5.x 服务端匹配 pip install neo4j==5.14.0 # 数据处理常用 pip install pandas参数说明：neo4j驱动 5.x 对应 Neo4j 5.x 服务端；如果你服务端还是 4.4，驱动要降到 4.4 系列，否则握手阶段会报协议不兼容。pandas不是驱动依赖，是源码里读 CSV 用的，换成csv标准库也能跑。
2.2 Neo4j 服务端的三个连接参数
连接信息集中在配置里，源码一般抽成一个config.py或环境变量。核心就三个：
| 参数 | 含义 | 常见取值 |
|---|---|---|
| URI | 连接地址 | bolt://localhost:7687 |
| AUTH | 用户名/密码 | ("neo4j", "你的密码") |
| DATABASE | 目标库 | "neo4j"（社区版默认） |
from neo4j import GraphDatabase URI = "bolt://localhost:7687" AUTH = ("neo4j", "your_password") driver = GraphDatabase.driver(URI, auth=AUTH) driver.verify_connectivity() # 连不上会在这里直接抛错逻辑说明：bolt://是二进制协议端口，比 HTTP 的 7474 更适合批量写入。verify_connectivity()是官方驱动提供的探活方法，放在脚本开头，能在真正写数据前把「地址错、密码错、服务没起」这三类问题挡掉，省得跑到一半才翻车。
参数说明：如果你把 Neo4j 装在另一台机器，URI 里的localhost要换成实际 IP，同时服务端neo4j.conf里的server.default_listen_address得放开，否则会出现「本机浏览器能开、Python 连不上」的情况——这是 neo4j 不能通过 IP 访问的典型原因，第 5 章会细说。
2.3 目录结构先扫一眼
源码包一般长这样，先认清每个文件干什么，改的时候不迷路：
neo4j-kg-upload/ ├── config.py # 连接参数、批量大小 ├── loader.py # 读 CSV/JSON，做字段清洗 ├── graph_builder.py # 节点、关系构建逻辑 ├── uploader.py # 批量写入 + 事务封装 ├── verify.py # 写入后校验 └── data/ └── sample.csv # 示例数据loader管输入，graph_builder管映射，uploader管落库，verify管对账。分层的好处是换数据源只动loader，换图谱模型只动graph_builder，上传逻辑不用碰。
3. 数据建模与上传：节点、关系、批量写入的完整链路
3.1 先定模型再写代码：节点标签和关系类型
知识图谱构建最容易返工的地方不是代码，是模型。拿到一份 CSV，先问三个问题：哪些列是实体、实体分几类、实体之间靠什么连。比如一份「论文-作者-机构」数据，实体是论文、作者、机构，关系是「作者-撰写-论文」「作者-隶属-机构」。
模型定完，落到 Neo4j 里就是标签（Label）和关系类型（Type）。标签用大驼峰，关系类型用大写下划线，这是社区惯例，别用中文标签，Cypher 里写起来要加反引号，麻烦。
# graph_builder.py NODE_LABELS = { "author": "Author", "paper": "Paper", "org": "Organization", } REL_TYPES = { "writes": "WRITES", "belongs": "BELONGS_TO", }逻辑说明：把标签和关系类型集中成字典，是为了后面拼 Cypher 时统一引用，避免字符串散落各处、改一个漏一个。参数说明：标签名不要和 Neo4j 内置保留字冲突，比如User、Role在某些版本里有特殊含义，换成AppUser更稳。
3.2 用 MERGE 而不是 CREATE：避免重复节点
批量上传最怕重复。同一作者出现十次，用CREATE就建十个节点，图谱直接脏掉。正确做法是MERGE，它按匹配条件「有则复用、无则创建」。
def upsert_author(tx, name, org): tx.run( """ MERGE (a:Author {name: $name}) SET a.org = $org """, name=name, org=org )逻辑说明：MERGE (a:Author {name: $name})以name为唯一键匹配，匹配不到才建。SET负责补属性，重复执行不会产生新节点。参数说明：MERGE的匹配字段最好加唯一约束，否则并发下仍可能建重，约束写法见 3.4。
3.3 关系写入：先匹配两端再建边
关系不能凭空建，两端节点必须已存在。标准写法是先MATCH两端，再MERGE关系：
def link_author_paper(tx, author, title): tx.run( """ MATCH (a:Author {name: $author}) MATCH (p:Paper {title: $title}) MERGE (a)-[:WRITES]->(p) """, author=author, title=title )逻辑说明：两个MATCH分别定位作者和论文，MERGE建关系。如果某个MATCH没命中，整条语句不产生任何写入，也不会报错——这是静默失败，第 5 章会讲怎么排查。参数说明：关系方向按语义定，(a)-[:WRITES]->(p)表示作者指向论文，查询时顺着方向走。
3.4 批量写入：UNWIND + 事务函数
逐条tx.run在几千条数据上还能忍，上万条就明显慢。官方驱动推荐的批量姿势是UNWIND，把一批数据当列表传进去，一条 Cypher 处理一批。
def batch_upsert_authors(tx, rows): tx.run( """ UNWIND $rows AS row MERGE (a:Author {name: row.name}) SET a.org = row.org """, rows=rows ) with driver.session(database="neo4j") as session: session.execute_write(batch_upsert_authors, author_rows)逻辑说明：UNWIND $rows把参数里的列表展开成多行，MERGE对每行执行一次。execute_write是官方驱动的事务封装，内部带自动重试，遇到瞬时冲突会自己重试，比手写begin_transaction省心。参数说明：rows是字典列表，字段名要和 Cypher 里的row.xxx对齐；单批大小建议 1000～5000，太大内存吃紧，太小事务开销占比高。
写入前给唯一键加约束，能同时提速和防重：
CREATE CONSTRAINT author_name IF NOT EXISTS FOR (a:Author) REQUIRE a.name IS UNIQUE;逻辑说明：唯一约束会让MERGE走索引查找，而不是全表扫，批量写入速度差好几倍。参数说明：约束名自定义但要唯一，IF NOT EXISTS保证重复执行不报错。
3.5 写入后校验：别信「没报错就是成功」
脚本跑完不报错，不代表数据进去了。校验这一步不能省：
def count_nodes(session, label): result = session.run(f"MATCH (n:{label}) RETURN count(n) AS c") return result.single()["c"] with driver.session(database="neo4j") as session: print("Author:", count_nodes(session, "Author")) print("Paper:", count_nodes(session, "Paper"))逻辑说明：按标签统计节点数，和源数据去重后的数量对一下，对不上就说明有静默失败。参数说明：f"MATCH (n:{label})"里的label来自代码常量，不要拼接用户输入，避免注入。
4. 性能与并发：批量大小、索引和连接池怎么调
4.1 批量大小不是越大越好
UNWIND的批大小直接影响吞吐。太小，网络往返和事务提交次数多；太大，单事务内存占用高，还可能触发服务端事务超时。经验区间是 1000～5000，具体看单行字段多少。
BATCH_SIZE = 2000 def chunked(rows, size): for i in range(0, len(rows), size): yield rows[i:i + size] for batch in chunked(author_rows, BATCH_SIZE): session.execute_write(batch_upsert_authors, batch)逻辑说明：chunked把大列表切片，逐批提交。参数说明：BATCH_SIZE从 2000 起调，观察服务端dbms.memory和脚本耗时，往上下各试一档，找到拐点。
4.2 索引和约束是性能的地基
没有索引的MERGE是全表扫，数据量一上来就是灾难。除了唯一约束，常用查询字段也建议建索引：
CREATE INDEX paper_title IF NOT EXISTS FOR (p:Paper) ON (p.title);逻辑说明：索引让按title匹配的查询走 B+ 树，而不是逐节点扫。参数说明：索引不是越多越好，每个索引都占存储、拖慢写入，只给高频查询字段建。
4.3 连接池与并发写入的边界
官方驱动的Driver自带连接池，默认上限 100。多线程写入时，共享一个driver实例即可，不要每个线程 new 一个：
from concurrent.futures import ThreadPoolExecutor def write_batch(batch): with driver.session(database="neo4j") as session: session.execute_write(batch_upsert_authors, batch) with ThreadPoolExecutor(max_workers=4) as pool: pool.map(write_batch, chunked(author_rows, BATCH_SIZE))逻辑说明：driver全局一个，session每批一个，session用完即关，连接归还池子。参数说明：max_workers不要超过服务端能承受的并发事务数，社区版并发能力有限，4～8 比较稳，盲目开到 32 反而因锁竞争变慢。
提示：并发写入前先确认唯一约束已建好，否则多线程
MERGE仍可能建出重复节点。
5. 避坑与排查：连接、静默失败、内存这几类问题
5.1 现象：浏览器能打开 Neo4j，Python 连不上
原因：Neo4j 默认只监听localhost，浏览器在本机访问没问题，Python 从别的机器连就被拒。这是 neo4j 不能通过 IP 访问的最常见原因。
解决：改服务端neo4j.conf，把server.default_listen_address设为0.0.0.0，重启服务。同时确认防火墙放行 7687 端口。改完先用verify_connectivity()探活再跑数据。
5.2 现象：脚本没报错，但节点数比预期少
原因：关系写入里的MATCH没命中，整条语句静默跳过。比如作者节点还没写就先写关系，MATCH (a:Author ...)匹配为空，MERGE关系自然不执行。
解决：调整执行顺序，先写所有节点、再写关系。或者在关系写入后统计关系数，和预期对账。养成「每类写入后 count 一次」的习惯。
5.3 现象：批量写入跑到一半报内存或超时
原因：单批数据太大，或者没建索引导致MERGE全表扫，事务持有时间过长。
解决：先把BATCH_SIZE降到 1000 试；再检查唯一约束和索引是否建全。服务端侧可以适当调大事务超时，但治本还是减小批和加索引。
5.4 现象：重复执行脚本后节点翻倍
原因：用了CREATE而不是MERGE，或者MERGE的匹配字段不唯一（比如用可空的org当键）。
解决：统一改MERGE，匹配字段选业务上真正唯一的列，并加唯一约束兜底。已经脏了的数据，用 Cypher 按属性分组去重后再重建。
5.5 现象：中文属性写入后查询乱码或匹配不上
原因：CSV 读取时编码没指定，默认按系统编码解析，中文变乱码。
解决：pandas.read_csv(path, encoding="utf-8")显式指定编码；如果源文件是 GBK，先转 UTF-8 再入库。写入前打印几行确认字段值正常。
6. 进阶：把上传脚本改成可复用的导入管道
跑通单次上传只是起点，真正省事的是把它做成可重复执行的导入管道。我一般会加三样东西：配置外置、幂等保证、增量标记。
配置外置，就是把 URI、账号、批大小、数据路径全挪到环境变量或config.yaml，脚本本身不含硬编码。这样换环境只改配置，不动代码：
import os URI = os.getenv("NEO4J_URI", "bolt://localhost:7687") AUTH = (os.getenv("NEO4J_USER", "neo4j"), os.getenv("NEO4J_PASSWORD")) BATCH_SIZE = int(os.getenv("BATCH_SIZE", "2000"))逻辑说明：os.getenv带默认值，本地开发不设环境变量也能跑，生产环境用环境变量覆盖。参数说明：密码不要写进代码仓库，用环境变量或密钥管理。
幂等保证靠MERGE+ 唯一约束，前面讲过，这里补一个增量思路：给每个节点加updated_at时间戳，重跑时只处理源数据里变更过的行。这样全量重跑变成增量更新，大数据量下差别很大。
def upsert_with_ts(tx, rows): tx.run( """ UNWIND $rows AS row MERGE (a:Author {name: row.name}) SET a.org = row.org, a.updated_at = datetime() """, rows=rows )逻辑说明：datetime()是 Cypher 内置函数，写入当前时间。参数说明：增量判断在 Python 侧做，比对源数据的修改时间和库里updated_at，只挑变更行进rows。
验证方法上，我习惯在管道末尾加一段「对账查询」，把源数据行数和图谱节点数、关系数三方对齐，对不上就退出码非零，方便挂到调度里。
| 校验项 | Cypher | 预期 |
|---|---|---|
| 节点总数 | MATCH (n) RETURN count(n) | 等于去重后实体数 |
| 关系总数 | MATCH ()-[r]->() RETURN count(r) | 等于关系记录数 |
| 孤立节点 | MATCH (n) WHERE NOT (n)--() RETURN count(n) | 接近 0 |
最后一条孤立节点检查特别有用，数量异常往往意味着关系写入那步有MATCH没命中。从那以后我每次改完映射逻辑，都强制先跑一遍对账再宣布导入完成，省了不止一次回头返工。希望帮到你。
本文还有配套的精品资源，点击获取
