[LangGraph] - Subgraph(서브그래프) 1

LangGraph에서는 하나의 graph를 다른 graph에서 노드로 사용할 수 있다. 모듈화라는 측면에서 보았을 때, 특정 작업이나 기능 단위로 독립적인 graph를 구축하면 이들의 조합으로 하나의 큰 작업을 처리하는 parent graph를 생성 가능하다. 이처럼 subgraph를 사용한다면 다음과 같은 장점이 있다.

1. 멀티 에이전트 시스템 구축: 각 에이전트 subgraph가 독립적으로 작동하면서 parent graph에서 유기적으로 결합
2. 노드 재사용: 여러 graph에서 동일한 작업이 필요한 경우, 이를 subgraph로 정의해 두면 한 번 작성한 subgraph를 여러 parent graph에서 쉽게 재사용 가능.
3. 독립적인 작업 분리: 서로 다른 사람이 graph의 각 부분을 독립적으로 작업해야 할 때, 입출력 스키마만 준수한다면 graph를 subgraph로 분리하면 효율적인 협업 가능.

Subgraph를 추가하는 방법

Subgraph를 parent graph에 추가하는 방법은 2가지가 있다.

1. 동일한 State key를 공유하는 경우
   
   이 경우에는 이미 compile 된 subgraph를 node로 추가할 수 있다.

subgraph = sub_flow.compile()
parent_graph = parent_flow.compile("subgraph", subgraph)

2. 서로 다른 State 스키마를 사용하는 경우
   
   일반적으로는 graph를 정의할 때 하나의 독립적인 작업을 수행하도록 작성하기 때문에 이러한 경우가 많다. 이처럼 parent graph와 subgraph 간 이 경우에는 state를 변환하는 함수를 작성해서 하나의 node로 추가하면 된다.
   
   여기서 "state를 변환하는 함수"라는 건 생각보다 간단하다. 단순히 node의 로직 내에서 parent graph의 state key를 subgraph의 state key와 맞춰주는 것을 의미한다.

subgraph = sub_flow.compile()

# subgraph 호출 노드 정의
def call_subgraph(state: State):
    parent_key = state["parent_key"]
    return subgraph.invoke({"subgraph_key": parent_key})

 

각 방법에 대해서 조금 코드를 통해 좀 더 자세히 알아보자!

1. 동일한 State key를 공유하는 경우: subgraph를 node로 직접 추가

Parent graph와 subgraph가 동일한 상태 키(shared state key)를 갖는 상황에는 컴파일된 subgraph를 parent graph에 추가한다.

여기서는 새해 인사를 생성하는 subgraph를 parent graph에 통합해 보자. subgraph는 parent state에서 전달된 name key를 기반으로 greeting 메시지를 생성하며, parent graph와 subgraph는 공유된 state key로 name과 greeting을 통해 데이터를 주고받는다.

• Parent graph와 subgraph가 name key를 공유하며, greeting key는 subgraph에서 최종 업데이트
• Parent graph는 node1을 통해 name key를 업데이트하고, 이를 subgraph에 전달
• subgraph의 두 번째 노드(subgraph_node2)에서 공유된 name을 활용해 greeting 메시지를 생성

즉, 여기서 parent graph는 subgraph에 전달되는 name key를 전처리하고 subgraph를 호출하는 역할을 한다.

from typing import TypedDict
from datetime import datetime
from langgraph.graph import StateGraph, START, END


# subgraph 정의
class SubgraphState(TypedDict):
    # name과 greeting key는 parent state와 공유됨
    name: str
    greeting: str
    year: int


def subgraph_node1(state: SubgraphState):
    return {"year": datetime.now().year}


def subgraph_node2(state: SubgraphState):
    # year key는 subgraph에서만 사용 가능
    # 공유되는 name key를 받아 greeting key에 업데이트
    return {"greeting": state["name"] + f"님, {state['year']}년 새해 복 많이 받으세요!"}


subgraph_flow = StateGraph(SubgraphState)

subgraph_flow.add_node("subgraph_node1", subgraph_node1)
subgraph_flow.add_node("subgraph_node2", subgraph_node2)

subgraph_flow.add_edge(START, "subgraph_node1")
subgraph_flow.add_edge("subgraph_node1", "subgraph_node2")
subgraph_flow.add_edge("subgraph_node2", END)
subgraph = subgraph_flow.compile()

 

 

 

이번에는 subgraph를 포함하여 parent graph를 정의하자.

class ParentState(TypedDict):
    # subgraph와 공유하는 key
    name: str
    greeting: str

def node1(state: ParentState):
    return {"name":  f"멋진 {state['name']}"}


flow = StateGraph(ParentState)

flow.add_node("node1", node1)
flow.add_node("node2", subgraph)

flow.add_edge(START, "node1")
flow.add_edge("node1", "node2")
flow.add_edge("node2", END)
graph = flow.compile()

 

 

 

의도한 대로 작동하는지 확인해 보자.

1. input으로 name을 전달하면 node1에서 멋진 이라는 수식어를 붙여주고 다시 name key에 저장
2. subgraph로 넘어가서, subgraph_node1에서 year key 업데이트
3. subgraph_node2에서 새해 인사말을 만들어 greeting key에 저장

graph.invoke({"name": "Sean"})

{'name': '멋진 Sean', 'greeting': '멋진 Sean님, 2025년 새해 복 많이 받으세요!'}

 

subgraphs=True 옵션을 사용하면, subgraph 내부의 출력도 확인할 수 있다.

for chunk in graph.stream({"name": "Sean"}, subgraphs=True):
    print(chunk)

((), {'node1': {'name': '멋진 Sean'}})
(('node2:1e1f3549-51f3-03dd-0261-520eaadf54b7',), {'subgraph_node1': {'year': 2025}})
(('node2:1e1f3549-51f3-03dd-0261-520eaadf54b7',), {'subgraph_node2': {'greeting': '멋진 Sean님, 2025년 새해 복 많이 받으세요!'}})
((), {'node2': {'name': '멋진 Sean', 'greeting': '멋진 Sean님, 2025년 새해 복 많이 받으세요!'}})

2. 서로 다른 State 스키마인 경우: subgraph를 호출하는 node 추가

parent graph와 호출하고자 하는 subgraphs 간 전혀 다른 state 스키마인 경우(공유하는 key가 없음), subgraph를 호출하는 node를 정의하면 된다. 즉, 해당 노드에서 subgraph를 호출하기 전에 parent graph의 state로부터 값을 꺼내오고 해당 값을 subgraph 호출 규약에 맞게 변경해서 호출하여 결과를 다시 parent graph state의 key에 저장하도록 하자.

 

말로는 복잡하지만 실제 코드로 보면 당연한 로직이라 이해가 더 쉽다. 한 번 봐보자.

하나의 노드에서 2개 이상의 subgraph를 호출하는 건 불가능
단, 하나의 노드에서 1개의 subgraph를 호출하고 이를 감싼 노드에서 subgraph를 호출하는 건 가능!

 

이번에는 subgraph는 동일한 graph를 사용한다. 하지만 parent graph에서 subgraph와 공유하는 key가 없는 상황이다.

따라서, parent state를 subgraph state로 변환하고, subgraph 결과를 다시 parent state로 변환하는 작업을 수행한다.

1. node1
   • parent state에서 first_name과 last_name key로 subgraph state가 받을 수 있게 전처리 후 last_name 키에 업데이트
2. node2
   • subgraph를 호출하여, parent state의 last_name 값을 subgraph의 name 입력으로 전달
   • subgraph 호출 결과에서 greeting 값을 받아와 parent state의 message key를 업데이트

class ParentState(TypedDict):
    # subgraph와 공유하는 key가 없음!
    first_name: str
    last_name: str
    message: str

def node1(state: ParentState):
    name = state["last_name"] + state["first_name"]
    return {"last_name": name}


def node2(state: ParentState):
    response = subgraph.invoke({"name": state["last_name"]})
    return {"message": response["greeting"]}


flow = StateGraph(ParentState)

flow.add_node("node1", node1)
flow.add_node("node2", node2)

flow.add_edge(START, "node1")
flow.add_edge("node1", "node2")
flow.add_edge("node2", END)
graph = flow.compile()

 

 

 

graph.invoke({"first_name": "길동", "last_name": "홍"})

{'first_name': '길동', 
 'last_name': '홍길동', 
 'message': '홍길동님, 2025년 새해 복 많이 받으세요!'}

중첩된 Subgraph

 

위의 예처럼 Parent graph와 subgraph가 서로 다른 state 스키마를 가진 경우, subgraph를 호출하는 node를 정의하고 parent state의 key를 사용해 subgraph를 호출할 수 있는 형태로 변환하는 로직이 필요하다.

 

예를 들어, 보고서를 작성하는 graph를 만드려고 한다고 가정해 보자. 주제에 대해 조사하는 여러 ReAct 에이전트가 있고, 이를 관리하는 관리자 에이전트가 있는 경우를 생각해 보자. 이때, ReAct 에이전트들은 메시지 목록을 추적해야 하지만, 관리자는 사용자 입력과 최종 보고서만 필요로 할 수 있다.

 

이런 상황에서는 subgraph로 데이터를 전달하기 전에 사용자 input을 변환하고, 결과를 받은 후에도 출력을 변환해야 한다.

여기서는 3개의 그래프를 만들고 부모 그래프의 노드에서 (자식 그래프 -> 손자 그래프)를 중첩해서 호출해 보자:

1. 부모 그래프 (Parent)
2. 자식 그래프 (Child)
3. 손자 그래프 (Grandchild)

각 그래프는 자신만의 독립적인 상태를 가지며, 다른 그래프의 상태에 직접 접근할 수 없다.

grandchild graph 정의

# define grandcild graph
from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class GrandChildState(TypedDict):
    grandchild_key: str


def grandchild_node(state: GrandChildState) -> GrandChildState:
    # child key와 parent key는 접근할 수 없는 노드
    return {"grandchild_key": state["grandchild_key"] + "님, 안녕하세요!"}


grandchild_flow = StateGraph(GrandChildState)
grandchild_flow.add_node("grandchild_node", grandchild_node)
grandchild_flow.add_edge(START, "grandchild_node")
grandchild_flow.add_edge("grandchild_node", END)

grandchild_graph = grandchild_flow.compile()

grandchild_graph.invoke({"grandchild_key": "Sean"})

{'grandchild_key': 'Sean님, 안녕하세요!'}

child graph 정의

이번에는 child_graph를 정의해 보자. child_graph의 child_node 는 내부적으로 입력받은 child_key를 grandchild_graph가 입력받을 수 있도록 변환한 뒤, 받은 결과의 grandchild_key에 ", 오늘 기분은 어때요?"를 덧붙여 child_key에 다시 저장한다.

# define child graph
class ChildState(TypedDict):
    child_key: str


def call_grandchild_graph(state: ChildState) -> ChildState:
    """grandchild_graph 호출"""
    # 입력 변환
    grandchild_input = {"grandchild_key": state["child_key"]}
    # 출력 변환
    grandchild_output = grandchild_graph.invoke(grandchild_input)
    return {"child_key": grandchild_output["grandchild_key"] + ", 오늘 기분은 어때요?"}


child_flow = StateGraph(ChildState)
child_flow.add_node("child_node", call_grandchild_graph)
child_flow.add_edge(START, "child_node")
child_flow.add_edge("child_node", END)

child_graph = child_flow.compile()

child_graph.invoke({"child_key": "Sean"})

{'child_key': 'Sean님, 안녕하세요!, 오늘 기분은 어때요?'}

parent graph 정의

이번에는 parent_graph를 정의해 보자.

• parent_node1: 입력받은 parent_key 앞에 수식어를 더해주는 노드
• child 노드: child_node 는 입력받은 parent_key를 child_graph가 입력받을 수 있도록 변환한 뒤, 받은 결과의 변환하여 저장한다. 내부적으로 child_graph에서는 다시 입력받은 key를 grandchild_graph가 입력받을 수 있는 형태로 가공한다.
• parent_node2: parent_key 뒤에 수식어를 더해주는 노드

# define parent graph
class ParentState(TypedDict):
    parent_key: str


def parent_node1(state: ParentState) -> ParentState:
    return {"parent_key": "행복한 " + state["parent_key"]}


def call_child_graph(state: ParentState) -> ParentState:
    """child_graph 호출"""
    # 입력 변환
    child_input = {"child_key": state["parent_key"]}
    # 출력 변환
    child_output = child_graph.invoke(child_input)
    return {"parent_key": child_output["child_key"]}


def parent_node2(state: ParentState) -> ParentState:
    return {"parent_key": state["parent_key"] + " 좋은 하루 보내세요🔥"}


flow = StateGraph(ParentState)
flow.add_node("parent_node1", parent_node1)
flow.add_node("child", call_child_graph)
flow.add_node("parent_node2", parent_node2)

flow.add_edge(START, "parent_node1")
flow.add_edge("parent_node1", "child")
flow.add_edge("child", "parent_node2")
flow.add_edge("parent_node2", END)

graph = flow.compile()

 

 

 

이제 의도한 대로 실행되는지 그래프를 호출해 보자.

graph.invoke({"parent_key": "Sean"})

{'parent_key': '행복한 Sean님, 안녕하세요!, 오늘 기분은 어때요? 좋은 하루 보내세요🔥'}

 

---

LangGraph Glossary. https://langchain-ai.github.io/langgraph/concepts/low_level/#subgraphs

LangGraph. "How to add and use subgraphs". https://langchain-ai.github.io/langgraph/how-tos/subgraph

LangGraph. "How to transform inputs and outputs of a subgraph". https://langchain-ai.github.io/langgraph/how-tos/subgraph-transform-state