如何添加节点重试策略¶

先决条件

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

LangGraph 词汇表

有许多用例，您可能希望您的节点具有自定义的重试策略，例如调用 API、查询数据库或调用 LLM 等。

设置¶

首先，让我们安装所需的包并设置我们的 API 密钥

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

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

为LangGraph开发设置LangSmith

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

为了配置重试策略，你需要将 retry 参数传递给添加节点。retry 参数接收一个名为 RetryPolicy 的命名元组对象。下面我们将使用默认参数实例化一个 RetryPolicy 对象：

from langgraph.pregel import RetryPolicy

RetryPolicy()

RetryPolicy(initial_interval=0.5, backoff_factor=2.0, max_interval=128.0, max_attempts=3, jitter=True, retry_on=<function default_retry_on at 0x78b964b89940>)

默认情况下，retry_on 参数使用 default_retry_on 函数，该函数在遇到以下异常时重试，除了以下异常之外：

ValueError
TypeError
ArithmeticError
ImportError
LookupError
NameError
SyntaxError
RuntimeError
ReferenceError
StopIteration
StopAsyncIteration
OSError

此外，对于流行 HTTP 请求库（如 requests 和 httpx）中的异常，仅在遇到 5xx 状态码时才进行重试。

将重试策略传递给节点¶

最后，我们可以在调用add_node函数时传递RetryPolicy对象。在下面的例子中，我们将两种不同的重试策略分别传递给我们的节点：

import operator
import sqlite3
from typing import Annotated, Sequence
from typing_extensions import TypedDict

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage

from langgraph.graph import END, StateGraph, START
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import AIMessage

db = SQLDatabase.from_uri("sqlite:///:memory:")

model = ChatAnthropic(model_name="claude-2.1")


class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]


def query_database(state):
    query_result = db.run("SELECT * FROM Artist LIMIT 10;")
    return {"messages": [AIMessage(content=query_result)]}


def call_model(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}


# Define a new graph
builder = StateGraph(AgentState)
builder.add_node(
    "query_database",
    query_database,
    retry=RetryPolicy(retry_on=sqlite3.OperationalError),
)
builder.add_node("model", call_model, retry=RetryPolicy(max_attempts=5))
builder.add_edge(START, "model")
builder.add_edge("model", "query_database")
builder.add_edge("query_database", END)

graph = builder.compile()

如何添加节点重试策略¶

设置¶

将重试策略传递给节点¶

Comments