[AI] LangChain - LangGraph ( State, Message )

LangGraph는 상태 기반 LLM을 만들기 위한 프레임워크입니다. 쉽게 말해 LLM의 현재 상태에 따라 조건문 반복문 등을 사용하여 유연한 LLM 워크플로우를 만들 수 있는 기능으로 StateGraph, MessageGraph, ReAct 등이 있습니다. 

 

구성요소

구성요소	설명
노드(Node)	실행 단위. LLM 호출, 함수 실행 등.
엣지(Edge)	다음에 어떤 노드로 갈지 정의. 조건 분기, 루프 가능.
상태(State)	전체 흐름에서 공유되는 데이터. 상태를 기준으로 흐름 결정 가능.

 

• 상태(State) - 모든 노드가 공유하는 데이터

상태는 전체 그래프 에서 사용하는 공용 데이터로 그래프가 처리하는 데이터 구조를 정의하는 객체입니다. 일반적으로 StateGraph에서는 기존 상태를 ovrride(덮어쓰기)합니다. 모든 노드는 이 상태를 읽거나 수정할 수 있으며 Dict형태를 갖습니다.

class ChatState(TypedDict):
    history: List[str]
    user_input: None

• 노드(Node)  - 하나의 작업 단위

노드는 LangGraph에서 실제 작업을 수행하는 함수 입니다..

def get_user_input(state: ChatState) -> ChatState:
    user_input = input("사용자: ")
    print(state['history'])
    return {"user_input": user_input, "history": state['history'] + [user_input]}

def get_user_exit(state: ChatState) -> ChatState:
    return {"user_input": "exit"}

 

• 엣지(Edge) - 노드 간의 흐름을 정의

엣지는 노드 간의 연결을 정의합니다. 

무조건 다음 노드로 이동 (A → B)

from langgraph.graph import StateGraph, START, END

# 그래프 빌더 생성
builder = StateGraph(ChatState)

# 노드 추가
builder.add_node("get_user_input", get_user_input)
builder.add_node("get_user_exit", get_user_input2)

# 엣지 추가
builder.add_edge(START, "get_user_input")
builder.add_edge("get_user_input", "get_user_exit")
builder.add_edge("get_user_exit", END)

# 그래프 컴파일
graph = builder.compile()

from IPython.display import Image, display

# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))

• 조건부 엣지

조건부 엣지는 노드 간의 연결을 정의하며 상태에 따라 분기합니다. 

상태 값을 보고 분기 (A → B1 또는 A → B2)

 

아래 소스 코드는 StateGraph로 일반적인 입력을 하면 계속 입력을 수행하고 chat history를 출력합니다. 그리고 exit를 입력하면 프로그램이 종료됩니다.

from typing import Literal
from typing import List
from IPython.display import Image, display
from typing import TypedDict

def decide_next_step(state: ChatState) -> Literal["get_user_input", "get_user_exit"]:
    if state['user_input'] == "exit":
        return "get_user_exit"
    else:
        return "get_user_input"  

from langgraph.graph import StateGraph, START, END

# 그래프 빌더 생성
builder = StateGraph(ChatState)

# 노드 추가
builder.add_node("get_user_input", get_user_input)
builder.add_node("get_user_exit", get_user_exit)
builder.add_node("decide_next_step", decide_next_step) 

# 엣지 추가
builder.add_edge(START, "get_user_input")
# 조건부 엣지 추가
builder.add_conditional_edges(
    "decide_next_step",
    decide_next_step,
    {
        "get_user_input": "get_user_input",
        "get_user_exit": "get_user_exit"
    }
)
builder.add_edge("get_user_input", END)
builder.add_edge("get_user_exit", END)

# 그래프 컴파일
graph = builder.compile()

# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))

initial_state = {'user_input': '', 'history': []}
new_state = graph.invoke(initial_state)
while True:
    new_state = graph.invoke(new_state)
    print(new_state['history'])
    if new_state["user_input"] == "exit":
        print("대화를 종료합니다. 감사합니다!")
        break

ADD Reducer

위에 조건부 엣지의 소스코드  상태기반 그래프로 구현되어 있다. 이때 상태 기반 그래프는 상태를 덮어쓰는 특징이 있다. 그래서 history를 출력하기 위해 강제적으로 아래소스와 같이 기존 history와 입력받은 값을 강제로 합치는 로직을 넣어놨다.

def get_user_input(state: ChatState) -> ChatState:
	...
    return {"user_input": user_input, "history": state['history'] + [user_input]}

이런 상태를 덮어 쓰는 특징은 데이터를 추적하고자 할 때 곤란할 수 있다. 이때 Reducer이란 키워드를 사용하면 현재 상태에 데이터를 덮어쓰는 게 아니라 추가할 수 있다. Annotated를 사용하여 LangGraph에서는 이 기능을 활용해서 상태 필드에 대한 자동 업데이트 방식을 지정한다.

from typing import Annotated, TypedDict
from operator import add

class ChatState(TypedDict):
    history: Annotated[List[str], add]
    user_input: str

Custom Reducer

상태 업데이트가 기본적인 덮어쓰기나 병합만으로 해결되지 않을 때 유용한 방법으로 예를 들어 중복 제거, 최대/최소 값 유지 조건에 따른 병합등 특정 로직이 필요한 경우에 적용할 수 있다.

 

아래 소스코드는 history에 대한 중복을 제거 해서 추가하는 코드이다.

from typing import Annotated, TypedDict

def reduce_unique_history(left: list | None, right: list | None) -> list:
    if not left:
        left = []
    if not right:
        right = []
    # 중복 제거: set을 사용하여 중복된 문서를 제거하고 다시 list로 변환
    return list(set(left + right))

class ChatState(TypedDict):
    history: Annotated[List[str], reduce_unique_history]  # Custom Reducer 적용
    user_input: str

MessageGraph

기존 LangGraph의 StateGraph는 딕셔너리 기반 상태(StateDict)를 중심으로 동작했다. 하지만 LLM 애플리케이션은 대부분 메시지의 흐름에 기반한 구조를 가지고 있으며, MessageGraph는 이러한 메시지 중심 구조에 최적화된 프레임워크이다.

MessageGraph는 LangChain의 ChatModel에 특화된 형태의 StateGraph로, HumanMessage, AIMessage 등 다양한 메시지 타입을 지원한다. 그래프 상태에는 대화 기록을 나타내는 메시지 목록(messages)이 저장되며, 이 목록을 상태 키로 추가하면 자동으로 operator.add 리듀서가 적용되어 메시지가 누적된다.

또한 키에 커스텀 리듀서 함수를 지정함으로써, 메시지의 병합 방식이나 업데이트 방식을 유연하게 제어할 수 있다. 이를 통해 각 노드에서 메시지를 추가하거나 수정할 때 상태 업데이트를 효율적으로 관리할 수 있다.

 

메시지 저장 구조

• 상태 딕셔너리에 messages 키가 자동 생성됨.
• 메시지 객체(HumanMessage, AIMessage, 등)가 리스트 형태로 저장됨.
• 메시지는 LangChain 내부에서 LLM 호출에 사용되는 표준 포맷.

자동 리듀서 (Reducer)

• messages 키에는 자동으로 operator.add가 적용됨.
• 즉, 새 메시지를 반환하면 기존 메시지 목록에 자동으로 누적됨.

메시지 제어 : add_messages 방법

• messages 필드 정의 해야함
• 내부적으로는 operator.add를 통해 messages 리스트가 합쳐짐.

from typing import Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

# 기본 State 초기화 방법을 사용
class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

 

메시지 제어 :  MessagesState 방법

• messages 필드 정하지 않아도 사용가능
• 추가적인 필드 아래documents 처럼 추가

# LangGraph MessagesState라는 미리 만들어진 상태를 사용
from langgraph.graph import MessagesState
from typing import List
from langchain_core.documents import Document

class GraphState(MessagesState):
    # messages 키는 기본 제공 - 다른 키를 추가하고 싶을 경우 아래 주석과 같이 적용 가능 
    documents: List[Document]