还在用单次RAG？多跳检索才是YYDS

“帮我查一下，QLeap-V4和Eco-AI-M2这两款AI芯片，哪个更适合做边缘计算？功耗分别是多少？”

你满怀信心地把这个问题抛给了你的RAG（检索增强生成）应用。几秒钟后，它自信满满地回复了一整段关于“Eco-AI-M2”的介绍，最后还不忘补上一句：“很抱歉，我没有找到关于QLeap-V4的相关信息。”

是不是很熟悉？这就是典型的“一个问题，半吊子回答”。我们的RAG系统，在处理需要综合多个信息来源的复杂问题时，往往会像一只只关注了半截大象的盲人，答案自然也就支离破碎。

今天，我们就来聊聊如何根治这个痛点。我们将构建一个多跳检索（Multi-Hop Retrieval） 系统，它能像一位经验丰富的研究员那样，把一个复杂问题拆解、分发给多个“专家”去并行查找，最后再将信息整合成一份完美的答案。这不仅仅是技术升级，更是RAG应用从“搜索引擎”向“研究助手”的质变。

核心概念类比：当AI学会“分而治之”

想象一下，你是一家大公司的CEO，需要同时了解“北美区”和“欧洲区”的销售数据，然后写一份对比报告。

• 单次RAG就像你派了一个实习生，让他去整个公司里翻找一份同时包含“北美区”和“欧洲区”所有数据的完美报告。结果大概率是，他要么只找到了北美的数据，要么只找到了欧洲的，要么找到了一份完全不相关的财务报表。效率低，且几乎不可能成功。
• 多跳检索（Multi-Hop Retrieval）则更像你训练了一个高效的“行政秘书”。这个秘书会先把你的问题拆解成两个独立的任务：1. 去问“北美区的销售经理”，拿到北美区的数据；2. 去问“欧洲区的销售经理”，拿到欧洲区的数据。两个任务可以同时并行进行。最后，秘书再把两份数据整合起来，为你呈现一份逻辑清晰的对比报告。分工明确，信息完整，逻辑自洽。

正式定义：

多跳检索是一种高级RAG架构，它通过一个“元代理”将用户的复杂问题分解为多个独立的子问题，然后并行地执行多次检索和生成过程，最后将子问题的答案合成，以形成一个全面、精准的最终回答。它模拟了人类解决复杂问题的“分而治之”的思维过程。

我们将使用Python和LangChain、LangGraph来构建这个系统。假设我们已经有了一个基础的向量检索器（retriever）和一个大语言模型（llm），下面就是核心代码。

首先，我们需要定义一个问题分解的结构。这里我们用Pydantic来规范“元代理”的输出格式。

# 环境说明：Python 3.10+, langchain, langgraph, openai等库已安装
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List

classSubQuestions(BaseModel):
"""用于存放分解后的子问题列表的Pydantic模型"""
    questions: List[str] = Field(description="2-3个简单、独立的子问题，它们的答案组合起来能完整回答原始复杂问题。")

这个SubQuestions模型就像是给“元代理”下发的任务书，强制它将复杂问题拆解成一个个简单、可独立回答的子问题。

接下来，我们使用LangGraph来定义整个工作流。首先定义图的状态。

from typing import TypedDict, List, Dict, Annotated
import operator

classMultiHopRAGState(TypedDict):
    original_question: str  # 原始问题
    sub_questions: List[str]  # 分解后的子问题列表
# 字典，用于存储每个子问题的答案，键为子问题，值为答案
    sub_question_answers: Annotated[Dict[str, str], operator.update]
    final_answer: str  # 最终合成的答案

节点1：问题分解器（元代理的“策略官”）

这个节点的任务，是制定“作战计划”，而不是执行。

# Node 1: 分解器 (元代理的第一步)
from langchain_core.prompts import ChatPromptTemplate

decomposer_prompt = ChatPromptTemplate.from_template(
"你是一个问题分解专家。你的工作是将一个复杂问题分解成几个简单、独立的子问题，这些子问题可以由检索系统回答。"
"不要试图自己回答问题。\n\n"
"问题: {question}"
)

# 使用 with_structured_output 强制输出为 SubQuestions 模型
decomposer_chain = decomposer_prompt | llm.with_structured_output(SubQuestions)

defdecomposer_node(state: MultiHopRAGState):
"""接收原始复杂问题，并将其分解为子问题列表。"""
    print("--- [元代理] 正在分解复杂问题... ---")
    result = decomposer_chain.invoke({"question": state['original_question']})
    print(f"--- [元代理] 已生成 {len(result.questions)} 个子问题。 ---")
return {"sub_questions": result.questions}

⚠️ 注意：这里90%的人会踩坑with_structured_output 要求LLM支持JSON模式或Function Calling。如果使用的模型不支持，需要退而求其次，用正则表达式从纯文本中解析出子问题列表。但强烈建议使用支持该功能的模型，以保证稳定性。

节点2：并行检索器（元代理的“执行军”）

现在，我们将为每个子问题启动一个独立的RAG流程，并让它们并行执行。这里的关键是使用ThreadPoolExecutor。

from concurrent.futures import ThreadPoolExecutor, as_completed
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 这是一个标准的RAG链，它将作为我们每个并行检索代理的“引擎”
# 假设我们已经定义好了 retriever（检索器）和 generator_prompt（生成提示词）
defformat_docs(docs):
return"\n\n".join([doc.page_content for doc in docs])

sub_question_rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | generator_prompt
    | llm
    | StrOutputParser()
)

defretrieval_agent_node(state: MultiHopRAGState):
"""为每个子问题并行执行完整的RAG过程。"""
    print(f"--- [检索代理] 正在并行回答 {len(state['sub_questions'])} 个子问题... ---")

    answers = {}
# 使用线程池实现并行
with ThreadPoolExecutor(max_workers=len(state['sub_questions'])) as executor:
# 为每个子问题提交一个任务
        future_to_question = {executor.submit(sub_question_rag_chain.invoke, q): q for q in state['sub_questions']}
for future in as_completed(future_to_question):
            question = future_to_question[future]
try:
                answer = future.result()
                answers[question] = answer
                print(f"  - 已找到子问题 '{question}' 的答案")
except Exception as e:
                answers[question] = f"回答该问题时出错: {e}"
# 将收集到的所有答案存入状态
return {"sub_question_answers": answers}

这个节点是系统的动力核心。ThreadPoolExecutor让多个RAG链同时运行，极大地缩短了总响应时间。每个“检索代理”都像一个小型的RAG系统，专注于解决一个简单明确的子问题，因此准确率也更高。

节点3：答案合成器（元代理的“汇报官”）

最后，我们需要一个“汇报官”将所有的碎片信息整合成一份完整的报告。

# Node 3: 合成器 (元代理的最后一步)
synthesizer_prompt = ChatPromptTemplate.from_template(
"你是一个信息合成专家。你的工作是将几个子问题的答案结合起来，形成一个连贯、全面的答案，来回答用户的原始复杂问题。\n\n"
"原始问题: {original_question}\n\n"
"子问题及其答案:\n{sub_question_answers}"
)

synthesizer_chain = synthesizer_prompt | llm | StrOutputParser()

defsynthesizer_node(state: MultiHopRAGState):
"""接收所有子问题的答案，并将它们合成为最终的全面答案。"""
    print("--- [元代理] 正在合成最终答案... ---")

# 将答案字典格式化成易读的字符串
    sub_answers_str = "\n".join([f"- Q: {q}\n- A: {a}"for q, a in state['sub_question_answers'].items()])

    final_answer = synthesizer_chain.invoke({
"original_question": state['original_question'],
"sub_question_answers": sub_answers_str
    })
return {"final_answer": final_answer}

这个合成器并不进行检索，它只做“思考”和“整合”。它拿到的是所有子问题的明确答案，就像拿到了所有证据，然后进行逻辑推理和文字润色，最终生成一份针对原始问题的完美答复。

组装流程图

现在，我们把这三个节点按照顺序组装起来。

from langgraph.graph import StateGraph, END

workflow = StateGraph(MultiHopRAGState)
workflow.add_node("decompose", decomposer_node)
workflow.add_node("retrieve_in_parallel", retrieval_agent_node)
workflow.add_node("synthesize", synthesizer_node)

workflow.set_entry_point("decompose")

workflow.add_edge("decompose", "retrieve_in_parallel")
workflow.add_edge("retrieve_in_parallel", "synthesize")
workflow.add_edge("synthesize", END)
multi_hop_rag_app = workflow.compile()

扩展思考

性能对决与避坑指南

现在，我们用一个原始素材中的复杂问题来测试一下，看看“单次RAG”和“多跳RAG”的表现差异。

user_query = "对比一下QLeap-V4和Eco-AI-M2，重点关注它们的目标应用场景和功耗。"

# 运行简单RAG... 结果如下：
# Simple RAG 输出：
# 根据提供的上下文，Eco-AI-M2芯片是为边缘计算和移动设备设计的，其主要特点是低功耗，满载下仅15W。上下文中没有关于QLeap-V4的信息，因此无法进行对比。

# 运行多跳RAG...
print("--- 子问题及答案 ---")
for q, a in multi_hop_result['sub_question_answers'].items():
    print(f"Q: {q}\nA: {a}")
print("--- 最终合成答案 ---")
print(multi_hop_result['final_answer'])

输出结果对比与分析：

--- 子问题及答案 ---
Q: QLeap-V4的目标应用场景和功耗是什么？
A: QLeap-V4处理器专为数据中心的最大性能而设计，主要用于大规模AI模型训练，满载功耗为1200W。
Q: Eco-AI-M2的目标应用场景和功耗是什么？
A: Eco-AI-M2芯片专为边缘计算和移动设备（如无人机、智能相机）设计，其核心优势是低功耗，满载下仅15W。
--- 最终合成答案 ---
QLeap-V4和Eco-AI-M2在设计目标上截然不同，主要体现在应用场景和功耗上。
- **QLeap-V4**: 这是一款面向数据中心的高性能处理器，主要用于大规模AI模型训练，功耗高达1200W。
- **Eco-AI-M2**: 这是一款面向边缘计算和移动设备的低功耗芯片，专注于能效，功耗仅为15W，适用于无人机和智能相机等场景。

分析：

• 简单RAG：彻底失败。 原因很简单，用户的问题同时包含了两款产品的信息。向量检索试图找到一个与整个问题语义最相似的文档，结果“平均”下来，只找到了关于Eco-AI-M2的内容。它没有能力去主动“寻找”另一个产品的信息。
• 多跳RAG：完美成功。 成功的关键在于其“智能”的分解步骤。元代理将比较性问题拆解成了两个简单、聚焦的子问题（分别问每个产品的信息）。并行检索代理毫无压力地找到了各自对应的文档。最后，合成器拿到了关于两款产品的完整事实，生成一份完美的对比报告就成了水到渠成的事。

避坑指南

生产中你必须知道的3点

1. 子问题的独立性：分解子问题时，务必保证它们是相互独立的。如果子问题之间有依赖关系（比如子问题B的答案需要子问题A的答案作为前提），那么并行执行就会失败。这种情况下，你需要构建“顺序多跳”，而非“并行多跳”的图。
2. 并行数的控制：max_workers 不是越大越好。过高的并行度会瞬间打爆你的检索系统（如向量数据库的连接数限制）和LLM API的并发限制。建议根据实际的系统容量设置一个合理的上限，比如5-10个。
3. 状态管理：在LangGraph中，状态（MultiHopRAGState）是共享的。这意味着多个并行节点同时向sub_question_answers字典写入数据时，可能会有线程安全问题。operator.update注解保证了字典更新的原子性，但在自定义节点中，如果操作更复杂，需要引入锁机制。

写在最后

我们从“一问一答”的简单RAG，进化到了“分而治之、并行探索”的多跳RAG。这不仅仅是代码架构上的升级，更是应用思维的跃迁。当你的业务需要处理复杂研究、产品对比、多源信息整合等高级场景时，多跳检索将是你手中最锋利的武器。

技术是冰冷的，但设计是有温度的。好的架构，应当像一位得力的同事，理解你的意图，拆分你的困难，并行处理任务，最终给你一份清晰、完整的答案。希望今天的分享，能帮助你构建出更懂用户、更智能的RAG应用。

如果你的RAG应用需要处理“今年双十一，A公司和B公司推出的新款手机，谁的性价比更高？”这类问题，你觉得我们的多跳检索框架需要做哪些调整才能完美应对？在评论区聊聊你的想法吧。