Graph RAG：2026完整实现

概述

Graph RAG是一种专注于知识图谱遍历以实现多跳推理的专用检索模式。本指南将展示如何使用Neo4j构建生产级系统。

为什么选择Graph RAG？

标准RAG在处理复杂查询、多跳推理或领域特定内容时往往力不从心。Graph RAG通过知识图谱遍历实现多跳推理，有效解决了这些局限性。

架构

查询 → [图预处理] → 向量搜索 → [上下文处理] → LLM → 响应
              ↓                                           ↑
         查询扩展                             重排序 + 过滤

实现

环境准备

bash
pip install langchain langchain-openai neo4j tiktoken

python
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
初始化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

图检索器

python
from langchain.retrievers import GraphRetriever
from langchain_neo4j import Neo4jVectorStore
构建向量存储
vectorstore = Neo4jVectorStore.from_documents(
    documents=your_documents,
    embedding=embeddings,
    index_name="my-rag-index"
)
创建专用于知识图谱遍历实现多跳推理的检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 6,
        "fetch_k": 25,
        "lambda_mult": 0.7
    }
)

文档处理

python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader
def load_and_process_documents(directory: str) -> list[Document]:
    """加载并处理文档，用于Graph RAG。"""
    
    # 加载文档
    loader = DirectoryLoader(directory, glob="**/*.txt")
    raw_docs = loader.load()
    
    # 带重叠切分以保留上下文
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=150,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    
    chunks = splitter.split_documents(raw_docs)
    
    # 添加元数据以支持知识图谱遍历实现多跳推理
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "chunk_id": i,
            "variant": "Graph",
            "chunk_length": len(chunk.page_content)
        })
    
    print(f"从{len(raw_docs)}个文档创建了{len(chunks)}个块")
    return chunks
chunks = load_and_process_documents("./documents/")

RAG链

python
def create_graph_chain(retriever):
    """创建针对知识图谱遍历实现多跳推理优化的Graph RAG链。"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个知识渊博的AI助手。
        请使用以下检索到的上下文准确回答问题。
        
        上下文：
        {context}
        
        知识图谱遍历实现多跳推理的指南：
        - 引用上下文中的具体信息
        - 如果信息不在上下文中，请明确说明
        - 尽可能引用来源
        - 简洁但完整"""),
        ("human", "{question}")
    ])
    
    def format_context(docs: list[Document]) -> str:
        formatted = []
        for doc in docs:
            source = doc.metadata.get('source', 'Unknown')
            formatted.append(f"[来源: {source}]\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)
    
    chain = (
        {
            "context": retriever | format_context,
            "question": RunnablePassthrough()
        }
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain
构建并使用链
rag_chain = create_graph_chain(retriever)
answer = rag_chain.invoke("你的问题")

进阶：带来源的流式输出

python
from langchain_core.runnables import RunnableParallel
def create_rag_with_sources(retriever):
    """返回答案和来源文档的RAG。"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "基于上下文回答。准确并引用来源。\n\n上下文: {context}"),
        ("human", "{question}")
    ])
    
    # 并行执行检索和格式化
    setup = RunnableParallel(
        context=retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
        question=RunnablePassthrough(),
        source_documents=retriever
    )
    
    chain = setup | {
        "answer": prompt | llm | StrOutputParser(),
        "sources": lambda x: [d.metadata.get('source') for d in x['source_documents']]
    }
    
    return chain
chain_with_sources = create_rag_with_sources(retriever)
result = chain_with_sources.invoke("主要话题是什么？")
print(f"答案: {result['answer']}")
print(f"来源: {result['sources']}")

评估

python
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
def evaluate_rag(test_cases: list[dict]) -> dict:
    """使用RAGAS评估Graph RAG质量。"""
    
    dataset = Dataset.from_list(test_cases)
    
    result = evaluate(
        dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )
    
    print(f"忠实度: {result['faithfulness']:.3f}")
    print(f"答案相关性: {result['answer_relevancy']:.3f}")
    print(f"上下文精确度: {result['context_precision']:.3f}")
    print(f"上下文召回率: {result['context_recall']:.3f}")
    
    return result
test_cases = [
    {
        "question": "关键特性有哪些？",
        "answer": rag_chain.invoke("关键特性有哪些？"),
        "contexts": [d.page_content for d in retriever.invoke("关键特性有哪些？")],
        "ground_truth": "预期答案..."
    }
]
evaluate_rag(test_cases)

性能优化技巧

结论

基于Neo4j的Graph RAG为知识图谱遍历实现多跳推理提供了出色的基础。本文展示的模式经过生产验证且可扩展。

从基础实现开始，使用RAGAS衡量质量，然后根据指标迭代优化。

---

*Graph RAG实现 | Neo4j | 2026年5月*