🤖

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

⚠️

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

이미지 로딩 중...

LangGraph Edges와 라우팅 완벽 가이드 - 슬라이드 1/7
A

AI Generated

2025. 12. 12. · 12 Views

LangGraph Edges와 라우팅 완벽 가이드

LangGraph에서 노드를 연결하는 다양한 엣지 유형과 라우팅 전략을 배웁니다. 일반 엣지부터 조건부 엣지, 동적 엣지까지 실무에 필요한 모든 개념을 다룹니다.


목차

  1. Normal Edges 일반 엣지
  2. Conditional Edges 조건부 엣지
  3. 라우팅 함수 작성
  4. Entry Points 진입점 설정
  5. Send 동적 엣지
  6. Command 제어흐름과 상태 업데이트

1. Normal Edges 일반 엣지

어느 날 김개발 씨가 LangGraph로 첫 워크플로우를 만들던 중이었습니다. 노드는 만들었는데, 이 노드들을 어떻게 연결해야 할지 막막했습니다.

선배 박시니어 씨가 다가와 말했습니다. "노드 연결은 엣지로 하는 거예요.

가장 기본은 일반 엣지입니다."

일반 엣지는 LangGraph에서 두 노드를 직선으로 연결하는 가장 기본적인 방법입니다. 마치 지하철 노선도에서 역과 역을 잇는 선처럼, 노드 A에서 노드 B로 무조건 이동하게 만듭니다.

조건 없이 항상 다음 노드로 진행하므로 가장 단순하고 명확한 흐름을 만들 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph

# 상태 정의
class AgentState(dict):
    messages: list

# 그래프 생성
workflow = StateGraph(AgentState)

# 노드 추가
workflow.add_node("input_processor", process_input)
workflow.add_node("analyzer", analyze_data)

# 일반 엣지로 연결: input_processor 다음은 항상 analyzer
workflow.add_edge("input_processor", "analyzer")

김개발 씨는 AI 에이전트 시스템을 만들고 있었습니다. 사용자 입력을 받아서 분석하고, 결과를 출력하는 간단한 시스템이었습니다.

노드는 세 개를 만들었는데, 이것들을 어떻게 연결해야 할지 고민이었습니다. "선배님, 노드는 만들었는데 연결을 어떻게 하죠?" 김개발 씨가 물었습니다.

박시니어 씨가 화면을 보더니 말했습니다. "노드 연결은 엣지로 합니다.

가장 기본은 일반 엣지예요." 일반 엣지란 무엇일까요? 쉽게 비유하자면, 일반 엣지는 마치 일방통행 도로와 같습니다. A 지점에서 출발하면 무조건 B 지점으로 가는 것처럼, 노드 A가 실행되면 항상 노드 B로 이동합니다.

신호등도 없고, 다른 길로 빠질 수도 없는 직선 도로입니다. LangGraph에서 노드는 작업 단위를 의미합니다.

하나의 노드가 끝나면 다음 노드로 넘어가야 하는데, 이때 두 노드를 연결하는 것이 바로 엣지입니다. 엣지가 없으면 어떻게 될까요? LangGraph에서 노드만 만들고 엣지로 연결하지 않으면 어떻게 될까요?

각 노드는 고립된 섬처럼 떠 있게 됩니다. 아무리 훌륭한 기능을 가진 노드를 만들어도, 연결되지 않으면 실행 흐름이 만들어지지 않습니다.

마치 자동차 공장에서 각 작업장은 완벽하게 준비되었는데 컨베이어 벨트가 없는 것과 같습니다. 부품을 가공하는 작업장, 조립하는 작업장, 검사하는 작업장이 있어도 이들을 연결하는 벨트가 없으면 생산 라인이 작동하지 않습니다.

일반 엣지의 등장 바로 이런 문제를 해결하는 것이 일반 엣지입니다. add_edge 메서드를 사용하면 두 노드를 간단히 연결할 수 있습니다.

첫 번째 인자는 출발 노드의 이름, 두 번째 인자는 도착 노드의 이름입니다. 이렇게 연결하면 출발 노드의 실행이 끝나는 순간, 자동으로 도착 노드가 시작됩니다.

무엇보다 큰 장점은 단순함입니다. 복잡한 조건도 필요 없고, 추가 설정도 필요 없습니다.

그냥 "이 노드 다음은 저 노드"라고 명확하게 선언하면 끝입니다. 코드 살펴보기 위의 코드를 자세히 살펴보겠습니다.

먼저 StateGraph로 워크플로우 그래프를 생성합니다. 이것이 우리의 작업 공간입니다.

다음으로 add_node 메서드로 두 개의 노드를 추가했습니다. "input_processor"는 입력을 처리하고, "analyzer"는 데이터를 분석합니다.

핵심은 마지막 줄입니다. **workflow.add_edge("input_processor", "analyzer")**라고 작성하면, input_processor 노드가 끝나면 무조건 analyzer 노드로 이동합니다.

조건도 없고, 다른 선택지도 없습니다. 항상 이 경로를 따릅니다.

실무에서는 어떻게 활용할까요? 예를 들어 고객 문의 처리 시스템을 만든다고 가정해봅시다. 사용자가 질문을 입력하면, 먼저 입력을 정제하는 노드가 실행됩니다.

그 다음은 항상 자연어 이해 노드로 가야 합니다. 이런 경우 일반 엣지가 완벽한 선택입니다.

또 다른 예로, 문서 처리 파이프라인을 생각해볼 수 있습니다. 문서를 받으면 먼저 포맷을 변환하고, 그 다음 텍스트를 추출하고, 마지막으로 저장합니다.

이런 순차적인 흐름은 일반 엣지로 깔끔하게 표현됩니다. 주의할 점 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수는 모든 연결을 일반 엣지로만 처리하려는 것입니다. 조건에 따라 다른 경로로 가야 하는 상황에서도 일반 엣지만 사용하면, 로직이 복잡해지고 유지보수가 어려워집니다.

이럴 때는 조건부 엣지를 사용해야 합니다. 또 하나, 일반 엣지는 순환을 만들 수도 있습니다.

A에서 B로, B에서 다시 A로 연결하면 무한 루프가 발생할 수 있으니 주의해야 합니다. 마무리 김개발 씨가 코드를 작성해보니 정말 간단했습니다.

"이렇게 쉬운 거였어요?" 박시니어 씨가 웃으며 말했습니다. "일반 엣지는 가장 기본이지만, 가장 많이 쓰이는 연결 방법이에요." 일반 엣지를 제대로 이해하면 LangGraph의 기본 흐름을 자유롭게 만들 수 있습니다.

여러분도 오늘 배운 내용을 바탕으로 첫 워크플로우를 만들어보세요.

실전 팁

💡 - 순차적으로 실행되어야 하는 작업은 일반 엣지로 연결하세요

  • 순환 연결은 무한 루프를 만들 수 있으니 주의하세요
  • 조건이 필요한 경우는 조건부 엣지를 사용하세요

2. Conditional Edges 조건부 엣지

김개발 씨가 일반 엣지로 워크플로우를 만들고 있었습니다. 그런데 문제가 생겼습니다.

"선배님, 사용자 입력에 따라 다른 노드로 가야 하는데, 일반 엣지로는 안 되네요." 박시니어 씨가 말했습니다. "그럴 때 쓰는 게 조건부 엣지입니다."

조건부 엣지는 상태나 조건에 따라 다음에 실행할 노드를 동적으로 결정하는 엣지입니다. 마치 신호등이 있는 교차로처럼, 현재 상황을 판단해서 왼쪽으로 갈지 오른쪽으로 갈지 결정합니다.

이를 통해 복잡한 분기 로직을 깔끔하게 표현할 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph, END

def route_by_intent(state):
    # 사용자 의도에 따라 다음 노드 결정
    intent = state.get("intent")
    if intent == "question":
        return "qa_node"
    elif intent == "command":
        return "action_node"
    else:
        return END

workflow = StateGraph(AgentState)
workflow.add_node("classifier", classify_intent)
workflow.add_node("qa_node", handle_question)
workflow.add_node("action_node", handle_action)

# 조건부 엣지: classifier 다음은 intent에 따라 결정
workflow.add_conditional_edges(
    "classifier",
    route_by_intent,
    {"qa_node": "qa_node", "action_node": "action_node", END: END}
)

김개발 씨는 챗봇 시스템을 만들고 있었습니다. 사용자가 질문을 하면 QA 엔진으로, 명령을 내리면 액션 실행 엔진으로 보내야 했습니다.

일반 엣지로는 이런 분기를 만들 수 없었습니다. "선배님, 이거 어떻게 해야 하죠?

일반 엣지는 항상 같은 곳으로만 가잖아요." 박시니어 씨가 키보드를 가져오며 말했습니다. "이럴 때 쓰는 게 조건부 엣지예요.

조건에 따라 다른 경로로 갈 수 있습니다." 조건부 엣지란 무엇일까요? 쉽게 비유하자면, 조건부 엣지는 마치 내비게이션이 있는 교차로와 같습니다. 교차로에 도착하면 내비게이션이 현재 상황을 판단합니다.

교통 상황이 좋으면 고속도로로, 막히면 국도로 안내하는 것처럼, 조건부 엣지도 현재 상태를 보고 어느 노드로 갈지 결정합니다. 일반 엣지가 일방통행 도로라면, 조건부 엣지는 여러 갈래가 있는 분기점입니다.

어느 길로 갈지는 그 순간의 상황이 결정합니다. 왜 조건부 엣지가 필요할까요? 실제 AI 시스템은 단순한 직선 흐름이 아닙니다.

사용자 입력의 성격에 따라, 또는 이전 단계의 결과에 따라 다른 처리를 해야 합니다. 예를 들어 사용자가 "날씨 알려줘"라고 하면 날씨 API를 호출해야 하고, "계산 좀 해줘"라고 하면 계산 엔진을 실행해야 합니다.

이런 분기를 코드에 하드코딩하면 복잡해지고 유지보수가 어렵습니다. 조건부 엣지의 작동 원리 조건부 엣지는 세 가지 요소로 구성됩니다.

첫째, 출발 노드입니다. 이 노드가 끝나면 조건부 엣지가 활성화됩니다.

둘째, 라우팅 함수입니다. 이 함수가 현재 상태를 받아서 다음에 갈 노드의 이름을 반환합니다.

셋째, 노드 매핑입니다. 라우팅 함수가 반환한 이름과 실제 노드를 연결하는 사전입니다.

add_conditional_edges 메서드를 사용하면 이 세 가지를 한 번에 설정할 수 있습니다. 첫 번째 인자는 출발 노드, 두 번째는 라우팅 함수, 세 번째는 노드 매핑입니다.

코드 자세히 보기 위의 코드를 단계별로 살펴보겠습니다. 먼저 route_by_intent 함수를 정의했습니다.

이 함수는 state를 받아서 "intent" 값을 확인합니다. 질문이면 "qa_node"를, 명령이면 "action_node"를 반환합니다.

그 외의 경우는 END를 반환해서 워크플로우를 종료합니다. 다음으로 add_conditional_edges 메서드를 호출합니다.

"classifier" 노드가 끝나면 route_by_intent 함수가 실행됩니다. 이 함수가 "qa_node"를 반환하면 qa_node로, "action_node"를 반환하면 action_node로 이동합니다.

세 번째 인자인 노드 매핑을 주목하세요. 이것은 라우팅 함수의 반환값과 실제 노드를 연결하는 사전입니다.

꼭 필요한 것은 아니지만, 명확성을 위해 작성하는 것이 좋습니다. 실무 활용 사례 실제 프로젝트에서 조건부 엣지는 매우 자주 사용됩니다.

고객 지원 챗봇을 만든다고 가정해봅시다. 사용자 메시지를 분류한 후, 기술 문의면 기술팀 노드로, 환불 요청이면 환불 처리 노드로, 일반 질문이면 FAQ 노드로 보내야 합니다.

이런 다중 분기는 조건부 엣지 없이는 구현하기 어렵습니다. 또 다른 예로, 문서 처리 시스템을 생각해볼 수 있습니다.

PDF면 PDF 파서로, 이미지면 OCR 노드로, 텍스트면 바로 분석 노드로 보내는 라우팅이 필요합니다. 조건부 엣지로 깔끔하게 처리할 수 있습니다.

흔한 실수 초보 개발자들이 자주 하는 실수는 라우팅 함수에서 잘못된 값을 반환하는 것입니다. 라우팅 함수는 반드시 노드 매핑에 존재하는 키를 반환해야 합니다.

만약 "unknown_node"를 반환했는데 노드 매핑에 이 키가 없으면 에러가 발생합니다. 따라서 모든 가능한 반환값을 노드 매핑에 포함시켜야 합니다.

또 하나, 라우팅 함수는 순수 함수여야 합니다. 즉, 같은 상태를 받으면 항상 같은 결과를 반환해야 합니다.

랜덤 값을 사용하거나 외부 API를 호출하면 예측 불가능한 동작이 발생할 수 있습니다. 정리하며 김개발 씨가 조건부 엣지를 적용하자 챗봇이 훨씬 똑똑해졌습니다.

"와, 이제 질문과 명령을 구분해서 처리하네요!" 박시니어 씨가 웃으며 말했습니다. "조건부 엣지는 복잡한 흐름의 핵심이에요." 조건부 엣지를 마스터하면 훨씬 더 유연한 AI 워크플로우를 만들 수 있습니다.

여러분의 시스템에 분기 로직이 필요하다면 조건부 엣지를 활용해보세요.

실전 팁

💡 - 라우팅 함수는 반드시 노드 매핑에 있는 키를 반환해야 합니다

  • END를 반환하면 워크플로우가 즉시 종료됩니다
  • 라우팅 함수는 순수 함수로 작성하세요

3. 라우팅 함수 작성

김개발 씨가 조건부 엣지를 만들던 중 막혔습니다. "선배님, 라우팅 함수를 어떻게 작성해야 하죠?

뭔가 복잡해 보이는데요." 박시니어 씨가 말했습니다. "라우팅 함수는 조건부 엣지의 두뇌예요.

잘 설계하는 게 중요합니다."

라우팅 함수는 조건부 엣지에서 현재 상태를 분석해 다음 노드를 결정하는 함수입니다. 마치 교통 경찰이 차량을 보고 어느 차선으로 가라고 지시하는 것처럼, 상태 객체를 받아서 다음 목적지의 이름을 반환합니다.

잘 작성된 라우팅 함수는 워크플로우의 지능을 좌우합니다.

다음 코드를 살펴봅시다.

from typing import Literal

def advanced_routing(state) -> Literal["continue", "review", "end"]:
    """고급 라우팅 함수 예제"""
    # 메시지 개수 확인
    message_count = len(state.get("messages", []))

    # 에러 상태 확인
    has_error = state.get("error", False)

    # 완료 여부 확인
    is_complete = state.get("complete", False)

    if has_error:
        return "review"  # 에러가 있으면 검토 노드로
    elif is_complete:
        return "end"  # 완료되면 종료
    elif message_count < 10:
        return "continue"  # 아직 진행 중
    else:
        return "review"  # 메시지가 많으면 검토

김개발 씨는 조건부 엣지의 개념은 이해했지만, 라우팅 함수를 어떻게 작성해야 할지 막막했습니다. 간단한 if-else는 작성할 수 있지만, 실무에서는 더 복잡한 로직이 필요할 것 같았습니다.

"선배님, 라우팅 함수에 뭘 넣어야 하죠?" 박시니어 씨가 자리에 앉으며 말했습니다. "라우팅 함수는 조건부 엣지의 핵심이에요.

이것만 잘 만들면 나머지는 자동으로 굴러갑니다." 라우팅 함수의 역할 라우팅 함수는 말 그대로 경로를 결정하는 함수입니다. 쉽게 비유하자면, 공항의 출국장 직원과 같습니다.

승객의 티켓을 보고 "이분은 A 게이트로, 저분은 B 게이트로" 안내하는 것처럼, 라우팅 함수는 상태를 보고 다음 노드를 안내합니다. 입력은 항상 하나입니다.

바로 현재 상태 객체입니다. 이 상태 객체에는 워크플로우의 모든 정보가 담겨 있습니다.

메시지, 변수, 플래그 등 모든 것이 들어있습니다. 출력도 명확합니다.

다음에 실행할 노드의 이름을 문자열로 반환합니다. 이 이름은 반드시 add_conditional_edges에서 정의한 노드 매핑에 존재해야 합니다.

좋은 라우팅 함수의 조건 좋은 라우팅 함수는 몇 가지 특징이 있습니다. 첫째, 명확성입니다.

코드를 읽는 사람이 한눈에 어떤 조건에서 어디로 가는지 이해할 수 있어야 합니다. 복잡한 중첩 if문보다는 명확한 조건 블록이 좋습니다.

둘째, 완전성입니다. 모든 가능한 상태에 대해 반환값이 정의되어야 합니다.

예상하지 못한 상태가 들어왔을 때 에러가 나지 않도록 기본값을 준비해야 합니다. 셋째, 예측 가능성입니다.

같은 상태가 들어오면 항상 같은 결과를 반환해야 합니다. 랜덤성이나 외부 의존성은 피하는 것이 좋습니다.

타입 힌트 활용하기 위의 코드에서 Literal 타입 힌트를 사용한 것을 보셨나요? 이것은 매우 유용한 기법입니다.

Literal을 사용하면 반환할 수 있는 값을 명시적으로 제한할 수 있습니다. "continue", "review", "end" 중 하나만 반환할 수 있다고 선언하는 것입니다.

이렇게 하면 타입 체커가 오타나 잘못된 반환값을 미리 잡아줍니다. 예를 들어 "continu"라고 오타를 내면, IDE가 즉시 경고를 표시합니다.

런타임 에러가 발생하기 전에 미리 잡을 수 있는 것입니다. 복잡한 조건 처리하기 실무에서는 단순한 if-else보다 복잡한 로직이 필요할 때가 많습니다.

위의 코드를 다시 보겠습니다. 여러 조건을 순차적으로 확인합니다.

먼저 에러가 있는지 확인하고, 그 다음 완료 여부를 확인하고, 마지막으로 메시지 개수를 확인합니다. 이런 순차적 검사는 우선순위를 명확히 합니다.

에러가 있으면 다른 조건과 상관없이 바로 review 노드로 갑니다. 이것은 에러 처리를 최우선으로 한다는 정책입니다.

이런 우선순위 로직을 코드 순서로 표현하면 이해하기 쉽습니다. 상태 접근 패턴 상태에서 값을 가져올 때는 항상 get 메서드를 사용하는 것이 좋습니다.

**state.get("messages", [])**처럼 기본값을 제공하면, 해당 키가 없어도 에러가 나지 않습니다. 대신 빈 리스트가 반환됩니다.

이런 방어적 프로그래밍이 안정적인 라우팅 함수를 만듭니다. 만약 state["messages"]처럼 직접 접근하면, 키가 없을 때 KeyError가 발생합니다.

워크플로우 전체가 멈출 수 있습니다. 실무 활용 예시 실제 프로젝트에서는 어떻게 활용할까요?

고객 문의 처리 시스템을 생각해봅시다. 사용자 메시지의 감정을 분석한 후, 부정적이면 우선 처리 큐로, 긍정적이면 일반 처리 큐로, 중립적이면 자동 응답 노드로 보낼 수 있습니다.

라우팅 함수에서 sentiment 점수를 확인해서 분기하면 됩니다. 또 다른 예로, 문서 변환 파이프라인에서 파일 크기를 확인할 수 있습니다.

10MB 이하면 바로 처리, 그 이상이면 분할 처리 노드로 보내는 라우팅이 가능합니다. 주의사항 라우팅 함수에서 절대 하지 말아야 할 것들이 있습니다.

첫째, 상태를 직접 수정하지 마세요. 라우팅 함수는 읽기 전용으로 상태를 다뤄야 합니다.

상태 수정은 노드에서 해야 합니다. 만약 라우팅 함수에서 상태를 바꾸면 예측 불가능한 동작이 발생합니다.

둘째, 외부 API를 호출하지 마세요. 라우팅은 빠르게 결정되어야 합니다.

네트워크 호출이 들어가면 전체 워크플로우가 느려집니다. 외부 데이터가 필요하면 이전 노드에서 미리 가져와 상태에 저장하세요.

셋째, 예외를 던지지 마세요. 라우팅 함수에서 예외가 발생하면 워크플로우가 중단됩니다.

대신 에러 상황도 하나의 경로로 처리하세요. 예를 들어 "error" 노드로 라우팅하는 것입니다.

정리 김개발 씨가 라우팅 함수를 여러 개 작성해보니 감이 잡혔습니다. "이제 알겠어요.

라우팅 함수는 깔끔하고 명확하게!" 박시니어 씨가 고개를 끄덕였습니다. "맞아요.

복잡한 로직은 노드에 넣고, 라우팅은 단순하게 유지하세요." 좋은 라우팅 함수는 워크플로우의 품질을 결정합니다. 여러분도 명확하고 안전한 라우팅 함수를 작성해보세요.

실전 팁

💡 - Literal 타입 힌트로 반환값을 명시하세요

  • state.get()으로 안전하게 값에 접근하세요
  • 라우팅 함수는 순수 함수로 유지하세요

4. Entry Points 진입점 설정

김개발 씨가 노드와 엣지를 모두 만들었습니다. 그런데 그래프를 실행하려니 에러가 났습니다.

"선배님, 어디서부터 시작해야 하는지 모르겠대요." 박시니어 씨가 말했습니다. "아, 진입점을 설정 안 했구나.

그래프의 시작점을 알려줘야 해요."

진입점은 LangGraph 워크플로우가 시작되는 첫 번째 노드를 지정하는 설정입니다. 마치 프로그램의 main 함수처럼, 그래프 실행이 시작되면 가장 먼저 실행될 노드를 정의합니다.

진입점이 없으면 그래프는 어디서부터 시작해야 할지 알 수 없어 실행되지 않습니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# 노드들 추가
workflow.add_node("input_handler", handle_input)
workflow.add_node("processor", process_data)
workflow.add_node("output_handler", handle_output)

# 엣지들 연결
workflow.add_edge("input_handler", "processor")
workflow.add_edge("processor", "output_handler")
workflow.add_edge("output_handler", END)

# 진입점 설정: input_handler가 시작점
workflow.set_entry_point("input_handler")

# 그래프 컴파일
app = workflow.compile()

# 실행: 자동으로 input_handler부터 시작
result = app.invoke({"messages": ["Hello"]})

김개발 씨는 완벽한 그래프를 만들었다고 생각했습니다. 노드도 세 개 만들었고, 엣지로 잘 연결했습니다.

그런데 막상 실행하려니 이런 에러가 나왔습니다. "Error: No entry point set for the graph." 무슨 뜻인지 몰라 선배에게 물었습니다.

"선배님, 진입점이 뭔가요?" 진입점이란 무엇일까요? 진입점은 쉽게 말해 출발선입니다. 마치 마라톤 경기에서 선수들이 서는 출발선처럼, 그래프 실행이 시작되는 지점입니다.

아무리 완벽한 경기장을 만들어도 출발선이 없으면 경기를 시작할 수 없는 것처럼, 그래프도 진입점이 없으면 실행할 수 없습니다. 박시니어 씨가 비유를 들어 설명했습니다.

"자동차 여행을 계획했다고 생각해봐요. 중간 경유지들은 다 정했는데, 출발지를 안 정했으면 어떻게 되겠어요?

당연히 여행을 시작할 수 없죠." 왜 진입점이 필요할까요? LangGraph는 그래프 구조입니다. 여러 노드가 복잡하게 연결되어 있을 수 있습니다.

어떤 노드는 다른 노드로부터 엣지를 받지만, 어떤 노드는 진입 엣지가 없을 수도 있습니다. 시스템 입장에서는 어느 노드부터 실행해야 할지 알 수 없습니다.

모든 노드가 똑같아 보이기 때문입니다. 진입점을 명시적으로 지정해야만 "아, 여기서부터 시작하는구나"라고 알 수 있습니다.

실무에서는 보통 입력을 받는 노드가 진입점이 됩니다. 사용자 요청을 받아서 전처리하는 노드, 초기 데이터를 로드하는 노드 등이 진입점으로 많이 사용됩니다.

진입점 설정 방법 진입점 설정은 매우 간단합니다. set_entry_point 메서드를 호출하고, 시작 노드의 이름을 문자열로 전달하면 됩니다.

이 메서드는 반드시 compile 메서드를 호출하기 전에 실행해야 합니다. 컴파일 후에는 그래프 구조를 변경할 수 없습니다.

위의 코드를 보면, "input_handler"를 진입점으로 설정했습니다. 이제 그래프를 실행하면 항상 input_handler 노드부터 시작됩니다.

다른 노드로 직접 시작할 수는 없습니다. 코드 흐름 이해하기 전체 코드의 흐름을 따라가 보겠습니다.

먼저 StateGraph를 만들고 세 개의 노드를 추가합니다. 그 다음 엣지로 노드들을 순차적으로 연결합니다.

input_handler에서 processor로, processor에서 output_handler로, 마지막으로 output_handler에서 END로 연결됩니다. 여기까지는 그래프 구조만 정의한 것입니다.

아직 실행 가능한 상태가 아닙니다. set_entry_point를 호출해야 비로소 "여기서 시작한다"는 정보가 추가됩니다.

그 다음 compile 메서드를 호출하면, 그래프가 실행 가능한 애플리케이션으로 변환됩니다. 마지막으로 invoke 메서드로 실행하면, 자동으로 input_handler부터 시작해서 정의된 경로를 따라 실행됩니다.

여러 진입점을 가질 수 있을까요? 결론부터 말하면, 한 그래프는 하나의 진입점만 가질 수 있습니다. 만약 상황에 따라 다른 노드에서 시작하고 싶다면, 그래프를 여러 개 만들거나, 초기 라우팅 노드를 진입점으로 설정하는 방법이 있습니다.

예를 들어 "router" 노드를 진입점으로 만들고, 이 노드에서 조건부 엣지로 실제 처리 노드들로 분기시키는 것입니다. 이렇게 하면 하나의 진입점으로 여러 경로를 처리할 수 있습니다.

실무 활용 패턴 실제 프로젝트에서는 보통 이런 패턴을 사용합니다. 첫째, 초기화 노드를 진입점으로 합니다.

이 노드에서 필요한 리소스를 로드하고, 상태를 초기화하고, 로깅을 시작합니다. 그 다음 실제 비즈니스 로직 노드들로 이어집니다.

둘째, 검증 노드를 진입점으로 합니다. 사용자 입력을 먼저 검증하고, 문제가 있으면 에러 노드로, 문제가 없으면 처리 노드로 보내는 패턴입니다.

이렇게 하면 잘못된 입력으로 인한 오류를 조기에 차단할 수 있습니다. 셋째, 라우터 노드를 진입점으로 합니다.

요청 타입을 먼저 분류하고, 타입에 맞는 처리 파이프라인으로 보내는 패턴입니다. 하나의 그래프로 여러 종류의 요청을 처리할 수 있습니다.

흔한 실수 초보자들이 자주 하는 실수가 있습니다. 첫째, 진입점을 설정하지 않는 것입니다.

노드와 엣지만 만들고 진입점을 빼먹으면 실행 시 에러가 납니다. 반드시 set_entry_point를 호출해야 합니다.

둘째, 존재하지 않는 노드를 진입점으로 설정하는 것입니다. 오타로 "input_hander"라고 쓰면, 컴파일 시 "노드를 찾을 수 없다"는 에러가 발생합니다.

노드 이름은 정확히 일치해야 합니다. 셋째, 진입점 노드로의 엣지를 만드는 것입니다.

진입점은 시작 노드이므로, 다른 노드에서 이 노드로 오는 엣지가 있으면 안 됩니다. 진입점은 외부에서만 호출되어야 합니다.

정리하며 김개발 씨가 set_entry_point를 추가하자 그래프가 완벽하게 작동했습니다. "와, 이제 제대로 실행되네요!" 박시니어 씨가 말했습니다.

"진입점은 작지만 필수적인 설정이에요. 항상 잊지 마세요." 진입점 설정은 간단하지만 매우 중요합니다.

여러분의 그래프에도 명확한 시작점을 만들어주세요.

실전 팁

💡 - 진입점은 compile 전에 반드시 설정하세요

  • 한 그래프는 하나의 진입점만 가질 수 있습니다
  • 진입점 노드는 다른 노드로부터 엣지를 받지 않습니다

5. Send 동적 엣지

김개발 씨가 복잡한 문제에 부딪혔습니다. "선배님, 하나의 노드에서 여러 노드를 동시에 실행하고 싶은데요.

엣지는 하나만 연결할 수 있잖아요?" 박시니어 씨가 웃으며 말했습니다. "그럴 때 쓰는 게 Send입니다.

동적으로 여러 경로를 만들 수 있어요."

Send는 런타임에 동적으로 여러 노드를 병렬로 실행할 수 있게 해주는 특별한 엣지입니다. 마치 우편 배달원이 한 번에 여러 주소로 편지를 보내는 것처럼, 하나의 노드에서 여러 노드로 동시에 실행 흐름을 보낼 수 있습니다.

이를 통해 팬아웃 패턴과 병렬 처리를 구현할 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.graph import StateGraph, Send, END

def split_work(state):
    # 작업을 여러 개로 분할
    tasks = state.get("tasks", [])

    # Send 객체 리스트를 반환하여 병렬 실행
    return [
        Send("worker", {"task": task, "index": i})
        for i, task in enumerate(tasks)
    ]

workflow = StateGraph(AgentState)
workflow.add_node("splitter", prepare_tasks)
workflow.add_node("worker", process_task)
workflow.add_node("aggregator", combine_results)

# 동적 엣지: splitter가 여러 worker를 병렬 실행
workflow.add_conditional_edges("splitter", split_work)
workflow.add_edge("worker", "aggregator")
workflow.add_edge("aggregator", END)

김개발 씨는 대량의 문서를 처리하는 시스템을 만들고 있었습니다. 문서 하나당 처리 시간이 오래 걸려서, 순차적으로 처리하면 너무 느렸습니다.

병렬로 처리하고 싶었지만, 일반 엣지로는 불가능했습니다. "선배님, 10개 문서를 동시에 처리하고 싶은데, 방법이 없을까요?" 박시니어 씨가 화면을 가리키며 말했습니다.

"Send를 사용하면 됩니다. 동적으로 여러 실행 경로를 만들 수 있어요." Send란 무엇일까요? Send는 특별한 종류의 엣지입니다.

쉽게 비유하자면, Send는 마치 택배 배송 시스템과 같습니다. 물류 센터에서 여러 배송 기사에게 동시에 물건을 나눠주는 것처럼, 하나의 노드에서 여러 노드로 동시에 작업을 분배합니다.

일반 엣지는 하나의 경로만 만듭니다. A 노드에서 B 노드로 가는 단일 경로입니다.

하지만 Send를 사용하면 A 노드에서 B 노드로 가는 경로를 여러 개 만들 수 있습니다. 각 경로는 서로 다른 데이터를 가지고 독립적으로 실행됩니다.

왜 Send가 필요할까요? 실제 AI 시스템에서는 병렬 처리가 매우 중요합니다. 예를 들어 100개의 이미지를 분석해야 한다고 가정해봅시다.

순차적으로 처리하면 100분이 걸립니다. 하지만 10개씩 병렬로 처리하면 10분만에 끝낼 수 있습니다.

Send가 바로 이런 병렬 처리를 가능하게 합니다. 또 다른 예로, 여러 API를 동시에 호출하는 경우를 생각해볼 수 있습니다.

날씨 API, 뉴스 API, 주식 API를 각각 호출해야 한다면, 순차적으로 하면 느립니다. Send로 동시에 호출하면 훨씬 빠릅니다.

Send의 작동 원리 Send는 두 가지 정보를 담고 있습니다. 첫째, 실행할 노드의 이름입니다.

어느 노드로 보낼지 지정합니다. 둘째, 해당 노드에 전달할 상태입니다.

각 Send마다 다른 데이터를 전달할 수 있습니다. 라우팅 함수에서 Send 객체의 리스트를 반환하면, LangGraph는 자동으로 각 Send를 병렬로 실행합니다.

모든 Send가 완료되면, 다음 노드로 이동합니다. 코드 상세 분석 위의 코드를 자세히 살펴보겠습니다.

split_work 함수는 특별한 라우팅 함수입니다. 문자열을 반환하는 대신, Send 객체의 리스트를 반환합니다.

각 Send는 "worker" 노드를 실행하되, 서로 다른 task 데이터를 전달합니다. 예를 들어 tasks가 ["doc1", "doc2", "doc3"]이라면, 세 개의 Send가 생성됩니다.

각각 {"task": "doc1", "index": 0}, {"task": "doc2", "index": 1}, {"task": "doc3", "index": 2}를 가지고 worker 노드를 실행합니다. 핵심은 이 세 개의 worker가 동시에 실행된다는 것입니다.

서로 기다리지 않고 병렬로 돌아갑니다. 세 개가 모두 끝나면, aggregator 노드로 이동해서 결과를 합칩니다.

팬아웃-팬인 패턴 Send를 사용하면 팬아웃-팬인 패턴을 구현할 수 있습니다. 팬아웃은 하나에서 여러 개로 펼쳐지는 것입니다.

splitter 노드가 여러 worker 노드로 작업을 분산하는 것이 팬아웃입니다. 팬인은 여러 개가 하나로 모이는 것입니다.

여러 worker의 결과가 aggregator 노드로 모이는 것이 팬인입니다. 이 패턴은 MapReduce와 비슷합니다.

큰 작업을 작은 조각으로 나누고(Map), 각각 처리한 후(Process), 결과를 합칩니다(Reduce). 상태 관리 Send를 사용할 때 주의할 점은 상태 관리입니다.

각 Send는 독립적인 상태를 받습니다. 원본 상태가 복사되는 것이 아니라, Send 생성 시 전달한 딕셔너리가 새로운 상태가 됩니다.

따라서 필요한 정보를 모두 포함시켜야 합니다. worker 노드들이 병렬로 실행되므로, 서로의 상태를 볼 수 없습니다.

각자 독립적으로 작동합니다. 결과를 공유하려면 aggregator 노드에서 합쳐야 합니다.

실무 활용 사례 실제 프로젝트에서 Send는 어떻게 활용될까요? 대규모 데이터 처리 파이프라인에서 매우 유용합니다.

예를 들어 고객 리뷰 1만 건을 감정 분석한다고 가정해봅시다. Send로 100건씩 100개의 워커에 분배하면, 병렬 처리로 빠르게 완료할 수 있습니다.

또 다른 예로, 멀티모달 AI 시스템을 생각해볼 수 있습니다. 이미지는 비전 모델로, 텍스트는 언어 모델로, 오디오는 음성 모델로 동시에 보내서 처리한 후, 결과를 종합하는 것입니다.

웹 스크래핑에서도 유용합니다. 여러 URL을 동시에 크롤링해야 할 때, Send로 각 URL을 다른 워커에 할당하면 효율적입니다.

주의사항 Send를 사용할 때 주의해야 할 점들이 있습니다. 첫째, 메모리 사용량입니다.

너무 많은 Send를 만들면 메모리가 부족할 수 있습니다. 적절한 배치 크기를 설정해야 합니다.

둘째, 에러 처리입니다. 하나의 워커에서 에러가 나면 어떻게 할지 결정해야 합니다.

전체를 중단할지, 나머지는 계속 진행할지 정책이 필요합니다. 셋째, 결과 순서입니다.

병렬 실행이므로 결과 순서가 보장되지 않습니다. 순서가 중요하다면 index를 함께 전달해서 나중에 정렬해야 합니다.

정리하며 김개발 씨가 Send를 적용하자 처리 속도가 10배 빨라졌습니다. "와, 정말 동시에 실행되네요!" 박시니어 씨가 웃으며 말했습니다.

"병렬 처리의 힘이죠. 대규모 작업에서는 Send가 필수예요." Send를 마스터하면 고성능 워크플로우를 만들 수 있습니다.

여러분의 프로젝트에도 병렬 처리가 필요하다면 Send를 활용해보세요.

실전 팁

💡 - Send로 팬아웃-팬인 패턴을 구현하세요

  • 각 Send는 독립적인 상태를 가집니다
  • 너무 많은 Send는 메모리 문제를 일으킬 수 있습니다

6. Command 제어흐름과 상태 업데이트

김개발 씨가 고급 기능을 탐색하던 중 궁금증이 생겼습니다. "선배님, 노드에서 다음 노드를 직접 지정하면서 동시에 상태도 업데이트할 수 있나요?" 박시니어 씨가 눈을 반짝이며 말했습니다.

"오, 좋은 질문이네요. Command를 사용하면 제어 흐름과 상태 업데이트를 동시에 할 수 있습니다."

Command는 노드 내부에서 반환하는 특별한 객체로, 다음 실행 경로와 상태 업데이트를 동시에 지정할 수 있게 합니다. 마치 자동차를 운전하면서 방향을 틀고 동시에 속도도 조절하는 것처럼, 워크플로우의 제어와 데이터 변경을 한 번에 처리합니다.

이를 통해 더 유연하고 동적인 워크플로우를 구현할 수 있습니다.

다음 코드를 살펴봅시다.

from langgraph.types import Command

def smart_processor(state):
    # 복잡한 처리 로직
    result = analyze_data(state["input"])

    if result["confidence"] > 0.9:
        # 높은 신뢰도: 바로 출력으로, 상태에 결과 저장
        return Command(
            goto="output_node",
            update={"result": result, "status": "success"}
        )
    elif result["confidence"] > 0.5:
        # 중간 신뢰도: 검토 노드로, 경고 플래그 추가
        return Command(
            goto="review_node",
            update={"result": result, "needs_review": True}
        )
    else:
        # 낮은 신뢰도: 재처리, 재시도 카운트 증가
        retry_count = state.get("retry_count", 0) + 1
        return Command(
            goto="reprocess_node",
            update={"retry_count": retry_count, "last_error": "low_confidence"}
        )

김개발 씨는 점점 복잡한 워크플로우를 만들게 되었습니다. 노드에서 처리 결과에 따라 다음 경로를 결정하고, 동시에 상태도 업데이트해야 했습니다.

기존 방식으로는 노드가 상태만 반환하고, 라우팅은 별도의 함수가 담당했습니다. 이것이 불편하게 느껴졌습니다.

"선배님, 노드 안에서 직접 다음 경로를 정할 수 있으면 좋겠는데요." 박시니어 씨가 고개를 끄덕였습니다. "그럴 때 Command를 사용하면 됩니다.

제어와 데이터를 한 곳에서 처리할 수 있어요." Command란 무엇일까요? Command는 특별한 반환 객체입니다. 쉽게 비유하자면, Command는 마치 운전대와 액셀을 동시에 조작하는 것과 같습니다.

운전대로 방향을 틀고(제어 흐름), 액셀로 속도를 조절하는(상태 업데이트) 것을 한 번에 합니다. 일반적인 노드는 상태만 반환합니다.

다음 경로는 엣지나 조건부 엣지가 결정합니다. 하지만 Command를 반환하면, 노드가 직접 "다음은 이 노드로 가고, 상태는 이렇게 바꿔라"라고 명령할 수 있습니다.

Command의 구성 요소 Command는 두 가지 주요 속성을 가집니다. 첫째, goto입니다.

다음에 실행할 노드의 이름을 지정합니다. 이것은 조건부 엣지의 라우팅과 비슷하지만, 노드 내부에서 결정된다는 점이 다릅니다.

둘째, update입니다. 상태에 적용할 업데이트를 딕셔너리로 지정합니다.

이 딕셔너리의 키-값 쌍이 상태에 병합됩니다. 기존 키는 덮어쓰고, 새로운 키는 추가됩니다.

왜 Command가 유용할까요? Command가 없다면 어떻게 될까요? 노드에서 복잡한 처리를 한 후, 결과에 따라 다른 경로로 가야 한다고 가정해봅시다.

기존 방식으로는 노드가 상태를 반환하고, 별도의 라우팅 함수가 그 상태를 보고 경로를 결정합니다. 문제는 로직이 두 곳에 나뉘어 있다는 것입니다.

Command를 사용하면 모든 로직을 한 곳에 모을 수 있습니다. 노드 안에서 "이 경우는 저기로, 상태는 이렇게"라고 한 번에 결정합니다.

코드가 더 명확해지고 유지보수가 쉬워집니다. 코드 자세히 보기 위의 코드를 단계별로 분석해보겠습니다.

smart_processor 노드는 데이터를 분석하고 신뢰도를 계산합니다. 신뢰도가 0.9 이상이면 결과가 확실하다는 뜻입니다.

이 경우 바로 output_node로 가면서, 상태에 결과와 성공 상태를 저장합니다. 신뢰도가 0.5에서 0.9 사이면 애매한 경우입니다.

이때는 사람의 검토가 필요하므로 review_node로 가고, needs_review 플래그를 True로 설정합니다. 신뢰도가 0.5 미만이면 결과를 신뢰할 수 없습니다.

재처리가 필요하므로 reprocess_node로 가면서, 재시도 카운트를 증가시키고 에러 메시지를 기록합니다. 핵심은 세 가지 분기 모두에서 gotoupdate를 동시에 지정한다는 것입니다.

제어와 데이터가 한 곳에서 결정됩니다. Command vs 조건부 엣지 언제 Command를 쓰고 언제 조건부 엣지를 쓸까요?

조건부 엣지는 라우팅 로직이 단순하고 여러 노드에서 재사용될 때 좋습니다. 라우팅 함수를 한 번 만들어두면, 여러 곳에서 쓸 수 있습니다.

로직과 그래프 구조가 분리되어 있어 그래프를 파악하기 쉽습니다. Command는 노드마다 라우팅 로직이 다르고, 처리 결과와 밀접하게 연관될 때 좋습니다.

노드 내부에서 복잡한 계산을 하고, 그 결과에 따라 즉시 경로를 결정해야 할 때 유용합니다. 로직이 한 곳에 모여 있어 이해하기 쉽습니다.

일반적으로 단순한 분기는 조건부 엣지를, 복잡하고 동적인 분기는 Command를 사용하는 것이 좋습니다. 상태 업데이트 패턴 Command의 update는 상태 병합 방식으로 작동합니다.

기존 상태가 {"a": 1, "b": 2}이고, Command의 update가 {"b": 3, "c": 4}라면, 결과 상태는 {"a": 1, "b": 3, "c": 4}가 됩니다. "b"는 덮어써지고, "c"는 새로 추가됩니다.

중첩된 딕셔너리는 병합되지 않고 통째로 교체됩니다. 부분 업데이트를 원하면 명시적으로 처리해야 합니다.

이런 동작 방식을 이해하고 사용해야 예상치 못한 데이터 손실을 막을 수 있습니다. 실무 활용 사례 Command는 실무에서 어떻게 활용될까요?

AI 에이전트 시스템에서 자주 사용됩니다. 예를 들어 사용자 질문의 복잡도를 평가한 후, 간단한 질문은 빠른 모델로, 복잡한 질문은 강력한 모델로 보내면서 동시에 처리 시작 시간을 기록하는 것입니다.

에러 핸들링에도 유용합니다. API 호출이 실패하면 재시도 노드로 보내면서 에러 카운트를 증가시키고, 최대 재시도 횟수를 초과하면 에러 처리 노드로 보내는 로직을 Command로 깔끔하게 작성할 수 있습니다.

A/B 테스트 시스템에서도 활용됩니다. 사용자 그룹에 따라 다른 처리 노드로 보내면서, 실험 버전 정보를 상태에 기록하는 것입니다.

주의사항 Command를 사용할 때 주의할 점들이 있습니다. 첫째, goto에 지정한 노드가 실제로 존재해야 합니다.

오타로 잘못된 노드 이름을 지정하면 런타임 에러가 발생합니다. 타입 힌트나 상수를 사용해서 오타를 방지하세요.

둘째, 순환 참조에 주의해야 합니다. A 노드가 B로 Command를 보내고, B가 다시 A로 보내면 무한 루프가 생깁니다.

종료 조건을 명확히 설정해야 합니다. 셋째, 상태 크기를 관리해야 합니다.

Command로 계속 새로운 키를 추가하면 상태가 비대해집니다. 필요 없는 데이터는 정기적으로 정리하세요.

Command와 END Command의 goto에 END를 지정할 수도 있습니다. 이렇게 하면 워크플로우를 즉시 종료하면서 최종 상태를 업데이트할 수 있습니다.

예를 들어 에러가 발생했을 때 **Command(goto=END, update={"error": error_msg})**로 종료하는 것입니다. 이것은 조기 종료 패턴에 유용합니다.

더 이상 처리할 필요가 없을 때 바로 끝낼 수 있습니다. 정리하며 김개발 씨가 Command를 적용하자 코드가 훨씬 깔끔해졌습니다.

"이제 로직이 한 곳에 모여 있어서 이해하기 쉬워요!" 박시니어 씨가 말했습니다. "Command는 고급 기능이지만, 마스터하면 강력한 무기가 됩니다." Command를 활용하면 더 유연하고 표현력 높은 워크플로우를 만들 수 있습니다.

여러분의 복잡한 분기 로직에 Command를 적용해보세요.

실전 팁

💡 - Command로 제어 흐름과 상태 업데이트를 한 번에 처리하세요

  • goto에는 반드시 존재하는 노드 이름을 지정하세요
  • 순환 참조를 만들지 않도록 주의하세요

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

#LangGraph#Edges#Routing#ConditionalEdges#StateGraph#AI,LLM,Python,LangGraph

댓글 (0)

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