[langgraph] 노드 재시도 정책 추가하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/node-retries/

API를 호출하거나 데이터베이스를 쿼리하거나 LLM을 호출하는 등 노드에 사용자 정의 재시도 정책을 설정해야 하는 다양한 사용 사례가 있을 수 있다.

준비

우선, 필요한 패키지를 설치하자.

pip install langgraph langchain-openai langchain-community

재시도 정책을 구성하려면 add_node에 retry 매개변수를 전달해야 한다. retry 매개변수는 RetryPolicy라는 named tuple 객체를 받는다. 아래에서는 기본 매개변수로 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

또한, requests와 httpx와 같은 인기 있는 HTTP 요청 라이브러리에서 발생하는 예외의 경우, 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()

LangGraph 참고 자료

저작자표시

준비

노드에 재시도 정책 전달

LangGraph 참고 자료

티스토리툴바