基于 LangGraph 构建可恢复的对话代理：从零实现天气查询机器人

背景与动机：为什么需要状态化对话流

在构建对话式 AI 应用时，一个常见的误区是将每次用户输入视为独立事件，依赖无状态的 LLM 调用直接生成回复。这种模式看似简单，但在真实场景中很快会暴露出根本性问题：上下文断裂。

想象用户发起多轮对话：“明天烟台天气怎么样？”、“那北京呢？”、“记得我之前问的城市吗？”。如果系统每次只看到当前问题而看不到历史消息，就无法正确理解“之前问的城市”指代什么。更严重的是，当涉及工具调用（如查询天气）、条件分支或人工干预时，缺乏统一的状态管理会导致流程失控——比如重复执行、跳过关键步骤。

这就是为什么我们需要状态化对话流。LangGraph 的出现正是为了解决这一类复杂交互的编排问题。它基于有向无环图（DAG）模型，将对话建模为一系列节点与边的流转过程，每个节点代表一个动作（如调用模型、执行工具），每条边决定下一步走向。更重要的是，整个流程围绕一个共享的 state 对象进行演化，确保所有参与者都能看到一致的上下文。

尤为关键的是，LangGraph 支持通过 checkpointer 机制持久化对话状态。这意味着你可以随时暂停会话、保存快照，并在未来任意时刻恢复到断点继续执行。这不仅支持 Web 应用中的长期对话保持，还极大提升了调试能力——开发者可以回放某次失败的交互路径，逐节点检查状态变化，就像调试传统程序一样精准。

简言之，从无状态调用迈向基于 LangGraph 的状态机架构，是从“问答机器人”进化到“可信赖代理”的必经之路。接下来我们将深入其核心抽象，看看它是如何实现这一切的。

关键概念解析：StateGraph 与 MessagesState

在 LangGraph 中，构建可恢复、可追踪的对话代理依赖于几个核心抽象：状态（State）、节点（Node）和边（Edge）。它们共同构成一个带状态的有向无环图（DAG），而这些概念的正确理解是设计稳健代理流程的前提。

首先来看状态定义。代码中通过 TypedDict 定义了 MyMessagesState，并使用 Annotated[List, add_messages] 标注 messages 字段：

python
class MyMessagesState(TypedDict):
    messages: Annotated[List, add_messages]

这里的 add_messages 是一个预置的归约函数（reducer），它确保每次对 messages 的更新不是简单覆盖，而是将新消息追加到历史列表中。这种累积策略对于聊天场景至关重要——模型需要看到完整的对话历史才能做出合理响应。如果不使用此类机制，状态更新很容易误写为替换操作，导致上下文丢失。

接着是 StateGraph，它是整个流程的骨架。你可以把它看作是对标准 DAG 的增强封装，专为状态驱动的代理任务设计。通过 StateGraph(MyMessagesState) 实例化后，你可以在其上注册函数作为节点，如 query_or_respond 或工具执行节点。每个节点接收当前状态作为输入，并返回部分更新的状态，图引擎负责自动合并结果到全局状态中。

值得一提的是，LangGraph 内置支持两种关键能力：

条件转移：借助 tools_condition 这样的路由函数，可以根据 LLM 输出是否包含工具调用，动态决定下一步走向（例如进入工具执行节点或直接结束）。
持久化集成点：StateGraph.compile() 时传入 checkpointer，即可实现每一步状态变更的自动保存与恢复。

最后提一下 MessagesState ——这是 LangGraph 提供的一个开箱即用的状态模式，其内部结构与我们自定义的 MyMessagesState 非常相似，专门用于处理以消息流为核心的对话应用。如果你不需要额外字段（如用户身份、会话元数据等），直接继承 MessagesState 可以进一步简化代码。

这些抽象看似简单，实则构成了可复用、可测试、可调试的代理架构基础。下一节我们将梳理这些组件是如何串联成完整执行流程的。

完整流程设计：从入口到终止的状态流转

一个高效的对话代理不仅需要正确的状态建模，还需要清晰的流程控制。在本例中，整个图结构通过一系列节点与条件边组织起来，形成从用户输入到最终响应的完整闭环。

流程起点由 query_or_respond 节点担任。它的核心职责是接收当前对话状态中的消息历史，并调用绑定工具的 LLM 实例生成回复：

python
def query_or_respond(state: MyMessagesState):
    llm_with_tools = llm.bind_tools([weather])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

这里的关键在于：LLM 是否决定调用工具（如查询天气），完全取决于其输出内容是否包含符合规范的工具调用指令。这正是 LangGraph 灵活性的体现——决策由模型驱动，而非硬编码逻辑。

接下来是流程的“大脑”：条件转移。通过以下代码实现路由判断：

python
graph_builder.add_conditional_edges(
    "query_or_respond",
    tools_condition,
    {
        "tools": "tools",
        END: END
    }
)

tools_condition 是一个内置判别函数，它检查上一步输出中是否存在待执行的工具调用。如果有，则跳转至 "tools" 节点；否则认为任务完成，直接流向 END。这种基于语义内容的动态跳转，使得同一节点既能处理普通问答，也能触发复杂操作。

真正执行工具的是 ToolNode([weather])。该节点会自动遍历当前消息中的工具调用请求，匹配注册的 @tool 函数并执行。例如当模型提出要调用 weather(city="烟台") 时，ToolNode 就会运行对应的函数，获取返回值，并将结果以工具消息（ToolMessage）形式追加回状态中，供后续步骤参考。

最后，test 节点虽然在此示例中为空实现，但它代表了典型的后处理阶段，比如记录日志、触发通知或进行数据清洗。你可以将其视为流程结束前的“收尾钩子”，便于未来扩展业务逻辑。

整体来看，这条路径清晰地表达了“思考 → 判断 → 执行 → 收尾”的典型代理行为模式。这种模块化设计不仅提升了可读性，也为调试和迭代提供了良好基础。

实践示例：实现一个可运行的天气查询机器人

下面是一个完整、可运行的天气查询机器人实现，基于 LangGraph 构建，支持多轮对话与状态恢复。我们将逐段解析其结构与关键设计。

python
from typing import Annotated, List, TypedDict
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START, END
from langgraph.graph import MessagesState, StateGraph, add_messages
from langchain_community.chat_models import ChatTongyi
from langgraph.prebuilt import ToolNode, tools_condition
from pydantic import SecretStr

首先导入所需模块。值得注意的是 SecretStr 来自 Pydantic，用于安全封装敏感信息如 API Key，避免日志泄露。

python
# 初始化通义千问模型
llm = ChatTongyi(
    model="qwen-max",
    api_key=SecretStr("sk-xxxxx")
)

这里使用 ChatTongyi 接入阿里云通义千问大模型。将 api_key 包装为 SecretStr 是最佳实践，确保在打印或序列化时不会暴露密钥。

python
@tool(response_format="content_and_artifact")
def weather(city: str):
    """查询指定城市的天气"""
    return "30℃", {"city": city, "temperature": "30℃"}

通过 @tool 装饰器注册一个工具函数。关键点在于 response_format="content_and_artifact"：它表示该工具返回两个部分：

content：供 LLM 阅读的简洁结果（字符串）
artifact：结构化数据，可用于后续节点处理

这种方式避免了 LLM 因输出格式不一致导致的解析错误，也便于系统内其他组件消费原始数据。

python
class MyMessagesState(TypedDict):
    messages: Annotated[List, add_messages]

定义状态结构。所有消息都会被累积到 messages 字段中，由 add_messages 自动合并新旧内容，防止上下文丢失。

python
def query_or_respond(state: MyMessagesState):
    llm_with_tools = llm.bind_tools([weather])
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

这是主推理节点，负责生成回复或发起工具调用。注意传入的是完整的 state["messages"]，保证模型能看到历史交互。

python
# 构建图结构
graph_builder = StateGraph(MyMessagesState)
graph_builder.add_node(query_or_respond)
graph_builder.add_node("tools", ToolNode([weather]))
graph_builder.add_node(test)

graph_builder.set_entry_point("query_or_respond")

# 条件跳转：根据是否需要工具执行决定流向
graph_builder.add_conditional_edges(
    "query_or_respond",
    tools_condition,
    {
        "tools": "tools",
        END: END
    }
)

graph_builder.add_edge("tools", "test")
graph_builder.add_edge("test", END)

构建状态图的核心逻辑。tools_condition 自动判断是否需进入工具执行流程。ToolNode 会自动处理所有注册工具的调用。

python
# 启用内存检查点，支持对话恢复
graph = graph_builder.compile(checkpointer=InMemorySaver())

InMemorySaver() 提供线程级状态持久化。每个会话可通过 thread_id 独立保存和恢复。虽然当前是内存存储，但未来可轻松替换为数据库后端（如 Redis 或 PostgreSQL）。

python
# 执行示例请求
config = RunnableConfig(configurable={"thread_id": 1})
inputs = {"messages": [HumanMessage(content="烟台天气如何？")]}
result = graph.invoke(inputs, config=config)

print(result["messages"][-1])

调用时通过 configurable={"thread_id": 1} 标识会话 ID，checkpointer 以此为键保存状态。下次只需使用相同 thread_id 即可从中断处继续。

这个最小系统展示了如何将模型、工具、状态与流程控制整合成一个健壮的代理应用，且具备向生产环境扩展的基础能力。

注意事项与常见陷阱

在使用 LangGraph 构建状态化代理时，尽管框架提供了强大的流程控制能力，但一些细微的误用仍可能导致难以排查的问题。以下是开发中常见的陷阱及应对建议。

首先，状态更新必须是合并而非覆盖。这是最易出错的一点。LangGraph 中的状态管理依赖于 reducer 函数（如 add_messages），它会在每次节点返回部分状态时自动累积消息。如果你不小心直接替换了 messages 列表（例如写成 {"messages": [new_msg]} 而非追加），就会丢失历史上下文，导致模型“失忆”。正确的做法是始终返回待添加的消息，并依赖 add_messages 的归约行为完成合并：

python
# ❌ 错误：覆盖式更新，导致上下文丢失
return {"messages": [response]}

# ✅ 正确：让 add_messages 处理累积
return {"messages": [response]}  # 实际上没问题 —— 因为是追加到已有列表

注意：虽然语法相同，关键在于你是否理解其背后是由 Annotated[List, add_messages] 驱动的增量更新机制。切勿手动操作 state["messages"] 并重新赋值整个字段。

其次，工具函数的返回格式必须严格匹配 response_format 声明。当你使用 @tool(response_format="content_and_artifact") 时，返回值必须是一个元组 (content: str, artifact: dict)。如果只返回字符串或字典，LangGraph 将无法正确解析，导致工具执行失败且无明确报错提示：

python
# ❌ 错误：仅返回字符串
@tool(response_format="content_and_artifact")
def weather(city):
    return "30℃"  # 缺少 artifact

# ✅ 正确：返回 (content, artifact) 元组
@tool(response_format="content_and_artifact")
def weather(city):
    return "30℃", {"city": city, "temperature": "30℃"}

建议在开发阶段启用调试日志，观察 ToolNode 是否成功捕获并处理了工具调用结果。

最后，RunnableConfig 中的 thread_id 是 checkpointer 持久化的关键标识。每个独立会话应使用唯一的 thread_id，否则不同用户的对话状态可能混淆或覆盖：

python
config = RunnableConfig(configurable={"thread_id": "user_123"})

若忽略此配置，即使启用了 InMemorySaver 或其他存储后端，系统也无法区分会话，导致恢复时错乱。未来扩展为持久化存储（如数据库）时，这一设计更是不可或缺的基础。

总结：保持对状态更新语义、工具返回格式和会话标识的敏感性，是构建可靠 LangGraph 应用的关键防线。这些细节看似微小，实则决定了系统能否稳定运行于生产环境。

总结与延伸：构建更复杂的代理工作流

本文实现的天气查询机器人虽小，却完整呈现了一个现代对话代理的核心架构模式：以状态机驱动流程、用工具扩展能力、通过 checkpointer 实现可恢复性。这一模式具有高度可迁移性，适用于客服工单处理、自动化运维、智能助手等需要多轮决策与外部交互的场景。

例如，在企业级客服系统中，你可以将 weather 工具替换为“查询订单”、“创建工单”或“触发退款”等业务接口；在运维自动化中，则可编排“检测异常 → 执行诊断脚本 → 通知管理员 → 等待审批 → 执行修复”的长周期任务。LangGraph 的条件边与节点机制天然支持这类复杂逻辑的建模。

进一步扩展时，以下几个方向值得探索：

集成检索增强生成（RAG）：在 query_or_respond 前插入一个检索节点，从知识库中提取相关信息注入上下文，提升回答准确性。
多工具协同与依赖管理：利用图结构的优势，设计并行或串行的工具调用链，例如先查天气再推荐穿搭。
引入人工干预节点：在关键决策点加入“等待人工确认”环节，实现人机协同流程，适用于风控、审批等高敏感场景。

此外，LangGraph 提供了 .get_graph().draw_mermaid() 方法，可自动生成流程图：

python
from IPython.display import display, Markdown
display(Markdown(graph.get_graph().draw_mermaid()))

该功能不仅能帮助开发者快速理解执行路径，还可嵌入文档或会议材料中，成为团队协作的可视化语言。