2. LangGraph - StateGraph 상태 기반 그래프
- 이 문서는 LangGraph의 핵심 구조인 StateGraph를 학습하기 위한 문서이다.
- StateGraph는 상태(State)를 기반으로 작동하는 방향성 그래프 구조로, LLM 애플리케이션의 복잡한 워크플로우를 선언적으로 정의할 수 있게 한다.
- 이 문서에서는 상태(State) 정의, 노드(Node) 구현, 엣지(Edge) 연결, 조건부 분기(Conditional Edge) 처리까지 StateGraph의 전체 라이프사이클을 단계적으로 학습한다.
- 이전 문서(1. LangGraph - LangChain Tool Calling 학습 매뉴얼)에서 학습한 도구 호출 개념이 StateGraph 내에서 어떻게 통합되는지를 이해하는 것이 이 문서의 목표이다.
1. 사전 작업
1.1. Env 환경변수
from dotenv import load_dotenv
load_dotenv(override=True)
1.2. 기본 라이브러리
import re
import os, json
from textwrap import dedent
from pprint import pprint
import warnings
warnings.filterwarnings("ignore")
2. StateGraph - 상태 기반 그래프 구조
- StateGraph는 LangGraph의 핵심 개념으로, 상태(State)를 기반으로 작동하는 방향성 그래프 구조이다.
- 전통적인 함수 호출 방식과 달리, StateGraph는 데이터(상태)가 노드 사이를 흐르면서 각 노드가 상태를 읽고 업데이트하는 방식으로 동작한다.
- 이 패턴이 필요한 이유는 다음과 같다:
- LLM 기반 애플리케이션은 여러 단계의 처리(입력 분석, 도구 호출, 응답 생성 등)를 거치는데, 각 단계의 결과를 다음 단계로 전달해야 한다.
- 조건에 따라 다른 경로로 분기하거나, 반복(loop)이 필요한 복잡한 워크플로우를 선언적으로 정의할 수 있다.
- 각 노드가 독립적이므로 디버깅, 테스트, 재사용이 용이하다.
- StateGraph의 구성 요소는 크게 세 가지이다: State(상태), Node(노드), Edge(엣지).
- 실습 주제: 레스토랑 메뉴 추천 시스템 -- 사용자의 선호도에 따라 메뉴를 추천하고, 메뉴에 대한 정보를 제공한다.
- StateGraph 구성 요소 전체 구조
- State 구성 요소 상세
- Node 구성 요소 상세
- Edge 구성 요소 상세
2.1. 상태(State) - TypedDict로 스키마 정의
- State(상태)란: State는 그래프의 모든 노드가 공유하는 중앙 데이터 저장소의 구조(스키마)를 정의하는 것이다. Python의 `TypedDict`를 상속한 클래스로 선언하며, 그래프가 실행되는 동안 노드 간에 전달되어야 하는 모든 데이터 필드의 이름과 타입을 명시한다. State는 그래프의 "메모리" 역할을 하여, 한 노드의 출력을 다른 노드의 입력으로 자연스럽게 연결한다. 예를 들어, `class ChatState(TypedDict): user_input: str; response: str`처럼 정의하면, 그래프 내 모든 노드는 `user_input`과 `response` 필드를 읽고 쓸 수 있다.
- StateGraph 라이프사이클에서의 위치: State 정의는 그래프를 구성하기 전 가장 먼저 수행해야 하는 단계이다. State가 정의되어야 노드 함수의 입출력 구조가 결정된다.
- State는 그래프가 처리하는 데이터의 구조(스키마)를 정의한다.
- Python의 `typing.TypedDict`를 사용하여 State의 각 필드와 타입을 선언한다.
- LangGraph에서 State의 핵심 동작 원리는 override(덮어쓰기) 방식이다:
- 그래프가 실행되면 초기 상태가 생성된다.
- 각 노드는 전체 상태를 인자로 받지만, 반환할 때는 업데이트하고 싶은 키만 포함한 딕셔너리를 반환한다.
- 반환된 키-값 쌍이 기존 상태에 덮어쓰기(override) 된다.
- 반환하지 않은 키는 이전 값이 그대로 유지된다.
- 이 override 방식 덕분에 각 노드는 자신이 담당하는 필드만 업데이트하면 되므로, 노드 간 결합도가 낮아지고 독립성이 보장된다.
- 참고 — Annotated 타입과 reducer 함수: LangGraph는 `typing.Annotated`를 사용하여 필드별로 커스텀 업데이트 로직(reducer)을 지정할 수 있다. 예를 들어, `langgraph.graph`에서 제공하는 `add_messages` reducer를 사용하면 메시지 리스트를 override 대신 append 방식으로 누적할 수 있다. 이 문서에서는 기본 override 방식만 다루며, `Annotated` + reducer 패턴은 이후 문서에서 학습한다.
from typing import TypedDict
# [사용자 정의] 상태 Schema 정의 - 사용자의 선호도, 추천된 메뉴, 그리고 메뉴 정보를 저장
# 이 MenuState는 섹션 2 전체(2.1~2.3)에서 사용되는 State 스키마이다.
# TypedDict는 Python 표준 라이브러리의 타입 힌트로, 딕셔너리의 키와 값 타입을 명시적으로 선언한다.
# StateGraph에 전달되면 그래프 내 모든 노드가 이 스키마에 따라 상태를 주고받게 된다.
class MenuState(TypedDict):
user_preference: str # 사용자 선호도 (예: "육류", "해산물", "채식")
recommended_menu: str # 추천된 메뉴 이름 (예: "스테이크")
menu_info: str # 메뉴 상세 정보 (예: 설명 + 가격)
- 위 코드에서 `MenuState`는 3개의 필드를 가진 상태 스키마이다.
- 그래프 실행 시 초기 상태로 `{"user_preference": ""}` 만 전달해도, 이후 노드들이 `recommended_menu`와 `menu_info`를 채워나간다.
- 이것이 StateGraph의 핵심 철학이다: 상태는 그래프를 통해 점진적으로 완성된다.
- State 점진적 업데이트 흐름
- 위 다이어그램은 상태가 각 노드를 거치면서 점진적으로 채워지는 과정을 보여준다.
- 각 노드는 자신이 담당하는 필드만 반환하며, 나머지 필드는 이전 상태 값이 유지된다.
2.2. 노드(Node) - 작업 수행 함수
- Node(노드)란: Node는 그래프에서 실제 비즈니스 로직을 실행하는 독립적인 함수 단위이다. 각 노드는 Python 함수로 구현되며, State 전체를 인자로 받아 필요한 데이터를 읽고, 자신이 담당하는 처리(LLM 호출, 데이터 변환, 외부 API 요청, 조건 판단 등)를 수행한 뒤, 업데이트할 필드만 딕셔너리로 반환한다.
- 예를 들어, `def analyze(state: MyState) -> dict: return {"category": llm.invoke(state["query"])}`처럼 정의하면, 이 노드는 `query`를 읽어 LLM으로 분석한 결과를 `category` 필드에 기록하는 역할을 한다.
- StateGraph 라이프사이클에서의 위치: State 정의 이후, 그래프 빌드 이전에 노드 함수를 정의한다. 노드는 그래프의 실제 "작업 단위"이다.
- 노드는 그래프에서 실제 작업을 수행하는 함수이다.
- 노드 함수의 시그니처 규칙:
- 인자: 전체 State를 받는다 (`state: MenuState`)
- 반환값: 업데이트할 키만 포함한 딕셔너리를 반환한다 (전체 State가 아님)
- 타입 힌트로 `-> MenuState`를 사용하지만, 실제로는 부분 딕셔너리를 반환해도 된다
- 이 패턴은 LangGraph의 핵심 설계 원칙이다: 각 노드는 상태 전체를 읽을 수 있지만, 자신이 책임지는 부분만 업데이트한다.
2.2.1. get_user_preference 노드
- 랜덤으로 사용자 선호도를 생성하는 노드이다.
- `user_preference` 키만 반환하여 상태를 업데이트한다.
- 참고: 이 코드의 `state` 파라미터는 섹션 2.1에서 정의한 `MenuState` TypedDict이다. `MenuState`는 `user_preference`, `recommended_menu`, `menu_info` 3개의 필드를 가진다.
import random
# [사용자 정의] get_user_preference 노드 함수
# 이 함수는 섹션 2.3에서 add_node("get_preference", get_user_preference)로 그래프에 등록된다.
# 역할: 랜덤으로 사용자의 음식 선호도를 생성하여 State의 user_preference 필드를 채운다.
def get_user_preference(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 2.1 참조] MenuState TypedDict (user_preference, recommended_menu, menu_info)
# 이 노드에서는 state를 읽지 않고, 새로운 값을 생성하여 반환한다.
print("---랜덤 사용자 선호도 생성---")
preferences = ["육류", "해산물", "채식", "아무거나"]
preference = random.choice(preferences)
print(f"생성된 선호도: {preference}")
# 부분 딕셔너리 반환: user_preference 키만 업데이트하고, 나머지 키(recommended_menu, menu_info)는 변경하지 않음
return {"user_preference": preference}
- `state` 인자를 통해 현재 상태 전체에 접근할 수 있으나, 이 노드에서는 사용하지 않는다.
- 반환값 `{"user_preference": preference}`는 State 전체가 아니라 부분 딕셔너리이다. LangGraph가 이 값을 기존 상태에 merge(override)한다.
2.2.2. recommend_menu 노드
- 이전 노드에서 설정한 `user_preference`를 읽어 메뉴를 추천하는 노드이다.
- 상태를 읽는 것(`state['user_preference']`)과 상태를 쓰는 것(`return {"recommended_menu": menu}`)이 분리되어 있다.
- 참고: 이 코드의 `state` 파라미터는 섹션 2.1에서 정의한 `MenuState` TypedDict이며, `state['user_preference']`는 섹션 2.2.1의 `get_user_preference` 노드에서 설정한 값이다.
# [사용자 정의] recommend_menu 노드 함수
# 이 함수는 섹션 2.3에서 add_node("recommend", recommend_menu)로 그래프에 등록된다.
# 역할: 이전 노드(get_user_preference)가 설정한 user_preference를 읽어서 그에 맞는 메뉴를 추천한다.
def recommend_menu(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 2.1 참조] MenuState TypedDict (user_preference, recommended_menu, menu_info)
print("---메뉴 추천---")
# state['user_preference']를 읽어옴: 이전 노드(get_user_preference, 섹션 2.2.1)가 설정한 값
# 예를 들어 "육류", "해산물", "채식", "아무거나" 중 하나가 들어있다
preference = state['user_preference']
if preference == "육류":
menu = "스테이크"
elif preference == "해산물":
menu = "랍스터 파스타"
elif preference == "채식":
menu = "그린 샐러드"
else:
menu = "오늘의 쉐프 특선"
print(f"추천 메뉴: {menu}")
# 부분 딕셔너리 반환: recommended_menu 키만 업데이트
return {"recommended_menu": menu}
- 이 노드는 `user_preference`를 읽기만 하고, `recommended_menu`를 쓰기만 한다.
- 이런 패턴 덕분에 노드 간의 데이터 의존성이 명확해진다.
2.2.3. provide_menu_info 노드
- `recommended_menu`를 읽어 해당 메뉴의 상세 정보를 반환하는 노드이다.
- 참고: 이 코드의 `state` 파라미터는 섹션 2.1에서 정의한 `MenuState` TypedDict이며, `state['recommended_menu']`는 섹션 2.2.2의 `recommend_menu` 노드에서 설정한 값이다.
# [사용자 정의] provide_menu_info 노드 함수
# 이 함수는 섹션 2.3에서 add_node("provide_info", provide_menu_info)로 그래프에 등록된다.
# 역할: 이전 노드(recommend_menu)가 추천한 메뉴의 상세 정보(설명, 가격)를 제공한다.
def provide_menu_info(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 2.1 참조] MenuState TypedDict (user_preference, recommended_menu, menu_info)
print("---메뉴 정보 제공---")
# state['recommended_menu']를 읽어옴: 이전 노드(recommend_menu, 섹션 2.2.2)가 설정한 추천 메뉴 이름
menu = state['recommended_menu']
if menu == "스테이크":
info = "최상급 소고기로 만든 juicy한 스테이크입니다. 가격: 30,000원"
elif menu == "랍스터 파스타":
info = "신선한 랍스터와 al dente 파스타의 조화. 가격: 28,000원"
elif menu == "그린 샐러드":
info = "신선한 유기농 채소로 만든 건강한 샐러드. 가격: 15,000원"
else:
info = "쉐프가 그날그날 엄선한 특별 요리입니다. 가격: 35,000원"
print(f"메뉴 정보: {info}")
# 부분 딕셔너리 반환: menu_info 키만 업데이트. 이로써 MenuState의 3개 필드가 모두 채워진다.
return {"menu_info": info}
- 세 노드를 종합하면, 각 노드가 상태의 서로 다른 필드를 담당하는 구조이다:
- `get_user_preference` -> `user_preference` 기록
- `recommend_menu` -> `user_preference` 읽기, `recommended_menu` 기록
- `provide_menu_info` -> `recommended_menu` 읽기, `menu_info` 기록
- 노드별 State 읽기/쓰기 관계
2.3. 그래프(Graph) 구성 - StateGraph 빌더 패턴
- Edge(엣지)란: Edge는 그래프에서 노드 간의 실행 순서와 데이터 흐름 방향을 정의하는 연결선이다. 엣지는 "어떤 노드가 완료되면 다음에 어떤 노드를 실행할 것인가"를 결정하며, 두 가지 형태가 있다.
- 첫째, `add_edge("A", "B")`는 A 노드 완료 후 항상 B 노드를 실행하는 선형(무조건) 엣지이다.
- 둘째, `add_conditional_edges("A", router_fn, {"조건1": "B", "조건2": "C"})`는 라우터 함수의 반환값에 따라 B 또는 C로 분기하는 조건부 엣지이다.
- 예를 들어, `add_edge(START, "입력분석")`은 그래프 시작 시 반드시 `입력분석` 노드를 실행하고, `add_conditional_edges("입력분석", classify, {"질문": "검색", "잡담": "응답"})`은 `classify` 함수가 `"질문"`을 반환하면 `검색` 노드로, `"잡담"`을 반환하면 `응답` 노드로 분기한다.
- StateGraph 라이프사이클에서의 위치: State와 Node가 정의된 후, 이들을 조합하여 실행 가능한 그래프를 만드는 단계이다. 이 단계는 빌더(Builder) -> 컴파일(Compile) -> 실행(Invoke)의 3단계로 구분된다.
- StateGraph는 빌더 패턴(Builder Pattern)을 사용한다:
- `StateGraph(MenuState)`: State 스키마를 인자로 빌더를 생성한다.
- `add_node(name, function)`: 노드를 등록한다. `name`은 문자열 식별자, `function`은 노드 함수이다.
- `add_edge(source, target)`: 두 노드를 연결하는 엣지를 추가한다.
- `compile()`: 빌더를 실행 가능한 그래프 객체로 변환한다.
- `START`와 `END`는 LangGraph에서 제공하는 특수 상수이다:
- `START`: 그래프의 진입점(entry point). 어떤 노드가 맨 처음 실행될지를 지정한다.
- `END`: 그래프의 종료점(exit point). 이 노드에 도달하면 그래프 실행이 완료된다.
- 이 두 상수를 사용하여 그래프의 시작과 끝을 명시적으로 선언한다.
- `compile()`은 빌더에 정의된 노드, 엣지 정보를 검증하고, 실행 가능한 `CompiledGraph` 객체를 반환한다. 컴파일 시점에 그래프의 구조적 오류(도달 불가능한 노드, 순환 경로 등)가 감지된다.
- 참고: 이 코드는 섹션 2.1에서 정의한 `MenuState`, 섹션 2.2.1의 `get_user_preference`, 섹션 2.2.2의 `recommend_menu`, 섹션 2.2.3의 `provide_menu_info` 함수를 사용합니다.
from langgraph.graph import StateGraph, START, END
# [LangGraph 내장] StateGraph: 상태 기반 그래프 빌더 클래스. State 스키마(TypedDict)를 인자로 받아 그래프를 구성한다.
# [LangGraph 내장] START: 그래프의 진입점을 나타내는 특수 상수. add_edge(START, "노드이름")으로 첫 번째 실행 노드를 지정한다.
# [LangGraph 내장] END: 그래프의 종료점을 나타내는 특수 상수. add_edge("노드이름", END)로 그래프 실행 완료를 선언한다.
# 1. 그래프 빌더 생성 - MenuState 스키마를 전달
# [사용자 정의 - 섹션 2.1 참조] MenuState: 사용자 선호도, 추천 메뉴, 메뉴 정보를 담는 TypedDict
# StateGraph(MenuState)를 호출하면 MenuState 스키마에 기반한 그래프 빌더가 생성된다.
builder = StateGraph(MenuState)
# 2. 노드 추가 - (문자열 이름, 함수) 쌍으로 등록
# 첫 번째 인자(문자열)는 그래프 내에서 노드를 식별하는 고유 이름이다.
# 두 번째 인자는 실제로 실행될 노드 함수이다.
# [사용자 정의 - 섹션 2.2.1 참조] get_user_preference: 랜덤 사용자 선호도를 생성하여 user_preference 필드를 채우는 함수
builder.add_node("get_preference", get_user_preference)
# [사용자 정의 - 섹션 2.2.2 참조] recommend_menu: user_preference를 읽어 recommended_menu를 설정하는 함수
builder.add_node("recommend", recommend_menu)
# [사용자 정의 - 섹션 2.2.3 참조] provide_menu_info: recommended_menu를 읽어 menu_info를 설정하는 함수
builder.add_node("provide_info", provide_menu_info)
# 3. 엣지 추가 - 노드 간 실행 순서를 정의
# [LangGraph 내장] add_edge(source, target): 두 노드를 선형(linear)으로 연결하는 메서드.
# source 노드 실행이 완료되면 반드시 target 노드가 실행된다.
builder.add_edge(START, "get_preference") # 시작 -> get_preference (그래프의 첫 번째 노드)
builder.add_edge("get_preference", "recommend") # get_preference -> recommend (선호도 생성 후 메뉴 추천)
builder.add_edge("recommend", "provide_info") # recommend -> provide_info (메뉴 추천 후 상세 정보 제공)
builder.add_edge("provide_info", END) # provide_info -> 종료 (상세 정보 제공 후 그래프 실행 완료)
# 4. 그래프 컴파일 - 빌더를 실행 가능한 그래프로 변환
# [LangGraph 내장] compile(): 빌더를 CompiledGraph 객체로 변환하는 메서드.
# 이 시점에 그래프 구조가 검증된다 (도달 불가능한 노드, 누락된 엣지 등).
graph = builder.compile()
- `add_node`의 첫 번째 인자(문자열)는 그래프 내에서 노드를 식별하는 이름이다. `add_edge`에서 이 이름을 사용하여 연결한다.
- `add_edge`는 선형(linear) 엣지로, source 노드 실행이 완료되면 반드시 target 노드가 실행된다.
- 메뉴 추천 그래프 실행 구조
- 위 다이어그램은 실제 LangGraph가 생성하는 그래프 구조와 동일하다.
- `__start__`와 `__end__`는 LangGraph 내부에서 START, END 상수에 대응하는 특수 노드이다.
2.3.1. 그래프 시각화
- LangGraph는 `get_graph().draw_mermaid_png()` 메서드를 통해 그래프를 이미지로 시각화할 수 있다.
- Jupyter 환경에서 IPython의 `display`와 `Image`를 사용하여 출력한다.
from IPython.display import Image, display
# 그래프 시각화
# graph: [사용자 정의 - 섹션 2.3 참조] builder.compile()로 생성된 CompiledGraph 인스턴스
# [LangGraph 내장] get_graph(): 그래프의 구조 정보를 반환하는 메서드
# [LangGraph 내장] draw_mermaid_png(): 그래프 구조를 mermaid 형식의 PNG 이미지로 변환하는 메서드
display(Image(graph.get_graph().draw_mermaid_png()))
- 출력되는 이미지는 위의 mermaid 다이어그램과 동일한 구조를 보여준다.
2.3.2. 그래프 실행 (invoke)
- `invoke()`는 컴파일된 그래프를 초기 상태와 함께 실행하는 메서드이다.
- 초기 상태는 State 스키마에 맞는 딕셔너리로 전달한다.
- 그래프는 START에서 시작하여 엣지를 따라 순서대로 노드를 실행하며, END에 도달하면 최종 상태를 반환한다.
- `invoke()`의 반환값은 모든 노드 실행이 완료된 후의 최종 상태 딕셔너리이다.
# [사용자 정의] 결과 출력을 위한 헬퍼 함수
# 그래프 실행 후 반환된 최종 상태를 보기 좋게 출력하기 위한 유틸리티 함수이다.
def print_result(result: MenuState):
# result: [사용자 정의 - 섹션 2.1 참조] MenuState TypedDict의 인스턴스 (최종 상태)
# 그래프의 모든 노드가 실행을 완료한 후의 상태로, 3개 필드가 모두 채워져 있다.
# - user_preference: get_user_preference 노드(섹션 2.2.1)가 설정한 사용자 선호도
# - recommended_menu: recommend_menu 노드(섹션 2.2.2)가 설정한 추천 메뉴
# - menu_info: provide_menu_info 노드(섹션 2.2.3)가 설정한 메뉴 상세 정보
print("\n=== 결과 ===")
print("선호도:", result['user_preference'])
print("추천 메뉴:", result['recommended_menu'])
print("메뉴 정보:", result['menu_info'])
print("============\n")
# 초기 상태 - user_preference만 빈 문자열로 설정
# recommended_menu와 menu_info는 그래프 실행 과정에서 각 노드가 채워넣는다.
inputs = {"user_preference": ""}
# 여러 번 실행하여 테스트 (랜덤 선호도이므로 매번 다른 결과)
for _ in range(2):
# [LangGraph 내장] invoke(): 초기 상태를 전달하여 그래프를 동기적으로 실행하는 메서드
# graph: [사용자 정의 - 섹션 2.3 참조] builder.compile()로 생성된 CompiledGraph 인스턴스
# 실행 흐름: START -> get_preference -> recommend -> provide_info -> END
result = graph.invoke(inputs)
# print_result: 위에서 정의한 헬퍼 함수. result의 3개 필드를 출력한다.
print_result(result)
print("*" * 100)
print()---랜덤 사용자 선호도 생성---
생성된 선호도: 채식
---메뉴 추천---
추천 메뉴: 그린 샐러드
---메뉴 정보 제공---
메뉴 정보: 신선한 유기농 채소로 만든 건강한 샐러드. 가격: 15,000원
=== 결과 ===
선호도: 채식
추천 메뉴: 그린 샐러드
메뉴 정보: 신선한 유기농 채소로 만든 건강한 샐러드. 가격: 15,000원
============
****************************************************************************************************
---랜덤 사용자 선호도 생성---
생성된 선호도: 해산물
---메뉴 추천---
추천 메뉴: 랍스터 파스타
---메뉴 정보 제공---
메뉴 정보: 신선한 랍스터와 al dente 파스타의 조화. 가격: 28,000원
=== 결과 ===
선호도: 해산물
추천 메뉴: 랍스터 파스타
메뉴 정보: 신선한 랍스터와 al dente 파스타의 조화. 가격: 28,000원
============
****************************************************************************************************
- 실행 결과를 분석하면:
- 첫 번째 실행: `get_preference`가 "채식"을 생성 -> `recommend`가 "그린 샐러드"를 추천 -> `provide_info`가 가격 정보를 제공
- 두 번째 실행: `get_preference`가 "해산물"을 생성 -> `recommend`가 "랍스터 파스타"를 추천 -> `provide_info`가 가격 정보를 제공
- 매번 `invoke()`를 호출할 때마다 새로운 상태가 생성되므로, 이전 실행의 상태가 다음 실행에 영향을 주지 않는다.
- 이것은 StateGraph의 중요한 특성이다: 각 invoke() 호출은 독립적인 실행이다.
- invoke() 실행 시퀀스 다이어그램
3. 조건부 엣지(Edge) - 조건에 따른 분기 처리
- StateGraph 라이프사이클에서의 위치: 이 섹션은 StateGraph의 고급 기능인 조건부 엣지(Conditional Edge)를 다룬다. 선형 엣지(`add_edge`)만으로는 모든 워크플로우를 표현할 수 없으며, 조건에 따라 다른 경로로 분기해야 하는 경우 조건부 엣지를 사용한다.
- 조건부 엣지는 런타임에 상태를 분석하여 다음에 실행할 노드를 동적으로 결정하는 엣지이다.
- 이 실습에서는 사용자 입력이 메뉴 관련 질문인지 아닌지에 따라 다른 처리 경로를 타는 시스템을 구현한다:
- 메뉴 관련 질문 -> 벡터저장소 검색 -> RAG 기반 응답 생성
- 메뉴 무관 질문 -> 일반 LLM 응답 생성
- 조건부 분기 그래프 개요
3.1. State 정의
- StateGraph 라이프사이클에서의 위치: 조건부 엣지가 포함된 그래프에서도 State 정의가 가장 먼저 수행된다. 이번 State는 이전 실습보다 복잡하며, 조건 분기에 사용할 boolean 필드(`is_menu_related`)가 포함된다.
- 사용자 입력이 메뉴 추천에 관한 것이면 벡터저장소에서 검색하여 RAG Chain을 실행하고, 그렇지 않으면 LLM이 직접 답변을 생성하는 구조이다.
- `is_menu_related` 필드는 조건부 엣지의 분기 조건으로 사용된다. 이 필드의 값에 따라 그래프의 실행 경로가 달라진다.
- 참고: 아래의 `MenuState`는 섹션 2.1에서 정의한 `MenuState`와는 별도의 클래스이다. 섹션 2.1의 `MenuState`는 `user_preference`, `recommended_menu`, `menu_info` 필드를 가지지만, 이 클래스는 조건부 분기를 위한 `user_query`, `is_menu_related`, `search_results`, `final_answer` 필드를 가진다. 같은 이름이지만 용도와 구조가 다르며, 섹션 3 전체(3.1~3.6)에서 사용되는 State 스키마이다.
from typing import TypedDict, List
# [사용자 정의] state 스키마 - 조건부 분기용 (섹션 2.1의 MenuState와는 별도의 클래스)
# 이 MenuState는 4개의 필드를 가지며, 각 필드는 그래프의 서로 다른 노드에서 읽고 쓰인다.
# 특히 is_menu_related 필드는 조건부 엣지(섹션 3.4)의 분기 조건으로 사용되는 핵심 필드이다.
class MenuState(TypedDict):
user_query: str # 사용자의 원래 질문 (get_user_query 노드에서 설정)
is_menu_related: bool # 메뉴 관련 질문 여부 (analyze_input 노드에서 설정, decide_next_step에서 분기 조건으로 사용)
search_results: List[str] # 벡터저장소 검색 결과 목록 (search_menu_info 노드에서 설정, 메뉴 관련 경로에서만 사용)
final_answer: str # 최종 생성된 답변 (generate_menu_response 또는 generate_general_response에서 설정)
- `search_results`는 `List[str]` 타입으로, 벡터저장소에서 검색된 여러 문서의 내용을 저장한다.
- 일반 응답 경로에서는 `search_results`가 사용되지 않으며, 이는 StateGraph에서 자연스러운 패턴이다. 모든 노드가 모든 필드를 사용할 필요는 없다.
- user_query 필드 사용 관계.
- is_menu_related 필드 사용 관계
- search_results 필드 사용 관계
- final_answer 필드 사용 관계
3.2. 벡터저장소 검색 도구
- StateGraph 라이프사이클에서의 위치: 이 단계는 노드에서 사용할 외부 도구(벡터저장소)를 초기화하는 단계이다. 노드 함수 내부에서 사용되므로 노드 정의 전에 준비해야 한다.
- 메뉴 검색을 위한 벡터저장소를 Chroma와 OllamaEmbeddings를 사용하여 초기화한다.
- `persist_directory`를 지정하여 기존에 저장된 벡터 인덱스를 로드한다.
- `collection_name="restaurant_menu"`는 레스토랑 메뉴 데이터가 저장된 컬렉션을 지정한다.
from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings
# [LangChain 내장] Chroma: 벡터저장소 클래스 (langchain_chroma 패키지). 텍스트를 벡터로 저장하고 유사도 검색을 수행한다.
# [LangChain 내장] OllamaEmbeddings: Ollama 기반 로컬 임베딩 모델 래퍼 클래스 (langchain_ollama 패키지).
# 텍스트를 고차원 벡터로 변환하여 의미적 유사도를 비교할 수 있게 한다.
# 임베딩 모델 초기화 - bge-m3는 다국어 지원 임베딩 모델
# Ollama 서버에서 로컬로 실행되므로 API 키 없이 사용 가능하다.
embeddings_model = OllamaEmbeddings(model="bge-m3")
# Chroma 인덱스 로드 - 기존 저장소에서 메뉴 데이터를 로드
# vector_db: 섹션 3.3.3의 search_menu_info 노드에서 similarity_search()로 유사도 검색에 사용됨
vector_db = Chroma(
embedding_function=embeddings_model,
collection_name="restaurant_menu",
persist_directory="./chroma_db",
)
- `OllamaEmbeddings`는 Ollama 서버에서 실행되는 로컬 임베딩 모델을 사용한다.
- `bge-m3` 모델은 한국어를 포함한 다국어 텍스트에 대해 고품질 임베딩을 생성한다.
- 벡터저장소는 `search_menu_info` 노드에서 사용자 질문과 유사한 메뉴 정보를 검색하는 데 활용된다.
3.3. 노드(Node) - 조건부 엣지용 노드 함수들
- StateGraph 라이프사이클에서의 위치: State 및 외부 도구가 준비된 후 노드 함수를 정의한다. 조건부 엣지 그래프에서는 분기 이후 서로 다른 경로의 노드들이 존재한다.
- 이 그래프에는 5개의 노드가 있으며, 각각의 역할이 명확하게 분리되어 있다.
- LLM을 사용하는 노드에서는 LangChain의 `ChatPromptTemplate`과 `StrOutputParser`를 체인으로 연결하여 사용한다.
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# [LangChain 내장] ChatPromptTemplate: 채팅 프롬프트 템플릿 클래스 (langchain_core.prompts).
# 문자열 템플릿에 변수를 삽입하여 LLM에 전달할 프롬프트를 생성한다.
# [LangChain 내장] StrOutputParser: LLM 출력을 문자열로 파싱하는 클래스 (langchain_core.output_parsers).
# LLM의 AIMessage 객체에서 텍스트 content만 추출한다.
# [LangChain 내장] ChatOpenAI: OpenAI 채팅 모델 래퍼 클래스 (langchain_openai).
# OpenAI API를 통해 GPT 모델을 호출한다.
# LLM 모델 초기화 - 섹션 3.3.2~3.3.5의 노드들에서 공통으로 사용됨
# gpt-4o-mini는 빠르고 비용 효율적인 모델로, 분류 및 응답 생성 작업에 적합하다.
llm = ChatOpenAI(model="gpt-4o-mini")
3.3.1. get_user_query 노드
- 사용자로부터 직접 입력을 받는 노드이다.
- `input()` 함수를 사용하여 대화형으로 질문을 입력받는다.
- 참고: 이 코드의 `state` 파라미터는 섹션 3.1에서 정의한 `MenuState` TypedDict이다 (섹션 2.1의 MenuState와는 다른 클래스).
# [사용자 정의] get_user_query 노드 함수
# 이 함수는 섹션 3.5에서 add_node("get_user_query", get_user_query)로 그래프에 등록된다.
# 역할: 사용자로부터 질문을 직접 입력받아 State의 user_query 필드를 채운다.
# 이후 analyze_input 노드(섹션 3.3.2)에서 이 user_query를 읽어 메뉴 관련 여부를 판단한다.
def get_user_query(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict (user_query, is_menu_related, search_results, final_answer)
user_query = input("무엇을 도와드릴까요? ")
return {"user_query": user_query} # user_query 키만 업데이트
- 이 노드는 외부 입력(사용자)을 State에 주입하는 역할을 한다.
- StateGraph에서 외부 입력은 초기 상태(`invoke` 인자)로 전달하거나, 이처럼 노드 내부에서 직접 받을 수 있다.
3.3.2. analyze_input 노드
- LLM을 사용하여 사용자 입력이 메뉴/음식 관련 질문인지 판단하는 노드이다.
- 이 노드의 출력(`is_menu_related`)이 조건부 엣지의 분기 조건이 된다.
- 참고: 이 코드는 섹션 3.1에서 정의한 `MenuState`의 `user_query` 필드를 읽고, 섹션 3.3 상단에서 초기화한 `llm`(ChatOpenAI 인스턴스)을 사용합니다.
# [사용자 정의] analyze_input 노드 함수
# 이 함수는 섹션 3.5에서 add_node("analyze_input", analyze_input)로 그래프에 등록된다.
# 역할: LLM을 사용하여 사용자 질문이 메뉴/음식 관련인지 여부를 판단한다.
# 이 노드의 출력(is_menu_related)이 decide_next_step 함수(섹션 3.4)에서 조건부 분기의 기준이 된다.
def analyze_input(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict (user_query, is_menu_related, search_results, final_answer)
analyze_template = """
사용자의 입력을 분석하여 레스토랑 메뉴 추천이나 음식 정보에 관한 질문인지 판단하세요.
사용자 입력: {user_query}
레스토랑 메뉴나 음식 정보에 관한 질문이면 "True", 아니면 "False"로 답변하세요.
답변:
"""
# [LangChain 내장] ChatPromptTemplate.from_template(): 문자열 템플릿으로 프롬프트 생성
analyze_prompt = ChatPromptTemplate.from_template(analyze_template)
# [LangChain 내장] | (파이프 연산자): 프롬프트 -> LLM -> 파서를 체인(LCEL)으로 연결
# llm: [사용자 정의 - 섹션 3.3 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# StrOutputParser(): LLM의 AIMessage에서 텍스트 content만 추출하는 파서
analyze_chain = analyze_prompt | llm | StrOutputParser()
# state['user_query']: [사용자 정의 - 섹션 3.3.1 참조] get_user_query 노드에서 설정한 사용자 입력
result = analyze_chain.invoke({"user_query": state['user_query']})
# LLM의 응답(예: "True", "true", " True ")을 strip/lower 후 비교하여 bool 변환
is_menu_related = result.strip().lower() == "true"
return {"is_menu_related": is_menu_related} # is_menu_related 키만 업데이트
- LLM의 응답을 `"true"`와 비교하여 boolean 값으로 변환한다.
- 이 패턴은 LLM을 "분류기(classifier)"로 사용하는 일반적인 방법이다.
- `is_menu_related` 값이 `True`면 메뉴 관련 경로, `False`면 일반 응답 경로로 분기된다.
3.3.3. search_menu_info 노드
- 벡터저장소에서 사용자 질문과 유사한 메뉴 정보를 검색하는 노드이다.
- 메뉴 관련 질문일 때만 실행되는 경로에 위치한다.
- 참고: 이 코드는 섹션 3.2에서 초기화한 `vector_db`(Chroma 벡터저장소 인스턴스)를 사용하고, 섹션 3.1에서 정의한 `MenuState`의 `user_query` 필드를 읽습니다.
# [사용자 정의] search_menu_info 노드 함수
# 이 함수는 섹션 3.5에서 add_node("search_menu_info", search_menu_info)로 그래프에 등록된다.
# 역할: 벡터저장소에서 사용자 질문과 의미적으로 유사한 메뉴 문서를 검색한다.
# 조건부 분기에 의해 is_menu_related가 True일 때만 실행되는 경로에 위치한다.
# 검색 결과(search_results)는 이후 generate_menu_response 노드(섹션 3.3.4)에서 RAG 응답 생성에 사용된다.
def search_menu_info(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict (user_query, is_menu_related, search_results, final_answer)
# vector_db: [사용자 정의 - 섹션 3.2 참조] Chroma 벡터저장소 인스턴스 (restaurant_menu 컬렉션)
# state['user_query']: [사용자 정의 - 섹션 3.3.1 참조] get_user_query 노드에서 설정한 사용자 입력
# 벡터저장소에서 최대 2개의 문서를 검색 (k=2)
# similarity_search는 사용자 질문을 임베딩 벡터로 변환한 후 코사인 유사도 기준으로 가장 유사한 문서를 반환한다.
results = vector_db.similarity_search(state['user_query'], k=2)
# 검색된 Document 객체에서 page_content(텍스트 내용)만 추출하여 문자열 리스트로 변환
search_results = [doc.page_content for doc in results]
return {"search_results": search_results} # search_results 키만 업데이트
- `similarity_search`는 사용자 질문을 임베딩으로 변환한 후, 벡터저장소에서 가장 유사한 k개의 문서를 검색한다.
- 검색 결과에서 `page_content`만 추출하여 문자열 리스트로 State에 저장한다.
3.3.4. generate_menu_response 노드
- 검색 결과를 활용하여 RAG(Retrieval-Augmented Generation) 방식으로 답변을 생성하는 노드이다.
- `user_query`와 `search_results` 두 필드를 읽어 프롬프트에 포함한다.
- 참고: 이 코드는 섹션 3.3 상단에서 초기화한 `llm`(ChatOpenAI 인스턴스)을 사용하고, 섹션 3.1에서 정의한 `MenuState`의 `user_query`(섹션 3.3.1에서 설정)와 `search_results`(섹션 3.3.3에서 설정) 필드를 읽습니다.
# [사용자 정의] generate_menu_response 노드 함수
# 이 함수는 섹션 3.5에서 add_node("generate_menu_response", generate_menu_response)로 그래프에 등록된다.
# 역할: 벡터저장소 검색 결과(search_results)를 컨텍스트로 활용하여 RAG 방식의 상세 답변을 생성한다.
# RAG(Retrieval-Augmented Generation)는 검색된 문서를 근거로 LLM이 답변을 생성하는 패턴이다.
def generate_menu_response(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict (user_query, is_menu_related, search_results, final_answer)
# state['user_query']: [사용자 정의 - 섹션 3.3.1 참조] get_user_query 노드에서 설정한 사용자 입력
# state['search_results']: [사용자 정의 - 섹션 3.3.3 참조] search_menu_info 노드에서 설정한 벡터저장소 검색 결과 리스트
response_template = """
사용자 입력: {user_query}
메뉴 관련 검색 결과: {search_results}
위 정보를 바탕으로 사용자의 메뉴 관련 질문에 대한 상세한 답변을 생성하세요.
검색 결과의 정보를 활용하여 정확하고 유용한 정보를 제공하세요.
답변:
"""
# [LangChain 내장] ChatPromptTemplate.from_template(): 문자열 템플릿으로 프롬프트 생성
response_prompt = ChatPromptTemplate.from_template(response_template)
# llm: [사용자 정의 - 섹션 3.3 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# LCEL 체인: 프롬프트 -> LLM -> 문자열 파서
response_chain = response_prompt | llm | StrOutputParser()
final_answer = response_chain.invoke({
"user_query": state['user_query'],
"search_results": state['search_results']
})
print(f"\n메뉴 어시스턴트: {final_answer}")
return {"final_answer": final_answer} # final_answer 키만 업데이트
- 이 노드는 RAG 체인의 "생성(Generation)" 단계에 해당한다.
- 프롬프트에 검색 결과를 컨텍스트로 포함하여 LLM이 근거 있는 답변을 생성하도록 유도한다.
3.3.5. generate_general_response 노드
- 메뉴와 관련 없는 질문에 대해 일반적인 답변을 생성하는 노드이다.
- 벡터저장소 검색 없이 LLM만으로 응답한다.
- 참고: 이 코드는 섹션 3.3 상단에서 초기화한 `llm`(ChatOpenAI 인스턴스)을 사용하고, 섹션 3.1에서 정의한 `MenuState`의 `user_query`(섹션 3.3.1에서 설정) 필드를 읽습니다.
# [사용자 정의] generate_general_response 노드 함수
# 이 함수는 섹션 3.5에서 add_node("generate_general_response", generate_general_response)로 그래프에 등록된다.
# 역할: 메뉴와 무관한 일반 질문에 대해 LLM만으로 답변을 생성한다.
# 조건부 분기에 의해 is_menu_related가 False일 때만 실행되는 경로에 위치한다.
# search_results 필드는 사용하지 않으며, user_query만 읽어 프롬프트에 전달한다.
def generate_general_response(state: MenuState) -> MenuState:
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict (user_query, is_menu_related, search_results, final_answer)
# state['user_query']: [사용자 정의 - 섹션 3.3.1 참조] get_user_query 노드에서 설정한 사용자 입력
response_template = """
사용자 입력: {user_query}
위 입력은 레스토랑 메뉴나 음식과 관련이 없습니다.
일반적인 대화 맥락에서 적절한 답변을 생성하세요.
답변:
"""
# [LangChain 내장] ChatPromptTemplate.from_template(): 문자열 템플릿으로 프롬프트 생성
response_prompt = ChatPromptTemplate.from_template(response_template)
# llm: [사용자 정의 - 섹션 3.3 참조] ChatOpenAI(model="gpt-4o-mini") 인스턴스
# LCEL 체인: 프롬프트 -> LLM -> 문자열 파서
response_chain = response_prompt | llm | StrOutputParser()
final_answer = response_chain.invoke({"user_query": state['user_query']})
print(f"\n일반 어시스턴트: {final_answer}")
return {"final_answer": final_answer} # final_answer 키만 업데이트
- `generate_menu_response`와 `generate_general_response`는 모두 `final_answer`를 기록하지만, 사용하는 프롬프트와 컨텍스트가 다르다.
- 두 노드 중 하나만 실행되므로, `final_answer`에 대한 충돌은 발생하지 않는다.
- 조건부 분기 노드 역할 요약
3.4. 엣지(Edge) - 조건부 엣지 정의
- StateGraph 라이프사이클에서의 위치: 노드 정의 후, 그래프 빌드 전에 조건부 분기 함수를 정의한다. 이 함수는 `add_conditional_edges`에 전달되어 런타임 분기를 수행한다.
- 조건부 엣지는 `add_conditional_edges()` 메서드로 추가하며, 핵심은 경로 결정 함수(path function)이다.
- 경로 결정 함수의 규칙:
- 현재 State를 인자로 받는다.
- 다음에 실행할 노드의 이름(문자열)을 반환한다.
- 반환값의 타입 힌트로 `Literal`을 사용하여 가능한 반환값을 명시한다.
- `Literal` 타입 힌트는 LangGraph가 그래프 구조를 검증하고 시각화하는 데 활용된다.
- 참고: 이 코드는 섹션 3.1에서 정의한 `MenuState`의 `is_menu_related` 필드(섹션 3.3.2의 `analyze_input` 노드에서 설정)를 읽어 분기 조건으로 사용합니다.
from typing import Literal
# [사용자 정의] decide_next_step 경로 결정 함수 - add_conditional_edges에 전달됨
# 이 함수는 섹션 3.5에서 add_conditional_edges("analyze_input", decide_next_step, {...})의
# 두 번째 인자(path)로 전달된다.
# 역할: analyze_input 노드 실행 후 호출되어, is_menu_related 값에 따라 다음 노드를 결정한다.
# 주의: 경로 함수는 순수 함수(pure function)로, State를 읽기만 하고 수정하지 않는다.
def decide_next_step(state: MenuState) -> Literal["search_menu_info", "generate_general_response"]:
"""상태의 is_menu_related 값에 따라 다음 노드를 결정하는 경로 함수"""
# state: [사용자 정의 - 섹션 3.1 참조] MenuState TypedDict
# state['is_menu_related']: [사용자 정의 - 섹션 3.3.2 참조] analyze_input 노드에서 설정한 bool 값
# True면 사용자 질문이 메뉴/음식 관련, False면 메뉴와 무관한 일반 질문
if state['is_menu_related']:
return "search_menu_info" # 메뉴 관련 -> 벡터저장소 검색 노드(섹션 3.3.3)로 이동
else:
return "generate_general_response" # 메뉴 무관 -> 일반 응답 노드(섹션 3.3.5)로 이동
- `Literal["search_menu_info", "generate_general_response"]` 타입 힌트의 의미:
- 이 함수가 반환할 수 있는 값은 `"search_menu_info"` 또는 `"generate_general_response"` 두 가지뿐이다.
- LangGraph는 이 타입 힌트를 사용하여 그래프 시각화 시 가능한 분기 경로를 표시한다.
- 런타임에 이 두 값 외의 문자열이 반환되면 오류가 발생한다.
- 경로 함수는 순수 함수(pure function)여야 한다: State를 읽기만 하고, 수정하지 않는다. 상태 수정은 노드의 책임이다.
- 조건부 엣지 동작 원리
3.5. 그래프(Graph) 구성 - 조건부 분기 포함 그래프
- StateGraph 라이프사이클에서의 위치: 모든 노드, 엣지, 조건부 분기 함수가 정의된 후, 이들을 조합하여 그래프를 빌드하고 컴파일하는 최종 단계이다.
- `add_conditional_edges()` 메서드의 3개 인자를 상세히 설명한다:
- source (첫 번째 인자): 조건부 분기가 시작되는 노드의 이름. 이 노드 실행 후 경로 함수가 호출된다.
- path (두 번째 인자): 경로 결정 함수. State를 받아 다음 노드 이름을 반환한다.
- path_map (세 번째 인자): 경로 함수의 반환값과 실제 노드 이름을 매핑하는 딕셔너리. 경로 함수의 반환값과 노드 이름이 동일한 경우에도 명시적으로 매핑을 작성하는 것이 관례이다.
- 참고: 이 코드는 섹션 3.1의 `MenuState`, 섹션 3.3.1~3.3.5의 노드 함수들(`get_user_query`, `analyze_input`, `search_menu_info`, `generate_menu_response`, `generate_general_response`), 섹션 3.4의 경로 결정 함수(`decide_next_step`)를 사용합니다.
from langgraph.graph import StateGraph, START, END
# [LangGraph 내장] StateGraph: 상태 기반 그래프 빌더 클래스
# [LangGraph 내장] START, END: 그래프의 진입점/종료점을 나타내는 특수 상수
# 그래프 구성
# [사용자 정의 - 섹션 3.1 참조] MenuState: 조건부 분기용 State (user_query, is_menu_related, search_results, final_answer)
builder = StateGraph(MenuState)
# 노드 추가 - 5개의 노드를 등록
# 각 노드의 첫 번째 인자(문자열)는 그래프 내에서의 고유 식별자이며, add_edge에서 이 이름을 참조한다.
# [사용자 정의 - 섹션 3.3.1 참조] get_user_query: 사용자 입력을 input()으로 받아 user_query 필드를 설정하는 함수
builder.add_node("get_user_query", get_user_query)
# [사용자 정의 - 섹션 3.3.2 참조] analyze_input: LLM으로 질문을 분류하여 is_menu_related 필드를 설정하는 함수
builder.add_node("analyze_input", analyze_input)
# [사용자 정의 - 섹션 3.3.3 참조] search_menu_info: 벡터저장소에서 유사 문서를 검색하여 search_results를 설정하는 함수
builder.add_node("search_menu_info", search_menu_info)
# [사용자 정의 - 섹션 3.3.4 참조] generate_menu_response: 검색 결과 기반 RAG 응답을 생성하여 final_answer를 설정하는 함수
builder.add_node("generate_menu_response", generate_menu_response)
# [사용자 정의 - 섹션 3.3.5 참조] generate_general_response: 일반 LLM 응답을 생성하여 final_answer를 설정하는 함수
builder.add_node("generate_general_response", generate_general_response)
# [LangGraph 내장] add_edge(): 선형 엣지 추가 - START에서 analyze_input까지는 순차 실행
builder.add_edge(START, "get_user_query") # 그래프 시작 -> 사용자 입력 수집
builder.add_edge("get_user_query", "analyze_input") # 사용자 입력 수집 -> 질문 분류
# [LangGraph 내장] add_conditional_edges(): 조건부 엣지 추가 - analyze_input 이후 분기
# analyze_input 노드 실행 완료 후 decide_next_step 함수가 호출되어 다음 노드가 결정된다.
builder.add_conditional_edges(
"analyze_input", # source: 이 노드 실행 후 경로 함수가 호출됨
decide_next_step, # path: [사용자 정의 - 섹션 3.4 참조] 경로 결정 함수 (is_menu_related 값에 따라 분기)
{ # path_map: 경로 함수의 반환값 -> 실제 실행할 노드 이름 매핑
"search_menu_info": "search_menu_info", # 메뉴 관련 -> 벡터저장소 검색
"generate_general_response": "generate_general_response" # 메뉴 무관 -> 일반 응답
}
)
# 분기 이후 선형 엣지 - 각 경로의 마지막 노드에서 END로
builder.add_edge("search_menu_info", "generate_menu_response") # 벡터저장소 검색 -> RAG 응답 생성
builder.add_edge("generate_menu_response", END) # RAG 응답 생성 -> 그래프 종료
builder.add_edge("generate_general_response", END) # 일반 응답 생성 -> 그래프 종료
# [LangGraph 내장] compile(): 빌더를 실행 가능한 CompiledGraph 객체로 변환
# 이 시점에 그래프 구조가 검증된다 (모든 노드에 도달 가능한지, 누락된 엣지는 없는지 등).
graph = builder.compile()
- 위 코드에서 핵심은 `add_conditional_edges` 호출이다.
- `path_map`의 키는 경로 함수(`decide_next_step`)가 반환할 수 있는 값이고, 값은 실제로 실행할 노드의 이름이다.
- 이 예제에서는 키와 값이 동일하지만, 경로 함수가 `"menu"`를 반환하고 실제 노드 이름이 `"search_menu_info"`인 경우처럼 다를 수도 있다:
- 예: `{"menu": "search_menu_info", "general": "generate_general_response"}`
3.5.1. 그래프 시각화
from IPython.display import Image, display
# graph: [사용자 정의 - 섹션 3.5 참조] builder.compile()로 생성된 CompiledGraph 인스턴스
# [LangGraph 내장] get_graph(): 그래프의 구조 정보를 반환하는 메서드
# [LangGraph 내장] draw_mermaid_png(): 그래프 구조를 mermaid 형식의 PNG 이미지로 변환하는 메서드
display(Image(graph.get_graph().draw_mermaid_png()))
- 시각화 결과는 아래 mermaid 다이어그램과 동일한 구조를 보여준다.
- 조건부 분기 포함 전체 그래프 구조
- 그래프 시각화에서 `analyze_input`에서 두 갈래로 분기되는 것을 확인할 수 있다.
- 각 분기의 라벨은 경로 함수의 반환값(또는 path_map의 키)이다.
3.6. Graph 실행 - while loop 대화
- StateGraph 라이프사이클에서의 위치: 컴파일된 그래프를 반복적으로 실행하여 대화형 시스템을 구현하는 최종 실행 단계이다.
- `while True` 루프를 사용하여 사용자가 대화를 종료할 때까지 반복적으로 그래프를 실행한다.
- 매 루프마다 새로운 초기 상태로 `invoke()`를 호출하므로, 각 대화 턴은 독립적인 그래프 실행이다.
while True:
# 매 반복마다 빈 초기 상태를 생성하여 독립적인 그래프 실행을 보장한다.
initial_state = {'user_query': ''}
# [LangGraph 내장] invoke(): 초기 상태를 전달하여 그래프를 동기적으로 실행
# graph: [사용자 정의 - 섹션 3.5 참조] builder.compile()로 생성한 CompiledGraph 인스턴스
# 실행 흐름: START -> get_user_query -> analyze_input -> (조건부 분기) -> ... -> END
graph.invoke(initial_state)
continue_chat = input("다른 질문이 있으신가요? (y/n): ").lower()
if continue_chat != 'y':
print("대화를 종료합니다. 감사합니다!")
break메뉴 어시스턴트: 티라미수는 이탈리아의 전통 디저트로, 부드러운 마스카포네 치즈 크림과 에스프레소에 적신
레이디핑거 비스킷을 층층이 쌓아 만든 디저트입니다. 이 디저트는 고소한 카카오 파우더를 듬뿍 뿌려 풍미를
더하였으며, 커피의 쌉싸름함과 치즈의 부드러움이 조화롭게 어우러지는 맛이 특징입니다.
가격은 ₩9,000이며, 주요 식재료로는 마스카포네 치즈, 에스프레소, 카카오 파우더, 레이디핑거 비스킷이
사용됩니다. 티라미수는 그 자체로도 훌륭한 디저트이지만, 커피와 함께 즐기면 더욱 맛있습니다.
부드럽고 달콤한 맛을 좋아하시는 분들께 추천드립니다!
대화를 종료합니다. 감사합니다!
- 위 출력은 사용자가 "티라미수" 관련 질문을 했을 때의 실행 결과이다.
- 실행 흐름: `get_user_query` -> `analyze_input`(메뉴 관련으로 판단) -> `search_menu_info`(벡터저장소에서 티라미수 정보 검색) -> `generate_menu_response`(검색 결과 기반 상세 답변 생성)
- 대화 루프 실행 시퀀스 다이어그램
4. StateGraph 핵심 개념 정리
- 이 섹션에서는 앞서 실습한 내용을 바탕으로 StateGraph의 핵심 개념을 체계적으로 정리한다.
4.1. StateGraph가 필요한 이유
- 단순한 함수 호출 체인(`func1() -> func2() -> func3()`)으로도 순차적 처리는 가능하다.
- 그러나 LLM 기반 애플리케이션에서는 다음과 같은 요구사항이 발생한다:
- 조건에 따른 동적 분기 (if/else가 아닌 LLM 판단 기반)
- 상태의 점진적 누적과 전달
- 실행 과정의 시각화와 디버깅
- 노드 단위의 재시도, 체크포인트, 스트리밍
- StateGraph는 이러한 요구사항을 선언적으로 해결한다. 개발자는 "무엇을" 할지(노드)와 "어떤 순서로" 할지(엣지)를 선언하면, LangGraph 런타임이 실행을 관리한다.
4.2. State 설계 원칙
- State는 `TypedDict`를 사용하여 정의한다. 이는 Python의 타입 시스템을 활용하여 IDE 자동완성과 타입 검사를 지원한다.
- 각 노드는 전체 State를 인자로 받지만, 반환할 때는 업데이트할 키만 포함한 부분 딕셔너리를 반환한다.
- 기본 동작은 override(덮어쓰기)이다: 노드가 반환한 키의 값이 기존 값을 대체한다.
- LangGraph는 `typing.Annotated` 타입과 `reducer` 함수를 통해 override 대신 append, merge 등의 커스텀 업데이트 로직을 지원한다. 예를 들어, `langgraph.graph`의 `add_messages` reducer를 사용하면 채팅 메시지 리스트를 누적할 수 있다:
from typing import Annotated
from langgraph.graph import add_messages
class ChatState(TypedDict):
messages: Annotated[list, add_messages] # 메시지가 override 대신 누적됨
- 이러한 `Annotated` + reducer 패턴은 이후 문서에서 상세히 다룬다.
4.3. 노드 함수 시그니처
- 모든 노드 함수는 동일한 시그니처를 따른다:
- 입력: `state: StateType` (전체 상태)
- 출력: `dict` (업데이트할 키-값 쌍)
- 이 일관된 시그니처 덕분에 노드를 자유롭게 교체, 추가, 삭제할 수 있다.
4.4. 엣지 종류 비교
- `add_edge(source, target)`: 선형 엣지. source 완료 후 반드시 target이 실행된다. 분기 없이 고정된 순서를 정의한다.
- `add_conditional_edges(source, path_fn, path_map)`: 조건부 엣지. source 완료 후 path_fn이 호출되어 다음 노드가 동적으로 결정된다.
- 선형 엣지와 조건부 엣지 비교
4.5. START와 END 상수
- `START`: `langgraph.graph` 모듈에서 임포트하는 특수 상수. 그래프의 진입점을 정의한다. `add_edge(START, "first_node")`로 첫 번째 노드를 지정한다.
- `END`: 그래프의 종료점. 노드가 `END`로 연결되면 해당 경로의 실행이 완료된다.
- 하나의 그래프에서 여러 노드가 `END`로 연결될 수 있다 (조건부 분기의 각 경로가 각각 END로 연결).
4.6. compile()과 invoke()
- `compile()`: 빌더 패턴으로 구성한 그래프를 실행 가능한 `CompiledGraph` 객체로 변환한다.
- 컴파일 시점에 그래프 구조가 검증된다 (도달 불가능한 노드, 누락된 엣지 등).
- 컴파일된 그래프는 `invoke()`, `stream()`, `batch()` 등의 실행 메서드를 제공한다.
- `invoke(initial_state)`: 초기 상태를 전달하여 그래프를 동기적으로 실행한다.
- START부터 END까지 모든 노드를 실행한 후 최종 상태를 반환한다.
- 각 `invoke()` 호출은 독립적이며, 이전 호출의 상태를 유지하지 않는다.
4.7. add_conditional_edges 파라미터 상세
- `add_conditional_edges`는 세 개의 인자를 받는다:
builder.add_conditional_edges(
"source_node", # 1) source: 분기 시작점 노드 이름
path_function, # 2) path: 경로 결정 함수 (State -> str)
{ # 3) path_map: 반환값 -> 노드 이름 매핑
"return_val_1": "target_node_1",
"return_val_2": "target_node_2",
}
)
- source: 이 노드의 실행이 완료된 후 경로 함수가 호출된다.
- path: `(state: StateType) -> str` 시그니처의 함수. 현재 상태를 분석하여 다음 노드를 결정한다.
- path_map: 경로 함수의 반환값을 실제 노드 이름에 매핑한다. path_map을 생략하면 경로 함수의 반환값이 곧 노드 이름으로 사용된다.
4.8. Literal 타입 힌트의 역할
- 경로 함수의 반환 타입에 `Literal`을 사용하면 두 가지 이점이 있다:
- 정적 분석: IDE와 타입 체커가 잘못된 반환값을 감지할 수 있다.
- 그래프 시각화: LangGraph가 가능한 분기 경로를 파악하여 시각화에 반영한다.
- `Literal` 없이도 코드는 동작하지만, 사용하는 것이 모범 사례(best practice)이다.
# Literal 사용 (권장)
def decide_next_step(state: MenuState) -> Literal["search_menu_info", "generate_general_response"]:
...
# Literal 미사용 (비권장)
def decide_next_step(state: MenuState) -> str:
...
5. 전체 흐름 요약
- StateGraph를 구성하는 전체 과정을 순서대로 정리한다.
- StateGraph 구성 7단계 프로세스
- 1단계: `TypedDict`로 State 스키마를 정의한다. 그래프를 흐르는 데이터의 구조를 결정한다.
- 2단계: 각 노드의 작업을 함수로 구현한다. 모든 노드는 동일한 시그니처(`state -> dict`)를 따른다.
- 3단계: `StateGraph(StateType)`으로 빌더 인스턴스를 생성한다.
- 4단계: `add_node(name, function)`으로 노드를 등록한다.
- 5단계: `add_edge` 또는 `add_conditional_edges`로 노드 간 연결을 정의한다. `START`와 `END`를 사용하여 진입점과 종료점을 지정한다.
- 6단계: `compile()`로 실행 가능한 그래프를 생성한다.
- 7단계: `invoke(initial_state)`로 그래프를 실행하고 최종 상태를 받는다.
'Study > LangChain' 카테고리의 다른 글
| 4. LangGraph - LangGraph ReAct Agent와 MemorySaver 학습 매뉴얼 (0) | 2026.05.05 |
|---|---|
| 3. LangGraph - LangGraph State Reducer & 메시지 상태 관리 학습 매뉴얼 (0) | 2026.05.04 |
| 1. LangGraph - LangChain Tool Calling 학습 매뉴얼 (0) | 2026.03.24 |
| 10. LangChain HuggingFace 오픈소스 언어모델 활용방법 (1) | 2026.03.16 |
| 8. LangChain 실무 설계 고려사항 총정리 (0) | 2026.03.08 |
댓글