传统RAG（检索增强生成）技术在处理基于私有文档的问答时面临瓶颈，主要依赖向量相似性而忽略了实体间的上下文关系。本文将深入探讨一种更强大的演进方案——GraphRAG，它将知识图谱与向量检索相结合，不仅理解语义相似性，还能洞悉概念间的关联。我们将通过一个实际案例，详细介绍如何使用LangChain框架，结合Google的Gemini大模型和Neo4j图数据库，构建一个生产级别的GraphRAG系统，重点关注内存向量存储以实现最佳性能和简化操作。
GraphRAG：超越传统RAG的强大技术
GraphRAG通过引入知识图谱，扩展了传统RAG的能力。它不仅能找到语义上相似的文本块，还能通过遍历实体间的关系，提供更全面、更符合上下文的答案。相较于传统RAG，GraphRAG的关键优势在于：
- 关系感知： 能够理解实体之间的连接，例如“员工属于部门”、“产品包含零件”等。这使得系统可以推断出隐藏的信息，提供更深入的回答。例如，如果用户询问“A公司的CEO是谁？”，传统RAG可能只能检索到包含“A公司”和“CEO”的段落，而GraphRAG可以根据知识图谱中“A公司”与“CEO”之间的关系，直接给出答案。
- 上下文检索： 通过图遍历找到相关信息。不再仅仅依赖向量相似度，而是可以根据实体之间的关系进行多跳检索。例如，用户询问“与A公司相关的政策有哪些？”，GraphRAG可以先找到A公司，然后通过“实施”或“遵循”等关系，找到相关的政策文件。
- 更高的准确性： 结合语义相似性和结构化关系。这降低了“噪声”的影响，提高了答案的精度。例如，在法律领域，一个词可能在不同的上下文中含义不同，GraphRAG可以通过知识图谱确定该词在特定法律条文中的具体含义，从而提供更准确的解释。
- 可解释性： 图关系提供了推理的透明度，用户可以追溯答案的来源和逻辑。这对于需要审计和验证的场景非常重要。例如，用户可以查看GraphRAG是如何通过一系列关系最终得出结论的，从而增加对系统回答的信任度。
系统架构：分层设计的GraphRAG
我们构建的GraphRAG系统采用分层架构，将向量检索与知识图谱相结合，实现了高效且灵活的查询处理流程。具体架构如下：
- 文档加载： 从各种来源（例如.docx文件）加载并预处理文档。
- 语义分块： 使用基于嵌入的文本分割技术将文档分割成语义相关的文本块。
- 图谱提取： 使用LLM（大型语言模型）驱动的实体和关系发现技术，从文本块中提取实体和关系。
- 双重存储： 将文本块存储在内存向量存储中，并将实体和关系存储在Neo4j图数据库中。
- 混合检索： 结合图遍历和向量相似性，检索相关信息。
这种架构允许系统同时利用文本的语义信息和知识图谱的结构化信息，从而提供更全面、更准确的答案。例如，在金融领域，我们可以构建一个包含公司、股票、行业、财务指标等实体的知识图谱，并结合新闻报道、研报等文本数据，构建一个强大的金融知识问答系统。
核心组件：LangChain、Gemini与Neo4j的协同
我们的GraphRAG系统依赖于以下关键组件：
- LangChain： 这是一个强大的框架，用于构建基于LLM的应用程序。它提供了各种模块，包括文档加载器、文本分割器、向量存储、图数据库集成和链式调用。
- Gemini： Google的下一代多模态大模型，用于生成文本嵌入和执行实体和关系提取。Gemini强大的语义理解能力能够准确识别文档中的实体和关系。例如，Gemini可以从一篇医学论文中准确提取出药物、疾病、症状等实体，以及它们之间的关系，例如“药物治疗疾病”、“症状是疾病的特征”。
- Neo4j： 一个高性能的图数据库，用于存储和查询知识图谱。Neo4j的Cypher查询语言提供了强大的图遍历能力，可以高效地查找实体之间的关系。例如，我们可以使用Cypher查询语言找到与某个特定疾病相关的所有药物和治疗方案。
- 内存向量存储： 是一种用于存储和检索向量嵌入的内存数据库。由于数据存储在内存中，因此可以实现极快的查询速度。这对于需要快速响应的实时问答系统至关重要。
这些组件协同工作，共同构建了一个高效、准确的GraphRAG系统。LangChain作为连接器，将Gemini的强大语义理解能力和Neo4j的图数据库能力整合在一起，实现了知识图谱与向量检索的无缝集成。
代码实现：一步步构建GraphRAG系统
以下是GraphRAG系统的核心代码示例：
import os
import glob
from typing import List
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI
from langchain_experimental.graph_transformers import LLMGraphTransformer
from langchain_experimental.text_splitter import SemanticChunker
from langchain_neo4j import Neo4jGraph
from langchain_community.document_loaders import Docx2txtLoader
from langchain_graph_retriever import GraphRetriever
from graph_retriever.strategies import Eager
from dotenv import load_dotenv
load_dotenv()
# 初始化 LLM 图转换器
print("Initializing LLM Graph Transformer...")
embeddings = GoogleGenerativeAIEmbeddings(
model="models/gemini-embedding-exp-03-07",
google_api_key=os.getenv("GOOGLE_API_KEY")
)
llm = ChatGoogleGenerativeAI(
model="models/gemini-2.5-flash-preview-05-20",
google_api_key=os.getenv("GOOGLE_API_KEY")
)
llm_transformer = LLMGraphTransformer(llm=llm)
graph = Neo4jGraph(
url=os.getenv("NEO4J_URI"),
username=os.getenv("NEO4J_USERNAME"),
password=os.getenv("NEO4J_PASSWORD"))
def load_and_chunk_documents() -> List[Document]:
"""加载并使用语义分块 .docx 文件"""
documents = []
docx_files = glob.glob("./documents/*.docx") # 根据需要调整路径
if not docx_files:
print("No .docx files found")
return documents
# 初始化语义分块器
text_splitter = SemanticChunker(embeddings)
for docx_file in docx_files:
print(f"Processing {docx_file}...")
# 加载文档
loader = Docx2txtLoader(docx_file)
raw_docs = loader.load()
# 应用语义分块
chunks = text_splitter.split_documents(raw_docs)
# 添加元数据
for chunk in chunks:
chunk.metadata["source"] = os.path.basename(docx_file)
documents.extend(chunks)
print(f"Loaded {len(documents)} document chunks")
return documents
def store_graph_documents(graph_documents: List):
"""将图文档存储在 Neo4j 中"""
print("Storing graph data in Neo4j...")
# 清除现有数据（可选）
graph.query("MATCH (n) DETACH DELETE n")
for graph_doc in graph_documents:
# 添加节点
for node in graph_doc.nodes:
node_type = node.type.replace(" ", "_") # 为 Neo4j 进行清理
graph.query(
f"MERGE (n:{node_type} {{id: $id}})",
{"id": node.id}
)
# 添加关系
for rel in graph_doc.relationships:
rel_type = rel.type.replace(" ", "_")
graph.query(
f"""
MATCH (source {{id: $source_id}})
MATCH (target {{id: $target_id}})
MERGE (source)-[r:{rel_type}]->(target)
""",
{
"source_id": rel.source.id,
"target_id": rel.target.id
}
)
print("Graph data stored successfully")
def create_rag_chain(vector_store):
"""使用图和向量检索创建 RAG 链"""
# 根据您的数据创建策略
traversal_config = Eager(
select_k=5,
start_k=2,
adjacent_k=3,
max_depth=2
)
# 创建图检索器
graph_retriever = GraphRetriever(
store=vector_store,
strategy=traversal_config,
)
# 文档格式化器
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# 提示模板
prompt = ChatPromptTemplate.from_template("""
Answer the question based on the provided context.
Context: {context}
Question: {question}
Answer:
""")
# 创建链
chain = (
{
"context": graph_retriever | format_docs,
"question": RunnablePassthrough()
}
| prompt
| llm
| StrOutputParser()
)
return chain
def test_graphrag_system(chain):
"""使用示例查询测试 GraphRAG 系统"""
test_queries = [
"What are the main topics discussed in the documents?",
"How are the entities connected in the knowledge graph?",
"Can you explain the relationships between key concepts?"
]
for query in test_queries:
print(f"\n🔍 Query: {query}")
try:
response = chain.invoke(query)
print(f"Response: {response[:200]}...")
except Exception as e:
print(f"Error: {e}")
# 检查图统计信息
try:
nodes_count = graph.query("MATCH (n) RETURN count(n) as count")[0]["count"]
rels_count = graph.query("MATCH ()-[r]->() RETURN count(r) as count")[0]["count"]
print(f"\nGraph Stats: {nodes_count} nodes, {rels_count} relationships")
except Exception as e:
print(f"Error getting graph stats: {e}")
def main():
"""完整的 GraphRAG 实现"""
print("Starting GraphRAG System")
# 1. 加载和分块文档
documents = load_and_chunk_documents()
# 2. 创建向量存储
vector_store = InMemoryVectorStore.from_documents(documents, embeddings)
# 3. 提取和存储图
graph_documents = llm_transformer.convert_to_graph_documents(documents)
store_graph_documents(graph_documents)
# 4. 创建 RAG 链
chain = create_rag_chain(vector_store)
print("GraphRAG system ready")
return chain
if __name__ == "__main__":
chain = main()
test_graphrag_system(chain)
上述代码展示了如何加载文档、进行语义分块、使用LLM提取实体和关系、将数据存储在内存向量存储和Neo4j中，并最终构建一个RAG链，用于回答用户的问题。
Neo4j与Docker：快速部署你的知识图谱
使用Docker可以快速部署Neo4j图数据库：
docker run \
--name neo4j-graphrag \
-p 7474:7474 -p 7687:7687 \
-d \
-v $PWD/neo4j/data:/data \
-v $PWD/neo4j/logs:/logs \
-v $PWD/neo4j/import:/var/lib/neo4j/import \
-v $PWD/neo4j/plugins:/plugins \
--env NEO4J_AUTH=neo4j/your-password \
neo4j:latest
然后，配置.env文件，包含Neo4j连接信息和Google AI API密钥：
# Neo4j Configuration
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=your-password
# Google AI Configuration
GOOGLE_API_KEY=your-google-api-key
通过访问http://localhost:7474/browser/preview/，你可以使用Neo4j Browser可视化你的知识图谱，并通过Cypher查询语句探索实体间的关系。
内存向量存储：性能与效率的平衡
我们选择内存向量存储的原因在于其卓越的性能和简易性。相较于磁盘存储，内存存储能够实现闪电般的搜索速度，无需复杂的外部数据库设置，非常适合原型设计和小型数据集。当然，内存存储也存在一定的局限性，即容量受限于内存大小。对于大型数据集，可能需要考虑使用磁盘向量数据库。但对于许多实际应用场景，内存向量存储在速度和易用性之间实现了理想的平衡。
结论：GraphRAG的未来展望
GraphRAG代表了文档检索和问答系统的重要演进。通过结合向量检索的语义理解和知识图谱的关系智能，我们可以构建能够真正理解文档上下文和连接的系统。本文介绍的内存向量存储方法在性能和简单性之间提供了极佳的平衡，使其非常适合许多生产用例。
构建GraphRAG系统的关键要点：
- 从小做起： 从核心实体类型和关系开始。
- 积极缓存： 嵌入和图遍历成本很高。
- 监控性能： 内存使用和响应时间很重要。
- 规划扩展： 设计时要考虑增长。
- 彻底测试： 验证准确性和性能。
随着该领域的不断发展，GraphRAG系统将变得越来越复杂，并将结合先进的图算法、多模态处理和实时功能。我们在此处构建的基础为这些未来的增强功能奠定了坚实的基础。