[langgraph] 서브 그래프(subgraph)를 추가하고 사용하는 방법

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

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

서브그래프는 그래프로 구성된 여러 컴포넌트를 가진 복잡한 시스템을 구축할 수 있게 해준다. 서브그래프를 사용하는 일반적인 사례는 다중 에이전트 시스템을 구축하는 것이다.

서브그래프를 추가할 때 가장 중요한 질문은 부모 그래프와 서브그래프가 어떻게 통신하는지, 즉 그래프 실행 중 상태를 어떻게 서로 전달하는지이다. 여기에 두 가지 시나리오가 있다.

1. 부모 그래프와 서브그래프가 스키마 키를 공유하는 경우: 이 경우, 컴파일된 서브그래프를 포함하는 노드를 추가할 수 있다.
2. 부모 그래프와 서브그래프가 서로 다른 스키마를 가진 경우: 이 경우, 서브그래프를 호출하는 노드 함수를 추가해야 한다. 이는 부모 그래프와 서브그래프가 서로 다른 상태 스키마를 가지고 있을 때, 서브그래프를 호출하기 전이나 후에 상태를 변환해야 할 때 유용하다.

아래에서는 각 시나리오에 대해 서브그래프를 추가하는 방법을 보여준다.

준비

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

pip install langgraph langchain-openai langchain-community

서브 그래프를 가진 노드 추가

부모 그래프와 서브그래프가 공유 상태 키(채널)를 통해 통신하는 경우가 일반적이다. 예를 들어, 다중 에이전트 시스템에서는 에이전트들이 종종 공유된 메시지 키를 통해 통신한다.

서브그래프가 부모 그래프와 상태 키를 공유하는 경우, 이를 그래프에 추가하는 방법은 다음과 같다.

1. 서브그래프 워크플로우(subgraph_builder 예제에서)를 정의하고 컴파일한다.
2. 컴파일된 서브그래프를 .add_node 메서드를 사용하여 부모 그래프 워크플로우를 정의할 때 전달한다.

간단한 예제를 살펴보자.

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


class SubgraphState(TypedDict):
    foo: str  # 이 키는 부모 그래프 상태와 공유한다
    bar: str


def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}


def subgraph_node_2(state: SubgraphState):
    # 이 노드는 서브그래프트에서만 가능한 상태키('bar')를 사용하고 있다.
    # 그리고 공유 상태 키('fool')에 수정을 하고 있다.
    return {"foo": state["foo"] + state["bar"]}


subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()


class ParentState(TypedDict):
    foo: str


def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

for chunk in graph.stream({"foo": "foo"}):
    print(chunk)

{'node_1': {'foo': 'hi! foo'}}
{'node_2': {'foo': 'hi! foobar'}}

부모 그래프의 최종 출력에 서브그래프 호출 결과("bar" 문자열)가 포함된 것을 볼 수 있다. 서브그래프에서 출력을 보고 싶다면 스트리밍 시 subgraphs=True를 지정할 수 있다. 서브그래프에서 스트리밍하는 방법에 대한 자세한 내용은 이 가이드에서 확인하자.

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

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})

서브그래프를 실행하는 노드 함수 추가

보다 복잡한 시스템에서는 부모 그래프와 완전히 다른 스키마를 가진 서브그래프를 정의할 수 있다(공유된 키가 없음). 예를 들어, 다중 에이전트 RAG 시스템에서 검색 에이전트는 쿼리와 검색된 문서만 추적하면 된다.

이 경우, 서브그래프를 호출하는 노드 함수를 정의해야 한다. 이 함수는 서브그래프를 호출하기 전에 입력(부모) 상태를 서브그래프 상태로 변환하고, 서브그래프의 결과를 부모 상태로 변환하여 노드에서 상태 업데이트를 반환해야 한다.

아래에서는 원래의 예제를 수정하여 노드 내부에서 서브그래프를 호출하는 방법을 보여준다.

class SubgraphState(TypedDict):
    # 여기 키들 중 어떤것도 부모 상태와 공유하지 않는다.
    bar: str
    baz: str


def subgraph_node_1(state: SubgraphState):
    return {"baz": "baz"}


def subgraph_node_2(state: SubgraphState):
    return {"bar": state["bar"] + state["baz"]}


subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()


class ParentState(TypedDict):
    foo: str


def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}


def node_2(state: ParentState):
    # 서브그래프 상태로 변환한다
    response = subgraph.invoke({"bar": state["foo"]})
    # 부모 상태값으로 변환한다
    return {"foo": response["bar"]}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
# 컴파일된 서브그래프를 사용하지 않고 서브그래프를 호출하는 'node_2' 함수를 사용한다
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

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

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:c47d7ea3-7798-87c4-adf4-2543a91d6891',), {'subgraph_node_1': {'baz': 'baz'}})
(('node_2:c47d7ea3-7798-87c4-adf4-2543a91d6891',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})

LangGraph 참고 자료

• Controllability
  • 병렬 실행 브랜치 생성하는 방법
  • map-reduce 브랜치를 병렬 실행하는 방법
  • 그래프 재귀 한도를 제한하는 방법
• Persistence
  • 그래프에 스레드 수준 영속성 추가하는 방법
  • 서브그래프에 스레드 수준 영속성을 추가하는 방법
  • 그래프에 여러 스레드 간 영속성을 추가하는 방법
  • Postgres 영속성 체크포인터 사용하는 방법
• Memory
  • 대화 이력을 관리하는 방법
  • 메시지 삭제하는 방법
  • 대화 이력을 요약하는 방법
• Human-in-the-loop
  • 중단 지점 추가하는 방법
  • 동적 중단 지점을 추가하는 방법
  • 그래프 상태를 수정하는 방법
  • 사용자 입력을 대기하는 방법
  • 그래프의 과거 상태를 보고 수정하는 방법
  • 도구 호출을 리뷰하는 방법
• Streaming
  • 그래프의 전체 상태를 스트리밍하는 방법
  • 그래프의 상태 업데이트를 스트리밍하는 방법
  • 그래프에서 LLM 토큰을 스트리밍하는 방법
  • LLM 토큰을 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 커스텀 데이터를 스트리밍하는 방법
  • 동시에 여러 스트리밍 모드를 구성하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 최종 노드에서 스트리밍하는 방법
  • 서브그래프에서 스트리밍하는 방법
  • 지원하지 않는 모델에서 스트리밍을 비활성화하는 방법
• Tool calling
  • ToolNode를 사용하여 도구를 호출하는 방법
  • 도구 호출 시 에러 처리하는 방법
  • 도구에 런타임 값을 전달하는 방법
  • 도구에 구성 전달하는 방법
  • 많은 수의 도구를 처리하는 방법
• Subgraphs
  • 서브 그래프를 추가하고 사용하는 방법
  • 서브 그래프에서 상태를 조회하고 업데이트하는 방법
  • 서브 그래프의 입/출력값을 변환하는 방법
• State Management
  • 상태에 pydantic 모델을 사용하는 방법
  • 그래프의 입출력 스키마를 정의하는 방법
  • 노드 간 비공개 상태 전달하는 방법
• Other
  • 그래프를 비동기로 실행하는 방법
  • 그래프를 시각화하는 방법
  • 그래프에 런타임 구성을 하는 방법
  • 노드 재시도 정책 추가하는 방법
  • ReAct 스타일 에이전트로 구조화된 출력을 리턴하는 방법
  • LangSmith에서 그래프 실행 시 실행 ID, 태그와 메타데이터를 전달하는 방법
  • 재귀 횟수에 도달하기 전에 상태를 리턴하는 방법
  • AutoGen, CrewAI 외 다른 프레임워크와 LangGraph 통합하는 방법
• Prebuilt ReAct Agent
  • ReAct 에이전트를 사용하는 방법
  • ReAct 에이전트에 메모리 추가하는 방법
  • ReAct 에이전트에 사용자 정의 시스템 프롬프트 추가하는 방법
  • ReAct 에이전트에 사람이 개입하는 프로세스를 추가하는 방법
  • 처음부터 ReAct 에이전트를 만드는 방법