LangGraph

LangGraph 受到众多引领智能体未来发展的企业信赖，包括 Klarna、Replit、Elastic 等。它是一个底层编排框架，可用于构建、管理和部署长期运行的有状态智能体。

一、快速开始

安装 LangGraph

pip install -U langgraph

使用预构建组件创建智能体

API 参考文档：create_react_agent（React 智能体创建函数）

# 如需调用该模型，请先执行：pip install -qU "langchain[anthropic]"
from langgraph.prebuilt import create_react_agent

def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"It's always sunny in {city}!"  # 译文：{city}（城市名）的天气总是晴朗的！

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",  # Anthropic 公司的 Claude-3-7-sonnet-latest 模型
    tools=[get_weather],  # 工具列表，此处包含“获取天气”工具
    prompt="You are a helpful assistant"  # 提示词：你是一个乐于助人的助手
)

# 运行智能体
agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}  # 译文：{"messages": [{"role": "user", "content": "旧金山的天气怎么样"}]}（注：“sf”为“San Francisco”缩写，即旧金山）
)

二、核心优势

LangGraph 为所有长期运行的有状态工作流或智能体提供底层支持架构。它不抽象提示词或架构，主要具有以下核心优势：

• 持久化执行：构建能够在故障中存续、可长时间运行的智能体，智能体可从暂停的确切位置自动恢复运行。
• 人机协同（Human-in-the-loop）：在执行过程中的任意节点，都能通过检查和修改智能体状态，无缝融入人工监督环节。
• 全面的记忆功能：打造真正的有状态智能体，既具备用于持续推理的短期工作记忆，也拥有跨会话的长期持久记忆。
• 借助 LangSmith 调试：通过可视化工具深入洞察复杂智能体的行为，这些工具可追踪执行路径、捕捉状态转换，并提供详细的运行时指标。
• 可投入生产的部署：依托为应对有状态、长期运行工作流独特挑战而设计的可扩展架构，放心部署复杂的智能体系统。

三、LangGraph 生态系统

LangGraph 可单独使用，同时也能与任何 LangChain 产品无缝集成，为开发者提供构建智能体所需的全套工具。若要提升大语言模型（LLM）应用开发效率，可将 LangGraph 与以下工具搭配使用：

• LangSmith：适用于智能体评估和可观测性。可调试性能不佳的大语言模型应用运行过程、评估智能体执行轨迹、掌握生产环境中的运行情况，并逐步提升性能。
• LangSmith Deployment（LangSmith 部署工具）：借助专为长期运行的有状态工作流设计的部署平台，轻松部署和扩展智能体。支持跨团队发现、复用、配置和共享智能体，并能通过 LangGraph Studio（LangGraph 工作室）的可视化原型设计快速迭代。
• LangChain：提供集成功能和可组合组件，简化大语言模型应用的开发流程。

注：正在寻找 LangGraph 的 JavaScript 版本？可查看 JS 代码仓库（JS repo）和 JS 文档（JS docs）。

五、鸣谢

LangGraph 的设计灵感源自 Pregel（一种图计算框架）和 Apache Beam（一种数据处理框架）。其公共接口的设计借鉴了 NetworkX（一种Python图论库）的理念。LangGraph 由 LangChain 公司（LangChain 框架的创建者）开发，但无需依赖 LangChain 即可独立使用。

LangGraph 智能体快速入门指南总结

该指南围绕LangGraph预构建可复用组件，提供了智能体搭建的完整流程，从环境准备到功能配置逐步展开，帮助开发者快速构建可靠的智能体系统，核心内容可分为前提条件、6个核心操作步骤两大部分，具体如下：

一、前提条件

开始前需准备Anthropic API密钥，用于调用Anthropic系列模型（如Claude-3-7-sonnet-latest）。

二、核心操作步骤

1. 安装依赖

通过pip命令安装LangGraph及LangChain的Anthropic扩展，确保智能体可调用对应模型：

pip install -U langgraph "langchain[anthropic]"

2. 创建基础智能体

使用create_react_agent函数创建智能体，需指定模型、工具（示例中为获取天气的get_weather函数）和基础提示词，同时支持直接调用智能体响应用户请求：

from langgraph.prebuilt import create_react_agent

# 定义天气查询工具
def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

# 初始化智能体
agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
    prompt="You are a helpful assistant"
)

# 运行智能体（查询旧金山天气）
agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)

3. 配置LLM参数

借助init_chat_model函数为大语言模型设置特定参数（如temperature，控制输出随机性，0表示确定性输出），再将配置好的模型传入智能体：

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import create_react_agent

# 配置模型参数
model = init_chat_model(
    "anthropic:claude-3-7-sonnet-latest",
    temperature=0
)

# 用配置好的模型创建智能体
agent = create_react_agent(
    model=model,
    tools=[get_weather]
)

4. 添加自定义提示词

支持两种提示词类型，满足不同场景需求：

• 静态提示词：固定不变的字符串或消息列表，直接定义智能体行为（如禁止回答天气问题）：

  agent = create_react_agent(
      model="anthropic:claude-3-7-sonnet-latest",
      tools=[get_weather],
      prompt="Never answer questions about the weather."  # 静态提示词
  )
  

• 动态提示词：通过函数生成，可结合智能体状态（state）和运行时配置（config）动态调整内容（如根据user_name称呼用户）：

  from langchain_core.messages import AnyMessage
  from langchain_core.runnables import RunnableConfig
  from langgraph.prebuilt.chat_agent_executor import AgentState
  
  # 定义动态提示词函数
  def prompt(state: AgentState, config: RunnableConfig) -> list[AnyMessage]:  
      user_name = config["configurable"].get("user_name")
      system_msg = f"You are a helpful assistant. Address the user as {user_name}."
      return [{"role": "system", "content": system_msg}] + state["messages"]
  
  # 用动态提示词创建智能体
  agent = create_react_agent(
      model="anthropic:claude-3-7-sonnet-latest",
      tools=[get_weather],
      prompt=prompt
  )
  
  # 调用时传入用户名称配置
  agent.invoke(
      {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
      config={"configurable": {"user_name": "John Smith"}}
  )
  

动态提示词可融入运行时信息（如user_id、API凭证）和多步推理中的智能体内部状态，灵活性更高。

5. 添加记忆功能（支持多轮对话）

通过提供checkpointer（检查点）实现状态持久化，需指定thread_id（对话唯一标识），确保多轮调用时智能体自动保留历史对话记录：

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver

# 初始化内存检查点（用于存储对话状态）
checkpointer = InMemorySaver()

# 创建带记忆功能的智能体
agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
    checkpointer=checkpointer
)

# 第一轮对话（查询旧金山天气）
config = {"configurable": {"thread_id": "1"}}  # 同一对话的唯一标识
sf_response = agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    config
)

# 第二轮对话（查询纽约天气，自动关联历史记录）
ny_response = agent.invoke(
    {"messages": [{"role": "user", "content": "what about new york?"}]},
    config
)

检查点会存储智能体每一步的状态，支持从内存（InMemorySaver）或数据库中读取历史信息。

6. 配置结构化输出

通过response_format参数指定输出 schema（支持Pydantic模型或TypedDict），使智能体输出符合固定格式，结果可从structured_response字段获取，需注意该功能需额外调用LLM进行格式处理：

from pydantic import BaseModel
from langgraph.prebuilt import create_react_agent

# 定义输出 schema（天气响应包含"conditions"字段）
class WeatherResponse(BaseModel):
    conditions: str

# 创建带结构化输出的智能体
agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
    response_format=WeatherResponse
)

# 调用智能体，获取结构化结果
response = agent.invoke(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)
print(response["structured_response"])  # 输出符合WeatherResponse格式的结果

LangGraph 核心价值与基础学习指南总结

该网页聚焦LangGraph的核心优势及入门学习路径，清晰阐述了开发者选择LangGraph的核心原因，并提供了系统化的基础教程体系，助力开发者掌握其关键能力，具体内容如下：

一、开发者选择LangGraph的核心原因

LangGraph专为构建强大、可适配的AI智能体设计，核心吸引力集中在三大维度，满足复杂场景下的开发需求：

1. 高可靠性与可控性：支持通过审核检查和“人机协同审批”（human-in-the-loop approvals）引导智能体行为，避免失控；同时能为长期运行的工作流持久化存储上下文信息，确保智能体始终围绕目标推进，不偏离任务方向。
2. 底层可扩展性：提供完全可描述的底层基础组件，无限制定制的僵化抽象，便于开发者构建自定义智能体；支持设计可扩展的多智能体系统，可根据具体业务场景为每个智能体分配专属角色，实现分工协作。
3. 原生流式支持：具备“逐token流式输出”和“中间步骤流式展示”能力，让用户能实时、清晰地看到智能体的推理过程与行动轨迹，提升过程透明度，便于调试和用户理解。

二、LangGraph基础学习路径

为帮助开发者掌握LangGraph的关键概念与功能，网页提供了一套系统化的基础教程系列，共6个核心模块，按能力递进顺序排列：

1. 构建基础聊天机器人（Build a basic chatbot）
2. 为智能体添加工具（Add tools）
3. 为智能体添加记忆功能（Add memory）
4. 添加人机协同控制（Add human-in-the-loop controls）
5. 自定义智能体状态（Customize state）
6. 实现“时间旅行”（Time travel，即回溯并探索对话的其他可能路径）

通过完成这套教程，开发者可构建出一个具备完整功能的客服聊天机器人，该机器人将拥有以下核心能力：

• ✅ 支持通过网页搜索回答常见问题
• ✅ 跨调用会话保持对话状态（即记忆历史交互内容）
• ✅ 可将复杂查询路由给人工审核处理
• ✅ 利用自定义状态控制自身行为逻辑
• ✅ 支持回溯对话过程，并探索不同的对话路径

构建LangGraph基础聊天机器人教程

该教程详细介绍了使用LangGraph搭建基础聊天机器人的完整流程，涵盖环境准备、核心组件配置、运行与可视化等关键步骤，同时为后续功能扩展（如添加工具）铺垫基础，具体内容如下：

一、前提条件

需准备支持工具调用功能的大语言模型（LLM），可选类型包括OpenAI（如GPT-4.1）、Anthropic（如Claude-3.5-Sonnet）、Google Gemini、Azure OpenAI、AWS Bedrock等。

二、核心操作步骤

1. 安装依赖包

通过pip命令安装LangGraph核心框架及LangSmith（用于后续追踪与调试）：

pip install -U langgraph langsmith

2. 创建StateGraph（状态图）

StateGraph是LangGraph的核心组件，用于定义聊天机器人的“状态机”结构，包含状态（State）、节点（Nodes）和边（Edges）三大要素：

• 定义State（状态）：通过TypedDict定义状态 schema，核心字段为messages（存储对话消息），并使用add_messages注解实现消息“追加”逻辑（而非覆盖）：

  from typing import Annotated
  from typing_extensions import TypedDict
  from langgraph.graph import StateGraph, START, END
  from langgraph.graph.message import add_messages
  
  class State(TypedDict):
      # 用add_messages确保新消息追加到现有列表
      messages: Annotated[list, add_messages]
  

• 初始化StateGraph：基于定义的State创建状态图构建器：

  graph_builder = StateGraph(State)
  

3. 添加“chatbot”节点（核心功能单元）

节点代表具体工作单元（通常为函数），此处“chatbot”节点负责调用LLM生成对话响应，需先配置目标LLM，再定义节点逻辑：

（1）配置LLM（按模型类型选择对应代码）

模型类型	安装命令与配置代码
OpenAI	安装：pip install -U "langchain[openai]" 配置： import os os.environ["OPENAI_API_KEY"] = "sk-..." llm = init_chat_model("openai:gpt-4.1")
Anthropic	安装：pip install -U "langchain[anthropic]" 配置： os.environ["ANTHROPIC_API_KEY"] = "sk-..." llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
Azure OpenAI	安装：pip install -U "langchain[openai]" 配置： os.environ["AZURE_OPENAI_API_KEY"] = "..." os.environ["AZURE_OPENAI_ENDPOINT"] = "..." llm = init_chat_model("azure_openai:gpt-4.1", azure_deployment=...)
Google Gemini	安装：pip install -U "langchain[google-genai]" 配置： os.environ["GOOGLE_API_KEY"] = "..." llm = init_chat_model("google_genai:gemini-2.0-flash")
AWS Bedrock	安装：pip install -U "langchain[aws]" 配置： llm = init_chat_model("anthropic.claude-3-5-sonnet-20240620-v1:0", model_provider="bedrock_converse")（需先配置AWS凭证）

（2）定义节点函数并添加到状态图

节点函数接收当前State作为输入，调用LLM生成响应后，返回更新后的messages状态：

def chatbot(state: State):
    # 调用LLM处理现有消息，生成新响应
    return {"messages": [llm.invoke(state["messages"])]}

# 向状态图添加节点（第一个参数为节点唯一名称，第二个为节点函数）
graph_builder.add_node("chatbot", chatbot)

4. 配置入口（START）与出口（END）

• 入口：指定状态图的起始节点，即从START直接进入“chatbot”节点：

  graph_builder.add_edge(START, "chatbot")
  

• 出口：指定状态图的结束节点，即“chatbot”节点执行完成后，流程终止于END：

  graph_builder.add_edge("chatbot", END)
  

5. 编译状态图

将构建好的状态图编译为可执行的CompiledGraph对象：

graph = graph_builder.compile()

6. 可视化状态图（可选）

通过draw_mermaid_png等方法生成状态图可视化结果（需额外依赖），便于直观查看流程结构：

from IPython.display import Image, display
try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # 依赖缺失时跳过，不影响核心功能
    pass

可视化结果为简单线性流程：START → chatbot → END。

7. 运行聊天机器人

实现交互式对话逻辑，支持用户输入并流式输出AI响应，同时提供退出机制（输入quit/exit/q退出）：

def stream_graph_updates(user_input: str):
    # 流式处理对话，输出AI响应
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

# 交互式循环
while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:
        # 输入功能不可用时的 fallback 逻辑
        user_input = "What do you know about LangGraph?"
        print("User:", user_input)
        stream_graph_updates(user_input)
        break

三、完整代码汇总

from typing import Annotated
from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 初始化状态图
graph_builder = StateGraph(State)

# 3. 配置LLM（以Anthropic为例，可替换为其他模型）
import os
os.environ["ANTHROPIC_API_KEY"] = "sk-..."
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")

# 4. 定义并添加chatbot节点
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)

# 5. 配置入口与出口
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 6. 编译状态图
graph = graph_builder.compile()

# 7. 运行聊天机器人
def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:
        user_input = "What do you know about LangGraph?"
        print("User:", user_input)
        stream_graph_updates(user_input)
        break

四、后续步骤

当前聊天机器人仅依赖LLM训练数据响应，知识存在局限性，下一阶段将通过“添加网页搜索工具”扩展其知识范围，提升实用性。

为LangGraph聊天机器人添加工具

该教程基于“基础聊天机器人”进一步扩展功能，核心是为机器人集成Tavily网页搜索工具，使其能处理超出LLM训练数据范围的查询（如实时信息、特定领域新知），同时引入条件边（conditional edges）实现“调用工具-处理结果-生成响应”的循环逻辑，具体内容如下：

一、前提条件

需准备Tavily搜索引擎的API密钥（用于调用网页搜索功能），同时确保已完成上一教程“构建基础聊天机器人”的环境配置。

二、核心操作步骤

1. 安装搜索工具依赖

通过pip安装LangChain与Tavily的集成包，支持调用搜索功能：

pip install -U langchain-tavily

2. 配置环境变量（添加Tavily API密钥）

将Tavily API密钥注入环境变量，确保工具能正常授权调用：

import os
os.environ["TAVILY_API_KEY"] = "tvly-..."  # 替换为实际API密钥

3. 定义网页搜索工具

初始化Tavily搜索工具，设置返回结果数量（示例中为2条），并可测试工具调用效果：

from langchain_tavily import TavilySearch

# 定义搜索工具，最多返回2条结果
tool = TavilySearch(max_results=2)
tools = [tool]

# 测试工具：查询“LangGraph中的node是什么”
tool.invoke("What's a 'node' in LangGraph?")

测试返回结果包含搜索query、相关网页标题、URL、内容摘要、相关性评分等信息，供机器人后续生成响应使用。

4. 配置带工具调用能力的LLM

在基础聊天机器人的StateGraph基础上，为LLM绑定工具（bind_tools），使其知晓工具调用的JSON格式，能判断何时需要调用搜索工具：

（1）选择并初始化LLM（与上一教程一致，支持多模型）

模型类型	安装命令与配置代码（示例，同前序教程）
OpenAI	安装：pip install -U "langchain[openai]" 配置：os.environ["OPENAI_API_KEY"] = "sk-..."；llm = init_chat_model("openai:gpt-4.1")
Anthropic	安装：pip install -U "langchain[anthropic]" 配置：os.environ["ANTHROPIC_API_KEY"] = "sk-..."；llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
Azure/Google Gemini/AWS Bedrock	安装对应依赖后，配置专属API密钥/端点，调用init_chat_model初始化

（2）绑定工具到LLM并定义chatbot节点

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态（与基础版一致，仅存储messages）
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 初始化状态图
graph_builder = StateGraph(State)

# 3. 为LLM绑定工具，使其支持工具调用
llm_with_tools = llm.bind_tools(tools)

# 4. 定义chatbot节点：调用带工具能力的LLM生成响应
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 添加chatbot节点到状态图
graph_builder.add_node("chatbot", chatbot)

5. 创建工具执行节点（BasicToolNode）

定义专门的“工具节点”，负责检查chatbot节点输出的消息中是否包含tool_calls（工具调用指令），若有则执行对应工具并返回结果：

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
    """执行LLM请求的工具调用，返回工具结果"""
    def __init__(self, tools: list) -> None:
        # 按工具名称建立映射，便于快速查找
        self.tools_by_name = {tool.name: tool for tool in tools}
    
    def __call__(self, inputs: dict):
        # 获取最新消息
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("输入中未找到消息")
        
        # 执行所有工具调用请求
        outputs = []
        for tool_call in message.tool_calls:
            # 调用对应工具
            tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])
            # 将结果封装为ToolMessage，便于chatbot节点后续处理
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),  # 工具结果转为JSON字符串
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"]
                )
            )
        return {"messages": outputs}

# 初始化工具节点
tool_node = BasicToolNode(tools=[tool])
# 添加工具节点到状态图
graph_builder.add_node("tools", tool_node)

6. 定义条件边（实现流程分支逻辑）

添加“条件边”（conditional edges），让状态图根据chatbot节点的输出动态决定下一步：若需调用工具则进入“tools”节点，否则直接结束流程（END）。

（1）定义路由函数（route_tools）

def route_tools(state: State):
    """
    路由逻辑：检查最新消息是否包含工具调用，是则去tools节点，否则结束
    """
    # 获取最新消息
    if messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError("状态中未找到消息")
    
    # 判断是否有工具调用
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"  # 有工具调用，进入tools节点
    return END  # 无工具调用，流程结束

（2）添加条件边到状态图

# 为chatbot节点添加条件边：根据route_tools结果路由
graph_builder.add_conditional_edges(
    "chatbot",  # 起点节点
    route_tools,  # 路由函数
    {"tools": "tools", END: END}  # 路由结果与目标节点的映射
)

# 工具执行完成后，返回chatbot节点，让LLM基于工具结果生成最终响应
graph_builder.add_edge("tools", "chatbot")

# 流程起点：从START进入chatbot节点
graph_builder.add_edge(START, "chatbot")

# 编译状态图
graph = graph_builder.compile()

7. 可视化状态图（可选）

通过Mermaid生成可视化图表，可直观看到“循环流程”：START → chatbot → （有工具调用→tools→chatbot；无→END）

from IPython.display import Image, display
try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # 依赖缺失时跳过，不影响核心功能
    pass

8. 运行带搜索功能的聊天机器人

沿用交互式运行逻辑，支持用户输入查询，机器人会自动判断是否调用搜索工具：

def stream_graph_updates(user_input: str):
    # 流式输出：先显示工具调用过程，再显示最终响应
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

# 交互式循环（输入quit/exit/q退出）
while True:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:
        # 输入功能不可用时的 fallback
        user_input = "What do you know about LangGraph?"
        print("User:", user_input)
        stream_graph_updates(user_input)
        break

运行示例：用户询问“LangGraph是什么”时，机器人会先调用Tavily搜索，获取最新信息后整理成结构化响应（如LangGraph的用途、开发者、核心特性等）。

9. 使用预构建组件简化代码（可选）

LangGraph提供预构建的ToolNode和tools_condition，可替代自定义的BasicToolNode和route_tools，减少重复代码：

from langgraph.prebuilt import ToolNode, tools_condition

# 1. 替换自定义BasicToolNode为预构建ToolNode
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 2. 替换自定义route_tools为预构建tools_condition
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition  # 预构建路由逻辑，功能与route_tools一致
)

# 其余配置（添加边、编译）不变
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

为LangGraph聊天机器人添加记忆功能教程总结

该教程基于“添加工具的聊天机器人”进一步升级，核心是通过LangGraph的持久化检查点（checkpointing）机制为机器人实现对话记忆，使其能跨多轮交互保留上下文，解决此前“无法记住历史对话”的问题，具体内容如下：

一、核心原理：检查点（Checkpointing）的作用

LangGraph的记忆功能依赖“检查点”实现：

• 编译状态图时传入checkpointer（检查点存储工具），状态图会在每一步执行后自动保存当前State（如对话消息列表）。
• 调用机器人时指定唯一thread_id（对话标识），相同thread_id的后续调用会自动加载历史状态，实现“续接对话”。
• 检查点的能力远超简单聊天记忆，还可用于错误恢复、人机协同流程、对话回溯（时间旅行）等复杂场景。

二、前提条件

需已完成上一教程“添加工具”的环境配置（如Tavily API密钥、LLM配置、工具节点与条件边定义），本教程在此基础上新增记忆功能。

三、核心操作步骤

1. 创建检查点存储工具（MemorySaver）

使用InMemorySaver（内存检查点）存储对话状态，适合教程演示；生产环境需替换为SqliteSaver（SQLite数据库）或PostgresSaver（PostgreSQL数据库）以实现持久化：

from langgraph.checkpoint.memory import InMemorySaver

# 初始化内存检查点（仅用于演示，重启后数据丢失）
memory = InMemorySaver()

2. 编译带检查点的状态图

编译时将checkpointer参数设为创建的memory，让状态图自动保存/加载对话状态：

# 对比上一教程：新增checkpointer=memory参数
graph = graph_builder.compile(checkpointer=memory)

3. 多轮交互测试：验证记忆功能

通过指定thread_id，测试机器人是否能记住历史对话：

（1）第一轮对话：告知机器人姓名

# 配置：指定对话标识thread_id="1"（同一对话需保持一致）
config = {"configurable": {"thread_id": "1"}}

# 发送用户输入：告知姓名
user_input = "Hi there! My name is Will."
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,  # 关键：第二个位置参数传入config，而非嵌套在inputs中
    stream_mode="values",  # 流式输出结果
)

# 打印响应
for event in events:
    event["messages"][-1].pretty_print()

响应结果：机器人会礼貌回应并称呼“Will”，状态图自动保存包含“用户姓名”的对话消息。

（2）第二轮对话：验证是否记住姓名

# 相同config（thread_id="1"），发送跟进问题
user_input = "Remember my name?"
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)

for event in events:
    event["messages"][-1].pretty_print()

响应结果：机器人会明确表示“记得你的名字是Will”，证明成功加载历史状态。

（3）对比测试：不同thread_id无记忆

更换thread_id为“2”，模拟新对话，机器人无法获取历史状态：

# 不同config：thread_id="2"（新对话）
events = graph.stream(
    {"messages": [{"role": "user", "content": "Remember my name?"}]},
    {"configurable": {"thread_id": "2"}},  # 新对话标识
    stream_mode="values",
)

for event in events:
    event["messages"][-1].pretty_print()

响应结果：机器人会回复“没有过往上下文，不知道你的名字”，验证thread_id是关联对话记忆的关键。

4. 查看当前状态（可选）

通过get_state(config)方法可查看指定thread_id的完整对话状态，包括历史消息、配置信息、下一步执行节点等：

# 获取thread_id="1"的状态快照
snapshot = graph.get_state(config)

# 打印快照内容（包含所有历史消息、配置、元数据等）
print(snapshot)

# 查看下一步执行节点（当前对话已结束，next为空）
print(snapshot.next)  # 输出：()

快照中values["messages"]字段会列出所有历史交互消息（用户消息+AI响应），next字段表示下一步需执行的节点（对话结束时为空）。

四、完整代码汇总（以Anthropic为例）

# 1. 安装依赖（若未安装）
# pip install -U langchain[anthropic] langchain-tavily langgraph

# 2. 导入模块
from typing import Annotated
from typing_extensions import TypedDict
from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
import os

# 3. 配置环境变量
os.environ["ANTHROPIC_API_KEY"] = "sk-..."  # 替换为实际API密钥
os.environ["TAVILY_API_KEY"] = "tvly-..."    # 替换为实际API密钥

# 4. 初始化LLM与工具
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
tool = TavilySearch(max_results=2)
tools = [tool]
llm_with_tools = llm.bind_tools(tools)

# 5. 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]  # 消息追加逻辑

# 6. 构建状态图
graph_builder = StateGraph(State)

# 添加chatbot节点
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)

# 添加工具节点
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 添加条件边（工具调用判断）
graph_builder.add_conditional_edges("chatbot", tools_condition)

# 工具执行后返回chatbot节点
graph_builder.add_edge("tools", "chatbot")

# 设置入口节点
graph_builder.set_entry_point("chatbot")

# 7. 初始化检查点并编译状态图
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# 8. 多轮交互测试
config = {"configurable": {"thread_id": "1"}}

# 第一轮：告知姓名
user_input1 = "Hi there! My name is Will."
events1 = graph.stream({"messages": [{"role": "user", "content": user_input1}]}, config, stream_mode="values")
for event in events1:
    event["messages"][-1].pretty_print()

# 第二轮：验证记忆
user_input2 = "Remember my name?"
events2 = graph.stream({"messages": [{"role": "user", "content": user_input2}]}, config, stream_mode="values")
for event in events2:
    event["messages"][-1].pretty_print()

五、核心改进与后续方向

1. 本次教程核心价值

• 实现自然多轮对话：机器人可记住用户过往输入（如姓名、偏好、前文问题），避免重复询问，交互更连贯。
• 灵活的状态管理：检查点机制支持跨会话保存状态（如用户退出后重新进入仍可续接对话），且可扩展至复杂场景。

2. 后续步骤

下一教程将为机器人添加“人机协同（Human-in-the-loop）”功能，让机器人在遇到复杂/不确定场景时（如敏感操作、模糊查询），可请求人工指导或验证后再继续执行。

为LangGraph聊天机器人自定义状态教程总结

该教程基于“添加人机协同控制”的聊天机器人进一步升级，核心是通过扩展状态（State）的自定义字段，实现更复杂的 workflow 管理（如存储特定任务数据并结合人工审核修正），突破此前仅依赖“messages”列表存储状态的局限，具体内容如下：

一、核心目标与原理

此前机器人的状态仅包含messages（对话消息列表），本次教程通过在State中新增name（实体名称）和birthday（实体发布/创建日期）字段，让机器人能：

1. 存储特定任务的关键数据（如查询“LangGraph发布日期”时的实体名与日期）；
2. 结合human_assistance工具实现“机器查询→人工审核修正→状态更新”的闭环；
3. 支持手动或通过工具自动更新自定义状态字段，且数据可被下游节点（如存储、分析节点）直接调用。

二、前提条件

需已完成上一教程“添加人机协同控制”的环境配置（如Tavily搜索工具、human_assistance工具、检查点记忆功能），本教程在此基础上扩展状态结构与逻辑。

三、核心操作步骤

1. 扩展State：新增自定义字段

通过TypedDict定义State时，除原有messages字段外，新增name（字符串类型，存储实体名）和birthday（字符串类型，存储实体日期）字段，并用add_messages注解保持消息追加逻辑：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 原有：对话消息列表（追加更新）
    name: str  # 新增：实体名称（如“LangGraph”）
    birthday: str  # 新增：实体日期（如发布日期“Jan 17, 2024”）

自定义字段的优势在于：数据结构化存储，可被所有节点直接访问，无需从messages中解析信息。

2. 升级human_assistance工具：支持状态更新

修改human_assistance工具，使其在人工审核后不仅返回结果，还能通过Command对象自动更新State中的name和birthday字段，核心逻辑包括：

• 用interrupt中断流程，向人工展示待审核的name和birthday；
• 根据人工反馈（“正确”或“修正”）确定最终的verified_name和verified_birthday；
• 构建state_update字典，包含更新后的字段与工具消息，通过Command(update=state_update)返回，触发状态自动更新。

具体代码如下：

from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from langgraph.types import Command, interrupt

@tool  # 标记为工具，供LLM调用
def human_assistance(
    name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]
) -> str:
    """请求人工协助审核实体名称与日期"""
    # 1. 中断流程，向人工展示待审核内容
    human_response = interrupt(
        {
            "question": "Is this correct?",  # 人工审核问题
            "name": name,  # 待审核的实体名
            "birthday": birthday  # 待审核的实体日期
        }
    )
    
    # 2. 处理人工反馈：确定最终数据
    if human_response.get("correct", "").lower().startswith("y"):
        # 人工确认正确：沿用原数据
        verified_name = name
        verified_birthday = birthday
        response = "Correct"
    else:
        # 人工修正：使用人工提供的新数据（无则沿用原数据）
        verified_name = human_response.get("name", name)
        verified_birthday = human_response.get("birthday", birthday)
        response = f"Made a correction: {human_response}"
    
    # 3. 构建状态更新内容：包含自定义字段与工具消息
    state_update = {
        "name": verified_name,  # 更新State的name字段
        "birthday": verified_birthday,  # 更新State的birthday字段
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)]  # 新增工具消息到messages
    }
    
    # 4. 返回Command对象，触发State自动更新
    return Command(update=state_update)

3. 配置工具与LLM

将升级后的human_assistance工具与Tavily搜索工具整合，绑定到LLM，确保机器人可调用这两个工具：

from langchain_tavily import TavilySearch

# 1. 初始化工具列表（含Tavily搜索和人工协助工具）
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]

# 2. 为LLM绑定工具，使其能判断何时调用搜索或人工协助
llm_with_tools = llm.bind_tools(tools)

# 3. 定义chatbot节点：调用带工具的LLM生成响应（限制单次最多1个工具调用）
def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    assert len(message.tool_calls) <= 1  # 确保单次仅调用1个工具
    return {"messages": [message]}

4. 构建并编译状态图

沿用此前的“chatbot→工具节点→条件边”结构，新增检查点以保留自定义状态字段的历史数据：

from langgraph.graph import StateGraph, START
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver

# 1. 初始化状态图
graph_builder = StateGraph(State)

# 2. 添加节点：chatbot节点（生成响应/工具调用）、tools节点（执行工具）
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)

# 3. 添加条件边：chatbot节点根据工具调用判断路由（去tools节点或结束）
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition  # 预构建路由逻辑：有工具调用则去tools节点
)

# 4. 工具执行后返回chatbot节点，继续处理结果
graph_builder.add_edge("tools", "chatbot")

# 5. 设置流程入口：从START进入chatbot节点
graph_builder.add_edge(START, "chatbot")

# 6. 初始化检查点，编译状态图（保留自定义状态）
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)

5. 测试：查询LangGraph发布日期（含人工修正）

通过多轮交互测试“搜索→人工审核→状态更新”流程：

（1）第一轮：用户请求查询并人工审核

# 配置对话标识（thread_id="1"，确保记忆连贯）
config = {"configurable": {"thread_id": "1"}}

# 用户输入：查询LangGraph发布日期，并要求人工审核
user_input = (
    "Can you look up when LangGraph was released? "
    "When you have the answer, use the human_assistance tool for review."
)

# 流式调用机器人
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values"
)

# 打印响应
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

流程与响应：

1. 机器人调用Tavily搜索“LangGraph release date”，但搜索结果仅提到2024年6月27日的“LangGraph Platform发布”，未明确原LangGraph库的发布日期；
2. 机器人调用human_assistance工具，提交待审核数据（name="Assistant"，birthday="2023-01-01"），触发人工中断。

（2）第二轮：人工修正并恢复流程

人工发现机器人提交的数据错误，通过Command对象传入正确信息（name="LangGraph"，birthday="Jan 17, 2024"），恢复流程：

# 人工发送修正指令：正确的实体名与日期
human_command = Command(
    resume={
        "name": "LangGraph",
        "birthday": "Jan 17, 2024",
        "correct": "n"  # 标记原数据不正确
    }
)

# 继续流式执行，应用人工修正
events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

流程与响应：

1. human_assistance工具接收人工修正，更新State的name和birthday字段；
2. 机器人生成最终响应，明确LangGraph的发布日期为2024年1月17日，并区分“LangGraph库”与“2024年6月27日发布的LangGraph Platform”。

（3）验证状态更新

通过get_state查看State中的自定义字段，确认已更新为人工修正后的数据：

# 获取当前对话状态
snapshot = graph.get_state(config)

# 提取并打印自定义字段
print({k: v for k, v in snapshot.values.items() if k in ("name", "birthday")})
# 输出：{'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}

6. 手动更新状态（可选）

LangGraph支持在任意阶段（包括流程中断时）通过update_state手动修改自定义状态字段，适合紧急修正或特殊场景：

# 手动更新State的name字段（添加“(library)”标识）
graph.update_state(config, {"name": "LangGraph (library)"})

# 验证更新结果
snapshot = graph.get_state(config)
print({k: v for k, v in snapshot.values.items() if k in ("name", "birthday")})
# 输出：{'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}

手动更新会在LangSmith中生成追踪记录，但推荐优先使用interrupt实现人机协同，避免状态更新与交互流程脱节。

四、完整代码汇总（以Anthropic为例）

# 1. 安装依赖（若未安装）
# pip install -U langchain[anthropic] langchain-tavily langgraph

# 2. 导入模块
from typing import Annotated
from langchain.chat_models import init_chat_model
from langchain_tavily import TavilySearch
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import Command, interrupt
import os

# 3. 配置环境变量
os.environ["ANTHROPIC_API_KEY"] = "sk-..."  # 替换为实际API密钥
os.environ["TAVILY_API_KEY"] = "tvly-..."    # 替换为实际API密钥

# 4. 初始化LLM
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")

# 5. 定义自定义State
class State(TypedDict):
    messages: Annotated[list, add_messages]
    name: str
    birthday: str

# 6. 定义升级后的human_assistance工具
@tool
def human_assistance(
    name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]
) -> str:
    human_response = interrupt(
        {"question": "Is this correct?", "name": name, "birthday": birthday}
    )
    if human_response.get("correct", "").lower().startswith("y"):
        verified_name, verified_birthday, response = name, birthday, "Correct"
    else:
        verified_name = human_response.get("name", name)
        verified_birthday = human_response.get("birthday", birthday)
        response = f"Made a correction: {human_response}"
    state_update = {
        "name": verified_name,
        "birthday": verified_birthday,
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)]
    }
    return Command(update=state_update)

# 7. 配置工具与LLM
tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    message = llm_with_tools.invoke(state["messages"])
    assert len(message.tool_calls) <= 1
    return {"messages": [message]}

# 8. 构建并编译状态图
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)

# 9. 测试流程
config = {"configurable": {"thread_id": "1"}}
# 第一轮：用户查询
user_input = "Can you look up when LangGraph was released? When you have the answer, use the human_assistance tool for review."
events1 = graph.stream({"messages": [{"role": "user", "content": user_input}]}, config, stream_mode="values")
for event in events1:
    if "messages" in event:
        event["messages"][-1].pretty_print()

# 第二轮：人工修正
human_command = Command(resume={"name": "LangGraph", "birthday": "Jan 17, 2024", "correct": "n"})
events2 = graph.stream(human_command, config, stream_mode="values")
for event in events2:
    if "messages" in event:
        event["messages"][-1].pretty_print()

# 验证状态
snapshot = graph.get_state(config)
print({k: v for k, v in snapshot.values.items() if k in ("name", "birthday")})

五、核心价值与总结

1. 状态结构化扩展：通过自定义字段（如name、birthday），机器人可存储任务专属数据，避免信息分散在messages中难以解析，为复杂 workflow 提供数据支撑；
2. 工具与状态联动：human_assistance工具通过Command实现“人工反馈→状态自动更新”，无需额外节点处理，简化流程逻辑；
3. 灵活的状态控制：支持工具自动更新与手动更新，适配不同场景需求，且状态数据可被下游节点直接复用（如存储到数据库、生成报告）。

该教程为后续构建更复杂的智能体（如多节点数据处理、跨任务状态共享）奠定了基础。

LangGraph聊天机器人“时间旅行”功能教程总结

该教程基于“自定义状态”的聊天机器人进一步拓展，核心是利用LangGraph的检查点历史遍历能力实现“时间旅行”——支持回溯到对话的任意历史节点，重新执行或探索不同流程分支，适用于调试错误、尝试替代策略（如自主软件工程师类应用）等场景，具体内容如下：

一、核心目标与原理

“时间旅行”的本质是基于LangGraph的检查点历史记录实现：

1. 此前教程中，检查点（checkpointer）会在状态图每一步执行后保存快照（含当前State、下一步节点next、配置config等）；
2. 通过get_state_history方法可获取某一对话（thread_id）的所有检查点快照，选择任意快照即可“回溯”到对应时间点；
3. 基于历史快照的config（含checkpoint_id）重新调用状态图，可从该节点继续执行，实现“重放”或“分支探索”。

二、前提条件

需已完成上一教程“自定义状态”的环境配置（如Tavily搜索工具、检查点记忆、自定义State结构），本教程复用此前的状态图结构（chatbot节点、tools节点、条件边），聚焦检查点历史的操作。

三、核心操作步骤

1. 初始化基础状态图与检查点

沿用含工具调用、记忆功能的状态图，确保检查点已启用（InMemorySaver），代码与“自定义状态”教程一致，核心结构如下：

from typing import Annotated
from langchain_tavily import TavilySearch
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# 1. 定义State（仅含messages，也可扩展自定义字段）
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 初始化工具与LLM（以Anthropic为例，需配置API密钥）
tool = TavilySearch(max_results=2)
tools = [tool]
llm_with_tools = llm.bind_tools(tools)  # llm需提前通过init_chat_model初始化

# 3. 定义chatbot节点
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 4. 构建状态图
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", ToolNode(tools=tools))
graph_builder.add_conditional_edges("chatbot", tools_condition)  # 工具调用判断
graph_builder.add_edge("tools", "chatbot")  # 工具执行后返回chatbot
graph_builder.add_edge(START, "chatbot")

# 5. 启用检查点，编译状态图
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)

2. 生成对话历史（创建检查点快照）

通过多轮用户交互，让状态图生成多个检查点（每一步执行后自动保存），为后续“时间旅行”提供历史数据：

（1）第一轮：用户请求研究LangGraph

# 配置对话标识（固定thread_id="1"，确保历史连贯）
config = {"configurable": {"thread_id": "1"}}

# 用户输入：请求研究LangGraph
events1 = graph.stream(
    {"messages": [{"role": "user", "content": "I'm learning LangGraph. Could you do some research on it for me?"}]},
    config,
    stream_mode="values"
)

# 打印响应（机器人会调用Tavily搜索，生成工具消息与总结）
for event in events1:
    if "messages" in event:
        event["messages"][-1].pretty_print()

流程与检查点：此轮会生成多个快照，包括“用户消息→chatbot调用工具→tools执行搜索→chatbot生成总结”等步骤。

（2）第二轮：用户表示要构建自主智能体

# 用户输入：反馈有用，并计划构建自主智能体
events2 = graph.stream(
    {"messages": [{"role": "user", "content": "Ya that's helpful. Maybe I'll build an autonomous agent with it!"}]},
    config,
    stream_mode="values"
)

# 打印响应（机器人会再次调用搜索，获取“用LangGraph构建自主智能体”的资料）
for event in events2:
    if "messages" in event:
        event["messages"][-1].pretty_print()

流程与检查点：此轮会新增快照，记录“用户新输入→chatbot再次调用工具→tools执行新搜索→chatbot生成构建建议”等步骤。

3. 查看检查点历史（获取回溯节点）

通过graph.get_state_history(config)获取thread_id="1"的所有检查点快照，每个快照包含：

• values：当前State（如messages列表长度）；
• next：下一步要执行的节点（如('tools',)表示即将执行工具节点）；
• config：含checkpoint_id（唯一标识该快照的时间点）。

代码示例：

to_replay = None  # 用于存储待回溯的快照

# 遍历所有检查点历史
for state in graph.get_state_history(config):
    # 打印每个快照的关键信息：消息数量、下一步节点
    print(f"Num Messages: {len(state.values['messages'])}, Next: {state.next}")
    print("-" * 80)
    
    # 选择“消息数量为6”的快照（可根据需求自定义筛选条件）
    if len(state.values["messages"]) == 6:
        to_replay = state

# 查看选中的快照信息：下一步节点、配置（含checkpoint_id）
print("待回溯快照的下一步节点：", to_replay.next)  # 输出示例：('tools',)
print("待回溯快照的配置：", to_replay.config)  # 含checkpoint_id，如"1efd43e3-0c1f-6c4e-8006-891877d65740"

关键发现：检查点覆盖整个对话历史，包括多轮调用的所有步骤，可跨调用回溯。

4. 实现“时间旅行”：从历史节点重启流程

基于选中的历史快照（to_replay），通过其config（含checkpoint_id）让状态图从该时间点重新执行，实现“重放”或“分支探索”：

# 从待回溯快照的config启动，inputs设为None（复用历史状态，无需新输入）
events_replay = graph.stream(
    None,  # 无新输入，基于历史状态执行
    to_replay.config,  # 关键：传入历史快照的config，指定回溯时间点
    stream_mode="values"
)

# 打印回溯后的执行结果
for event in events_replay:
    if "messages" in event:
        event["messages"][-1].pretty_print()

回溯效果：

• 选中的快照next为('tools',)，因此重启后会先执行tools节点（即重放搜索步骤）；
• 输出会从“工具消息”开始，重现后续流程（如chatbot基于搜索结果生成建议）；
• 若在回溯后传入新输入（而非None），可基于历史状态探索新分支（如用户此时改变需求，机器人从该节点开始响应新请求）。

四、核心价值与应用场景

1. 本次教程核心优势

• 调试效率提升：无需重新执行完整对话，可直接回溯到错误节点（如工具调用失败、回答偏差的步骤），定位问题原因；
• 流程分支探索：支持从某一历史节点尝试不同输入或策略（如用户在中途改变需求，机器人无需从头开始，直接基于已有上下文响应）；
• 透明化流程：通过get_state_history可查看每一步的状态变化，清晰理解智能体的推理与执行路径。

2. 典型应用场景

• 自主软件工程师：当智能体编写代码出错时，回溯到错误前的节点，修正指令后重新生成，避免重复前期正确步骤；
• 复杂任务调试：多步骤任务（如数据分析、报告生成）中，某一步结果不符合预期时，回溯到该步骤前重新执行；
• 交互式实验：用户可“反悔”并探索不同选择（如“如果刚才不调用搜索，直接回答会怎样”），提升交互灵活性。