LangGraph学习——快速入门

发表于 2025-07-15 更新于 2025-08-28 分类于 ai框架， langgraph 阅读次数：

构建langgraph聊天机器人的基本流程

创建一个 `StateGraph`

首先创建一个 StateGraph。一个 StateGraph 对象将我们的聊天机器人结构定义为“状态机”。我们将添加 节点 来表示 LLM 和聊天机器人可以调用的函数，并添加 边 来指定机器人应如何在这些函数之间进行转换。

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]

#定义状态
graph_builder = StateGraph(State)

在 langgraph 中，状态会在图的各个节点之间传递。当一个节点产生新的消息时，它会更新 State 中的 messages 字段。

我们的图现在可以处理两个关键任务

每个 节点 都可以接收当前 状态 作为输入，并输出状态的更新。
对 消息 的更新将追加到现有列表而不是覆盖它，这得益于与 Annotated 语法一起使用的预构建 add_messages 函数。

langgraph中每个消息对象通常包含以下关键属性：

role : 一个字符串，标识消息的发送者（例如 ‘human’ , ‘ai’ , ‘system’ ）。

content : 消息的具体内容，通常是字符串，但也可以是更复杂的结构（例如，用于多模态输入）。

id : 一个可选的唯一标识符。

Annotated 的作用 : 通过使用 Annotated[list, add_messages] ，你改变了这个默认行为。 add_messages 函数（由 langgraph 提供或由你自定义）的逻辑是追加而不是覆盖。所以，当一个新节点返回消息时， langgraph 会调用 add_messages 函数，将新消息追加到现有 messages 列表的末尾。

定义一个聊天模型

两种方法：

1.使用 init_chat_model(通用高层封装) 这是一个通用的辅助函数，旨在提供一个统一的接口来初始化来自不同提供商的聊天模型。

init_chat_model — 🦜🔗 LangChain 文档 — init_chat_model — 🦜🔗 LangChain documentation

2.使用如ChatOpenAI (特定于提供商的类) 这是一个专门为 OpenAI API 设计的类，提供了对 OpenAI 模型所有功能的完全访问。

from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "sk-"
#使用‘{model_provider}:{model}’格式在单个参数中指定模型和模型提供者，例如“openai:o1”
llm = init_chat_model("openai:qwen-plus-2025-04-28",
                      base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
                      )

添加一个节点

现在我们可以将聊天模型集成到一个简单的节点中

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

graph_builder.add_node("chatbot", chatbot)

chatbot 节点函数如何将当前 状态 作为输入，并返回一个包含更新的 消息 列表的字典，键为“messages”。这是所有 LangGraph 节点函数的基本模式。

我们 状态 中的 add_messages 函数会将 LLM 的响应消息追加到状态中已有的消息之后。

添加一个 `入口` 点

添加一个 入口 点，以告诉图每次运行时从何处开始工作

1	graph_builder.add_edge(START, "chatbot")

编译图

在运行图之前，我们需要对其进行编译。我们可以通过在图构建器上调用 compile() 来完成。这将创建一个 CompiledGraph，我们可以在我们的状态上调用它。

1	graph = graph_builder.compile()

可视化图

您可以使用 get_graph 方法和其中一个“绘图”方法（例如 draw_ascii 或 draw_png）来可视化图。这些 draw 方法都需要额外的依赖项。

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

运行聊天机器人

def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        print(event)
        #stream 返回的每个 event 通常是一个字典，键是图中节点的名称，值是该节点完成后的状态更新。
        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 if input() is not available
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

graph.stream() 是 LangGraph 的核心功能之一。它会执行整个图（Graph），但不是一次性返回最终结果，而是像视频流一样，一步一步地返回中间过程的更新。这使得您可以实时看到模型生成内容的每一个部分。

添加网页搜索工具

获取Tavily api

Tavily 的搜索 API 是一款专为 AI 代理 (LLM) 构建的搜索引擎，能够快速提供实时、准确和基于事实的结果。

每月 1,000 次免费搜索

Tavily Search | 🦜️🔗 LangChain 框架

获取apiTavily AI — Tavily AI

添加工具

from langchain_tavily import TavilySearch

tool = TavilySearch(
    tavily_api_key="tvly-dev-",
    max_results=2)
tools = [tool]
tool.invoke("李超是谁?")

定义图

在LLM上添加bind_tools。这让LLM知道如果它想使用搜索引擎，应使用正确的JSON格式。

定义聊天模型llm（代码同上）

将tools整合到StateGraph中

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):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

# 将一个或多个**工具（tools） 绑定到一个 大型语言模型（LLM）**上，从而创建一个新的、具备工具调用能力的 LLM 实例
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

创建一个运行工具的函数

现在，创建一个函数来运行被调用的工具。通过将工具添加到一个名为BasicToolNode的新节点来完成，该节点检查状态中的最新消息，如果消息包含tool_calls，则调用工具。它依赖于LLM的tool_calling支持，该支持在Anthropic、OpenAI、Google Gemini以及许多其他LLM提供商中可用。

# 定义 __call__ 方法，让这个类的实例可以像函数一样被调用
# inputs 是 langgraph 传进来的当前状态，是一个字典
def __call__(self, inputs: dict):
    # 1. 从状态中获取最新的消息
    # 使用了“海象操作符” :=，先从 inputs 中获取 'messages' 列表，如果不存在则返回空列表 []
    # 然后检查列表是否为空。如果不为空，则取出最后一条消息。
    if messages := inputs.get("messages", []):
        message = messages[-1]  # 通常，最后一条消息是 AI 发出的，其中包含工具调用请求
    else:
        # 如果没有消息，就报错，因为这个节点不知道该做什么
        raise ValueError("No message found in input")

    # 2. 准备一个列表，用来存放所有工具的执行结果
    outputs = []

    # 3. 遍历 AI 消息中请求的所有工具调用
    for tool_call in message.tool_calls:
        
        # 4. 执行工具
        # a. tool_call["name"] 获取工具名称 (例如 'tavily_search_results_json')
        # b. self.tools_by_name[...] 从预存的工具字典中找到对应的工具对象
        # c. .invoke(tool_call["args"]) 使用 LLM 提供的参数来调用该工具
        tool_result = self.tools_by_name[tool_call["name"]].invoke(
            tool_call["args"]
        )

        # 5. 将工具执行结果打包成 ToolMessage
        # 这是 langgraph/langchain 的标准格式，用于告诉 LLM 工具执行的结果是什么
        outputs.append(
            ToolMessage(
                content=json.dumps(tool_result),  # 工具结果必须是字符串，所以用 json.dumps 序列化
                name=tool_call["name"],  # 告诉 LLM 这是哪个工具的结果
                tool_call_id=tool_call["id"],  # 必须提供原始请求的 ID，以便 LLM 知道这个结果对应哪个请求
            )
        )
    
    # 6. 返回结果，更新图的状态
    # 返回一个字典，其中 'messages' 键对应着包含所有 ToolMessage 的列表
    # langgraph 会将这个列表中的消息追加到主状态的 'messages' 列表中
    return {"messages": outputs}

call 是 Python 中一个非常特殊的“魔术方法”（magic method）。它的作用是让一个类的实例（对象）能够像函数一样被调用。

这在 langgraph 中是一种常见且核心的设计模式。它的含义是：

节点即函数： BasicToolNode 的实例（比如 tool_node ）本身就代表了图中的一个可执行节点。

执行逻辑：当 langgraph 的状态机运行到这个 tool_node 节点时，它会直接“调用”这个节点对象，并把当前的状态（ inputs 字典）传递给它

可以使用LangGraph预构建的ToolNode。

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

定义`conditional_edges`

添加了工具节点后，现在您可以定义conditional_edges。

边（Edges）将控制流从一个节点路由到下一个节点。条件边（Conditional edges）从单个节点开始，通常包含“if”语句，根据当前图状态路由到不同的节点。这些函数接收当前的图state并返回一个字符串或字符串列表，指示接下来要调用哪个（或哪些）节点。

接下来，定义一个名为route_tools的路由函数，它检查聊天机器人输出中的tool_calls。通过调用add_conditional_edges将此函数提供给图，这会告诉图，无论何时chatbot节点完成，都要检查此函数以确定下一步去哪里。

如果存在工具调用，条件将路由到tools；如果不存在，则路由到END。由于条件可以返回END，因此这次您不需要明确设置finish_point。

def route_tools(
     state: State,
 ):
     """
     用于 conditional_edge 的路由函数：
     - 如果最后一条消息包含工具调用，则路由到 ToolNode
     - 否则路由到结束节点
     """
     # 处理 state 为列表的情况（可能是消息列表）
     if isinstance(state, list):
         ai_message = state[-1]  # 获取最后一条消息
     # 处理 state 为字典的情况（包含 messages 字段）
     elif messages := state.get("messages", []):
         ai_message = messages[-1]  # 获取最后一条消息
     else:
         raise ValueError(f"输入状态中没有找到消息: {state}")
     
     # 检查消息是否有工具调用
     if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
         return "tools"  # 有工具调用，返回 "tools" 路由到工具节点
     return END  # 没有工具调用，返回 END 结束流程

可以使用预构建的tools_condition代替route_tools以使其更简洁。

tools_condition 函数在聊天机器人需要使用工具时返回 “tools”，如果可以不使用响应则返回 “END”。

from langgraph.prebuilt import tools_condition
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
    {"tools": "tools", END: END},
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

可视化图

如上

向机器人提问

现在您可以向聊天机器人提出超出其训练数据范围的问题。

如上

添加记忆功能

LangGraph 通过持久性检查点解决了这个问题。如果您在编译图时提供一个checkpointer，并在调用图时提供一个thread_id，LangGraph 会在每一步之后自动保存状态。当您使用相同的thread_id再次调用图时，图会加载其保存的状态，允许聊天机器人从上次中断的地方继续。

我们稍后会看到，检查点比简单的聊天记忆功能强大得多——它允许您随时保存和恢复复杂状态，用于错误恢复、人工干预工作流、时间旅行交互等。但首先，让我们添加检查点以实现多轮对话。

创建 `MemorySaver` 检查点

1
2
3

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

这是一个内存中的检查点，方便本教程使用。然而，在生产应用程序中，您可能会将其更改为使用 SqliteSaver 或 PostgresSaver 并连接数据库。

编译图

使用提供的检查点编译图，图在遍历每个节点时将对 State 进行检查点。

1	graph = graph_builder.compile(checkpointer=memory)

与您的聊天机器人互动

选择一个线程作为此对话的键。

thread_id决定对话窗口
1
config = {"configurable": {"thread_id": "1"}}

调用您的聊天机器人

user_input = "我是谁"

# The config is the **second positional argument** to stream() or invoke()!
events = graph.stream(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print()

添加人工干预

代理可能不可靠，并且可能需要人工输入才能成功完成任务。同样，对于某些操作，您可能需要在运行前要求人工批准，以确保一切按预期运行。

LangGraph 的持久化层支持人工干预工作流，允许根据用户反馈暂停和恢复执行。此功能的主要接口是interrupt函数。在节点内调用interrupt将暂停执行。通过传入Command，可以恢复执行并接收来自人工的新输入。interrupt在功能上类似于 Python 的内置input()，但有一些注意事项。

添加`human_assistance`工具

将human_assistance工具添加到聊天机器人。此工具使用interrupt从人工接收信息。

# 添加人工干预功能
# 导入LangGraph的中断机制和命令类型
from langgraph.types import Command, interrupt
# 导入LangChain的工具装饰器
from langchain_core.tools import tool

# 使用@tool装饰器将函数标记为可被LLM调用的工具
@tool
def human_assistance(query: str) -> str:
    """请求人工协助的工具函数。
    当LLM遇到需要人工判断或帮助的情况时，会调用此工具。
    该函数会暂停图的执行，等待人工操作员提供响应。
    """
    # interrupt()函数会暂停图的执行，等待人工输入
    # 传入的字典包含查询信息，人工操作员会看到这个查询
    human_response = interrupt({"query": query})
    
    # 从人工响应中提取数据并返回给LLM
    # human_response是一个字典，"data"字段包含人工提供的实际响应
    return human_response["data"]

简单来说， 调用哪个工具，以及何时调用，完全是由大语言模型（LLM）根据你给它的指令（Prompt）来决定的。

1
2
3

tool = TavilySearch(max_results=2)
tools = [tool, human_assistance]
llm_with_tools = llm.bind_tools(tools)

定义chatbot

def chatbot(state: State):
    # 调用绑定了工具的LLM（llm_with_tools），传入当前的消息历史
    # LLM会根据最新的消息决定是生成文本回复，还是调用一个或多个工具
    message = llm_with_tools.invoke(state["messages"])
    
    # --- 关键断言逻辑 ---
    assert len(message.tool_calls) <= 1
    
    # 将LLM生成的新消息（可能是文本回复，也可能是工具调用请求）返回
    # 这个返回值会以字典的形式更新到状态（State）对象中
    return {"messages": [message]}

# 将 chatbot 函数作为名为 "chatbot" 的节点添加到图构建器中
graph_builder.add_node("chatbot", chatbot)

中断安全性的断言 ( assert ) : assert len(message.tool_calls) <= 1 是在实现人工干预时一个非常重要的安全措施。

问题 : 现代 LLM 支持并行工具调用（一次请求执行多个工具）。但如果其中一个工具是 human_assistance 并触发了中断，整个图会暂停。当人工操作完成后，图会从中断点恢复。此时，如果不对工具调用数量做限制，LangGraph 可能会重新尝试执行所有在中断前请求的工具，导致已经执行过的工具被再次调用。

解决方案 : 这个断言强制要求 LLM 在每一步最多只能请求调用一个工具。这样就保证了当中断发生并恢复后，不会有重复执行工具的风险，确保了流程的稳定性和可预测性。

编译图

1
2
3

memory = MemorySaver()

graph = graph_builder.compile(checkpointer=memory)

调用聊天机器人并中断

user_input = "我需要一些关于构建 AI 代理的专家指导。你能帮我请求协助吗？"
config = {"configurable": {"thread_id": "1"}}

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 2	snapshot = graph.get_state(config) snapshot.next

恢复执行

要恢复执行，请传入一个包含工具所需数据的Command对象。此数据的格式可以根据需要进行自定义。对于本示例，请使用一个带有键"data"的字典（由human_assistance决定）

human_response = (
"我们专家在此为您提供帮助！我们建议您查看 LangGraph 来构建您的代理。它比简单的自主代理更可靠、更具可扩展性。"
)   
#从暂停状态恢复执行
human_command = Command(resume={"data": human_response})

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 function):

当 LLM 调用 human_assistance 工具时，这个函数被执行。

函数内部， interrupt() 被调用，导致图暂停，并等待人工输入。

在图恢复后， interrupt() 函数会返回一个值，这个值就是您通过 Command(resume=…) 注入的内容，也就是 {“data”: human_response} 。

因此， human_assistance 函数中的 human_response 变量实际上就等于 {“data”: human_response} 。

最后， return human_response[“data”] 从这个字典中提取出 “data” 键对应的值，并将其作为 human_assistance 工具的最终返回结果。

2.恢复指令 ( Command(resume=…) ):

当您构建 Command(resume={“data”: human_response}) 时，您正在创建一个符合 human_assistance 函数期望的结构。

您将人工回复包装在一个字典里，并使用 “data” 作为键。

这个结构被传递回 interrupt() ，然后被 human_assistance 函数接收和解析。

因为 human_assistance 函数的 return 语句期望从返回的字典中访问 “data” 键，所以我们在恢复执行时必须提供一个具有相同结构的字典。这是为了确保数据能够正确地在中断和恢复的过程中传递。

在 LangGraph 中，Command 是一个用于控制图执行流程、更新状态、实现人机交互的核心类。它支持以下四个参数：

参数名	类型	说明
`update`	`dict`	用于更新图的状态（state）。例如：`Command(update={"foo": "bar"})`。
`resume`	`Any`	与 `interrupt()` 配合使用，用于恢复被中断的图执行，并传递用户输入。
`goto`	`str` 或 `Send` 或 `List[str\|Send]`	控制下一步要执行的节点，支持跳转到指定节点、多个节点序列，或使用 `Send` 对象。
`graph`	`str`	可选，指定命令作用的图。默认是当前图，也可以设为 `Command.PARENT` 表示父图。

自定义状态

在本教程中，您将向状态添加额外字段，以定义复杂行为，而无需依赖消息列表。聊天机器人将使用其搜索工具查找特定信息，并将其转发给人工进行审查。

向状态添加键

通过向状态添加 name 和 birthday 键，更新聊天机器人以研究实体的生日

class State(TypedDict):
    messages: Annotated[list, add_messages]
    name: str
    birthday: str

将此信息添加到状态中，可以使其轻松被其他图节点（例如存储或处理信息的下游节点）以及图的持久层访问。

在工具内部更新状态

现在，在 human_assistance 工具内部填充状态键。这允许人工在信息存储到状态之前对其进行审查。使用 Command 从工具内部发出状态更新。

# 从 langchain_core.messages 导入 ToolMessage，用于创建工具调用的响应消息
from langchain_core.messages import ToolMessage 
# 从 langchain_core.tools 导入 InjectedToolCallId（用于自动注入工具调用ID）和 tool（工具装饰器）
from langchain_core.tools import InjectedToolCallId, tool 

# 从 langgraph.types 导入 Command（用于向图发送指令）和 interrupt（用于中断图的执行）
from langgraph.types import Command, interrupt 
from typing import Annotated
# @tool 装饰器将这个函数声明为一个可供 LLM 调用的工具
@tool 
def human_assistance(
    name: str, 
    birthday: str, 
    # tool_call_id 这个参数非常特殊。Annotated[...] 和 InjectedToolCallId 告诉 LangGraph：
    # 1. 这个参数不应暴露给 LLM，LLM 在调用此工具时不需要提供它。
    # 2. LangGraph 在执行时，会自动将触发此工具的那个工具调用的 ID 注入到这个参数中。
    # 这个 ID 对于创建与原始请求相关联的 ToolMessage 至关重要。
    tool_call_id: Annotated[str, InjectedToolCallId]
) -> str: 
    """当需要人工确认或更正信息时，请求人类协助。""" 
    # 调用 interrupt() 来暂停图的执行，并向人类审核者呈现一个包含问题和待确认数据的字典。
    # 图会在此处暂停，直到人类通过 resume 指令提供了响应。
    human_response = interrupt( 
        { 
            "question": "Is this correct?", 
            "name": name, 
            "birthday": birthday, 
        },
    ) 
    # 检查人类的响应。如果响应中 'correct' 键的值是 'yes' 或 'y' 开头，
    # 则认为信息是正确的。
    if human_response.get("correct", "").lower().startswith("y"): 
        # 如果信息正确，直接使用从 LLM 获取的原始信息。
        verified_name = name 
        verified_birthday = birthday 
        response = "Correct" 
    # 否则，认为人类审核者提供了更正后的信息。
    else: 
        # 从人类的响应中获取更正后的姓名和生日。
        # 如果人类没有提供新的值，则使用 .get() 的默认值，即原始值。
        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, # 更新状态中的 'name' 字段
        "birthday": verified_birthday, # 更新状态中的 'birthday' 字段
        # 创建一个 ToolMessage，将其添加到状态的 'messages' 列表中。
        # 这个消息将作为此工具调用的正式“答复”出现在对话历史中。
        # tool_call_id 是必需的，用于将此答复与 LLM 的原始工具调用请求关联起来。
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)], 
    } 
    # 这个工具不返回一个简单的字符串或数字，而是返回一个 Command 对象。
    # Command(update=...) 是一个明确的指令，告诉 LangGraph 执行器：
    # “请不要将我的返回值当作普通工具输出，而是用 state_update 字典里的内容来直接更新当前的图状态。”
    return Command(update=state_update)

图的其余部分保持不变。

提示聊天机器人调用人工审查

提示聊天机器人查找 LangGraph 库的“生日”，并在其获取所需信息后，指示聊天机器人使用 human_assistance 工具。通过在工具参数中设置 name 和 birthday，您将强制聊天机器人为这些字段生成提议。

user_input = (
    "你能查一下 LangGraph 是什么时候发布的吗？ "
    "当你有了答案后，使用 human_assistance 工具进行审查。"
)
config = {"configurable": {"thread_id": "1"}}

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()

我们再次在 human_assistance 工具中触发了 interrupt。

添加人工协助

聊天机器人未能识别正确的日期，因此为其提供信息

human_command = Command(
    resume={
        "name": "LangGraph",
        "birthday": "Jan 17, 2024",
    },
)

events = graph.stream(human_command, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

请注意，这些字段现在已反映在状态中

1
2
3

snapshot = graph.get_state(config)

{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}

这使得下游节点（例如，进一步处理或存储信息的节点）可以轻松访问它们。

时间功能（从之前的某个状态开始）

在典型的聊天机器人工作流程中，用户与机器人进行一次或多次交互以完成任务。记忆和人工干预功能可以为图状态启用检查点并控制未来的响应。

如果您希望用户能够从之前的响应开始并探索不同的结果，该怎么办？或者，如果您希望用户能够回溯聊天机器人的工作以纠正错误或尝试不同的策略，这在自主软件工程师等应用程序中很常见，那又该怎么办？

您可以使用 LangGraph 内置的时光旅行功能创建这些类型的体验。

回溯您的图

通过使用图的get_state_history方法获取检查点来回溯您的图。然后，您可以从之前的这个时间点恢复执行。

# 初始化一个变量 to_replay 为 None，它将用于存储我们想要“时间旅行”回去的特定状态。
# to_replay 将在循环中被赋值为我们感兴趣的那个历史状态快照。
to_replay = None

# 遍历 `graph` 在给定 `config` 下的所有历史状态。
# `graph.get_state_history(config)` 会返回一个迭代器，其中包含了从开始到当前的所有状态快照。
for state in graph.get_state_history(config):
    # 打印当前状态快照中的一些信息，以便我们观察和选择。
    # `len(state.values["messages"])` 显示了到该状态为止，对话历史中的消息总数。
    # `state.next` 显示了在该状态之后，图将要执行的下一个节点或步骤的名称。
    print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
    
    # 打印一条分隔线，使输出更易读。
    print("-" * 80)
    
    # 这里是选择“时间旅行”目标点的关键逻辑。
    # 我们设定一个条件：当对话历史中的消息数量正好等于4时，我们就找到了想要回到的那个点。
    # 这是一个为了演示而设定的任意条件，在实际应用中，您可以根据需要设置更复杂的选择逻辑。
    if len(state.values["messages"]) == 4:
        # We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
        # 将当前这个符合条件的状态（state）保存到 to_replay 变量中。
        # 循环结束后，to_replay 变量将持有我们选中的那个历史时刻的完整状态，
        # 之后我们就可以用它来恢复或修改执行流程。
        to_replay = state

图的每一步都会保存检查点。这跨越了调用，因此您可以回溯整个线程的历史。

从特定时间点加载状态

从to_replay状态恢复。从这一点恢复将接下来调用action节点。

1 2	print(to_replay.next) print(to_replay.config)

检查点的to_replay.config包含一个checkpoint_id时间戳。提供此checkpoint_id值会告诉 LangGraph 的检查点器从该时间点加载状态。

1
2
3

for event in graph.stream(None, to_replay.config, stream_mode="values"):
    if "messages" in event:
        event["messages"][-1].pretty_print()

运行本地服务器

安装 LangGraph CLI

1
2
3

# Python >= 3.11 is required.

pip install --upgrade "langgraph-cli[inmem]"

创建 LangGraph 应用 🌱

从 new-langgraph-project-python 模板或 new-langgraph-project-js 模板创建一个新应用。此模板展示了一个单节点应用程序，您可以根据自己的逻辑进行扩展。

1	langgraph new . --template new-langgraph-project-python

使用 langgraph new 而不指定模板，系统将显示一个交互式菜单，您可以从中选择可用的模板列表。

使用uv安装依赖项

uv 是一个用 Rust 编写的极速 Python 包和项目管理器。它旨在解决传统 Python 包管理工具（如 pip 、 poetry 等）在速度和效率方面的痛点，提供更快的安装、依赖解析和环境管理。

1	pip install uv#安装uv

运行uv sync会根据 pyproject.toml的依赖创建虚拟环境并安装依赖

pyproject.toml 文件是 Python 项目中用于统一配置项目元数据、构建系统、依赖管理和各种工具设置的标准化文件。它通常用于替代旧的 requirements.txt 文件，提供更现代和集中的项目配置方式。

创建一个 `.env` 文件

您将在新 LangGraph 应用的根目录下找到一个 .env.example 文件。在新 LangGraph 应用的根目录下创建一个 .env 文件，并将 .env.example 文件的内容复制到其中，填入所需的 API 密钥。

添加环境变量如LANGSMITH_API_KEY，OPENAI_API_KEY等

启动 LangGraph 服务器

在本地启动 LangGraph API 服务器

1	langgraph dev

示例输出

>    Ready!
>
>    - API: [https://:2024](https://:2024/)
>
>    - Docs: https://:2024/docs
>
>    - LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

LangGraph 服务器（如您通过 langgraph dev 命令启动的服务器）的主要作用是提供一个运行环境和接口，用于开发、测试、部署和管理基于 LangGraph 构建的 AI 代理和应用程序。具体来说，它有以下几个主要用途：

API 接口暴露：它将您用 LangGraph 定义的复杂代理逻辑（即图结构）通过标准的 RESTful API 接口暴露出来。这意味着其他应用程序、前端界面或者其他服务可以通过 HTTP 请求与您的 LangGraph 代理进行交互，而无需直接集成 LangGraph 的 Python 代码。
简化部署：通过将 LangGraph 应用程序打包成一个可运行的服务，您可以更容易地将其部署到云服务器、容器（如 Docker）或其他生产环境中。这使得 LangGraph 代理可以作为一个独立的微服务运行，方便扩展和管理。
开发和调试便利：
- 实时预览和调试：服务器通常会提供一个 Studio UI（如您在 http://127.0.0.1:2024/studio 看到的），让开发者能够可视化地查看代理的图结构、执行流程、状态变化和中间步骤，这对于理解和调试复杂的代理行为至关重要。
- API 文档：自动生成的 API 文档（如 http://127.0.0.1:2024/docs ）提供了所有可用接口的详细说明和交互式测试功能，极大地加速了开发和集成过程。
状态管理和持久化：LangGraph 代理通常涉及复杂的状态管理。服务器可以负责处理这些状态的持久化，确保代理在多次交互之间能够记住上下文和历史信息。

在 LangGraph Studio 中测试您的应用程序

LangGraph Studio 是一个专门的 UI，您可以连接到 LangGraph API 服务器，以便在本地可视化、交互和调试您的应用程序。通过访问 langgraph dev 命令输出中提供的 URL，在 LangGraph Studio 中测试您的图。

1	> - LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

参考资料

构建一个基本聊天机器人 - LangChain 框架

Overview - Docs by LangChain

官方教程，但是英文https://academy.langchain.com/collections

3.5 小时出证！LangGraph 官方课程 🆓 重磅上线🔥🔥🔥_哔哩哔哩_bilibili

构建langgraph聊天机器人的基本流程

创建一个 StateGraph

定义一个聊天模型

添加一个节点

添加一个 入口 点

编译图

可视化图

运行聊天机器人

添加网页搜索工具

获取Tavily api

添加工具

定义图

创建一个运行工具的函数

定义conditional_edges

可视化图

向机器人提问

添加记忆功能

创建 MemorySaver 检查点

编译图

与您的聊天机器人互动

添加人工干预

添加human_assistance工具

定义chatbot

编译图

调用聊天机器人并中断

恢复执行

自定义状态

向状态添加键

在工具内部更新状态

提示聊天机器人调用人工审查

添加人工协助

时间功能（从之前的某个状态开始）

回溯您的图

从特定时间点加载状态

运行本地服务器

安装 LangGraph CLI

创建 LangGraph 应用 🌱

使用uv安装依赖项

创建一个 .env 文件

启动 LangGraph 服务器

在 LangGraph Studio 中测试您的应用程序

参考资料

创建一个 `StateGraph`

添加一个 `入口` 点

定义`conditional_edges`

创建 `MemorySaver` 检查点

添加`human_assistance`工具

创建一个 `.env` 文件