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

前提条件

本指南假设你熟悉以下内容：

设置图的递归限制可以帮助你控制图的运行时长，但如果达到了递归限制，图会返回一个错误 —— 这可能并非适用于所有用例。相反，你可能希望返回刚好在达到递归限制*之前*的状态值。本操作指南将向你展示如何实现这一点。

安装设置¶

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

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

为 LangGraph 开发设置 LangSmith

注册 LangSmith，以便快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 — 点击此处了解更多关于如何开始使用的信息。

不返回状态¶

在这个示例中，我们将定义一个虚拟图，它总会达到递归限制。首先，我们将在不返回状态的情况下实现它，并展示它会达到递归限制。这个图基于 ReAct 架构，但它不会实际做出决策和采取行动，只是永远循环下去。

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 IPython.display import Image, display

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

让我们验证一下，我们的图是否总会达到递归限制：

from langgraph.errors import GraphRecursionError

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

Recursion Error

带有返回状态¶

为了避免达到递归限制，我们可以在状态中引入一个名为 remaining_steps 的新键。它将跟踪达到递归限制之前的步数。然后，我们可以检查 remaining_steps 的值，以确定是否应该终止图执行，并将状态返回给用户，而不会引发 RecursionError。

为此，我们将使用特殊的 RemainingSteps 注解。在底层，它会创建一个特殊的 ManagedValue 通道 —— 一个在图运行期间存在且仅在此期间存在的状态通道。

由于我们的 action 节点总是会为我们的图至少额外引入 2 步（因为 action 节点之后总是会调用 decision 节点），我们将使用这个通道来检查我们是否距离递归限制只剩 2 步。

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

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

from langgraph.managed.is_last_step import RemainingSteps


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


def router(state: State):
    # Force the agent to end
    if state["remaining_steps"] <= 2:
        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()

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

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

完美！正如我们所预期的那样，我们的代码运行时没有出现任何错误！

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

安装设置¶

不返回状态¶

带有返回状态¶

Comments