Multi-Agent · 14

LangGraph State 키: 덮어쓰기 vs. 쌓기 (Reducer)

모든 State key가 같은 방식으로 업데이트된다고 생각하기 쉽다.

근거 · 교안 p41-p44

1장: 루나: 모든 State 키가 똑같이 업데이트될까, 솔라?

솔라는 노트북 화면에 떠 있는 코드 블록을 보며 의자 등받이에 몸을 깊게 기댔다. 방금 전까지 읽던 LangGraph 공식 문서의 한 구절이 머릿속을 맴돌았다. ‘State의 기본 업데이트 동작은 덮어쓰기(Overwrite)입니다.’

너무나 당연하고 명쾌한 설명이었다. 변수에 새로운 값을 할당하면 이전 값은 사라지고 새 값만 남는다. 프로그래밍의 기본 원칙과 다를 바 없었다. 솔라는 고개를 끄덕이며 옆에서 조용히 책을 읽고 있던 언니 루나에게 말을 걸었다.

“언니, LangGraph에서 상태가 업데이트되는 방식, 생각보다 간단한 거였어. 그냥 마지막에 실행된 노드가 새로운 값을 던져주면 그걸로 기존 값을 덮어쓰는 거잖아. 늘 하던 변수 할당처럼 말이야.”

솔라는 확신에 차 있었다. 하나의 그래프 안에서 상태(State)는 계속해서 최신 정보로 교체되며 흘러가는 것, 그게 전부라고 생각했다.

루나는 책에서 눈을 떼고 솔라가 보고 있던 화면으로 시선을 옮겼다. 화면에는 에이전트의 상태를 정의하는 AgentState 클래스가 보였다. 여러 개의 필드(키)가 정의되어 있었다.

from typing import List, Optional
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    # 채팅 기록을 저장하는 필드
    messages: List[BaseMessage]

    # 에이전트의 다음 행동 계획을 저장하는 필드
    plan: Optional[str]

    # ... 기타 다른 필드들

루나는 다른 말을 하는 대신, 화면을 향해 손가락을 뻗었다. 그리고 messages: List[BaseMessage] 라고 적힌 줄을 가볍게 톡, 쳤다.

“정말 모든 키가 그럴까, 솔라? messages 필드도 덮어쓰는 게 맞을까?”

“어…?”

솔라는 순간 말문이 막혔다. ‘모든 키가?’ 루나의 질문은 단순했지만, 솔라가 세운 명쾌한 가정의 한가운데에 작은 돌멩이를 던졌다. 솔라는 다시 코드로 시선을 돌려, 에이전트가 작동하는 과정을 머릿속으로 그려보기 시작했다.

‘처음에 사용자가 “오늘 날씨 어때?” 하고 물으면, 그 질문이 HumanMessage 객체가 되어서 messages 리스트에 담기겠지. 그럼 상태는 messages: [HumanMessage(content="오늘 날씨 어때?")] 가 될 거야.’

여기까지는 문제가 없었다.

‘그다음, AI 모델이 답변을 생성하는 노드를 거치면 “오늘은 맑습니다” 같은 답변이 AIMessage 객체로 만들어져. 그리고 이 노드가 상태를 업데이트하겠지. 만약… 여기서 덮어쓰기가 일어난다면?’

솔라의 미간이 살짝 찌푸려졌다.

‘덮어쓴다면, messages는 messages: [AIMessage(content="오늘은 맑습니다")] 로 바뀌어 버려. 사용자의 원래 질문이었던 HumanMessage는 사라지고. 그 상태에서 사용자가 “그럼 내일은?” 하고 다시 물으면, AI는 무엇에 대한 ‘내일’을 묻는지 알 수 없게 돼. 대화의 맥락이 통째로 날아가 버리니까.’

“아…!”

솔라는 나지막이 탄식하며 무릎을 쳤다.

“안되네. messages 필드를 덮어쓰면 대화가 이어지질 않잖아. AI가 방금 자기가 한 말만 기억하는 셈이 되니까. 이건 대화가 아니라 그냥… 단발적인 외침의 나열일 뿐이네.”

스스로 내린 결론이었지만, 충격은 작지 않았다. 모든 것을 한 가지 규칙으로 설명할 수 있다는 깔끔한 모델이 깨지는 순간이었다.

루나는 이번엔 바로 아래에 있는 plan: Optional[str] 라인으로 손가락을 옮겼다.

솔라는 루나의 의도를 금세 알아차렸다. messages 필드를 통해 얻은 깨달음을 plan 필드에 적용해 보라는 뜻이었다. 솔라는 다시 생각에 잠겼다.

“아하. 그런데 plan 필드는 덮어쓰는 게 맞을 것 같아. 에이전트가 ‘A 도구를 사용해야지’라고 계획을 세웠다가, 새로운 정보를 얻고 ‘아니야, B 도구를 쓰는 게 더 좋겠어’라고 계획을 수정한다면, 이전 계획인 A는 더 이상 필요 없잖아. 가장 최신의, 가장 유효한 계획 하나만 있으면 되니까. 이건 덮어쓰는 게 맞아.”

이제 솔라는 명확히 알 수 있었다. 어떤 데이터는 대화 기록처럼 차곡차곡 쌓여야만 의미가 있고, 어떤 데이터는 최신 계획처럼 계속해서 교체되어야만 유용했다. 상태를 구성하는 모든 키가 똑같은 방식으로 업데이트될 것이라는 처음의 생각은 완전히 틀렸다.

바로 그 지점에서, 해결된 의문은 더 큰 질문을 낳았다.

“그렇구나. 데이터의 목적에 따라 업데이트 방식이 달라야 하네. 그런데 언니, LangGraph의 기본 동작이 덮어쓰기라면… messages처럼 기록을 쌓아야 하는 건 대체 어떻게 처리하라고 알려주지? 코드에 뭔가 특별한 약속이라도 해야 하는 건가?”

2장: 루나: 덮어쓰기와 쌓기, AgentState에서 선언하기

솔라의 질문이 채 끝나기도 전에, 루나는 조용히 솔라의 노트북을 자기 쪽으로 돌려 키보드에 손을 얹었다. 솔라가 방금까지 골똘히 들여다보던 AgentState 클래스였다. 루나는 몇 글자를 더 타이핑했다. 코드가 미세하게 바뀌었다.

루나는 다시 노트북 화면을 솔라 쪽으로 돌려주었다. 화면에는 익숙하면서도 낯선 코드가 떠 있었다.

from typing import List, Optional
from typing_extensions import TypedDict, Annotated
from langchain_core.messages import BaseMessage

# 'add_messages'는 두 메시지 리스트를 합치는 간단한 함수
def add_messages(left: list, right: list):
    return left + right

class AgentState(TypedDict):
    # 채팅 기록: Annotated를 사용해 업데이트 방식을 지정
    messages: Annotated[List[BaseMessage], add_messages]

    # 에이전트의 계획: 특별한 지정 없이 기본 덮어쓰기 방식을 사용
    plan: Optional[str]

이전과 무엇이 달라졌는지 한눈에 들어왔다. messages 필드의 타입 힌트가 복잡하게 변해 있었다. List[BaseMessage]가 Annotated라는 것으로 감싸여 있었고, 그 안에는 add_messages라는 이름이 쉼표 뒤에 붙어 있었다. 반면 plan 필드는 이전과 똑같았다.

솔라는 미간을 찌푸렸다. “이게 언니가 말하려는 답이야?”

“응. 네 질문에 대한 답이 이 코드 안에 있어.”

“Annotated…? 주석을 달았다는 뜻인가? 그리고 add_messages는 위에 정의된 함수 이름이고. 이게 어떻게 messages가 덮어쓰기 되지 않고 쌓이게 만든다는 거지? plan에는 아무것도 안 붙어 있고….”

솔라는 혼란스러웠다. Annotated는 파이썬 타입 힌트에서 본 적이 있는 기능이었지만, LangGraph의 상태 업데이트와 무슨 관련이 있는지 전혀 감을 잡을 수 없었다. 마치 아는 단어들의 조합으로 만들어졌지만 전혀 해석할 수 없는 외계 문장 같았다.

솔라는 add_messages 함수를 먼저 살펴봤다. 함수 자체는 지극히 단순했다. 왼쪽 리스트와 오른쪽 리스트를 받아서 두 리스트를 합친(+) 새로운 리스트를 반환하는 것. 그게 전부였다.

“이 함수는 그냥 리스트 두 개를 합치는 거잖아. 이게 뭐 어쨌다는…”

말을 내뱉던 솔라는 순간 멈칫했다. ‘리스트 두 개를 합친다.’ 머릿속에서 뭔가 번쩍했다.

‘잠깐. 노드가 실행될 때마다 상태가 업데이트되지. 만약… left가 그래프의 현재 messages 리스트이고, right가 이번 노드에서 새로 생성된 메시지 리스트라면? 이 함수는 기존 기록에 새로운 기록을 그냥 더해주는 거네! 덮어쓰는 게 아니라!’

생각이 꼬리를 물고 이어졌다.

“아하! 그럼 Annotated는 LangGraph에게 주는 특별 지시 같은 건가? ‘이 messages 필드를 업데이트할 때는, 그냥 새 값으로 덮어쓰지 말고, add_messages라는 함수를 사용해서 기존 값과 새 값을 합쳐줘!’ 라고 알려주는 약속인 거야?”

루나는 말없이 고개를 살짝 끄덕였다. 정답이라는 신호였다.

솔라는 다시 코드를 보았다. 이제 암호문 같던 코드가 명확한 설계도로 보이기 시작했다.

“그러니까… LangGraph는 상태를 업데이트할 때 각 키를 확인하는 거구나. plan 키를 봤는데, 아무런 특별한 지시(Annotated)가 없어. 그럼 ‘아, 이건 기본 규칙대로 처리하면 되는구나’ 하고 그냥 최신 값으로 덮어쓰는 거고. 반면에 messages 키를 봤더니 Annotated와 함께 add_messages라는 ‘병합 규칙’이 붙어있어. 그럼 ‘아, 이건 기본 규칙이 아니네. 지정된 add_messages 함수를 호출해서 값을 합쳐야겠다’ 라고 판단하는 거지.”

LangGraph에서는 이렇게 값을 합치는 규칙, 즉 기존 상태와 새로운 상태 조각을 어떻게 결합할지 정의하는 함수를 리듀서(Reducer) 라고 부른다. 솔라는 이제 그 의미를 온전히 이해할 수 있었다. 리듀서는 단순히 데이터를 ‘줄이는’ 것이 아니라, 이전 값과 새 값을 ‘어떻게 하나의 값으로 만들지’ 결정하는 규칙이었던 것이다. add_messages 함수는 messages 필드를 위한 리듀서였다.

“와… 대박이다. 그러니까 상태의 각 필드가 어떻게 행동할지를 타입 힌트 레벨에서 선언하는 거였네. 기본값은 ‘덮어쓰기’고, 다른 행동이 필요하면 Annotated로 ‘쌓기’ 같은 커스텀 규칙을 지정해주고.”

솔라는 자신이 처음 가졌던 ‘모든 키는 똑같이 작동할 것’이라는 생각이 얼마나 단순했는지 깨달았다. LangGraph는 훨씬 더 유연하고 정교한 시스템이었다. 키의 정체성을 정의하는 단계에서부터 업데이트 동작까지 설계할 수 있도록 만든 것이다.

명쾌한 해답을 얻었지만, 솔라의 마음 한구석에는 작은 의심이 남았다. 코드를 읽고 머릿속으로 시뮬레이션하는 것과, 실제로 그렇게 돌아가는 것을 눈으로 보는 것은 다른 차원의 문제였다.

“알겠어, 언니. 코드 상으로는 이제 완벽하게 이해했어. plan은 덮어쓰기 되고, messages는 add_messages 함수 때문에 계속 쌓일 거라는 거. 그런데… 진짜로 그래프를 실행했을 때도 그렇게 될까? 노드를 여러 개 통과하고 난 최종 상태가 정말 우리가 설계한 대로 남아있는지, 결과를 직접 보고 싶어.”

3장: 루나: 노드 실행으로 덮어쓰기와 쌓기 확인하기

솔라의 말이 끝나자, 루나는 기다렸다는 듯 노트북 화면에 새로운 코드 파일을 열었다. 방금 전까지 보았던 AgentState 선언부와는 다른, 더 구체적인 실행 코드였다. 화면에는 두 개의 간단한 함수, node_a와 node_b가 정의되어 있었고, 그 아래에는 이 노드들을 연결한 작은 그래프가 만들어져 있었다. 마치 솔라의 의심을 증명하기 위해 미리 준비해 둔 실험실 같았다.

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import StateGraph

# ... 이전에 정의한 AgentState와 add_messages 함수 ...

# 노드 A: 메시지를 추가하고 plan을 업데이트한다.
def node_a(state: AgentState):
    return {
        "messages": [AIMessage(content="A 노드 실행 완료")],
        "plan": "A 노드 실행 계획"
    }

# 노드 B: 메시지를 추가하고 plan을 업데이트한다.
def node_b(state: AgentState):
    return {
        "messages": [AIMessage(content="B 노드 실행 완료")],
        "plan": "B 노드 실행 계획"
    }

# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("node_a", node_a)
workflow.add_node("node_b", node_b)
workflow.set_entry_point("node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", "__end__")
graph = workflow.compile()

코드를 훑어본 솔라는 금세 구조를 파악했다. 그래프는 node_a에서 시작해서 node_b를 거쳐 끝나는 단순한 일직선 구조였다. 그리고 두 노드는 각각의 AgentState를 반환하는데, 하나는 messages 키에 자신의 실행 기록을, plan 키에 새로운 계획을 담아 돌려준다. 솔라가 머릿속으로만 그렸던 상황을 직접 실행해 볼 수 있는 완벽한 실험 환경이었다.

루나는 코드 아래에 마지막으로 실행할 한 줄을 보여주었다.

# 실행 시작
initial_input = {"messages": [HumanMessage(content="시작!")], "plan": "초기 계획 없음"}
result = graph.invoke(initial_input)
print(result)

“자, 솔라. 이 코드를 실행하면 result에 최종 상태가 담길 거야. print(result)를 했을 때, messages와 plan 필드가 각각 어떤 값을 가지고 있을지, 네가 한번 예측해 볼래?”

드디어 직접 확인할 시간이 왔다. 솔라는 심호흡을 한번 하고 화면에 집중했다. 이것은 더 이상 추상적인 개념에 대한 토론이 아니었다. 코드의 각 줄이 상태(State)라는 공유된 캔버스 위에 어떤 그림을 그릴지, 한 단계씩 따라가는 구체적인 추적이었다.

솔라는 손가락으로 화면을 짚어가며 생각의 흐름을 소리 내어 말했다.

“좋아. 먼저, initial_input으로 시작. 이때 상태는 messages 리스트에 HumanMessage 하나, plan은 문자열 ‘초기 계획 없음’이야.”

솔라는 머릿속에 첫 번째 상태 스냅샷을 그렸다.

“그다음, 그래프는 node_a를 실행하겠지. node_a는 AIMessage("A 노드 실행 완료")와 plan("A 노드 실행 계획")을 반환해. 여기서 업데이트가 일어나.”

솔라는 잠시 멈춰, 이전 장에서 루나와 나누었던 대화를 떠올렸다. 키마다 다른 업데이트 규칙.

“plan 키는 특별한 지정이 없으니까 기본 동작, 즉 덮어쓰기가 일어나겠네. 그럼 ‘초기 계획 없음’은 사라지고 ‘A 노드 실행 계획’으로 바뀔 거야. 하지만 messages 키는 다르지. 우리는 AgentState를 정의할 때 add_messages라는 리듀서(Reducer) 함수를 지정해 줬어. 그러니까 기존 messages 리스트에 node_a가 반환한 새 메시지가 뒤에 쌓이게 될 거야. 그럼 node_a 실행 후의 상태는 messages에 메시지 두 개, plan은 ‘A 노드 실행 계획’이겠네.”

이제 마지막 단계가 남았다. 그래프는 node_a 다음으로 node_b를 실행한다.

“마지막으로 node_b가 실행돼. node_b도 똑같이 AIMessage("B 노드 실행 완료")와 plan("B 노드 실행 계획")을 반환해. 아까랑 같은 규칙이 적용되겠지. plan은 또 덮어쓰기가 되어서 ‘A 노드 실행 계획’은 사라지고 ‘B 노드 실행 계획’만 남을 거고, messages에는 node_b의 메시지가 하나 더 추가될 거야.”

솔라는 예측을 마쳤다. 확신에 찬 목소리로 최종 결과를 말했다.

“최종 result는 이거야. plan 필드의 값은 ‘B 노드 실행 계획’이고, messages 필드는 HumanMessage, AIMessage from A, AIMessage from B, 이렇게 총 세 개의 메시지가 순서대로 들어있는 리스트일 거야!”

루나는 솔라의 예측을 조용히 듣고만 있다가, 말이 끝나자 엔터 키를 눌러 코드를 실행했다. 잠시 후, 화면에 graph.invoke의 최종 결과가 출력되었다.

{'messages': [HumanMessage(content='시작!'), AIMessage(content='A 노드 실행 완료'), AIMessage(content='B 노드 실행 완료')], 'plan': 'B 노드 실행 계획'}

솔라의 얼굴에 환한 미소가 번졌다. 출력된 결과는 그녀의 예측과 토씨 하나 다르지 않았다. messages는 시작점의 HumanMessage부터 node_a와 node_b의 AIMessage까지, 모든 대화 기록을 차곡차곡 쌓아두고 있었다. 반면 plan은 node_a가 만들었던 계획을 미련 없이 버리고, 가장 마지막에 실행된 node_b의 계획으로 완전히 대체되어 있었다.

눈으로 직접 확인한 결과는 강력했다. ‘덮어쓰기’와 ‘쌓기’는 더 이상 코드 위에 적힌 막연한 선언이 아니었다. 그래프가 살아 움직이며 상태를 변화시키는, 명백하고 실질적인 동작 방식이었다.

솔라는 다시 AgentState 정의 코드를 바라보았다.

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages] # 쌓기
    plan: Optional[str]                                  # 덮어쓰기

이제 이 코드는 솔라에게 단순한 데이터 구조 정의가 아니었다. 에이전트의 기억이 어떻게 형성될지를 결정하는 설계도였다. 어떤 기억은 시간이 흘러도 사라지지 않고 축적되어야 하고(messages), 어떤 기억은 항상 최신의 것으로 유지되어야 한다(plan). LangGraph는 바로 그 기억의 메커니즘을 개발자가 직접 설계할 수 있도록 열어준 것이었다.

“이제 알겠어.”

솔라는 작은 노트를 펼쳐 새로운 에이전트를 구상하며 중얼거렸다.

“만약 내가 파일 시스템을 다루는 에이전트를 만든다면… AgentState에 ‘처리한 파일 목록’이란 키를 둘 거야. 이건 messages처럼 Annotated를 써서 계속 쌓아야겠지. 하지만 ‘현재 작업 중인 파일 경로’라는 키는? 그건 plan처럼 그냥 덮어쓰는 게 맞아. 작업 대상이 바뀔 때마다 이전 경로는 필요 없으니까.”

솔라는 노트에 processed_files: Annotated[List[str], ...]와 current_file: str이라고 적었다. 더 이상 어떤 업데이트 방식을 선택해야 할지 망설이지 않았다. 데이터의 목적을 생각하면 답은 명확했다. 쌓을 것인가, 덮어쓸 것인가. 솔라는 이제 스스로 그 답을 찾고 설계할 수 있게 되었다.