상태 그래프 내에서는 모든 상태가 공유된다고 했다. 그럼 굳이 상태를 반환할 필요가 있나?
이전에 살펴본 것 처럼 각 노드의 반환값은 기존 상태에 병합(merge)된다. 이걸 가능하게 하는 langgraph의 컴포넌트를 리듀서(reducer)라고 한다. 기본 리듀서는 노드가 반환한 새로운 값을 기존 상태에 병합하는 일을 한다.
물론 리듀서에도 종류가 있다. 메시지 리스트 등에서는 이전 상태에 새로운 값을 추가해야하는 경우도 있다. 이는 add 리듀서가 수행한다.
중복제거나 정렬 등 특수한 상태관리를 하도록 사용자 정의 리듀서를 만들 수도 있다.
정리하면 리듀서는 merge, add, 그리고 custom 리듀서 세가지가 있다.
리듀서는 언제 지정해서 사용하는가? 상태를 정의할때 지정한다. 기본적으로 아무것도 지정하지 않으면 기본 리듀서가 사용되고, 파이썬 내장 tpying모듈인 Annotated 를 이용해 지정하면 해당 리듀서가 사용된다.
class ReducerState(TypedDict):
query: str
documents: **Annotated[List[str], add]**
위에서는 add리듀서를 사용하겠다고 설정. operator.add 를 이용해 리듀서를 변경할 수 있다. 이제 documents는 str타입의 배열에 상태값들이 계속 add된다.
중복 제거, 최대/최소 값 유지, 조건부 병합 등의 특정 비즈니스 로직이 필요한 경우 리듀서를 만들어서 사용할 수 있다.
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()))
여기서 의문인건, 커스텀 리듀서를 구현할때 입출력이 추상화되어있지 않냐는 것이다. 그냥
def연산자만 사용하면 다 되는건가?# 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))각 리듀서는 State내의 각 키마다 별도로 적용할 수 있다. 이 별도 키와 입출력 타입만 일치하면 문제는 없는 듯. 나는 추상화 수준에서 안전한 코드를 작성하고 싶어서 찾아본건데 그런 장치는 없는거 같고, 그냥 컴파일 오류로 잡아내야 하나 보다.