财报分析里有一个很常见的问题文本明明都在但回答“A公司和B公司之间什么关系”“谁持有谁股权”“营收增长率是多少”这类关系型问题时向量检索却总是答不准。原因在于财报是实体密集、关系密集的文本而纯向量 RAG 擅长的是“找相似段落”不是“推关系”。这一集要解决的就是把财报文本中的实体和关系自动抽出来并构建成 Neo4j 知识图谱为后续的图谱增强 RAG 提供结构化底座。本集是“从零搭建财报 RAG 知识图谱”系列的第 6 集目标很集中完成实体关系抽取与图谱入库流水线。我们会从 Neo4j 环境准备开始依次覆盖图谱 Schema 设计、LLM 关系抽取提示词、批量入库写入、接口封装和批量任务调度最终得到一个能自动把财报文本变成图谱数据的可用链路。如果你已经在跑这个系列的 RAG 问答链路或者准备做投研领域的知识库这一集可以直接当作 Neo4j 落地参考。1. 核心能力速览能力项说明项目定位财报文本到 Neo4j 知识图谱的全自动构建流水线核心链路文本清洗 - LLM 实体关系抽取 - 结构化校验 - 批量入库 Neo4j图数据库Neo4j Community / Desktop / Docker 均可抽取方式大模型结构化输出支持本地模型或云端模型入库方式Cypher UNWIND 批量写入MERGE 去重支持批量支持按目录或文件列表批量处理接口能力可封装 FastAPI 服务暴露抽取、建图、查询接口查询能力通过 Cypher 查询实体关系、路径、子图适合读者RAG 开发者、投研系统开发者、知识图谱初学者合规提醒只使用公开财报或已授权数据注意访问控制从实现角度看这套流水线不依赖特定 RAG 框架LangChain、LlamaIndex、自研 Prompt 都能接。第 5 集搭好的文档解析和分块逻辑到这里直接复用即可。2. 适用场景与使用边界知识图谱最大的价值是把“可检索的文本”变成“可推理的关系网络”。财报场景尤其适合公司、股东、高管、行业、产品、客户、供应商之间存在大量显式关系这种关系一旦结构化就能支撑“公司 A 的控股股东是谁”“前五大客户中哪些是上市公司”“供应商之间是否存在股权关联”这类查询。适合场景包括财报问答增强把图谱查询结果作为检索上下文喂给 RAG 回答链路。产业链与股权穿透通过股权关系路径查询多层持股结构。风险识别整合诉讼、质押、关联交易等实体关系辅助风险分析。批量建仓处理数十到数百份年报、季报形成公司关系库。不适用或需要谨慎的场景实时性要求极高的场景图谱构建依赖 LLM 抽取耗时明显高于纯文本检索。非结构化闲聊类问答知识图谱不是通用聊天数据源。敏感财务数据涉及内部未公开数据、个人信息或未授权数据时不应该直接入图。需要特别强调版权与授权边界。如果使用公开年报注意数据来源的版权条款如果使用内部财务数据必须在受控环境中部署配置访问权限并对输出结果做人工抽检避免错误三元组影响后续决策。3. 环境准备与前置条件本集流水线涉及以下几个组成部分建议按顺序检查。3.1 基础环境Python 3.10 或更高版本。Neo4j Community Edition本地单机或 Docker Desktop。LLM 服务本地模型Ollama、vLLM或云端 API 均可。磁盘空间Neo4j 数据目录预留 10GB 以上比较稳妥具体按语料量评估。3.2 Python 依赖建议新建虚拟环境避免与系统 Python 环境互相污染。python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate pip install neo4j pip install langchain langchain-openai # 如果使用 LangChain 封装 pip install fastapi uvicorn pydantic pip install python-dotenv如果使用本地模型按模型框架补充依赖。例如 Ollama 场景下直接通过 OpenAI 兼容接口访问即可。3.3 启动前检查清单检查项说明Neo4j 是否启动浏览器访问http://localhost:7474确认Bolt 端口是否开放默认 7687Docker 部署需映射认证密码是否修改首次启动后立即修改默认密码LLM API Key 是否配置云端模型需要写入.env待处理财报文本清洗为纯文本按文件存放4. Neo4j 安装与启动配置Neo4j 有几种常见部署方式这里给出两种最常用的。4.1 Docker 部署对本地开发和自动恢复场景Docker 是最省心的方式。docker run -d \ --name neo4j-finance \ -p 7474:7474 \ -p 7687:7687 \ -e NEO4J_AUTHneo4j/finance123! \ -e NEO4J_server_memory_heap_initial__size1G \ -e NEO4J_server_memory_heap_max__size2G \ -e NEO4J_server_memory_pagecache_size1G \ -e NEO4J_PLUGINS[apoc] \ -v neo4j_finance_data:/data \ neo4j:community参数说明7474Neo4j Browser 的 HTTP 端口。7687Bolt 协议端口Python 驱动连接用的是这个。NEO4J_AUTH第一次启动会创建用户neo4j密码为finance123!。NEO4J_PLUGINS可选安装 APOC便于后续使用一些图算法工具函数。-v数据卷持久化容器删除后数据不丢。启动后建议先访问http://localhost:7474用密码登录然后执行一条简单查询确认服务正常。RETURN 1 AS ok;4.2 Desktop 或 Community 安装如果本机已经安装 Neo4j Desktop直接新建 Project新建 Local DBMS选择 Community 版本即可。首次连接会在浏览器弹出修改密码页面。修改密码后Python 连接信息按实际填from neo4j import GraphDatabase URI bolt://localhost:7687 AUTH (neo4j, finance123!) driver GraphDatabase.driver(URI, authAUTH) with driver.session() as session: record session.run(RETURN 1 AS ok).single() print(record[ok]) driver.close()这里有一个容易踩的坑如果 Windows 上使用 Neo4j Desktop要在项目设置里确认 Bolt 端口没有被其他进程占用。多个 DBMS 同时运行时端口可能变化连接前查看实际端口。5. 图谱 Schema 设计入库前先把 Schema 设计好。Schema 不是死板的约束而是为了后续查询和分析效率。5.1 节点类型财报图谱建议先控制实体类型数量不要一上来就拆很细。节点类型说明示例Company公司主体宁德时代、贵州茅台Person自然人董事长、高管、股东Product产品/品牌极氪 001、飞天茅台Industry行业动力电池、白酒Indicator财务指标营业收入、毛利率Stock证券标的300750.SZ把实体统一放在Entity标签下再用type属性区分查询时更灵活但这会导致索引设计复杂一些。更稳妥的方式是给每种实体单独建标签例如:Company、:Person并统一保留entity_id和name属性。5.2 关系类型关系命名直接影响查询可读性。建议使用动词短语统一大写 下划线关系类型含义HOLDS_SHARES_OF持股IS_CUSTOMER_OF客户关系IS_SUPPLIER_OF供应商关系HAS_PRODUCT拥有产品BELONGS_TO_INDUSTRY所属行业EMPLOYS雇佣/高管任职DEAL_WITH交易/合作5.3 建索引和约束写入前为关键属性创建唯一约束防止重复实体堆积。CREATE CONSTRAINT company_name IF NOT EXISTS FOR (c:Company) REQUIRE c.name IS UNIQUE; CREATE CONSTRAINT person_name IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE;如果使用统一标签也可以做成组合约束CREATE CONSTRAINT entity_name_type IF NOT EXISTS FOR (e:Entity) REQUIRE (e.name, e.type) IS UNIQUE;索引和约束建立以后MERGE 写入效率会明显提升。6. 实体关系抽取提示词与结构化解析关系抽取是整个流水线的核心。设计目标不是“一次完美”而是“可控、可校验、可重试”。6.1 LLM 抽取提示词模板建议使用结构化输出要求模型返回 JSON并通过外部校验程序保证格式正确。EXTRACTION_PROMPT 你是财务领域知识图谱构建引擎。 请从财报文本中抽取实体和关系只输出 JSON 数组不要输出额外文字。 实体类型company, person, product, industry, indicator, stock 关系类型HOLDS_SHARES_OF, IS_CUSTOMER_OF, IS_SUPPLIER_OF, HAS_PRODUCT, BELONGS_TO_INDUSTRY, EMPLOYS, DEAL_WITH 输出 JSON 格式 [ { head: {type: company, name: 公司名}, relation: HOLDS_SHARES_OF, tail: {type: company, name: 公司名}, confidence: 0.95, evidence: 支持这句话的原文片段 } ] 财报文本 {text} 注意两点confidence字段让下游可以过滤低置信度关系。evidence字段保留来源原文方便追溯和复核。6.2 用 Pydantic 做结构化校验LLM 返回的 JSON 不一定合法用 Pydantic 校验可以拦住大部分格式错误。from pydantic import BaseModel, Field from typing import List class Entity(BaseModel): type: str Field(description实体类型) name: str Field(description实体名称) class Triple(BaseModel): head: Entity relation: str tail: Entity confidence: float Field(ge0.0, le1.0) evidence: str class TripleList(BaseModel): triples: List[Triple]在调用 LLM 后把返回文本解析为 JSON再加载到 Pydantic。解析失败时记录日志并考虑重试一次。6.3 多模型一致性策略如果同一个文本片段被多个模型抽取或者在不同批次中重复抽取可以按(head.name, relation, tail.name)做合并。置信度取平均或取最大值evidence保留第一次出现的原文。这一步可以在入库前完成也可以直接在 Cypher 中通过 MERGE 动态更新取决于你的数据量和业务要求。7. 图谱入库批量写入与去重抽取完成后下一步是把三元组写入 Neo4j。这里推荐使用 Cypher 的 UNWIND 批量写入而不是逐条写入因为逐条写入的网络和事务开销太大。7.1 统一标签 MERGE 写入UNWIND $triples AS t MERGE (s:Entity {name: t.head.name, type: t.head.type}) MERGE (o:Entity {name: t.tail.name, type: t.tail.type}) MERGE (s)-[r:RELATION {type: t.relation}]-(o) SET r.confidence t.confidence, r.evidence t.evidence, r.source t.source, r.period t.period这里把关系统一为RELATION用type属性区分。如果后续查询依赖固定关系类型最好拆成具体关系标签例如HOLDS_SHARES_OF。不过拆开以后MERGE 语句需要按关系类型写 CASE代码会复杂一些。7.2 拆分关系标签的写入写法如果使用具体关系标签可以这样处理UNWIND $triples AS t MERGE (s:Entity {name: t.head.name, type: t.head.type}) MERGE (o:Entity {name: t.tail.name, type: t.tail.type}) CALL apoc.merge.relationship(s, t.relation, {}, {}, o, {}) YIELD rel SET rel.confidence t.confidence, rel.evidence t.evidence, rel.source t.source, rel.period t.period该写法依赖 APOC 插件如果 Docker 启动时配置了NEO4J_PLUGINS[apoc]可以直接使用。7.3 Python 批量入库封装from typing import List, Dict WRITE_TRIPLES UNWIND $triples AS t MERGE (s:Entity {name: t.head.name, type: t.head.type}) MERGE (o:Entity {name: t.tail.name, type: t.tail.type}) MERGE (s)-[r:RELATION {type: t.relation}]-(o) SET r.confidence t.confidence, r.evidence t.evidence, r.source t.source, r.period t.period def write_triples(driver, triples: List[Dict], batch_size: int 500): for i in range(0, len(triples), batch_size): batch triples[i:i batch_size] with driver.session() as session: session.run(WRITE_TRIPLES, triplesbatch) print(f写入完成共 {len(triples)} 条三元组)建议单批控制在 200 到 500 条。批次太大单个事务执行时间过长容易出现内存压力或锁等待。7.4 增量入库思路后续新增财报时不需要清空图谱。只需要在处理新文本时给三元组打上新的source和period通过 MERGE 天然完成增量合并。同一实体和关系再次出现时只会更新属性和来源。如果业务要求保留多个报告期的关系变化可以在关系节点上添加期次属性或者在入库时增加时间维度MERGE (s)-[r:RELATION {type: t.relation, period: t.period}]-(o)这样同一个关系在不同报告期会形成多条边查询时可以按period过滤。8. 接口 API 设计与批量任务调度流水线不能只停留在命令行里。下面给出一个最小可用的 FastAPI 服务设计覆盖抽取、建图、查询三个入口。8.1 FastAPI 服务示例from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel app FastAPI(title财报图谱流水线服务) class BuildRequest(BaseModel): text: str period: str 2024Q3 source: str unknown class BuildResponse(BaseModel): status: str triples_count: int entities_count: int app.post(/graph/build, response_modelBuildResponse) def build_graph(req: BuildRequest): # 1. 调用 LLM 抽取三元组 triples extract_triples(req.text) # 2. 校验并写入 Neo4j write_triples(driver, triples, periodreq.period, sourcereq.source) return { status: ok, triples_count: len(triples), entities_count: count_entities(triples), } app.get(/graph/query) def query_graph(cypher: str): with driver.session() as session: result session.run(cypher) return [record.data() for record in result]启动服务uvicorn main:app --host 0.0.0.0 --port 8000接口启动后可以通过浏览器访问http://localhost:8000/docs查看接口文档。8.2 批量处理目录文件实际处理财报时通常是多个文件一起处理。写一个批量任务脚本按目录扫描文件逐个抽取入库。import os import json import time INPUT_DIR ./reports SUPPORTED_EXTS [.txt, .md] def batch_build(input_dir: str): files [ os.path.join(input_dir, f) for f in os.listdir(input_dir) if os.path.splitext(f)[1].lower() in SUPPORTED_EXTS ] for filepath in files: with open(filepath, r, encodingutf-8) as f: text f.read() # 这里替换为实际抽取与入库调用 # resp requests.post(http://127.0.0.1:8000/graph/build, json{...}) print(f完成: {filepath}) time.sleep(1) if __name__ __main__: batch_build(INPUT_DIR)批量任务建议记录每个文件的处理状态失败后重试避免中断后从头再来。8.3 失败重试与断点续跑最简单的做法是维护一个处理日志文件{ 2024年报.txt: success, 2023年报.txt: failed, 2024半年报.txt: pending }每个文件成功后立即写入状态。重跑时跳过 success只处理 failed 和 pending。9. 资源占用与性能观察图谱构建阶段资源瓶颈通常在两个地方LLM 抽取耗时和 Neo4j 写入内存。9.1 Neo4j 内存观察使用 Neo4j Browser 或系统监控工具观察指标说明Heap 堆内存事务执行、查询运算使用PageCache 页缓存数据文件缓存影响点查速度磁盘 I/O大批量写入时常见瓶颈Docker 部署时通过环境变量控制堆内存和页缓存。如果宿主机只有 8G 内存堆内存建议 1G页缓存 1G不要贪大。9.2 写入性能优化先建好约束和索引再执行 MERGE 写入。单批 200 到 500 条观察事务时间再调整批次大小。不要每条三元组都单独开事务。大批量写入前可以先关闭周期性的索引重建任务。9.3 显存占用说明如果使用本地大模型抽取实体关系显存占用取决于模型参数量和量化方式。7B 模型在量化状态下通常需要 6GB 到 8GB 显存如果使用云端 API本地不消耗额外显存。实际占用以本机测试为准建议先用单条财报文本验证抽取质量再进入批量阶段。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Neo4j 浏览器打不开服务未启动或端口被占用Docker logs 查看日志检查端口重启容器或修改端口映射Python 连接报unauthorized due to authentication failure密码错误或密码未修改在浏览器中登录验证重置密码后更新连接配置Docker 启动后数据访问 403默认密码未修改需强制改密访问 7474 控制台登录后修改默认密码LLM 返回内容无法解析为 JSON模型输出格式漂移打印原始返回文本增加解析失败重试或使用 JSON mode写入速度很慢缺少索引、批次过小或过大检查执行计划观察事务耗时建索引后重试调整批次大小中文实体乱码终端编码或文本处理编码问题检查源码文件编码和读取编码统一使用 UTF-8 读取和写入DBeaver 连接 Neo4j 报content is not allowed in prolog连接配置被错误保存为 XML 格式或驱动配置损坏重新建立数据库连接确认驱动版本优先使用 Neo4j Browser 或 Python 驱动验连接其中“连接失败但认证已通过”的情况多数是URI写错或端口没有映射。Docker 部署时Python 连接必须使用bolt://localhost:7687不要用 7474。11. 最佳实践与合规建议把流水线从“能跑”提升到“能稳定跑”下面几条值得从一开始就执行。11.1 数据层面每个三元组保留evidence和source出问题可以追溯到原文。低置信度关系单独标记不直接参与关键问答。财报文件按报告期命名入库时写入period属性方便按时间切分查询。11.2 工程层面抽取和入库解耦抽取结果先落盘成 JSON入库失败时可以重放。批量任务必须有日志和状态记录。API 服务不要直接暴露到公网设置访问认证或内网访问限制。定期备份 Neo4j 数据目录。11.3 合规层面只使用公开披露的年报、季报或已获得授权的财务数据。不把个人敏感信息、未公开经营数据直接入图。对抽取结果做人工抽检财务关系错误可能造成误导。11.4 提示词与模型选择先用小片段测试抽取质量再扩展到全文。提示词固定好后不要频繁修改否则图谱一致性会被破坏。云端模型和本地模型的抽取结果可能有差异切换模型前先在测试集上对比。12. 总结与下一步这一集完成了 Neo4j 知识图谱自动构建链路的三个关键部分实体关系抽取、批量入库、接口封装。整个流水线处理财报文本时最终产出的是一批带来源和置信度的三元组这些三元组存放在 Neo4j 中可以被 Cypher 查询、被图算法分析也能在下一集接入 RAG 问答链路作为结构化检索来源。最建议先验证的功能是单文本的POST /graph/build接口确认实体、关系正确入库后再扩大到目录批量处理。最容易踩的坑还是两个一是连接 Neo4j 时的端口和密码配置二是 LLM 抽取结果没有做结构化校验就直接入库。下一步可以沿着两个方向扩展一是把图谱查询结果接入 RAG 的检索重排流程实现“向量检索 图谱检索”双路召回二是用 APOC 或图算法做链路上推理比如股权穿透、关联方发现让图谱从“能查”变成“能分析”。建议收藏备用后面几集会继续围绕财报 RAG 展开这些工程细节。