解析 ‘Multi-hop Reasoning over Unstructured Data’：利用 LangGraph 驱动 Agent 在非结构化文档间进行逻辑跃迁 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位编程专家、架构师与技术爱好者，大家好！

今天，我们将深入探讨一个在人工智能领域极具挑战性也极富前景的话题：如何在非结构化数据上进行多跳推理，并利用 LangGraph 驱动的 Agent 来实现这一目标。 随着大模型（LLMs）能力的日益增强，我们发现它们在处理复杂任务时，并非总是能一蹴而就。特别是当问题需要从海量、异构的非结构化信息中抽丝剥茧，经过多步骤的逻辑关联和验证才能得出结论时，传统的单次查询或简单链式调用往往力不从心。

多跳推理是模拟人类高级认知过程的关键能力，它要求系统能够识别问题中的多个子目标，独立或并行地检索相关信息，然后将这些碎片化的信息拼接、整合、推断，最终形成一个完整、连贯的答案。而当这些信息散落在非结构化的文本（如网页、文档、PDF、聊天记录等）中时，挑战更是指数级上升。我们不仅要理解文本内容，还要知道如何有效搜索、如何判断信息的关联性、如何处理不确定性和冲突，并最终构建出清晰的推理路径。

LangGraph，作为 LangChain 生态系统中的一个高级组件，正是为了应对这类复杂场景而生。它提供了一个强大且灵活的框架，允许我们以图形化的方式定义 Agent 的行为流程、状态管理和条件转换。通过 LangGraph，我们可以将一个复杂的推理任务分解为一系列明确的步骤（节点），并根据Agent的决策或外部条件进行路径选择（边），从而实现高度可控、可调试、可扩展的多跳推理系统。

本次讲座，我将从多跳推理的本质挑战入手，逐步深入到 LangChain 和 LangGraph 的核心概念，并通过具体的代码示例，演示如何构建一个能够跨越信息鸿沟、在非结构化数据间进行逻辑跃迁的智能 Agent。我们将涵盖工具的使用、状态的管理、图的构建与执行，以及一些高级的实践考量。

一、多跳推理：驾驭非结构化数据中的认知挑战

多跳推理（Multi-hop Reasoning）指的是解决一个问题需要多个独立的推理步骤，每个步骤的输出可能作为下一个步骤的输入。这个过程通常涉及从不同的信息源中提取事实，并将这些事实通过逻辑关系连接起来，最终形成一个完整的答案。

1.1 多跳推理的本质

想象一个问题：“某公司在收购其主要竞争对手时，其CEO是谁？这家被收购的公司当时最畅销的产品是什么？”要回答这个问题，你需要：

识别目标公司。
查找该公司进行的收购事件。
确定收购事件中被收购的竞争对手。
在收购发生的时间点，查询目标公司的CEO。
查询被收购公司在收购发生时间点左右的最畅销产品。
综合所有信息，形成最终答案。

这个过程显然不是一次性查询就能解决的。它需要多次信息检索、时间点匹配、实体识别和事实关联。

1.2 非结构化数据的挑战

当上述信息散布在非结构化数据中时，挑战变得更加严峻：

信息分散与异构： 相关事实可能存在于不同的文档、网页、数据库条目中，格式各异。
语义理解与歧义： 文本的自然语言特性使得信息提取变得复杂，词语可能有多重含义，实体可能存在别名。
信息冗余与冲突： 大量文本中存在重复信息，甚至相互矛盾的信息，需要Agent具备鉴别和整合能力。
上下文缺失： 单个文本片段可能不足以提供完整上下文，需要Agent主动寻找更多背景信息。
规模性： 在海量数据中高效地定位和提取相关信息是一项巨大的工程。

传统的基于规则或简单的信息检索系统很难有效处理这种复杂性。它们往往缺乏对问题背后深层语义的理解，也无法灵活地规划和执行多步推理。大语言模型（LLMs）的出现，为解决这些问题带来了曙光，它们强大的文本理解、生成和推理能力，使得构建能够进行复杂决策和行动的Agent成为可能。然而，LLMs本身也需要一个结构化的框架来引导其在多步骤任务中保持连贯性、管理状态并有效利用外部工具。

二、LangChain Agent 与 LangGraph：构建智能系统的基石

在深入 LangGraph 之前，我们必须先理解 LangChain Agent 的基础。LangChain 是一个用于开发由语言模型驱动的应用程序的框架，它提供了构建 Agent 所需的各种组件。

2.1 LangChain Agent 的核心思想

LangChain Agent 的核心在于赋予语言模型规划、执行和反思的能力。一个 Agent 通常由以下几个关键部分组成：

LLM (Language Model)： Agent 的“大脑”，负责理解指令、生成推理步骤和最终答案。
Prompt (提示)： 指导 LLM 如何思考、如何行动的指令集，通常包含角色设定、任务描述、可用工具列表等。
Tools (工具)： Agent 的“手脚”，是Agent与外部世界交互的接口，例如搜索引擎、文档检索器、代码解释器、API调用等。
Agent Executor (Agent 执行器)： 负责接收 LLM 的决策（通常是选择一个工具并提供其输入），执行该工具，并将工具的输出反馈给 LLM，形成一个循环。

工作流程概览：

用户提出问题。
Agent Executor 将问题和历史对话（如果存在）传递给 LLM。
LLM 根据Prompt和当前状态，决定下一步行动：是直接回答，还是需要使用某个工具。
如果决定使用工具，LLM 会输出一个结构化的工具调用指令（工具名称、参数）。
Agent Executor 解析指令，调用相应的工具。
工具执行并返回结果。
Agent Executor 将工具结果添加到上下文，再次传递给 LLM。
重复步骤3-7，直到 LLM 认为问题已解决并生成最终答案。

这种循环机制使得Agent能够动态地规划和执行任务，而非简单的链式调用。然而，标准的 LangChain Agent Executor 在处理更复杂的控制流、状态管理和显式多步规划时，会遇到局限。例如，它通常是线性的，依赖 LLM 的自由发挥来决定下一步，难以强制执行特定的逻辑路径或进行复杂的条件跳转。

2.2 LangGraph 的登场：显式控制与状态管理

LangGraph 正是为了解决 LangChain Agent 在复杂流程控制上的不足而设计的。它将 Agent 的执行过程抽象为一个有向图（Graph），其中：

节点 (Nodes)： 代表Agent工作流中的一个步骤或一个Agent本身。一个节点可以是一个工具调用、一个LLM的推理步骤、一个自定义函数或另一个子图。
边 (Edges)： 连接节点，定义了工作流的转换逻辑。边可以是无条件的（从A到B），也可以是条件性的（根据某个函数的返回值决定走向B或C）。
状态 (State)： 图在执行过程中维护的上下文信息。所有节点都可以读取和更新这个状态，状态的更新是幂等的，即多次更新相同的值不会产生副作用。
Checkpoints (检查点)： 允许我们持久化图的状态，从而在中断后恢复执行，或者进行调试和回溯。

LangGraph 优势在于：

显式流程控制： 不再完全依赖 LLM 的自由决策，我们可以通过图的结构，强制 Agent 按照预设的逻辑路径前进。
复杂 Agent 编排： 能够轻松组合多个 Agent、工具和自定义逻辑，构建出高度复杂的决策流。
状态管理： 内置的状态管理机制使得Agent在多步骤任务中能够保持一致的上下文，避免信息丢失。
可调试性与可观测性： 图的结构使得 Agent 的执行路径清晰可见，便于调试和理解其推理过程。
循环与迭代： 能够轻松实现循环结构，例如“尝试X，如果失败则重试Y”或“不断搜索直到找到满意结果”。

LangGraph 与 LangChain Agent 的关系：
LangGraph 并非取代 LangChain Agent，而是对其能力的增强与扩展。我们可以将一个 LangChain Agent 作为 LangGraph 中的一个节点，从而将 Agent 的动态决策能力融入到更宏观、更具结构化的工作流中。

三、构建多跳Agent的基石：工具、Prompt与状态

在着手构建 LangGraph 之前，我们需要准备好 Agent 所需的基本组件。

3.1 Agent 的“感官与行动”：工具 (Tools)

工具是 Agent 与外部世界交互的唯一途径。对于多跳推理，我们通常需要以下几类工具：

搜索工具 (Search Tool)： 用于在互联网上查找实时或广泛的信息。例如 TavilySearchResults、Google Search。
文档检索工具 (Document Retriever)： 用于在预先索引的私有文档库中查找特定信息。这通常涉及向量数据库（Vector Store）和检索增强生成（RAG）技术。
数据解析工具 (Data Parsing Tool)： 用于从特定格式（如JSON、XML、HTML）中提取信息。
自定义工具 (Custom Tools)： 根据特定需求编写的函数，例如调用内部API、执行计算、访问特定数据库等。

代码示例：定义一个文档检索工具

假设我们有一个包含公司历史、产品信息、收购公告等非结构化文档的向量数据库。我们将使用一个简单的内存型 Chroma 向量库作为示例，实际应用中会是 Pinecone, Weaviate, FAISS 等。

import os
from typing import List, Dict, Any, TypedDict
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage

# 设置环境变量，实际使用中请替换为您的API密钥
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
os.environ["TAVILY_API_KEY"] = "YOUR_TAVILY_API_KEY"

# 模拟一些非结构化文档数据
docs_data = [
    "Company A was founded in 2005 by John Doe. Its initial product was 'Alpha Software'.",
    "In 2010, Company A launched 'Beta Platform', which quickly became its flagship product.",
    "Company A acquired Company B in April 2015. At the time of acquisition, Mary Smith was the CEO of Company A.",
    "Company B, founded in 2012, was known for its innovative 'Gamma Analytics' product, which had high market share.",
    "John Doe stepped down as CEO of Company A in late 2014 and was succeeded by Mary Smith.",
    "Company A's CEO in 2005 was John Doe. Its main competitor was Company C.",
    "Gamma Analytics' primary feature was real-time data visualization."
]

# 将文档加载并切分
# 实际应用中，您会从文件或数据库加载
documents = [TextLoader(f"temp_doc_{i}.txt").load()[0] for i, doc_content in enumerate(docs_data)]
for i, doc_content in enumerate(docs_data):
    with open(f"temp_doc_{i}.txt", "w") as f:
        f.write(doc_content)

text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 创建并填充向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(texts, embeddings, collection_name="company_info")
retriever = vectorstore.as_retriever()

# 定义一个自定义工具：文档检索工具
@tool
def retrieve_company_info(query: str) -> str:
    """
    根据查询语句从公司信息文档库中检索相关信息。
    当需要查找关于公司历史、产品、收购事件、人事变动等具体事实时使用。
    输入应是一个清晰的问题或关键词。
    """
    docs = retriever.invoke(query)
    return "n---n".join([doc.page_content for doc in docs])

# 定义一个互联网搜索工具
from langchain_community.tools.tavily_research import TavilySearchResults
tavily_search_tool = TavilySearchResults(max_results=3)

# 将所有工具放在一个列表中
tools = [retrieve_company_info, tavily_search_tool]

print("工具已定义并准备就绪。")

3.2 Agent 的“思考方式”：Prompt Engineering

一个好的 Prompt 是 Agent 成功的关键。对于多跳推理，Prompt 需要引导 LLM：

识别子问题： 帮助 LLM 将复杂问题分解为可管理的子问题。
工具选择： 明确告诉 LLM 何时以及如何使用可用工具。
信息整合： 引导 LLM 将从不同来源获取的信息进行整合、去重、验证。
推理链： 鼓励 LLM 显式地展示其思考过程（Chain-of-Thought）。
反思与纠错： 在必要时引导 LLM 评估其当前状态并进行自我修正。

代码示例：为多跳Agent设计Prompt

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import MessagesPlaceholder

system_prompt = (
    "你是一个高级多跳推理助手，擅长在非结构化数据和互联网上查找信息并进行复杂的逻辑推理。n"
    "你的任务是回答用户提出的问题，这些问题可能需要多个步骤的搜索、提取和整合才能得出答案。n"
    "你拥有以下工具来帮助你完成任务：n"
    "{tools}n"
    "请根据以下步骤进行思考和行动：n"
    "1. **理解问题：** 仔细分析用户的问题，识别其中的关键实体、时间点和需要解决的子问题。n"
    "2. **规划：** 制定一个多步骤的计划来解决问题。每个步骤都应该明确你需要什么信息，以及打算使用哪个工具来获取它。如果信息缺失，请说明你需要查找什么。n"
    "3. **执行：** 使用合适的工具来执行你的计划。确保工具的输入清晰准确。n"
    "4. **整合与推理：** 将从工具获得的信息进行整合。识别相关事实，消除矛盾，并进行逻辑推断以得出中间结论。n"
    "5. **反思与调整：** 评估你当前的进展。是否还需要更多信息？是否需要修正计划？如果发现不确定或不完整的信息，考虑再次使用工具进行验证或补充。n"
    "6. **生成答案：** 当你确信已收集到所有必要信息并完成推理后，以清晰、简洁、直接的方式回答用户的问题。如果无法找到确切答案，请说明原因。n"
    "始终在回答前进行思考，并以 JSON 格式调用工具。你的思考过程（thought）应该清晰地展示你的推理步骤。"
)

# Agent的Prompt模板
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder("messages"), # 这里的messages会包含历史对话和工具输出
    MessagesPlaceholder("agent_scratchpad") # 这里是Agent的思考过程和工具调用
])

print("Agent的Prompt模板已创建。")

3.3 Agent 的“记忆”：状态管理 (State Management)

在 LangGraph 中，Agent 的“记忆”通过图的 State 来实现。State 是一个 TypedDict，定义了在整个图执行过程中需要共享和更新的数据结构。

代码示例：定义图的状态

from typing import Literal
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    """
    表示Agent的当前状态。
    messages: 存储对话历史，包括用户消息、Agent思考、工具调用及工具输出。
    question: 原始的用户问题。
    intermediate_steps: 存储Agent执行的中间步骤（可选，用于调试）。
    """
    messages: List[BaseMessage]
    question: str
    intermediate_steps: List[Any] # 可以存储工具调用、观察结果等
    # 还可以添加其他状态变量，例如：
    # retrieved_info: str # 存储已经检索到的关键信息
    # sub_questions: List[str] # 存储分解出的子问题
    # final_answer: str # 存储最终答案
    # plan: List[str] # 存储Agent的规划步骤

messages 列表至关重要，它会随着 Agent 的思考、工具调用和工具输出而不断增长，作为 LLM 的上下文。

四、利用 LangGraph 设计多跳推理 Agent

现在，我们将结合以上构建块，设计一个能够进行多跳推理的 LangGraph Agent。我们将围绕前面提到的复杂问题：“某公司在收购其主要竞争对手时，其CEO是谁？这家被收购的公司当时最畅销的产品是什么？”来构建我们的 Agent。

4.1 核心组件：LLM 与 Agent Executor

首先，我们需要一个 LLM 和一个 AgentExecutor 的基础实现。

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.runnables import RunnablePassthrough

# 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0) # gpt-4o 在推理和工具调用方面表现优秀

# 创建一个LangChain工具调用Agent
# 这个Agent负责根据Prompt和可用工具，决定下一步是回答问题还是调用工具
langchain_agent = create_tool_calling_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(agent=langchain_agent, tools=tools, verbose=True)

print("LangChain Agent Executor 已初始化。")

4.2 图的节点 (Nodes) 定义

在 LangGraph 中，每个节点都是一个可执行的函数，它接收当前 AgentState 作为输入，并返回一个更新后的 AgentState。

我们将定义以下几种节点：

agent_node: 这是核心的Agent节点，它将调用我们上面定义的 agent_executor。它负责思考、规划、调用工具，并决定是否完成任务。
tool_node: 一个专门用于执行工具调用的节点。当 agent_node 决定调用工具时，我们将转换到此节点。
start_node: 初始节点，负责接收用户输入并初始化状态。
end_node: 结束节点，表示推理完成。

代码示例：定义节点函数

from langgraph.prebuilt import ToolExecutor
from langchain_core.messages import ToolMessage

# 1. ToolExecutor 节点
# 这个节点负责实际执行Agent决定调用的工具
tool_executor = ToolExecutor(tools)

def call_tool(state: AgentState) -> dict:
    """
    根据Agent的决策调用工具。
    Agent的决策通常是AIMessage，其中包含tool_calls。
    """
    messages = state["messages"]
    last_message = messages[-1]

    # 检查最后一个消息是否包含工具调用
    if not last_message.tool_calls:
        raise ValueError("Last message does not contain tool calls to execute.")

    # 执行所有工具调用
    tool_outputs = []
    for tool_call in last_message.tool_calls:
        output = tool_executor.invoke(tool_call)
        tool_outputs.append(ToolMessage(tool_call_id=tool_call.id, content=str(output)))

    # 将工具输出作为新的消息添加到状态中
    return {"messages": messages + tool_outputs}

# 2. Agent 节点
# 这个节点是LangChain Agent Executor的包装，负责核心的决策逻辑
def call_agent(state: AgentState) -> dict:
    """
    调用Agent Executor进行决策（思考、规划、工具调用或生成最终答案）。
    """
    messages = state["messages"]
    # 由于Agent Executor需要完整的历史，我们直接传入所有消息
    # Agent Executor会自行处理prompt和agent_scratchpad
    result = agent_executor.invoke({"messages": messages})

    # 将Agent的响应添加到消息历史中
    return {"messages": messages + [result["output"]]} # result["output"]是AIMessage

4.3 图的边 (Edges) 和条件逻辑

边的定义是 LangGraph 灵活性的核心。它决定了图的流程控制。我们可以定义：

普通边： 从一个节点无条件地转换到另一个节点。
条件边： 根据一个函数的返回值来决定下一步走向哪个节点。这是实现复杂决策逻辑的关键。

我们将需要一个条件函数来判断 agent_node 的输出：

如果 agent_node 决定调用工具，则转换到 tool_node。
如果 agent_node 认为已完成任务并生成了最终答案，则转换到 END。

代码示例：定义条件函数 should_continue

def should_continue(state: AgentState) -> Literal["continue", "end", "call_tool"]:
    """
    根据Agent的最新消息判断下一步应该做什么：
    - 如果Agent调用了工具，则继续调用工具。
    - 如果Agent生成了最终答案，则结束。
    - 否则，继续让Agent思考。
    """
    messages = state["messages"]
    last_message = messages[-1]

    # LangChain Agent的输出是AIMessage，它可能有tool_calls属性
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        print(f"DEBUG: Agent decided to call tool(s): {last_message.tool_calls}")
        return "call_tool"
    elif isinstance(last_message, AIMessage) and last_message.content:
        # 如果Agent的AIMessage有内容但没有tool_calls，我们假设它已生成最终答案
        # 这需要LLM在最终回答时不要包含tool_calls
        print(f"DEBUG: Agent provided final answer: {last_message.content}")
        return "end"
    else:
        # 理论上不会走到这里，除非Agent输出不符合预期
        print(f"DEBUG: Agent's last message is unexpected type or empty: {last_message}")
        return "continue" # 强制继续，防止死循环或未定义行为

4.4 构建并编译 LangGraph

现在，我们有了节点和条件逻辑，可以开始构建我们的 StateGraph。

from langgraph.graph import StateGraph, START

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", call_agent)
workflow.add_node("tool", call_tool)

# 设置入口点
workflow.set_entry_point("agent")

# 添加边
# 从agent节点出发，根据should_continue函数判断下一步
workflow.add_conditional_edges(
    "agent",       # 当前节点
    should_continue, # 判断函数
    {
        "call_tool": "tool", # 如果需要调用工具，转到tool节点
        "end": END,          # 如果任务完成，转到END
        "continue": "agent"  # 如果还需要Agent继续思考（例如，Agent想等待更多信息或重新规划），则回到agent节点
    }
)

# 从tool节点出发，无条件回到agent节点，让Agent处理工具输出
workflow.add_edge("tool", "agent")

# 编译图
app = workflow.compile()

print("LangGraph 工作流已编译。")

4.5 运行多跳推理 Agent

现在，我们可以运行我们的 LangGraph Agent，并观察它如何进行多跳推理。

# 运行示例查询
question = "Company A在收购Company B时，其CEO是谁？Company B当时最畅销的产品是什么？"

# 初始化状态，第一个消息是用户问题
initial_state = {
    "messages": [HumanMessage(content=question)],
    "question": question,
    "intermediate_steps": []
}

print(f"n--- 启动 Agent 对话 ---")
print(f"问题: {question}n")

# 迭代执行图，每一步都会打印DEBUG信息
for s in app.stream(initial_state):
    print(f"--- 当前状态更新 ---")
    for key, value in s.items():
        if key == "messages":
            last_msg = value[-1]
            if isinstance(last_msg, HumanMessage):
                print(f"Human: {last_msg.content}")
            elif isinstance(last_msg, AIMessage):
                if last_msg.tool_calls:
                    print(f"Agent (Thinking, Calling Tool): {last_msg.tool_calls}")
                else:
                    print(f"Agent (Final Answer): {last_msg.content}")
            elif isinstance(last_msg, ToolMessage):
                print(f"Tool Output: {last_msg.content}")
            else:
                print(f"Other Message: {last_msg}")
        else:
            print(f"{key}: {value}")
    print("n")

print("--- Agent 对话结束 ---")

# 打印最终答案
final_state = app.invoke(initial_state)
final_answer_message = final_state["messages"][-1]
if isinstance(final_answer_message, AIMessage):
    print(f"n最终答案: {final_answer_message.content}")
else:
    print(f"n无法获取最终答案，最后一条消息不是AIMessage: {final_answer_message}")

# 清理临时文件
import glob
for f in glob.glob("temp_doc_*.txt"):
    os.remove(f)

预期输出（简化版，实际会有更多思考和工具调用细节）：

--- 启动 Agent 对话 ---
问题: Company A在收购Company B时，其CEO是谁？Company B当时最畅销的产品是什么？

--- 当前状态更新 ---
messages: Human: Company A在收购Company B时，其CEO是谁？Company B当时最畅销的产品是什么？

--- 当前状态更新 ---
DEBUG: Agent decided to call tool(s): [tool_call(id='call_xxx', name='retrieve_company_info', args={'query': 'Company A acquired Company B date'})]
messages: Agent (Thinking, Calling Tool): [tool_call(id='call_xxx', name='retrieve_company_info', args={'query': 'Company A acquired Company B date'})]

--- 当前状态更新 ---
messages: Tool Output: Company A acquired Company B in April 2015. At the time of acquisition, Mary Smith was the CEO of Company A.

--- 当前状态更新 ---
DEBUG: Agent decided to call tool(s): [tool_call(id='call_yyy', name='retrieve_company_info', args={'query': 'Company B best-selling product April 2015'})]
messages: Agent (Thinking, Calling Tool): [tool_call(id='call_yyy', name='retrieve_company_info', args={'query': 'Company B best-selling product April 2015'})]

--- 当前状态更新 ---
messages: Tool Output: Company B, founded in 2012, was known for its innovative 'Gamma Analytics' product, which had high market share. Gamma Analytics' primary feature was real-time data visualization.

--- 当前状态更新 ---
DEBUG: Agent provided final answer: Company A在收购Company B时，其CEO是Mary Smith。Company B当时最畅销的产品是Gamma Analytics。
messages: Agent (Final Answer): Company A在收购Company B时，其CEO是Mary Smith。Company B当时最畅销的产品是Gamma Analytics。

--- Agent 对话结束 ---

最终答案: Company A在收购Company B时，其CEO是Mary Smith。Company B当时最畅销的产品是Gamma Analytics。

这个例子展示了 Agent 如何通过多次调用 retrieve_company_info 工具来获取不同方面的信息，然后将这些信息整合以回答原始的多跳问题。LangGraph 在背后负责了状态的传递和流程的控制。

五、高级考量与最佳实践

构建一个能够有效进行多跳推理的 Agent 并非易事，需要考虑多个维度。

5.1 上下文窗口管理

大型模型虽然强大，但其上下文窗口是有限的。在多跳推理过程中，随着对话历史和工具输出的增加，上下文可能会迅速膨胀，导致：

截断： 最早的信息被丢弃，导致模型失去关键上下文。
性能下降： 处理更长的上下文需要更多计算资源和时间。
“迷失在中间”问题： LLM 难以在大量无关信息中识别关键事实。

应对策略：

摘要 (Summarization)： 在每次工具调用后，让 LLM 总结关键信息，只将摘要添加到状态中，而非原始长文本。
迭代式检索 (Iterative Retrieval)： 不一次性检索所有可能相关的信息，而是根据当前推理步骤逐步细化查询并检索。
长短期记忆 (Long-Term Memory)： 将核心事实抽取出来，存储在向量数据库或知识图中，以便 Agent 随时查阅，而无需每次都将其放入 LLM 的上下文。
上下文压缩 (Context Compression)： 使用 ContextualCompressionRetriever 等技术，在检索后进一步压缩相关文档，只保留最相关的信息。

5.2 错误处理与鲁棒性

Agent 在执行过程中可能会遇到各种问题：

工具调用失败： API 超时、返回错误数据。
信息缺失： 搜索或检索工具找不到相关信息。
信息冲突： 不同来源提供相互矛盾的事实。
Agent 陷入循环： 不断重复相同的思考或工具调用。

应对策略：

重试机制： 为工具调用添加指数退避（exponential backoff）重试逻辑。
回退机制： 当某个工具失败时，尝试使用另一个备用工具，或者通知用户。
最大迭代次数限制： 在 LangGraph 中设置 max_iterations 参数，防止 Agent 无限循环。
反思节点 (Reflection Node)： 增加一个专门的节点，当 Agent 陷入困境或产生不确定性时，引导 LLM 对当前状态进行反思，重新规划或寻求用户帮助。

代码片段：增加一个简单的反思机制

# 在 should_continue 函数中可以增加对Agent输出的更细致判断
# 例如，如果Agent的AIMessage中包含“我无法找到确切信息”或“我需要进一步澄清”，则可以路由到ReflectionNode
# 这里我们仅作概念性演示
def should_continue_with_reflection(state: AgentState) -> Literal["call_tool", "end", "reflect", "continue"]:
    messages = state["messages"]
    last_message = messages[-1]

    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return "call_tool"
    elif isinstance(last_message, AIMessage) and last_message.content:
        # 简单判断是否包含“无法”或“不确定”等关键词
        if "无法" in last_message.content or "不确定" in last_message.content:
            print(f"DEBUG: Agent expressed uncertainty, routing to reflect.")
            return "reflect"
        return "end"
    else:
        return "continue"

def reflect_on_state(state: AgentState) -> dict:
    """
    Agent在此节点反思当前状态，并尝试重新规划。
    可以通过LLM生成新的思考或子问题。
    """
    messages = state["messages"]
    # 构造一个Prompt，引导LLM反思
    reflection_prompt = ChatPromptTemplate.from_messages([
        ("system", "你正在进行多跳推理，但似乎遇到了困难或不确定性。请回顾当前的消息历史，n"
                   "识别你遇到的问题（例如，信息缺失、信息冲突、工具调用失败），n"
                   "并提出一个修正计划或下一个需要采取的明确步骤。你的回复应该是一个新的思考，n"
                   "以帮助Agent继续前进。如果需要，你可以提出新的工具调用建议。"),
        MessagesPlaceholder("messages")
    ])

    reflection_chain = reflection_prompt | llm
    reflection_output = reflection_chain.invoke({"messages": messages})

    # 将反思结果添加到消息历史中，并标记为AI的思考
    return {"messages": messages + [AIMessage(content=f"Reflection: {reflection_output.content}")]}

# 在workflow中添加 'reflect' 节点和相应的边
# workflow.add_node("reflect", reflect_on_state)
# workflow.add_conditional_edges("agent", should_continue_with_reflection, {"call_tool": "tool", "end": END, "reflect": "reflect", "continue": "agent"})
# workflow.add_edge("reflect", "agent") # 反思后回到agent节点继续规划

5.3 评估与监控

多跳推理 Agent 的质量评估是一个复杂的问题，因为它涉及多个步骤的正确性和最终答案的综合质量。

人工评估： 最可靠但成本最高的方式，由人类专家评估 Agent 的推理路径、工具使用、事实准确性和答案完整性。
自动化评估：
- RAGAS： 专门用于评估 RAG 系统的框架，可用于衡量回答的忠实度、上下文相关性、答案相关性等。
- 自定义指标： 设计特定于任务的指标，例如“找到所有关键事实的比例”、“推理路径的效率”等。
日志与可观测性： 记录 Agent 的每次运行、状态变化、工具调用和 LLM 响应，以便进行事后分析和调试。LangGraph 的 checkpoints 功能对此非常有帮助。

5.4 持久化与恢复 (Checkpoints)

LangGraph 内置了 checkpointer 功能，允许我们将 Agent 的完整状态持久化到数据库中。这对于：

长时间运行的任务： 允许 Agent 在中断后从上次的状态恢复执行。
调试与回放： 可以检查 Agent 在任何时间点的状态，理解其决策过程。
A/B 测试与迭代： 可以在不同的 Agent 版本上运行相同的输入，比较它们的行为。

代码片段：使用 Checkpointer

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:") # 使用内存SQLite作为示例

# 编译图时传入checkpointer
app_with_checkpoint = workflow.compile(checkpointer=memory)

# 运行Agent时，可以通过config传入thread_id，实现持久化
config = {"configurable": {"thread_id": "multi_hop_session_1"}}
# for s in app_with_checkpoint.stream(initial_state, config=config):
#     # ... (省略打印逻辑)
#
# # 之后可以加载同一thread_id恢复会话
# restored_state = app_with_checkpoint.get_state(config)
# print(f"Restored state for session 'multi_hop_session_1': {restored_state}")

5.5 并行处理

有时，Agent 需要同时执行多个工具调用或并行探索多个推理路径。LangGraph 可以通过在节点中触发多个工具调用并等待它们全部完成后再继续，或者通过设计分支路径来实现并行。

六、案例研究：构建一个研究与报告生成 Agent

让我们构想一个更复杂的场景：一个研究 Agent，它不仅要回答特定问题，还要围绕一个主题进行深入研究，识别关键实体、关系，并最终生成一份结构化的报告。

Graph 设计思路：

InitializeResearch 节点： 接收用户提供的研究主题，并生成初始的搜索查询列表。
IterativeSearch 节点： 循环执行搜索工具，根据搜索结果更新知识库，并生成新的、更深入的查询。
InformationExtraction 节点： 从检索到的文档中提取关键实体、事件、关系和事实。
KnowledgeGraphBuilder 节点（可选）： 将提取的事实组织成一个简单的知识图谱结构。
DraftReportSection 节点： 根据当前收集到的知识，起草报告的某个章节或部分。
RefineReport 节点： 审查已起草的报告部分，识别信息缺失、不一致或需要改进的地方，并可能触发新的搜索或提取任务。
SynthesizeFinalReport 节点： 整合所有章节，生成最终的结构化报告。
ShouldContinueResearch 条件边： 判断是否需要进行更多搜索，或者报告草稿是否已足够完善。

AgentState 示例：

class ResearchAgentState(TypedDict):
    question: str
    messages: List[BaseMessage]
    research_topic: str
    search_queries: List[str]
    raw_search_results: List[str]
    extracted_facts: List[Dict[str, Any]] # 例如：{"entity1": "X", "relation": "CEO_OF", "entity2": "Y", "source": "doc_id"}
    report_sections: Dict[str, str] # 例如：{"Introduction": "...", "Company_A_History": "..."}
    current_section_to_draft: str # 当前正在起草的报告章节
    # ... 其他状态，如研究深度、时间限制等

节点职能概览：

节点名称	职能	输入 (部分)	输出 (更新状态)
`initialize_research`	解析初始研究主题，生成第一批搜索查询	`research_topic`	`search_queries`
`iterative_search`	执行搜索工具，获取原始搜索结果，并根据结果生成新的搜索查询或确定已收集足够信息。	`search_queries`	`raw_search_results`, `search_queries` (更新)
`information_extraction`	从 `raw_search_results` 中提取结构化的事实、实体和关系	`raw_search_results`	`extracted_facts`
`draft_report_section`	根据 `extracted_facts` 和 `research_topic` 起草报告的特定章节。	`extracted_facts`	`report_sections`
`refine_report`	评估 `report_sections` 的质量、完整性，并决定是否需要进一步搜索或修订。	`report_sections`	`search_queries` (如果需要更多信息), `report_sections` (修订后)
`synthesize_final_report`	整合所有 `report_sections`，生成最终的完整报告。	`report_sections`	`messages` (包含最终报告)

条件边概览：

源节点	条件函数/逻辑	目标节点
`initialize_research`	无条件	`iterative_search`
`iterative_search`	`should_continue_search(state)`	"continue_search" -> `iterative_search` (循环) "extract_info" -> `information_extraction`
`information_extraction`	`should_draft_next_section(state)`	"draft_section" -> `draft_report_section` "refine_existing" -> `refine_report` "finalize" -> `synthesize_final_report`
`draft_report_section`	`should_refine_or_continue_drafting(state)`	"refine" -> `refine_report` "next_section" -> `draft_report_section` (起草下一章节) "finalize" -> `synthesize_final_report`
`refine_report`	`should_re_search_or_finalize(state)`	"re_search" -> `iterative_search` "finalize" -> `synthesize_final_report` "re_draft" -> `draft_report_section`
`synthesize_final_report`	无条件	`END`

这个更复杂的 Agent 能够体现 LangGraph 在编排复杂、多阶段、迭代式工作流方面的强大能力。通过精心设计的节点和条件边，我们可以构建出高度自治、能够进行深度研究和内容创作的智能系统。

七、挑战与未来展望

尽管 LangGraph 为多跳推理 Agent 提供了强大的框架，但该领域仍面临诸多挑战，并拥有广阔的未来发展空间。

7.1 当前挑战

事实一致性与幻觉： LLMs 仍然可能生成不准确或捏造的信息，尤其是在复杂的多跳推理中。确保推理过程中的每个事实都有可靠的来源并能被验证至关重要。
效率与成本： 多跳推理涉及多次 LLM 调用和工具执行，这会显著增加延迟和计算成本。优化推理路径、减少不必要的步骤是持续的挑战。
知识表示与推理： 目前主要依赖 LLMs 的自由文本推理，与结构化知识图谱的深度融合仍是研究热点。如何有效地将非结构化文本中的信息转化为结构化知识，并在此基础上进行更严谨的逻辑推理，是未来的方向。
可解释性与透明度： LLM 的“黑箱”特性使得理解 Agent 的决策过程和推理路径变得困难。提高 Agent 行为的可解释性，让用户能够理解其为何得出某个结论，对于信任和调试至关重要。
动态环境适应性： 真实世界的知识是不断变化的。Agent 需要具备捕捉实时信息、更新内部知识并适应新情况的能力。

7.2 未来展望

更智能的规划器： 结合强化学习或规划算法，使 Agent 能够更自主地学习和优化其多跳推理策略。
多模态推理： 将多跳推理扩展到文本、图像、音频等多种模态的数据。例如，从图片中识别实体，并结合文本信息进行推理。
人机协作： 发展更灵活的 Human-in-the-Loop 机制，让人类专家在 Agent 遇到困难时介入，提供指导或纠正。
自主学习与进化： Agent 能够从每次推理任务中学习，不断改进其工具使用、规划能力和知识库，实现自我优化。
集成高级推理模块： 结合专门的逻辑推理器、数值计算引擎等，弥补 LLMs 在特定类型推理上的不足。

LangGraph 在多跳推理中的核心优势在于其提供了一个清晰、可控的框架，将复杂的逻辑分解为可管理的节点和可编程的转换，从而使开发者能够构建出更智能、更鲁棒的 Agent 系统，有效驾驭非结构化数据带来的挑战。通过不断迭代和优化工具、Prompt以及图的结构，我们将能够解锁Agent在复杂认知任务中的巨大潜力，让它们真正成为我们处理海量信息、解决复杂问题的得力助手。