如何添加节点重试策略¶

先决条件

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

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")

设置 LangSmith 以进行 LangGraph 开发

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

为了配置重试策略，您必须将 retry 参数传递给 add_node。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()

如何添加节点重试策略¶

设置¶

将重试策略传递给节点¶

评论