GraphMind：构建具备深度推理能力的全栈Agentic RAG架构

本文介绍一套可用于生产环境的完整架构，该架构利用GraphRAG流水线将复杂的非结构化数据转化为高准确度、可检索的知识。我们将阐述Chonkie如何通过语义切分保留上下文，Neo4j如何同时存储向量和图表示以实现双重检索，以及LiteLLM如何编排推理流程。同时，文章将解释系统如何通过智能路由动态切换快速直答与深度多跳推理代理。通过本文，您将了解如何构建一个稳健的、具备代理能力的RAG系统，将向量搜索、图遍历与编排的LLM推理无缝融合。

架构概览

该架构首先摄取原始文档，并通过Chonkie（一个遵循句子边界与逻辑流的语义切分引擎）进行处理，确保切分后不丢失上下文。随后，LiteLLM从这些文本块中抽取原子事实与实体，并将其写入Neo4j。这一过程会形成一个“知识图谱”，其中的数据点不再是孤立的向量，而是相互连接的节点。

当用户提交查询时，智能路由器会分析其复杂度。简单问题将走“快速路径”——直接进行向量检索，毫秒级返回答案。复杂问题则会触发“GraphReader代理”，执行多跳推理：遍历图谱、收集相关事实并综合形成深入的回答。最终，系统既能高效处理简单查询（成本低），又能可靠应对复杂查询（具备深度推理能力）。

为什么选择GraphRAG？

GraphRAG为企业级基础设施带来多项优势。

1. 深度推理：标准RAG基于关键词相似度检索文档。GraphRAG理解的是关系。即使“PD-1”与“免疫疗法”不在同一句话中出现，系统也能“知道”它们之间的关联。
2. 成本效益：借助智能路由，简单问题不会消耗昂贵的Token。只有在确实需要时，才会启用复杂的代理推理。
3. 可追溯性：每个答案都由一条具体的事实路径构建而成。您可以精确追踪代理访问了哪些节点，从而降低幻觉风险。
4. 鲁棒性：系统包含“混合回退”机制。如果高级代理推理失败，系统会优雅降级到标准向量检索，确保用户总能获得答案。

技术栈

我们使用现代、模块化的技术栈来构建这个“认知图谱”。每个组件都因其在流水线中的特定优势而被选用。

• Chonkie：专用切分库，将长文档切分为有意义的“块”（非随机切分），保留语义上下文。
• PyMuPDF4LLM：稳健的PDF解析器，保留版式与表格结构。
• Neo4j：系统的大脑。在同一数据库中同时存储文本（向量）与连接关系（图），支持混合检索。
• LiteLLM：推理引擎。提供统一接口调用多种LLM（如Ollama、Mistral、GPT-4）以进行代理决策。
• LangChain & LangSmith：编排器与调试器。LangChain管理状态与流程，LangSmith提供对代理“思维过程”的可观测性。

数据摄取深潜

在回答问题之前，我们需要先构建知识基础。摄取流水线使用GraphReader提取器将原始文本转化为结构化的知识图谱。

1. 提取原子事实

我们并非将文本整块存入数据库，而是使用LLM提炼“原子事实”——最小且不可再分的信息单元。这样可以避免大块文本常见的“中间迷失”问题。

示例：
输入文本：“Keytruda is an immunotherapy that targets PD-1 to treat lung cancer.”

提取的原子事实（JSON）：
json
{
"atomic_facts": [
{
"key_elements": ["Keytruda", "Immunotherapy"],
"atomic_fact": "Keytruda is a type of immunotherapy."
},
{
"key_elements": ["Keytruda", "PD-1", "Lung Cancer"],
"atomic_fact": "Keytruda targets PD-1 to treat lung cancer."
}
]
}

2. 实体消歧（去重）

在真实数据中，“Google”、“Google Inc.”与“Alphabet”可能指向同一实体。我们使用SpaCy与模糊搜索来识别并合并这些重复项。

• 技术：语义相似度与Levenshtein距离
• 阈值：设置严格阈值（如0.95）以避免误判
• 结果：获得更干净的图谱，关于“Google”的所有事实都指向同一个节点，而非分散成三个。

3. 索引策略

我们构建两种索引以最大化召回率：

1. 向量索引：通过Mistral/Ollama生成1024维嵌入向量，并使用cosine相似度查找语义相近的事实。
2. 关键词索引（全文）：标准的Lucene索引，用于精确匹配（如药品名或编号），弥补向量检索可能遗漏的情况。

在Neo4j中创建向量索引的示例：
cypher
CREATE VECTOR INDEX fact_embeddings IF NOT EXISTS
FOR (n:FactNode) ON (n.embedding)
OPTIONS {indexConfig: {
`vector.dimensions`: 1024,
`vector.similarity_function`: 'cosine'
}}

检索深潜

1. GraphReader检索器

这是代理能力的核心。与普通检索器仅执行查询并返回结果不同，GraphReaderRetriever是一个实现了完整“思维循环”（发现 -> 遍历 -> 评估）的类。

2. 带评分的向量检索

我们不会盲目选取 Top-K 结果，而是通过分析相似度分数（similarity score）来评估检索结果的置信度。

“`python

Inside graph_reader_retriever.py

def initial_discovery(self, state):
# … setup retriever …
results = retriever.search(query_vector=question_embedding, top_k=5)

# 提取分数以评估相关性
scores = [item.score for item in results.items]
print(f"Vector hits: {len(results.items)} (Scores: {scores})")

if max(scores) < 0.7:
    print("Warning: Low confidence match. Expanding search...")
    # 触发备用方案或更广泛的搜索

“`

实现

我们使用现代、模块化的技术栈来构建这个“认知图谱”。每个组件都因其在流水线中的特定优势而被选用。

• Chonkie：专用文档切分库，将长文档切分为有意义的“块”，保留语义上下文。
• Neo4j：作为系统的大脑。在同一数据库中同时存储文本（向量）与连接关系（图），支持混合检索。
• LiteLLM：推理引擎。提供统一接口调用多种 LLM（如 Ollama、Mistral、GPT-4），以执行智能体决策。
• LangChain：编排器。管理智能体图谱的状态与流程。

数据摄取流水线

在回答问题之前，我们需要先“构建大脑”。该摄取流水线负责将原始文本转换为结构化的知识图谱。

工作流

流水线遵循严格的顺序以确保数据质量：

1. 切分：将文档切分为语义块。
2. 提取：使用 LLM 识别实体（如 “PD-1”）与原子事实（如 “PD-1 inhibits T-cells”）。
3. 索引：将事实嵌入为向量，以实现快速检索。
4. 消歧：使用 SpaCy 合并重复实体（例如，将 “Google” 与 “Google Inc.” 收敛为同一个节点）。

代码片段：构建流水线

以下展示如何使用 LangGraph 定义该工作流：

“`python
def _build_workflow(self):
“””Build LangGraph workflow for ingestion.”””
workflow = StateGraph(GraphRAGState)

# 添加所有节点
workflow.add_node("document_loader", self._load_documents)
workflow.add_node("text_chunker", self._chunk_documents)
workflow.add_node("entity_extractor", self._extract_graph)
workflow.add_node("fact_extractor", self._extract_atomic_facts)
workflow.add_node("fact_writer", self._write_atomic_facts)
workflow.add_node("vector_indexer", self._create_vector_index)
workflow.add_node("entity_resolver", self._resolve_entities)

# 定义流程
workflow.set_entry_point("document_loader")
workflow.add_edge("document_loader", "text_chunker")
workflow.add_edge("text_chunker", "entity_extractor")
workflow.add_edge("entity_extractor", "fact_extractor")
workflow.add_edge("fact_extractor", "fact_writer")
workflow.add_edge("fact_writer", "vector_indexer")
workflow.add_edge("vector_indexer", "entity_resolver")
workflow.add_edge("entity_resolver", END)

return workflow.compile()

“`

项目结构

项目脚手架如下：

.
├── src
│ ├── GraphRAG
│ │ ├── Unstructured-kgpipeline-lexical
│ │ │ ├── langchain_graphrag_pipeline.py # 核心流水线
│ │ │ ├── graph_reader_retriever.py # 智能体检索器
│ │ │ ├── retrieval_chain.py # 路由逻辑
│ │ │ ├── langchain_config.yaml # 配置文件
│ │ │ └── requirements.txt
├── .env
└── data
└── medical_research.pdf

阶段 1：智能路由（retrieval_chain.py）

阶段 1：查询分类与路由决策

系统的核心设计在于对查询进行智能分类。我们并非对所有查询一视同仁，而是通过一个轻量级的 LLM 调用来评估问题的复杂度。这个分类器扮演着“交通警察”的角色，将简单查询导向快速检索路径，而将复杂查询交给具备深度推理能力的智能体处理。

python
def _classify_question_complexity(self, question: str) -> str:
"""使用 LLM 对问题复杂度进行分类。"""
try:
prompt = f"""分析以下问题并对其复杂度进行分类：
问题：{question}
分类标准：
- SIMPLE：单事实查找、直接实体查询（例如，“X是什么？”）
- COMPLEX：多跳推理、比较分析（例如，“X如何影响Y？”）
请仅返回一个 JSON 对象：{{“complexity”: “simple”}} 或 {{“complexity”: “complex”}}"""
response = self.llm.invoke(prompt, format=“json”)
result = json.loads(response.content)
return result.get(“complexity”, “simple”).lower()
except Exception:
return “simple” # 默认回退到简单模式

阶段 2：路由逻辑与回退机制

分类完成后，系统根据结果进行路由。设计中包含了一个稳健的回退机制：如果智能体路径（例如图查询）执行失败，系统会自动回退到标准的混合检索器，确保服务的可用性。

“`python
def answer_question(self, question: str, use_agentic: Optional[bool] = None) -> str:
# 1. 确定复杂度
if use_agentic is None:
complexity = self._classify_question_complexity(question)
else:
complexity = “complex” if use_agentic else “simple”

# 2. 尝试图检索器（主路径）
if self.agentic_enabled:
    try:
        return self.graph_reader_retriever.run_retrieval(question, complexity=complexity)
    except Exception as e:
        print(f“[INFO] 回退到标准混合检索”)
        # 主路径失败，执行回退

# 3. 回退路径（次选）
return self.hybrid_retriever.answer_question(question)

“`

阶段 3：智能体核心：图阅读器

这是整个架构的“大脑”。GraphReader 不仅仅是一个函数，它是一个状态机，包含了用于知识发现、分析与评估的多个节点。

跃迁分析器（探索者节点）：该节点执行“思考”过程。它审视已发现的事实，并决定在图数据库中下一步探索的方向。

python
def hop_analyzer(self, state: RetrievalState) -> RetrievalState:
"""智能体遍历：执行无向跃迁以发现相关事实。"""
# Cypher 查询以查找邻居节点
query = “””
MATCH (f:FactNode)-[:HAS_ENTITY]->(e:EntityNode)<-[:HAS_ENTITY]-(neighbor:FactNode)
WHERE f.fact IN $current_facts
RETURN DISTINCT neighbor.fact AS new_fact
“””
# … 执行查询并更新状态 …
return state

评估器（法官节点）：在生成最终答案之前，智能体会暂停并进行自检。当前收集的信息是否足够？如果不够，它将决定是继续探索还是深入挖掘。

“`python
def evaluate_answer(self, state: RetrievalState) -> dict:
“””判断答案是否充分、需要深入挖掘还是需要更多跃迁。”””
prompt = f“””
原始问题：{state[‘original_question’]}
当前收集到的事实：{json.dumps(state[‘notebook’])}

问题是否已得到解答？
1. ‘sufficient’：答案已完备。
2. ‘hop_more’：需要更多相关事实。
3. ‘deepdive’：需要开启新的问题探究线。
“””
# … 调用 LLM 并返回决策 …

“`

实战：多跳推理示例

让我们通过一个复杂的医学问题来观察系统的实际执行轨迹：“免疫疗法如何通过靶向 PD-1 来治疗肺癌？”

第一步：初始发现
系统首先定位到知识图谱中的切入点节点。
* 发现：“PD-1 是一种免疫检查点受体。”
* 发现：“免疫疗法使用检查点抑制剂。”

第二步：跃迁（图遍历）
跃迁分析器识别出“PD-1”与“T细胞”之间的联系，并决定沿着这条路径进行探索。

[[IMAGE_X]]

• Hop：“PD-1 acts as a brake on T-cells.”
• Hop：“Blocking PD-1 releases this brake.”

第三步：Synthesis（Final Answer）

Context Manager 将这些事实整合为最终响应。

Final Answer：Immunotherapy targets PD-1, a checkpoint protein on immune T-cells, to treat lung cancer by blocking its interaction with PD-L1 on cancer cells. Normally, this interaction acts as an “off switch” that prevents T-cells from attacking the cancer. By using checkpoint inhibitors (a form of immunotherapy), this signal is blocked, allowing the T-cells to recognize and destroy the lung cancer cells.

结语

这套实践落地展示了 GraphRAG 流水线如何从根本上提升企业级 AI 系统的质量与可靠性。通过结合 Chonkie 的语义切分、Neo4j 的图能力，以及 Agentic Routing ，我们构建出比仅靠文本的系统更具“看、读、推理”精度的技术栈。

最终得到的是一个将 “thinking” 设为一等公民的稳健工作流。智能路由策略确保每个查询都获得恰到好处的算力——简单问题快速解答，复杂问题深入推理。该架构证明，RAG 的下一次进化不只是更好的 embeddings，而是关于真正的 “cognitive architecture”。

关注“鲸栖”小程序，掌握最新AI资讯