[langgraph] 서브 그래프(subgraph)의 입/출력값을 변환하는 방법

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

https://langchain-ai.github.io/langgraph/how-tos/subgraph-transform-state/

서브그래프 상태가 부모 그래프 상태와 완전히 독립적일 수 있다. 즉, 두 그래프 간에 중복된 채널(키)이 없을 수 있다. 예를 들어, 여러 개의 ReAct 에이전트를 활용하여 보고서를 작성해야 하는 감독(supervisor) 에이전트가 있을 수 있다. ReAct 에이전트 서브그래프는 메시지 목록을 추적할 수 있지만, 감독자는 사용자 입력과 최종 보고서만 상태에 저장하고 메시지는 추적할 필요가 없다.

이러한 경우 서브그래프를 호출하기 전에 입력을 변환하고, 출력이 반환되기 전에 출력을 변환해야 한다. 이 가이드는 이를 수행하는 방법을 설명한다.

준비

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

pip install langgraph

그래프와 서브그래프 정의

세 개의 그래프를 정의해 보자.

부모 그래프
부모 그래프에 의해 호출되는 자식 서브그래프
자식 그래프에 의해 호출되는 손자 서브그래프

손자 정의

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END


class GrandChildState(TypedDict):
    my_grandchild_key: str


def grandchild_1(state: GrandChildState) -> GrandChildState:
    return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}


grandchild = StateGraph(GrandChildState)
grandchild.add_node("grandchild_1", grandchild_1)

grandchild.add_edge(START, "grandchild_1")
grandchild.add_edge("grandchild_1", END)

grandchild_graph = grandchild.compile()

result = grandchild_graph.invoke({"my_grandchild_key": "hi Bob"})
print(result)

{'my_grandchild_key': 'hi Bob, how are you'}

자식 정의

class ChildState(TypedDict):
    my_child_key: str


def call_grandchild_graph(state: ChildState) -> ChildState:
    # state를 child state channels (`my_child_key`)에서 child state channels (`my_grandchild_key`)로 변환한다.
    grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]}
    # 상태를 grandchild state channels (`my_grandchild_key`)에서 child state channels (`my_child_key`)로 변환한다.
    grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
    return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"}


child = StateGraph(ChildState)
child.add_node("child_1", call_grandchild_graph)
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
child_graph = child.compile()

result = child_graph.invoke({"my_child_key": "hi Bob"})
print(result)

{'my_child_key': 'hi Bob, how are you today?'}

자식과 손자 서브그래프는 부모 그래프와 공유되지 않는 독립적인 상태를 가지고 있다는 점에 유의하자.

부모 정의

class ParentState(TypedDict):
    my_key: str


def parent_1(state: ParentState) -> ParentState:
    return {"my_key": "hi " + state["my_key"]}


def parent_2(state: ParentState) -> ParentState:
    return {"my_key": state["my_key"] + " bye!"}


def call_child_graph(state: ParentState) -> ParentState:
    # 상태를 parent state channels (`my_key`)에서 child state channels (`my_child_key`)로 변환한다.
    child_graph_input = {"my_child_key": state["my_key"]}
    # 상태를 child state channels (`my_child_key`)에서 parent state channels (`my_key`)로 변환한다.
    child_graph_output = child_graph.invoke(child_graph_input)
    return {"my_key": child_graph_output["my_child_key"]}


parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
parent.add_node("child", call_child_graph)
parent.add_node("parent_2", parent_2)

parent.add_edge(START, "parent_1")
parent.add_edge("parent_1", "child")
parent.add_edge("child", "parent_2")
parent.add_edge("parent_2", END)

parent_graph = parent.compile()

부모 그래프를 실행하여 자식 및 손자 서브그래프가 올바르게 호출되는지 확인하다.

result = parent_graph.invoke({"my_key": "Bob"})
print(result)

{'my_key': 'hi Bob, how are you today? bye!'}

부모 그래프가 자식과 손자 서브그래프를 올바르게 호출한다 (원래의 "my_key" 상태 값에 ", how are you"와 "today?"가 추가된 것을 보았기 때문이다).

LangGraph 참고 자료

저작자표시

준비

그래프와 서브그래프 정의

손자 정의

자식 정의

부모 정의

LangGraph 참고 자료

티스토리툴바