Skip to content

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

先决条件

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

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

环境搭建

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

pip install -U langgraph

为LangGraph开发设置LangSmith

注册LangSmith可以快速发现并解决您的LangGraph项目中的问题，并提高性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。

不返回状态

在这个示例中，我们将定义一个始终会达到递归限制的假图。首先，我们将在不返回状态的情况下实现它，并展示它如何达到递归限制。该图基于ReAct架构，但与其实际做出决策并采取行动不同，它只是无限循环。

API Reference: StateGraph | START | END

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步以内。

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

API Reference: StateGraph

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