[LangGraph] Reducer

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

---

LangGraph의 첫번째 포스팅에서 언급했던 내용을 다시 짚어볼 차례입니다. LangGraph의 StateGraph는 순수 함수들의 연결 그래프이고, Reducer는 “여러 노드가 같은 상태 필드를 수정할 때, 그 값들을 어떻게 병합(merge)할지 정의하는 규칙”입니다. 

 

1. Reducer의 역할

새 상태(new state)가 반환될 때, 이전 상태(previous state)와 새 상태(new state)를 받아 병합된 최종 상태를 반환하는 함수를 말합니다. 개념적으로 보면 아래와 같죠.

merged_state[key] = reducer(previous_state[key], new_state[key])

 

이 방식 덕분에 LangGraph는 각 필드별로 업데이트 정책(reduction rule) 을 다르게 지정할 수 있습니다.

 

2. LangGraph Reducer의 종류

Reducer는 default로는 덮어쓰기로 되어 있습니다만 append한다든지 누적합을 구한다든지 하는 방식, 혹은 더 나아가서 사용자 정의 방식으로 변경해 줄 수도 있습니다. 

Reducer	기능
add_messages	메시지 리스트를 자동으로 append (대화 히스토리 유지용)
add	수치형 값을 누적 (합산)
append	일반 리스트를 단순히 뒤에 붙임
replace (기본값)	새 값으로 덮어씀
사용자 정의 리듀서	lambda old, new: ... 형태로 직접 정의 가능

 

Default 방식이 아닌 방식으로 Reducer를 변경하고자 한다면 Annotated를 통해 상태 필드의 타입과 reducer 규칙을 정의해야합니다. 

from typing import Annotated
from langgraph.graph import StateGraph, add_messages

class MyState(TypedDict):
    messages: Annotated[list[dict], add_messages]

 

 

한 번 예시를 통해서 여러 방식의 reducer를 알아볼까요?

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END, add, append, add_messages

class MyState(TypedDict):
    count: Annotated[int, add]                     # 숫자 누적
    logs: Annotated[list[str], append]             # 리스트 이어붙이기
    messages: Annotated[list[dict], add_messages]  # LangChain message 누적
    value: int                                     # 기본값: replace

def add_one(state):
    return {"count": 1, "logs": ["node1"], "messages": [{"role": "assistant", "content": "hi"}], "value": 1}

def add_two(state):
    return {"count": 2, "logs": ["node2"], "messages": [{"role": "assistant", "content": "hello"}], "value": 999}

graph = StateGraph(MyState)
graph.add_node("n1", add_one)
graph.add_node("n2", add_two)
graph.set_entry_point("n1")
graph.add_edge("n1", "n2")
graph.add_edge("n2", END)
app = graph.compile()

final_state = app.invoke({"count": 0, "logs": [], "messages": [], "value": 0})

print(final_state)

 

{
  'count': 3,                           # add → 1 + 2
  'logs': ['node1', 'node2'],           # append → 리스트 이어붙이기
  'messages': [ {..."hi"...}, {..."hello"...} ],  # add_messages → message 누적
  'value': 999                          # replace (기본) → 마지막 값으로 덮어씀
}

 

최근 N개의 message만 유지하는 사용자 정의 reducer는 아래와 같이 만들어 볼 수 있습니다. 

def keep_last_n(old, new, n=3):
    return (old + new)[-n:]

class MyState(TypedDict):
    messages: Annotated[list[dict], lambda o, n: keep_last_n(o, n, 3)]