AI创想

标题: 【LangGraph】langgraph.managed.IsLastStep 类：工作流中跟踪当前步骤是否为最后一步 [打印本页]

---

作者: 创想小编    时间: 9 小时前
标题: 【LangGraph】langgraph.managed.IsLastStep 类：工作流中跟踪当前步骤是否为最后一步
作者：彬彬侠
本文是对 LangGraph 中 langgraph.managed.IsLastStep 类的详细介绍，涵盖其定义、功能、使用场景、代码示例以及注意事项。

---

1. IsLastStep 类的概述

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

动态计算：IsLastStep 的值由 LangGraph 运行时根据图的执行状态自动确定，通常在节点执行时计算。布尔值：其值是一个布尔类型（bool），表示当前步骤是否为图的最后一步（即是否到达 END 或终止节点）。状态集成：通过 Annotated 类型注解集成到状态中，允许节点访问 IsLastStep 值以调整逻辑。用途：帮助节点根据是否为最后一步执行不同的行为，例如在最后一步触发总结、保存结果或跳过某些操作。

典型应用场景

条件逻辑：在节点中检查是否为最后一步，以决定是否执行特定操作（例如生成最终回复或清理资源）。对话系统：在多轮对话中，判断是否结束对话并返回总结。工作流优化：在最后一步执行特殊处理，如记录日志或发送通知。

---

2. IsLastStep 类的定义

2.1 类定义

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

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

复制代码

继承：IsLastStep 继承自 ManagedValue[bool]，表示它是一个托管值，专门管理布尔类型的值。功能：在图的每次执行步骤中，LangGraph 运行时会调用 IsLastStep 的 __call__ 方法，动态计算当前步骤是否为最后一步。

• 参数：
  
  step：当前执行的步骤编号（整数），由 LangGraph 内部跟踪。config：运行时配置（RunnableConfig），包含上下文信息（如 thread_id）。**kwargs：其他运行时参数，通常由 LangGraph 传递。
  
  
  

2.2 返回值

类型：bool

• 描述：
  
  返回 True：如果当前步骤是图的最后一步（即执行后跳转到 END 或终止节点）。返回 False：如果当前步骤不是最后一步（图将继续执行其他节点）。
  
  
  

2.3 工作原理

IsLastStep 的值由 LangGraph 运行时根据图的结构和执行状态动态计算。它检查当前节点的后续边（无条件边或条件边）是否指向 END，或者是否满足图的终止条件。开发者通过状态中的 Annotated 字段访问 IsLastStep 值，无需手动计算。

---

3. IsLastStep 的使用方式

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

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

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

复制代码

is_last_step：状态字段，类型为 bool，由 IsLastStep 动态提供值。LangGraph 运行时会在每次节点执行时自动更新 is_last_step 的值。

3.2 节点逻辑

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

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

复制代码

---

4. 使用场景和代码示例

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

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

1. from typing import TypedDict, Annotated
2. from langgraph.graph import StateGraph, START, END
3. from langgraph.managed import IsLastStep
5. # 定义状态classState(TypedDict):input:str
6.     output:str
7.     is_last_step: Annotated[bool, IsLastStep]# 定义节点defprocess(state: State)-> State:
8.     state["output"]=f"中间处理: {state['input']}"return state
10. deffinalize(state: State)-> State:if state["is_last_step"]:
11.         state["output"]=f"最后一步: {state['input']}"else:
12.         state["output"]=f"非最后一步: {state['input']}"return state
14. # 创建 StateGraph
15. workflow = StateGraph(State)
16. workflow.add_node("process", process)
17. workflow.add_node("finalize", finalize)
18. workflow.add_edge(START,"process")
19. workflow.add_edge("process","finalize")
20. workflow.add_edge("finalize", END)# 编译和运行
21. graph = workflow.compile()
22. result = graph.invoke({"input":"Hello"})print(result)# {'input': 'Hello', 'output': '最后一步: Hello', 'is_last_step': True}

复制代码

说明：

process 节点执行中间处理。finalize 节点检查 is_last_step，因为它是最后一步（连接到 END），所以输出“最后一步”。is_last_step 由 LangGraph 自动设置为 True。

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

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

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

复制代码

说明：

agent 节点检查 is_last_step，在最后一步添加总结提示。图循环 3 次（由 route 控制），最后一步触发总结。is_last_step 在最后一次循环（跳转到 END）时为 True。

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

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

1. from typing import TypedDict, Annotated, List
2. from langgraph.graph import StateGraph, START, END, add_messages
3. from langgraph.managed import IsLastStep
4. from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
5. from langchain_core.tools import tool
6. from langchain_openai import ChatOpenAI
8. # 定义工具@tooldefsearch(query:str):returnf"搜索结果: {query}"# 定义状态classState(TypedDict):
9.     messages: Annotated[List[BaseMessage], add_messages]
10.     is_last_step: Annotated[bool, IsLastStep]# 定义节点defagent(state: State)-> State:
11.     llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([search])
12.     response = llm.invoke(state["messages"])return{"messages": response}deftool_node(state: State)-> State:
13.     tool_call = state["messages"][-1].tool_calls[0]
14.     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"]:
15.         last_message = state["messages"][-1]
16.         state["messages"].append(AIMessage(content=f"最终结果: {last_message.content}"))return state
18. # 条件函数defroute(state: State)->str:return"tool"if state["messages"][-1].tool_calls else"finalize"# 创建 StateGraph
19. workflow = StateGraph(State)
20. workflow.add_node("agent", agent)
21. workflow.add_node("tool", tool_node)
22. workflow.add_node("finalize", finalize)
23. workflow.add_edge(START,"agent")
24. workflow.add_conditional_edges("agent", route,{"tool":"tool","finalize":"finalize"})
25. workflow.add_edge("tool","agent")
26. workflow.add_edge("finalize", END)# 编译和运行
27. graph = workflow.compile()
28. result = graph.invoke({"messages":[HumanMessage(content="搜索 LangGraph")]})print(result["messages"][-1].content)# 最终结果: 搜索结果: LangGraph

复制代码

说明：

finalize 节点检查 is_last_step，在最后一步格式化输出。is_last_step 在 finalize 节点（连接到 END）时为 True。

---

5. 注意事项

托管值特性：

IsLastStep 是只读的，由 LangGraph 运行时动态计算，节点无法直接修改其值。确保状态字段使用 Annotated[..., IsLastStep]，否则无法访问。

状态定义：

is_last_step 字段必须显式定义在状态中，且类型为 bool。不要在节点中手动设置 is_last_step，否则会覆盖托管值。

最后一步的定义：

is_last_step 为 True 的条件是当前节点执行后跳转到 END 或图终止。如果图有循环或条件边，is_last_step 可能在多个节点动态变化。

异步支持：

IsLastStep 支持异步工作流（graph.astream），其值在异步节点中同样有效。

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

  复制代码
版本兼容性：

确保使用最新版本的 LangGraph（如 0.2.x +），以支持 IsLastStep 的完整功能。旧版本可能对托管值的支持有限，建议升级 langgraph 和 langchain-core。

性能考虑：

IsLastStep 的计算由 LangGraph 运行时高效处理，开销极低。如果状态字段过多，注意优化状态结构，避免不必要的复杂性。

---

6. 总结

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

动态布尔值：由 LangGraph 运行时计算，指示是否到达 END。状态集成：通过 Annotated[..., IsLastStep] 嵌入状态，节点可读取。灵活应用：适合对话总结、工具调用后处理、工作流优化等场景。

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

LangGraph 官方文档：https://langchain-ai.github.io/langgraph/GitHub 仓库：https://github.com/langchain-ai/langgraphLangChain 文档：https://python.langchain.com/docs/

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

---

欢迎光临 AI创想 (https://www.llms-ai.com/)

Powered by Discuz! X3.4