LangGraph 기반 AI 채팅의 상태 관리: Branching과 Custom Reducer의 역할

LangGraph를 활용하여 AI 챗봇이나 대화형 애플리케이션을 만들 때, 상태(State) 관리는 가장 중요한 요소 중 하나입니다.
✅ 어떻게 이전 메시지를 기억하고, 새 메시지를 추가할 것인가?
✅ 병렬 실행 중 충돌을 방지하고 안정적인 대화 흐름을 유지하는 방법은?

LangGraph는 기본적으로 상태를 관리하는 여러 가지 방법을 제공합니다.
하지만, 기본 동작(Default Overwriting State)을 이해하지 않으면, 메시지가 예상치 못하게 사라지거나 덮어씌워질 수도 있습니다.

이 글에서는 LangGraph에서 상태가 어떻게 업데이트되는지를 살펴보고,
Branching(분기), Custom Reducer, 그리고 add_messages Reducer를 활용해 AI 챗봇의 대화 상태를 안전하게 유지하는 방법을 설명하겠습니다. 🚀

Default overwriting state

LangGraph에서 기본적인 상태(State) 업데이트 방식은 "덮어쓰기(Overwriting)" 입니다.

 

✅ 실행 흐름

1. graph.invoke({"foo": 2})를 실행하면, foo의 초기 값은 2입니다.
2. node_1(state)가 실행되면서 foo = foo + 1이 됩니다.
3. 새로운 상태 {"foo": 3}이 반환되며 기존 상태가 덮어씌워집니다.

from typing_extensions import TypedDict
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

def node_1(state):
    print("---Node 1---")
    return {"foo": state['foo'] + 1}

 

graph.invoke({"foo" : 2})

• Node 1 → END
• 상태는 순차적으로 업데이트됨 → {'foo': 3}이 정상적으로 출력됨.

🔹 이 경우 문제 없음!

 

Branching

여러 개의 노드(Node)가 실행될 때, 특히 병렬 실행(Branching)이 발생하면 상태 관리 방식이 중요해집니다.

아래와 같은 노드가 있다라고 할때 

def node_1(state):
    print("---Node 1---")
    return {"foo": state['foo'] + 1}

def node_2(state):
    print("---Node 2---")
    return {"foo": state['foo'] + 1}

def node_3(state):
    print("---Node 3---")
    return {"foo": state['foo'] + 1}

 

from langgraph.errors import InvalidUpdateError
try:
    graph.invoke({"foo" : 1})
except InvalidUpdateError as e:
    print(f"InvalidUpdateError occurred: {e}")

✅ 실행 결과

---Node 1---
---Node 2---
---Node 3---
InvalidUpdateError occurred: At key 'foo': Can receive only 
one value per step. Use an Annotated key to handle multiple 
values.

🛑 왜 오류가 발생하는가?

1. Node 1 실행 후 foo = 2
2. Node 1에서 Node 2와 Node 3으로 동시에 분기 (병렬 실행)
3. Node 2와 Node 3이 동시에 실행되면서 각각 foo 값을 1 증가
   • Node 2의 결과 → {'foo': 3}
   • Node 3의 결과 → {'foo': 3}
4. LangGraph 입장에서 상태를 하나만 유지해야 하는데, 두 개의 서로 다른 결과가 발생!
5. 어떤 값을 선택할지 명확하지 않기 때문에 InvalidUpdateError가 발생!

 

Reducers (리듀서)

Reducers는 이 문제를 해결할 수 있는 일반적인 방법을 제공합니다.

Reducers는 상태(state)를 어떻게 업데이트할지 지정하는 역할을 합니다.

우리는 Annotated 타입을 사용하여 Reducer 함수를 지정할 수 있습니다.

예를 들어, 이 경우 각 노드에서 반환된 값을 덮어쓰지 않고 리스트에 추가(append)하고 싶습니다.

이를 위해 Reducer 함수로 operator.add를 사용할 수 있습니다.

✅ operator.add는 Python의 내장 operator 모듈에 있는 함수

// 기존 
class State(TypedDict):
    foo: int

def node_1(state):
    print("---Node 1---")
    return {"foo": state['foo'] + 1}

// 수정
class State(TypedDict):
    foo: Annotated[list[int], add]

def node_1(state):
    print("---Node 1---")
    return {"foo": [state['foo'][0] + 1]}

 

수정된 node로 실행

graph.invoke({"foo" : [1]})

Reducer 함수로 operator.add를 해서 결과가 [1,2]로 나옴

---Node 1---
{'foo': [1, 2]}

 

🛑 다시 위에 예제를 operator.add로 해보자

 

✅ 실행 흐름

📌 값과 노드 매칭

1. Start → foo = [1] (초기값)
2. Node 1 → foo = [2]
3. Node 2 (병렬 실행) → foo = [3]
4. Node 3 (병렬 실행) → foo = [3]
5. End → 최종적으로 foo = [1, 2, 3, 3]

✅ Node 2와 Node 3이 같은 값(3)을 생성했기 때문에 리스트에 중복으로 포함됨!

🔹 왜 3이 두 개 있는가?

• Node 2와 Node 3이 동시에 실행되었기 때문.
• 둘 다 Node 1의 결과 2를 기반으로 동작했으므로, 동일한 값 3을 생성함.

 

🔍 Custom Reducers (사용자 정의 Reducer)란?

LangGraph에서 기본적으로 operator.add 같은 Reducer를 사용할 수 있지만, 입력이 None일 경우 문제가 발생할 수 있습니다.

➡ 이를 해결하기 위해 None을 안전하게 처리하는 Custom Reducer를 정의할 수 있습니다.

1️⃣ 기존 문제: operator.add는 None을 처리하지 못함

graph.invoke({"foo": None})

🔴 기존 방식 (operator.add 사용)

• None + [new_value] → TypeError 발생!
• operator.add는 리스트를 병합하는 함수이지만, None이 입력되면 동작하지 않음.

2️⃣ 해결 방법: Custom Reducer 정의

우리는 리스트를 병합하면서도 None이 입력될 경우 자동으로 빈 리스트([])로 처리하는 Reducer를 만들 수 있습니다.

✅ reduce_list 함수 (Custom Reducer)

python
복사편집
def reduce_list(left: list | None, right: list | None) -> list:
    """두 리스트를 안전하게 병합하는 함수 (None도 처리 가능)

    Args:
        left (list | None): 첫 번째 리스트 (None일 수도 있음)
        right (list | None): 두 번째 리스트 (None일 수도 있음)

    Returns:
        list: 두 리스트를 병합한 새 리스트 (None이면 빈 리스트로 변환)
    """
    if not left:
        left = []  # None이면 빈 리스트로 변경
    if not right:
        right = []  # None이면 빈 리스트로 변경
    return left + right  # 리스트 병합

✅ None을 자동으로 빈 리스트로 변환하여 오류 방지!

3️⃣ 두 가지 상태 정의

LangGraph에서 사용할 두 가지 상태(State)를 정의할 수 있습니다.

class DefaultState(TypedDict):
    foo: Annotated[list[int], add]  # 기존 방식 (operator.add 사용)

class CustomReducerState(TypedDict):
    foo: Annotated[list[int], reduce_list]  # 새로운 방식 (Custom Reducer 사용)

🔄 기존 방식과 Custom Reducer 방식 비교

입력 값 기존 방식 (operator.add) Custom Reducer (reduce_list)

{"foo": [1]}	✅ 정상 실행 ([1, 2])	✅ 정상 실행 ([1, 2])
{"foo": None}	❌ TypeError 발생	✅ 정상 실행 ([2])

✅ 이제 None이 입력되어도 안전하게 LangGraph가 실행

🔍 Reducer 에서 MessagesState 활용

1️⃣ Reducer의 기본 개념 복습

Reducer는 여러 개의 상태 업데이트를 하나로 합치는 역할을 합니다.

기존에 우리가 다룬 예제에서는:

• 숫자를 합치는 max, sum
• 리스트를 병합하는 operator.add, reduce_list

➡ 하지만 LangGraph에서는 대화를 저장하는 messages 상태도 관리해야 합니다.

➡ 이때, 기존 메시지에 새로운 메시지를 추가하는 역할을 하는 Reducer가 add_messages입니다.

📌 MessagesState의 특징

1. 기본적으로 messages라는 키를 가짐
2. messages는 add_messages Reducer를 사용하여 관리됨
3. 즉, 메시지를 누적하면서 저장하는 역할을 함

📌 add_messages Reducer의 동작 방식

LangGraph에서 대화를 유지하려면, 기존 메시지에 새로운 메시지를 추가하는 것이 중요합니다.

이 역할을 하는 것이 바로 add_messages Reducer

1. Reducer는 단순 숫자 합산뿐만 아니라, "대화 기록" 같은 데이터도 관리할 수 있다.
2. LangGraph에서는 AI 대화를 지속적으로 저장할 필요가 있다.
3. 이를 위해 메시지를 저장하는 messages 상태가 필요하고, 이를 편리하게 제공하는 MessagesState가 있다.
4. MessagesState는 messages 키를 기본적으로 포함하고, add_messages Reducer를 내장하고 있다.
5. 즉, MessagesState를 알려면 Reducer 동작 원리를 알아두는게 좋다

🔍 add_messages Reducer의 추가 기능: 메시지 덮어쓰기 & 삭제

• add_messages를 사용하면 같은 id의 메시지를 덮어씌울 수 있음! (Re-writing)
• RemoveMessage를 사용하면 특정 메시지를 쉽게 삭제할 수 있음! (Removal)

 

✅ 우리가 살펴본 주요 개념들

1. Default Overwriting State
   • 기본적으로 LangGraph는 새로운 상태를 반환하면 기존 상태를 덮어씌움(Overwrite)
   • 단일 실행에서는 문제가 없지만, 여러 경로(Branching)에서 상태를 업데이트할 경우 충돌 가능성이 있음.
2. Branching과 상태 충돌
   • 동일한 상태에서 두 개 이상의 노드가 병렬 실행될 경우, 각 노드가 서로 다른 값을 반환하면 충돌이 발생
   • LangGraph는 이를 명확하게 정의하지 않으면 InvalidUpdateError를 발생시킴.
3. Custom Reducer를 활용한 상태 병합
   • 상태가 덮어씌워지는 것이 아니라, 사용자 정의 로직을 통해 상태를 병합할 수 있도록 설계 가능
   • 예를 들어, max, sum, reduce_list 등의 Custom Reducer를 사용하면 상태를 더 유연하게 관리 가능.
4. add_messages Reducer를 활용한 채팅 메시지 관리
   • add_messages Reducer를 사용하면, 기존 메시지 리스트에 새로운 메시지를 추가할 수 있음.
   • 메시지 ID가 같다면 자동으로 덮어쓰며, RemoveMessage를 통해 특정 메시지를 삭제할 수도 있음.
   • 이를 활용하면 LangGraph 기반 AI 챗봇이 대화 기록을 유지하면서도 동적으로 업데이트 가능.

---

🚀 결론 및 다음 단계

LangGraph에서 상태를 관리하는 것은 단순한 값 업데이트 이상의 의미를 가집니다.
특히, Branching이 발생하는 환경에서 올바른 상태 병합 전략을 설정하지 않으면, 데이터 충돌이 발생할 수 있습니다.

📌 앞으로의 활용 방향

✅ LangGraph를 활용해 AI 챗봇을 구축할 경우, 메시지 상태를 안전하게 유지하는 방법을 고민해야 합니다.
✅ Custom Reducer를 설계하여 병렬 실행 중에도 안정적으로 상태를 업데이트할 수 있도록 해야 합니다.
✅ add_messages Reducer를 활용하면 채팅 기록을 효율적으로 관리할 수 있습니다.