【LangGraph】langgraph.managed.IsLastStep 类：工作流中跟踪当前步骤是否为最后一步

创想小编 · 发表于 8 小时前

作者：彬彬侠
本文是对 LangGraph 中 langgraph.managed.IsLastStep 类的详细介绍，涵盖其定义、功能、使用场景、代码示例以及注意事项。

1. IsLastStep 类的概述

langgraph.managed.IsLastStep 是 LangGraph 提供的一个 托管值（Managed Value） 类，位于 langgraph.managed 模块中。它用于在工作流的状态中跟踪当前执行是否为图的最后一步。IsLastStep 是一个特殊的动态值，由 LangGraph 运行时自动管理，开发者无需手动设置其值。
核心特点

动态计算

布尔值

状态集成

用途

典型应用场景

条件逻辑

对话系统

工作流优化

2. IsLastStep 类的定义

2.1 类定义

根据 LangGraph 文档，IsLastStep 的定义如下：

from langgraph.managed import IsLastStep, ManagedValue
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
classIsLastStep(ManagedValue[bool]):def__call__(self, step:int, config: Optional[RunnableConfig]=None,**kwargs: Any)->bool:...

复制代码

继承

功能

参数：

2.2 返回值

描述：

2.3 工作原理

3. IsLastStep 的使用方式

IsLastStep 通常通过 Annotated 类型注解集成到 StateGraph 的状态定义中，作为状态的一个字段。节点函数可以读取该字段的布尔值，以根据是否为最后一步调整逻辑。
3.1 状态注解

在状态定义中，使用 Annotated[..., IsLastStep] 指定某个字段由 IsLastStep 管理：

from typing import TypedDict, Annotated
from langgraph.managed import IsLastStep
classState(TypedDict):input:str
output:str
is_last_step: Annotated[bool, IsLastStep]

复制代码

3.2 节点逻辑

节点函数可以通过状态访问 is_last_step，根据其值执行不同逻辑：

defnode(state: State)-> State:if state["is_last_step"]:
state["output"]=f"最后一步: {state['input']}"else:
state["output"]=f"中间步骤: {state['input']}"return state

复制代码

4. 使用场景和代码示例

以下是 IsLastStep 的典型使用场景和详细代码示例。
4.1 简单工作流：检查最后一步

构建一个工作流，节点根据 is_last_step 执行不同逻辑。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.managed import IsLastStep
# 定义状态classState(TypedDict):input:str
output:str
is_last_step: Annotated[bool, IsLastStep]# 定义节点defprocess(state: State)-> State:
state["output"]=f"中间处理: {state['input']}"return state
deffinalize(state: State)-> State:if state["is_last_step"]:
state["output"]=f"最后一步: {state['input']}"else:
state["output"]=f"非最后一步: {state['input']}"return state
# 创建 StateGraph
workflow = StateGraph(State)
workflow.add_node("process", process)
workflow.add_node("finalize", finalize)
workflow.add_edge(START,"process")
workflow.add_edge("process","finalize")
workflow.add_edge("finalize", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"Hello"})print(result)# {'input': 'Hello', 'output': '最后一步: Hello', 'is_last_step': True}

复制代码

说明：

4.2 对话系统：最后一步总结

在对话系统中，使用 IsLastStep 判断是否结束对话并生成总结。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.managed import IsLastStep
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
# 定义状态classState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
is_last_step: Annotated[bool, IsLastStep]# 定义节点defagent(state: State)-> State:
llm =(ChatOpenAI(model="gpt-4o-mini"))if state["is_last_step"]:
response = llm.invoke(state["messages"]+[HumanMessage(content="请总结对话")])else:
response = llm.invoke(state["messages"])return{"messages": response}# 条件函数defroute(state: State)->str:return END iflen(state["messages"])>=3else"agent"# 创建 StateGraph
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_edge(START,"agent")
workflow.add_conditional_edges("agent", route,{"agent":"agent", END: END})# 编译和运行
graph = workflow.compile()
result = graph.invoke({"messages":[HumanMessage(content="Hello!")],"is_last_step":False})print(result["messages"][-1].content)# 对话总结（在第 3 次循环后）

复制代码

说明：

4.3 工具调用：最后一步处理

在工具调用工作流中，使用 IsLastStep 决定是否格式化最终输出。

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.managed import IsLastStep
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
# 定义工具@tooldefsearch(query:str):returnf"搜索结果: {query}"# 定义状态classState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
is_last_step: Annotated[bool, IsLastStep]# 定义节点defagent(state: State)-> State:
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([search])
response = llm.invoke(state["messages"])return{"messages": response}deftool_node(state: State)-> State:
tool_call = state["messages"][-1].tool_calls[0]
result = search.invoke(tool_call["args"]["query"])return{"messages": ToolMessage(content=result, tool_call_id=tool_call["id"])}deffinalize(state: State)-> State:if state["is_last_step"]:
last_message = state["messages"][-1]
state["messages"].append(AIMessage(content=f"最终结果: {last_message.content}"))return state
# 条件函数defroute(state: State)->str:return"tool"if state["messages"][-1].tool_calls else"finalize"# 创建 StateGraph
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tool", tool_node)
workflow.add_node("finalize", finalize)
workflow.add_edge(START,"agent")
workflow.add_conditional_edges("agent", route,{"tool":"tool","finalize":"finalize"})
workflow.add_edge("tool","agent")
workflow.add_edge("finalize", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"messages":[HumanMessage(content="搜索 LangGraph")]})print(result["messages"][-1].content)# 最终结果: 搜索结果: LangGraph

复制代码

说明：

5. 注意事项

托管值特性

状态定义

最后一步的定义

异步支持

示例：
1. asyncdefasync_node(state: State)-> State:if state["is_last_step"]:
2. state["output"]="异步最后一步"return state
复制代码

版本兼容性

性能考虑

6. 总结

langgraph.managed.IsLastStep 是一个托管值类，用于在 LangGraph 工作流中跟踪当前步骤是否为最后一步。其核心特点包括：

动态布尔值

状态集成

灵活应用

通过在状态中定义 is_last_step 字段并在节点中检查其值，你可以轻松实现基于最后一步的条件逻辑。IsLastStep 特别适合对话系统和多步骤工作流，与 MessageGraph 和 StateGraph 无缝配合。
参考文献：

原文地址：https://blog.csdn.net/u013172930/article/details/147986016

25位大厂高管转战生成式AI创业！吸金猛，不

【LangGraph】langgraph.managed.IsLastStep 类：工作流中跟踪当前步骤是否为最后一步

浏览过的版块

LangChain入门2 RAG详解

关于我们

产品与服务

解决方案

产品与服务