做财报分析时，我踩过不少坑：PDF 解析出来的数据散落在文本里，想要回答“这家公司到底参股了哪些新能源企业”，纯靠 RAG 检索经常答非所问。文本检索只能匹配相似片段，但像股权关系、上下游供应链、参股控股这类“关系型问题”，用知识图谱来管理才真正靠谱。第 4、5 集我们完成了财报文本清洗和向量检索基础建设，这一集把重心放在图谱侧：基于 Python 从零搭建实体关系抽取与图谱入库流水线，让 Neo4j 里的节点和关系能够全自动构建。
这篇实战教程会围绕“财报 RAG + 知识图谱”这一主线展开，重点讲清楚实体关系抽取的两种实现思路、Cypher 批量入库的完整代码、以及如何把抽取结果安全地写进 Neo4j。最终你会得到一条可复用的流水线：财报文本进，结构化知识图谱出。
1. 为什么 RAG 场景需要知识图谱
1.1 纯文本 RAG 的局限
常规 RAG 流程是把文档切片、向量化，再基于相似度检索。它在回答“某个指标在哪个章节出现”这类问题时表现不错，但一旦遇到复杂关系问题，比如“A 公司是否为 B 公司的第一大股东”“C 公司与 D 公司是否存在关联交易”，纯文本检索就非常吃力。
原因是向量检索本质上是在做语义相似度匹配，它并不能真正理解两个实体之间的逻辑关系。即便检索到了相关段落，模型依然要从长文本中自行推断关系，既容易漏信息，也容易产生幻觉。
1.2 知识图谱如何补位
知识图谱把实体和关系显式建模，例如“贵州茅台”持股“某销售公司”的 5%，在图中就是一个节点加一条带属性关系。RAG 系统在检索阶段先查询图谱，拿到精确关系路径，再把这些结构化结果作为上下文交给大模型，既能减少幻觉，也能显著提升多跳问答的效果。
财报数据天然适合建图谱：公司、股东、子公司、高管、供应商、客户都是实体；持股、任职、交易、担保则是清晰的关系。把财报中的关系抽取出来建成图谱，是解决复杂问答最直接的方法。
1.3 本文的目标
这一集我们不会只写概念，而是完整落地一条“实体关系抽取→数据清洗→图谱入库”的流水线。你将看到：
- 如何抽取实体与关系并输出标准化 JSON
- 如何通过 Python 连接 Neo4j 并批量写入
- 如何设计可重跑的幂等入库逻辑
- 如何用最简单的办法验证图谱已正确入库
2. 整体架构与流水线设计
先来看整条流水线的数据流向，我们这里不用复杂的图表示意，用文字描述更利于理解：
第 1 步：财报文本输入。可以是清洗后的 TXT，也可以是 PDF 解析后的纯文本，先做段落切分。
第 2 步：实体关系抽取。支持两种方式：基于大模型的结构化抽取，以及基于规则的轻量抽取。二者输出统一结构。
第 3 步：中间格式标准化。统一抽取结果的字段格式，过滤无效实体，避免把空字符串、纯数字噪声写入图谱。
第 4 步：Neo4j 批量入库。使用 MERGE 语法完成“有则更新，无则创建”的幂等写入。
第 5 步：验证与使用。通过 Cypher 查询验证实体关系是否正确挂载，再供 RAG 问答阶段调用。
模块化设计的好处是每一层都可以独立替换。今天先用一个可运行的简化版把闭环跑通，后续再逐步扩展模型或数据源。
3. 环境准备与版本说明
3.1 安装 Neo4j 与基础配置
Neo4j 推荐使用 Desktop 版本管理多个数据库实例，也可以直接使用 Community Server。安装完成后默认地址一般是bolt://localhost:7687，浏览器管理界面是http://localhost:7474。
首次登录时默认用户名是neo4j，密码需要在初始化时自行设置。请妥善保存密码，后面的 Python 驱动连接需要用到。
3.2 Python 环境准备
本文示例以 Python 3.9+ 为例，需要安装 Neo4j 官方 Python 驱动：
pip install neo4j如果后续要接入大模型接口，再根据你实际使用的服务安装对应 SDK。这里我们尽量不引入多余依赖，让代码更透明。
3.3 版本说明
Neo4j 的版本迭代较快，不同小版本的驱动兼容性也存在差异。本文示例基于常见版本编写，重点演示连接、写入和查询思路。如果你使用的是 Neo4j 5.x，官方驱动 5.x 均可兼容；如果使用更早的 4.x 版本，连接参数差别也不大，按实际环境调整即可。
4. 实体关系抽取模块
4.1 确定实体与关系的 Schema
设计抽取 Schema 是第一步，不能等代码写完了再想。针对财报场景，我们定义两类基础实体和两类基础关系，做到够用且清晰：
实体类型一：公司。代表财报中出现的上市公司、子公司、参股公司、客户、供应商等。
实体类型二：人员。代表公司高管、董事、监事等角色。
关系类型一：持股。从“股东公司/人”指向“被投资公司”，属性包括持股比例和持股数量。
关系类型二：任职。从“人员”指向“公司”，属性包括职位名称。
示例：
{ "relations": [ { "source": "A公司", "source_type": "Company", "relation": "持股", "target": "B公司", "target_type": "Company", "properties": { "ratio": 0.15, "amount": "1500万股" } }, { "source": "张三", "source_type": "Person", "relation": "任职", "target": "A公司", "target_type": "Company", "properties": { "position": "董事长" } } ] }这个结构很直观，后续入库时可以直接遍历并执行 Cypher。如果你想抽取供应链关系、产品关系、融资关系，按同样方式扩展即可。
4.2 基于规则的关系抽取示例
为了让你在没有任何大模型 API Key 的情况下也能立刻运行，先写一个基于规则的抽取器。它使用正则表达式匹配财报中常见的“持有……股权”句式。
核心代码如下：
import re import json def extract_relations_by_rule(text): relations = [] # 匹配模式：XX公司持有YY公司Z%股权 pattern = re.compile( r'([\u4e00-\u9fa5A-Za-z0-9]+?公司)\s*(?:持有|参股|入股)\s*' r'([\u4e00-\u9fa5A-Za-z0-9]+?公司)\s*' r'(\d+(?:\.\d+)?)%\s*股权' ) for m in pattern.finditer(text): source, target, ratio = m.group(1), m.group(2), m.group(3) relations.append({ "source": source, "source_type": "Company", "relation": "持股", "target": target, "target_type": "Company", "properties": { "ratio": float(ratio) } }) return relations这个正则适合最基本的表述，实际财报中还有很多变体，比如“持有……股份 5000 万股”“期末持股比例为 10.5%”等。你需要根据数据形态不断补充规则。规则抽取的优势是零成本、可控性强，缺点是覆盖度有限。
4.3 基于大模型的结构化抽取
当财报结构复杂、句式多变时，更推荐用大模型做信息抽取。核心思想是设计一份严格的输出格式要求，让模型只返回 JSON，不返回其他解释文本。
下面给出一个不绑定具体厂商的示意实现，关键逻辑是构造 Prompt 并解析模型输出：
def extract_relations_by_llm(text, llm_func): prompt = f""" 你是一个财报信息抽取助手。 请从下面的财报文本中抽取“公司实体”和“关系”。 公司实体类型包括：Company、Person。 关系类型包括：持股、任职。 输出必须是 JSON 格式，不要输出多余内容，结构如下： {{ "relations": [ {{ "source": "实体名称", "source_type": "Company 或 Person", "relation": "持股 或 任职", "target": "实体名称", "target_type": "Company 或 Person", "properties": {{ "ratio": "持股比例，没有则留空", "amount": "持股数量，没有则留空", "position": "职位名称，没有则留空" }} }} ] }} 财报文本： {text} """ raw_output = llm_func(prompt) # 这里建议做一层 JSON 解析保护 try: start = raw_output.find("{") end = raw_output.rfind("}") + 1 json_str = raw_output[start:end] data = json.loads(json_str) return data.get("relations", []) except Exception as e: print("解析大模型输出失败:", e) return []llm_func是一个函数对象，你可以把它接到 OpenAI 兼容接口、本地部署模型或其他服务上。这样做的好处是业务代码不需要绑定具体模型实现。
5. 图谱入库模块
5.1 连接 Neo4j 驱动
首先封装一个 Neo4j 连接类，便于流水线中复用：
from neo4j import GraphDatabase class Neo4jClient: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def run_query(self, query, params=None): with self.driver.session() as session: result = session.run(query, params) return list(result)连接地址建议从配置文件中读取，不要硬编码在代码里。生产环境中更推荐使用环境变量或配置中心管理。
5.2 创建约束与索引
在批量入库之前，建议先为实体创建唯一约束，避免同一实体重复创建。以下是常用 Cypher：
CREATE CONSTRAINT company_name_unique IF NOT EXISTS FOR (c:Company) REQUIRE c.name IS UNIQUE;CREATE CONSTRAINT person_name_unique IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE;如果使用 Neo4j 4.x，语法是CREATE CONSTRAINT ON (c:Company) ASSERT c.name IS UNIQUE；5.x 则使用上面这种新语法。注意区分版本。
在 Python 中执行：
client.run_query( "CREATE CONSTRAINT company_name_unique IF NOT EXISTS " "FOR (c:Company) REQUIRE c.name IS UNIQUE" ) client.run_query( "CREATE CONSTRAINT person_name_unique IF NOT EXISTS " "FOR (p:Person) REQUIRE p.name IS UNIQUE" )5.3 使用 MERGE 实现幂等入库
MERGE是 Cypher 里的“有则匹配，无则创建”操作，比CREATE更适合数据重复入库场景。这里的关键点是先 MERGE 实体，再 MERGE 关系，避免重复边。
入库代码：
def import_relations(client, relations): create_entity_company_query = """ MERGE (c:Company {name: $name}) """ create_entity_person_query = """ MERGE (p:Person {name: $name}) """ create_relation_hold_query = """ MATCH (a:Company {name: $source}) MATCH (b:Company {name: $target}) MERGE (a)-[r:持股]->(b) SET r.ratio = $ratio, r.amount = $amount """ create_relation_employment_query = """ MATCH (a:Person {name: $source}) MATCH (b:Company {name: $target}) MERGE (a)-[r:任职]->(b) SET r.position = $position """ for rel in relations: source = rel["source"] target = rel["target"] if not source or not target: continue if rel["source_type"] == "Company": client.run_query(create_entity_company_query, {"name": source}) elif rel["source_type"] == "Person": client.run_query(create_entity_person_query, {"name": source}) if rel["target_type"] == "Company": client.run_query(create_entity_company_query, {"name": target}) elif rel["target_type"] == "Person": client.run_query(create_entity_person_query, {"name": target}) props = rel.get("properties", {}) if rel["relation"] == "持股": client.run_query(create_relation_hold_query, { "source": source, "target": target, "ratio": props.get("ratio"), "amount": props.get("amount", "") }) elif rel["relation"] == "任职": client.run_query(create_relation_employment_query, { "source": source, "target": target, "position": props.get("position", "") })注意：上面的写法便于理解，但逐条执行在数据量大时效率不高。更推荐使用UNWIND批量操作，见下一节。
5.4 批量入库优化：UNWIND 用法
当关系数量达到几百上千时，逐条执行 Cypher 会产生大量网络往返。推荐的做法是把数据组装成列表，在一条 Cypher 语句中用UNWIND批量处理。
下面是批量写入“持股”关系的优化版本：
def batch_import_hold_relations(client, relations): filtered = [] for rel in relations: if rel["relation"] != "持股": continue props = rel.get("properties", {}) filtered.append({ "source": rel["source"], "target": rel["target"], "ratio": props.get("ratio"), "amount": props.get("amount", "") }) if not filtered: return query = """ UNWIND $rows AS row MERGE (a:Company {name: row.source}) MERGE (b:Company {name: row.target}) MERGE (a)-[r:持股]->(b) SET r.ratio = row.ratio, r.amount = row.amount """ client.run_query(query, {"rows": filtered})这样一次会话就能完成批量写入，速度提升非常明显。生产级流水线建议都采用这种UNWIND思路。
6. 全自动流水线整合
6.1 定义统一的数据入口
为了保证流水线可扩展，我们把“读取文本”也抽象成函数。真实项目中可能从 PDF、数据库或对象存储读取，这里先用字符串模拟：
def get_financial_report_text(): return """ 华远智能装备股份有限公司持有湖南华远新能源有限公司25%的股权。 李明担任华远智能装备股份有限公司董事长。 华远智能装备股份有限公司持有浙江远动科技有限公司10.5%的股权。 """6.2 编写主流程
把上面几个模块串起来，就是一条最小可运行的自动入库流水线：
def main(): uri = "bolt://localhost:7687" user = "neo4j" password = "your_password" client = Neo4jClient(uri, user, password) # 1. 读取财报文本 text = get_financial_report_text() # 2. 实体关系抽取 llm_relations = extract_relations_by_llm(text, llm_func=None) if not llm_relations: # 如果没有大模型，就使用规则抽取 llm_relations = extract_relations_by_rule(text) print("抽取到关系数量:", len(llm_relations)) # 3. 统一入库 batch_import_hold_relations(client, llm_relations) # 4. 关闭连接 client.close() print("流水线执行完成")示例中llm_func=None时走了规则抽取分支，这样即使没有外部服务也能完整跑通。等你需要更强抽取能力时，再传入大模型的调用函数即可。
6.3 运行与预期输出
运行脚本后，你会看到类似输出：
抽取到关系数量: 2 流水线执行完成这里因为示例文本中只有“持股”关系，所以batch_import_hold_relations处理了两条。随后打开 Neo4j Browser，执行查询：
MATCH p=(a:Company)-[r:持股]->(b:Company) RETURN p LIMIT 10;你应该能看到“华远智能装备股份有限公司”指向两家公司节点的持股关系边。
7. 与 RAG 检索结合验证
7.1 通过图谱查询增强问答上下文
图谱入库不是终点，最终要服务于问答。一种常见做法是：用户提问后，先使用 LLM 将问题转换成 Cypher 查询或关键词，检索图谱获得候选实体和关系，再把结果作为上下文传回大模型。
以“华远智能装备股份有限公司持股了哪些公司？”为例，可以将问题映射为：
MATCH (c:Company {name: '华远智能装备股份有限公司'})-[r:持股]->(target:Company) RETURN c.name AS source, target.name AS target, r.ratio AS ratio结果可以直接格式化为文本：
华远智能装备股份有限公司 持股 湖南华远新能源有限公司，持股比例 25% 华远智能装备股份有限公司 持股 浙江远动科技有限公司，持股比例 10.5%这段文本比财报原稿更精炼，放入 RAG 后既省 token，又能保证关系正确。
7.2 验证抽取质量
建议从三个角度验证入库质量：
第一，节点数量是否符合预期。可以查询所有公司节点数量：MATCH (c:Company) RETURN count(c)。
第二，关系数量是否匹配抽取结果。MATCH ()-[r:持股]->() RETURN count(r)。
第三，抽样检查属性是否正确。例如查看某个关系的ratio字段是否与原文一致。
只有这三层验证都通过，才能说知识图谱构建是可靠的。
8. 常见问题与排查思路
在搭建和运行流水线时，下面几个问题出现频率最高，整理成表格方便快速排查：
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
连接失败The client is unauthorized due to authentication failure | 用户名或密码错误 | 检查 Neo4j 账号密码，重新设置后重试 |
| 驱动连接超时 | 数据库未启动或地址端口错误 | 确认 Neo4j 服务已启动，地址是否为bolt://localhost:7687 |
| Cypher 语法报错 | Neo4j 版本语法差异 | 4.x 使用ASSERT，5.x 使用REQUIRE，按版本调整 |
| 实体重复创建 | 没有唯一约束或使用了 CREATE | 先建唯一约束，再使用 MERGE |
| 抽取结果为空 | 正则未匹配到目标句式 | 打印原始文本检查格式，补充更多匹配模式 |
| 批量写入速度慢 | 逐条执行 Cypher | 改用 UNWIND 批量写入 |
| JSON 解析失败 | 大模型输出夹杂了多余文字 | 截取第一个{到最后一个}，再做解析 |
最让人头疼的其实是“连接失败”。排查时不要只盯密码，还要关注 Neo4j 版本里是否启用了认证、是否允许远程连接。如果是在 Docker 容器中运行，还需要确认端口映射是否正常。
9. 最佳实践与工程建议
9.1 抽取层：宁可空，不可错
实体关系抽取是错误传播的源头。一个错误实体入库，可能导致后续多跳查询全部异常。所以建议在抽取后增加一层过滤规则，例如剔除实体名称为空、实体长度小于 2、实体名称与关系两端相同的记录。
对于不确定的关系，不要强行入库，可以单独记录到日志表，后续人工审核。
9.2 入库层：利用 MERGE 保证幂等
流水线可能因为异常中断而重复执行。使用MERGE而不是CREATE可以从根本上避免重复节点和重复边。同时配合唯一约束，让数据库自己也具备防重能力。
9.3 数据更新：区分增量与重建
如果只是新增财报，建议使用增量入库。如果发现抽取算法升级导致已有数据需要重新构建，那么建议先清理旧图再重建。清理时注意安全风险，务必在测试环境验证后再操作生产库。
9.4 安全与权限
生产环境中 Neo4j 不建议开放公网访问，尤其是使用默认账号时。应当为不同业务场景创建专用账号，并分配最小权限。即使是读取账号，也应仅授权给指定数据库或指定标签。
9.5 日志与监控
每批数据入库后，记录抽取数量、入库数量、失败数量。如果接入大模型抽取，还要记录调用耗时和 token 消耗。这些指标能帮助你判断流水线是否健康，也为后续调优提供依据。
9.6 从最小闭环开始迭代
我建议你不要一上来就设计一套覆盖几十种实体的超级 Schema。先把“公司-持股→公司”这种最核心的关系跑通，验证整个流水线稳定后，再逐步扩展“人员任职”“供应关系”“客户关系”等。这样每一步都能得到验证，也更容易查错。
9.7 RAG 与图谱的分工
要明确一点：知识图谱不是替代向量库，而是补充。向量库擅长模糊语义召回，图谱擅长精确关系查询。实际 RAG 架构中，建议先通过图谱获取精确关系证据，再结合向量检索召回相关片段，最后一起交给大模型生成答案。这样一来，答案既有事实依据，又有上下文支持。
本文完整实现了一条从财报文本到 Neo4j 知识图谱的自动化流水线，核心代码可以复制直接运行。建议你动手跑一遍最小示例，再做三件事：替换成真实财报文本、补充自己的正则规则或接入大模型抽取、把入库结果接入你之前的 RAG 查询链路。图谱建好之后，你会发现很多原先回答不了的复杂关系问题，慢慢地都能给出有依据的答案了。
