[LangGraph] GraphState

2025.04.03 - [Data & Research] - [Langchain & Langgraph] Table of Contents

---

1. LangGraph의 State

LangChain에 비해서 좀 더 높은 자유도를 갖는 LangGraph에 대해서 알아보려 합니다. LangGraph는 말그대로 그래프 구조를 따라 node에서 작업이 일어나고(Airflow와 비슷한 느낌?) edge를 타고 또 다른 node로 이어지는 구조인데요. 그러면 이렇게 그래프를 따라 움직이는 게 대체 무엇이냐? 하면 "State"라는 것이 계속 처리되면서 그래프 구조를 진행해나갑니다. State라는 녀석의 데이터 타입에 대해서 물어본다면 TypeDict 혹은 Pydantic이라는 타입을 사용합니다. LangGraph에서는 기본적으로 TypeDict로 상태를 정의하는 것을 권장합니다. TypedDict와 Pydantic은 파이썬에서 데이터 구조의 타입을 정의한다는 공통점이 있지만, 작동하는 시점과 제공하는 기능에서 근본적으로 다릅니다. TypedDict가 '설계도'라면, Pydantic은 그 설계도대로 지어졌는지 '검사' 하고, 필요하면 '자재를 변환'까지 해주는 도구입니다.

 

1) TypedDict State의 장/단점

• 장점:
  1. 간결성: 파이썬 표준 라이브러리의 일부이므로 가볍고 추가적인 의존성이 없습니다.
  2. 호환성: 순수한 딕셔너리로 작동하므로 파이썬의 다른 부분과 통합하기 쉽습니다.
• 단점:
  1. 검증 부족: 런타임에 데이터가 들어올 때 (예: LLM의 응답) 해당 데이터가 실제로 정의된 타입과 일치하는지 검사하지 않습니다. 만약 잘못된 타입의 값이 들어오면 런타임 오류로 이어질 수 있습니다.
  2. 기능 제한: 복잡한 제약 조건(예: 문자열 길이, 특정 값 범위)을 지정할 수 없습니다.

 

2) Pydantic State의 장/단점

• 장점:
  1. 강력한 런타임 검증: 외부(LLM 등)에서 입력되는 데이터의 유효성을 확실하게 검증하고, 문제가 있을 경우 사용자 친화적인 오류를 즉시 발생시킵니다.
  2. 자동 변환: LLM이 정수형을 문자열로 반환하더라도 Pydantic이 자동으로 변환하여 State에 안전하게 저장합니다.
  3. 추가 기능: validator, field 등 복잡한 데이터 제약 조건이나 설정 관리를 쉽게 추가할 수 있습니다.
• 단점:
  1. 오버헤드: TypedDict 대비 약간의 라이브러리 의존성과 처리 오버헤드가 있습니다.
  2. 객체 변환: State가 딕셔너리가 아닌 Pydantic 객체로 관리되므로, 때로는 딕셔너리로 변환하는 추가 작업이 필요할 수 있습니다.

 

만약에 아래와 같이 정의된 GraphState가 있다고 해봅시다. 

from typing import TypedDict
from pydantic import BaseModel, ValidationError, Field

# 1. TypedDict를 사용한 상태 정의 (정적 힌트)
class TypedDictState(TypedDict):
    """LangGraph 기본 State 스키마"""
    count: int
    message: str

# 2. Pydantic BaseModel을 사용한 상태 정의 (런타임 검증 및 파싱)
class PydanticState(BaseModel):
    """런타임 검증 및 자동 변환 기능 포함"""
    count: int = Field(ge=0) # count는 0 이상이어야 함
    message: str

 

그런데, 만약 잘못된 타입을 입력해주면 어떻게 될까요? 예: data = {'count': '1', 'message': 'Update'}

• TypedDictState: '1'이 int가 아니라고 경고. 파이썬은 경고를 무시하고 딕셔너리로 저장. 이후 data['count']를 수학 연산에 사용하면 TypeError 발생 가능.
• PydanticState자동 변환 성공: Pydantic이 '1'을 정수 1로 자동 파싱하여 모델 생성. 런타임 오류 없이 안전하게 데이터 저장.

 

여튼 중요한 점은 TypedDict가 됐건 Pydantic이 됐건 LangGraph의 State를 이러한 타입에 담아서 표현하는 것이죠.

(마치 Hamiltoian 같은 물리량을 Matrix의 형태로 표기하는 것에 비유할 수 있을까요?)

 

2. GraphState의 정의

GraphState는 예를들어 아래와 같이 정의할 수 있습니다. 

class User(TypedDict):
    name: str
    age: int

 

User라는 클래스의 TypeDict는 문자열 형식의 name과 정수 형식의 age를 갖는다고 정의했습니다. 좀 더 복잡한 구조인 List나 Dictionary도 물론 포함할 수 있습니다. 

class ChatState(TypedDict):
    messages: list[HumanMessage | AIMessage]
    refer: List[str] # Python 3.9_부터는 굳이 import해야하는 List말고 바로 list로 쓰는것이 좋습니다
    eye: dict[str, Any]
    result: Optional[str]

 

Chatstate라는 state의 messages라는 원소는 리스트 형식이고 messages의 원소는 HumanMessage나 AIMessage의 형식을 가질 수 있습니다.

refer 역시 리스트 형식이고 그 원소는 문자열 형식입니다. 

eye 는 dictionary 형식인데 eye 의 key는 문자열, value는 어떤 형식도 가질 수 있습니다. 

result는 문자열 형식인데 Optional이라서 있어도 되고 없어도(None이어도 되는) 됩니다.

 

그래서 LangGraph를 사용하는 예시를 보면 아래와 같습니다. 세세한 내용은 추후 포스팅에서 이어서 쓰겠습니다. 

# 예시: LangGraph + TypedDict를 활용한 간단한 챗봇 흐름
from typing import Annotated, Any, cast
from typing_extensions import TypedDict, NotRequired
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages  # reducer helper
# (실제 LLM 호출은 여기선 모의로 구현)

# 1) 상태 스키마 정의: 고정 필드, 선택 필드, 누적 메시지(reducer) 포함
class ChatState(TypedDict):
    # messages: add_messages 리듀서를 통해 누적되는 필드
    messages: Annotated[list[dict[str, Any]], add_messages]

    # conversation_id는 항상 존재해야 하는 필수 필드
    conversation_id: str

    # last_intent는 있을 수도, 없을 수도 있음 (NotRequired -> 키 자체가 없어도 OK)
    last_intent: NotRequired[str]

    # optionally store an error object (키는 있어도 None 허용하려면 Optional 사용)
    # 여기서는 키가 있을 수 있고 값은 dict 또는 None 가능
    last_error: NotRequired[dict[str, Any]]

# 2) 노드들 정의 (각 노드는 state를 받아 부분 상태 업데이트 dict를 반환)
def receive_user_node(state: ChatState) -> dict[str, object]:
    # 실제 앱에선 사용자 입력을 받아 메시지로 추가. 여기선 모의.
    user_msg = {"role": "user", "content": "Find me a pizza place nearby"}
    # add_messages 리듀서가 messages에 합쳐질 수 있도록 부분 업데이트만 반환
    return {"messages": [user_msg]}

def classify_intent_node(state: ChatState) -> dict[str, object]:
    # 마지막 메시지 내용을 보고 의도 추출 (단순 키워드 룰)
    last = state["messages"][-1]["content"]
    if "pizza" in last.lower():
        intent = "find_restaurant"
    else:
        intent = "chat"
    return {"last_intent": intent}

def call_llm_node(state: ChatState) -> dict[str, object]:
    # LLM 호출을 모의로 수행하고 AI 메시지를 추가
    intent = state.get("last_intent", "chat")
    if intent == "find_restaurant":
        ai_text = "I found 'Tony's Pizza' 0.5 mi away."
    else:
        ai_text = "Sorry, I didn't get that."
    ai_msg = {"role": "assistant", "content": ai_text, "meta": {"source": "mock-llm"}}
    return {"messages": [ai_msg]}

def finalize_node(state: ChatState) -> dict[str, object]:
    # 그래프 끝내기 전에 간단한 요약 결과를 상태에 둠
    return {"result_summary": f"intent={state.get('last_intent')}"}

# 3) 그래프 구성
g = StateGraph(state_schema=ChatState)
g.add_node("receive_user", receive_user_node)
g.add_node("classify", classify_intent_node)
g.add_node("call_llm", call_llm_node)
g.add_node("finalize", finalize_node)

# edges: START -> receive_user -> classify -> call_llm -> finalize -> END
g.add_edge(START, "receive_user")
g.add_edge("receive_user", "classify")
g.add_edge("classify", "call_llm")
g.add_edge("call_llm", "finalize")
g.add_edge("finalize", END)

compiled = g.compile()

# 4) 초기 상태 (messages는 빈 리스트로 시작)
initial_state: ChatState = {
    "messages": [],
    "conversation_id": "conv-1234",
    # last_intent, last_error는 생략 가능 (NotRequired)
}

# 5) 실행 (실제 환경에선 compile()/invoke 등 호출 방식이 다를 수 있음)
result = compiled.invoke(initial_state)
print("최종 상태:", result)

 

1) State 클래스 정의: 일단 그래프를 통해 계속 이어지며 전달할 State가 어떤 형식인지 먼저 정의합니다.

2) Node 정의: State를 처리할 node를 정의합니다(node는 위에서 정의한 state class를 상속하도록 하면됩니다).

3) 그래프 생성: Graph를 StateGraph 명령어를 통해 생성한 후  node와 edge를 추가해줍니다. 

• Node에서 state를 return할 때 꼭 state의 모든 원소를 입력할 필요는 없습니다. 각 노드는 입력 상태를 직접 변경하지 않고(불변성 권장) 부분 상태 업데이트만 반환합니다(예: {"messages":[new_msg]} 또는 {"last_intent": intent}). LangGraph는 이러한 부분 업데이트들을 병합하여 최종 상태를 만듭니다. 이 방식은 테스트와 디버깅을 쉽게 하고 사이드 이펙트를 줄입니다.
• edge는 출발하는 node와 도착하는 node를 명시해주어야 합니다. START, END노드를 이용해서 처음과 끝을 표기해줍니다. 초기 버전의 LangGraph에는 내부적으로set_entry_point() 같은 메서드가 있었지만 현재 공식 API (2025 기준)에서는 사라지고 대신 add_edge(START, "first_node")를 사용하는 방식으로 통일되었습니다.

4) Graph compile: compile은 그래프를 검증(START - END 사이의 경로가 올바른지, node이름이 중복되진 않았는 등을 검사), 내부 데이터 구조 생성, 병합함수 및 reducer등록, 최적화 및 실행기 build 작업 등을 수행합니다. 그런데 g.invoke와 같이 호출하는 경우에는 내부적으로 자동으로 compile을 수행하기에 생략해도 됩니다. 

5) Input입력: 이렇게 그래프의 구조가 완료됐다면 이제 invoke 등을 통해서 input을 입력하고 output을 받아낼 수 있습니다. 

 

---

 

참고) 추후에 다시 언급할 나름 고급 기능을 짚어보겠습니다. 

messages: Annotated[list[dict], add_messages] — 리듀서(누적)

Annotated[..., add_messages]는 LangGraph의 리듀서 헬퍼를 붙여서, 각 노드가 {"messages": [new_msg]} 같은 부분 업데이트를 반환하면 런타임이 이를 자동으로 누적(append)하게 합니다. 이렇게 하면 노드들이 전체 메시지 리스트를 수동으로 병합할 필요가 없어집니다. 실무적으로 메시지 히스토리처럼 “계속 추가되는” 필드는 리듀서를 쓰면 깔끔합니다.