에이전트 메모리의 아키텍처: LangGraph가 실제로 작동하는 방식
Source: Dev.to
LangGraph의 상태 이해하기
State는 LangGraph 에이전트의 실행 메모리입니다. 그래프가 순회되는 동안 모든 입력, 중간 사고, 도구 출력, 그리고 결정을 기록합니다. 각 노드는 현재 상태를 받아 로직을 수행하고 오직 업데이트하고 싶은 상태 부분만 반환합니다. LangGraph는 제공한 스키마에 따라 그 업데이트를 기존 상태와 병합합니다. 이 결정론적 병합은 상태 정의 방식에 따라 데이터를 추가, 덮어쓰기 또는 결합할 수 있습니다.
TypedDict 로 간단한 상태 정의
from typing_extensions import TypedDict
class State(TypedDict):
messages: list
extra_field: int
TypedDict는 그래프를 흐르는 메모리의 형태를 정의합니다. 대화 기록 외에도 상태는 도구 출력, 메타데이터, 작업 상태, 카운터, 검색된 문서 등 다양한 정보를 담을 수 있습니다.
왜 TypedDict가 Pydantic이나 Dataclass보다 선호되는가
- 부분 업데이트: 노드는 변경하고 싶은 키만 포함한 딕셔너리를 반환할 수 있습니다. LangGraph는 이러한 부분 업데이트를 자동으로 병합합니다.
- 성능: 단일 필드만 업데이트해도 전체 객체를 재구성할 필요가 없어, 큰 히스토리에서도 비용이 적습니다.
- 불변성 보장: LangGraph는 결정론적 병합에 의존합니다; 전체 Pydantic 혹은 dataclass 인스턴스를 변형하면 이러한 보장이 깨질 수 있습니다.
- 캐시 문제: Pydantic 내부 메타데이터가 인스턴스마다 달라질 수 있어 예기치 않은 캐시 동작을 일으킬 수 있습니다.
이러한 트레이드오프 때문에 대부분의 프로덕션 LangGraph 그래프는 상태 스키마에 TypedDict를 사용합니다.
Reducer: 업데이트 병합 방식 제어
Reducer는 기존 값과 새로운 값을 어떻게 결합할지 정의합니다. 시그니처는 다음과 같습니다:
def reducer(current_value, new_value) -> merged_value:
...
- 기본 동작: Reducer가 지정되지 않으면 새로운 값이 기존 값을 덮어씁니다.
- 커스텀 동작: 리스트에 추가하거나, 카운터를 누적하거나, 딕셔너리를 병합하는 등 원하는 로직을 구현할 수 있습니다.
Python의 operator.add 사용
from typing import Annotated
from operator import add
from typing_extensions import TypedDict
class State(TypedDict):
history: Annotated[list[str], add]
count: int
add reducer는 리스트를 연결하므로 history에 대한 여러 업데이트가 덮어쓰기 대신 추가됩니다.
LangChain 메시지를 위한 내장 add_messages Reducer
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
add_messages는 LangChain HumanMessage, AIMessage, ToolMessage 객체를 지능적으로 병합하고, ID 기반 중복 제거와 스트리밍 혹은 도구 호출 시 올바른 순서를 보장합니다.
실용 예시
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
class MyState(TypedDict):
logs: Annotated[list[str], add]
counter: Annotated[int, add]
노드 구현
def start_node(state: MyState):
return {"logs": ["Started"], "counter": 1}
def step_node(state: MyState):
return {"logs": ["Step done"], "counter": 2}
def finish_node(state: MyState):
return {"logs": ["Finished"], "counter": 3}
실행 흐름
start_node이후 –logs = ["Started"],counter = 1step_node이후 –logs = ["Started", "Step done"],counter = 3finish_node이후 –logs = ["Started", "Step done", "Finished"],counter = 6
Reducer 덕분에 로그가 추가되고 카운터가 누적됩니다; Reducer가 없으면 각 업데이트가 이전 상태를 덮어쓰게 됩니다.
커스텀 Reducer 작성하기
LangGraph는 reducer 시그니처를 가진 어떤 호출 가능한 객체도 허용합니다. 커스텀 reducer는 다음과 같은 작업을 할 수 있습니다:
- 리스트가 너무 커지면 오래된 대화 컨텍스트를 잘라내기.
- 기존 키를 보존하면서 딕셔너리 병합하기.
- 중복되지 않은 항목만 누적하기.
- 애플리케이션에 특화된 복잡한 변환 로직 구현하기.
def unique_items(current: list, new: list) -> list:
return list(dict.fromkeys(current + new))
class CustomState(TypedDict):
items: Annotated[list, unique_items]
위 예시에서 unique_items는 각 병합 후 items 리스트에 중복이 없도록 보장합니다.