[LangGraph] StateGraph 상태그래프

StateGraph 상태 그래프

1. 개념

• 상태(State)를 기반으로 구축하는 가장 기본적인 그래프
• 작업자인 노드(Node)는 상태(State)에 데이터를 읽고 쓰면서 다른 노드와 통신한다.
• 다음 포스팅부터 살펴볼 상태(State), 노드(Node), 엣지(Edge)로 구성된다.

상태(State), 노드(Node), 엣지(Edge) 포스팅들을 모두 둘러본 후 이 글을 다시 읽어보는 것을 권장한다.

2. StateGraph 클래스 살펴보기

(1) 생성자

• langgraph.graph 의 StateGraph 클래스를 사용한다.
• StateGraph 는 빌더 클래스이므로, 직접 실행할 수 없고 compile() 메서드로 실행 가능한 그래프를 만들어야 한다.
• 필수 파라미터는 state_schema 즉, “상태 스키마” 이다.
• 그 외로 context_schema, input_schema, output_schema 등의 선택적 파라미터가 있다.

StateGraph(
  self,
  state_schema: type[StateT],
  context_schema: type[ContextT] | None = None,
  *,
  input_schema: type[InputT] | None = None,
  output_schema: type[OutputT] | None = None,
  **kwargs: Unpack[DeprecatedKwargs] = {}
)

(2) 생성자 파라미터

파라미터	타입	설명	필수
state_schema	type[StateT]	상태(State)의 스키마를 정의한 클래스	필수
context_schema	type[ContextT]	런타임 컨텍스트를 정의하는 스키마 클래스 user_id, db_conn 과 같이 실행 내내 고정으로 참조하는 공통 정보
input_schema	type[InputT]	그래프의 가장 앞단 입력 스키마 클래스
output_schema	type[OutputT]	그래프의 최종 출력 스키마 클래스

2. 사용 방법

(1) StateGraph 인스턴스 생성과 상태(State) 할당

• langgraph.graph 의 StateGraph 클래스를 사용해 인스턴스를 생성한다.
• 생성할 때 필수적으로 상태(State)에 대한 스키마 클래스를 파라미터로 넣어줘야 한다.
• 상태(State) 외의 파라미터는 추후 심화 포스팅에서 살펴보기로 한다.

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

class MyState(TypedDict):
    count: Annotated[int, add]
    win_logs: Annotated[list[str], add]
    winner: str

graph = StateGraph(state_schema = MyState)

(2) 노드(Node) 추가

• add_node("노드별칭", 노드함수) 메서드를 이용해 그래프에 노드를 추가한다.

def first_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["철수"]
    }

def second_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["민수"]
    }

def judge_node(state: State) -> State:
    c = Counter(state["win_logs"])
    return {
        "winner": c.most_common()[0][0]
    }

graph.add_node("first", first_round_node)
graph.add_node("second", second_round_node)
graph.add_node("judge", judge_node)

(3) 시작노드 및 종료노드 지정

• set_entry_point 메서드로 시작 노드를 지정하며
• set_finish_point 메서드로 종료 노드를 지정한다.

graph.set_entry_point("first")
graph.set_finish_point("judge")

(4) 엣지(Edge)로 경로 추가

• add_edge 메서드로 노드와 노드간의 연결 경로를 추가한다.

graph.add_edge("first", "second")
graph.add_edge("second", "judge")

(5) 컴파일(Compile)로 실행 가능한 그래프 만들기

• compile 메서드를 이용해 실행 가능한 그래프 인스턴스를 생성한다.

compiled = graph.compile()

(6) 실행

• invoke 메서드를 이용해 그래프를 실행할 수 있다.
• 이 외로도 실행하는 메서드가 존재하며, 이는 추후 심화 포스팅에서 살펴보도록 한다.

result = compiled.invoke({"count":0})

3. 전체 예시 코드

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

class MyState(TypedDict):
    count: Annotated[int, add]
    win_logs: Annotated[list[str], add]
    winner: str

graph = StateGraph(state_schema = MyState)

def first_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["철수"]
    }

def second_round_node(state: State) -> State:
    return {
        "count": 1,
        "win_logs": ["민수"]
    }

def judge_node(state: State) -> State:
    c = Counter(state["win_logs"])
    return {
        "winner": c.most_common()[0][0]
    }

graph.add_node("first", first_round_node)
graph.add_node("second", second_round_node)
graph.add_node("judge", judge_node)

graph.set_entry_point("first")
graph.set_finish_point("judge")

graph.add_edge("first", "second")
graph.add_edge("second", "judge")

compiled = graph.compile()
result = compiled.invoke({"count":0})

# 출력
{'count': 2, 'win_logs': ['철수', '민수'], 'winner': '철수'}

Reference

https://reference.langchain.com/python/langgraph/graph/state/StateGraph