Skill不是插件，而是Agent的能力契约
2026/10/27 8:19:48
影刀RPA Neo4j图数据库自动化：知识图谱构建与查询
作者：林焱 | 适用人群：影刀RPA新手 | 难度：★★★★☆
一、什么情况用这个
做数据分析的兄弟，经常需要搞清楚「A公司和B公司是什么关系」「这个人和哪些企业有关联」「这条供应链上有几个环节」。
这种关系型查询，用MySQL写SQL能把人写吐——三度以内的人脉关系，SQL要JOIN三四次，性能堪忧。
图数据库（Neo4j）专门解决这种问题。但Neo4j的Cypher查询语言和传统SQL差别大，很多人不熟悉。用影刀RPA定时从各种数据源采集关系数据 → 自动写入Neo4j → 定时查询生成关系图谱报告，省去手动维护的麻烦。
实际场景：企业股权穿透分析、供应链上下游梳理、社交网络关系挖掘、知识图谱自动更新。
这篇文章能解决的问题：
在影刀Python节点中连接Neo4j
自动从Excel/网页采集关系数据并写入图数据库
用Cypher查询做关联分析
将查询结果导出为Excel报告
二、怎么做
2.1 环境准备
影刀Python节点需要安装neo4j驱动：
拼多多店群自动化报活动上架！
影刀安装目录\python\python.exe-mpipinstallneo4j确保你有Neo4j数据库的访问地址（默认bolt://localhost:7687）和账号密码。
2.2 影刀主流程
┌──────────────────────────────────────────────┐ │ ① 【执行Python代码】 │ │ 连接Neo4j，初始化数据库 │ │ ↓ │ │ ② 【执行Python代码】 │ │ 批量创建节点（公司/人物/产品） │ │ ↓ │ │ ③ 【执行Python代码】 │ │ 批量创建关系（持股/任职/供应/竞争） │ │ ↓ │ │ ④ 【执行Python代码】 │ │ 执行Cypher查询，做关联分析 │ │ ↓ │ │ ⑤ 【写入Excel】 │ │ 将分析结果输出为报表 │ └──────────────────────────────────────────────┘2.3 连接Neo4j并创建节点
步骤①②：节点创建
fromneo4jimportGraphDatabaseimportjson# ===== Neo4j连接配置 =====NEO4J_URI="bolt://localhost:7687"NEO4J_USER="neo4j"NEO4J_PASSWORD="your_password"# ===== 创建驱动 =====driver=GraphDatabase.driver(NEO4J_URI,auth=(NEO4J_USER,NEO4J_PASSWORD))defcreate_company_node(tx,name,industry,registered_capital):"""创建公司节点"""query=""" MERGE (c:Company {name: $name}) SET c.industry = $industry, c.registered_capital = $registered_capital, c.updated_at = datetime() RETURN c """result=tx.run(query,name=name,industry=industry,registered_capital=registered_capital)returnresult.single()defcreate_person_node(tx,name,title,phone=''):"""创建人物节点"""query=""" MERGE (p:Person {name: $name}) SET p.title = $title, p.phone = $phone, p.updated_at = datetime() RETURN p """result=tx.run(query,name=name,title=title,phone=phone)returnresult.single()# ===== 批量创建节点 =====# 从Excel读取的数据（这里用示例数据，实际从影刀Excel节点获取）companies=[{"name":"山海科技有限公司","industry":"信息技术","registered_capital":"1000万"},{"name":"星辰贸易有限公司","industry":"批发零售","registered_capital":"500万"},{"name":"云帆投资控股","industry":"金融投资","registered_capital":"5000万"},]persons=[{"name":"张三","title":"总经理","phone":"13800001111"},{"name":"李四","title":"法人代表","phone":"13800002222"},{"name":"王五","title":"监事","phone":"13800003333"},]withdriver.session()assession:forcompanyincompanies:session.execute_write(create_company_node,company['name'],company['industry'],company['registered_capital'],)print(f"创建公司节点：{company['name']}")forpersoninpersons:session.execute_write(create_person_node,person['name'],person['title'],person['phone'],)print(f"创建人物节点：{person['name']}")print("节点创建完成")2.4 创建关系
步骤③：建立节点之间的关联
defcreate_shareholding_relation(tx,shareholder,company,ratio):"""创建持股关系"""query=""" MATCH (p:Person {name: $shareholder}) MATCH (c:Company {name: $company}) MERGE (p)-[r:持股]->(c) SET r.ratio = $ratio, r.updated_at = datetime() RETURN r """result=tx.run(query,shareholder=shareholder,company=company,ratio=ratio)returnresult.single()defcreate_supply_relation(tx,supplier,customer,product):"""创建供应关系"""query=""" MATCH (s:Company {name: $supplier}) MATCH (c:Company {name: $customer}) MERGE (s)-[r:供应]->(c) SET r.product = $product, r.updated_at = datetime() RETURN r """result=tx.run(query,supplier=supplier,customer=customer,product=product)returnresult.single()defcreate_employment_relation(tx,person,company,position):"""创建任职关系"""query=""" MATCH (p:Person {name: $person}) MATCH (c:Company {name: $company}) MERGE (p)-[r:任职于]->(c) SET r.position = $position, r.updated_at = datetime() RETURN r """result=tx.run(query,person=person,company=company,position=position)returnresult.single()# ===== 批量创建关系 =====relations=[# (类型, 源, 目标, 附加属性)('shareholding','张三','山海科技有限公司','60%'),('shareholding','李四','山海科技有限公司','40%'),('shareholding','云帆投资控股','星辰贸易有限公司','80%'),('employment','张三','山海科技有限公司','总经理'),('employment','李四','山海科技有限公司','法人'),('employment','王五','星辰贸易有限公司','监事'),('supply','山海科技有限公司','星辰贸易有限公司','ERP系统'),]withdriver.session()assession:forrel_type,src,dst,attrinrelations:ifrel_type=='shareholding':session.execute_write(create_shareholding_relation,src,dst,attr)print(f"创建持股关系：{src}→{dst}({attr})")elifrel_type=='employment':session.execute_write(create_employment_relation,src,dst,attr)print(f"创建任职关系：{src}→{dst}({attr})")elifrel_type=='supply':session.execute_write(create_supply_relation,src,dst,attr)print(f"创建供应关系：{src}→{dst}({attr})")print("关系创建完成")2.5 Cypher查询做关联分析
步骤④：核心分析查询
defquery_shareholding_chain(tx,company_name,depth=3):"""查询股权穿透链（向上追溯N层股东）"""query=f""" MATCH path = (shareholder)-[r:持股*1..{depth}]->(c:Company {{name: $company_name}}) RETURN shareholder.name AS 股东, [rel IN relationships(path) | rel.ratio] AS 持股比例, length(path) AS 穿透层数 ORDER BY 穿透层数 """result=tx.run(query,company_name=company_name)return[record.data()forrecordinresult]defquery_related_persons(tx,company_name):"""查询与公司关联的所有人员（任职+持股）"""query=""" MATCH (p:Person)-[r]->(c:Company {name: $company_name}) WHERE type(r) IN ['任职于', '持股'] RETURN p.name AS 姓名, type(r) AS 关系类型, COALESCE(r.position, r.ratio) AS 详情 """result=tx.run(query,company_name=company_name)return[record.data()forrecordinresult]defquery_supply_chain(tx,company_name):"""查询供应链上下游"""query=""" MATCH (upstream)-[r1:供应]->(c:Company {name: $company_name}) OPTIONAL MATCH (c)-[r2:供应]->(downstream) RETURN c.name AS 当前公司, upstream.name AS 上游供应商, r1.product AS 供应产品, downstream.name AS 下游客户, r2.product AS 输出产品 """result=tx.run(query,company_name=company_name)return[record.data()forrecordinresult]# ===== 执行查询 =====analysis_results=[]withdriver.session()assession:# 1. 股权穿透chain=session.execute_read(query_shareholding_chain,'星辰贸易有限公司')analysis_results.append({'分析类型':'股权穿透','数据':chain})# 2. 关联人员persons_related=session.execute_read(query_related_persons,'山海科技有限公司')analysis_results.append({'分析类型':'关联人员','数据':persons_related})# 3. 供应链supply=session.execute_read(query_supply_chain,'山海科技有限公司')analysis_results.append({'分析类型':'供应链','数据':supply})# 输出结果供影刀Excel写入output=json.dumps(analysis_results,ensure_ascii=False)driver.close()2.6 写入Excel报告
影刀中：
【遍历列表】→ 遍历analysis_results 【写入Excel】→ 每个分析类型写一个Sheet页 Sheet1「股权穿透」→ 写入chain数据 Sheet2「关联人员」→ 写入persons_related数据  Sheet3「供应链」→ 写入supply数据三、有什么坑
坑1：MERGE vs CREATE的区别
新手最容易搞混的两个操作：
CREATE：无条件创建，即使节点/关系已经存在也会创建一份新的（产生重复）MERGE：如果不存在则创建，如果存在则匹配已有节点
数据更新的场景，一定用MERGE：
-- ✅ 正确：合并添加，不会重复 MERGE (c:Company {name: '山海科技'}) -- ❌ 错误：每执行一次就多一条重复记录 CREATE (c:Company {name: '山海科技'})坑2：Neo4j连接不关闭导致资源泄漏
影刀流程如果报错中途退出，driver对象不会被关闭，Neo4j服务端会积累大量空闲连接。
TEMU店群矩阵自动化运营核价报活动
确保关闭：
# 方案1：with语句（推荐，自动关闭）withGraphDatabase.driver(...)asdriver:withdriver.session()assession:# ... 操作# 方案2：手动try-finallydriver=GraphDatabase.driver(...)try:# ... 操作finally:driver.close()坑3：Cypher注入风险
如果你在Cypher语句里用字符串拼接拼接用户输入（比如公司名），有注入风险：
# ❌ 危险：字符串拼接query=f"MATCH (c:Company {{name: '{company_name}'}}) RETURN c"# ✅ 安全：参数化查询query="MATCH (c:Company {name: $company_name}) RETURN c"tx.run(query,company_name=company_name)坑4：深度穿透查询性能差
查询3层以上的股权穿透，数据量一大就慢。建议：
- 限制最大深度（不超过5层）
- 对大图建立索引：
CREATE INDEX FOR (c:Company) ON (c.name) - 考虑在Neo4j中预先计算路径，而不是实时查询
坑5：影刀Python节点中安装neo4j驱动失败
neo4j驱动依赖一些系统库，Windows下偶尔安装失败。如果遇到：
# 尝试安装旧版本（兼容性更好）pipinstallneo4j==4.4.0# 或者用py2neo（另一个驱动库，更轻量）pipinstallpy2neopy2neo的API不同，但思路一样，文档很全。
总结：影刀 + Neo4j 的核心价值在于「自动化采集 + 图分析」的组合。影刀负责从各种地方（网页、Excel、API）把关系数据抓回来，Neo4j负责存储和分析关系。MERGE别写成CREATE，查询一定要参数化，连接用完记得关。