自动化构建Neo4j知识图谱的财报RAG流水线实践

最近在搭财报分析的 RAG 系统时，很多人卡在同一个点上：文档能切、能向量化、能召回，但回答“这家公司的股东是谁、和哪家子公司存在关联、营收增长靠什么驱动”这类结构化问题时，纯向量检索经常答得含糊。

原因也很直接，财报里的实体关系和数字逻辑是强结构化的，适合放进知识图谱。如果你正在做财报 RAG，或者准备把 Neo4j 接进自己的知识库体系，这篇文章可以直接看下去。

这里要展开的是财报 RAG + 知识图谱系列第 6 集：全自动构建 Neo4j 知识图谱，从实体关系抽取到图谱入库的完整流水线。核心内容包括怎么抽取财报中的公司实体、人物、股权关系、业务关联，怎么通过大模型自动生成实体关系三元组，怎么把这些三元组批量写入 Neo4j，以及最后怎么在图谱里做查询和校验。目标是让图谱构建这条链路从手工整理变成自动化流水线，跑完一遍就能直接在 Neo4j 里看到一张可以查询的财报关系网络。

1. 核心能力速览

这套流水线不是简单地把文档导入 Neo4j，而是把非结构化的财报正文，抽成结构化的三元组，再做图谱入库。整个过程分三段：文本预处理、实体关系抽取、图谱写入。三段各自独立，任务失败可以单独重跑。

2. 适用场景与使用边界

2.1 适合什么场景

从实际搭建经验看，这套流水线最适用的场景有三个。

第一个是财报深度分析。一份财报里有公司基本信息、股东持股、高管变动、主营业务、上下游供应商、收入构成、风险提示。这些内容天然适合用图谱表达。比如“控股股东是深圳某资本”“三季报营收同比增长 18%”“子公司经营范围包含新能源”等，表达成节点和关系后，回答归因类问题时远比向量全文检索直观。

第二个是同一集团下多份文档的关联分析。比如你要分析一家上市公司及其子公司的历年财报，文档之间存在的实体重叠非常多。用图谱可以直接查“某子公司某年的营收增长和母公司之间的关系”，而纯 RAG 很难跨文档做这种关联推理。

第三个是构建企业级 RAG 知识库。把知识图谱作为 RAG 的结构化索引层，与向量检索形成双路召回。图谱负责回答关系型问题，向量库负责回答开放型问题，两者互补。

2.2 不推荐的场景

财报 PDF 本身是扫描件或图片型文档时，需要先接 OCR 再做文本抽取，流水线链路会变长，图谱质量会被 OCR 误差直接干扰。这种情况需要先单独建 OCR 预处理任务，不应该直接进实体关系抽取。

体量非常小的需求，例如只分析一份几十页的财报，手工整理关系可能比搭流水线更快。流水线的价值在文档数量多、需要反复更新场景下才明显。

2.3 合规边界

财报中可能包含未公开披露的信息、敏感数据、个人隐私（如高管身份证号、联系方式）。在做实体关系抽取和入库前，必须确认数据来源的合法性，建议只使用公开披露的年报、季报或已经获得授权的材料。涉及个人实体的信息要注意脱敏，图谱入库时对敏感属性做删除或匿名化。搭建系统时，在配置里预留字段过滤和属性白名单，避免把多余信息写进图谱。

3. 环境准备与前置条件

从零搭建这条流水线，需要准备以下几部分环境。

3.1 软件环境

3.2 Neo4j 安装与启动

Neo4j 有多种安装方式。这里给两种常用路径。

路径一：Neo4j Desktop。

到 Neo4j 官网下载 Desktop 安装包，安装后创建本地数据库，设置密码，启动。这是最省事的方式，适合本机调试。有个需要留意的地方：给数据库设置的密码和账号会用于后续 Python 连接，建议单独建一个专用账号，避免使用默认的 neo4j 账号做业务写入。

路径二：Docker 启动。

如果本机已经装好 Docker，用下面的命令可以快速起一个 Neo4j 实例。

docker run -d \ --name neo4j-finance \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTH=neo4j/yourpassword \ -e NEO4J_PLUGINS='["apoc"]' \ -v /data/neo4j:/data \ neo4j:5.26.0-community

启动后，浏览器访问http://localhost:7474，使用账号neo4j和设置的密码登录，就能看到 Neo4j Browser 管理界面。7687端口是 Bolt 协议连接端口，Python 驱动连接时会用到。

3.3 Python 依赖

建议单独创建一个虚拟环境，避免和其他项目依赖互相污染。

python -m venv venv_finance source venv_finance/bin/activate # Windows 下使用 venv_finance\Scripts\activate

安装核心依赖。

pip install neo4j openai pydantic tiktoken

如果文档来源是 PDF，还需要安装 PDF 解析相关库。

pip install pypdf pdfplumber

这里把pydantic单独列出来，是因为实体关系抽取结果最好用 Pydantic 模型做结构约束。大模型返回的 JSON 经过 Pydantic 校验后，再进入 Neo4j 写入，可以大幅减少脏数据入库的情况。

4. 整体流水线设计

完整的图谱入库流水线可以拆成四个阶段。

4.1 阶段一：财报文本解析

输入是一份或多份财报 PDF，输出是分章节的纯文本片段。

这部分要处理的核心问题是：PDF 转出来的文本往往包含页眉、页脚、目录、大量空白字符，需要清洗。财报的章节结构比较固定，通常包括重要提示、公司简介、主营业务、经营情况讨论与分析、董事会报告、财务数据、股份变动、股东情况、董监高信息等。

实际处理时，可以根据章节标题将财报拆分成多个片段，而不是把整份财报塞给模型。这样做有两个收益：一是降低单次 LLM 调用的 token 成本，二是实体关系抽取时上下文更聚焦，减少实体歧义。

建议清洗步骤：

• 删除页眉页脚；
• 合并断行；
• 过滤目录页；
• 按章节标记切分。

4.2 阶段二：Schema 定义

实体关系抽取不能直接让大模型“自由发挥”，必须提前定义好 Schema。Schema 决定了图谱的类型体系，定义得好不好直接影响入库质量和查询便利度。

财报知识图谱的 Schema 一般需要覆盖以下实体类型：

关系类型定义：

Schema 越清晰，抽取准确率越高。比如同样是“持股”关系，如果 Schema 里同时定义了HOLDS_SHARE和INVESTED，模型就有可能在两者之间摇摆。建议初期 Schema 控制在一张表内，包含 5 种左右实体类型和 5 种左右关系类型，通过测试后再逐步扩展。

4.3 阶段三：实体关系抽取

这是整个流水线的核心模块。目标是从财报文本片段中抽取出符合 Schema 的三元组。

抽取方式采用 LLM + JSON Schema 约束。每次给模型一段文本，让模型输出一个 JSON 数组，数组中每个元素包含source_entity、relation、target_entity、confidence、evidence字段。

这样做的好处是：

• 结果可以直接被 Pydantic 校验；
• 可以保留置信度和证据文本，方便后续人工复核；
• 批量任务时可以通过字段判断哪些三元组需要过滤。

4.4 阶段四：图谱入库与去重

抽取完成后的三元组，通过 Neo4j Python Driver 写入 Neo4j。写入的关键是使用MERGE而不是CREATE。

MERGE会先检查节点或关系是否已存在，存在则不重复创建，不存在才创建。这对于批量任务非常重要，因为同一个实体可能在不同财报片段中反复出现，如果使用CREATE，会产生大量重复节点，导致图谱出现数据膨胀和查询混乱。

入库流程中还建议加一步去重逻辑：在写入前，用实体的唯一标识（如公司名 + 股票代码）做一次预检查。如果实体已经存在，则只更新属性，不新增节点。

5. 实体关系抽取实现

下面给出可运行的抽取代码示例。

5.1 定义 Pydantic 模型

from typing import List, Optional from pydantic import BaseModel, Field class Entity(BaseModel): entity_type: str = Field(description="实体类型，如 Company、Person") name: str = Field(description="实体名称，必须是文本中出现的原文或标准化名称") properties: dict = Field(default_factory=dict, description="实体属性") class Relation(BaseModel): source: str = Field(description="头实体的名称") source_type: str = Field(description="头实体类型") relation: str = Field(description="关系类型，必须在 Schema 定义范围内") target: str = Field(description="尾实体的名称") target_type: str = Field(description="尾实体类型") confidence: float = Field(default=1.0, ge=0.0, le=1.0, description="置信度") evidence: Optional[str] = Field(default=None, description="抽取依据的原文片段") class GraphExtractionResult(BaseModel): entities: List[Entity] = Field(description="抽取出的实体列表") relations: List[Relation] = Field(description="抽取出的关系列表")

5.2 构建抽取 Prompt

KG_SCHEMA_PROMPT = """ 你是一个财报结构化分析专家。请从下面的财报文本中抽取实体和关系。 实体类型定义： - Company: 公司实体，包含公司名称、股票代码、行业属性 - Person: 人物实体，包含姓名、职位 - Shareholder: 股东实体，包含股东名称、持股比例 - Product: 产品或业务实体，包含名称、收入和增长情况 - Event: 事件实体，包含事件名称、日期、描述 关系类型定义： - HOLDS_SHARE: 持有股份，从股东到被持股公司 - SENIOR_MANAGER: 高管任职，从人到公司 - MAIN_BUSINESS: 主营业务，从公司到业务/产品 - SUPPLIES: 供应关系，从供应商到采购方 - INVESTED: 投资关系，从投资方到被投方 抽取要求： 1. 只抽取文本中明确存在的实体和关系，不要进行主观推测 2. 实体名称尽量使用原文，必要时归一化 3. 如果持股比例、收入金额等数值信息在原文中出现，请在 properties 中记录 4. 不要输出不存在的实体关系 5. 输出必须是 JSON 数组格式，可以解析为 Python json 对象 财报文本片段： {text_chunk} 请只输出 JSON，不要输出解释。 """.strip()

5.3 调用 LLM 抽取

import json import openai client = openai.OpenAI( api_key="your-api-key", base_url="your-api-base-url" # 兼容 OpenAI 格式的服务 ) def extract_graph_from_text(text_chunk: str) -> GraphExtractionResult: prompt = KG_SCHEMA_PROMPT.format(text_chunk=text_chunk) response = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是一个严格按 JSON Schema 输出的结构化抽取助手。"}, {"role": "user", "content": prompt}, ], temperature=0.1, response_format={"type": "json_object"}, # 如果接口支持 ) content = response.choices[0].message.content data = json.loads(content) # 兼容不同返回结构 if isinstance(data, list): payload = {"entities": [], "relations": []} for item in data: if "source" in item: payload["relations"].append(item) data = payload return GraphExtractionResult(**data)

这里有几个重要细节：

temperature要调低，最好在 0.1 到 0.2 之间。实体关系抽取不是创意生成任务，低温度可以显著降低模型“发挥”出来的不实实体比例。

response_format如果服务支持，直接约束为 JSON 输出，可以避免解析失败。如果接口不支持，需要在 Prompt 里强调“只输出 JSON”，并在代码里做容错。

部分模型返回的 JSON 结构不统一，有时直接返回关系数组，有时返回封装后的{"entities": [...], "relations": [...]}对象，代码里要做一层兼容转换。

5.4 批量抽取

单份财报有几十个片段，需要循环调用。批量抽取时建议加一个简单的重试机制。

import time def batch_extract(text_chunks: list[str], max_retries: int = 3) -> List[GraphExtractionResult]: results = [] for idx, chunk in enumerate(text_chunks): for attempt in range(max_retries): try: result = extract_graph_from_text(chunk) results.append(result) print(f"[{idx+1}/{len(text_chunks)}] success, " f"entities={len(result.entities)}, relations={len(result.relations)}") break except Exception as e: print(f"[{idx+1}] attempt {attempt+1} failed: {e}") time.sleep(2 ** attempt) # 指数退避 else: print(f"[{idx+1}] failed after {max_retries} retries, skip") return results

批量任务一定要有日志。每条片段处理成功或失败都输出一条日志，方便排查是哪一段文本导致的问题，也方便后续做断点续跑。

6. 知识图谱入库流水线

抽取结果出来后，需要写入 Neo4j。入库逻辑可以用一个独立的模块实现。

6.1 Neo4j 连接与驱动封装

from neo4j import GraphDatabase class Neo4jWriter: def __init__(self, uri: str, user: str, password: str): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def write_extraction(self, result: GraphExtractionResult): with self.driver.session() as session: session.execute_write(self._create_entities, result) session.execute_write(self._create_relations, result) @staticmethod def _create_entities(tx, result: GraphExtractionResult): for entity in result.entities: entity_type = entity.entity_type props = dict(entity.properties) props["name"] = entity.name tx.run( f""" MERGE (n:`{entity_type}` {{name: $name}}) SET n += $props """, name=entity.name, props=props, ) @staticmethod def _create_relations(tx, result: GraphExtractionResult): for rel in result.relations: tx.run( f""" MATCH (a {{name: $source}}) MATCH (b {{name: $target}}) MERGE (a)-[r:`{rel.relation}`]->(b) SET r.confidence = $confidence SET r.evidence = $evidence """, source=rel.source, target=rel.target, confidence=rel.confidence, evidence=rel.evidence or "", )

注意一个点：上面的_create_entities方法里，实体节点上直接设置了name属性。MERGE (n:Company {name: $name})会以name作为实体唯一标识，这意味着同名的不同实体可能会发生冲突。这在真实场景中是一个真实存在的问题，比如两个人同名，或者两个子公司重名。

实际的工程方案通常是在实体属性中增加一个唯一业务键，比如公司用code字段（股票代码或统一社会信用代码），人则用name + position + company组合。使用唯一业务键作为MERGE的匹配条件，更准确。

6.2 入库去重策略

当同一条关系被多个文本片段重复抽取到时，不能重复建立边。上面的做法是利用MERGE保证不会重复创建关系。如果需要保留多条证据来源，可以在关系上维护一个evidence_list属性。

@staticmethod def _create_relations_with_evidence(tx, result: GraphExtractionResult): for rel in result.relations: tx.run( """ MATCH (a {name: $source}) MATCH (b {name: $target}) MERGE (a)-[r:REL]->(b) ON CREATE SET r.confidence = $confidence, r.evidence = [$evidence] ON MATCH SET r.evidence = CASE WHEN $evidence IN r.evidence THEN r.evidence ELSE r.evidence + $evidence END """, source=rel.source, target=rel.target, confidence=rel.confidence, evidence=rel.evidence or "", )

这种方式的好处是：同一关系的多条证据可以累积保存，后续如果要做人工审核，审计链路是完整的。

6.3 主流水线入口

把上面的模块串起来，完整的流水线脚本如下。

import os from pathlib import Path # 1. 解析 PDF 获取文本片段 def parse_finance_pdfs(pdf_dir: str) -> list[str]: import pdfplumber chunks = [] for pdf_path in Path(pdf_dir).glob("*.pdf"): with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages[:60]: # 示例限制页数 text = page.extract_text() if text: # 简单按 1500 字切块，实际可根据章节边界切分 text = " ".join(text.split()) for i in range(0, len(text), 1500): chunks.append(text[i:i+1500]) return chunks # 2. 抽取实体关系 def extract_all(chunks: list[str]) -> List[GraphExtractionResult]: return batch_extract(chunks) # 3. 写入 Neo4j def write_to_neo4j(results: List[GraphExtractionResult]): writer = Neo4jWriter( uri=os.getenv("NEO4J_URI", "bolt://localhost:7687"), user=os.getenv("NEO4J_USER", "neo4j"), password=os.getenv("NEO4J_PASSWORD", "yourpassword"), ) try: for idx, result in enumerate(results): writer.write_extraction(result) print(f"[{idx+1}/{len(results)}] written") finally: writer.close() if __name__ == "__main__": chunks = parse_finance_pdfs("./finance_pdfs") print(f"parsed {len(chunks)} chunks") results = extract_all(chunks) write_to_neo4j(results) print("done")

流水线拆分得很清楚。后续要单独重跑某个阶段，只需要调用对应函数。比如模型更新后不想重新解析 PDF，可以直接从extract_all这一步开始跑。

7. 功能测试与效果验证

流水线跑完，怎么判断图谱构建成功？建议按下面的验证流程走一遍。

7.1 检查入库统计

在 PyCharm 或命令行看完日志后，到 Neo4j Browser 执行下面的 Cypher 查询，看节点和关系数量。

MATCH (n) RETURN count(n) AS node_count;

MATCH ()-[r]->() RETURN count(r) AS relation_count;

如果节点数或关系数是 0，说明实体关系抽取或入库环节出了问题，需要逐步排查。

7.2 查询公司节点

MATCH (c:Company) RETURN c.name, c.code LIMIT 20;

这条查询确认公司实体是否正常入库。如果c.name为空，说明节点属性写入有问题，回头检查_create_entities里的属性赋值逻辑。

7.3 查询股权关系

MATCH (s:Shareholder)-[r:HOLDS_SHARE]->(c:Company) WHERE s.name CONTAINS '控股' RETURN s.name, c.name, r.shareholding_ratio LIMIT 50;

这条查询用来验证关系是否按预期方向建立。这里需要注意方向问题：HOLDS_SHARE从股东指向被持股公司。在实际抽取中，模型很可能把方向搞反，所以入库时要检查。

如果发现方向反了，可以在抽取结果写入前做一次规则校验，也可以使用 Cypher 的ORDER BY和可视化检查来确定问题。

7.4 查询某个具体实体的完整子图

MATCH (n {name: '贵州茅台'})-[r]-(m) RETURN n, r, m LIMIT 100;

在 Neo4j Browser 中执行这条查询，可以看到贵州茅台这个节点相关的所有关系和节点。这一步能直观看到图谱构建效果，也能快速发现数据异常，比如某个实体出现了几十次重名节点，或者关系明显不相关的连接。

7.5 断点重跑验证

删掉部分错误关系后重新执行入库，验证MERGE是否正常工作。

MATCH (:Company)-[r:SUPPLIES]->(:Company) DELETE r;

删掉SUPPLIES关系后，重新跑一次入库脚本，确认这些关系重新出现且没有重复创建。

7.6 常见验证失败现象

8. 图谱与 RAG 的联动设计

实体关系入库只是第 6 集要完成的重点。最终目标还是服务于财报 RAG。图谱与 RAG 的联动可以按两个方向设计。

8.1 图谱作为结构化召回源

用户在 RAG 问答系统里问“这家公司前十大股东是谁”，系统先解析出问题中的公司实体，然后走图谱查询。

MATCH (s:Shareholder)-[r:HOLDS_SHARE]->(c:Company) WHERE c.name = $company_name RETURN s.name AS shareholder, r.shareholding_ratio AS ratio ORDER BY ratio DESC

查询结果作为上下文拼进 Prompt，再让大模型组织自然语言回答。这个方案的优点是可解释性强，答案完全基于图库中的结构化数据，基本不会出现幻觉。

8.2 图谱与向量检索双路召回

更完整的方案是把图谱召回和向量检索并行：向量检索负责找相关文档片段，图谱召回负责找结构化关系，两者结果融合后一起送去生成答案。

def retrieve_rag_context(question: str): # 1. 向量召回片段 text_chunks = vector_search(question, top_k=5) # 2. 图谱召回结构化信息 entities = extract_entities_from_question(question) graph_triples = [] for entity in entities: triples = search_neo4j_entity(entity) graph_triples.extend(triples) return {"text_chunks": text_chunks, "graph_triples": graph_triples}

这种方式对“股东是谁”“与某公司什么关系”“去年营收多少”这类问题效果提升非常明显。但也需要注意：图谱召回的质量完全依赖图谱数据质量。图谱里如果存在错误关系，回答会错得更“自信”，所以实体关系抽取的置信度阈值设置很重要。

建议在入库时设置一个置信度阈值，低于阈值的三元组先进入“待审核”队列，不直接用于 RAG 回答。

8.3 面向 Agent 的图谱工具

更进一步，可以把图谱查询封装成工具函数，供 Agent 调用。

def query_shareholder_ratio(company_name: str) -> str: with driver.session() as session: result = session.run( """ MATCH (s:Shareholder)-[r:HOLDS_SHARE]->(c:Company) WHERE c.name = $name RETURN s.name AS shareholder, r.shareholding_ratio AS ratio """, name=company_name ) rows = [{"shareholder": r["shareholder"], "ratio": r["ratio"]} for r in result] return json.dumps(rows, ensure_ascii=False)

这样 RAG 系统或者 Agent 就可以通过函数调用直接获取结构化答案，不需要每次重新组织 Cypher。长期维护时只需要完善函数定义和参数校验即可。

9. 资源占用与性能观察

流水线的性能瓶颈通常不在 Neo4j，而在实体关系抽取阶段。

9.1 不同阶段的资源占用

如果使用云端 LLM API，实体关系抽取阶段的耗时取决于文本长度和接口响应速度。建议对单片段长度做限制，1500 字以内是比较稳妥的区间。片段太长会导致输出 JSON 变大，增加解析失败概率；太短则抽取出的关系稀疏，上下文不足容易漏抽。

9.2 如何加速批量任务

• 增加并发请求数，但要注意 API 限流；
• 使用异步 HTTP 客户端，如httpx.AsyncClient；
• 入库时使用 Neo4j 的批量事务提交，而不是每一条三元组单独一个事务。

伪代码示例：

def write_batch(tx, relations): for rel in relations: tx.run(...)

with driver.session() as session: for i in range(0, len(all_relations), 100): batch = all_relations[i:i+100] session.execute_write(lambda tx: write_batch(tx, batch))

将事务按 100 条左右批量提交，可以显著降低提交次数，提高写入吞吐量。

10. 常见问题与排查方法

11. 最佳实践与使用建议

从系统搭建和后续维护角度，有几个工程建议值得保留。

11.1 实体命名归一化

实体名称不一致是图谱项目最常见的脏数据来源。同一个公司，在财报正文里可能有“贵州茅台酒股份有限公司”“贵州茅台”“茅台集团”等多种写法。实体关系抽取阶段需要做名称归一化。

一个可落地的做法是：在 Prompt 中增加“实体名称标准化规则”，声明公司实体优先使用全称，如果没有全称则使用文本中的标准简称。同时在入库前再做一次字典映射。

11.2 保留原始证据链

每条关系都应保留evidence字段或原始文本片段引用。这样图谱中的每一条关系都能溯源到原文，方便人工审核和最终 RAG 回答校验。

11.3 设置置信度阈值

抽取完成后，把低于阈值的三元组放到待审核列表，可以用 CSV 导出后人工快速过一遍。不要直接入库参与 RAG 回答。

11.4 先小样本验证再全量

建议先拿一份财报中的“股东情况”章节作为测试样本，跑通整个流水线，确认 Neo4j 中的图谱结构和查询语句没有问题，再扩展到全量财报。全量跑完后，再按股票代码或财报年份分目录重新入库，方便出问题时局部重跑。

11.5 数据备份与版本管理

Neo4j 数据文件需要定期备份。建议在每次批量入库前做一次数据导出，防止错误的数据覆盖原有的正确图谱。

# Neo4j Admin 备份示例（需要进入容器或安装目录执行） neo4j-admin database dump neo4j --to-path=/backup/

11.6 合规与安全

财报数据虽然以公开披露为主，但仍要注意：入库过程中不要写入未公开信息；涉及个人实体的敏感属性（手机号、身份证号、家庭住址）在抽取阶段就要过滤；接口服务在开放给外部使用前，需要加上访问权限控制和审计日志。当前项目如果只在本机调试，建议 Neo4j 只监听 localhost，不要暴露到公网。

12. 总结与下一步

到这一步，“从零搭建财报 RAG + 知识图谱”第 6 集的完整链路已经打通：PDF 解析、Schema 定义、实体关系抽取、Neo4j 图谱入库、抽取结果验证、RAG 联动设计。流水线最值得先验证的部分是实体关系抽取的准确性和 Neo4j 入库的幂等性。先用一份公开财报的“股东信息”章节跑通，再扩展全量，是这个项目最快的上手路径。

最容易踩的坑集中在两个点：LLM 抽取结果不稳定导致实体名称漂移，以及关系方向写反。前者靠 Schema 约束和归一化解决，后者靠入库前的规则校验解决。建议后续把这两块单独做成可复用模块，同一套代码可以直接迁移到招股书、研报、公告等类似文档场景。

下一步扩展方向可以考虑：把实体关系抽取的提示词模板做成配置化；在 Neo4j 中建立全文索引；把图谱查询封装为 RAG Agent 可调用的工具函数；增加人工审核界面，对低置信度三元组做人工确认。图谱数据稳定后，整个财报 RAG 系统的结构化问答能力会有一个明显提升。