LangGraph의 Annotated reducer를 사용하여 후보 리스트가 늘어나지 않도록 제약 사항 표현하기

LangGraph의 State, 즉 그래프 전체에서 전달되는 상태에서는 여러 노드가 동일한 필드에 값을 작성할 수 있습니다. 단순한 로그라면 내용을 추가(append)하면 되지만, 후보 리스트를 좁혀나가는(filtering) 경우에는 처리 방식이 달라집니다.

후보를 단계적으로 좁혀나가는 파이프라인에서는, 후속 노드가 이전 단계에 존재하지 않는 후보를 State에 추가해 버리면 필터링의 전제가 무너집니다. 이 글에서는 "기존 후보에 없는 후보를 채택하지 않는다"라는 제약을 Annotated로 지정하는 reducer에 담는 방법을 다룹니다.

Annotated는 Python의 일반적인 실행 구문이 아니라, typing 모듈에 있는 타입 힌트(type hint)용 메커니즘입니다. Annotated[T, metadata]라고 쓰면, "타입으로는 T이지만, 추가적인 메타데이터도 붙어 있다"라고 표현할 수 있습니다. LangGraph에서는 이 메타데이터 부분에 reducer 함수를 작성합니다. 이는 LangGraph 공식 문서의 reducers 절에서도 설명되어 있는 사용법입니다.

추가형(append-type) reducer로 자주 사용되는 operator.add는 Python 표준 라이브러리의 함수이며, LangGraph는 Annotated[..., add]의 메타데이터를 reducer로 읽습니다. 반면, 본 기사에서 말하는 intersection reducer는 LangGraph나 Python 표준 라이브러리에 준비된 이름 있는 기능이 아닙니다. 현재 값과 업데이트 값의 교집합(intersection)만을 남기는 reducer 함수를, 이 글에서는 편의상 그렇게 부릅니다.

Annotated의 기본적인 읽는 법이나 Python 자체가 이 메타데이터를 어떻게 다루는지는, 전제 기사인 "Python의 typing.Annotated 입문 — 타입에 추가 정보를 곁들이기"에서 정리되어 있습니다. 본 기사에서는 그 지식을 전제로 하여, LangGraph의 reducer로 사용하는 상황에 집중합니다.

filtered_pairs: Annotated[PairList | None, intersect_pairs]

이 예시에서 filtered_pairs의 값 자체는 PairList | None입니다. 뒤에 붙어 있는 intersect_pairs는 동일한 필드에 여러 업데이트가 들어왔을 때 LangGraph가 어떻게 합성(composition)할지를 지정하는 함수입니다.

여기서는 이전 단계의 후보 집합에 존재하지 않는 후보를 후속 노드가 추가해 버리는 업데이트를 "확대 쓰기(expansion write)"라고 부릅니다. filtered_pairs에 대한 reducer를 통한 업데이트에서는, 이러한 확대 쓰기를 채택하지 않도록 합니다.

결론

로그와 같이 "늘리는" 필드에는 operator.add가 적합합니다. 반면, 후보를 좁혀나가는 경우에는 업데이트 값을 그대로 채택하는 것이 아니라, 현재 값과의 교집합만을 남기는 reducer를 사용합니다.

from __future__ import annotations
from operator import add
from typing import Annotated, TypedDict
...

filtered_pairs에 대한 reducer를 통한 업데이트에서는 current에 포함되어 있는 후보들만 채택됩니다. 후속 노드가 실수로 새로운 후보를 반환하더라도, 그 후보는 State 상의 filtered_pairs에 남지 않습니다.

LangGraph의 reducer란

LangGraph의 State는 TypedDict로 정의합니다. 각 노드는 State 전체를 반환하는 것이 아니라, 변경하고 싶은 키만 포함하는 partial update를 dict 형태로 반환합니다.

async def filter_node(state: MatchingState) -> dict:
    return {"filtered_pairs": [("c1", "j1")]}

reducer를 지정하지 않은 필드는 기본적으로 새로운 값으로 덮어쓰기(overwrite)되는 업데이트로 취급됩니다. 동일한 필드에 여러 업데이트가 들어올 가능성이 있다면, 의도한 합성 규칙을 reducer로서 명시해 두어야 합니다.

추가하고 싶은 필드에서는 operator.add를 reducer로 지정할 수 있습니다. operator.add

자체는 Python 표준 라이브러리의 함수이며, LangGraph는 State 정의의 Annotated[..., add]를 읽어 해당 함수를 reducer로 사용합니다.

from operator import add
from typing import Annotated
class MatchingState(TypedDict, total=False):
...

이 경우, 각 노드가 반환한 audit_log 리스트는 결합됩니다.

[{"event": "start"}] + [{"event": "filtered"}]
→ [{"event": "start"}, {"event": "filtered"}]

하지만 후보 리스트를 필터링(filtering)할 때는 add를 사용할 수 없습니다. 후보를 늘리고 싶은 것이 아니라, 이전 단계에 존재하는 후보만 남기고 싶기 때문입니다.

필터링 시 발생하는 문제

후보 쌍(candidate pairs)을 단계적으로 필터링하는 과정을 생각해 보겠습니다.

초기 후보
→ 도메인 조건으로 필터링
→ 스코어 조건으로 필터링
...

정직하게 작성한다면, 각 단계마다 별도의 필드를 준비할 수 있습니다.

class MatchingState(TypedDict):
candidate_pairs: list[Pair]
domain_filtered_pairs: list[Pair]
...

이 설계는 명시적이지만, 필터 단계가 늘어날 때마다 필드도 늘어납니다. 하류(downstream) 노드는 "어떤 필드가 직전의 결과인가"를 알아야 하며, 이는 노드의 추가나 교체에 취약해집니다.

그래서 모든 노드가 동일한 filtered_pairs를 읽고, 필터링된 결과를 동일한 filtered_pairs에 반환하는 설계로 만듭니다.

async def domain_filter_node(state: MatchingState) -> dict:
pairs = state.get("filtered_pairs") or []
narrowed = [pair for pair in pairs if passes_domain_check(pair)]
...

이 형태에서는 State의 필드가 늘어나지 않습니다. 하지만 reducer가 없다면, 후속 노드가 실수로 초기 후보를 반환하더라도 그 값이 그대로 채택됩니다. 이것이 확장 쓰기(additive write)입니다.

current = [("c1", "j1")]
update = [("c1", "j1"), ("c_new", "j_new")]
reducer를 지정하지 않을 경우:
...

이 제약 사항은 각 노드의 주석으로 주의를 주는 것보다, State의 reducer로서 표현하는 것이 유지하기 더 쉽습니다.

intersection reducer 구현하기

후보 리스트용 reducer는 현재 값 current와 업데이트 값 update를 받아, 양쪽 모두에 존재하는 후보만을 반환합니다.

Python에는 set.intersection이나 operator.and_가 있습니다. 하지만 이것들을 그대로 사용하면 None을 업데이트 없음으로 취급하거나, update 측의 순서를 유지하거나, PairList | None을 받는 등의 이 글에서 다루는 State 업데이트 규칙을 표현할 수 없습니다. 따라서 여기서는 작은 reducer 함수를 직접 정의합니다.

Pair = tuple[str, str]
PairList = list[Pair]
def intersect_pairs(
...

이 글에서는 None을 "후보 없음"이 아니라 "업데이트 없음"으로 취급합니다. 후보를 0개로 필터링하고 싶다면 None이 아니라 []를 반환합니다. 즉, 이 reducer에서 {"filtered_pairs": None}은 현재 값을 유지하고, {"filtered_pairs": []}는 후보를 0건으로 필터링하는 업데이트가 됩니다.

current is None은 방어적인 분기입니다. LangGraph의 일반적인 실행에서는 첫 번째 쓰기 시에 reducer를 호출하지 않고 그 값을 직접 설정하는 경우가 있습니다. 이 분기는 LangGraph의 동작을 전제로 한 필수 처리라기보다, reducer를 일반적인 함수처럼 다루기 쉽게 만들기 위한 분기입니다.

이 reducer는 집합(set) 관점에서는 "후보를 늘리지 않는" 방향으로 작동하지만, 리스트의 순서는 update 측에 의존합니다. 병렬 브랜치(parallel branch)에서 여러 update가 들어오는 경우, 후보 집합은 좁혀지더라도 최종 리스트의 순서는 적용 순서에 영향을 받을 가능성이 있습니다. 순서까지 의미를 갖는다면, 마지막에 안정 정렬(stable sort)을 수행하거나 스코어(score)가 포함된 별도의 필드로 순서를 관리해야 합니다.

JSON을 경유하면 tuple이 list로 반환되는 경우가 있습니다. 그럴 때는 Pair를 정규화하는 작은 함수를 준비하여, 비교 키를 (str(pair[0]), str(pair[1]))와 같이 통일합니다. 이 기사에서는 reducer의 핵심 목적을 보여주기 위해 입력 타입을 tuple[str, str]로 한정했습니다.

State에 reducer를 지정하기

MatchingState에서는 추가하고 싶은 audit_log에는 add를 사용하고, 필터링하고 싶은 filtered_pairs에는 intersect_pairs를 지정합니다.

from __future__ import annotations
from operator import add
from typing import Annotated, TypedDict
...

total=False로 설정해 두면, 타입 상으로도 각 키를 생략 가능하도록 취급할 수 있습니다. LangGraph의 노드(node)는 변경하고 싶은 키만 포함하는 부분 업데이트(partial update)를 반환하므로, 이 State 정의와 궁합이 좋습니다. 노드의 반환값에 filtered_pairs가 포함되지 않으면 해당 필드는 업데이트되지 않습니다.

명시적으로 {"filtered_pairs": None}을 반환하는 경우에는 reducer에 update=None이 전달될 가능성이 있습니다. 키 생략과 None의 명시적 반환은 내부적으로는 다르게 처리되지만, 이 기사에서는 둘 다 "변경 없음"으로 취급하도록 설계했습니다. 후보를 비우고 싶다면 []를 반환합니다.

최소 샘플

후보 쌍(candidate pair)을 초기화하고 2단계로 필터링하는 최소 예제입니다.

from langgraph.graph import END, START, StateGraph
async def initialize_node(state: MatchingState) -> dict:
    pairs = [
        ...
    ]

이 예제에서 filtered_pairs는 단계적으로 줄어들기만 합니다. 각 노드는 동일한 필드에 쓰기를 수행하지만, State 측의 reducer가 "이전 단계에 없는 후보는 채택하지 않는다"라는 제약 사항을 가지고 있습니다.

reducer 단체 테스트하기

reducer는 일반적인 함수이므로 LangGraph의 그래프를 구성하지 않고도 테스트할 수 있습니다. 최소한 다음 4가지 케이스를 확인합니다.

def test_reducer_first_write():
    pairs = [("c1", "j1"), ("c2", "j2")]
    assert intersect_pairs(None, pairs) == pairs
    ...

실제 운영 환경에서는 빈 리스트, JSON 경유 시의 list화, 순서 유지 등도 필요에 따라 확인해야 합니다. 다만 reducer의 핵심 사양은 최초 쓰기, update=None, 일반적인 필터링, 확장 쓰기(expansion write)의 제외 여부로 확인할 수 있습니다.

요약

LangGraph의 Annotated[T, reducer]를 사용하면, 동일한 필드에 여러 노드가 쓸 때의 합성(composition) 방법을 State 측에 정의할 수 있습니다.

로그와 같이 늘리고 싶은 필드에는 operator.add를 사용할 수 있습니다. 반면, 후보 쌍의 필터링에서는 이전 단계에 존재하지 않는 후보를 채택하지 않는 교집합 reducer(intersection reducer)를 사용합니다. filtered_pairs에 대한 업데이트가 이 reducer를 통하는 한, 후속 노드가 반환한 새로운 후보는 State 상에서 채택되지 않습니다.

reducer는 부작용(side effect)이 없는 함수로 구현합니다. 병렬 실행 시 동일한 필드에 여러 update가 들어오는 경우, 후보 집합뿐만 아니라 최종 순서에 대한 의존성도 확인하십시오. 우선 reducer를 단체 테스트하여, State에 작성한 제약 사항이 구현 변경으로 인해 무너지지 않도록 관리하면 다루기 훨씬 수월해집니다.

참고 정보

Discussion