GraphRAG实战：用Neo4j打造AI Agent知识图谱问答系统

先问一个问题：把一堆知识库文档切片塞进向量存储，再交给大模型，这真的是AI Agent落地问答场景的最优解吗？我做客服类Agent做了大半年，传统RAG该踩的坑基本都踩过：提问稍微绕一点，答案就开始拼凑、自相矛盾，更别提让模型做多跳推理。后来被逼着把方案升级成graphRAG，存储层选了Neo4j，这才把准确率和可解释性拉回到能交付的水平。这篇就把整个项目的来龙去脉、架构选型和核心实现完整写出来，给正在折腾AI Agent、又对graphRAG感兴趣的朋友做个参考。

整个过程我会按实际开发顺序讲：先解释为什么放弃纯向量RAG，再讲Neo4j怎么选型和部署，然后是实体关系抽取、Cypher写入、图检索和Agent工具编排，最后是问题排查和优化经验。内容会带上大量可直接抄的代码和配置，你复现的时候能少走不少弯路。

1. 先想清楚：为什么传统RAG不够用了，graphRAG改了什么

1.1 传统RAG的三个硬伤

我最早做的问答Agent是标准套路：文档切块 -> embedding -> 向量检索 -> 拼接上下文 -> LLM生成。单点知识问答效果确实还行，比如“某产品的退款政策是什么”，因为答案就在某一片段里。但业务场景根本不会这么温柔，问题稍微复杂一点，这套架构就露馅。

第一个问题是上下文割裂。文档被切割成独立片段后，片段之间的引用关系全部丢失。比如合同里先定义“乙方”，后面所有“乙方”的条款都依赖这个定义，切片之后，检索到的片段里可能只有行为描述，没有主体定义，模型只能靠猜。

第二个问题是多跳推理崩溃。用户问“A产品用了B供应商的芯片，B供应商最近财报里提到产能紧张，会不会影响A的交货周期”，这需要跨越多个文档、多个实体做关联推理。向量检索按相似度召回，召回来的往往是和问题字面最像的片段，而不是逻辑上相关的节点，模型拿不到完整的推理路径，自然答不对。

第三个问题是不可解释。传统RAG返回答案时，你很难说清楚它依据的是哪条确定性的知识链。出了问题排查困难，业务方提出“为什么这么回答”时，你没法给出有说服力的证据链。

1.2 graphRAG把RAG从“相似度匹配”升级成“图遍历”

graphRAG的核心区别是：先把非结构化文本转成结构化的知识图谱，把“实体”作为节点、“关系”作为边存起来，然后再在图上做检索。检索的对象从“相似的文本片段”变成了“相关的实体和它们之间的关系”。

举个例子，传统RAG处理“B供应商产能紧张对A产品的影响”这条问题时，它会去向量库找“B供应商”相关的文本片段，很可能只找到一段财务摘要，语义相关，但不知道它和A产品是什么关系。graphRAG则不同，它在知识图谱里存了这样的路径：A产品 -> 使用元器件 -> 芯片X -> 供应商 -> B供应商，查询时沿图遍历，把整条路径上的节点和关系全部取出来作为上下文，LLM看到的是完整的因果链，答案自然靠谱得多。

而且图结构天然支持多跳查询，这正好对应业务里最常见的复杂问题。Cypher查询每多跳一层，就相当于给模型追加了一个关系事实，比塞多少文本片段都管用。

1.3 为什么数据库偏偏选Neo4j

可选图数据库不少，但我最后选了Neo4j，理由很实际。

首先是原生图存储。Neo4j不是用关系表模拟图，而是节点和边在物理存储上就是相邻的，遍历时不需要做昂贵的关联查询，这对多跳检索的性能非常重要，尤其在知识图谱节点数达到百万级之后。

其次是Cypher查询语言。Cypher的语法非常接近人类描述关系的方式，比如MATCH (p:Product)-[:USES]->(c:Chip)-[:SUPPLIED_BY]->(s:Supplier) RETURN p, c, s，这句话基本就是问题本身。图RAG里最关键的“Text2Cypher”环节，就是让LLM将自然语言翻译成这种查询，语言越直观，翻译准确率越高。

第三是生态成熟度。Neo4j官方维护了和LangChain、LlamaIndex的集成组件，也提供了内置向量索引。也就是说你可以在同一个数据库里同时做向量检索和图遍历，不需要额外再搭一套向量库，架构会轻很多。社区版免费，单机部署对中小项目完全够用。

2. 环境准备与Neo4j部署

2.1 版本选择和安装方式

Neo4j当前主推5.x版本，graphRAG相关能力基本都在5.x上完善。我用的是5.19.0社区版，支持向量索引、全文索引和内置的db.index.vector.queryNodes查询函数，这几个能力后面都要用到。

安装方式我建议优先用Docker，隔离干净、版本切换方便，也不用处理本地Java环境的依赖问题：

docker run \ --name neo4j-graphrag \ -p 7474:7474 -p 7687:7687 \ -e NEO4J_AUTH=neo4j/yourpassword \ -e NEO4J_PLUGINS='["apoc"]' \ -d neo4j:5.19.0

如果你不想用Docker，也可以直接去Neo4j官网下载社区版安装包，或者用Neo4j Desktop这个桌面客户端。需要说明的是，如果官网下载速度不理想，可以找国内高校镜像或开源软件镜像站，文件不大，拉下来很快。Docker镜像本身走的是镜像仓库，一般没有障碍。

参数解释一下：7474是浏览器管理界面的HTTP端口，7687是Bolt协议端口，Python驱动连接走的就是7687。NEO4J_PLUGINS预装了APOC，后面做数据加工、图算法时会用到。

2.2 基础配置与连接验证

启动后先打开http://localhost:7474，用neo4j/yourpassword登录，完成初始密码修改。然后打开同一个Docker容器，把内存参数调一下，防止因为并发查询出现OOM：

docker exec -it neo4j-graphrag bash # 编辑 /var/lib/neo4j/conf/neo4j.conf

我实际用的配置供参考：

server.memory.heap.initial_size=512m server.memory.heap.max_size=1g server.memory.pagecache.size=512m

页面缓存的大小取决于你的图数据量，数据在几百万节点以内，512m到1g够用。改完配置重启容器：

docker restart neo4j-graphrag

然后用Python验证连接：

from neo4j import GraphDatabase driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "yourpassword")) def ping(): with driver.session() as session: result = session.run("RETURN 'connected' AS status") print(result.single()["status"]) ping()

这一步能把80%的环境问题暴露出来：端口没映射、密码不对、驱动版本不兼容，全会在连接阶段报错。

注意：如果你是远程连接，记得在neo4j.conf里把server.default_listen_address改成0.0.0.0，否则只有容器内部能连。

2.3 确认向量索引和APOC插件可用

链路越提前验证越好，免得写到后面才发现在这个版本里函数名不对。建索引前先执行：

RETURN apoc.version()

能返回版本号说明APOC没问题。向量索引用下面的语句创建，里面的维度要和之后生成embedding的模型维度保持一致：

CREATE VECTOR INDEX entity_vector IF NOT EXISTS FOR (n:Entity) ON (n.embedding) OPTIONS { indexConfig: { `vector.dimensions`: 1536, `vector.similarity_function`: 'cosine' }};

我用的embedding模型是text-embedding-3-small这类按维度区分的接口，1536维比较主流。如果你用别的模型，比如768维或1024维，必须同步改这里。

3. 核心实现：从文档到知识图谱

3.1 用LLM做实体关系抽取

graphRAG的第一步是把原始文档变成三元组。这一步质量直接决定整个系统的上限，我在实际项目中甚至觉得它比检索和生成都重要。

抽取的核心是设计好提示词，让LLM输出结构化JSON。我的工程实现用LangChain的with_structured_output配合Pydantic模型，比自己解析JSON字符串稳定得多：

from pydantic import BaseModel, Field from typing import List from langchain_openai import ChatOpenAI class Entity(BaseModel): name: str = Field(description="实体名称，尽量使用全称") type: str = Field(description="实体类型，如产品/供应商/指标/事件") description: str = Field(description="一句话描述该实体") class Relation(BaseModel): source: str = Field(description="源实体名称") target: str = Field(description="目标实体名称") type: str = Field(description="关系类型，如供应/属于/影响") detail: str = Field(description="关系详细说明") class KnowledgeGraph(BaseModel): entities: List[Entity] relations: List[Relation] extract_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) structured_llm = extract_llm.with_structured_output(KnowledgeGraph) def extract_graph(text: str) -> KnowledgeGraph: prompt = f""" 从下面的文本中抽取与业务相关的实体和关系。 实体类型包括但不限于：产品、供应商、技术、指标、事件、组织、人物。 关系要明确主客体，比如"芯片X的供应商是B"应抽为 (芯片X) -[supplied_by]-> (B)。 只抽取文本中明确出现或能直接推断的信息，不要臆造。 如果同一实体出现多个名称，优先使用全称。 文本内容： {text[:3000]} """ return structured_llm.invoke(prompt)

提示词里我特别加了两个约束：“只抽取文本中明确出现或能直接推断的信息”和“如果同一实体出现多个名称，优先使用全称”。前者用来抑制模型在抽取时强行联想，后者用来减少图谱里同一实体多个节点的问题。如果你不做实体对齐，后面所有查询都会因为名字不一致而查不到，这个坑我踩过。

3.2 把三元组写入Neo4j

抽取的结果是一堆元组，但写入时不能盲目用CREATE，否则重复处理同一批文档就会生成大量重复节点。正确做法是用MERGE，它只在节点不存在时才创建，存在则匹配复用：

def write_graph(tx, graph: KnowledgeGraph): for entity in graph.entities: tx.run( """ MERGE (e:Entity {name: $name}) ON CREATE SET e.type = $type, e.description = $description """, name=entity.name, type=entity.type, description=entity.description ) for rel in graph.relations: tx.run( """ MATCH (a:Entity {name: $source}) MATCH (b:Entity {name: $target}) MERGE (a)-[r:RELATION {type: $type}]->(b) SET r.detail = $detail """, source=rel.source, target=rel.target, type=rel.type, detail=rel.detail ) with driver.session() as session: for chunk in chunks: # chunks 是切分后的文档块 graph = extract_graph(chunk) session.execute_write(write_graph, graph)

这里有个细节：MERGE关系前必须先MATCH两个端点，所以我在写实体时用ON CREATE SET保证实体首次创建就带属性，写关系时再通过名字找到端点。如果你在同一个事务里先创建实体再创建关系，也没问题，但跨事务时一定要确保实体已经存在，否则MATCH会匹配不到导致关系写入失败。

批量写入时建议用session.execute_write自动管理事务，每批写一个chunk的抽取结果，出错只会回滚当前批次，不会污染全量数据。

3.3 节点embedding化

知识图谱建好后，需要给每个节点生成embedding，用来支持后面“实体语义召回+图遍历”的混合检索。给整个节点批量打向量，我写过这样一个操作：

from langchain_openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings(model="text-embedding-3-small") def update_embeddings(tx): result = tx.run("MATCH (n:Entity) WHERE n.embedding IS NULL RETURN n.name AS name") for record in result: name = record["name"] vec = embeddings.embed_query(name) tx.run( "MATCH (n:Entity {name: $name}) SET n.embedding = $vec", name=name, vec=vec ) with driver.session() as session: session.execute_write(update_embeddings)

只对embedding IS NULL的节点生成向量，避免重复调用模型接口烧钱。向量存储就是前面建好的entity_vector索引，查询时用db.index.vector.queryNodes。

4. AI Agent编排：把Neo4j变成Agent的工具

4.1 为什么需要Agent而不是直接问答

图建好之后，许多教程就停在“写一段Cypher查询然后让LLM生成答案”了，但真实业务里用户问题千变万化，一段固定的查询逻辑根本覆盖不了。我的做法是引入Agent，让模型自己做“检索规划”。

用户问“B供应商产能紧张对A产品的交付有什么影响”，Agent需要拆解成几步：先定位B供应商节点，再沿supplied_by关系找到相关产品，再查这些产品的交付状态，最后综合多段上下文生成回答。这个“拆解-执行-汇总”的过程，用ReAct模式的Agent来编排最自然。

我用的框架是LangGraph，比纯LangChain链式调用灵活得多，它把Agent建模成状态图，节点之间可以条件跳转、循环、并行，完全符合复杂的查询流程。

4.2 三个核心工具的实现

Agent手上的工具我做成了三个，各有分工：

工具1：实体语义搜索。用户问题先走向量召回，找到可能的实体节点：

def semantic_search(question: str, top_k: int = 5): q_vec = embeddings.embed_query(question) with driver.session() as session: result = session.run( """ WITH $question_vec AS qv CALL db.index.vector.queryNodes('entity_vector', $top_k, qv) YIELD node, score RETURN node.name AS name, node.description AS description, score """, question_vec=q_vec, top_k=top_k ) return [record.data() for record in result]

工具2：Text2Cypher查询。让LLM根据图谱schema把用户问题翻译成Cypher并执行。这是整个graphRAG最核心的环节，我的提示词写得比较讲究：

schema = """ 节点: Entity(name, type, description, embedding) 关系: RELATION(type, detail) 示例查询: 1. 某实体的直接关系: MATCH (n:Entity {name:$name})-[r:RELATION]-(m) RETURN n,r,m 2. 两跳关系: MATCH (a:Entity {name:$name1})-[:RELATION*1..2]-(b:Entity {name:$name2}) RETURN a,b 3. 按类型过滤: MATCH (n:Entity {type:'供应商'}) RETURN n LIMIT 20 """ cypher_llm = ChatOpenAI(model="gpt-4o", temperature=0) def text_to_cypher(question: str) -> str: prompt = f""" 你是Neo4j数据库的Cypher查询专家。根据图数据库schema将用户问题转换为可执行的Cypher查询。 schema: {schema} 要求： 1. 实体名称匹配时，必须使用大小写不敏感的模糊匹配，例如 toLower(n.name) CONTAINS toLower($keyword) 2. 如果用户没有指定具体实体名称，不要创建查询，返回错误说明 3. 限制返回结果为30条以内，防止超时 4. 只输出Cypher语句，不要输出多余解释 用户问题: {question} """ response = cypher_llm.invoke(prompt) return response.content

执行时用session.run(cypher)，如果语法错误或超时，我会把错误信息回传给LLM让它重写一次，最多重试两次。

工具3：图路径遍历。当用户问题涉及明确的实体对，比如“A和B之间有什么关系”，直接跑路径查询比让LLM生成Cypher更稳：

def find_path(source: str, target: str, max_depth: int = 3): with driver.session() as session: result = session.run( """ MATCH path = shortestPath( (a:Entity)-[:RELATION*..{max_depth}]-(b:Entity) ) WHERE toLower(a.name) CONTAINS toLower($source) AND toLower(b.name) CONTAINS toLower($target) RETURN [n IN nodes(path) | n.name] AS nodes_path, [r IN relationships(path) | r.type] AS rels_path """, source=source, target=target, max_depth=max_depth ) return [record.data() for record in result]

这里我用shortestPath防止深遍历导致查询爆炸，最大深度限制到3.业务里大多数逻辑链2跳就能说清，3跳基本是上限。

4.3 Agent调度逻辑与回退策略

Agent的调度我画成如下状态流，虽不是复杂图，但能说明几个关键节点：

1. 收到问题，先用语义搜索找候选实体，候选置信度超过阈值就直接进图查询；
2. 候选置信度低或没有候选，转Text2Cypher，尝试直接生成查询；
3. 如果Text2Cypher生成的查询执行失败，回退到纯语义检索；
4. 图查询结果和多段文本上下文一起汇总给LLM生成最终答案；
5. 最终答案必须附上图谱里的证据路径，例如“A产品 <- 使用 <- 芯片X <- 供应商 <- B供应商”。

我用的LangGraph代码骨架是这样：

from langgraph.graph import StateGraph, END class AgentState(dict): question: str entities: list cypher_result: str answer: str def node_semantic_search(state): # 语义搜索得到候选实体 pass def node_graph_query(state): # 基于候选实体做图查询 pass def node_answer(state): # 汇总上下文生成答案 pass def route_after_search(state): # 根据候选实体质量决定走图查询还是回退 if state.get("entities"): return "graph_query" return "answer" graph = StateGraph(AgentState) graph.add_node("semantic_search", node_semantic_search) graph.add_node("graph_query", node_graph_query) graph.add_node("answer", node_answer) graph.set_entry_point("semantic_search") graph.add_conditional_edges("semantic_search", route_after_search) graph.add_edge("graph_query", "answer") graph.add_edge("answer", END)

关于并发，我在项目后期还接了一层FastAPI，用asyncio包装驱动查询，并对Neo4j连接池做了上限控制。这里有个经验：Neo4j Python驱动本身有连接池，默认大小就够了，但如果你用Docker容器，宿主机和容器的文件描述符限制要调大，否则并发一高连接数会卡在Linux层的ulimit上，表现就是请求全部超时但数据库负载很低，非常诡异。

5. 常见问题与排查技巧实录

5.1 实体抽取质量差，图谱一团乱麻

这是我遇到的第一个大坑。最初我让LLM自由抽取，结果出现大量类似“信息”、“问题”、“方案”这种泛化实体，图谱中心化严重，没有任何查询价值。后来加了两条硬性约束才缓解：一是实体类型白名单，只允许产品、供应商、指标、事件、组织、人物等业务类型；二是抽取时要求描述必须具体到可以被事实核查的粒度，比如“影响交付周期”这种模糊描述直接过滤。

如果还不行，建议做一步后处理去重：用apoc.text.similarity或简单按名称归一化（去掉空格、统一大小写）合并同类实体。实体对齐这件事，别指望一次做好，图谱是迭代出来的。

5.2 Text2Cypher生成错误Cypher

LLM生成Cypher错误是常态，我的经验是按三类错误分别处理：

提前在提示词里声明“只允许MATCH，不允许CREATE/DELETE/SET”能有效防止模型生成破坏性语句。即使如此，我还是建议给Agent用的数据库账号权限收窄到只读，别用neo4j超级账号跑生产服务。

5.3 向量检索和图查询结果怎么合并

这是graphRAG最容易被忽视的工程问题。向量检索出来的是“语义相似的实体”，图查询出来的是“结构相关的关系链”，两者得分体系不同，不能简单相加排序。

我的做法是分两阶段：先用向量检索定位起始实体，再用图查询扩展上下文。向量召回结果只用于找出“用户问题对应哪个实体”，进入图查询后，上下文主要以图路径为主，不再受向量得分影响。如果你需要在最终生成时给出相关性排序，优先关注图路径跳数——路径越短，逻辑越紧密。

5.4 答案质量评估

很多项目上线后不知道怎么衡量graphRAG比传统RAG好多少。我在项目里做了一个简单的评估集：30条问题，分成单实体事实类、多实体关系类、跨文档推理类。分别记录AI Agent用传统RAG和graphRAG的准确率。实测下来，跨文档推理类从传统RAG的不到40%提升到graphRAG的80%以上，效果非常直观。

建议你也留一组固定的评测问题，每次改动抽取提示词、图结构或检索逻辑后都跑一遍，防止改一处坏一处。

6. 个人经验与扩展空间

整个项目做下来，最大的体会是graphRAG的瓶颈不在技术链路，而在知识工程。Neo4j和LangGraph这些工具都是成熟的，真正决定系统上限的是你怎样设计实体类型、怎样定义关系、怎样约束LLM去抽取。不要一上来就追求大而全的图谱，先针对业务里最常问的三类问题，建一个不超过500个节点的核心图谱，把链路跑通，再逐步扩大，这个节奏最稳。

一个小技巧是：给每个关系都尽量带上detail属性，把原始文本里那句支撑性的句子存下来。最终问答时把detail一并作为上下文交给大模型，它就有了事实依据，而不是只靠图谱骨架自己发挥。这一步对答案的可信度提升非常明显。

后续可扩展的方向也比较清晰：一是引入时间维度，把Entity节点加上有效期属性，做时序知识图谱；二是把断言都挂上来源文档ID，形成一个可追溯的知识链路，方便审计；三是接入更强的Text2Cypher微调模型或专项prompt。这些都是在上面的基础之上涨点快的优化。如果你也在做AI Agent的落地项目，卡在RAG效果上，不妨试试把Neo4j加进你的技术栈，它带来的不是某一个指标提升，而是整个问题求解方式的变化。