LangGraph 이론 정리

wldbs._.·2025년 8월 19일
AILLMlanggraphpython스터디실습정리
0

AI-LLM

목록 보기
21/21
post-thumbnail

AI-LLM을 활용하는 프로젝트 진행 시, LangGraph를 많이 사용한다고 들었다.
아는 바가 없어서, 아래 문서를 읽으며 정리하였다.

LangGraph 가이드북 - 에이전트 RAG with 랭그래프

🟦 LangGraph의 세 가지 핵심 요소

  • state(상태) - 데이터 저장소

    • 모든 노드가 공유하는 데이터 저장소

    • Python의 TypedDict를 사용해서 정의한다

    • 어떤 종류의 데이터든 저장할 수 있다

      # 모든 노드가 같은 State를 보고 수정할 수 있다 
class SimpleState(TypedDict):
	count: int
	name: str
  • node(노드) - 작업하는 함수

    • 실제 작업을 수행하는 Python 함수

    • 현재 상태를 받아서 새로운 상태를 돌려준다

    • 항상 state를 첫 번째 매개변수로 받는다

    • 딕셔너리 형태로 새로운 상태를 반환한다

    • 반환하지 않은 필드는 기존 값을 유지한다

      # 카운터를 1 증가시키는 노드
def add_one(state):
	return {"count": state["count"] + 1}
	
# 이름을 설정하는 노드	
def set_name(state):
	return ("name": "Alice"}
  • Edge(엣지) - 연결하는 화살표

    • 노드들 사이의 연결을 정의

    • “이 작업 다음에 저 작업을 하자”라고 알려주는 역할

      # 기본적인 엣지 연결
graph.add_edge("노드1", "노드2") # 노드1 → 노드2

# 시작과 끝 연결
graph.add_edge(START, "첫번째노드") # 시작 → 첫번째노드
graph.add_edge("마지막노드", END) # 마지막노트 → 끝

🟩 LangGraph의 상태(State) 관리

  • LangGraph에서 상태는 그래프의 전체 데이터 흐름을 관리하는 핵심 요소
  • 주로 TypedDict나 Pydantic BaseModel을 사용하여 정의
  • 그래프의 모든 노드와 엣지에 대한 입력 스키마 역할 수행
from typing import TypedDict

clas MyState(TypedDict):
	counter: int
  1. TypedDict는 딕셔너리의 키와 값의 타입을 미리 정의할 수 있게 해주는 도구
  2. 클래스 MyState는 TypedDict를 기반으로 함
  3. MyState 클래스 안에 counter라는 항목을 만들고, 이는 정수(int) 타입

🟩 상태 관리란 무엇인가

LangGraph에서 상태 관리는 AI 시스템의 핵심이다.

🥨 상태(State)는 시스템이 처리하는 모든 정보를 담는 그릇이며, 노드 간에 데이터를 전달하고 축적하는 역할을 한다.

  1. 상태가 왜 필요한가?
    1. LangGraph는 메시지 전달 아키텍처를 사용한다.
      1. 노드는 독립적인 처리 단위: 각 노드는 독립적으로 실행되며, 서로 직접 통신하지 않는다
      2. 상태를 통한 간접 통신: 노드들은 공유 상태를 통해 데이터를 주고받는다
      3. Super-step 실행 모델: 병렬 실행 가능한 노드들은 같은 super-step에서 동시에 실행된다
# 간단한 상태 예시
class ChatState(TypedDict):
    user_message: str      # 현재 사용자 메시지 -- 각 노드 실행 시 덮어쓰기된다
    chat_history: list     # 이전 대화 기록 -- 대화의 연속성을 위해 누적되는 리스트
    user_context: dict     # 사용자 정보
    system_status: str     # 시스템 상태
  • TypedDict 사용 이유
    • Python의 일반 dict와 달리, 각 키의 타입이 명시되어 있다
    • IDE에서 자동완성과 타입 체크를 지원한다
    • 런타임에는 일반 dict처럼 동작하지만, 개발 시 타입 안정성을 제공한다

🟩 상태 관리의 중요성

  1. 일관성 유지: 복잡한 대화 흐름에서 맥락을 일관되게 유지하여 사용자 경험 향상

    1. 일관성: 대화나 작업의 전체 흐름에서 정보가 손실되지 않고 유지되는 것
    # 대화 맥락 유지 예시
def conversation_node(state: ChatState):
    # 이전 대화 내용을 참조하여 일관된 응답 생성
    context = "\n".join(state["chat_history"][-3:])  # 최근 3개 대화
    response = generate_contextual_response(state["user_message"], context)
    return {"ai_response": response} # 변경할 필드만 반환 -- 자동으로 기존 상태와 병합

  2. 맥락 이해: 단순한 키워드 매칭을 넘어선다

  3. 복잡한 작업 처리

    1. 다단계 작업이나 다주제 대화를 원활하게 관리할 수 있다

  4. 오류 복구 및 중단점 관리

    1. 체크포인팅: 프로그램의 현재 상태를 저장하는 기술
    2. 문제가 발생했을 때 마지막 체크포인트부터 다시 시작할 수 있다

  5. 성능 최적화: 불필요한 계산을 줄여 전반적인 시스템 성능을 향상시킨다

    1. 상태에 계산 결과를 캐싱하면 동일한 계산을 반복하지 않아도 된다

🟩 기본 상태 구조

from typing_extensions import TypedDict
from typing import Annotated, List
from operator import add

class BasicState(TypedDict):
    # 단순 값들 - 덮어쓰기 방식
    current_step: str
    user_id: str

    # 누적되는 값들 - 추가 방식
    # -- Annotated는 Python 3.9에서 도입된 타입 힌팅 기능
    # -- 첫 번째 인자: 실제 타입, 두 번째 인자: 메타데이터(여기서는 리듀서 함수 add)
    # -- LangGraph는 이 메타데이터를 읽어서 상태 업데이트 방식 결정
    messages: Annotated[List[str], add] # “List[str] 타입 힌팅: 문자열만 들어있는 리스트다”
    processing_log: Annotated[List[str], add]
  • 기본 리듀서: current_step, user_id → 새 값이 들어오면 기존 값은 완전히 사라진다
  • 커스텀 리듀서: messages, processing_log
from operator import add

# 리스트의 경우
result = add([1, 2], [3, 4])  # [1, 2, 3, 4]

# 문자열의 경우
result = add("Hello", " World")  # "Hello World"

# 숫자의 경우
result = add(5, 3)  # 8

🥨 리듀서는 LangGraph에서 상태 업데이트 로직을 정의하는 핵심 메커니즘

  • 각 노드가 반환하는 업데이트를 기존 상태에 어떻게 적용할지 결정

왜 필요한가?

  • 여러 노드가 동시에 또는 순차적으로 실행될 때, 각 노드의 출력을 상태에 통합하는 방법 필요
  • 단순히 덮어쓰기만 하면 이전 정보 손실될 수 있다
  • 리듀서 → 누적, 병합, 최대값 유지 등 다양한 업데이트 전략 구현 가능
# 리듀서 없음 = 덮어쓰기 (Default Reducer)
old_state = {"counter": 5}
new_update = {"counter": 10}
result = {"counter": 10}  # 기존 값이 완전히 대체됨

# 리듀서 있음 = 사용자 정의 동작
from operator import add
old_state = {"items": [1, 2, 3]}
new_update = {"items": [4, 5]}
result = {"items": [1, 2, 3, 4, 5]}  # 리스트가 연결됨
  • 노드는 전체 상태가 아닌 업데이트할 부분만 반환
  • 리듀서는 이 부분 업데이트를 기존 상태와 결합

🟩 상태 스키마 설계

  • 각 필드의 목적과 사용법이 명확해야 한다
  • 각 상태 스키마는 하나믜 명확한 목적을 가져야 한다
  • 내부 처리용 데이터와 외부 인터페이스를 구분한다
  • 입/출력 스키마를 분리하여 내부 구현을 숨긴다
  1. 기본 스키마
    1. 입력과 출력이 동일한 단일 스키마를 사용
    2. 모든 노드가 같은 상태 구조를 공유
  2. 명시적 입출력 스키마
    1. 입력과 출력을 별도로 정의하여 인터페이스 제어
    2. 외부에서 보는 인터페이스(입력/출력)와 내부 처리용 데이터를 분리
    3. 내부 구현의 변경이 외부 인터페이스에 영향 주지 않도록 한다
  3. 다중 스키마
    1. 내부 노드 간 통신을 위한 private 스키마를 포함
    2. 각 단계에 최적화된 스키마를 사용하면서도 데이터 흐름을 체계적으로 관리

🟨 노드(Node)

  • 노드는 LangGraph에서 실제 작업을 수행하는 단위
  • 각 노드는 특정 기능을 수행하는 Python 함수로 구현됨
  • 노드는 현재 상태를 입력으로 받아 처리하고, 업데이트된 상태를 출력
def increment(state): # state라는 매개변수를 받는다 - 이는 현재 그래프 상태
    return {"counter": state["counter"] + 1} # state의 counter 값을 1 증가시킨 새로운 상태를 반환

graph.add_node("increment", increment) # 정의한 함수를 그래프의 노드로 추가
# 첫 번째 인자: 노드의 이름
# 두 번째 인자: 앞서 정의한 함수

🥨 노드의 역할

  1. 상태 처리: 노드는 현재 상태를 받아 필요한 작업을 수행
  2. 로직 실행: 이 예에서는 카운터를 증가시키는 간단한 로직을 수행
  3. 상태 업데이트: 작업 결과를 바탕으로 새로운 상태를 생성하여 반환

→ 이렇게 정의된 노드는 그래프 내에서 하나의 작업 단위로 동작하며, 그래프의 전체 흐름에 따라 순차적으로 또는 조건에 따라 실행된다.

🟨 노드란 무엇인가?

: 각 노드는 그래프의 한 단계를 나타내며, 데이터를 처리하고 변환하는 역할을 한다

  • 함수 기반: Python 함수로 구현
  • 상태 중심: 현재 상태를 입력으로 받아 처리
  • 독립적 실행: 각 노드는 독립적으로 실행 가능
  • 조합 가능: 여러 노드를 연결하여 복잡한 워크플로우 구성

State 클래스는 그래프가 관리할 상태의 구조를 정의한다

StateGraph(State)로 그래프를 생성하고, add_node로 노드를 추가한다

  • 첫 번째 매개변수는 노드 이름, 두 번째 매개변수는 실행할 함수

노드는 변경할 필드만 딕셔너리로 반환한다

  • 반환하지 않은 필드는 기존 값이 그대로 유지된다

🟨 노드의 역할과 책임

LangGraph에서 노드는 애플리케이션의 핵심 구성 요소로, 그래프 내에서 특정 작업을 수행하는 독립적인 처리 단위이다.

  • 각 노드는 입력 상태를 받아 특정 로직을 실행하고, 그 결과를 다시 상태로 반환하는 함수형 패러다임을 따른다.

노드의 가장 중요한 특징은 상태 중심 설계이다.

  • 모든 노드는 공통된 상태 스키마를 통해 데이터를 주고받는다
    • → 이를 통해 복잡한 워크플로우에서도 데이터 흐름을 보장할 수 있다
  • 각 노드는 상태의 일부/전체를 읽어들이고, 필요한 처리를 수행한 후, 업데이트된 상태를 반환한다

노드의 주요 책임은 크게 세 가지로 나눌 수 있다.

  1. 데이터 변환 및 처리: 원시 데이터를 구조화하거나, 복잡한 계산을 수행하거나, 외부 API와의 통신 담당
  2. 상태 관리: 애플리케이션의 현재 상태를 읽고, 필요한 변경사항을 적용, 다음 단계로 전달할 정보 준비
  3. 흐름 제어: 조건부 로직을 통해 다음에 실행될 노드를 결정하거나, 특정 조건에서 그래프의 실행 종료

LangGraph의 노드는 모듈성과 재사용성을 강조한다.

  • 각 노드는 명확한 입력과 출력을 가진 독립적인 단위로 설계된다
    • → 다른 그래프나 워크플로우에서도 쉽게 재사용할 수 있다
  • 타입 안전성을 제공하여 TypedDict를 활용한 상태 스키마를 통해 컴파일 시점에서 많은 오류를 방지
  • 상태 업데이트 방식에서도 유연성을 제공한다
    • 기본적인 덮어쓰기 방식 외에도
    • 리듀서(reducer)를 사용하여 복잡한 상태 병합 로직을 구현할 수 있다
      • Annotated[list, operator.add]와 같은 어노테이션 → 리스트에 새 항목을 추가
      • 사용자 정의 리듀서 함수를 통해 특별한 병합 규칙을 적용

서브그래프와 Private State 개념을 통해 더욱 복잡한 아키텍처도 지원한다

  • 서브그래프는 독립적인 상태를 가지면서도 부모 그래프와 필요한 정보만을 공유할 수 있다
  • → 대규모 멀티 에이전트 시스템이나 복잡한 워크플로우를 효과적으로 관리할 수 있다

노드는 또한 에러 처리와 복구에도 중요한 역할을 한다.

  • 각 노드에서 발생할 수 있는 예외 상황을 적절히 처리하고, 필요에 따라 재시도 로직이나 대안 경로 제공
  • → 전체 시스템의 안정성과 신뢰성 ↑

노드는 성능 최적화의 핵심 단위이다.

  • 각 노드의 실행 시간을 모니터링하고, 병목 지점을 식별하여 최적화, 필요에 따라 병렬 처리나 캐싱 적용

⇒ 이러한 특성들이 결합되어 LangGraph의 노드는 확장 가능하고 유지보수가 용이한 AI 애플리케이션 개발의 기반을 제공

🟨 노드 타입과 패턴

노드의 타입은 주로 실행 방식과 처리 패턴에 따라 구분되며, 이는 애플리케이션의 성능과 구조에 큰 영향을 미친다.

  1. 동기 노드 (Sync Node)
  • 작업이 순차적으로 실행되며, 완료될 때까지 다음 단계로 진행 X
  • CPU 집약적인 작업이나 로컬 데이터 처리에 주로 사용
def sync_node(state: State) -> Dict[str, Any]:
    """
    동기 노드 - 일반적인 노드 타입
    순차적으로 실행되며 결과를 즉시 반환
    """
    # 동기 작업 수행
    result = perform_sync_operation(state["input"])

    return {"output": result}

def perform_sync_operation(data):
    """동기 작업"""
    import time
    time.sleep(0.1)  # 시뮬레이션
    return f"Sync result: {data}"
  1. 비동기 노드 (Async Node)
  • I/O 작업이나 외부 API 호출과 같이 대기 시간이 발생하는 작업에 효율적
  • async/await 패턴을 활용하여 여러 작업을 동시에 실행
  • asyncio.gather()를 통해 병렬 처리 구현
  • 네트워크 요청, 데이터베이스 쿼리, 파일 I/O 등에서 성능 향상
async def async_node(state: State) -> Dict[str, Any]: # async def로 정의된 비동기 노드
    """
    비동기 노드 - I/O 작업에 효율적
    동시에 여러 작업을 처리할 수 있음
    """
    # 비동기 작업 수행
    result = await perform_async_operation(state["input"])

    # 여러 비동기 작업 동시 실행
    results = await asyncio.gather( # 여러 비동기 작업을 동시에 실행하고 모든 결과를 기다림
		    # 독립적인 비동기 작업을 시뮬레이션 & 동시 실행
        fetch_data_1(),
        fetch_data_2(),
        fetch_data_3()
    )

    return {
        "output": result,
        "additional_data": results
    }
  1. 조건부 노드 (Conditional Node)
  • 상태나 입력에 따라 다른 처리 로직을 선택하는 유연한 패턴
  • 비즈니스 로직의 복잡성을 효과적으로 관리
  • 런타임 조건에 따라 다양한 처리 경로 제공
def conditional_node(state: State) -> Dict[str, Any]:
    """
    조건부 노드 - 상태에 따라 다른 처리 수행
    """
    mode = state.get("mode", "default")

    if mode == "fast":
        # 빠른 처리
        result = quick_process(state)
        execution_time = "fast"
    elif mode == "thorough":
        # 정밀 처리
        result = thorough_process(state)
        execution_time = "slow"
    else:
        # 기본 처리
        result = default_process(state)
        execution_time = "normal"

    return {
        "result": result,
        "execution_time": execution_time,
        "mode_used": mode
    }
  1. 검증 노드 (Validation Node)
  • 데이터 품질과 시스템 안정성 보장
  • 입력 데이터의 타입, 크기, 형식, 범위 등을 체계적으로 검증
  • 모든 오류를 수집하여 한 번에 반환 → 사용자 경험 향상
  • 보안 취약점을 방지하고 다운스트림 노드의 안정성 보장
from typing import Optional

class ValidationState(TypedDict):
    input_data: Any
    validation_errors: list[str]
    is_valid: bool

def validation_node(state: ValidationState) -> Dict[str, Any]:
    """
    검증 노드 - 데이터 유효성 검사
    """
    input_data = state["input_data"]
    errors = []

    # 다양한 검증 수행
    if not input_data:
        errors.append("입력 데이터가 비어있습니다.")

    if isinstance(input_data, str):
        if len(input_data) < 3:
            errors.append("문자열이 너무 짧습니다 (최소 3자).")
        if len(input_data) > 1000:
            errors.append("문자열이 너무 깁니다 (최대 1000자).")
        if not input_data.isascii():
            errors.append("ASCII 문자만 허용됩니다.")
            
		...
		
	  # 검증 결과 반환
    return {
        "validation_errors": errors,
        "is_valid": len(errors) == 0
    }

각 노드 타입은 특정 사용 사례와 성능 요구사항에 최적화되어 있다

  • 동기 노드는 단순성과 예측 가능성 제공
  • 비동기 노드는 처리량과 응답성 향상
  • 조건부 노드는 비즈니스 로직의 유연성 제공
  • 검증 노드는 시스템 신뢰성 보장

노드 패턴의 조합을 통해 더욱 복잡하고 강력한 워크플로우를 구성할 수 있다

  • 검증 노드(데이터 품질 확인) → 조건부 노드(처리 방식 결정) → 비동기 노드(병렬 처리 수행 구축)

각 노드 타입은 에러 처리와 모니터링에서도 다른 접근 방식이 필요하다

  • 동기 노드는 간단한 try-catch 구조로 충분
  • 비동기 노드는 타임아웃, 재시도, 부분 실패 등 고려
  • 조건부 노드는 모든 분기에서의 에러 처리 보장
  • 검증 노드는 검증 실패 시의 적절한 복구 전략 필요

타입 안전성도 중요한 고려사항이다

  • TypedDict 를 활용한 상태 스키마와 함께 각 노드 타입의 입출력을 명확히 정의
    • → 컴파일 시점에서 많은 오류 방지
  • 특히 조건부 노드에서는 모든 분기가 일관된 출력 스키마를 반환하도록 보장

🟨 노드의 구성

LangGraph에서 노드의 구성은 런타임에 노드의 동작을 동적으로 조정할 수 있는 메커니즘이다.

  • 같은 그래프 구조를 유지하면서도 다양한 설정에 따라 다른 처리 방식을 적용할 수 있게 한다.

RunnableConfig는 LangChain의 핵심 구성 객체로, 노드가 실행될 때 런타임 설정을 전달 받게 한다.

  • 정적으로 정의된 그래프에 동적 유연성을 부여
  • 노드가 RunnableConfig를 매개변수로 받으면, 실행 시점에 구성 값들을 참조하여 처리 로직 조정

핵심 활용 영역

  • 모델 선택과 매개변수 조정: 같은 노드에서 런타임에 다른 LLM 모델을 사용하거나, temperature/max_tokens 등의 매개변수 동적으로 설정
  • 재시도 및 오류 처리 정책: 최대 재시도 횟수, 백오프 전략, 타임아웃 설정 등을 환경이나 요청의 중요도에 따라 조정
  • 환경별 설정: 개발, 테스트, 프로덕션 환경에 따라 다른 DB 연결, API 엔드포인트, 로깅 수준 등 적용

구성 접근 패턴은 계층적 구조를 따른다.

동적 처리 로직의 구현에서는 구성값에 따라 다른 알고리즘이나 처리 방식을 선택할 수 있다.

사용자별 개인화도 중요한 활용 사례이다.

구성의 전파와 상속에서는 부모 그래프의 구성이 서브그래프나 하위 노드로 자동 전파되며, 필요에 따라 특정 노드에서 구성을 오버라이드할 수 있다.

성능 최적화에서도 구성이 중요한 역할을 한다.

모니터링과 관찰 가능성을 위해 구성을 활용한다.

보안 고려사항에는 적절한 암호화와 접근 제어가 필요하다.

🟨 노드 구성 고급 패턴 유형

  1. 클래스 기반 노드
  • 객체지향 설계의 장점을 LangGraph에 적용
  • 노드가 내부 상태를 유지해야 하거나, 복잡한 로직을 캡슐화해야 하거나, 여러 메서드를 통해 기능을 분리해야 할 때 유용
  • 생성자를 통해 초기 설정을 받고, __call__ 메서드를 구현하여 호출 가능한 객체로 만들어짐
  • 장점: 상태 지속성과 메서드 분리
    • 처리 횟수 추적, 캐시 관리, 설정 저장 등을 클래스의 인스턴스 변수로 관리
    • 각기 다른 처리 방식을 별도의 메서드로 분리 → 코드의 가독성과 유지보수성
    • 상속을 통해 기본 기능을 확장하거나 특화된 버전 → 코드 재사용성
  1. 데코레이터 패턴 노드
  • 기존 노드 함수에 추가 기능을 동적으로 부여
  • 함수형 프로그래밍의 고차 함수 개념 활용 → 원본 노드 로직을 수정하지 않고도 새로운 기능 추가 O
  • 핵심 가치는 관심사의 분리와 조합 가능성
  1. 팩토리 패턴 노드
  • 런타임에 다양한 타입의 노드를 동적으로 생성
  • 설정이나 조건에 따라 다른 동작을 하는 노드들을 일관된 인터페이스로 생성할 수 있게 함
  • 노드 타입과 매개변수를 받아 해당하는 노드 함수를 생성하고 반환
  • 주요 장점은 유연성과 확장성

패턴들의 조합과 응용 관점

  • 이러한 패턴들을 단독으로 사용하는 것보다 조합하여 사용할 때 더 큰 시너지 효과를 얻을 수 있다.
  • 예를 들어, 팩토리 패턴으로 생성된 노드에 데코레이터를 적용하거나, 클래스 기반 노드의 메서드에 데코레이터를 사용하는 등의 조합이 가능

성능과 메모리 고려사항

  • 클래스 기반 노드는 인스턴스 상태를 유지하므로 메모리 사용량이 늘어날 수 있으며
  • 데코레이터는 함수 호출 오버헤드를 추가할 수 있다
    • 하지만 이러한 오버헤드는 일반적으로 제공하는 이점에 비해 미미하며, 적절한 설계를 통해 최소화할 수 있습니다.

테스트와 디버깅

  • 이러한 고급 패턴들은 각각 고유한 테스트 전략이 필요
  • 클래스 기반 노드는 상태 변화를 테스트
  • 데코레이터는 각 계층을 독립적으로 테스트
  • 팩토리 패턴은 생성된 다양한 노드 타입들을 모두 검증

🟪 엣지 (Edge)

엣지는 LangGraph에서 노드 간의 연결을 나타낸다.

  • 그래프의 실행 흐름을 결정
  • 엣지를 통해 한 노드에서 다음 노드로 상태 전달
# 그래프 실행이 시작되면 바로 increment 노드로 이동
graph.add_edge(START, "increment") # START는 그래프의 시작점을 나타내는 특별한 상수

# increment 노드의 실행이 끝나면 그래프 실행을 종료
graph.add_edge("increment", END) # END는 그래프의 종료점을 나타내는 특별한 상수
  1. 실행 순서 정의: 엣지부 노드들이 어떤 순서로 실행될지 결정
  2. 흐름 제어: 조건부 엣지를 사용하면 특정 조건에 따라 다른 노드로 이동
  3. 상태 전달: 한 노드에서 다음 노드로 상태 전달

엣지의 핵심 특징

  • 방향성: 한 노드에서 다른 노드로의 단방향 연결
  • 상태 전달: 노드 간 상태 정보 전달
  • 흐름 제어: 실행 순서와 조건 결정
  • 병렬 실행: 여러 엣지가 동시에 활성화 가능

엣지 타입

  • 일반 엣지: 두 노드를 직접 연결, 첫 번째 노드 완료되면 자동으로 두 번째 노드 실행
  • 시작 엣지: 그래프 실행의 진입점 정의, Start에서 지정된 노드로 실행 시작
  • 종료 엣지: 그래프 실행의 종료점 정의, 해당 노드 완료되면 그래프 실행 종료
  • 조건부 엣지: 상태나 조건 따라 다음 노드 동적으로 결정
# 1. 일반 엣지 (Normal Edge)
graph.add_edge("node_a", "node_b")

# 2. 시작 엣지 (Start Edge)
graph.add_edge(START, "first_node")

# 3. 종료 엣지 (End Edge)
graph.add_edge("last_node", END)

# 4. 조건부 엣지 (Conditional Edge)
graph.add_conditional_edges(
    "decision_node",
    routing_function, # routing_function이 반환하는 값에 따라 다른 노드로 분기
    {
        "option_a": "node_a",
        "option_b": "node_b"
    }
)

🟪 엣지의 역할과 기능

  1. 실행 순서 정의
  • 엣지를 통해 노드들 간의 순차적 실행 체인 구성
  • 선형 흐름뿐만 아니라, 분기와 병합이 포함된 복잡한 실행 경로 정의
  • 의존성 관리 중요 → 각 노드가 이전 단계의 완료를 전제로 동작할 수 있도록 보장, 데이터 처리 파이프라인에서 각 단계가 올바른 입력 받을 수 있도록 조율
  1. 흐름 제어
  • 조건부 분기를 통해 런타임 상황에 따라 다른 실행 경로 선택
  • add_conditional_edges → 상태 값, 계산 결과, 외부 조건 등에 따라 동적으로 다음 노드 결정
  • 핵심은 라우팅 함수 → 현재 상태를 분석하여 다음에 실행할 노드를 결정
    • Literal 타입 사용 → 가능한 경로들을 명시적으로 정의 → 타입 안전성 보장
  1. 상태 전달
  • 노드 간 이동 시 상태 정보가 손실 없이 전달되도록 보장
  • 필요에 따라 상태 변환이나 필터링 적용
  • 리듀서(reducer)와 함께 사용될 때, 여러 노드의 출력을 병합하거나 누적
  • 상태 불변성과 부분 업데이트 중요
    • 각 노드는 전체 상태를 덮어쓰는 것이 아니라, 필요한 부분만 업데이트하고, 나머지 상태는 유지

🟪 조건부 엣지

: 런타임 상태에 따라 동적으로 실행 경로를 결정할 수 있게 해주는 기능

기본 조건부 엣지

  • 핵심은 라우팅 함수와 경로 매핑
  • 현재 상태를 분석하여 다음에 실행할 노드를 결정하는 로직을 담는다
  • Literal 타입을 사용하여 가능한 모든 경로를 명시적으로 정의한다 → 타입 안전성 보장
  • 소스 노드 실행 - 상태 업데이트 → 라우팅 함수 호출 - 업데이트된 상태 기반으로 다음 노드 선택 → 선택된 노드는 경로 매핑 딕셔너리 통해 실제 노드 이름으로 변환 - 해당 노드 실행
# 조건부 엣지 설정
sentiment_graph.add_edge(START, "analyze")
sentiment_graph.add_conditional_edges(
    "analyze",
    sentiment_router,
    {
        "positive": "positive",
        "negative": "negative",
        "neutral": "neutral"
    }
)

🟪 Command

LangGraph에서 “다음에 어디로 갈지”와 “상태를 어떻게 바꿀지”를 한 번에 결정

  • 핵심 가치는 원자적 연산
    • 전통적 방식 → 라우팅 결정과 상태 업데이트가 별도의 단계로 분리 ⇒ 중간 상태의 불일치나 복잡한 동기화 문제 발생
    • Command는 이 두 작업을 하나의 원자적 연산으로 통합 → 데이터 일관성, 시스템 안정성
if stock >= quantity:
        # 재고 충분 - 결제 처리로 이동
        # "결제 처리로 가면서 동시에 상태 메시지 업데이트"를 한 번에 수행
        return Command(
            goto="process_payment",
            update={
                "status": "재고 확인 완료",
                "messages": state["messages"] + [f"{item} {quantity}개 재고 확인됨"]
            }
        )

Command를 사용하면 좋은 경우:

  • 다음 단계로 가면서 동시에 여러 상태를 업데이트해야 할 때
  • 재시도나 복구 로직이 복잡한 경우
  • 동적으로 조건을 판단하여 다른 경로로 보내야 할 때
  • 비즈니스 로직이 복잡하고 여러 요소를 동시에 고려해야 할 때

조건부 엣지를 사용하면 좋은 경우:

  • 간단한 분기만 필요한 경우
  • 라우팅 로직과 비즈니스 로직을 명확히 분리하고 싶은 경우
  • 여러 노드로 병렬 실행이 필요한 경우

🟧 Graph 연결(compile) 및 실행

  • graph = StateGraph(MyState): StateGraph 객체 생성 → MyState 상태 기반의 그래프 정의
  • app -= graph.compile(): 정의된 그래프를 실행 가능한 형태로 변환
  • result = app.invoke({"counter": 0}): invoke()는 컴파일된 그래프 실행, 인자로 전달된 {”counter”: 0}는 초기 상태(MyState 클래스의 구조와 일치)

🥨 Annotated란?

  • Python에서 Annotated는 타입 힌트에 “추가 메타데이터(부가 정보)”를 달아주는 문법
from typing import Annotated

Annotated[원래_타입, 부가정보...]

  • 첫 번째 인자: 원래 타입 (int, str, list[str], …)
  • 그 뒤 인자들: 메타데이터 (아무 Python 객체나 넣을 수 있음)
profile
wldbs._.
공부 기록용 24.08.05~ #LLM #RAG
이전 포스트

LLM 문서 요약 방법론

0개의 댓글