CodeDeck

🤖

본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.

⚠️

본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.

이미지 로딩 중...

LangGraph 노드 정의 완벽 가이드 - 슬라이드 1/7
A

AI Generated

2025. 12. 12. · 11 Views

LangGraph 노드 정의 완벽 가이드

LangGraph에서 노드를 정의하고 활용하는 방법을 초급 개발자도 쉽게 이해할 수 있도록 설명합니다. 노드 함수 시그니처부터 상태 업데이트까지 실무에 필요한 모든 내용을 담았습니다.

목차

  1. 노드 함수 시그니처
  2. state, config, runtime 파라미터
  3. START 진입점 노드
  4. END 종료 노드
  5. 동기/비동기 노드 함수
  6. 노드에서 상태 업데이트

1. 노드 함수 시그니처

어느 날 김개발 씨가 LangGraph 튜토리얼을 보다가 고개를 갸우뚱했습니다. "노드가 뭐지?

그냥 함수를 만들면 되는 건가?" 선배 박시니어 씨가 웃으며 말했습니다. "노드는 그래프의 핵심이에요.

제대로 된 시그니처를 알아야 합니다."

노드 함수 시그니처는 LangGraph에서 노드를 정의하는 기본 형식입니다. 마치 레고 블록이 정해진 모양으로 끼워져야 하는 것처럼, 노드 함수도 정해진 형식을 따라야 그래프에서 제대로 작동합니다.

이 시그니처를 이해하면 복잡한 AI 워크플로우를 손쉽게 구성할 수 있습니다.

다음 코드를 살펴봅시다.

from typing import TypedDict

class State(TypedDict):
    messages: list[str]
    count: int

# 기본 노드 함수 시그니처
def my_node(state: State) -> State:
    """노드는 state를 받아서 state를 반환합니다"""
    # 여기서 비즈니스 로직을 처리합니다
    new_messages = state["messages"] + ["새로운 메시지"]
    new_count = state["count"] + 1

    # 업데이트된 state를 반환합니다
    return {"messages": new_messages, "count": new_count}

김개발 씨는 입사 2개월 차 주니어 개발자입니다. 회사에서 AI 챗봇 프로젝트에 투입되었는데, LangGraph라는 새로운 도구를 처음 접하게 되었습니다.

튜토리얼을 따라 하다가 노드라는 개념에 막혔습니다. "함수인 것 같기도 하고, 뭔가 다른 것 같기도 하고..." 박시니어 씨가 자리에 앉아 천천히 설명하기 시작했습니다.

"LangGraph에서 노드는 특별한 함수예요. 아무 함수나 노드가 될 수 있는 건 아니죠." 그렇다면 노드 함수 시그니처란 정확히 무엇일까요?

쉽게 비유하자면, 노드 함수 시그니처는 마치 공장의 컨베이어 벨트와 같습니다. 컨베이어 벨트에서 물건이 들어오면 작업을 거쳐 다시 컨베이어 벨트로 나갑니다.

입구와 출구의 형식이 정해져 있어야 다음 공정으로 자연스럽게 이어질 수 있습니다. 이처럼 노드 함수도 state를 받아서 state를 반환하는 명확한 형식을 가져야 합니다.

왜 이런 형식이 필요할까요? 초창기 워크플로우 시스템에서는 각 단계마다 다른 형식의 입출력을 사용했습니다.

A 함수는 딕셔너리를 받아 리스트를 반환하고, B 함수는 리스트를 받아 문자열을 반환하는 식이었죠. 이렇게 하면 함수들을 연결할 때마다 변환 코드를 작성해야 했습니다.

프로젝트가 커질수록 이런 변환 로직은 스파게티처럼 엉켜버렸습니다. 더 큰 문제는 디버깅이었습니다.

중간에 어떤 데이터가 흐르는지 추적하기가 매우 어려웠습니다. 각 단계마다 데이터 형식이 달랐기 때문에 로그를 봐도 무슨 일이 일어나는지 파악하기 힘들었습니다.

바로 이런 문제를 해결하기 위해 LangGraph의 노드 함수 시그니처가 탄생했습니다. 노드 함수 시그니처를 사용하면 일관된 데이터 흐름이 가능해집니다.

모든 노드가 같은 형식의 state를 주고받기 때문에 복잡한 변환 없이 자연스럽게 연결됩니다. 또한 타입 안정성도 얻을 수 있습니다.

TypedDict를 사용하면 IDE가 자동완성을 제공하고 오타를 잡아줍니다. 무엇보다 예측 가능한 동작이라는 큰 이점이 있습니다.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 State를 TypedDict로 정의합니다.

이것은 state가 어떤 필드를 가지는지 명시적으로 선언하는 것입니다. 다음으로 노드 함수는 반드시 state 파라미터를 받아야 합니다.

함수 안에서는 기존 state를 읽어서 새로운 값을 계산합니다. 마지막으로 업데이트된 state를 딕셔너리 형태로 반환합니다.

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

문의 접수 노드에서는 고객 메시지를 state에 추가하고, 분류 노드에서는 카테고리를 state에 추가하고, 응답 생성 노드에서는 답변을 state에 추가합니다. 모든 노드가 같은 state 형식을 사용하기 때문에 흐름이 매우 자연스럽습니다.

많은 AI 스타트업에서 이런 패턴을 적극적으로 사용하고 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 state를 직접 수정하는 것입니다. state 딕셔너리를 받아서 그대로 수정하고 반환하면 예상치 못한 부작용이 발생할 수 있습니다.

따라서 새로운 딕셔너리를 생성해서 반환하는 방식으로 사용해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 설명을 들은 김개발 씨는 눈이 반짝였습니다. "아, state를 받아서 state를 반환하는 거군요!

생각보다 단순하네요." 노드 함수 시그니처를 제대로 이해하면 복잡한 AI 워크플로우도 레고 블록 조립하듯 쉽게 만들 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - TypedDict를 사용해 state 구조를 명확히 정의하세요

  • 노드 함수는 순수 함수처럼 작성하여 부작용을 최소화하세요
  • 반환 시 전체 state가 아닌 업데이트할 필드만 포함해도 됩니다

2. state, config, runtime 파라미터

김개발 씨가 첫 번째 노드를 성공적으로 만들었습니다. 그런데 튜토리얼을 더 읽다 보니 어떤 노드는 파라미터가 3개나 있었습니다.

"state만 받으면 되는 거 아니었나요?" 박시니어 씨가 고개를 끄덕였습니다. "상황에 따라 더 많은 정보가 필요할 때가 있죠."

노드 함수는 state, config, runtime 세 가지 파라미터를 받을 수 있습니다. state는 작업 데이터를, config는 설정 정보를, runtime은 실행 환경 정보를 담고 있습니다.

마치 요리사가 재료뿐만 아니라 레시피와 주방 도구도 필요한 것처럼, 노드도 상황에 맞는 다양한 정보가 필요합니다.

다음 코드를 살펴봅시다.

from langgraph.types import Command
from typing import TypedDict

class State(TypedDict):
    messages: list[str]
    user_id: str

# 세 가지 파라미터를 모두 사용하는 노드
def advanced_node(state: State, config: dict, runtime) -> Command:
    """config와 runtime 정보를 활용하는 고급 노드"""
    # config에서 설정 값을 가져옵니다
    max_retries = config.get("configurable", {}).get("max_retries", 3)

    # runtime에서 실행 환경 정보를 가져옵니다
    current_thread = runtime.thread_id

    # state를 활용해 비즈니스 로직을 처리합니다
    user_message = f"사용자 {state['user_id']}: {state['messages'][-1]}"

    # Command 객체로 반환하여 다음 노드를 지정할 수도 있습니다
    return Command(
        update={"messages": state["messages"] + [user_message]},
        goto="next_node"
    )

김개발 씨는 이제 기본 노드를 자유롭게 만들 수 있게 되었습니다. 하지만 실제 프로젝트를 진행하다 보니 문제가 생겼습니다.

개발 환경과 운영 환경에서 다른 설정을 사용해야 하는데, 노드 함수 안에 하드코딩할 수는 없었습니다. 박시니어 씨에게 상황을 설명하자, "그럴 때는 config 파라미터를 사용하면 돼요"라는 답변이 돌아왔습니다.

state, config, runtime 파라미터는 정확히 무엇일까요? 쉽게 비유하자면, 이 세 가지는 마치 배달 기사의 세 가지 도구와 같습니다.

state는 배달할 물건입니다. 실제 작업 대상이 되는 데이터죠.

config는 배달 지시서입니다. 어떤 경로로 갈지, 몇 번까지 재시도할지 같은 설정이 담겨 있습니다.

runtime은 배달 차량과 GPS입니다. 현재 위치, 스레드 ID, 실행 환경 같은 메타 정보를 제공합니다.

왜 이런 구분이 필요할까요? 초기 버전의 워크플로우 시스템에서는 모든 정보를 state에 넣었습니다.

작업 데이터도 state, 설정 값도 state, 환경 정보도 state에 우겨넣었죠. 그 결과 state가 비대해지고, 어떤 데이터가 실제 작업용이고 어떤 데이터가 설정용인지 구분하기 어려워졌습니다.

더 큰 문제는 재사용성이었습니다. 같은 노드를 다른 설정으로 사용하려면 state를 수정해야 했고, 이는 코드 중복으로 이어졌습니다.

테스트할 때도 state에 온갖 더미 데이터를 넣어야 했습니다. 바로 이런 문제를 해결하기 위해 파라미터가 세 가지로 분리되었습니다.

config를 사용하면 환경별 설정 분리가 가능해집니다. 같은 노드 코드를 개발, 스테이징, 운영 환경에서 다른 설정으로 실행할 수 있습니다.

runtime을 활용하면 실행 컨텍스트 추적이 가능합니다. 어떤 스레드에서 실행 중인지, 몇 번째 실행인지 같은 정보를 얻을 수 있죠.

무엇보다 깔끔한 관심사 분리라는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

함수 시그니처에서 세 개의 파라미터를 선언합니다. config는 딕셔너리 형태로 전달되는데, 보통 configurable 키 아래에 사용자 설정이 들어갑니다.

runtime 객체에서는 thread_id 같은 실행 정보를 가져올 수 있습니다. state는 평소처럼 작업 데이터를 담고 있습니다.

Command 객체로 반환하면 state 업데이트와 함께 다음 실행할 노드를 지정할 수 있습니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 이메일 발송 시스템을 개발한다고 가정해봅시다. state에는 수신자와 메시지 내용이 담겨 있고, config에는 SMTP 서버 주소와 재시도 횟수가 설정되어 있고, runtime에서는 현재 요청의 추적 ID를 가져옵니다.

같은 노드를 테스트 환경에서는 가짜 SMTP로, 운영 환경에서는 실제 SMTP로 실행할 수 있습니다. 글로벌 SaaS 기업들이 이런 패턴을 널리 사용합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수는 작업 데이터를 config에 넣는 것입니다.

config는 어디까지나 설정 정보를 담는 곳입니다. 실제 처리할 데이터는 반드시 state에 넣어야 합니다.

또한 runtime은 읽기 전용으로 사용해야 합니다. runtime 객체를 수정하려고 하면 예상치 못한 오류가 발생할 수 있습니다.

김개발 씨는 이제 확신이 생겼습니다. "state는 데이터, config는 설정, runtime은 환경이군요!" 박시니어 씨가 웃으며 대답했습니다.

"정확해요. 이걸 제대로 이해하면 훨씬 유연한 시스템을 만들 수 있어요." 세 가지 파라미터를 적재적소에 활용하면 확장 가능하고 유지보수하기 쉬운 LangGraph 애플리케이션을 만들 수 있습니다.

여러분도 다음 프로젝트에서 이 개념을 적용해 보세요.

실전 팁

💡 - 작업 데이터는 state, 환경 설정은 config에 분리하세요

  • runtime은 로깅과 추적에 매우 유용합니다
  • 모든 파라미터가 항상 필요한 것은 아니므로 필요한 것만 선언하세요

3. START 진입점 노드

김개발 씨가 여러 노드를 만들어 그래프에 추가했습니다. 그런데 실행하려니 어디서부터 시작해야 할지 막막했습니다.

"첫 번째 노드를 어떻게 지정하죠?" 박시니어 씨가 화이트보드에 그림을 그리며 설명했습니다. "그래프에는 시작점이 필요해요.

바로 START 노드죠."

START 노드는 LangGraph에서 워크플로우가 시작되는 특별한 진입점입니다. 마치 미로의 입구처럼, 모든 실행은 반드시 START에서 시작됩니다.

START는 실제 노드가 아니라 그래프의 시작을 표시하는 가상의 지점이며, 여기서 첫 번째 실제 노드로 연결됩니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    input_text: str
    processed: bool

def first_node(state: State) -> State:
    """그래프에서 가장 먼저 실행되는 노드"""
    print(f"첫 번째 노드 실행: {state['input_text']}")
    return {"processed": True}

# StateGraph 생성
graph = StateGraph(State)

# 노드 추가
graph.add_node("first_node", first_node)

# START에서 first_node로 연결 - 이것이 진입점입니다
graph.add_edge(START, "first_node")

# 컴파일
app = graph.compile()

김개발 씨는 드디어 완성된 노드들을 연결할 차례가 되었습니다. first_node, second_node, third_node를 만들어 놓고 보니 문득 궁금증이 생겼습니다.

"이 노드들을 순서대로 실행하려면 어떻게 해야 하지?" 박시니어 씨가 다가와 화면을 보더니 바로 문제를 파악했습니다. "아, START 연결을 안 했네요.

그래프는 어디서 시작할지 알아야 해요." START 노드란 정확히 무엇일까요? 쉽게 비유하자면, START는 마치 지하철 노선도의 출발역과 같습니다.

서울 지하철을 탈 때 어느 역에서 시작할지 정해야 하듯이, 그래프도 어느 노드에서 시작할지 명확히 지정해야 합니다. 차이점이 있다면 START는 실제 역이 아니라 노선도상의 표시일 뿐이라는 것입니다.

승객이 탑승하는 곳은 START가 아니라 START에 연결된 첫 번째 역입니다. 왜 START가 필요할까요?

초기 워크플로우 라이브러리들은 첫 번째 노드를 암묵적으로 정했습니다. 코드에 추가한 순서대로 첫 번째가 시작점이 되는 식이었죠.

하지만 이는 매우 혼란스러웠습니다. 코드를 리팩토링하면서 노드 추가 순서가 바뀌면 갑자기 다른 노드가 먼저 실행되었습니다.

더 큰 문제는 의도의 명확성이었습니다. 코드를 처음 보는 사람은 어떤 노드가 시작점인지 알 수 없었습니다.

주석을 달거나 별도 문서를 작성해야 했고, 시간이 지나면 문서와 코드가 따로 놀았습니다. 바로 이런 문제를 해결하기 위해 START라는 명시적 진입점이 도입되었습니다.

START를 사용하면 시작점이 명확해집니다. 코드를 보는 누구나 graph.add_edge(START, "어디")를 보고 바로 알 수 있습니다.

또한 순서 독립성을 얻습니다. 노드를 어떤 순서로 추가하든 START 연결만 보면 되죠.

무엇보다 실행 예측 가능성이라는 큰 장점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 StateGraph 객체를 생성합니다. 이것이 우리의 워크플로우 그래프입니다.

add_node로 실제 작업을 수행할 노드를 추가합니다. 가장 중요한 부분은 add_edge(START, "first_node")입니다.

이 한 줄이 그래프의 시작점을 정의합니다. START는 LangGraph에서 제공하는 특수 상수로, 임포트해서 사용합니다.

compile을 호출하면 실행 가능한 애플리케이션이 됩니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 주문 처리 시스템을 개발한다고 가정해봅시다. validate_order, process_payment, send_confirmation 같은 여러 노드가 있습니다.

START를 validate_order에 연결하면, 모든 주문은 반드시 검증부터 시작됩니다. 나중에 fraud_check 노드를 추가하더라도 START 연결은 변하지 않아 시작점이 일관됩니다.

이커머스 플랫폼들이 이런 명확한 진입점 패턴을 선호합니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수는 여러 노드를 START에 연결하려는 것입니다. START는 하나의 노드에만 연결되어야 합니다.

만약 여러 시작점이 필요하다면 조건부 엣지를 사용하거나 라우터 노드를 만들어야 합니다. 또한 START 자체를 노드처럼 사용하려고 하면 오류가 발생합니다.

김개발 씨는 이제 완벽히 이해했습니다. "아, START는 실제 노드가 아니라 시작 지점을 표시하는 거군요!" 박시니어 씨가 만족스럽게 고개를 끄덕였습니다.

"맞아요. 이제 그래프를 실행하면 정확히 first_node부터 시작될 거예요." START를 올바르게 사용하면 누구나 이해하기 쉬운 워크플로우를 만들 수 있습니다.

여러분의 다음 프로젝트에서 명확한 진입점을 정의해 보세요.

실전 팁

💡 - START는 항상 하나의 노드에만 연결하세요

  • 그래프를 그릴 때 START를 먼저 표시하면 흐름이 명확해집니다
  • START 연결을 보면 코드 리뷰 시 워크플로우를 빠르게 파악할 수 있습니다

4. END 종료 노드

그래프가 잘 실행되기 시작했습니다. 그런데 김개발 씨가 로그를 보니 이상한 점을 발견했습니다.

마지막 노드가 실행된 후에도 뭔가 계속 대기하고 있었습니다. "왜 끝나지 않죠?" 박시니어 씨가 웃으며 대답했습니다.

"종료점을 알려주지 않았으니까요. END가 필요해요."

END 노드는 LangGraph에서 워크플로우가 종료되는 지점을 표시합니다. START와 마찬가지로 실제 노드가 아닌 가상의 종료 지점입니다.

마치 마라톤의 결승선처럼, 어떤 노드가 END로 연결되면 그 노드가 실행된 후 워크플로우가 완료됩니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    count: int
    result: str

def process_node(state: State) -> State:
    """데이터를 처리하는 노드"""
    return {"count": state["count"] + 1}

def final_node(state: State) -> State:
    """마지막 작업을 수행하는 노드"""
    result = f"처리 완료: 총 {state['count']}번 실행"
    return {"result": result}

# 그래프 구성
graph = StateGraph(State)
graph.add_node("process", process_node)
graph.add_node("final", final_node)

# 연결: START -> process -> final -> END
graph.add_edge(START, "process")
graph.add_edge("process", "final")
graph.add_edge("final", END)  # 종료점 지정

app = graph.compile()

김개발 씨의 첫 번째 그래프가 드디어 실행되었습니다. 노드들이 순서대로 잘 실행되는 것을 보니 뿌듯했습니다.

그런데 뭔가 이상했습니다. 마지막 노드까지 실행되었는데도 프로그램이 종료되지 않았습니다.

터미널에는 계속 "실행 중..." 메시지가 떠 있었습니다. 김개발 씨는 당황해서 Ctrl+C를 눌러 강제 종료했습니다.

"뭐가 잘못된 거지?" 박시니어 씨가 다가와 코드를 살펴보더니 바로 문제를 찾아냈습니다. "END 연결이 없네요.

그래프는 어디서 끝나야 할지 모르고 있어요." END 노드란 정확히 무엇일까요? 쉽게 비유하자면, END는 마치 책의 마지막 페이지와 같습니다.

소설을 읽다가 "끝"이라는 표시가 없으면 다음 페이지를 계속 넘기게 됩니다. 페이지가 비어 있어도 혹시 뒷장에 내용이 있을까 확인하죠.

이처럼 그래프도 명시적인 종료 신호가 없으면 다음에 실행할 노드를 계속 기다립니다. END는 "더 이상 실행할 것이 없습니다"라고 선언하는 것입니다.

왜 END가 필요할까요? 초기 워크플로우 시스템에서는 암묵적으로 종료를 처리했습니다.

더 이상 연결된 노드가 없으면 자동으로 끝나는 식이었죠. 하지만 복잡한 그래프에서는 이것이 버그의 원인이 되었습니다.

개발자가 실수로 엣지를 빼먹으면 의도치 않게 워크플로우가 중간에 끝나버렸습니다. 더 큰 문제는 디버깅이었습니다.

워크플로우가 예상보다 일찍 끝났을 때, 그것이 의도된 것인지 실수인지 알 수 없었습니다. 로그를 뒤져가며 추적해야 했고, 원인을 찾는 데 몇 시간씩 걸렸습니다.

바로 이런 문제를 해결하기 위해 명시적인 END가 도입되었습니다. END를 사용하면 종료 지점이 명확해집니다.

코드를 보는 누구나 어떤 노드가 마지막인지 바로 알 수 있습니다. 또한 실수 방지가 가능합니다.

모든 경로가 END로 끝나는지 컴파일 시점에 확인할 수 있죠. 무엇보다 의도의 명시성이라는 큰 장점이 있습니다.

위의 코드를 한 줄씩 살펴보겠습니다. process_node와 final_node 두 개의 노드를 정의합니다.

그래프에 두 노드를 추가합니다. START에서 process로 엣지를 연결하여 시작점을 지정합니다.

process에서 final로 연결하여 순서를 정합니다. 가장 중요한 부분은 final에서 END로의 연결입니다.

이 한 줄이 "final_node가 마지막이다"라고 선언합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 데이터 파이프라인을 개발한다고 가정해봅시다. extract, transform, load 세 단계가 있습니다.

load 노드가 끝나면 파이프라인이 완료되어야 합니다. load에서 END로 연결하면, 데이터가 저장된 후 정확히 종료됩니다.

만약 나중에 notify 노드를 추가한다면, load에서 END로 가는 엣지를 제거하고 notify에서 END로 연결하면 됩니다. 데이터 엔지니어링팀들이 이런 명확한 종료 패턴을 선호합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수는 모든 노드를 END에 연결하는 것입니다.

END는 워크플로우의 최종 종료점에만 연결해야 합니다. 중간 노드들은 다음 노드로 연결되어야 하죠.

또 다른 실수는 조건부 분기에서 일부 경로에만 END를 연결하는 것입니다. 모든 가능한 경로가 결국 END로 이어지는지 확인해야 합니다.

김개발 씨는 코드를 수정해서 final_node에 END를 연결했습니다. 다시 실행하자 이번에는 깔끔하게 종료되었습니다.

"이제 제대로 끝나네요!" 박시니어 씨가 만족스럽게 웃었습니다. "좋아요.

START와 END로 시작과 끝을 명확히 하면 그래프가 훨씬 이해하기 쉬워져요." END를 올바르게 사용하면 예측 가능하고 안정적인 워크플로우를 만들 수 있습니다. 여러분도 다음 그래프를 만들 때 종료점을 명확히 정의해 보세요.

실전 팁

💡 - 모든 실행 경로가 최종적으로 END에 도달하는지 확인하세요

  • 조건부 분기가 있다면 각 분기의 끝에서 END 연결을 체크하세요
  • END 연결을 보면 그래프의 출구를 한눈에 파악할 수 있습니다

5. 동기/비동기 노드 함수

김개발 씨의 그래프가 점점 복잡해졌습니다. 어떤 노드는 외부 API를 호출하고, 어떤 노드는 데이터베이스를 조회했습니다.

그런데 실행 시간이 너무 오래 걸렸습니다. "API 응답을 기다리는 동안 다른 일을 할 수는 없나요?" 박시니어 씨가 고개를 끄덕였습니다.

"비동기 노드를 사용하면 돼요."

LangGraph 노드는 동기 함수와 비동기 함수 모두 지원합니다. 동기 노드는 작업이 끝날 때까지 기다리고, 비동기 노드는 대기 시간 동안 다른 작업을 수행할 수 있습니다.

마치 빨래를 돌리면서 설거지를 하는 것처럼, 비동기 노드는 효율적인 멀티태스킹을 가능하게 합니다.

다음 코드를 살펴봅시다.

import asyncio
from typing import TypedDict
from langgraph.graph import StateGraph

class State(TypedDict):
    user_id: str
    user_data: dict
    api_result: str

# 동기 노드: 순차적으로 실행됩니다
def sync_node(state: State) -> State:
    """간단한 계산은 동기로 처리합니다"""
    user_id = state["user_id"]
    return {"user_data": {"id": user_id, "processed": True}}

# 비동기 노드: 대기 시간 동안 다른 작업 가능
async def async_node(state: State) -> State:
    """외부 API 호출은 비동기로 처리합니다"""
    # await를 사용해 API 응답을 기다립니다
    await asyncio.sleep(1)  # API 호출 시뮬레이션
    result = f"API 응답: {state['user_id']}"
    return {"api_result": result}

# 두 가지 노드를 모두 사용할 수 있습니다
graph = StateGraph(State)
graph.add_node("sync", sync_node)
graph.add_node("async", async_node)

김개발 씨는 챗봇 서비스를 개발하고 있었습니다. 사용자 메시지를 받아서 여러 단계를 거쳐 응답을 생성하는 복잡한 워크플로우였습니다.

처음에는 모든 노드를 일반 함수로 만들었습니다. 그런데 문제가 생겼습니다.

외부 API를 호출하는 노드가 응답을 받을 때까지 3초가 걸렸고, 그동안 전체 워크플로우가 멈춰 있었습니다. 사용자는 답답하게 기다려야 했습니다.

박시니어 씨가 코드를 보더니 "비동기로 바꿔 보세요"라고 조언했습니다. 동기와 비동기 노드는 정확히 무엇이 다를까요?

쉽게 비유하자면, 동기 노드는 마치 은행 창구와 같습니다. 한 명의 고객이 업무를 마칠 때까지 다음 고객은 무조건 기다려야 합니다.

창구 직원은 한 번에 한 가지 일만 처리합니다. 반면 비동기 노드는 레스토랑 주방과 같습니다.

오븐에 피자를 넣어두고 굽는 동안 셰프는 샐러드를 준비할 수 있습니다. 대기 시간을 낭비하지 않고 다른 작업을 병렬로 수행합니다.

왜 비동기가 필요할까요? 초기 웹 애플리케이션들은 대부분 동기 방식이었습니다.

데이터베이스 쿼리를 실행하면 결과가 올 때까지 프로그램이 멈춰 있었죠. 한두 개의 요청을 처리할 때는 문제없었습니다.

하지만 사용자가 늘어나면서 심각한 병목이 발생했습니다. 더 큰 문제는 자원 낭비였습니다.

CPU는 놀고 있는데 네트워크 응답을 기다리며 시간만 보냈습니다. 서버를 더 추가해도 근본적인 해결책이 되지 못했습니다.

비용은 증가했지만 효율은 여전히 낮았습니다. 바로 이런 문제를 해결하기 위해 비동기 프로그래밍이 발전했습니다.

비동기 노드를 사용하면 I/O 대기 시간 활용이 가능해집니다. API 응답을 기다리는 동안 다른 노드를 실행할 수 있습니다.

또한 동시성 향상을 얻습니다. 여러 외부 요청을 동시에 보내고 결과를 모을 수 있죠.

무엇보다 응답 시간 단축이라는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

동기 노드는 일반 함수로 정의합니다. 간단한 계산이나 즉시 처리되는 작업에 적합합니다.

비동기 노드는 async def로 정의하고, I/O 작업에 await를 사용합니다. asyncio.sleep은 실제로는 API 호출이나 데이터베이스 쿼리가 될 것입니다.

LangGraph는 두 가지 유형을 자동으로 구분하여 처리합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 뉴스 요약 서비스를 개발한다고 가정해봅시다. 여러 뉴스 사이트에서 기사를 가져와야 합니다.

동기로 구현하면 첫 번째 사이트 응답을 받고, 그다음 두 번째 사이트 응답을 받는 식으로 순차 처리됩니다. 비동기로 구현하면 모든 사이트에 동시에 요청을 보내고 응답을 병렬로 받습니다.

전체 처리 시간이 획기적으로 줄어듭니다. AI 스타트업들이 이런 비동기 패턴을 적극 활용합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수는 모든 노드를 비동기로 만드는 것입니다.

간단한 계산이나 즉시 처리되는 작업까지 비동기로 만들면 오히려 오버헤드가 발생합니다. 비동기는 I/O 바운드 작업에만 사용하고, CPU 바운드 작업은 동기로 처리하는 것이 좋습니다.

또한 비동기 함수 안에서 await 없이 블로킹 작업을 하면 비동기의 이점이 사라집니다. 김개발 씨는 API 호출 노드를 비동기로 바꿨습니다.

실행 시간이 3초에서 1초로 줄어들었습니다. "와, 정말 빨라졌어요!" 박시니어 씨가 웃으며 대답했습니다.

"I/O 작업은 비동기가 정답이에요. 하지만 적재적소에 사용해야 합니다." 동기와 비동기를 올바르게 선택하면 빠르고 효율적인 워크플로우를 만들 수 있습니다.

여러분도 다음 프로젝트에서 I/O 작업을 비동기로 처리해 보세요.

실전 팁

💡 - I/O 작업(API, DB, 파일)은 비동기로, 계산 작업은 동기로 처리하세요

  • async 함수 안에서는 반드시 await를 사용해 비동기 작업을 기다리세요
  • LangGraph는 동기/비동기 노드를 자동으로 감지하므로 혼용해도 됩니다

6. 노드에서 상태 업데이트

김개발 씨의 그래프가 복잡해지면서 새로운 고민이 생겼습니다. 어떤 노드는 state의 일부만 수정하고, 어떤 노드는 기존 값을 유지하면서 새로운 필드를 추가해야 했습니다.

"state를 부분적으로 업데이트할 수 있나요?" 박시니어 씨가 자신 있게 대답했습니다. "물론이죠.

여러 가지 방법이 있어요."

노드에서 상태 업데이트는 전체 state를 반환하거나 일부만 반환하는 방식으로 이루어집니다. LangGraph는 반환된 딕셔너리를 기존 state와 자동으로 병합합니다.

마치 문서를 수정할 때 전체를 다시 쓰지 않고 바뀐 부분만 고치는 것처럼, 노드도 필요한 필드만 업데이트할 수 있습니다.

다음 코드를 살펴봅시다.

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

class State(TypedDict):
    messages: Annotated[list[str], add]  # 리스트는 합쳐집니다
    count: int  # 기본값은 덮어씁니다
    user_name: str

def partial_update_node(state: State) -> dict:
    """일부 필드만 업데이트합니다"""
    # count만 업데이트하고 나머지는 유지됩니다
    return {"count": state["count"] + 1}

def append_message_node(state: State) -> dict:
    """messages에 새 메시지를 추가합니다"""
    # Annotated[list, add] 덕분에 기존 메시지에 추가됩니다
    return {"messages": ["새 메시지"]}

def full_update_node(state: State) -> State:
    """여러 필드를 동시에 업데이트합니다"""
    return {
        "count": state["count"] + 1,
        "user_name": "김개발",
        "messages": ["안녕하세요"]
    }

김개발 씨는 채팅 히스토리를 관리하는 노드를 만들고 있었습니다. state에는 messages 리스트, 메시지 개수를 세는 count, 그리고 user_name이 있었습니다.

각 노드마다 업데이트해야 하는 필드가 달랐습니다. 처음에는 모든 노드에서 전체 state를 복사해서 반환했습니다.

하지만 코드가 너무 길어지고 실수하기 쉬웠습니다. "더 간단한 방법은 없을까요?" 박시니어 씨가 화면을 보며 설명했습니다.

"필요한 필드만 반환하면 돼요. LangGraph가 알아서 병합해 줍니다." 상태 업데이트는 정확히 어떻게 작동할까요?

쉽게 비유하자면, state 업데이트는 마치 화이트보드에 메모를 추가하는 것과 같습니다. 전체 화이트보드를 지우고 다시 쓸 필요 없이, 바뀐 부분만 덮어쓰거나 추가하면 됩니다.

"count: 5"를 "count: 6"으로 바꾸고 싶다면 count만 새로 쓰면 되고, 메시지를 추가하고 싶다면 messages 영역에 한 줄을 더하면 됩니다. 나머지 내용은 그대로 유지됩니다.

왜 부분 업데이트가 필요할까요? 초기 상태 관리 시스템에서는 항상 전체 state를 반환해야 했습니다.

한 필드만 바꾸려 해도 모든 필드를 포함한 객체를 만들어야 했죠. 코드가 길어지고 중복이 많았습니다.

더 심각한 문제는 실수였습니다. 어떤 필드를 빼먹으면 의도치 않게 그 값이 사라졌습니다.

또 다른 문제는 리스트와 딕셔너리 같은 컬렉션 타입이었습니다. 메시지 리스트에 한 개를 추가하고 싶은데, 기존 메시지를 모두 복사해서 새 메시지를 붙이는 코드를 매번 작성해야 했습니다.

번거로울 뿐만 아니라 성능도 낮았습니다. 바로 이런 문제를 해결하기 위해 부분 업데이트와 Reducer 패턴이 도입되었습니다.

부분 업데이트를 사용하면 코드가 간결해집니다. 바꿀 필드만 반환하면 되니까요.

Annotated를 활용하면 리스트나 딕셔너리를 자동으로 병합할 수 있습니다. 무엇보다 실수 방지라는 큰 장점이 있습니다.

위의 코드를 한 줄씩 살펴보겠습니다. State 정의에서 messages 필드에 Annotated[list[str], add]를 사용했습니다.

이것은 "messages는 덮어쓰지 말고 기존 값에 add 연산을 적용하라"는 의미입니다. partial_update_node는 count만 반환합니다.

user_name과 messages는 자동으로 유지됩니다. append_message_node는 messages만 반환하는데, add Reducer 덕분에 기존 메시지에 추가됩니다.

full_update_node는 여러 필드를 동시에 업데이트하는 예시입니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 문서 처리 파이프라인을 개발한다고 가정해봅시다. state에는 원본 텍스트, 처리 단계별 결과, 메타데이터가 있습니다.

첫 번째 노드는 원본을 정제하고, 두 번째 노드는 요약을 생성하고, 세 번째 노드는 메타데이터를 추가합니다. 각 노드는 자신이 담당하는 필드만 업데이트하면 됩니다.

코드가 깔끔해지고 책임이 명확해집니다. 문서 AI 서비스들이 이런 패턴을 널리 사용합니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수는 Reducer를 잘못 사용하는 것입니다.

리스트에 add를 쓰면 합쳐지지만, 숫자에 add를 쓰면 덮어쓰기가 아니라 더하기가 됩니다. 각 Reducer의 동작을 정확히 이해해야 합니다.

또한 None을 반환하면 state가 업데이트되지 않으니 주의하세요. 빈 딕셔너리 {}를 반환하는 것과는 다릅니다.

김개발 씨는 코드를 리팩토링해서 각 노드가 필요한 필드만 반환하도록 바꿨습니다. 코드가 절반으로 줄어들었습니다.

"훨씬 깔끔해졌어요!" 박시니어 씨가 만족스럽게 고개를 끄덕였습니다. "맞아요.

각 노드가 자기 책임만 지면 코드가 훨씬 이해하기 쉬워져요." 상태 업데이트 패턴을 올바르게 사용하면 간결하고 유지보수하기 쉬운 노드를 만들 수 있습니다. 여러분도 다음 프로젝트에서 부분 업데이트를 적극 활용해 보세요.

실전 팁

💡 - 바꿀 필드만 반환하여 코드를 간결하게 유지하세요

  • 리스트나 딕셔너리는 Annotated와 Reducer를 활용하세요
  • 각 노드가 자신의 책임 범위만 업데이트하도록 설계하세요

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

#LangGraph#Node#State#Graph#Workflow#AI,LLM,Python,LangGraph

이 카드뉴스가 포함된 코스

LangGraph 완전 정복

장시간 실행되는 상태 유지 에이전트를 구축하기 위한 저수준 오케스트레이션 프레임워크 LangGraph를 완벽하게 마스터합니다. StateGraph, 노드, 엣지, Persistence, Human-in-the-Loop, 서브그래프까지 모든 핵심 개념을 다룹니다.

advanced14개 카드뉴스약 22시간

댓글 (0)

댓글을 작성하려면 로그인이 필요합니다.

함께 보면 좋은 카드 뉴스

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

Kubernetes 환경에서 마이크로서비스를 효율적으로 패키징하고 배포하는 Helm의 핵심 기능을 실무 중심으로 학습합니다. Chart 생성부터 릴리스 관리까지 체계적으로 다룹니다.

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

프로젝트의 보안을 처음부터 설계하는 방법을 배웁니다. AWS 환경에서 VPC부터 WAF, 암호화, 접근 제어까지 실무에서 바로 적용할 수 있는 보안 아키텍처를 단계별로 구성해봅니다.

AWS Organizations 완벽 가이드

여러 AWS 계정을 체계적으로 관리하고 통합 결제와 보안 정책을 적용하는 방법을 실무 스토리로 쉽게 배워봅니다. 초보 개발자도 바로 이해할 수 있는 친절한 설명과 실전 예제를 제공합니다.

AWS KMS 암호화 완벽 가이드

AWS KMS(Key Management Service)를 활용한 클라우드 데이터 암호화 방법을 초급 개발자를 위해 쉽게 설명합니다. CMK 생성부터 S3, EBS 암호화, 봉투 암호화까지 실무에 필요한 모든 내용을 담았습니다.

AWS Secrets Manager 완벽 가이드

AWS에서 데이터베이스 비밀번호, API 키 등 민감한 정보를 안전하게 관리하는 Secrets Manager의 핵심 개념과 실무 활용법을 배워봅니다. 초급 개발자도 쉽게 따라할 수 있도록 실전 예제와 함께 설명합니다.

이전4/14다음