如何在达到递归限制之前返回状态¶

先决条件

本指南假设您熟悉以下内容

• 图
• 递归限制
• 节点

设置图递归限制 可以帮助您控制图运行的时间，但是如果达到递归限制，您的图会返回错误 - 这对于所有用例来说可能并不理想。相反，您可能希望返回在达到递归限制之前的状态值。本操作指南将向您展示如何执行此操作。

设置¶

首先，让我们安装所需的包

在 [ ]

%%capture --no-stderr
%pip install -U langgraph

%%capture --no-stderr %pip install -U langgraph

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控您使用 LangGraph 构建的 LLM 应用程序 - 阅读有关如何入门的更多信息 这里.

不返回状态¶

我们将在本示例中定义一个虚拟图，该图将始终达到递归限制。首先，我们将实现不返回状态的图，并展示它会达到递归限制。此图基于 ReACT 架构，但它不实际做出决定和采取行动，而是永远循环。

在 [1]

from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph import START, END


class State(TypedDict):
    value: str
    action_result: str


def router(state: State):
    if state["value"] == "end":
        return END
    else:
        return "action"


def decision_node(state):
    return {"value": "keep going!"}


def action_node(state: State):
    # Do your action here ...
    return {"action_result": "what a great result!"}


workflow = StateGraph(State)
workflow.add_node("decision", decision_node)
workflow.add_node("action", action_node)
workflow.add_edge(START, "decision")
workflow.add_conditional_edges("decision", router, ["action", END])
workflow.add_edge("action", "decision")
app = workflow.compile()

from typing_extensions import TypedDict from langgraph.graph import StateGraph from langgraph.graph import START, END class State(TypedDict): value: str action_result: str def router(state: State): if state["value"] == "end": return END else: return "action" def decision_node(state): return {"value": "keep going!"} def action_node(state: State): # 执行您的操作 ... return {"action_result": "what a great result!"} workflow = StateGraph(State) workflow.add_node("decision", decision_node) workflow.add_node("action", action_node) workflow.add_edge(START, "decision") workflow.add_conditional_edges("decision", router, ["action", END]) workflow.add_edge("action", "decision") app = workflow.compile()

在 [2]

from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png()))

让我们验证一下我们的图将始终达到递归限制

在 [3]

from langgraph.errors import GraphRecursionError

try:
    app.invoke({"value": "hi!"})
except GraphRecursionError:
    print("Recursion Error")

from langgraph.errors import GraphRecursionError try: app.invoke({"value": "hi!"}) except GraphRecursionError: print("Recursion Error")

Recursion Error

返回状态¶

如果我们想要实际返回状态，我们将要做的就是在我们的状态中引入一个名为is_last_step的新键，它跟踪我们是否处于递归限制的最后一步。如果是这样，我们将绕过所有其他图决策，只终止图，将状态返回给用户，而不会导致错误。

我们将使用ManagedValue通道来执行此操作。ManagedValue通道是一个状态通道，它将在我们的图运行期间一直存在，并且不会更长时间。由于我们的action节点将始终至少为我们的图引入 2 个额外的步骤（因为action节点总是随后调用decision节点），我们将使用此通道来检查我们是否在限制的 2 步之内。请参阅以下IsLastOrSecondToLastStepManager的实现。

此实现非常类似于isLastStep的实现（您可以通过调用from langgraph.managed import IsLastStep，然后使用isLastStep类型装饰状态键来使用它），但在这种情况下，我们检查我们是否处于最后一步或倒数第二步，而不是仅仅是最后一步。

现在，当我们运行图时，我们应该不会收到任何错误，而是收到在达到递归限制之前状态的最后一个值。

在 [4]

from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from typing import Annotated

from langgraph.managed.base import ManagedValue


class IsLastOrSecondToLastStepManager(ManagedValue[bool]):
    def __call__(self, step: int) -> bool:
        limit = self.config.get("recursion_limit", 0)
        return step >= limit - 2


class State(TypedDict):
    value: str
    action_result: str
    is_last_step: Annotated[bool, IsLastOrSecondToLastStepManager]


def router(state: State):
    # Force the agent to end if it is on the last step
    if state["is_last_step"]:
        return END
    if state["value"] == "end":
        return END
    else:
        return "action"


def decision_node(state):
    return {"value": "keep going!"}


def action_node(state: State):
    # Do your action here ...
    return {"action_result": "what a great result!"}


workflow = StateGraph(State)
workflow.add_node("decision", decision_node)
workflow.add_node("action", action_node)
workflow.add_edge(START, "decision")
workflow.add_conditional_edges("decision", router, ["action", END])
workflow.add_edge("action", "decision")
app = workflow.compile()

from typing_extensions import TypedDict from langgraph.graph import StateGraph from typing import Annotated from langgraph.managed.base import ManagedValue class IsLastOrSecondToLastStepManager(ManagedValue[bool]): def __call__(self, step: int) -> bool: limit = self.config.get("recursion_limit", 0) return step >= limit - 2 class State(TypedDict): value: str action_result: str is_last_step: Annotated[bool, IsLastOrSecondToLastStepManager] def router(state: State): # 如果处于最后一步，则强制代理结束 if state["is_last_step"]: return END if state["value"] == "end": return END else: return "action" def decision_node(state): return {"value": "keep going!"} def action_node(state: State): # 执行您的操作 ... return {"action_result": "what a great result!"} workflow = StateGraph(State) workflow.add_node("decision", decision_node) workflow.add_node("action", action_node) workflow.add_edge(START, "decision") workflow.add_conditional_edges("decision", router, ["action", END]) workflow.add_edge("action", "decision") app = workflow.compile()

在 [5]

app.invoke({"value": "hi!"})

app.invoke({"value": "hi!"})

Out[5]

{'value': 'keep going!', 'action_result': 'what a great result!'}

完美！我们的代码按预期运行，没有任何错误！

评论