[LangGraph] Reducer

📌 LangGraph - Reducer

1️⃣ Reducer (리듀서)

• State Reducer는 LangGraph의 상태 관리 핵심 메커니즘

• 각 노드의 출력을 전체 그래프 상태에 통합하는 방식을 정의

• Reducer의 필요성:

  • 상태 덮어쓰기 문제: 기본적으로 각 노드의 반환값은 해당 상태 키의 이전 값을 덮어쓰는 방식으로 동작 (override)
  • 누적 업데이트 필요: 특히 메시지 리스트와 같은 경우, 이전 상태에 새로운 값을 추가하고 싶을 때

---

1. Reducer를 별도로 지정하지 않은 경우

• 기본 Reducer는 이전 값을 자동으로 덮어쓰는 방식으로 작동

• Reducer 설정이 없는 경우 자동으로 기본값이 적용

• 이는 단순한 상태 업데이트에는 적합하나 데이터 누적이 필요한 경우 부적절

• 기본 Reducer는 간단한 상태 관리에 적합하지만 복잡한 데이터 처리에는 한계가 있음

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

# 상태 정의 
class DocumentState(TypedDict):
    query: str
    documents: List[str]

# Node 1: query 업데이트
def node_1(state: DocumentState) -> DocumentState:
    print("---Node 1 (query update)---")
    query = state["query"]
    return {"query": query}

# Node 2: 검색된 문서 추가 
def node_2(state: DocumentState) -> DocumentState:
    print("---Node 2 (add documents)---")
    return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}

# Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: DocumentState) -> DocumentState:
    print("---Node 3 (add more documents)---")
    return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}


# 그래프 빌드
builder = StateGraph(DocumentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

# 그래프 실행
graph = builder.compile()

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

- 출력

---

그래프 실행

# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요.", "documents": None}

# 그래프 실행 
final_state = graph.invoke(initial_state)

# 최종 상태 출력
print("-"*100)
print("최종 상태:")
print("쿼리:", final_state['query'])
print("검색된 문서:", final_state['documents'])

- 출력

---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
----------------------------------------------------------------------------------------------------
최종 상태:
쿼리: 채식주의자를 위한 비건 음식을 추천해주세요.
검색된 문서: ['doc2.pdf', 'doc4.pdf', 'doc5.pdf']

---

2. Reducer를 별도로 지정하는 경우

• Annotated를 통해 사용자 정의 Reducer를 지정할 수 있음

• operator.add를 사용하면 리스트 형태의 데이터를 누적 관리할 수 있음

• 여기서는 기존 리스트에 새로운 메시지를 추가하는 방식으로 작동

from operator import add
from typing import Annotated, TypedDict

class ReducerState(TypedDict):
    query: str
    documents: Annotated[List[str], add]

# Node 1: query 업데이트
def node_1(state: ReducerState) -> ReducerState:
    print("---Node 1 (query update)---")
    query = state["query"]
    return {"query": query}

# Node 2: 검색된 문서 추가 
def node_2(state: ReducerState) -> ReducerState:
    print("---Node 2 (add documents)---")
    return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}

# Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: ReducerState) -> ReducerState:
    print("---Node 3 (add more documents)---")
    return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}

# 그래프 빌드
builder = StateGraph(ReducerState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

# 그래프 실행
graph = builder.compile()

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

- 출력

---

그래프 실행

# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요.", "documents": []}

# 그래프 실행 
final_state = graph.invoke(initial_state)

# 최종 상태 출력
print("-"*100)
print("최종 상태:")
print("쿼리:", final_state['query'])
print("검색된 문서:", final_state['documents'])

- 출력

• 리스트에 데이터가 누적된 모습

---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
----------------------------------------------------------------------------------------------------
최종 상태:
쿼리: 채식주의자를 위한 비건 음식을 추천해주세요.
검색된 문서: ['doc1.pdf', 'doc2.pdf', 'doc3.pdf', 'doc2.pdf', 'doc4.pdf', 'doc5.pdf']

---

3. Custom Reducer 사용

• Custom Reducer는 복잡한 상태 관리가 필요할 때 사용

• 중복 제거나 최대/최소값 유지와 같은 특수한 로직을 구현할 수 있음

• 비즈니스 요구사항에 맞는 맞춤형 상태 관리가 가능

• 상황에 따라 조건부 병합과 같은 고급 기능을 구현할 수 있음

from typing import TypedDict, List, Annotated

# Custom reducer: 중복된 문서를 제거하며 리스트 병합
def reduce_unique_documents(left: list | None, right: list | None) -> list:
    """Combine two lists of documents, removing duplicates."""
    if not left:
        left = []
    if not right:
        right = []
    # 중복 제거: set을 사용하여 중복된 문서를 제거하고 다시 list로 변환
    return list(set(left + right))

# 상태 정의 (documents 필드 포함)
class CustomReducerState(TypedDict):
    query: str
    documents: Annotated[List[str], reduce_unique_documents]  # Custom Reducer 적용

# Node 1: query 업데이트
def node_1(state: CustomReducerState) -> CustomReducerState:
    print("---Node 1 (query update)---")
    query = state["query"]
    return {"query": query}

# Node 2: 검색된 문서 추가 
def node_2(state: CustomReducerState) -> CustomReducerState:
    print("---Node 2 (add documents)---")
    return {"documents": ["doc1.pdf", "doc2.pdf", "doc3.pdf"]}

# Node 3: 추가적인 문서 검색 결과 추가
def node_3(state: CustomReducerState) -> CustomReducerState:
    print("---Node 3 (add more documents)---")
    return {"documents": ["doc2.pdf", "doc4.pdf", "doc5.pdf"]}

# 그래프 빌드
builder = StateGraph(CustomReducerState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 논리 구성
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)

# 그래프 실행
graph = builder.compile()

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

- 출력

---

그래프 실행

# 초기 상태
initial_state = {"query": "채식주의자를 위한 비건 음식을 추천해주세요.", "documents": []}

# 그래프 실행 
final_state = graph.invoke(initial_state)

# 최종 상태 출력
print("-"*100)
print("최종 상태:")
print("쿼리:", final_state['query'])
print("검색된 문서:", final_state['documents'])

- 출력

---Node 1 (query update)---
---Node 2 (add documents)---
---Node 3 (add more documents)---
----------------------------------------------------------------------------------------------------
최종 상태:
쿼리: 채식주의자를 위한 비건 음식을 추천해주세요.
검색된 문서: ['doc4.pdf', 'doc1.pdf', 'doc2.pdf', 'doc5.pdf', 'doc3.pdf']