贫瘠之地

华北无浪漫，死海扬起帆
多少个夜晚，独自望着天

0%

LangGraph - 快速开始

发表于 分类于 > >

介绍

LangGraph 是 LangChain 社区下的新框架，主要用于构建基于状态的、多行动者的 LLM 应用，应用于 Agent 和复合 Agent 工作流

提供了对应用程序的流和状态的细粒度控制


其核心功能

  • 循环和分支：在应用程序中实现循环和条件
  • 持续：每个步骤后自动保存状态，随时暂停、恢复流程图执行
  • 人工循环：中断流程图执行以批准或编辑代理计划的下一个操作
  • 流式传输支持：流式传输每个节点产生的输出（包括令牌流）
  • 和 LangChain 集成：和 LangChain 及 LangSmith 无缝集成


实现一个流程图的步骤

  1. 初始化 LLM 和工具
  2. 初始化流程图中的状态（state）
  3. 定义流程图的节点（node）
  4. 定义流程图的顶点（entry point）和边（graph）
  5. 构建（compile）流程图
  6. 执行

快速开始

这里使用官方文档中的 Tutorials - Introduction 内容，来简单介绍 LangGraph 的使用，个别地方会进行修改

这里只记录代码部分，官方文档中的细节解释、验证、LangSmith 上的日志观察等内容因为太多，可以直接参考官方文档

在快速开始中将会基于 LangGraph 构建一个聊天机器人，它能够：

  • 通过搜索网络回答常见问题
  • 在通话中保持通话状态
  • 将复杂的查询发送给人类进行审查
  • 使用自定义状态控制其行为
  • 回放并探索其他对话路径

将从一个基本的聊天机器人开始，逐步添加更复杂的功能，在此过程中引入关键的 LangGraph 概念


配置

安装相应的包（LangChain 就不在这里列举了）

1
pip install -U langgraph langsmith

设置对应的 API key；这里我没有使用官方教程里的 langchain_anthropic ，直接使用的 LangChain 中的 AzureChatOpenAI

所以这里认为只需要配置 LangSmith 相关的 key 即可

1
2
3
os.environ["ANTHROPIC_API_KEY"] = ""
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "LangGraph Tutorial"


Part.1 构建一个基础的聊天机器人

状态机

首先创建一个 StateGraph ，一个 StateGraph 对象会将聊天机器人的结构定义为状态机

我们将添加节点来表示 LLM 可以调用的函数，并添加边来指定机器人应该如何在这些函数之间转换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from typing import Annotated

from typing_extensions import TypedDict

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

class State(TypedDict):
    # Messages have the type "list". The `add_messages` function
    # in the annotation defines how this state key should be updated
    # (in this case, it appends messages to the list, rather than overwriting them)
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

首先定义了一个 TypedDict ，它只有一个 key messages ，messages 使用 add_messages 函数进行注释，该函数告诉 LangGraph 将新消息添加到现有列表中，而不是进行覆盖

现在我们的流程图知道两件事：

  1. 我们定义的每个节点都将接收当前状态作为输入，并返回更新该状态的值
  2. 消息将附加到当前列表中，而不是直接覆盖。这是通过注释语法中预先构建的 add_messages 函数进行设置的

聊天机器人节点

节点代表工作单元，它们通常是常规的 python 函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langchain_anthropic import ChatAnthropic
import Azure

# 这里是我的实现
# 只要提供出一个 LLM 即可
llm = Azure.chat_model_4o

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

# The first argument is the unique node name
# The second argument is the function or object that will be called whenever
# the node is used.
graph_builder.add_node("chatbot", chatbot)

chatbot 节点功能如何将当前状态作为输入并返回更新的 messages，这是所有 LangGraph 节点函数的基本模式

我们的 State 中的 add_messages 函数将把 LLM 的响应消息附加到该状态中已经存在的任何消息上

添加顶点

接下来，添加一个入口点 entry，这告诉我们的流程图每次运行时从哪里开始工作

1
graph_builder.set_entry_point("chatbot")

类似的添加一个终点 finish，这指示流程图 any time this node is run, you can exit.

1
graph_builder.set_finish_point("chatbot")

构建

最后能够运行我们的图，在图形生成器上调用 compile

这将创建一个 CompiledGraph，我们就可以进行 invoke 调用了

1
graph = graph_builder.compile()

最终的流程图结构应该为

Run

最后使用 invoke 或 stream 方法进行执行

1
2
3
4
5
6
7
8
while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    for event in graph.stream({"messages": ("user", user_input)}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)


Part.2 使用工具增强

这里我们使用维基百科的查询工具

1
2
3
4
5
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [wikipedia]

LLM 绑定工具

我们需要将工具先和 LLM 绑定，即将工具调用以 JSON schema 的方式请求 LLM

1
2
model = Azure.chat_model_4o
llm = model.bind_tools(tools)

定义 Tool 节点

这里将自己实现一个对工具执行的节点，其实这个操作 LangGraph 是有预先构建好的实现的，但是这里文档为了让其深入了解原理，先手动实现，后面会进行替换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import json

from langchain_core.messages import ToolMessage

class BasicToolNode:
    """A node that runs the tools requested in the last AIMessage."""

    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("No message found in input")
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    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)

实现的基本逻辑是，从入参中获取消息，获取最后一条消息，取出 tool_calls 的内容后进行调用

定义条件边

当定义了节点后，接下来需要定义 conditional_edges

边将控制流从一个节点路由到下一个节点，这些函数接收当前图形状态，并返回一个字符串或字符串列表，指示下一个调用哪个节点

下面会定义一个路由器函数，用于检查聊天机器人的输出中的 tool_calls

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from typing import Literal

def route_tools(
    state: State,
) -> Literal["tools", "__end__"]:
    """
    Use in the conditional_edge to route to the ToolNode if the last message
    has tool calls. Otherwise, route to the end.
    """
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return "__end__"

# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "__end__" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
    # It defaults to the identity function, but if you
    # want to use a node named something else apart from "tools",
    # You can update the value of the dictionary to something else
    # e.g., "tools": "my_tools"
    {"tools": "tools", "__end__": "__end__"},
)
# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()

如果没有进行任何工具调用，我们的函数将返回 __end__ 字符串，当流程图需要切换到 __end__ 时，它没有其他任务要完成，所以停止执行

因为条件可以返回 __end__，所以这次不需要显式设置 finish_point

最终的流程图应该为


Part.3 添加记忆

上面实现的流程图其调用都是无状态的

LangGraph 通过持久的检查点（persistent checkpointing）来解决这个问题，需要提供一个 checkpointer 并且在调用时提供 thread_id

checkpointing 比记忆更加强大，它可以实现随时保存和恢复复杂的状态，用于错误恢复、人工循环工作流、时间旅行交互等

创建 SqliteSaver 检查点

1
2
3
from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")

这里也可以替换为其他实现

提供 checkpointer 编译

1
2
# 注意这里的 checkpointer 入参
graph = graph_builder.compile(checkpointer=memory)

此时流程图应该为

Run

根据官方文档，调用时需要传参 thread_id

1
2
3
4
5
6
7
8
9
# The only difference is we change the `thread_id` here to "2" instead of "1"
events = graph.stream(
    {"messages": [("user", user_input)]},
    # here
    {"configurable": {"thread_id": "2"}},
    stream_mode="values",
)
for event in events:
    event["messages"][-1].pretty_print()

get state

可以通过 config 随时检查给定配置的流程图状态

1
2
3
snapshot = graph.get_state(config)
snapshot
snapshot.next  # (since the graph ended this turn, `next` is empty. If you fetch a state from within a graph invocation, next tells which node will execute next)

上面的快照包含当前状态值、相应的配置以及要处理的下一个节点

在我们的例子中，流程图已经达到 __end__ 状态，所以 next 为空


Part.4 循环中人类介入

代理并不完全可靠，所以在某些节点需要人工介入

LangGraph 提供了多种方式支持人工循环工作流 human-in-the-loop，在本节中将使用 LangGraph 的 interrupt_before 功能来始终中断工具节点

前中断构建

编译流程图，指定在操作节点之前中断

1
2
3
4
5
6
7
graph = graph_builder.compile(
    checkpointer=memory,
    # This is new!
    interrupt_before=["tools"],
    # Note: can also interrupt __after__ actions, if desired.
    # interrupt_after=["tools"]
)
1
2
3
4
5
6
7
8
9
user_input = "I'm learning LangGraph. Could you do some research on it for me?"
config = {"configurable": {"thread_id": "1"}}
# The config is the **second positional argument** to stream() or invoke()!
events = graph.stream(
    {"messages": [("user", user_input)]}, 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)
snapshot.next
# ('action',)

不同于之前，这次下一个状态为 action，我们已经成功拦截在了工具执行之前

继续

接下来继续流程

传入 None 只会让流程继续从停止的位置进行，不会向状态添加任何新内容

1
2
3
4
5
# `None` will append nothing new to the current state, letting it resume as if it had never been interrupted
events = graph.stream(None, config, stream_mode="values")
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

最终实现了使用了一个中断将人工在环执行添加到聊天机器人中

允许在需要时进行人工监督和干预，添加了一个检查指针，流程图可以无限期地暂停，并在任何时候恢复，就像什么都没有发生一样


Part.5 手动更新状态

上面人类中断时，选择继续执行；下面需要来实现如果希望对过程进行修改、不进行工具执行该如何实现

通过手动更新状态，可以实现纠正 Agent 的错误、探索替代路径、引导 Agent 实现特定目标等

直接提供正确的回应

和上一个流程一致，当我们通过检查点暂停到调用工具之前

1
2
3
snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
existing_message.pretty_print()
1
2
3
4
5
6
7
8
================================== Ai Message ==================================

[{'id': 'toolu_01DTyDpJ1kKdNps5yxv3AGJd', 'input': {'query': 'LangGraph'}, 'name': 'tavily_search_results_json', 'type': 'tool_use'}]
Tool Calls:
  tavily_search_results_json (toolu_01DTyDpJ1kKdNps5yxv3AGJd)
 Call ID: toolu_01DTyDpJ1kKdNps5yxv3AGJd
  Args:
    query: LangGraph

这里显示的消息在工具调用之前（Tool Calls）

接下来提供指定回答

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.messages import AIMessage

answer = (
    "LangGraph is an apple."
)
new_messages = [
    # The LLM API expects some ToolMessage to match its tool call. We'll satisfy that here.
    ToolMessage(content=answer, tool_call_id=existing_message.tool_calls[0]["id"]),
    # And then directly "put words in the LLM's mouth" by populating its response.
    AIMessage(content=answer),
]

new_messages[-1].pretty_print()
graph.update_state(
    # Which state to update
    config,
    # The updated values to provide. The messages in our `State` are "append-only", meaning this will be appended
    # to the existing state. We will review how to update existing messages in the next section!
    {"messages": new_messages},
)

print("\n\nLast 2 messages;")
print(graph.get_state(config).values["messages"][-2:])

这时我们再拿到 messages，可以看到最终结果就是我们定义好的内容

为什么是 append

上面的流程会衍生出一个疑问，我们手动添加的新消息为什么会被追加到已经存在状态中的消息后

因为一开始就是这样定义

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

上面的实现高速流程图始终将值 append 到现有列表，而不是直接覆盖列表

这里应用了相同的逻辑，所以我们传递给 update_state 的消息是以相同的方式附加的

以哪个节点的名义

默认情况下，update_state 使用上次执行的节点

也可以通过传参方式告诉流程图更新来自哪个节点

1
2
3
4
5
6
7
graph.update_state(
    config,
    {"messages": [AIMessage(content="I'm an AI expert!")]},
    # Which node for this function to act as. It will automatically continue
    # processing as if this node just ran.
    as_node="chatbot",
)

再来看一次流程图，如果我们以 chatbot 的名义更新了消息，那么根据流程图，整个流程就会直接结束

1
2
3
snapshot = graph.get_state(config)
print(snapshot.values["messages"][-3:])
print(snapshot.next) # ()

查看状态可以发现，历史消息中不会有工具调用的记录

覆盖现有消息

可以通过 message 上的 id 来覆盖已有消息

在工具执行前的观察点进行暂停

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_core.messages import AIMessage

snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
print("Original")
print("Message ID", existing_message.id)
print(existing_message.tool_calls[0])
new_tool_call = existing_message.tool_calls[0].copy()
new_tool_call["args"]["query"] = "Where is China?"
new_message = AIMessage(
    content=existing_message.content,
    tool_calls=[new_tool_call],
    # Important! The ID is how LangGraph knows to REPLACE the message in the state rather than APPEND this messages
    id=existing_message.id,
)

print("Updated")
print(new_message.tool_calls[0])
print("Message ID", new_message.id)
graph.update_state(config, {"messages": [new_message]})

print("\n\nTool calls")
graph.get_state(config).values["messages"][-1].tool_calls

这里构造了一个新的工具调用消息，通过 update_state 进行覆盖

注意如果需要覆盖 tool calls 消息，需要保证 id 一致，即上述代码中的 existing_message.id

接下来继续流程

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

我这里会发现当执行了覆盖的工具调用后，又到了新的工具调用（因为 LLM 根据上下文发现没有查询需要查询的问题）

但是我们覆盖的消息已经被成功加入流程中了


Part.6 自定义状态

上述示例中都是使用一个简单的状态来操作消息列表，也可以通过状态中的更多字段来拓展其功能

上面的例子中每次工具调用时，流程都会暂停等待人工，下面将实现依靠 LLM 来选择是否需要进行人工处理

一种方法是创建一个 human 节点，在该节点之前图形将始终停止，只有当 LLM 调用 human 工具时，我们才会执行此节

为了方便起见，我们将在流程图状态中包含一个 ask_human 标识，如果 LLM 调用此工具，我们将切换该标识

定义一个带属性的状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from typing import Annotated

from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list, add_messages]
    # This flag is new
    ask_human: bool

定义请求协助工具

接下来定义一个 schema（tool）来让模型决定是否请求帮助

1
2
3
4
5
6
7
8
9
from langchain_core.pydantic_v1 import BaseModel

class RequestAssistance(BaseModel):
    """Escalate the conversation to an expert. Use this if you are unable to assist directly or if the user requires support beyond your permissions.

    To use this function, relay the user's 'request' so the expert can provide the right guidance.
    """

    request: str

chatbot 节点

接下来定义聊天机器人节点。如果我们看到聊天机器人调用了 RequestAssistance，就切换 ask_human 标识

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [wikipedia]

llm = Azure.chat_model_4o
llm_with_tools = llm.bind_tools(tools + [RequestAssistance])

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    ask_human = False
    if (
        response.tool_calls
        and response.tool_calls[0]["name"] == RequestAssistance.__name__
    ):
        ask_human = True
    return {"messages": [response], "ask_human": ask_human}

创建流程图图生成器，并将聊天机器人和工具节点添加到图中

1
2
3
4
5
graph_builder = StateGraph(State)

graph_builder.add_node("chatbot", chatbot)
# 看起来这里没必要将 RequestAssistance 也加入，因为 RequestAssistance 只是一个标记
graph_builder.add_node("tools", ToolNode(tools=[tool]))

human 节点

接下来创建 human 节点，这个节点函数在我们的图中主要是一个占位符，它将触发中断

如果人工在中断期间没有手动更新状态，它会插入一条工具消息，以便 LLM 知道用户被请求但没有响应

该节点还取消设置 ask_human 标志，以便流程图知道除非发出进一步的请求，否则不要重新访问该节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langchain_core.messages import AIMessage, ToolMessage

def create_response(response: str, ai_message: AIMessage):
    return ToolMessage(
        content=response,
        tool_call_id=ai_message.tool_calls[0]["id"],
    )

def human_node(state: State):
    new_messages = []
    if not isinstance(state["messages"][-1], ToolMessage):
        # Typically, the user will have updated the state during the interrupt.
        # If they choose not to, we will include a placeholder ToolMessage to
        # let the LLM continue.
        new_messages.append(
            create_response("No response from human.", state["messages"][-1])
        )
    return {
        # Append the new messages
        "messages": new_messages,
        # Unset the flag
        "ask_human": False,
    }

graph_builder.add_node("human", human_node)

条件边

接下来定义条件边，如果设置了标志，select_next_node 将路由到人工节点

否则让预构建的 tools_condition 函数选择下一个节点

tools_condition 函数只是检查聊天机器人是否在其响应消息中使用了任何 tool_calls 进行了响应，如果是将路由到动作节点，否则将结束流程

1
2
3
4
5
6
7
8
9
10
11
def select_next_node(state: State):
    if state["ask_human"]:
        return "human"
    # Otherwise, we can route as before
    return tools_condition(state)

graph_builder.add_conditional_edges(
    "chatbot",
    select_next_node,
    {"human": "human", "tools": "tools", "__end__": "__end__"},
)

编译

最后添加简单的有向边并编译图，这些边指示每当 a 完成执行时，图总是从节点 a->b 流出

1
2
3
4
5
6
7
8
9
10
# The rest is the same
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge("human", "chatbot")
graph_builder.set_entry_point("chatbot")
memory = SqliteSaver.from_conn_string(":memory:")
graph = graph_builder.compile(
    checkpointer=memory,
    # We interrupt before 'human' here instead.
    interrupt_before=["human"],
)

流程图如上，具体的执行步骤就参考官方文档即可


Part.7 时间旅行

时间旅行（Time Travel）可以实现

  • 从以前的响应开始，分支探索单独的结果
  • 用户能够 “倒带” 助理的工作来纠正一些错误或尝试不同的策略（在自主软件工程师等应用程序中很常见）
  • 以及更多功能

通过使用流程图的 get_state_history 方法获取检查点来 “倒带” （rewind）流程图

设置记忆检查点

基于上面的流程不用变更任何代码

可以确认，我们的 checkpointer 基于 memory

1
2
3
4
graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"]
)

模拟流程

让流程图走几步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
"""
第一步
"""
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
    {
        "messages": [
            ("user", "Where is China?")
        ]
    },
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()
        

"""
第二步
"""
events = graph.stream(
    {
        "messages": [
            ("Where is its capital?")
        ]
    },
    config,
    stream_mode="values",
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

这里放一下结果吧

1
2
3
4
5
6
7
8
9
10
11
12
================================ Human Message =================================

Where is China?
================================== Ai Message ==================================

China is a country located in East Asia. It is the world's most populous country, with a population of over 1.4 billion people. China is bordered by 14 countries: Afghanistan, Bhutan, India, Kazakhstan, Kyrgyzstan, Laos, Mongolia, Myanmar, Nepal, North Korea, Pakistan, Russia, Tajikistan, and Vietnam. To the east, it has a coastline along the East China Sea, Korea Bay, Yellow Sea, and South China Sea. The capital city of China is Beijing.
================================ Human Message =================================

Where is its capital?
================================== Ai Message ==================================

The capital of China is Beijing. Beijing is located in the northern part of the country. It is situated in the northern tip of the North China Plain, near the western coast of the Bohai Sea. The city is bordered by Hebei Province to the north, west, south, and a small section to the east, and by Tianjin Municipality to the southeast. Beijing is one of the most populous cities in the world and serves as the political, cultural, and educational center of China.

回看历史

上面让流程图进行了几次调用

下面可以通过 get_state_history 获取所有历史，并且保存 to_replay

1
2
3
4
5
6
7
to_replay = None
for state in graph.get_state_history(config):
    print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
    print("-" * 80)
    if len(state.values["messages"]) == 2:
        # We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
        to_replay = state

检查点是为流程图每个步骤保存的，会跨越多次调用

因此可以在整个线程的历史记录中进行倒带。我们选择了 to_replay 作为恢复状态

1
2
print(to_replay.next)
print(to_replay.config)
1
2
()
{'configurable': {'thread_id': '1', 'thread_ts': '1ef3228b-2fee-631a-8001-1697d5779962'}}

这里是第一次调用结束，通过 to_replay.next 可以看出

基于检查点配置调用

检查点的配置 to_replay.config 包含一个 thread_ts 时间戳

提供这个 thread_ts 值告诉 LangGraph 的检查指针从那个时刻开始加载状态

1
2
3
4
5
6
7
8
# The `thread_ts` in the `to_replay.config` corresponds to a state we've persisted to our checkpointer.
for event in graph.stream({
    "messages": [
        ("Where is its economic center?")
    ]
}, to_replay.config, stream_mode="values"):
    if "messages" in event:
        event["messages"][-1].pretty_print()
1
2
3
4
5
6
7
8
================================ Human Message =================================

Where is its economic center?
================================== Ai Message ==================================

China's economic center is often considered to be Shanghai. Shanghai is the largest city in China by population and one of the largest urban areas in the world. It is a global financial hub and a major center for commerce, trade, and industry. The city hosts the Shanghai Stock Exchange, which is one of the largest stock exchanges in the world by market capitalization.

Shanghai's strategic location on the Yangtze River Delta and its extensive port facilities make it a key player in both domestic and international trade. The city is also known for its modern skyline, including landmarks such as the Shanghai Tower and the Oriental Pearl TV Tower.


总结

通过上面的流程就完成了介绍教程，并在 LangGraph 中构建了一个聊天机器人

它支持

  • 工具调用
  • 内存记忆
  • 人工交互
  • 时间旅行

更多内容将在后续的文档中展开


参考

Tutorials - LangGraph (langchain-ai.github.io)