在实际的检索增强生成（RAG）系统中，多跳检索（Multi-hop Retrieval）是一个核心且复杂的挑战。它要求系统能够像人类推理一样，通过多个步骤、从多个文档中收集信息来回答一个复杂问题。例如，回答“苹果公司最新款手机的电池容量是多少？”可能需要先检索“苹果公司最新款手机是哪个型号”，再用这个型号去检索“该型号的电池规格”。传统的多跳RAG实现严重依赖大语言模型（LLM）在查询路径中进行推理和查询改写，这不仅增加了延迟和成本，也让整个系统的稳定性与LLM的可用性深度绑定。

Hubmesh 提出了一种新颖的思路：在查询路径中实现零次 LLM 调用的多跳 RAG 检索。这意味着从用户提问到最终检索出相关文档片段的整个过程中，完全不调用任何 LLM。其核心在于将多跳检索的“推理”逻辑，从运行时（查询时）提前到了构建时（索引时），通过预计算和建立文档间的语义链接网络来实现。对于需要高并发、低延迟、高稳定性且成本敏感的生产场景，这种架构具有显著的吸引力。本文将深入解析 Hubmesh 的设计理念、核心工作机制，并提供一个从零开始的实践指南，帮助你理解如何构建一个不依赖实时 LLM 的、高效的多跳检索系统。

1. 理解 Hubmesh 的核心设计：将推理移至索引时

要掌握 Hubmesh，首先要跳出传统 RAG 的思维定式。传统多跳 RAG 在收到用户查询时，其流程通常是：LLM 分析问题 -> 生成第一个查询 -> 检索初步结果 -> LLM 分析初步结果并生成下一个查询 -> 如此循环，直到收集到足够信息。这个“分析-生成”的循环发生在查询路径中，是延迟和成本的主要来源。

Hubmesh 的设计哲学截然不同。它认为，文档集合内部的知识关联关系是相对静态的。与其在每次查询时动态分析，不如在构建知识库（索引）时，就预先分析并固化这些关联。具体来说，Hubmesh 在索引阶段完成以下关键工作：

1.1 构建文档语义图（Document Semantic Graph）

Hubmesh 的核心数据结构是一个图（Graph）。图中的节点（Node）是经过切分后的文档片段（Chunks）。图中的边（Edge）则表示两个片段之间存在强烈的语义关联。这种关联不是简单的关键词匹配，而是通过嵌入模型（Embedding Model）计算出的语义相似度，或者基于规则、实体共现等启发式方法建立的。

例如，有两段文本：

• 片段A: “iPhone 15 Pro 搭载了 A17 Pro 芯片。”
• 片段B: “A17 Pro 芯片采用 3nm 制程工艺，能效比提升显著。”

在索引时，Hubmesh 的嵌入模型会计算这两个片段的向量表示，并判断它们的语义相关性。如果相关性超过预设阈值，系统就会在片段A和片段B之间建立一条有向或无向的边，表示“提及芯片型号”和“描述芯片工艺”这两个知识点是紧密相连的。

1.2 预计算多跳检索路径

建立了文档语义图之后，Hubmesh 会利用图算法进行预计算。它可能会模拟一些常见的多跳问题模式，或者简单地计算图中节点之间在若干跳数内的可达性与关联强度。这些预计算的结果可以被存储为一种“跳转表”或“路径索引”。

当用户查询“iPhone 15 Pro 的芯片制程是多少？”时，系统不再需要 LLM 来分解问题。查询流程变为：

1. 将用户查询向量化。
2. 在图中找到与查询向量最匹配的初始节点（例如，找到关于“iPhone 15 Pro 芯片”的片段A）。
3. 根据预先建立的图结构和预计算的路径，自动、直接地“跳转”到与之强关联的节点（例如，跳转到描述“A17 Pro 制程”的片段B）。
4. 将沿途收集到的节点（片段A和B）作为最终检索结果返回。

这个过程完全在向量检索和图遍历的范畴内完成，无需生成任何中间的自然语言查询，因此实现了查询路径的零 LLM 调用。

1.3 优势与权衡

这种设计的优势非常明确：

• 低延迟：避免了 LLM 生成和推理的时间，检索速度取决于向量检索和图遍历的效率，通常更快。
• 高稳定性与可控性：系统性能不再受外部 LLM API 波动、限流或内容审核政策的影响。检索逻辑完全由内置的图模型和算法决定，可预测、可调试。
• 低成本：消除了查询时 LLM 调用的费用，成本主要集中于一次性的索引构建和嵌入模型调用。

当然，它也需要权衡：

• 索引构建成本高：需要离线处理整个文档集，计算所有片段间的关联，复杂度较高。
• 灵活性受限：对于训练数据中未出现过的、全新的多跳问题模式，其检索效果依赖于预构建的图是否覆盖了相应的关联路径。而传统方法中，LLM 的泛化能力可以应对一些新奇的组合查询。
• 知识更新开销：当源文档更新时，需要重新分析更新部分与全局的关联，更新图结构，这可能比单纯新增向量到传统向量数据库更复杂。

2. 环境准备与核心组件选择

在动手实现一个 Hubmesh 风格的系统之前，我们需要规划技术栈。由于 Hubmesh 本身是一个展示概念的项目，我们可以选用一系列成熟的开源组件来搭建一个具有类似能力的原型系统。

2.1 基础环境与工具

• 操作系统：Linux (Ubuntu 20.04/22.04) 或 macOS， Windows 建议使用 WSL2。
• Python：3.9 或 3.10 版本。这是大多数 AI 库兼容性较好的版本。
• 版本控制：Git。
• 包管理：使用venv或conda创建独立的 Python 环境，避免依赖冲突。

# 创建并激活虚拟环境 (以 venv 为例) python3.9 -m venv hubmesh_env source hubmesh_env/bin/activate # Linux/macOS # hubmesh_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip

2.2 核心库依赖

我们将需要以下几类库：

1. 文档加载与处理：用于读取 PDF、Word、Markdown 等格式的文档。
2. 文本分割：将长文档切分为语义片段。
3. 嵌入模型：生成文本的向量表示。我们将使用一个本地运行的轻量级模型。
4. 向量数据库与图数据库：存储向量和关联图。为了简化，我们可以使用支持向量索引的图数据库，或组合使用两个数据库。
5. 图计算库：用于执行图遍历和路径查找算法。

以下是一个requirements.txt文件的示例内容：

# 文档处理 langchain==0.1.0 langchain-community==0.0.10 unstructured[pdf,docx]==0.10.30 pypdf==3.17.0 # 文本分割与嵌入 sentence-transformers==2.2.2 # 或者使用 OpenAI 嵌入（非零调用，仅用于对比或索引）。此处我们选本地模型。 # openai==1.6.1 # 向量与图存储（以 Neo4j 为例，它支持向量索引插件） neo4j==5.14.0 # 需要单独安装并运行 Neo4j 数据库 # 图算法与通用工具 networkx==3.1 numpy==1.24.3 pandas==2.0.3 tqdm==4.66.1

使用 pip 安装：

pip install -r requirements.txt

2.3 数据库部署：Neo4j

我们选择 Neo4j 作为后端存储，因为它既是属性图数据库，又通过neo4j-vector-index插件支持向量索引，完美契合 Hubmesh 的“图节点+向量”数据模型。

1. 安装 Neo4j：访问 Neo4j 下载中心 ，选择社区版，按照官方指南进行安装。
2. 启动 Neo4j：

   # 假设使用默认安装路径和配置 neo4j console

   首次启动后，通过浏览器访问http://localhost:7474，默认用户名和密码都是neo4j，登录后会要求更改密码。
3. 安装向量索引插件：对于 Neo4j 5.x，插件通常已内置或可通过APOC库实现。确保你的 Neo4j 版本支持db.index.vector相关过程。你也可以使用graphacademy/neo4j-vector-index等 Docker 镜像，它们预装了相关组件。

3. 构建 Hubmesh 风格的多跳检索系统

现在，我们开始实现核心流程。整个过程分为两个主要阶段：离线索引构建和在线查询检索。

3.1 第一阶段：离线索引构建

这个阶段的目标是处理原始文档，构建出文档语义图并存入 Neo4j。

3.1.1 文档加载与分割

首先，准备你的文档。假设我们有一个docs/文件夹，里面存放着各种格式的文件。

# build_index.py import os from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, UnstructuredWordDocumentLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer import numpy as np from neo4j import GraphDatabase import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 1. 加载文档 def load_documents(data_path): loaders = [] # 加载 PDF 文件 if any(fname.endswith('.pdf') for fname in os.listdir(data_path)): loaders.append(DirectoryLoader(data_path, glob="**/*.pdf", loader_cls=PyPDFLoader)) # 加载 Word 文件 if any(fname.endswith(('.docx', '.doc')) for fname in os.listdir(data_path)): loaders.append(DirectoryLoader(data_path, glob="**/*.docx", loader_cls=UnstructuredWordDocumentLoader)) # 可以添加更多格式... documents = [] for loader in loaders: documents.extend(loader.load()) logger.info(f"Loaded {len(documents)} documents.") return documents # 2. 分割文本 def split_documents(documents): # 使用递归字符分割器，尽量保持语义完整性 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个片段大约500字符 chunk_overlap=50, # 片段间重叠50字符，保持上下文 separators=["\n\n", "\n", "。", "？", "！", "；", "，", " ", ""] ) chunks = text_splitter.split_documents(documents) logger.info(f"Split into {len(chunks)} chunks.") return chunks if __name__ == "__main__": DATA_PATH = "./docs" raw_docs = load_documents(DATA_PATH) text_chunks = split_documents(raw_docs) # 后续步骤...

3.1.2 生成嵌入向量并建立关联图

这是最关键的步骤。我们需要为每个片段生成向量，并计算片段之间的语义相似度，以此建立边。

# build_index.py (续) # 3. 初始化嵌入模型和 Neo4j 驱动 EMBEDDING_MODEL_NAME = 'all-MiniLM-L6-v2' # 一个轻量且效果不错的句子嵌入模型 embedder = SentenceTransformer(EMBEDDING_MODEL_NAME) NEO4J_URI = "bolt://localhost:7687" NEO4J_USER = "neo4j" NEO4J_PASSWORD = "your_new_password" # 替换成你修改后的密码 driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD)) def create_vector_index(tx): # 在 Neo4j 中创建向量索引（如果插件支持） # 注意：社区版可能不支持原生向量索引，以下查询仅为示例。 # 实际中可能需要使用 APOC 库或依赖外部向量数据库。 # 这里我们假设使用一个支持向量的属性来模拟。 tx.run(""" CREATE VECTOR INDEX `chunk_embeddings` IF NOT EXISTS FOR (n:Chunk) ON (n.embedding) OPTIONS {indexConfig: { `vector.dimensions`: 384, `vector.similarity_function`: 'cosine' }} """) logger.info("Vector index created or already exists.") def store_chunks_and_edges(chunks): with driver.session() as session: # 清空旧数据（谨慎操作，生产环境应增量更新） session.run("MATCH (n) DETACH DELETE n") session.execute_write(create_vector_index) chunk_data = [] for i, chunk in enumerate(chunks): text = chunk.page_content metadata = chunk.metadata # 生成嵌入向量 embedding = embedder.encode(text).tolist() # 转换为 list chunk_data.append({ 'id': i, 'text': text, 'embedding': embedding, 'source': metadata.get('source', 'unknown'), 'page': metadata.get('page', 0) }) # 批量创建节点 logger.info("Creating chunk nodes...") session.execute_write(lambda tx: tx.run(""" UNWIND $data AS row CREATE (c:Chunk { id: row.id, text: row.text, embedding: row.embedding, source: row.source, page: row.page }) """, data=chunk_data)) # 建立语义边：为每个节点寻找最相似的Top-K个其他节点 logger.info("Building semantic edges...") all_embeddings = np.array([d['embedding'] for d in chunk_data]) # 这里使用简单的余弦相似度计算，大数据集需使用近似最近邻(ANN)库如FAISS from sklearn.metrics.pairwise import cosine_similarity sim_matrix = cosine_similarity(all_embeddings) edges_to_create = [] top_k = 3 # 每个节点连接最相似的3个其他节点 for i in range(len(chunk_data)): # 获取相似度（排除自身） similarities = sim_matrix[i] similarities[i] = -1 # 将自身相似度设为-1，确保不被选中 top_indices = np.argsort(similarities)[-top_k:][::-1] # 取相似度最高的top_k个 for j in top_indices: if similarities[j] > 0.6: # 设置一个相似度阈值，例如0.6 edges_to_create.append({ 'from_id': i, 'to_id': j, 'similarity': float(similarities[j]) }) logger.info(f"Creating {len(edges_to_create)} semantic edges...") session.execute_write(lambda tx: tx.run(""" UNWIND $edges AS edge MATCH (a:Chunk {id: edge.from_id}) MATCH (b:Chunk {id: edge.to_id}) MERGE (a)-[r:SEMANTIC_LINK]->(b) SET r.similarity = edge.similarity """, edges=edges_to_create)) logger.info("Index build complete.") if __name__ == "__main__": # ... 接前面的加载和分割代码 text_chunks = split_documents(raw_docs) store_chunks_and_edges(text_chunks) driver.close()

注意：上述代码中的向量索引创建和相似度计算是简化版。在生产环境中，面对成千上万的片段，全量计算余弦相似度矩阵（O(N²)复杂度）是不可行的。你需要使用像 FAISS、Annoy 或 HNSW 这样的近似最近邻（ANN）库来高效地找到每个节点的 Top-K 最近邻。Neo4j 的向量索引插件内部也使用了类似技术。

3.1.3 优化：引入实体链接增强关联

纯粹的语义相似度有时会漏掉一些重要的逻辑关联（例如，“苹果公司”和“iPhone”）。我们可以引入实体识别（NER）和链接来增强图的边。

# 可选步骤：使用 spaCy 或 NLTK 进行实体识别 # pip install spacy # python -m spacy download en_core_web_sm import spacy nlp = spacy.load("en_core_web_sm") def extract_entities(text): doc = nlp(text) return [ent.text for ent in doc.ents if ent.label_ in ['ORG', 'PRODUCT', 'PERSON', 'GPE']] # 在存储节点时，可以添加一个 `entities` 属性 # 在建立边时，除了语义相似度，如果两个片段共享重要实体，也可以建立一条边，权重可以不同。 # 例如： # if len(set(chunk_a_entities) & set(chunk_b_entities)) > 0: # create_edge(a, b, type='ENTITY_CO_OCCURRENCE', weight=1.0)

3.2 第二阶段：在线查询检索（零 LLM 调用）

索引构建完成后，在线查询就变得非常高效和直接。

# query.py from sentence_transformers import SentenceTransformer from neo4j import GraphDatabase import numpy as np class HubmeshRetriever: def __init__(self, neo4j_uri, neo4j_user, neo4j_password, embedding_model_name='all-MiniLM-L6-v2'): self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password)) self.embedder = SentenceTransformer(embedding_model_name) self.vector_dim = 384 # 与模型维度一致 def retrieve(self, query_text, max_hops=2, top_k_init=5, top_k_per_hop=3): """ 执行多跳检索。 :param query_text: 用户查询 :param max_hops: 最大跳数 :param top_k_init: 初始向量检索返回的节点数 :param top_k_per_hop: 每跳探索的边数（按相似度排序） :return: 检索到的文本片段列表，按相关性排序 """ # 1. 将查询向量化 query_embedding = self.embedder.encode(query_text).tolist() # 2. 第一跳：基于向量的初始检索 initial_nodes = self._vector_search(query_embedding, top_k=top_k_init) if not initial_nodes: return [] # 3. 多跳图遍历 visited_node_ids = set() result_nodes = [] # 使用一个简单的队列进行广度优先搜索 (BFS) # 队列元素：(node_id, current_hop) queue = [(node['id'], 0, node['score']) for node in initial_nodes] # (id, hop, cumulative_score) while queue: node_id, current_hop, cum_score = queue.pop(0) if node_id in visited_node_ids: continue visited_node_ids.add(node_id) # 获取该节点对应的文本信息 node_info = self._get_node_by_id(node_id) if node_info: # 可以简单用累积分数作为相关性，更复杂的可以加权平均 result_nodes.append((node_info, cum_score)) # 如果未达到最大跳数，继续探索 if current_hop < max_hops: # 获取该节点的出边（语义链接），按相似度排序 neighbor_edges = self._get_semantic_links(node_id, limit=top_k_per_hop) for edge in neighbor_edges: neighbor_id = edge['target_id'] edge_similarity = edge['similarity'] # 新的累积分数 = 当前分数 * 边权重（这里用相似度） new_score = cum_score * edge_similarity queue.append((neighbor_id, current_hop + 1, new_score)) # 4. 去重并按分数排序 unique_results = {} for node_info, score in result_nodes: node_id = node_info['id'] if node_id not in unique_results or score > unique_results[node_id][1]: unique_results[node_id] = (node_info, score) sorted_results = sorted(unique_results.values(), key=lambda x: x[1], reverse=True) return [node_info for node_info, _ in sorted_results] def _vector_search(self, query_embedding, top_k=5): # 使用 Neo4j 的向量索引进行搜索 # 注意：此为示意查询，实际语法取决于你的 Neo4j 版本和插件 with self.driver.session() as session: result = session.run(""" CALL db.index.vector.queryNodes('chunk_embeddings', $top_k, $embedding) YIELD node, score RETURN node.id AS id, node.text AS text, score ORDER BY score DESC """, top_k=top_k, embedding=query_embedding) return [{"id": record["id"], "text": record["text"], "score": record["score"]} for record in result] def _get_node_by_id(self, node_id): with self.driver.session() as session: result = session.run(""" MATCH (c:Chunk {id: $id}) RETURN c.id AS id, c.text AS text, c.source AS source """, id=node_id) record = result.single() if record: return {"id": record["id"], "text": record["text"], "source": record["source"]} return None def _get_semantic_links(self, source_id, limit=3): with self.driver.session() as session: result = session.run(""" MATCH (c:Chunk {id: $source_id})-[r:SEMANTIC_LINK]->(neighbor:Chunk) RETURN neighbor.id AS target_id, r.similarity AS similarity ORDER BY r.similarity DESC LIMIT $limit """, source_id=source_id, limit=limit) return [{"target_id": record["target_id"], "similarity": record["similarity"]} for record in result] def close(self): self.driver.close() if __name__ == "__main__": retriever = HubmeshRetriever("bolt://localhost:7687", "neo4j", "your_new_password") query = "What is the manufacturing process of the chip used in iPhone 15 Pro?" results = retriever.retrieve(query, max_hops=2) print(f"Query: {query}") print("="*50) for i, node in enumerate(results[:5]): # 显示前5个结果 print(f"Result {i+1} (ID: {node['id']}, Source: {node.get('source', 'N/A')}):") print(node['text'][:300] + "...") # 打印前300字符 print("-"*50) retriever.close()

这个retrieve方法实现了零 LLM 调用的多跳检索：

1. 向量化查询：使用与索引时相同的嵌入模型。
2. 初始检索：在向量空间中找到与查询最匹配的几个初始节点。
3. 图遍历：以这些初始节点为起点，沿着预建立的SEMANTIC_LINK边进行广度优先搜索（BFS），探索多跳范围内的关联节点。
4. 结果聚合与排序：收集所有访问到的节点，根据路径上的累积相似度（或其他评分策略）进行排序和去重，返回最相关的文本片段。

整个过程没有生成任何中间查询，完全基于预计算的图结构进行检索。

4. 运行验证与结果分析

4.1 准备测试数据

为了验证系统，你需要准备一些包含明确多跳关系的文档。例如，创建几个文本文件：

apple_company.txt:

Apple Inc. is an American multinational technology company headquartered in Cupertino, California. Its best-known hardware products include the iPhone smartphone.

iphone_15.txt:

The iPhone 15 Pro is the flagship smartphone announced by Apple in September 2023. It is powered by the Apple A17 Pro system-on-a-chip (SoC).

a17_chip.txt:

The Apple A17 Pro is a 64-bit ARM-based system on a chip (SoC). It is manufactured by TSMC using its first-generation 3 nm process, known as N3B.

将这些文件放入docs/文件夹。

4.2 执行索引构建

运行构建脚本：

python build_index.py

观察日志，确认文档被加载、分割、向量化并存储到 Neo4j 中。你可以通过 Neo4j 浏览器 (http://localhost:7474) 执行MATCH (n) RETURN n LIMIT 25来可视化查看创建的节点和边。

4.3 执行查询测试

运行查询脚本：

python query.py

对于查询 “What is the manufacturing process of the chip used in iPhone 15 Pro?”，理想的检索路径应该是：

1. 查询向量匹配到关于 “iPhone 15 Pro” 的片段。
2. 通过SEMANTIC_LINK跳转到描述 “A17 Pro chip” 的片段。
3. 再通过SEMANTIC_LINK跳转到描述 “3 nm process” 的片段。

最终返回的结果列表中，应该包含描述芯片制造工艺的文本片段。你可以在控制台看到输出的结果摘要。

4.4 验证关键特性

• 零 LLM 调用：检查整个query.py执行过程中，没有调用任何 LLM API（如 OpenAI, Claude 等）。所有操作都是本地嵌入模型计算和数据库查询。
• 多跳能力：尝试更复杂的问题，如“苹果公司总部所在州的缩写是什么？”。这可能需要：苹果公司 -> 位于加州 -> 加州缩写是 CA。观察系统是否能通过图连接找到答案。
• 速度：对比传统 RAG（需要串行调用 LLM 进行查询改写）和本方案的速度。本方案的延迟应主要集中在第一次向量检索和少量的图遍历查询上。

5. 常见问题排查

在实现和运行上述系统时，你可能会遇到以下问题：

关于向量索引的替代方案： 如果 Neo4j 社区版向量索引不可用，一个更通用的生产架构是：

1. 使用专门向量数据库（如 Qdrant, Weaviate, Pinecone）存储片段向量和提供向量检索。
2. 使用 Neo4j存储片段元数据（id, text, source）和语义边。
3. 检索时，先从向量数据库拿到 Top-K 初始片段 ID，再用这些 ID 到 Neo4j 中查询它们的邻居。

6. 生产环境最佳实践与扩展方向

将原型转化为生产可用的系统，需要考虑更多因素。

6.1 性能与可扩展性

1. 向量检索优化：必须使用工业级 ANN 库（如 FAISS, HNSWLib, ScaNN）或向量数据库。它们能处理百万甚至十亿级向量，并保证毫秒级检索。
2. 图数据库优化：为Chunk节点的id和边上的similarity属性建立索引，加速遍历查询。

   CREATE INDEX chunk_id IF NOT EXISTS FOR (c:Chunk) ON (c.id); CREATE INDEX semantic_link_similarity IF NOT EXISTS FOR ()-[r:SEMANTIC_LINK]-() ON (r.similarity);

3. 缓存策略：对高频或热点查询的结果进行缓存，可以显著降低数据库压力和响应延迟。
4. 异步索引更新：对于频繁更新的文档源，设计增量索引更新流水线，而不是全量重建。

6.2 检索质量提升

1. 混合检索：不要只依赖语义图。可以将基于图的检索结果与传统的 BM25 关键词检索结果进行融合（Hybrid Search），提高召回率。
2. 重排序（Re-ranking）：图检索返回的片段集合，可以经过一个轻量级的交叉编码器（Cross-Encoder）模型进行精排，提升最终结果的相关性。注意：重排序发生在检索之后，不违反“查询路径零 LLM 调用”的原则，因为它不生成查询。
3. 动态边权重：边的权重（相似度）可以不是静态的。可以考虑结合共现频率、实体关联强度、用户反馈等信息动态调整。
4. 路径多样性：在 BFS 遍历时，可以考虑探索多条路径，而不仅仅是分数最高的一条，以避免陷入局部最优。

6.3 系统监控与运维

1. 指标监控：监控查询延迟、QPS、图数据库和向量数据库的负载、缓存命中率。
2. 效果评估：定期用一组标准问题集评估检索系统的准确率、召回率，跟踪效果变化。
3. 日志与追踪：为每个查询记录详细的日志，包括初始检索结果、遍历的路径、最终返回的片段 ID。这对于调试检索逻辑至关重要。

6.4 架构扩展：Agentic RAG

Hubmesh 的零 LLM 多跳检索可以作为更复杂的 Agentic RAG（智能体化 RAG）系统的一个高效“子模块”。在这个架构中：

• 规划器（Planner）：一个轻量级模型或规则系统，判断问题是否需要多跳检索，以及大致需要哪些“知识块”。
• 检索器（Retriever）：即我们构建的 Hubmesh 系统，负责高效、准确地获取相关片段。
• 生成器（Generator）：LLM。它接收规划器的指令和检索器提供的精准上下文，生成最终答案。由于上下文已经过精炼和关联，LLM 的负担减轻，幻觉减少，生成质量更高。

在这种架构下，LLM 被移出了高并发的检索路径，只用于最终的答案合成，从而在保证答案质量的同时，提升了系统的整体吞吐量和稳定性。

通过本文的实践，你不仅实现了一个零 LLM 调用的多跳检索原型，更重要的是理解了“将推理从查询时转移到索引时”这一核心设计思想。这种思想可以应用于许多对延迟、成本和稳定性有苛刻要求的 AI 应用场景。下一步，你可以尝试用更真实的业务文档进行测试，优化图构建算法，并将其集成到一个完整的 RAG 问答应用中，亲身体验其与传统方案的区别。