提升AI智能体检索精度：LangGraph实现并行混合搜索融合实践

本系列旨在介绍增强现代智能体系统可靠性的设计模式，通过直观的方式逐一拆解每个概念，阐述其目的，并实现一个简单可运行的版本，以演示其如何融入现实世界的智能体系统。本系列共包含14篇文章，本文是第12篇。原文为 Building the 14 Key Pillars of Agentic AI。

要优化智能体解决方案的性能与可靠性，离不开软件工程方法的加持，确保各个组件能够协调、并行地运行，并与系统高效交互。例如预测执行，会尝试处理可预测的查询以降低时延；又或者进行冗余执行，即对同一智能体任务重复执行多次，以防单点故障。其他能够增强现代智能体系统可靠性的常见设计模式还包括：

• 并行工具：智能体同时执行独立的 API 调用，以隐藏 I/O 时延。
• 层级智能体：管理者将任务拆分为由执行智能体处理的小步骤。
• 竞争性智能体组合：多个智能体分别提出答案，系统从中选出最佳。
• 冗余执行：两个或多个智能体解决同一任务，以检测错误并提高可靠性。
• 并行检索和混合检索：多种检索策略协同运行，以提升上下文的完整性。
• 多跳检索：智能体通过迭代的检索步骤，收集更深入、更相关的信息。

当然，还有更多其他模式。本系列将实现这些最常用智能体模式背后的核心概念，帮助读者直观理解。

所有相关的理论和实现代码都位于以下 GitHub 仓库中：🤖 Agentic Parallelism: A Practical Guide 🚀

代码库的组织结构如下：

agentic-parallelism/
    ├── 01_parallel_tool_use.ipynb
    ├── 02_parallel_hypothesis.ipynb
    ...
    ├── 06_competitive_agent_ensembles.ipynb
    ├── 07_agent_assembly_line.ipynb
    ├── 08_decentralized_blackboard.ipynb
    ...
    ├── 13_parallel_context_preprocessing.ipynb
    └── 14_parallel_multi_hop_retrieval.ipynb

并行混合搜索融合：获取高保真上下文的关键

向量搜索（语义搜索） 擅长理解查询背后的概念含义，但它有时会漏掉那些包含特定、精确关键词的文档。相反，关键词搜索（词法搜索） 非常擅长查找包含精确术语的文档，却无法理解概念间的语义关系。

并行混合搜索融合（Parallel Hybrid Search Fusion） 架构通过结合这两种方法的优势，为我们提供了一个强有力的解决方案。

该架构的核心思路是同时执行向量搜索和关键词搜索，然后将两者的独特发现“融合”成一个单一、全面的结果集。

这种模式对于任何处理混合了自然语言文本和特定标识符（如产品代码、错误消息、法律案件编号）的 RAG（检索增强生成）系统都至关重要。

接下来，我们将构建并对比三个 RAG 系统：纯向量模式、纯关键词模式以及混合模式。通过这个对比，我们可以清晰地展示混合方法如何检索到更优质的上下文，并最终生成更完整、准确的答案。

首先，我们需要构建两种不同的检索机制。从大家熟悉的向量搜索开始：

from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings

# 为语义搜索创建标准 FAISS 向量存储
vector_store = FAISS.from_documents(kb_docs, embedding=embeddings)
vector_retriever = vector_store.as_retriever(search_kwargs={"k": 2})

这是一个标准的语义搜索引擎，由向量嵌入驱动，在理解查询的“核心意思”方面表现出色。

接下来，我们将利用 scikit-learn 的 TfidfVectorizer 从头开始构建一个经典的关键词检索器，作为基于词汇匹配的搜索引擎。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from typing import List
from langchain_core.documents import Document

class TfidfRetriever(BaseRetriever):
    """利用 TF-IDF 进行关键词搜索的自定义 LangChain 检索器"""
    # 存储拟合的矢量器和原始文档
    vectorizer: TfidfVectorizer
    docs: List[Document]
    k: int = 2

    class Config:
        arbitrary_types_allowed = True

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        # 将查询转换为 TF-IDF 向量
        query_vec = self.vectorizer.transform([query])
        # 获取所有文档的预计算 TF-IDF 向量
        doc_vectors = self.vectorizer.transform([doc.page_content for doc in self.docs])

        # 计算查询和文档之间的余弦相似度
        similarities = cosine_similarity(query_vec, doc_vectors).flatten()

        # 获取前 k 个最相似文档的索引
        top_k_indices = np.argsort(similarities)[-self.k:][::-1]

        # 返回对应的 Document 对象
        return [self.docs[i] for i in top_k_indices]

# 在知识库内容上拟合 TF-IDF 矢量
vectorizer = TfidfVectorizer().fit([doc.page_content for doc in kb_docs])

# 创建自定义检索器实例
keyword_retriever = TfidfRetriever(vectorizer=vectorizer, docs=kb_docs, k=2)

TfidfRetriever 是词汇检索的专家。与向量检索器不同，它不“理解”含义，纯粹基于词频和逆文档频率工作，专门查找那些包含用户查询中确切的、字面关键词的文档，无论这些关键词多么罕见或语义权重多低。

现在，让我们构建混合 RAG 系统的核心。我们将使用一个 LangGraph 节点来并行运行两个检索器，然后融合它们的结果。

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class HybridRAGState(TypedDict):
    question: str
    retrieved_docs: List[Document]
    final_answer: str

def parallel_retrieval_node(state: HybridRAGState):
    """该模式的核心是：并行运行矢量和关键字搜索并融合结果"""
    print("--- [Hybrid Retriever] Running Vector and Keyword searches in parallel... ---")

    # 用同样的问题来调用这两个系统
    # 由于其独立性，所以两个调用可以在真正的多线程系统中并发执行
    vector_docs = vector_retriever.invoke(state['question'])
    keyword_docs = keyword_retriever.invoke(state['question'])

    # “融合”步骤：合并两个文档列表，对数据去重
    all_docs = vector_docs + keyword_docs
    unique_docs = list({doc.page_content: doc for doc in all_docs}.values())

    print(f"--- [Hybrid Retriever] Fused results: Found {len(unique_docs)} unique documents. ---")
    return {"retrieved_docs": unique_docs}

# 用生成节点组装完整的图
workflow = StateGraph(HybridRAGState)
workflow.add_node("parallel_retrieval", parallel_retrieval_node)
workflow.add_node("generate_answer", generation_node)
workflow.set_entry_point("parallel_retrieval")
workflow.add_edge("parallel_retrieval", "generate_answer")
workflow.add_edge("generate_answer", END)
hybrid_rag_app = workflow.compile()

parallel_retrieval_node 是整个融合过程发生的地方。它同时调用 vector_retriever 和 keyword_retriever，然后合并它们的输出。简单的去重操作 list({doc.page_content: doc for doc in all_docs}.values()) 是一种基本但有效的融合策略：如果两个检索器找到了相同的文档，则只保留一份；如果找到了不同的文档，那么所有独特的发现都会被保留在最终的上下文中。

最后，让我们进行一次直接的对比分析。我们将创建一个特定的查询，这个查询被设计成会让单一搜索系统“失效”——它包含一个高级的语义概念（“节能措施”）和一个非常具体、罕见的关键词（ERR_THROTTLE_900）。要给出完整的答案，必须结合两种搜索方法的优势才能找到所有必需的信息。

# 查询包含概念部分和特定关键字部分
user_query = "What are our company's power saving efforts, and what is the error code for QLeap-V4 overheating?"

# --- 执行向量 RAG ---
vector_answer = rag_chain_vector.invoke(user_query)
# --- 执行关键词 RAG ---
keyword_answer = rag_chain_keyword.invoke(user_query)
# --- 执行混合 RAG ---
hybrid_answer = hybrid_result['final_answer']

# --- 最终分析 ---
print("\n" + "="*60)
print("                     ACCURACY & QUALITY ANALYSIS")
print("="*60 + "\n")
print("The User's Goal: The user asked two distinct questions: 1. What are our power saving efforts? (A semantic question) and 2. What is the error code for overheating? (A specific/lexical question).\n")
print("-" * 60)
print("Vector-Only RAG Performance:")
print("- Result: FAILED to answer completely.")
print("- Final Answer:", vector_answer)

print("Keyword-Only RAG Performance:")
print("- Result: FAILED to answer completely.")
print("- Final Answer:", keyword_answer)

print("-" * 60)
print("Hybrid Search RAG Performance:")
print("- Final Answer:", hybrid_answer)

运行后的输出结果清晰地展示了差异：

============================================================
                     ACCURACY & QUALITY ANALYSIS
============================================================

The User Goal: The user asked two distinct questions: 1. What are our power saving efforts? (A semantic question) and 2. What is the error code for overheating? (A specific/lexical question).

------------------------------------------------------------
Vector-Only RAG Performance:
- Result: FAILED to answer completely.
- Final Answer: Based on the provided context, the company power saving effort is an initiative called Project 'Titan', which is focused on developing energy-efficient hardware to reduce power consumption in data centers and is part of the green computing strategy. The context does not contain information about an error code for QLeap-V4 overheating.
- Reason: It excelled at the semantic part, finding documents about 'Project Titan' by matching the concept of 'power saving efforts' to 'energy-efficient hardware'. However, the specific error code 'ERR_THROTTLE_900' was not semantically close enough to the query to be retrieved. The agent correctly stated it could not find the answer.

------------------------------------------------------------
Keyword-Only RAG Performance:
- Result: FAILED to answer completely.
- Final Answer: Based on the context, the error code for QLeap-V4 overheating is 'ERR_THROTTLE_900'. Project 'Titan' is an initiative to reduce power consumption.
- Reason: It excelled at the lexical part, perfectly matching the keyword 'ERR_THROTTLE_900' in the query to the document containing it. However, it missed the second, more conceptual document about 'Project Titan' being part of the 'green computing strategy' because the keywords didnt overlap strongly.

------------------------------------------------------------
Hybrid Search RAG Performance:
- Result: SUCCESS. Answered both parts of the question accurately.
- Final Answer: Our companys power saving effort is called Project 'Titan', which is a core part of our green computing strategy aimed at developing energy-efficient hardware to reduce data center power consumption. The official error code for QLeap-V4 overheating is 'ERR_THROTTLE_900'.
- Reason: The parallel execution and fusion step provided the best of both worlds. The vector search contributed the two documents about 'Project Titan', and the keyword search contributed the document with the specific error code. By combining these unique results into a single, rich context, the generator had all the information it needed to construct a complete and correct answer.

最终的分析对混合方法的优越性做出了明确判断：

• 单一方法的局限性：仅使用向量或仅使用关键词的系统都产生了不完整、部分正确的答案。每个系统都只“看到”了问题的一部分。
• 各自为战的后果：向量搜索理解了“节能”的概念，但错过了错误代码的具体关键词。关键词搜索精准找到了错误代码，但却遗漏了关于公司可持续发展战略（绿色计算）的概念性文档。
• 融合制胜的关键：混合搜索系统之所以成功，不是因为它是一个全知全能的“专家”，而是因为它是一个由专家构成的“协调者”。其并行的“分散-聚集”方法确保了能够从两种截然不同的检索方法中捕获独特的信息片段。
• 上下文完整性的价值：融合步骤创建了一个包含所有必要事实的、单一且全面的上下文。这使得最终的生成式代理能够轻松地综合信息，产出完整且准确的答案，一次性解决了用户复杂查询的两个部分。

这种通过并行与融合来提升系统高可用性和答案质量的设计思路，非常值得在构建复杂 AI 应用时借鉴。如果你对这类 AI智能体的工程化实践感兴趣，欢迎到 云栈社区 与更多开发者交流探讨。