深入 ‘Multi-step Retrieval’：将复杂问题拆解为多个步骤，并在图中循环进行针对性知识补充 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位同仁，各位对人工智能与信息检索技术充满热情的开发者们，下午好！

今天，我们将深入探讨一个在构建智能系统、特别是问答系统中日益重要的范式——“多步检索”（Multi-step Retrieval）。在当今信息爆炸的时代，我们面对的问题不再是简单的关键词匹配，而是充满复杂性、需要深度理解和推理的挑战。传统的一次性检索往往难以招架，而多步检索正是为了应对这些挑战而生，它模拟了人类解决复杂问题的思维过程：将大问题拆解为小问题，逐步获取知识，并根据中间结果动态调整策略，最终构建出完整而准确的答案。

复杂问题的本质与传统检索的局限性

在进入多步检索的核心机制之前，我们首先需要明确什么是“复杂问题”，以及为什么传统的单步检索（例如，基于向量相似度或关键词匹配的检索增强生成RAG）在此类问题面前显得力不从心。

一个“复杂问题”通常具备以下一个或多个特征：

多实体/多关系： 问题涉及多个独立的实体及其之间的复杂关联。
- 示例： “找出所有在过去两年内获得过A轮融资，并且其核心产品使用Rust语言开发的SaaS公司。”
多跳推理： 需要通过一系列逻辑步骤才能得出答案，每个步骤都依赖前一个步骤的结果。
- 示例： “如果A依赖B，B依赖C，而C在服务器X上运行，那么服务器X宕机时，A会受到什么影响？”
隐式知识/背景知识缺失： 问题陈述中没有直接给出所有必要信息，需要系统自行补充常识或领域知识。
- 示例： “为什么Python的GIL会影响多线程程序的性能？”（需要理解GIL是什么，以及其对并发的影响）
模棱两可/开放性： 问题可能存在多种合理解释，或者答案不是唯一的、确定的。
- 示例： “如何优化Web应用的首次加载时间？”（有多种优化策略，需要系统权衡或提供全面建议）
时间/空间/条件约束： 问题中包含复杂的过滤条件或范围。
- 示例： “查找所有2020年后发布，且在GitHub上星标数超过5000的开源机器学习框架，排除掉仅支持GPU的框架。”

传统RAG，虽然通过将检索到的文档与大型语言模型（LLM）结合，在一定程度上缓解了LLM的幻觉问题并注入了外部知识，但其基本模式仍是“一次性检索一批文档，然后LLM基于这批文档生成答案”。这种模式在面对上述复杂问题时，面临以下局限：

上下文窗口限制： 复杂问题往往需要大量的背景信息，但LLM的上下文窗口是有限的。一次性检索过多不相关或冗余信息，会稀释掉真正有用的信息，甚至导致LLM无法有效处理。
“信息过载”与“信息不足”的矛盾： 如果检索范围太广，可能检索到大量无关信息；如果检索范围太窄，又可能遗漏关键信息。精确地一次性检索到所有必要且仅必要的知识，对于复杂查询几乎是不可能完成的任务。
缺乏推理和规划能力： 传统RAG模型本身不具备将复杂问题分解、规划检索路径、逐步推理的能力。它期望检索器能一次性提供所有“拼图块”，然后由LLM来完成“拼图”。
无法动态适应： 如果第一次检索的结果不足以回答问题，传统RAG无法根据现有信息动态调整检索策略，例如生成新的、更具体的子查询。

这些局限性促使我们思考，如何让机器像人类专家一样，面对复杂问题时，不是一下子跳到答案，而是通过一系列有计划、有反馈、有修正的步骤来解决。

多步检索的核心原理：分解、针对性检索与循环补充

多步检索的核心思想，正是对人类解决问题过程的模拟。它将一个复杂的、多方面的问题拆解为一系列更小、更具体的子问题。针对每个子问题，系统执行一次或多次“针对性检索”，获取相应的知识片段。然后，这些知识片段被LLM用于生成中间结果，并根据这些中间结果判断是否需要进一步的检索、修正或推理，从而形成一个循环迭代的过程，直至问题得到满意解答。

让我们用一个简化的流程图来描述这个核心循环（尽管这里无法画图，我会用文字清晰描述其逻辑）：

接收复杂问题
问题分解 (Decomposition)
- LLM或其他规划器将复杂问题拆解为一系列相互关联的、可独立回答的子问题。
- 例如： “分析Python中协程的实现原理与异步编程的性能优势，并对比其与传统多线程的异同。”
- 分解为：
  - 子问题1: “Python协程的实现原理是什么？”
  - 子问题2: “Python异步编程的性能优势有哪些？”
  - 子问题3: “Python协程/异步与传统多线程的异同点是什么？”
子问题执行与针对性检索 (Targeted Retrieval)
- 针对当前待解决的子问题，选择最合适的检索工具（例如，向量数据库、关键词搜索引擎、知识图谱查询、结构化数据库查询等）。
- 执行检索，获取高度相关的知识片段（Documents/Facts）。
知识合成与中间结果生成 (Synthesis & Intermediate Reasoning)
- 将检索到的知识片段提供给LLM，结合当前的子问题，进行信息合成、摘要、推理，生成一个中间答案或判断。
- 这个中间结果可能是一个事实、一个列表、一个观点，甚至是发现了一个新的疑问。
状态评估与迭代决策 (State Evaluation & Iteration Decision)
- LLM或一个控制器评估当前的中间结果：
  - 是否已经充分回答了当前的子问题？
  - 是否需要进一步的检索来补充信息或澄清歧义？
  - 是否可以进入下一个子问题的处理？
  - 是否所有子问题都已解决？
  - 是否需要对最初的问题分解进行调整？
- 根据评估结果，决定下一步行动：
  - 继续检索： 如果信息不足，生成新的、更具体的子查询，回到步骤3。
  - 处理下一个子问题： 如果当前子问题已解决，转到下一个子问题，回到步骤3。
  - 最终答案合成： 如果所有子问题都已解决，将所有中间结果综合起来，生成最终答案，结束循环。
  - 回溯/修正： 如果发现某个中间步骤的结论有误或导致死胡同，可能需要回溯到之前的步骤，甚至重新分解问题。

这个循环的核心在于“针对性知识补充”：每次迭代都基于当前已获取的信息，动态地识别知识缺口，并精确地去填补这些缺口。这避免了盲目地一次性获取大量信息，提高了检索效率和最终答案的质量。

架构模式与关键组件

实现多步检索，通常需要一个灵活的架构，它能够协调不同的功能模块。以下是一些常见的架构模式及其关键组件：

1. Agentic Frameworks (基于代理的框架)

这是当前最流行且强大的实现模式之一，例如LangChain Agents、LlamaIndex Agents等。其核心思想是将LLM作为一个智能“代理”（Agent），赋予它一系列“工具”（Tools），并让它通过观察、思考、行动的循环来解决问题。

LLM (Large Language Model) 作为大脑： 负责理解问题、规划步骤、选择工具、执行推理、合成答案。
工具 (Tools)： 封装了各种外部能力，例如：
- 搜索引擎工具： 调用Google Search、Bing Search等获取最新或广域信息。
- 向量数据库检索工具： 在私有知识库中进行语义搜索。
- 关键词检索工具： 在特定文档集合中进行精确关键词匹配。
- SQL/Cypher查询工具： 与结构化数据库或知识图谱交互。
- API调用工具： 调用外部服务获取特定数据。
- 代码解释器工具： 执行代码进行计算或数据处理。
Agent Executor (代理执行器)： 负责驱动Agent的循环，解析Agent的决策（例如，选择哪个工具，工具的输入是什么），执行工具，并将工具的输出返回给Agent。

Agent工作流示例：

用户输入问题。
Agent (LLM) 接收问题，思考： “我需要哪些信息来回答这个问题？我有哪些工具可以使用？”
Agent 决定： “为了回答这个问题，我首先需要使用搜索引擎查找相关背景信息。”
Agent 生成工具调用指令： search_tool.run("Python 协程实现原理")
Executor 执行 search_tool，获取结果。
结果返回给Agent。
Agent 接收结果，思考： “根据这些信息，我对协程原理有了初步了解。接下来我需要查找异步编程的性能优势。”
Agent 决定： “使用向量数据库检索关于异步编程性能的内部文档。”
Agent 生成工具调用指令： vector_db_tool.run("Python 异步编程性能优势")
Executor 执行 vector_db_tool，获取结果。
结果返回给Agent。
Agent 持续循环，直到它认为收集到足够的信息，然后合成最终答案。

代码示例：使用LangChain构建一个简化的Agent

首先，我们需要设置环境，安装必要的库，并准备一些模拟的工具。

# pip install langchain langchain-openai faiss-cpu wikipedia
import os
from typing import List, Dict, Any, Union
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_core.prompts import PromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import wikipedia # 模拟一个外部搜索工具

# 假设你的OpenAI API Key已经设置在环境变量中
# os.environ["OPENAI_API_KEY"] = "your_openai_api_key_here"

# --- 1. 定义工具 ---

# 模拟一个维基百科搜索工具
def wikipedia_search(query: str) -> str:
    """在维基百科中搜索信息。"""
    try:
        results = wikipedia.summary(query, sentences=3)
        return f"维基百科搜索结果：{results}"
    except wikipedia.exceptions.PageError:
        return f"维基百科未找到关于 '{query}' 的页面。"
    except wikipedia.exceptions.DisambiguationError as e:
        return f"维基百科搜索 '{query}' 存在歧义，请提供更具体的信息。可选页面: {e.options}"

# 模拟一个本地知识库（向量数据库）检索工具
class LocalKnowledgeBaseTool:
    def __init__(self, docs_path: str):
        self.vectorstore = self._init_vectorstore(docs_path)

    def _init_vectorstore(self, docs_path: str):
        # 假设我们有一些本地文档，例如关于Python异步编程的教程
        # 在实际应用中，这里会加载真实的文档并嵌入
        if not os.path.exists(docs_path):
            with open(docs_path, "w", encoding="utf-8") as f:
                f.write("""
                # Python 异步编程基础
                Python的异步编程主要通过`asyncio`库实现，它基于协程（Coroutines）的概念。协程是一种可以暂停和恢复执行的函数，允许程序在等待I/O操作（如网络请求、文件读写）完成时，切换到其他任务，从而提高程序的并发性。

                ## 协程的实现
                协程使用`async def`定义，并用`await`关键字暂停执行，等待一个“可等待对象”（awaitable object）完成。`asyncio.run()`是启动异步程序的入口点。

                ## 性能优势
                相比传统的多线程/多进程，异步编程在I/O密集型任务中表现出显著的性能优势。因为它避免了线程切换的开销，且单个线程可以管理多个并发I/O操作。对于CPU密集型任务，异步编程的优势不明显，甚至可能因为上下文切换的开销而略逊于多线程（尽管Python的GIL限制了多线程的CPU并行）。

                ## 传统多线程的局限
                Python的全局解释器锁（GIL）是多线程并发执行CPU密集型任务的主要障碍。GIL确保同一时刻只有一个线程在执行Python字节码。因此，即使有多个线程，它们也无法在CPU层面真正并行。但在I/O密集型任务中，当一个线程等待I/O时，GIL会被释放，允许其他线程运行。

                ## 常见异步库
                除了`asyncio`，还有`aiohttp`用于异步HTTP请求，`fastapi`用于构建异步Web服务。
                """)

        loader = TextLoader(docs_path, encoding="utf-8")
        documents = loader.load()
        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
        texts = text_splitter.split_documents(documents)
        embeddings = OpenAIEmbeddings()
        vectorstore = FAISS.from_documents(texts, embeddings)
        return vectorstore

    def retrieve_info(self, query: str) -> str:
        """从本地知识库中检索相关信息。"""
        docs = self.vectorstore.similarity_search(query, k=3)
        return "本地知识库检索结果：n" + "n---n".join([doc.page_content for doc in docs])

# 初始化本地知识库
knowledge_base_path = "python_async_knowledge.txt"
local_kb_tool = LocalKnowledgeBaseTool(knowledge_base_path)

# 将函数和方法封装成LangChain的Tool对象
tools = [
    Tool(
        name="Wikipedia Search",
        func=wikipedia_search,
        description="用于在维基百科中搜索通用知识或定义。输入应为具体的搜索词。"
    ),
    Tool(
        name="Local Knowledge Base",
        func=local_kb_tool.retrieve_info,
        description="用于在本地私有知识库中检索关于Python异步编程和并发的详细信息。输入应为关于异步/并发的查询。"
    )
]

# --- 2. 定义Agent的Prompt ---
# 这里使用ReAct模式的Prompt
# ReAct (Reasoning and Acting) 模式鼓励LLM在每次行动前进行思考 (Thought)，
# 然后决定采取哪个行动 (Action)，以及行动的输入 (Action Input)。
# 观察 (Observation) 是工具返回的结果。
prompt_template = PromptTemplate.from_template("""
你是一个经验丰富的编程专家，能够分解复杂问题，并利用提供的工具逐步获取信息来回答问题。
你的目标是提供准确、全面且易于理解的答案。

你有以下工具可以使用:
{tools}

请按照以下格式进行回复:

Question: 用户提出的复杂问题
Thought: 我应该如何分解这个问题？我需要哪些信息？我应该使用哪个工具来获取这些信息？
Action: 选择的工具名称
Action Input: 传递给工具的输入
Observation: 工具的输出
...（这个Thought/Action/Action Input/Observation 循环会一直重复，直到你认为问题已解决）
Thought: 我已经收集到足够的信息来回答原始问题了。
Final Answer: 最终的、全面的答案

现在开始！

Question: {input}
{agent_scratchpad}
""")

# --- 3. 初始化LLM和Agent ---
llm = ChatOpenAI(model="gpt-4o", temperature=0) # 使用GPT-4o，设置温度为0以获得确定性结果

# 创建ReAct Agent
agent = create_react_agent(llm, tools, prompt_template)

# 创建Agent Executor来执行Agent的步骤
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

# --- 4. 运行Agent ---
print("--- 启动多步检索Agent ---")
complex_question = "分析Python中协程的实现原理与异步编程的性能优势，并对比其与传统多线程的异同。"
response = agent_executor.invoke({"input": complex_question})

print("n--- 最终答案 ---")
print(response["output"])

运行上述代码，你将看到Agent的详细思考过程（verbose=True的作用）：

Thought: Agent会首先思考如何分解问题，并决定从哪里开始检索。它可能会先去本地知识库查找关于“Python协程原理”的信息。
Action: Local Knowledge Base
Action Input: Python 协程实现原理
Observation: 本地知识库返回关于asyncio和协程定义的信息。
Thought: Agent根据Observation，继续思考下一个子问题，例如“异步编程的性能优势”。
Action: Local Knowledge Base
Action Input: Python 异步编程性能优势
Observation: 本地知识库返回关于I/O密集型任务性能优势的信息。
Thought: Agent现在有了协程原理和异步性能的知识，接下来它需要对比与传统多线程的异同，特别是GIL的影响。它可能会决定再次查询本地知识库，或者如果觉得本地知识库不够全面，可能会去维基百科搜索Python GIL。
Action: Local Knowledge Base (或 Wikipedia Search)
Action Input: Python 多线程 GIL 限制
Observation: 检索结果会包含GIL对多线程CPU密集型任务的限制。
Thought: Agent现在拥有了所有子问题所需的信息，它会开始综合这些信息，形成一个全面的答案。
Final Answer: Agent生成一个结构化的答案，涵盖协程原理、异步性能优势以及与多线程的对比（包括GIL的影响）。

这个过程清晰地展示了“分解 -> 针对性检索 -> 合成 -> 评估 -> 迭代”的循环。LLM在这里不仅仅是一个答案生成器，更是一个决策者和协调者。

2. Tree-of-Thought (ToT) / Graph-of-Thought (GoT) Approaches

这些方法是Agentic Frameworks的进一步扩展，它们允许Agent在思考过程中探索多个推理路径，而不是线性地进行。

Tree-of-Thought: Agent不只生成一个Thought，而是生成多个可能的Thought路径。它会评估这些路径的潜力，选择最有前途的路径继续探索。如果发现某个路径是死胡同，它可以回溯到之前的决策点，选择另一条路径。这增强了解决复杂问题时的鲁棒性和探索性。
Graph-of-Thought: 比ToT更通用，允许思考过程形成一个任意的图结构，节点代表思考状态或中间结论，边代表推理步骤或工具调用。这使得复杂的非线性推理成为可能。

这些方法通常通过更复杂的Prompt工程或专门的框架来实现，它们的核心思想都是为了在思考和行动的循环中引入更多的“规划”和“自我修正”能力。

3. Structured Query Generation (结构化查询生成)

当知识存储在结构化数据库（如关系型数据库SQL、图数据库Cypher/SPARQL）中时，多步检索可以利用LLM将自然语言问题逐步转换为精确的结构化查询。

工作流示例：

用户问题： “查找所有年销售额超过100万美元，且位于加利福尼亚州的软件公司，并列出其主要联系人。”
LLM接收问题，识别实体和关系： 公司、销售额、州、软件行业、联系人。
LLM生成第一个SQL查询（分解）： SELECT company_id FROM companies WHERE annual_sales > 1000000 AND state = 'California' AND industry = 'Software';
执行SQL查询，获取公司ID列表。
LLM接收查询结果（中间信息），生成第二个SQL查询（迭代）： SELECT contact_name FROM contacts WHERE company_id IN (list_of_company_ids_from_prev_step);
执行第二个SQL查询，获取联系人列表。
LLM综合结果，生成最终答案。

这种模式的优势在于，结构化查询的执行是确定性的，并且可以高效地从大型数据库中提取精确信息，避免了语义检索可能带来的模糊性。

代码示例：简化版NL到SQL多步查询

import sqlite3
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 假设我们有一个简单的SQLite数据库
def create_sample_db():
    conn = sqlite3.connect(':memory:')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE companies (
            id INTEGER PRIMARY KEY,
            name TEXT,
            industry TEXT,
            annual_sales REAL,
            state TEXT
        );
    ''')
    cursor.execute('''
        CREATE TABLE contacts (
            id INTEGER PRIMARY KEY,
            company_id INTEGER,
            name TEXT,
            email TEXT
        );
    ''')
    cursor.execute("INSERT INTO companies (name, industry, annual_sales, state) VALUES ('Tech Innovations Inc.', 'Software', 1500000, 'California');")
    cursor.execute("INSERT INTO companies (name, industry, annual_sales, state) VALUES ('Data Solutions LLC', 'IT Services', 800000, 'Texas');")
    cursor.execute("INSERT INTO companies (name, industry, annual_sales, state) VALUES ('Global AI Co.', 'Software', 2200000, 'California');")
    cursor.execute("INSERT INTO companies (name, industry, annual_sales, state) VALUES ('Web Dev Pros', 'Software', 1200000, 'New York');")

    cursor.execute("INSERT INTO contacts (company_id, name, email) VALUES (1, 'Alice Smith', '[email protected]');")
    cursor.execute("INSERT INTO contacts (company_id, name, email) VALUES (1, 'Bob Johnson', '[email protected]');")
    cursor.execute("INSERT INTO contacts (company_id, name, email) VALUES (2, 'Charlie Brown', '[email protected]');")
    cursor.execute("INSERT INTO contacts (company_id, name, email) VALUES (3, 'David Lee', '[email protected]');")
    cursor.execute("INSERT INTO contacts (company_id, name, email) VALUES (4, 'Eve White', '[email protected]');")
    conn.commit()
    return conn

# LLM用于生成SQL
llm_sql_generator = ChatOpenAI(model="gpt-4o", temperature=0)

def execute_sql_query(conn, query: str) -> str:
    """执行SQL查询并返回结果。"""
    try:
        cursor = conn.cursor()
        cursor.execute(query)
        rows = cursor.fetchall()
        column_names = [description[0] for description in cursor.description]
        if not rows:
            return "SQL查询结果为空。"

        # 格式化输出
        formatted_results = [column_names] + [list(row) for row in rows]
        return str(formatted_results) # 简单转换为字符串，实际中可能需要更精细的表格格式
    except Exception as e:
        return f"SQL查询执行失败: {e}"

def multi_step_nl_to_sql_retrieval(question: str, db_conn):
    """
    多步NL到SQL检索函数。
    """
    llm = ChatOpenAI(model="gpt-4o", temperature=0) # 用于规划和最终回答的LLM

    # 定义数据库表结构信息给LLM
    table_info = """
    表 companies:
    id INTEGER PRIMARY KEY,
    name TEXT,
    industry TEXT,
    annual_sales REAL,
    state TEXT

    表 contacts:
    id INTEGER PRIMARY KEY,
    company_id INTEGER,
    name TEXT,
    email TEXT

    relationships: companies.id = contacts.company_id
    """

    # 步骤1：分解问题，生成第一个SQL查询
    prompt1 = ChatPromptTemplate.from_messages([
        ("system", f"""你是一个SQL专家，任务是将自然语言问题转换为SQLite SQL查询。
        数据库模式信息如下:
        {table_info}

        请针对以下问题，首先生成一个SQL查询来获取符合条件的公司ID。
        只输出SQL查询语句，不要包含任何解释或其他文本。
        """),
        ("user", "{question}")
    ])

    sql_chain1 = prompt1 | llm_sql_generator

    # 假设问题是：查找所有年销售额超过100万美元，且位于加利福尼亚州的软件公司，并列出其主要联系人。
    print(f"原始问题: {question}")

    # 生成第一个SQL查询，获取公司ID
    sql_query_companies = sql_chain1.invoke({"question": "查找所有年销售额超过100万美元，且位于加利福尼亚州的软件公司，列出其ID"})
    print(f"n步骤1: 生成SQL获取公司ID:n{sql_query_companies.content}")

    # 提取SQL语句（这里简化，假设LLM只返回SQL）
    company_ids_query = sql_query_companies.content.strip()

    # 执行第一个SQL查询
    company_results_str = execute_sql_query(db_conn, company_ids_query)
    print(f"n步骤1: SQL执行结果:n{company_results_str}")

    # 解析公司ID
    import ast
    try:
        parsed_results = ast.literal_eval(company_results_str)
        if len(parsed_results) > 1: # 排除列名行
            company_ids = [row[0] for row in parsed_results[1:]]
        else:
            company_ids = []
    except (ValueError, SyntaxError):
        company_ids = []

    if not company_ids:
        return "未能找到符合条件的公司。"

    company_ids_str = ", ".join(map(str, company_ids))
    print(f"n提取到的公司ID: {company_ids_str}")

    # 步骤2：根据公司ID，生成第二个SQL查询，获取联系人
    prompt2 = ChatPromptTemplate.from_messages([
        ("system", f"""你是一个SQL专家，任务是将自然语言问题转换为SQLite SQL查询。
        数据库模式信息如下:
        {table_info}

        现在你已经知道符合条件的公司ID是: {company_ids_str}
        请生成一个SQL查询，列出这些公司的所有联系人的姓名和邮箱。
        只输出SQL查询语句，不要包含任何解释或其他文本。
        """),
        ("user", "列出公司ID为 {company_ids_str} 的所有联系人姓名和邮箱")
    ])

    sql_chain2 = prompt2 | llm_sql_generator

    sql_query_contacts = sql_chain2.invoke({"company_ids_str": company_ids_str})
    print(f"n步骤2: 生成SQL获取联系人:n{sql_query_contacts.content}")

    # 提取SQL语句
    contacts_query = sql_query_contacts.content.strip()

    # 执行第二个SQL查询
    contact_results_str = execute_sql_query(db_conn, contacts_query)
    print(f"n步骤2: SQL执行结果:n{contact_results_str}")

    # 步骤3：综合所有信息，生成最终答案
    final_answer_prompt = ChatPromptTemplate.from_messages([
        ("system", f"""根据以下信息，生成一个清晰、简洁的最终答案：
        原始问题：{question}
        公司查询结果：{company_results_str}
        联系人查询结果：{contact_results_str}

        请以自然语言总结这些信息。
        """),
        ("user", "请综合上述信息，回答原始问题。")
    ])

    final_answer_chain = final_answer_prompt | llm
    final_response = final_answer_chain.invoke({"question": question, 
                                                 "company_results_str": company_results_str,
                                                 "contact_results_str": contact_results_str})

    return final_response.content

# 创建并连接到数据库
db_connection = create_sample_db()

# 运行多步NL到SQL检索
question_nl_sql = "查找所有年销售额超过100万美元，且位于加利福尼亚州的软件公司，并列出其主要联系人。"
final_output = multi_step_nl_to_sql_retrieval(question_nl_sql, db_connection)

print("n--- 最终答案 ---")
print(final_output)

db_connection.close()

这个例子展示了LLM如何作为“翻译官”和“协调者”，将自然语言问题逐步转化为数据库可以理解的结构化查询，并通过多次查询和结果合并来解决复杂问题。

实践中的挑战与高级考量

多步检索并非没有挑战，以下是一些在实际部署时需要考虑的问题：

性能与成本： 每次迭代都需要调用LLM和检索工具，这会增加延迟和API成本。优化迭代次数和工具选择是关键。
错误传播： 早期步骤的错误或不准确的中间结果可能会导致后续步骤的偏离，最终产生错误答案。健壮的错误处理、回溯机制和不确定性处理非常重要。
上下文管理： 随着迭代的进行，LLM需要跟踪越来越多的中间信息和历史对话。如何有效地管理和总结上下文，避免超出上下文窗口限制，同时保留关键信息，是一个持续的挑战。
工具选择和设计： 针对不同的数据源和任务，设计合适的工具至关重要。工具的描述（description）需要非常清晰，以便LLM能够正确地选择和使用。
评估： 如何衡量多步检索系统的性能？传统的单一指标（如准确率）可能不足以反映其在复杂推理和信息整合方面的能力。可能需要更细粒度的评估，例如每一步的决策质量、推理路径的效率等。
自修正与回溯： 一个高级的多步检索系统应具备在发现问题时，能够回溯到之前的决策点，并尝试不同的路径或策略的能力。这通常需要更复杂的规划和决策逻辑。
人机协作： 在某些极端复杂或模糊的情况下，纯自动化的多步检索可能仍无法给出满意答案。引入人机协作机制，允许人类专家在关键决策点介入，可以显著提高系统的鲁棒性。
知识图谱融合： 将多步检索与知识图谱（Knowledge Graph）结合，可以为LLM提供更结构化的事实知识和推理能力，尤其是在处理多跳推理和实体关系时。

展望未来

多步检索代表了问答系统和智能代理发展的一个重要方向。它使我们能够从简单的信息查找，迈向复杂的知识发现和智能推理。随着LLM能力的不断增强，以及Agentic Frameworks的日益成熟，未来的多步检索系统将变得更加智能、自适应和高效。我们可以预见，这些系统将在科学研究、法律分析、医疗诊断、软件开发等需要深度分析和持续学习的领域发挥越来越关键的作用。

通过将复杂问题分解、针对性获取知识并循环迭代优化，多步检索不仅提升了LLM处理复杂任务的能力，也为我们构建更接近人类智能的AI系统提供了强大的范式。它不再是一个“黑箱”，而是一个透明且可追踪的推理过程，让我们能够更好地理解AI如何“思考”和“解决”问题。