LangGraph 활용 - 상태 그래프(StateGraph) 구현 개념 설명
LangGraph는 상태 기반 그래프(StateGraph) 구조를 활용하여 대화 흐름을 체계적으로 관리하는 데 사용됩니다. 쉽게 말해, 챗봇이나 복잡한 데이터 흐름을 제어하기 위한 "노드(Node)와 엣지(Edge)를 활용한 시스템" 입니다.
1. StateGraph란?
StateGraph는 "상태(State) 기반의 그래프(Graph) 구조" 를 사용하여 대화 흐름을 체계적으로 관리하는 방법입니다.
💡 비유적으로 설명하면?
일반적인 함수 기반 처리는 직선 도로와 같아서, 입력을 넣으면 정해진 방식대로 처리되고 끝납니다.
반면, StateGraph 기반의 처리는 지하철 노선도처럼 여러 노드(Node)와 엣지(Edge) 가 있어, 상황에 따라 다른 경로로 이동할 수 있습니다.
🎯 StateGraph의 핵심 구조
여러 개의 노드(Node) 가 엣지(Edge, 연결선) 로 연결되어 흐름을 형성합니다.
데이터(State)는 흐름에 따라 여러 노드에서 처리됩니다.
특정 조건에 따라 다른 경로로 분기(조건부 Edge) 할 수 있습니다.
2. State(상태)란?
State는 그래프에서 다루는 데이터의 기본 구조를 말합니다. 쉽게 말해, 현재 챗봇이나 시스템이 알고 있는 정보라고 볼 수 있습니다.
📌 예를 들어, 챗봇이 사용자의 이름을 알고 있다면?
State = {"username": "Soyee"}
📌 주문 시스템이라면?
State = {"order_id": 12345, "status": "processing"}
🔹 상태(State)의 특징
상태는 그래프 내에서 공유되는 데이터입니다.
특정 노드에서 상태를 수정(override, 덮어쓰기) 할 수 있습니다.
TypedDict(타입 지정 딕셔너리)를 사용하여 상태 구조를 미리 정의합니다.
🛠 예제: TypedDict를 활용한 상태 정의
from typing import TypedDict
class ChatState(TypedDict):
username: str
messages: list[str]💡 위처럼 TypedDict를 사용하면 상태 구조를 미리 정의할 수 있어, 어떤 데이터가 들어올지 예측할 수 있습니다.
3. Node(노드)란?
Node는 그래프의 기본 구성 요소로, 특정 작업을 수행하는 독립적인 단위입니다.
💡 비유적으로 설명하면?
노드는 하나의 기차역과 같고,
노드 간의 연결(Edge)은 기차 선로와 같습니다.
사용자가 A에서 B로 가려면, 특정 노드를 거쳐야 합니다.
🔹 노드의 역할
특정 입력을 받아 처리 후 새로운 상태를 반환합니다.
다음 노드와 연결될 수 있습니다.
개별적으로 실행될 수도 있고, 여러 개가 연결되어 흐름을 만들 수도 있습니다.
🛠 예제: 노드 구현
def greet_user(state: ChatState) -> ChatState:
username = state.get("username", "Guest")
state["messages"].append(f"Hello, {username}!")
return state
💡 greet_user 노드는 사용자의 이름을 불러서 인사 메시지를 추가하는 역할을 합니다.
4. Graph(그래프)란?
Graph는 여러 개의 노드(Node)들을 엣지(Edge)로 연결한 구조입니다.
💡 비유적으로 설명하면?
그래프는 도시의 도로망과 같습니다.
어떤 길(Edge)을 선택하느냐에 따라 목적지(Node)가 달라질 수 있습니다.
🔹 그래프의 특징
여러 개의 노드와 엣지가 연결된 집합체입니다.
각 노드 간의 연결 관계가 전체 데이터 흐름을 결정합니다.
빌드(Build) 과정이 완료되면 실행할 수 있습니다.
🛠 예제: 그래프 구성
from langgraph.graph import StateGraph
graph = StateGraph(ChatState)
graph.add_node("greet", greet_user)
graph.set_entry_point("greet")
💡 위 코드에서는 StateGraph(ChatState)를 생성하고, greet_user 노드를 등록한 뒤, 실행의 시작점을 "greet"으로 설정했습니다.
5. invoke 실행
invoke는 그래프 실행의 가장 기본적인 방법입니다.
🔹 invoke 실행 방식
모든 노드가 순차적으로 실행된 후 최종 결과를 반환합니다.
실행이 완료될 때까지 동기적으로 대기해야 합니다.
중간 상태를 확인할 수 없습니다.
🛠 예제: invoke 실행
result = graph.invoke({"username": "Soyee", "messages": []})
print(result["messages"])
💡 invoke는 모든 처리가 끝난 후 최종 결과만 반환하기 때문에 단순한 작업에 적합합니다.
6. 조건부 엣지(Edge)란?
Edge는 노드와 노드를 연결하는 경로입니다.
🔹 조건부 엣지(Conditional Edge)의 특징
사용자의 입력이나 상태에 따라 다른 노드로 이동할 수 있습니다.
챗봇의 대화 흐름을 동적으로 변경할 수 있습니다.
📌 예를 들어, 사용자가 "안녕"이라고 입력하면 greet 노드로 이동하고,
"문제 있어"라고 입력하면 support 노드로 이동하도록 만들 수 있습니다.
🛠 예제: 조건부 엣지 설정
def route_message(state: ChatState):
message = state["messages"][-1]
return "support" if "문제" in message else "greet"
graph.add_node("support", handle_support)
graph.add_edge("greet", "route", route_message)
graph.add_edge("route", "support", condition=lambda state: "문제" in state["messages"][-1])
💡 사용자의 입력에 따라 "support" 또는 "greet" 경로로 이동합니다.
7. Stream 실행
stream 방식은 그래프의 실행 과정을 실시간으로 확인할 수 있는 실행 방식입니다.
🔹 Stream 실행의 장점
노드의 실행 결과를 순차적으로 확인 가능합니다.
실시간 피드백이 필요할 때 유용합니다.
대화형 인터페이스에서 사용자 경험을 개선할 수 있습니다.
🛠 예제: stream 실행
for update in graph.stream({"username": "Soyee", "messages": []}):
print(update)
💡 stream을 사용하면 처리 중인 데이터를 실시간으로 확인할 수 있어 대화형 챗봇에 적합합니다.
8. Stream Mode 옵션
stream_mode 옵션을 활용하면, 스트리밍 실행 시 어떤 정보를 받을지 결정할 수 있습니다.
🛠 예제: stream_mode 적용
for update in graph.stream({"username": "Soyee", "messages": []}, stream_mode="updates"):
print(update)
💡 "updates" 모드를 사용하면 디버깅과 로그 분석에 유용합니다.
🚀 정리
StateGraph: 상태(State) 기반의 그래프 구조로 대화 흐름을 관리
State(상태): 현재 챗봇이 알고 있는 데이터
Node(노드): 독립적인 작업 단위
Graph(그래프): 여러 노드들이 연결된 구조
invoke: 동기 실행 (최종 결과만 반환)
조건부 Edge: 입력에 따라 경로를 동적으로 변경
stream: 실시간 실행 결과 확인 가능
