LangGraph 서브그래프 완벽 가이드

LangGraph의 서브그래프를 활용하여 복잡한 워크플로우를 모듈화하고 재사용 가능한 그래프 구조를 만드는 방법을 배웁니다. 노드에서 그래프 호출, 상태 공유, 체크포인터, 스트리밍까지 실무에 필요한 모든 내용을 다룹니다.

1. 서브그래프 개념과 용도

어느 날 김개발 씨는 LangGraph로 복잡한 AI 워크플로우를 구축하던 중 문제에 부딪혔습니다. 그래프가 점점 커지면서 코드가 스파게티처럼 엉켜버린 것입니다.

선배 박시니어 씨가 다가와 물었습니다. "이 부분, 서브그래프로 분리하면 어떨까요?"

서브그래프는 LangGraph에서 그래프 안에 또 다른 그래프를 포함시키는 기법입니다. 마치 프로그래밍에서 함수를 나눠서 사용하듯이, 복잡한 워크플로우를 작은 단위로 분리할 수 있습니다.

이를 통해 코드의 재사용성을 높이고 유지보수를 쉽게 만들 수 있습니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    messages: list
    count: int

# 서브그래프 정의
def create_subgraph():
    subgraph = StateGraph(State)
    subgraph.add_node("process", lambda s: {"count": s["count"] + 1})
    subgraph.add_edge(START, "process")
    subgraph.add_edge("process", END)
    return subgraph.compile()

# 메인 그래프에서 서브그래프 활용
main_graph = StateGraph(State)
main_graph.add_node("sub_task", create_subgraph())
main_graph.add_edge(START, "sub_task")
main_graph.add_edge("sub_task", END)

김개발 씨는 입사 6개월 차 AI 엔지니어입니다. 요즘 LangGraph로 고객 응대 시스템을 구축하고 있는데, 그래프가 점점 복잡해지면서 코드를 읽기가 어려워졌습니다.

노드가 수십 개나 되고, 엣지 연결도 복잡하게 얽혀 있습니다. 선배 개발자 박시니어 씨가 김개발 씨의 코드를 살펴보더니 말했습니다.

"이 부분은 서브그래프로 분리하는 게 좋겠네요. 지금은 모든 로직이 한 그래프에 몰려 있어서 관리하기 힘들어요." 서브그래프란 무엇일까요? 쉽게 비유하자면, 서브그래프는 마치 레고 블록과 같습니다.

큰 레고 작품을 만들 때 작은 부품들을 먼저 조립한 후, 그것들을 다시 결합하여 완성품을 만들듯이 말입니다. 서브그래프도 작은 그래프들을 먼저 만들고, 이를 큰 그래프에 조립하여 완성된 워크플로우를 구성합니다.

프로그래밍에 익숙한 분들이라면 함수의 개념을 떠올리면 됩니다. 긴 코드를 여러 함수로 나누듯이, 복잡한 그래프를 여러 서브그래프로 나누는 것입니다.

왜 서브그래프가 필요할까요? 서브그래프가 없던 시절을 상상해 봅시다. 모든 로직을 하나의 거대한 그래프에 담아야 했습니다.

노드가 50개, 100개로 늘어나면서 어떤 노드가 무슨 일을 하는지 파악하기 어려워졌습니다. 더 큰 문제는 코드 재사용이었습니다.

비슷한 처리 과정이 여러 곳에서 필요할 때마다 똑같은 노드들을 복사해서 붙여넣어야 했습니다. 당연히 버그가 생기면 모든 곳을 찾아서 수정해야 했고, 이는 엄청난 시간 낭비였습니다.

서브그래프가 가져온 변화 바로 이런 문제를 해결하기 위해 서브그래프가 등장했습니다. 서브그래프를 사용하면 모듈화가 가능해집니다.

특정 기능을 담당하는 부분을 독립적인 서브그래프로 만들어 놓으면, 필요할 때마다 가져다 쓸 수 있습니다. 또한 재사용성도 크게 향상됩니다.

한 번 만든 서브그래프를 여러 프로젝트에서 사용할 수 있습니다. 무엇보다 유지보수가 쉬워진다는 큰 이점이 있습니다.

특정 기능에 버그가 있다면 해당 서브그래프만 수정하면 되고, 새로운 기능을 추가할 때도 독립적인 서브그래프를 추가하면 됩니다. 코드 분석 위의 코드를 단계별로 살펴보겠습니다.

먼저 create_subgraph 함수에서 독립적인 서브그래프를 생성합니다. 이 서브그래프는 process라는 노드 하나를 가지고 있으며, 상태의 count 값을 1 증가시키는 간단한 작업을 수행합니다.

서브그래프도 일반 그래프처럼 START와 END를 가지며, 엣지로 연결됩니다. 마지막에 compile() 메서드를 호출하여 실행 가능한 형태로 만듭니다.

메인 그래프에서는 이렇게 만들어진 서브그래프를 일반 노드처럼 추가합니다. add_node("sub_task", create_subgraph())를 보면 서브그래프 자체가 하나의 노드로 취급되는 것을 알 수 있습니다.

실무 활용 사례 실제 현업에서는 어떻게 활용할까요? 예를 들어 고객 문의 처리 시스템을 개발한다고 가정해봅시다.

감정 분석, 의도 파악, 답변 생성, 품질 검증 등 여러 단계가 필요합니다. 각 단계를 독립적인 서브그래프로 만들어 두면, 다른 프로젝트에서도 재사용할 수 있습니다.

예를 들어 감정 분석 서브그래프는 이메일 분류 시스템에서도 그대로 사용할 수 있습니다. 주의사항 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 지나치게 작은 단위로 서브그래프를 나누는 것입니다. 노드 하나짜리 서브그래프를 수십 개 만들면 오히려 복잡도가 증가합니다.

적절한 크기로 의미 있는 단위로 나누는 것이 중요합니다. 또한 서브그래프 간의 상태 공유를 잘못 설계하면 예상치 못한 버그가 발생할 수 있습니다.

상태 스키마를 명확히 정의하고, 어떤 데이터가 공유되는지 문서화해야 합니다. 정리 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 조언에 따라 코드를 서브그래프로 분리한 김개발 씨는 감탄했습니다. "코드가 훨씬 깔끔해졌어요!

이제 어디서 무슨 일이 일어나는지 한눈에 보이네요." 서브그래프를 제대로 활용하면 복잡한 AI 워크플로우도 체계적으로 관리할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - 의미 있는 단위로 서브그래프를 분리하세요. 너무 작게 나누면 오히려 복잡해집니다.

서브그래프의 이름은 그 역할을 명확히 나타내도록 짓습니다.
재사용 가능한 서브그래프는 별도 파일로 분리하여 관리하세요.

2. 노드에서 그래프 호출

김개발 씨는 서브그래프의 개념을 이해했지만, 실제로 노드 안에서 어떻게 그래프를 호출하는지 궁금했습니다. 박시니어 씨가 화면을 가리키며 말했습니다.

"일반 함수처럼 호출하면 되는데, 몇 가지 알아야 할 게 있어요."

노드에서 그래프 호출은 그래프를 컴파일한 후 일반 함수처럼 실행하는 방식입니다. 컴파일된 그래프는 상태를 입력받아 처리 후 결과를 반환합니다.

이를 통해 복잡한 로직을 캡슐화하고, 필요할 때마다 재사용할 수 있습니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    input_text: str
    result: str

# 서브그래프 생성
def create_processor():
    graph = StateGraph(State)

    def process(state):
        # 텍스트 처리 로직
        return {"result": state["input_text"].upper()}

    graph.add_node("transform", process)
    graph.add_edge(START, "transform")
    graph.add_edge("transform", END)
    return graph.compile()

# 노드 함수에서 그래프 호출
processor = create_processor()

def node_function(state):
    # 서브그래프를 함수처럼 호출
    output = processor.invoke({"input_text": state["input_text"]})
    return {"result": output["result"]}

김개발 씨는 서브그래프의 강력함에 매료되었습니다. 하지만 막상 코드를 작성하려니 막막했습니다.

"서브그래프를 만들었는데, 이걸 어떻게 사용하죠?" 박시니어 씨가 웃으며 대답했습니다. "생각보다 간단해요.

일반 함수 호출하듯이 하면 됩니다." 그래프를 함수처럼 사용하기 서브그래프를 노드에서 호출하는 것은 마치 파이썬의 함수를 호출하는 것과 비슷합니다. 함수를 정의한 후 필요할 때 호출하듯이, 그래프를 컴파일한 후 필요할 때 실행하면 됩니다.

핵심은 compile() 메서드입니다. 이 메서드는 그래프 정의를 실행 가능한 형태로 변환합니다.

컴파일된 그래프는 invoke, stream, batch 같은 메서드를 제공하여 다양한 방식으로 실행할 수 있습니다. 컴파일 이전과 이후의 차이 그래프를 정의만 한 상태에서는 실제로 아무 작업도 수행되지 않습니다.

이는 마치 요리 레시피를 작성한 것과 같습니다. 레시피만 있다고 요리가 완성되지는 않습니다.

compile() 메서드를 호출하면 그래프가 실행 가능한 객체로 변환됩니다. 이제 실제로 데이터를 넣고 처리 결과를 받을 수 있습니다.

레시피대로 실제 요리를 하는 것과 같습니다. 컴파일 후에는 그래프 구조를 변경할 수 없습니다.

노드를 추가하거나 엣지를 수정하려면 다시 정의부터 시작해야 합니다. 이는 안정성을 위한 설계입니다.

invoke 메서드의 동작 방식 invoke 메서드는 가장 기본적인 실행 방법입니다. 상태 딕셔너리를 입력으로 받아, 그래프의 모든 노드를 순서대로 실행하고, 최종 상태를 반환합니다.

이 과정은 동기적으로 진행되므로 결과가 나올 때까지 기다려야 합니다. 코드에서 processor.invoke({"input_text": state["input_text"]})를 보면, 입력 상태를 전달하고 처리 결과를 받는 것을 알 수 있습니다.

반환되는 값은 최종 상태 전체입니다. 실무에서의 패턴 실제 프로젝트에서는 서브그래프를 미리 컴파일해서 전역 변수나 클래스 속성으로 저장해 둡니다.

예를 들어 텍스트 감정 분석 서브그래프를 만들었다면, 애플리케이션 시작 시 한 번만 컴파일하고 계속 재사용합니다. 매번 컴파일하면 성능이 떨어지기 때문입니다.

이커머스 추천 시스템을 생각해 봅시다. 사용자 행동 분석, 상품 필터링, 순위 계산 등 각 단계를 서브그래프로 만들어 놓습니다.

메인 그래프의 노드에서는 필요한 서브그래프를 호출하여 결과를 조합합니다. 상태 전달의 중요성 노드에서 그래프를 호출할 때 가장 중요한 것은 상태 전달입니다.

서브그래프가 기대하는 상태 스키마와 실제로 전달하는 상태가 일치해야 합니다. 필요한 키가 누락되면 에러가 발생합니다.

타입도 맞아야 합니다. TypedDict를 활용하면 타입 체크를 통해 실수를 방지할 수 있습니다.

반환값도 마찬가지입니다. 서브그래프가 어떤 형태의 상태를 반환하는지 정확히 알고 있어야, 이후 처리를 올바르게 할 수 있습니다.

에러 처리 서브그래프 호출 시 에러가 발생할 수 있습니다. 네트워크 호출이나 외부 API를 사용하는 노드가 있다면 타임아웃이나 연결 실패가 일어날 수 있습니다.

적절한 try-except 블록으로 에러를 처리하고, 기본값을 반환하거나 재시도 로직을 구현해야 합니다. 정리 김개발 씨는 실제로 코드를 작성해 보며 이해했습니다.

"아, 그래프를 컴파일해서 함수처럼 쓰면 되는 거네요!" 박시니어 씨가 고개를 끄덕였습니다. "맞아요.

한 번 익숙해지면 정말 강력한 도구가 될 거예요." 노드에서 그래프를 호출하는 방식을 이해하면 복잡한 워크플로우도 체계적으로 구성할 수 있습니다. 각 부분을 독립적으로 개발하고 테스트한 후 조합하는 방식은 대규모 시스템 개발에 필수적입니다.

실전 팁

💡 - 서브그래프는 애플리케이션 시작 시 한 번만 컴파일하고 재사용하세요.

상태 스키마를 TypedDict로 명확히 정의하여 타입 안정성을 확보하세요.
서브그래프 호출 시 에러 처리를 잊지 마세요.

3. 그래프를 노드로 추가

"그런데 매번 노드 함수 안에서 invoke를 호출하는 게 번거롭지 않나요?" 김개발 씨가 물었습니다. 박시니어 씨가 미소 지으며 답했습니다.

"더 간단한 방법이 있어요. 그래프 자체를 노드로 바로 추가할 수 있거든요."

그래프를 노드로 추가하는 방식은 컴파일된 그래프를 add_node 메서드에 직접 전달하는 것입니다. 별도의 래퍼 함수 없이 서브그래프를 노드로 사용할 수 있어 코드가 간결해집니다.

LangGraph가 자동으로 상태를 전달하고 결과를 병합합니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    messages: list
    count: int
    processed: bool

# 서브그래프 1: 카운터
def create_counter_graph():
    graph = StateGraph(State)
    graph.add_node("increment", lambda s: {"count": s.get("count", 0) + 1})
    graph.add_edge(START, "increment")
    graph.add_edge("increment", END)
    return graph.compile()

# 서브그래프 2: 플래그 설정
def create_flag_graph():
    graph = StateGraph(State)
    graph.add_node("set_flag", lambda s: {"processed": True})
    graph.add_edge(START, "set_flag")
    graph.add_edge("set_flag", END)
    return graph.compile()

# 메인 그래프에 서브그래프를 노드로 추가
main = StateGraph(State)
main.add_node("counter", create_counter_graph())  # 직접 추가
main.add_node("flagger", create_flag_graph())     # 직접 추가
main.add_edge(START, "counter")
main.add_edge("counter", "flagger")
main.add_edge("flagger", END)

app = main.compile()

김개발 씨는 앞에서 배운 방식대로 코드를 작성했습니다. 노드 함수를 만들고, 그 안에서 서브그래프의 invoke를 호출하는 방식이었습니다.

동작은 잘 되었지만, 뭔가 코드가 장황하다는 느낌을 지울 수 없었습니다. 박시니어 씨가 김개발 씨의 화면을 보더니 말했습니다.

"이렇게 단순히 서브그래프를 호출만 하는 노드라면, 굳이 함수로 감쌀 필요가 없어요." 직접 추가의 편리함 그래프를 노드로 직접 추가하는 방식은 마치 플러그 앤 플레이와 같습니다. USB 기기를 컴퓨터에 꽂으면 바로 작동하듯이, 컴파일된 그래프를 add_node에 넣으면 바로 사용할 수 있습니다.

이전 방식에서는 매번 래퍼 함수를 만들어야 했습니다. 함수를 정의하고, 서브그래프를 호출하고, 결과를 반환하는 코드를 반복해서 작성했습니다.

서브그래프가 10개라면 래퍼 함수도 10개를 만들어야 했습니다. 하지만 직접 추가 방식을 사용하면 이런 보일러플레이트 코드가 사라집니다.

컴파일된 그래프를 바로 노드로 등록하면 끝입니다. LangGraph의 자동 처리 어떻게 이게 가능할까요?

LangGraph는 add_node에 전달된 객체의 타입을 확인합니다. 컴파일된 그래프 객체라면, 내부적으로 자동으로 invoke를 호출하는 래퍼를 생성합니다.

개발자는 이 과정을 신경 쓸 필요가 없습니다. 상태 전달도 자동으로 처리됩니다.

현재 그래프의 상태가 서브그래프로 전달되고, 서브그래프가 반환한 상태는 현재 상태와 병합됩니다. 이 과정에서 같은 키가 있다면 서브그래프의 값으로 덮어씁니다.

상태 병합의 메커니즘 상태 병합은 중요한 개념입니다. 메인 그래프의 상태가 {"count": 5, "messages": ["hello"]}라고 가정해 봅시다.

서브그래프가 {"count": 6}을 반환하면, 최종 상태는 {"count": 6, "messages": ["hello"]}가 됩니다. count는 업데이트되고, messages는 그대로 유지됩니다.

이는 딕셔너리의 update 메서드와 유사합니다. 기존 값은 유지하되, 새로운 값으로 덮어쓰는 방식입니다.

실무 활용 사례 실제 프로젝트에서 이 패턴은 매우 유용합니다. 문서 처리 파이프라인을 생각해 봅시다.

PDF 파싱, 텍스트 추출, 요약, 키워드 추출 등 각 단계를 독립적인 서브그래프로 만듭니다. 메인 그래프에서는 이들을 순서대로 노드로 추가하기만 하면 됩니다.

각 서브그래프는 독립적으로 개발하고 테스트할 수 있습니다. PDF 파싱 서브그래프에 버그가 있다면, 그 부분만 수정하면 됩니다.

나머지 파이프라인은 영향을 받지 않습니다. 코드의 간결함 위의 예제 코드를 보면 메인 그래프 구성이 매우 간결합니다.

main.add_node("counter", create_counter_graph())처럼 한 줄로 서브그래프를 추가합니다. 별도의 함수 정의가 필요 없습니다.

코드를 읽는 사람도 "counter 노드는 counter 그래프를 실행하는구나"라고 직관적으로 이해할 수 있습니다. 엣지 연결도 일반 노드와 동일합니다.

add_edge("counter", "flagger")처럼 서브그래프 간 연결도 자연스럽게 표현됩니다. 언제 사용할까 모든 경우에 직접 추가 방식을 사용하는 것은 아닙니다.

서브그래프를 호출하기 전에 전처리가 필요하거나, 호출 후 후처리가 필요하다면 여전히 래퍼 함수를 사용해야 합니다. 예를 들어 입력 형식을 변환하거나, 결과를 검증하는 로직이 있다면 함수로 감싸는 게 맞습니다.

하지만 단순히 서브그래프를 실행하기만 한다면, 직접 추가 방식이 훨씬 깔끔합니다. 정리 김개발 씨는 코드를 수정해 보았습니다.

래퍼 함수들을 모두 제거하고 서브그래프를 직접 추가했습니다. "와, 코드가 절반으로 줄었어요!" 박시니어 씨가 만족스러운 표정으로 말했습니다.

"간결한 코드가 좋은 코드예요." 그래프를 노드로 직접 추가하는 방식을 익히면 코드가 훨씬 읽기 쉬워집니다. 보일러플레이트를 줄이고 핵심 로직에 집중할 수 있습니다.

실전 팁

💡 - 단순히 서브그래프만 실행한다면 직접 추가 방식을 사용하세요.

전처리나 후처리가 필요하다면 래퍼 함수를 사용하세요.
노드 이름은 서브그래프의 역할을 명확히 나타내도록 짓습니다.

4. 상태 공유 메커니즘

서브그래프를 사용하다 보니 김개발 씨는 새로운 의문이 생겼습니다. "메인 그래프와 서브그래프가 같은 상태를 공유하는 건가요?

아니면 복사본을 사용하는 건가요?" 박시니어 씨가 화이트보드를 꺼내들며 설명을 시작했습니다.

LangGraph의 상태 공유 메커니즘은 서브그래프가 메인 그래프의 상태를 직접 수정하는 것이 아니라, 상태 업데이트를 반환하여 병합하는 방식입니다. 각 서브그래프는 독립적으로 동작하며, 반환된 상태만 메인 그래프에 반영됩니다.

이를 통해 예측 가능하고 안전한 상태 관리가 가능합니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    # Annotated로 리스트 병합 방식 지정
    messages: Annotated[list, add]
    count: int
    metadata: dict

def create_message_adder():
    graph = StateGraph(State)

    def add_message(state):
        # 새 메시지 추가 (기존 메시지와 병합됨)
        return {"messages": ["서브그래프에서 추가"]}

    graph.add_node("add", add_message)
    graph.add_edge(START, "add")
    graph.add_edge("add", END)
    return graph.compile()

# 메인 그래프
main = StateGraph(State)
main.add_node("sub", create_message_adder())
main.add_edge(START, "sub")
main.add_edge("sub", END)

# 실행
app = main.compile()
result = app.invoke({"messages": ["초기 메시지"], "count": 0, "metadata": {}})
# result["messages"]는 ["초기 메시지", "서브그래프에서 추가"]

김개발 씨는 코드를 작성하다가 이상한 현상을 발견했습니다. 서브그래프에서 상태를 변경했는데, 메인 그래프의 상태도 함께 변경된 것입니다.

"이게 맞는 건가요? 아니면 버그인가요?" 박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다.

상태는 어떻게 전달될까 LangGraph의 상태 관리는 함수형 프로그래밍의 불변성 원칙을 따릅니다. 각 노드는 현재 상태를 입력으로 받습니다.

하지만 이 상태를 직접 수정하지 않습니다. 대신 새로운 상태 업데이트를 반환합니다.

LangGraph는 이 업데이트를 기존 상태와 병합하여 다음 상태를 만듭니다. 이는 마치 Git의 커밋과 비슷합니다.

이전 커밋을 직접 수정하는 게 아니라, 변경 사항을 새 커밋으로 만들어 쌓아가는 방식입니다. 각 단계마다 상태의 스냅샷이 생성되고, 히스토리를 추적할 수 있습니다.

기본 병합 규칙 기본적으로 LangGraph는 딕셔너리 병합 방식을 사용합니다. 서브그래프가 {"count": 10}을 반환하면, 메인 상태의 count 키만 업데이트됩니다.

다른 키들은 그대로 유지됩니다. 이는 파이썬의 dict.update()와 동일한 동작입니다.

하지만 리스트나 집합 같은 컬렉션 타입은 어떻게 될까요? 기본적으로는 덮어쓰기가 됩니다.

{"messages": ["new"]}를 반환하면 기존 메시지들이 모두 사라지고 새 메시지로 교체됩니다. Annotated를 활용한 병합 제어 이런 기본 동작을 바꾸고 싶다면 Annotated 타입을 사용합니다.

코드에서 Annotated[list, add]를 보세요. 이는 "리스트 타입이며, 병합 시 add 연산자를 사용하라"는 의미입니다.

operator.add는 두 리스트를 합치는 연산자입니다. 이제 서브그래프가 {"messages": ["new"]}를 반환하면, 기존 메시지들 뒤에 추가됩니다.

["old"] + ["new"] = ["old", "new"]가 되는 것입니다. 이 방식은 채팅 메시지, 로그, 이벤트 히스토리 등 누적되어야 하는 데이터에 매우 유용합니다.

커스텀 병합 로직 add 외에도 다양한 병합 방식을 지정할 수 있습니다. 직접 함수를 만들어 전달할 수도 있습니다.

예를 들어 중복을 제거하며 병합하고 싶다면, 커스텀 함수를 작성하여 Annotated[list, custom_merge]로 지정합니다. python def unique_merge(existing, new): return list(set(existing + new)) class State(TypedDict): tags: Annotated[list, unique_merge] 이렇게 하면 태그 같은 데이터를 중복 없이 관리할 수 있습니다.

실무에서의 패턴 실제 프로젝트에서는 상태 스키마 설계가 매우 중요합니다. 채팅봇을 만든다고 가정해 봅시다.

메시지 히스토리는 누적되어야 하므로 Annotated[list, add]를 사용합니다. 현재 사용자 입력은 덮어쓰기가 맞으므로 일반 str 타입으로 정의합니다.

컨텍스트 정보는 딕셔너리 병합으로 처리하면 됩니다. 각 데이터의 성격에 맞게 병합 방식을 정하는 것이 핵심입니다.

상태 격리의 장점 서브그래프가 독립적인 상태 업데이트를 반환하는 방식은 여러 장점이 있습니다. 첫째, 예측 가능합니다.

서브그래프가 무엇을 변경했는지 반환값을 보면 명확히 알 수 있습니다. 둘째, 테스트하기 쉽습니다.

서브그래프에 특정 상태를 주고, 반환값을 검증하면 됩니다. 셋째, 병렬 실행이 가능합니다.

여러 서브그래프가 동시에 실행되어도 상태 충돌이 없습니다. 주의사항 하지만 주의할 점도 있습니다.

상태 스키마가 메인 그래프와 서브그래프 간에 호환되어야 합니다. 서브그래프가 메인 그래프에 없는 키를 반환하면 어떻게 될까요?

기본적으로 무시되거나 에러가 발생할 수 있습니다. 상태 스키마를 명확히 공유하는 것이 중요합니다.

정리 김개발 씨는 화이트보드의 그림을 보며 이해했습니다. "아, 상태를 직접 수정하는 게 아니라 업데이트를 반환해서 병합하는 거군요!" 박시니어 씨가 고개를 끄덕였습니다.

"맞아요. 이 방식이 안전하고 예측 가능한 상태 관리의 핵심이에요." 상태 공유 메커니즘을 제대로 이해하면 복잡한 그래프에서도 상태를 안정적으로 관리할 수 있습니다.

각 데이터의 성격에 맞게 병합 방식을 설정하는 것을 잊지 마세요.

실전 팁

💡 - 누적되어야 하는 데이터는 Annotated[list, add]를 사용하세요.

상태 스키마는 메인 그래프와 서브그래프 간에 명확히 공유하세요.
커스텀 병합 로직이 필요하다면 직접 함수를 작성하여 지정할 수 있습니다.

5. 서브그래프 체크포인터

"서브그래프도 체크포인트를 저장할 수 있나요?" 김개발 씨가 물었습니다. 장애가 발생했을 때 복구하려면 중간 상태를 저장해야 한다는 걸 알고 있었기 때문입니다.

박시니어 씨가 노트북을 열며 답했습니다. "당연하죠.

서브그래프도 자체 체크포인터를 가질 수 있어요."

서브그래프 체크포인터는 서브그래프 내부의 상태를 저장하고 복원하는 메커니즘입니다. 메인 그래프와 독립적으로 체크포인트를 관리할 수 있으며, 장애 복구나 디버깅에 유용합니다.

메모리, SQLite, Redis 등 다양한 백엔드를 지원합니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict

class State(TypedDict):
    value: int
    steps: list

# 체크포인터를 가진 서브그래프
def create_checkpointed_subgraph():
    checkpointer = MemorySaver()  # 메모리 기반 체크포인터
    graph = StateGraph(State)

    def step1(state):
        return {"value": state["value"] * 2, "steps": state.get("steps", []) + ["doubled"]}

    def step2(state):
        return {"value": state["value"] + 10, "steps": state.get("steps", []) + ["added"]}

    graph.add_node("double", step1)
    graph.add_node("add", step2)
    graph.add_edge(START, "double")
    graph.add_edge("double", "add")
    graph.add_edge("add", END)

    # 체크포인터와 함께 컴파일
    return graph.compile(checkpointer=checkpointer)

# 메인 그래프
main = StateGraph(State)
main.add_node("process", create_checkpointed_subgraph())
main.add_edge(START, "process")
main.add_edge("process", END)

# 체크포인트와 함께 실행
config = {"configurable": {"thread_id": "session_1"}}
app = main.compile()
result = app.invoke({"value": 5, "steps": []}, config)

김개발 씨는 프로덕션 환경에서 중요한 교훈을 얻었습니다. 긴 작업 중간에 서버가 재시작되면서 모든 진행 상황이 날아간 것입니다.

"중간 상태를 저장했어야 했는데..." 후회했습니다. 박시니어 씨가 옆에서 조언했습니다.

"체크포인터를 사용하면 그런 일을 막을 수 있어요." 체크포인터란 무엇인가 체크포인터는 그래프 실행 중간 상태를 저장하는 시스템입니다. 게임을 하다가 세이브 포인트에서 저장하는 것과 비슷합니다.

나중에 게임을 다시 시작하면 저장된 지점부터 이어할 수 있습니다. LangGraph의 체크포인터도 마찬가지입니다.

각 노드 실행 후 상태를 저장하고, 필요 시 복원합니다. 특히 오래 걸리는 작업이나 외부 API 호출이 많은 워크플로우에서 필수적입니다.

중간에 실패해도 처음부터 다시 시작할 필요가 없습니다. 메인 그래프 vs 서브그래프 체크포인터 메인 그래프에만 체크포인터를 설정하면 어떻게 될까요?

메인 그래프의 노드 단위로만 체크포인트가 생성됩니다. 서브그래프는 하나의 노드로 취급되므로, 서브그래프 내부의 중간 상태는 저장되지 않습니다.

서브그래프 실행 중 실패하면 서브그래프 전체를 다시 실행해야 합니다. 서브그래프에 자체 체크포인터를 설정하면 달라집니다.

서브그래프 내부의 각 노드마다 체크포인트가 생성됩니다. 서브그래프 중간에 실패해도 해당 지점부터 재개할 수 있습니다.

MemorySaver의 특징 코드에서 사용한 MemorySaver는 메모리 기반 체크포인터입니다. 가장 간단하고 빠른 방식입니다.

별도 설정 없이 바로 사용할 수 있습니다. 하지만 프로세스가 종료되면 모든 체크포인트가 사라집니다.

개발 환경이나 테스트에는 적합하지만, 프로덕션에는 적합하지 않습니다. 프로덕션에서는 영속성 있는 백엔드를 사용해야 합니다.

SQLite, PostgreSQL, Redis 등을 지원합니다. 체크포인트 ID와 스레드 체크포인트는 thread_id로 구분됩니다.

같은 thread_id를 사용하면 이전 실행의 체크포인트를 불러옵니다. 다른 thread_id를 사용하면 새로운 세션이 시작됩니다.

이를 통해 여러 사용자나 세션을 독립적으로 관리할 수 있습니다. 예를 들어 채팅봇에서 각 사용자마다 다른 thread_id를 부여하면, 사용자별로 대화 히스토리가 별도로 저장됩니다.

실무 활용 사례 실제 프로젝트에서 체크포인터는 매우 유용합니다. 데이터 처리 파이프라인을 생각해 봅시다.

수백 개의 파일을 처리하는 작업이 있습니다. 각 파일 처리를 서브그래프로 만들고, 체크포인터를 설정합니다.

중간에 실패해도 처리된 파일은 건너뛰고 실패한 지점부터 재개할 수 있습니다. 또 다른 예로 복잡한 승인 워크플로우가 있습니다.

여러 단계의 검토와 승인이 필요한 프로세스에서, 각 단계마다 체크포인트를 저장하면 어느 단계에서든 재개할 수 있습니다. 체크포인트 복원 체크포인트에서 복원하는 방법은 간단합니다.

동일한 thread_id로 다시 실행하면 됩니다. LangGraph는 자동으로 마지막 체크포인트를 찾아 그 시점부터 재개합니다.

개발자가 명시적으로 복원 코드를 작성할 필요가 없습니다. 특정 체크포인트로 돌아가고 싶다면 체크포인트 ID를 지정할 수도 있습니다.

이는 디버깅이나 타임 트래블 디버깅에 유용합니다. 메모리 관리 체크포인트가 계속 쌓이면 저장 공간이 문제가 될 수 있습니다.

주기적으로 오래된 체크포인트를 정리하는 로직이 필요합니다. LangGraph는 체크포인트 삭제 API를 제공합니다.

완료된 작업의 체크포인트는 일정 시간 후 삭제하는 정책을 구현할 수 있습니다. 정리 김개발 씨는 체크포인터를 적용한 후 안심이 되었습니다.

"이제 중간에 문제가 생겨도 처음부터 다시 시작하지 않아도 되겠어요." 박시니어 씨가 웃으며 말했습니다. "체크포인터는 안정적인 시스템의 필수 요소예요." 서브그래프 체크포인터를 활용하면 장애에 강한 워크플로우를 구축할 수 있습니다.

특히 오래 걸리는 작업이나 외부 의존성이 많은 경우 필수적입니다.

실전 팁

💡 - 개발 환경에서는 MemorySaver를 사용하고, 프로덕션에서는 영속성 백엔드를 사용하세요.

thread_id를 활용하여 세션이나 사용자별로 체크포인트를 관리하세요.
오래된 체크포인트는 주기적으로 정리하여 저장 공간을 관리하세요.

6. 서브그래프 스트리밍

"서브그래프 실행 과정을 실시간으로 보고 싶은데, 가능할까요?" 김개발 씨가 물었습니다. 사용자에게 진행 상황을 보여주고 싶었기 때문입니다.

박시니어 씨가 고개를 끄덕이며 답했습니다. "스트리밍을 사용하면 각 노드의 실행 결과를 실시간으로 받을 수 있어요."

서브그래프 스트리밍은 그래프 실행 과정을 실시간으로 관찰하는 기능입니다. stream 메서드를 사용하면 각 노드의 실행 결과를 이벤트로 받을 수 있습니다.

서브그래프 내부 이벤트도 포함되며, 사용자에게 진행 상황을 보여주거나 디버깅에 활용할 수 있습니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    messages: list
    progress: int

def create_streaming_subgraph():
    graph = StateGraph(State)

    def step1(state):
        return {"messages": state.get("messages", []) + ["Step 1 완료"], "progress": 33}

    def step2(state):
        return {"messages": state.get("messages", []) + ["Step 2 완료"], "progress": 66}

    def step3(state):
        return {"messages": state.get("messages", []) + ["Step 3 완료"], "progress": 100}

    graph.add_node("first", step1)
    graph.add_node("second", step2)
    graph.add_node("third", step3)
    graph.add_edge(START, "first")
    graph.add_edge("first", "second")
    graph.add_edge("second", "third")
    graph.add_edge("third", END)
    return graph.compile()

# 메인 그래프
main = StateGraph(State)
main.add_node("workflow", create_streaming_subgraph())
main.add_edge(START, "workflow")
main.add_edge("workflow", END)

# 스트리밍 실행
app = main.compile()
for event in app.stream({"messages": [], "progress": 0}):
    print(f"이벤트 수신: {event}")
    # 각 노드 실행마다 이벤트가 발생

김개발 씨는 긴 워크플로우를 실행할 때마다 답답했습니다. 결과가 나올 때까지 아무것도 볼 수 없었기 때문입니다.

"지금 어느 단계까지 진행됐는지 알 수가 없어요." 박시니어 씨가 화면을 가리키며 말했습니다. "스트리밍을 사용하면 실시간으로 확인할 수 있어요." 스트리밍이란 무엇인가 스트리밍은 그래프 실행을 관찰하는 방식입니다.

invoke 메서드는 모든 실행이 끝난 후 최종 결과만 반환합니다. 중간 과정은 볼 수 없습니다.

반면 stream 메서드는 각 노드가 실행될 때마다 이벤트를 생성합니다. 이 이벤트를 받아서 진행 상황을 추적할 수 있습니다.

마치 파일 다운로드할 때 진행률 바를 보는 것과 같습니다. 전체 과정이 투명하게 보입니다.

stream 메서드의 동작 stream 메서드는 제너레이터를 반환합니다. 파이썬의 제너레이터는 값을 하나씩 생성하는 이터레이터입니다.

for 루프로 순회하면서 각 이벤트를 받을 수 있습니다. 각 이벤트는 노드 이름과 해당 노드가 반환한 상태를 포함합니다.

코드에서 for event in app.stream(...)을 보세요. 각 노드가 실행될 때마다 이벤트가 발생하고, print로 출력됩니다.

실제로는 이 정보를 웹소켓으로 클라이언트에 전송하거나, 로그에 기록하거나, 진행률 바를 업데이트하는 데 사용할 수 있습니다. 서브그래프 내부 이벤트 흥미로운 점은 서브그래프 내부의 이벤트도 볼 수 있다는 것입니다.

메인 그래프에서 stream을 호출하면, 서브그래프의 각 노드 실행도 이벤트로 전달됩니다. 이는 디버깅에 매우 유용합니다.

어느 노드에서 시간이 오래 걸리는지, 어떤 데이터가 전달되는지 모두 확인할 수 있습니다. 이벤트에는 계층 정보도 포함됩니다.

메인 그래프의 노드인지, 서브그래프의 노드인지 구분할 수 있습니다. 실시간 피드백 제공 실무에서 스트리밍은 사용자 경험을 크게 향상시킵니다.

AI 에이전트가 복잡한 작업을 수행한다고 가정해 봅시다. 문서 분석, 정보 검색, 답변 생성 등 여러 단계를 거칩니다.

각 단계마다 "문서 분석 중...", "정보 검색 중...", "답변 생성 중..." 같은 메시지를 사용자에게 보여줄 수 있습니다. 사용자는 시스템이 멈춘 게 아니라 열심히 작업 중임을 알 수 있습니다.

이는 신뢰도를 높이고 사용자 만족도를 개선합니다. 진행률 계산 코드 예제에서 progress 필드를 주목하세요.

각 노드가 진행률을 업데이트합니다. 33%, 66%, 100%로 증가합니다.

이 정보를 클라이언트에 전달하면 정확한 진행률 바를 표시할 수 있습니다. 물론 진행률을 정확히 계산하기 어려운 경우도 있습니다.

반복 횟수를 미리 알 수 없거나, 각 단계의 소요 시간이 다를 때입니다. 이런 경우 "현재 X 단계 진행 중" 같은 정성적인 정보를 제공하는 것도 좋습니다.

성능 고려사항 스트리밍에도 비용이 있습니다. 각 이벤트를 생성하고 전달하는 데 오버헤드가 발생합니다.

노드가 수천 개인 그래프라면 이벤트도 수천 개 생성됩니다. 네트워크로 전송한다면 대역폭도 고려해야 합니다.

필요한 이벤트만 필터링하는 것이 좋습니다. LangGraph는 이벤트 타입을 지정할 수 있습니다.

예를 들어 특정 노드의 이벤트만 받거나, 에러 이벤트만 받을 수 있습니다. 에러 처리 스트리밍 중 에러가 발생하면 어떻게 될까요?

에러 이벤트가 발생하고, 스트림이 종료됩니다. try-except 블록으로 에러를 잡아서 적절히 처리해야 합니다.

사용자에게 에러 메시지를 보여주거나, 재시도 로직을 실행할 수 있습니다. 디버깅 활용 개발 중에는 스트리밍이 강력한 디버깅 도구가 됩니다.

각 노드의 입출력을 실시간으로 보면서 문제를 파악할 수 있습니다. 어느 노드에서 예상과 다른 데이터가 나오는지, 어떤 노드가 너무 오래 걸리는지 즉시 알 수 있습니다.

로그 파일에 이벤트를 기록해 두면, 나중에 분석할 때도 유용합니다. 정리 김개발 씨는 스트리밍을 적용한 후 사용자 피드백이 크게 개선되었습니다.

"이제 사용자들이 AI가 뭘 하고 있는지 알 수 있어서 좋다고 하네요!" 박시니어 씨가 만족스러운 표정으로 말했습니다. "투명성은 신뢰로 이어져요." 서브그래프 스트리밍을 활용하면 사용자 경험을 향상시키고, 디버깅도 쉬워집니다.

긴 워크플로우를 실행할 때는 꼭 스트리밍을 고려해 보세요.

실전 팁

💡 - 사용자에게 진행 상황을 보여줄 때는 stream 메서드를 사용하세요.

필요한 이벤트만 필터링하여 성능을 최적화하세요.
개발 중에는 스트리밍으로 각 노드의 입출력을 확인하며 디버깅하세요.

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

#LangGraph#Subgraph#StateGraph#Checkpoint#Streaming#AI,LLM,Python,LangGraph

LangGraph 서브그래프 완벽 가이드

# LangGraph 서브그래프 완벽 가이드 LangGraph의 서브그래프를 활용하여 복잡한 워크플로우를 모듈화하고 재사용 가능한 그래프 구조를 만드는 방법을 배웁니다. 노드에서 그래프 호출, 상태 공유, 체크포인터, 스트리밍까지 실무에 필요한 모든 내용을 다룹니다. --- ## 목차 1. [서브그래프_개념과_용도](#서브그래프-개념과-용도) 2. [노드에서_그래프_호출](#노드에서-그래프-호출) 3. [그래프를_노드로_추가](#그래프를-노드로-추가) 4. [상태_공유_메커니즘](#상태-공유-메커니즘) 5. [서브그래프_체크포인터](#서브그래프-체크포인터) 6. [서브그래프_스트리밍](#서브그래프-스트리밍) --- ## 1. 서브그래프_개념과_용도 어느 날 김개발 씨는 LangGraph로 복잡한 AI 워크플로우를 구축하던 중 문제에 부딪혔습니다. 그래프가 점점 커지면서 코드가 스파게티처럼 엉켜버린 것입니다. 선배 박시니어 씨가 다가와 물었습니다. "이 부분, 서브그래프로 분리하면 어떨까요?" **서브그래프**는 LangGraph에서 그래프 안에 또 다른 그래프를 포함시키는 기법입니다. 마치 프로그래밍에서 함수를 나눠서 사용하듯이, 복잡한 워크플로우를 작은 단위로 분리할 수 있습니다. 이를 통해 코드의 재사용성을 높이고 유지보수를 쉽게 만들 수 있습니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): messages: list count: int # 서브그래프 정의 def create_subgraph(): subgraph = StateGraph(State) subgraph.add_node("process", lambda s: {"count": s["count"] + 1}) subgraph.add_edge(START, "process") subgraph.add_edge("process", END) return subgraph.compile() # 메인 그래프에서 서브그래프 활용 main_graph = StateGraph(State) main_graph.add_node("sub_task", create_subgraph()) main_graph.add_edge(START, "sub_task") main_graph.add_edge("sub_task", END) ``` 김개발 씨는 입사 6개월 차 AI 엔지니어입니다. 요즘 LangGraph로 고객 응대 시스템을 구축하고 있는데, 그래프가 점점 복잡해지면서 코드를 읽기가 어려워졌습니다. 노드가 수십 개나 되고, 엣지 연결도 복잡하게 얽혀 있습니다. 선배 개발자 박시니어 씨가 김개발 씨의 코드를 살펴보더니 말했습니다. "이 부분은 서브그래프로 분리하는 게 좋겠네요. 지금은 모든 로직이 한 그래프에 몰려 있어서 관리하기 힘들어요." **서브그래프란 무엇일까요?** 쉽게 비유하자면, 서브그래프는 마치 레고 블록과 같습니다. 큰 레고 작품을 만들 때 작은 부품들을 먼저 조립한 후, 그것들을 다시 결합하여 완성품을 만들듯이 말입니다. 서브그래프도 작은 그래프들을 먼저 만들고, 이를 큰 그래프에 조립하여 완성된 워크플로우를 구성합니다. 프로그래밍에 익숙한 분들이라면 함수의 개념을 떠올리면 됩니다. 긴 코드를 여러 함수로 나누듯이, 복잡한 그래프를 여러 서브그래프로 나누는 것입니다. **왜 서브그래프가 필요할까요?** 서브그래프가 없던 시절을 상상해 봅시다. 모든 로직을 하나의 거대한 그래프에 담아야 했습니다. 노드가 50개, 100개로 늘어나면서 어떤 노드가 무슨 일을 하는지 파악하기 어려워졌습니다. 더 큰 문제는 코드 재사용이었습니다. 비슷한 처리 과정이 여러 곳에서 필요할 때마다 똑같은 노드들을 복사해서 붙여넣어야 했습니다. 당연히 버그가 생기면 모든 곳을 찾아서 수정해야 했고, 이는 엄청난 시간 낭비였습니다. **서브그래프가 가져온 변화** 바로 이런 문제를 해결하기 위해 서브그래프가 등장했습니다. 서브그래프를 사용하면 **모듈화**가 가능해집니다. 특정 기능을 담당하는 부분을 독립적인 서브그래프로 만들어 놓으면, 필요할 때마다 가져다 쓸 수 있습니다. 또한 **재사용성**도 크게 향상됩니다. 한 번 만든 서브그래프를 여러 프로젝트에서 사용할 수 있습니다. 무엇보다 **유지보수가 쉬워진다**는 큰 이점이 있습니다. 특정 기능에 버그가 있다면 해당 서브그래프만 수정하면 되고, 새로운 기능을 추가할 때도 독립적인 서브그래프를 추가하면 됩니다. **코드 분석** 위의 코드를 단계별로 살펴보겠습니다. 먼저 `create_subgraph` 함수에서 독립적인 서브그래프를 생성합니다. 이 서브그래프는 `process`라는 노드 하나를 가지고 있으며, 상태의 `count` 값을 1 증가시키는 간단한 작업을 수행합니다. 서브그래프도 일반 그래프처럼 `START`와 `END`를 가지며, 엣지로 연결됩니다. 마지막에 `compile()` 메서드를 호출하여 실행 가능한 형태로 만듭니다. 메인 그래프에서는 이렇게 만들어진 서브그래프를 일반 노드처럼 추가합니다. `add_node("sub_task", create_subgraph())`를 보면 서브그래프 자체가 하나의 노드로 취급되는 것을 알 수 있습니다. **실무 활용 사례** 실제 현업에서는 어떻게 활용할까요? 예를 들어 고객 문의 처리 시스템을 개발한다고 가정해봅시다. 감정 분석, 의도 파악, 답변 생성, 품질 검증 등 여러 단계가 필요합니다. 각 단계를 독립적인 서브그래프로 만들어 두면, 다른 프로젝트에서도 재사용할 수 있습니다. 예를 들어 감정 분석 서브그래프는 이메일 분류 시스템에서도 그대로 사용할 수 있습니다. **주의사항** 하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 지나치게 작은 단위로 서브그래프를 나누는 것입니다. 노드 하나짜리 서브그래프를 수십 개 만들면 오히려 복잡도가 증가합니다. 적절한 크기로 의미 있는 단위로 나누는 것이 중요합니다. 또한 서브그래프 간의 상태 공유를 잘못 설계하면 예상치 못한 버그가 발생할 수 있습니다. 상태 스키마를 명확히 정의하고, 어떤 데이터가 공유되는지 문서화해야 합니다. **정리** 다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 조언에 따라 코드를 서브그래프로 분리한 김개발 씨는 감탄했습니다. "코드가 훨씬 깔끔해졌어요! 이제 어디서 무슨 일이 일어나는지 한눈에 보이네요." 서브그래프를 제대로 활용하면 복잡한 AI 워크플로우도 체계적으로 관리할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요. **실전 팁** 💡 - 의미 있는 단위로 서브그래프를 분리하세요. 너무 작게 나누면 오히려 복잡해집니다. - 서브그래프의 이름은 그 역할을 명확히 나타내도록 짓습니다. - 재사용 가능한 서브그래프는 별도 파일로 분리하여 관리하세요. --- ## 2. 노드에서_그래프_호출 김개발 씨는 서브그래프의 개념을 이해했지만, 실제로 노드 안에서 어떻게 그래프를 호출하는지 궁금했습니다. 박시니어 씨가 화면을 가리키며 말했습니다. "일반 함수처럼 호출하면 되는데, 몇 가지 알아야 할 게 있어요." **노드에서 그래프 호출**은 그래프를 컴파일한 후 일반 함수처럼 실행하는 방식입니다. 컴파일된 그래프는 상태를 입력받아 처리 후 결과를 반환합니다. 이를 통해 복잡한 로직을 캡슐화하고, 필요할 때마다 재사용할 수 있습니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): input_text: str result: str # 서브그래프 생성 def create_processor(): graph = StateGraph(State) def process(state): # 텍스트 처리 로직 return {"result": state["input_text"].upper()} graph.add_node("transform", process) graph.add_edge(START, "transform") graph.add_edge("transform", END) return graph.compile() # 노드 함수에서 그래프 호출 processor = create_processor() def node_function(state): # 서브그래프를 함수처럼 호출 output = processor.invoke({"input_text": state["input_text"]}) return {"result": output["result"]} ``` 김개발 씨는 서브그래프의 강력함에 매료되었습니다. 하지만 막상 코드를 작성하려니 막막했습니다. "서브그래프를 만들었는데, 이걸 어떻게 사용하죠?" 박시니어 씨가 웃으며 대답했습니다. "생각보다 간단해요. 일반 함수 호출하듯이 하면 됩니다." **그래프를 함수처럼 사용하기** 서브그래프를 노드에서 호출하는 것은 마치 파이썬의 함수를 호출하는 것과 비슷합니다. 함수를 정의한 후 필요할 때 호출하듯이, 그래프를 컴파일한 후 필요할 때 실행하면 됩니다. 핵심은 `compile()` 메서드입니다. 이 메서드는 그래프 정의를 실행 가능한 형태로 변환합니다. 컴파일된 그래프는 `invoke`, `stream`, `batch` 같은 메서드를 제공하여 다양한 방식으로 실행할 수 있습니다. **컴파일 이전과 이후의 차이** 그래프를 정의만 한 상태에서는 실제로 아무 작업도 수행되지 않습니다. 이는 마치 요리 레시피를 작성한 것과 같습니다. 레시피만 있다고 요리가 완성되지는 않습니다. `compile()` 메서드를 호출하면 그래프가 실행 가능한 객체로 변환됩니다. 이제 실제로 데이터를 넣고 처리 결과를 받을 수 있습니다. 레시피대로 실제 요리를 하는 것과 같습니다. 컴파일 후에는 그래프 구조를 변경할 수 없습니다. 노드를 추가하거나 엣지를 수정하려면 다시 정의부터 시작해야 합니다. 이는 안정성을 위한 설계입니다. **invoke 메서드의 동작 방식** `invoke` 메서드는 가장 기본적인 실행 방법입니다. 상태 딕셔너리를 입력으로 받아, 그래프의 모든 노드를 순서대로 실행하고, 최종 상태를 반환합니다. 이 과정은 동기적으로 진행되므로 결과가 나올 때까지 기다려야 합니다. 코드에서 `processor.invoke({"input_text": state["input_text"]})`를 보면, 입력 상태를 전달하고 처리 결과를 받는 것을 알 수 있습니다. 반환되는 값은 최종 상태 전체입니다. **실무에서의 패턴** 실제 프로젝트에서는 서브그래프를 미리 컴파일해서 전역 변수나 클래스 속성으로 저장해 둡니다. 예를 들어 텍스트 감정 분석 서브그래프를 만들었다면, 애플리케이션 시작 시 한 번만 컴파일하고 계속 재사용합니다. 매번 컴파일하면 성능이 떨어지기 때문입니다. 이커머스 추천 시스템을 생각해 봅시다. 사용자 행동 분석, 상품 필터링, 순위 계산 등 각 단계를 서브그래프로 만들어 놓습니다. 메인 그래프의 노드에서는 필요한 서브그래프를 호출하여 결과를 조합합니다. **상태 전달의 중요성** 노드에서 그래프를 호출할 때 가장 중요한 것은 상태 전달입니다. 서브그래프가 기대하는 상태 스키마와 실제로 전달하는 상태가 일치해야 합니다. 필요한 키가 누락되면 에러가 발생합니다. 타입도 맞아야 합니다. `TypedDict`를 활용하면 타입 체크를 통해 실수를 방지할 수 있습니다. 반환값도 마찬가지입니다. 서브그래프가 어떤 형태의 상태를 반환하는지 정확히 알고 있어야, 이후 처리를 올바르게 할 수 있습니다. **에러 처리** 서브그래프 호출 시 에러가 발생할 수 있습니다. 네트워크 호출이나 외부 API를 사용하는 노드가 있다면 타임아웃이나 연결 실패가 일어날 수 있습니다. 적절한 try-except 블록으로 에러를 처리하고, 기본값을 반환하거나 재시도 로직을 구현해야 합니다. **정리** 김개발 씨는 실제로 코드를 작성해 보며 이해했습니다. "아, 그래프를 컴파일해서 함수처럼 쓰면 되는 거네요!" 박시니어 씨가 고개를 끄덕였습니다. "맞아요. 한 번 익숙해지면 정말 강력한 도구가 될 거예요." 노드에서 그래프를 호출하는 방식을 이해하면 복잡한 워크플로우도 체계적으로 구성할 수 있습니다. 각 부분을 독립적으로 개발하고 테스트한 후 조합하는 방식은 대규모 시스템 개발에 필수적입니다. **실전 팁** 💡 - 서브그래프는 애플리케이션 시작 시 한 번만 컴파일하고 재사용하세요. - 상태 스키마를 TypedDict로 명확히 정의하여 타입 안정성을 확보하세요. - 서브그래프 호출 시 에러 처리를 잊지 마세요. --- ## 3. 그래프를_노드로_추가 "그런데 매번 노드 함수 안에서 invoke를 호출하는 게 번거롭지 않나요?" 김개발 씨가 물었습니다. 박시니어 씨가 미소 지으며 답했습니다. "더 간단한 방법이 있어요. 그래프 자체를 노드로 바로 추가할 수 있거든요." **그래프를 노드로 추가**하는 방식은 컴파일된 그래프를 `add_node` 메서드에 직접 전달하는 것입니다. 별도의 래퍼 함수 없이 서브그래프를 노드로 사용할 수 있어 코드가 간결해집니다. LangGraph가 자동으로 상태를 전달하고 결과를 병합합니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): messages: list count: int processed: bool # 서브그래프 1: 카운터 def create_counter_graph(): graph = StateGraph(State) graph.add_node("increment", lambda s: {"count": s.get("count", 0) + 1}) graph.add_edge(START, "increment") graph.add_edge("increment", END) return graph.compile() # 서브그래프 2: 플래그 설정 def create_flag_graph(): graph = StateGraph(State) graph.add_node("set_flag", lambda s: {"processed": True}) graph.add_edge(START, "set_flag") graph.add_edge("set_flag", END) return graph.compile() # 메인 그래프에 서브그래프를 노드로 추가 main = StateGraph(State) main.add_node("counter", create_counter_graph()) # 직접 추가 main.add_node("flagger", create_flag_graph()) # 직접 추가 main.add_edge(START, "counter") main.add_edge("counter", "flagger") main.add_edge("flagger", END) app = main.compile() ``` 김개발 씨는 앞에서 배운 방식대로 코드를 작성했습니다. 노드 함수를 만들고, 그 안에서 서브그래프의 `invoke`를 호출하는 방식이었습니다. 동작은 잘 되었지만, 뭔가 코드가 장황하다는 느낌을 지울 수 없었습니다. 박시니어 씨가 김개발 씨의 화면을 보더니 말했습니다. "이렇게 단순히 서브그래프를 호출만 하는 노드라면, 굳이 함수로 감쌀 필요가 없어요." **직접 추가의 편리함** 그래프를 노드로 직접 추가하는 방식은 마치 플러그 앤 플레이와 같습니다. USB 기기를 컴퓨터에 꽂으면 바로 작동하듯이, 컴파일된 그래프를 `add_node`에 넣으면 바로 사용할 수 있습니다. 이전 방식에서는 매번 래퍼 함수를 만들어야 했습니다. 함수를 정의하고, 서브그래프를 호출하고, 결과를 반환하는 코드를 반복해서 작성했습니다. 서브그래프가 10개라면 래퍼 함수도 10개를 만들어야 했습니다. 하지만 직접 추가 방식을 사용하면 이런 보일러플레이트 코드가 사라집니다. 컴파일된 그래프를 바로 노드로 등록하면 끝입니다. **LangGraph의 자동 처리** 어떻게 이게 가능할까요? LangGraph는 `add_node`에 전달된 객체의 타입을 확인합니다. 컴파일된 그래프 객체라면, 내부적으로 자동으로 `invoke`를 호출하는 래퍼를 생성합니다. 개발자는 이 과정을 신경 쓸 필요가 없습니다. 상태 전달도 자동으로 처리됩니다. 현재 그래프의 상태가 서브그래프로 전달되고, 서브그래프가 반환한 상태는 현재 상태와 병합됩니다. 이 과정에서 같은 키가 있다면 서브그래프의 값으로 덮어씁니다. **상태 병합의 메커니즘** 상태 병합은 중요한 개념입니다. 메인 그래프의 상태가 `{"count": 5, "messages": ["hello"]}`라고 가정해 봅시다. 서브그래프가 `{"count": 6}`을 반환하면, 최종 상태는 `{"count": 6, "messages": ["hello"]}`가 됩니다. `count`는 업데이트되고, `messages`는 그대로 유지됩니다. 이는 딕셔너리의 `update` 메서드와 유사합니다. 기존 값은 유지하되, 새로운 값으로 덮어쓰는 방식입니다. **실무 활용 사례** 실제 프로젝트에서 이 패턴은 매우 유용합니다. 문서 처리 파이프라인을 생각해 봅시다. PDF 파싱, 텍스트 추출, 요약, 키워드 추출 등 각 단계를 독립적인 서브그래프로 만듭니다. 메인 그래프에서는 이들을 순서대로 노드로 추가하기만 하면 됩니다. 각 서브그래프는 독립적으로 개발하고 테스트할 수 있습니다. PDF 파싱 서브그래프에 버그가 있다면, 그 부분만 수정하면 됩니다. 나머지 파이프라인은 영향을 받지 않습니다. **코드의 간결함** 위의 예제 코드를 보면 메인 그래프 구성이 매우 간결합니다. `main.add_node("counter", create_counter_graph())`처럼 한 줄로 서브그래프를 추가합니다. 별도의 함수 정의가 필요 없습니다. 코드를 읽는 사람도 "counter 노드는 counter 그래프를 실행하는구나"라고 직관적으로 이해할 수 있습니다. 엣지 연결도 일반 노드와 동일합니다. `add_edge("counter", "flagger")`처럼 서브그래프 간 연결도 자연스럽게 표현됩니다. **언제 사용할까** 모든 경우에 직접 추가 방식을 사용하는 것은 아닙니다. 서브그래프를 호출하기 전에 전처리가 필요하거나, 호출 후 후처리가 필요하다면 여전히 래퍼 함수를 사용해야 합니다. 예를 들어 입력 형식을 변환하거나, 결과를 검증하는 로직이 있다면 함수로 감싸는 게 맞습니다. 하지만 단순히 서브그래프를 실행하기만 한다면, 직접 추가 방식이 훨씬 깔끔합니다. **정리** 김개발 씨는 코드를 수정해 보았습니다. 래퍼 함수들을 모두 제거하고 서브그래프를 직접 추가했습니다. "와, 코드가 절반으로 줄었어요!" 박시니어 씨가 만족스러운 표정으로 말했습니다. "간결한 코드가 좋은 코드예요." 그래프를 노드로 직접 추가하는 방식을 익히면 코드가 훨씬 읽기 쉬워집니다. 보일러플레이트를 줄이고 핵심 로직에 집중할 수 있습니다. **실전 팁** 💡 - 단순히 서브그래프만 실행한다면 직접 추가 방식을 사용하세요. - 전처리나 후처리가 필요하다면 래퍼 함수를 사용하세요. - 노드 이름은 서브그래프의 역할을 명확히 나타내도록 짓습니다. --- ## 4. 상태_공유_메커니즘 서브그래프를 사용하다 보니 김개발 씨는 새로운 의문이 생겼습니다. "메인 그래프와 서브그래프가 같은 상태를 공유하는 건가요? 아니면 복사본을 사용하는 건가요?" 박시니어 씨가 화이트보드를 꺼내들며 설명을 시작했습니다. LangGraph의 **상태 공유 메커니즘**은 서브그래프가 메인 그래프의 상태를 직접 수정하는 것이 아니라, 상태 업데이트를 반환하여 병합하는 방식입니다. 각 서브그래프는 독립적으로 동작하며, 반환된 상태만 메인 그래프에 반영됩니다. 이를 통해 예측 가능하고 안전한 상태 관리가 가능합니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated from operator import add class State(TypedDict): # Annotated로 리스트 병합 방식 지정 messages: Annotated[list, add] count: int metadata: dict def create_message_adder(): graph = StateGraph(State) def add_message(state): # 새 메시지 추가 (기존 메시지와 병합됨) return {"messages": ["서브그래프에서 추가"]} graph.add_node("add", add_message) graph.add_edge(START, "add") graph.add_edge("add", END) return graph.compile() # 메인 그래프 main = StateGraph(State) main.add_node("sub", create_message_adder()) main.add_edge(START, "sub") main.add_edge("sub", END) # 실행 app = main.compile() result = app.invoke({"messages": ["초기 메시지"], "count": 0, "metadata": {}}) # result["messages"]는 ["초기 메시지", "서브그래프에서 추가"] ``` 김개발 씨는 코드를 작성하다가 이상한 현상을 발견했습니다. 서브그래프에서 상태를 변경했는데, 메인 그래프의 상태도 함께 변경된 것입니다. "이게 맞는 건가요? 아니면 버그인가요?" 박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다. **상태는 어떻게 전달될까** LangGraph의 상태 관리는 함수형 프로그래밍의 불변성 원칙을 따릅니다. 각 노드는 현재 상태를 입력으로 받습니다. 하지만 이 상태를 직접 수정하지 않습니다. 대신 새로운 상태 업데이트를 반환합니다. LangGraph는 이 업데이트를 기존 상태와 병합하여 다음 상태를 만듭니다. 이는 마치 Git의 커밋과 비슷합니다. 이전 커밋을 직접 수정하는 게 아니라, 변경 사항을 새 커밋으로 만들어 쌓아가는 방식입니다. 각 단계마다 상태의 스냅샷이 생성되고, 히스토리를 추적할 수 있습니다. **기본 병합 규칙** 기본적으로 LangGraph는 딕셔너리 병합 방식을 사용합니다. 서브그래프가 `{"count": 10}`을 반환하면, 메인 상태의 `count` 키만 업데이트됩니다. 다른 키들은 그대로 유지됩니다. 이는 파이썬의 `dict.update()`와 동일한 동작입니다. 하지만 리스트나 집합 같은 컬렉션 타입은 어떻게 될까요? 기본적으로는 덮어쓰기가 됩니다. `{"messages": ["new"]}`를 반환하면 기존 메시지들이 모두 사라지고 새 메시지로 교체됩니다. **Annotated를 활용한 병합 제어** 이런 기본 동작을 바꾸고 싶다면 `Annotated` 타입을 사용합니다. 코드에서 `Annotated[list, add]`를 보세요. 이는 "리스트 타입이며, 병합 시 `add` 연산자를 사용하라"는 의미입니다. `operator.add`는 두 리스트를 합치는 연산자입니다. 이제 서브그래프가 `{"messages": ["new"]}`를 반환하면, 기존 메시지들 뒤에 추가됩니다. `["old"] + ["new"]` = `["old", "new"]`가 되는 것입니다. 이 방식은 채팅 메시지, 로그, 이벤트 히스토리 등 누적되어야 하는 데이터에 매우 유용합니다. **커스텀 병합 로직** `add` 외에도 다양한 병합 방식을 지정할 수 있습니다. 직접 함수를 만들어 전달할 수도 있습니다. 예를 들어 중복을 제거하며 병합하고 싶다면, 커스텀 함수를 작성하여 `Annotated[list, custom_merge]`로 지정합니다. ```python def unique_merge(existing, new): return list(set(existing + new)) class State(TypedDict): tags: Annotated[list, unique_merge] ``` 이렇게 하면 태그 같은 데이터를 중복 없이 관리할 수 있습니다. **실무에서의 패턴** 실제 프로젝트에서는 상태 스키마 설계가 매우 중요합니다. 채팅봇을 만든다고 가정해 봅시다. 메시지 히스토리는 누적되어야 하므로 `Annotated[list, add]`를 사용합니다. 현재 사용자 입력은 덮어쓰기가 맞으므로 일반 `str` 타입으로 정의합니다. 컨텍스트 정보는 딕셔너리 병합으로 처리하면 됩니다. 각 데이터의 성격에 맞게 병합 방식을 정하는 것이 핵심입니다. **상태 격리의 장점** 서브그래프가 독립적인 상태 업데이트를 반환하는 방식은 여러 장점이 있습니다. 첫째, 예측 가능합니다. 서브그래프가 무엇을 변경했는지 반환값을 보면 명확히 알 수 있습니다. 둘째, 테스트하기 쉽습니다. 서브그래프에 특정 상태를 주고, 반환값을 검증하면 됩니다. 셋째, 병렬 실행이 가능합니다. 여러 서브그래프가 동시에 실행되어도 상태 충돌이 없습니다. **주의사항** 하지만 주의할 점도 있습니다. 상태 스키마가 메인 그래프와 서브그래프 간에 호환되어야 합니다. 서브그래프가 메인 그래프에 없는 키를 반환하면 어떻게 될까요? 기본적으로 무시되거나 에러가 발생할 수 있습니다. 상태 스키마를 명확히 공유하는 것이 중요합니다. **정리** 김개발 씨는 화이트보드의 그림을 보며 이해했습니다. "아, 상태를 직접 수정하는 게 아니라 업데이트를 반환해서 병합하는 거군요!" 박시니어 씨가 고개를 끄덕였습니다. "맞아요. 이 방식이 안전하고 예측 가능한 상태 관리의 핵심이에요." 상태 공유 메커니즘을 제대로 이해하면 복잡한 그래프에서도 상태를 안정적으로 관리할 수 있습니다. 각 데이터의 성격에 맞게 병합 방식을 설정하는 것을 잊지 마세요. **실전 팁** 💡 - 누적되어야 하는 데이터는 Annotated[list, add]를 사용하세요. - 상태 스키마는 메인 그래프와 서브그래프 간에 명확히 공유하세요. - 커스텀 병합 로직이 필요하다면 직접 함수를 작성하여 지정할 수 있습니다. --- ## 5. 서브그래프_체크포인터 "서브그래프도 체크포인트를 저장할 수 있나요?" 김개발 씨가 물었습니다. 장애가 발생했을 때 복구하려면 중간 상태를 저장해야 한다는 걸 알고 있었기 때문입니다. 박시니어 씨가 노트북을 열며 답했습니다. "당연하죠. 서브그래프도 자체 체크포인터를 가질 수 있어요." **서브그래프 체크포인터**는 서브그래프 내부의 상태를 저장하고 복원하는 메커니즘입니다. 메인 그래프와 독립적으로 체크포인트를 관리할 수 있으며, 장애 복구나 디버깅에 유용합니다. 메모리, SQLite, Redis 등 다양한 백엔드를 지원합니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict class State(TypedDict): value: int steps: list # 체크포인터를 가진 서브그래프 def create_checkpointed_subgraph(): checkpointer = MemorySaver() # 메모리 기반 체크포인터 graph = StateGraph(State) def step1(state): return {"value": state["value"] * 2, "steps": state.get("steps", []) + ["doubled"]} def step2(state): return {"value": state["value"] + 10, "steps": state.get("steps", []) + ["added"]} graph.add_node("double", step1) graph.add_node("add", step2) graph.add_edge(START, "double") graph.add_edge("double", "add") graph.add_edge("add", END) # 체크포인터와 함께 컴파일 return graph.compile(checkpointer=checkpointer) # 메인 그래프 main = StateGraph(State) main.add_node("process", create_checkpointed_subgraph()) main.add_edge(START, "process") main.add_edge("process", END) # 체크포인트와 함께 실행 config = {"configurable": {"thread_id": "session_1"}} app = main.compile() result = app.invoke({"value": 5, "steps": []}, config) ``` 김개발 씨는 프로덕션 환경에서 중요한 교훈을 얻었습니다. 긴 작업 중간에 서버가 재시작되면서 모든 진행 상황이 날아간 것입니다. "중간 상태를 저장했어야 했는데..." 후회했습니다. 박시니어 씨가 옆에서 조언했습니다. "체크포인터를 사용하면 그런 일을 막을 수 있어요." **체크포인터란 무엇인가** 체크포인터는 그래프 실행 중간 상태를 저장하는 시스템입니다. 게임을 하다가 세이브 포인트에서 저장하는 것과 비슷합니다. 나중에 게임을 다시 시작하면 저장된 지점부터 이어할 수 있습니다. LangGraph의 체크포인터도 마찬가지입니다. 각 노드 실행 후 상태를 저장하고, 필요 시 복원합니다. 특히 오래 걸리는 작업이나 외부 API 호출이 많은 워크플로우에서 필수적입니다. 중간에 실패해도 처음부터 다시 시작할 필요가 없습니다. **메인 그래프 vs 서브그래프 체크포인터** 메인 그래프에만 체크포인터를 설정하면 어떻게 될까요? 메인 그래프의 노드 단위로만 체크포인트가 생성됩니다. 서브그래프는 하나의 노드로 취급되므로, 서브그래프 내부의 중간 상태는 저장되지 않습니다. 서브그래프 실행 중 실패하면 서브그래프 전체를 다시 실행해야 합니다. 서브그래프에 자체 체크포인터를 설정하면 달라집니다. 서브그래프 내부의 각 노드마다 체크포인트가 생성됩니다. 서브그래프 중간에 실패해도 해당 지점부터 재개할 수 있습니다. **MemorySaver의 특징** 코드에서 사용한 `MemorySaver`는 메모리 기반 체크포인터입니다. 가장 간단하고 빠른 방식입니다. 별도 설정 없이 바로 사용할 수 있습니다. 하지만 프로세스가 종료되면 모든 체크포인트가 사라집니다. 개발 환경이나 테스트에는 적합하지만, 프로덕션에는 적합하지 않습니다. 프로덕션에서는 영속성 있는 백엔드를 사용해야 합니다. SQLite, PostgreSQL, Redis 등을 지원합니다. **체크포인트 ID와 스레드** 체크포인트는 `thread_id`로 구분됩니다. 같은 `thread_id`를 사용하면 이전 실행의 체크포인트를 불러옵니다. 다른 `thread_id`를 사용하면 새로운 세션이 시작됩니다. 이를 통해 여러 사용자나 세션을 독립적으로 관리할 수 있습니다. 예를 들어 채팅봇에서 각 사용자마다 다른 `thread_id`를 부여하면, 사용자별로 대화 히스토리가 별도로 저장됩니다. **실무 활용 사례** 실제 프로젝트에서 체크포인터는 매우 유용합니다. 데이터 처리 파이프라인을 생각해 봅시다. 수백 개의 파일을 처리하는 작업이 있습니다. 각 파일 처리를 서브그래프로 만들고, 체크포인터를 설정합니다. 중간에 실패해도 처리된 파일은 건너뛰고 실패한 지점부터 재개할 수 있습니다. 또 다른 예로 복잡한 승인 워크플로우가 있습니다. 여러 단계의 검토와 승인이 필요한 프로세스에서, 각 단계마다 체크포인트를 저장하면 어느 단계에서든 재개할 수 있습니다. **체크포인트 복원** 체크포인트에서 복원하는 방법은 간단합니다. 동일한 `thread_id`로 다시 실행하면 됩니다. LangGraph는 자동으로 마지막 체크포인트를 찾아 그 시점부터 재개합니다. 개발자가 명시적으로 복원 코드를 작성할 필요가 없습니다. 특정 체크포인트로 돌아가고 싶다면 체크포인트 ID를 지정할 수도 있습니다. 이는 디버깅이나 타임 트래블 디버깅에 유용합니다. **메모리 관리** 체크포인트가 계속 쌓이면 저장 공간이 문제가 될 수 있습니다. 주기적으로 오래된 체크포인트를 정리하는 로직이 필요합니다. LangGraph는 체크포인트 삭제 API를 제공합니다. 완료된 작업의 체크포인트는 일정 시간 후 삭제하는 정책을 구현할 수 있습니다. **정리** 김개발 씨는 체크포인터를 적용한 후 안심이 되었습니다. "이제 중간에 문제가 생겨도 처음부터 다시 시작하지 않아도 되겠어요." 박시니어 씨가 웃으며 말했습니다. "체크포인터는 안정적인 시스템의 필수 요소예요." 서브그래프 체크포인터를 활용하면 장애에 강한 워크플로우를 구축할 수 있습니다. 특히 오래 걸리는 작업이나 외부 의존성이 많은 경우 필수적입니다. **실전 팁** 💡 - 개발 환경에서는 MemorySaver를 사용하고, 프로덕션에서는 영속성 백엔드를 사용하세요. - thread_id를 활용하여 세션이나 사용자별로 체크포인트를 관리하세요. - 오래된 체크포인트는 주기적으로 정리하여 저장 공간을 관리하세요. --- ## 6. 서브그래프_스트리밍 "서브그래프 실행 과정을 실시간으로 보고 싶은데, 가능할까요?" 김개발 씨가 물었습니다. 사용자에게 진행 상황을 보여주고 싶었기 때문입니다. 박시니어 씨가 고개를 끄덕이며 답했습니다. "스트리밍을 사용하면 각 노드의 실행 결과를 실시간으로 받을 수 있어요." **서브그래프 스트리밍**은 그래프 실행 과정을 실시간으로 관찰하는 기능입니다. `stream` 메서드를 사용하면 각 노드의 실행 결과를 이벤트로 받을 수 있습니다. 서브그래프 내부 이벤트도 포함되며, 사용자에게 진행 상황을 보여주거나 디버깅에 활용할 수 있습니다. 다음 코드를 살펴봅시다. ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict class State(TypedDict): messages: list progress: int def create_streaming_subgraph(): graph = StateGraph(State) def step1(state): return {"messages": state.get("messages", []) + ["Step 1 완료"], "progress": 33} def step2(state): return {"messages": state.get("messages", []) + ["Step 2 완료"], "progress": 66} def step3(state): return {"messages": state.get("messages", []) + ["Step 3 완료"], "progress": 100} graph.add_node("first", step1) graph.add_node("second", step2) graph.add_node("third", step3) graph.add_edge(START, "first") graph.add_edge("first", "second") graph.add_edge("second", "third") graph.add_edge("third", END) return graph.compile() # 메인 그래프 main = StateGraph(State) main.add_node("workflow", create_streaming_subgraph()) main.add_edge(START, "workflow") main.add_edge("workflow", END) # 스트리밍 실행 app = main.compile() for event in app.stream({"messages": [], "progress": 0}): print(f"이벤트 수신: {event}") # 각 노드 실행마다 이벤트가 발생 ``` 김개발 씨는 긴 워크플로우를 실행할 때마다 답답했습니다. 결과가 나올 때까지 아무것도 볼 수 없었기 때문입니다. "지금 어느 단계까지 진행됐는지 알 수가 없어요." 박시니어 씨가 화면을 가리키며 말했습니다. "스트리밍을 사용하면 실시간으로 확인할 수 있어요." **스트리밍이란 무엇인가** 스트리밍은 그래프 실행을 관찰하는 방식입니다. `invoke` 메서드는 모든 실행이 끝난 후 최종 결과만 반환합니다. 중간 과정은 볼 수 없습니다. 반면 `stream` 메서드는 각 노드가 실행될 때마다 이벤트를 생성합니다. 이 이벤트를 받아서 진행 상황을 추적할 수 있습니다. 마치 파일 다운로드할 때 진행률 바를 보는 것과 같습니다. 전체 과정이 투명하게 보입니다. **stream 메서드의 동작** `stream` 메서드는 제너레이터를 반환합니다. 파이썬의 제너레이터는 값을 하나씩 생성하는 이터레이터입니다. `for` 루프로 순회하면서 각 이벤트를 받을 수 있습니다. 각 이벤트는 노드 이름과 해당 노드가 반환한 상태를 포함합니다. 코드에서 `for event in app.stream(...)`을 보세요. 각 노드가 실행될 때마다 이벤트가 발생하고, `print`로 출력됩니다. 실제로는 이 정보를 웹소켓으로 클라이언트에 전송하거나, 로그에 기록하거나, 진행률 바를 업데이트하는 데 사용할 수 있습니다. **서브그래프 내부 이벤트** 흥미로운 점은 서브그래프 내부의 이벤트도 볼 수 있다는 것입니다. 메인 그래프에서 `stream`을 호출하면, 서브그래프의 각 노드 실행도 이벤트로 전달됩니다. 이는 디버깅에 매우 유용합니다. 어느 노드에서 시간이 오래 걸리는지, 어떤 데이터가 전달되는지 모두 확인할 수 있습니다. 이벤트에는 계층 정보도 포함됩니다. 메인 그래프의 노드인지, 서브그래프의 노드인지 구분할 수 있습니다. **실시간 피드백 제공** 실무에서 스트리밍은 사용자 경험을 크게 향상시킵니다. AI 에이전트가 복잡한 작업을 수행한다고 가정해 봅시다. 문서 분석, 정보 검색, 답변 생성 등 여러 단계를 거칩니다. 각 단계마다 "문서 분석 중...", "정보 검색 중...", "답변 생성 중..." 같은 메시지를 사용자에게 보여줄 수 있습니다. 사용자는 시스템이 멈춘 게 아니라 열심히 작업 중임을 알 수 있습니다. 이는 신뢰도를 높이고 사용자 만족도를 개선합니다. **진행률 계산** 코드 예제에서 `progress` 필드를 주목하세요. 각 노드가 진행률을 업데이트합니다. 33%, 66%, 100%로 증가합니다. 이 정보를 클라이언트에 전달하면 정확한 진행률 바를 표시할 수 있습니다. 물론 진행률을 정확히 계산하기 어려운 경우도 있습니다. 반복 횟수를 미리 알 수 없거나, 각 단계의 소요 시간이 다를 때입니다. 이런 경우 "현재 X 단계 진행 중" 같은 정성적인 정보를 제공하는 것도 좋습니다. **성능 고려사항** 스트리밍에도 비용이 있습니다. 각 이벤트를 생성하고 전달하는 데 오버헤드가 발생합니다. 노드가 수천 개인 그래프라면 이벤트도 수천 개 생성됩니다. 네트워크로 전송한다면 대역폭도 고려해야 합니다. 필요한 이벤트만 필터링하는 것이 좋습니다. LangGraph는 이벤트 타입을 지정할 수 있습니다. 예를 들어 특정 노드의 이벤트만 받거나, 에러 이벤트만 받을 수 있습니다. **에러 처리** 스트리밍 중 에러가 발생하면 어떻게 될까요? 에러 이벤트가 발생하고, 스트림이 종료됩니다. `try-except` 블록으로 에러를 잡아서 적절히 처리해야 합니다. 사용자에게 에러 메시지를 보여주거나, 재시도 로직을 실행할 수 있습니다. **디버깅 활용** 개발 중에는 스트리밍이 강력한 디버깅 도구가 됩니다. 각 노드의 입출력을 실시간으로 보면서 문제를 파악할 수 있습니다. 어느 노드에서 예상과 다른 데이터가 나오는지, 어떤 노드가 너무 오래 걸리는지 즉시 알 수 있습니다. 로그 파일에 이벤트를 기록해 두면, 나중에 분석할 때도 유용합니다. **정리** 김개발 씨는 스트리밍을 적용한 후 사용자 피드백이 크게 개선되었습니다. "이제 사용자들이 AI가 뭘 하고 있는지 알 수 있어서 좋다고 하네요!" 박시니어 씨가 만족스러운 표정으로 말했습니다. "투명성은 신뢰로 이어져요." 서브그래프 스트리밍을 활용하면 사용자 경험을 향상시키고, 디버깅도 쉬워집니다. 긴 워크플로우를 실행할 때는 꼭 스트리밍을 고려해 보세요. **실전 팁** 💡 - 사용자에게 진행 상황을 보여줄 때는 stream 메서드를 사용하세요. - 필요한 이벤트만 필터링하여 성능을 최적화하세요. - 개발 중에는 스트리밍으로 각 노드의 입출력을 확인하며 디버깅하세요. --- 이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!

카테고리: AI,LLM,Python,LangGraph

언어: Python

태그: LangGraph, Subgraph, StateGraph, Checkpoint, Streaming, AI,LLM,Python,LangGraph

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

CodeDeck

LangGraph 서브그래프 완벽 가이드

목차

1. 서브그래프 개념과 용도

2. 노드에서 그래프 호출

3. 그래프를 노드로 추가

4. 상태 공유 메커니즘

5. 서브그래프 체크포인터

6. 서브그래프 스트리밍

이 카드뉴스가 포함된 코스

LangGraph 완전 정복

댓글 (0)

함께 보면 좋은 카드 뉴스

Helm 마이크로서비스 패키징 완벽 가이드

보안 아키텍처 구성 완벽 가이드

AWS Organizations 완벽 가이드

AWS KMS 암호화 완벽 가이드

AWS Secrets Manager 완벽 가이드