langgraph / / 2024. 12. 3. 07:48

[langgraph] 재귀 횟수에 도달하기 전에 상태를 리턴하는 방법

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

https://langchain-ai.github.io/langgraph/how-tos/return-when-recursion-limit-hits/

그래프 재귀 제한을 설정하면 그래프 실행 시간을 제어할 수 있지만, 재귀 제한에 도달하면 그래프가 오류를 반환하게 된다. 모든 사용 사례에서 좋은 방법은 아니다. 대신, 재귀 제한에 도달하기 직전의 상태 값을 반환하도록 설정할 수 있다. 여기서는 이를 구현하는 방법을 보여준다.

준비

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

pip install 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):
    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):
    # agent가 강제로 종료하도록 한다
    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):
    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!'}

코드가 예상한 대로 에러없이 실행되었다.

LangGraph 참고 자료

반응형
  • 네이버 블로그 공유
  • 네이버 밴드 공유
  • 페이스북 공유
  • 카카오스토리 공유