LangGraph State와 Reducer 완벽 가이드

LangGraph에서 상태를 정의하고 관리하는 방법을 배웁니다. TypedDict부터 Pydantic, 그리고 다양한 Reducer 패턴까지, 실무에서 바로 사용할 수 있는 상태 관리 기법을 소개합니다.

---

목차

1. TypedDict로 상태 스키마 정의
2. Pydantic BaseModel 사용
3. 기본 Reducer: 값 덮어쓰기
4. 커스텀 Reducer: Annotated 사용
5. operator.add 메시지 누적
6. MessagesState 사전 구축 상태

---

1. TypedDict로 상태 스키마 정의

어느 날 김개발 씨가 LangGraph로 첫 AI 에이전트를 만들기 시작했습니다. 코드를 작성하던 중 문득 궁금해졌습니다.

"이 에이전트가 기억해야 할 정보들을 어떻게 정의하지?" 선배 박시니어 씨가 다가와 말했습니다. "상태 스키마부터 정의해야죠."

TypedDict는 LangGraph에서 상태의 구조를 명확하게 정의하는 가장 기본적인 방법입니다. 마치 건물을 짓기 전에 설계도를 그리는 것처럼, 에이전트가 어떤 정보를 담을지 미리 설계합니다.

이를 통해 타입 안정성을 확보하고 코드의 가독성을 높일 수 있습니다.

다음 코드를 살펴봅시다.

from typing import TypedDict
from langgraph.graph import StateGraph

# 상태 스키마 정의
class AgentState(TypedDict):
    messages: list[str]  # 대화 메시지 목록
    user_name: str       # 사용자 이름
    counter: int         # 처리 횟수
    is_complete: bool    # 완료 여부

# 상태를 사용하는 그래프 생성
graph = StateGraph(AgentState)

# 초기 상태 예시
initial_state: AgentState = {
    "messages": [],
    "user_name": "김개발",
    "counter": 0,
    "is_complete": False
}

김개발 씨는 입사 3개월 차 주니어 개발자입니다. 최근 팀에서 LangGraph를 도입하면서 AI 에이전트 개발을 맡게 되었습니다.

첫 번째 미션은 고객 문의를 처리하는 챗봇을 만드는 것이었습니다. 코드를 작성하던 중, 김개발 씨는 고민에 빠졌습니다.

챗봇이 대화 내용, 사용자 정보, 진행 상황 등 여러 정보를 기억해야 하는데, 이것들을 어떻게 관리해야 할지 막막했습니다. 선배 박시니어 씨가 모니터를 들여다보며 물었습니다.

"상태 스키마는 정의했어요?" 상태 스키마란 무엇일까요? 쉽게 비유하자면, 상태 스키마는 마치 서랍장의 칸막이와 같습니다.

옷장을 정리할 때 양말 칸, 속옷 칸, 셔츠 칸을 나누듯이, 에이전트가 기억해야 할 정보들을 체계적으로 분류하는 것입니다. 각 칸에는 어떤 종류의 물건이 들어갈지 미리 정해져 있습니다.

TypedDict는 바로 이런 칸막이를 만드는 도구입니다. Python의 딕셔너리처럼 키-값 쌍으로 데이터를 저장하지만, 각 키에 어떤 타입의 값이 들어가야 하는지 명확하게 정의할 수 있습니다.

상태 정의가 없던 시절에는 어땠을까요? 초기 LangGraph 사용자들은 그냥 일반 딕셔너리를 사용했습니다.

"messages"에 리스트가 들어갈지, 문자열이 들어갈지 코드를 보기 전까지는 알 수 없었습니다. 실수로 잘못된 타입의 데이터를 넣으면 런타임 에러가 발생했습니다.

더 큰 문제는 협업할 때였습니다. 다른 개발자가 작성한 코드를 볼 때마다 "이 키에는 뭐가 들어가지?"라고 물어봐야 했습니다.

프로젝트가 커질수록 상태 구조를 파악하기가 점점 어려워졌습니다. 바로 이런 문제를 해결하기 위해 TypedDict를 사용합니다.

TypedDict를 사용하면 상태의 구조가 코드 레벨에서 명확하게 드러납니다. IDE는 자동 완성을 제공하고, 타입 체커는 잘못된 타입 사용을 미리 경고해줍니다.

무엇보다 코드를 읽는 사람이 한눈에 상태의 구조를 이해할 수 있다는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 3번째 줄부터 8번째 줄을 보면 AgentState라는 TypedDict 클래스를 정의하는 것을 알 수 있습니다. 이 부분이 핵심입니다.

messages는 문자열 리스트, user_name은 문자열, counter는 정수, is_complete는 불리언 타입으로 정의되어 있습니다. 다음으로 11번째 줄에서는 이 상태 스키마를 사용하여 StateGraph를 생성합니다.

이제 그래프는 AgentState의 구조를 따르는 상태만 허용합니다. 마지막으로 14번째 줄부터는 실제 초기 상태 객체를 생성합니다.

TypedDict에 정의된 모든 필드를 채워넣은 것을 확인할 수 있습니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 전자상거래 고객 지원 챗봇을 개발한다고 가정해봅시다. 상태에는 고객 ID, 주문 번호, 문의 유형, 대화 히스토리, 해결 여부 등이 필요합니다.

TypedDict로 이런 구조를 명확하게 정의하면, 팀원 누구나 챗봇의 상태를 쉽게 이해하고 수정할 수 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 너무 많은 필드를 상태에 추가하는 것입니다. 상태는 에이전트가 진정으로 기억해야 할 정보만 담아야 합니다.

임시 변수나 계산 가능한 값까지 상태에 넣으면 관리가 복잡해집니다. 따라서 꼭 필요한 정보만 상태로 정의해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 고개를 끄덕였습니다.

"아, 그래서 상태 스키마를 먼저 설계하는 거군요!" TypedDict를 제대로 이해하면 더 체계적이고 유지보수하기 쉬운 LangGraph 애플리케이션을 만들 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - 상태 필드명은 명확하고 직관적으로 작성하세요. "data"보다는 "user_messages"처럼 구체적인 이름이 좋습니다.

• 선택적 필드가 필요하다면 typing.NotRequired를 사용하여 표시할 수 있습니다.
• 상태 스키마는 프로젝트 초기에 팀과 함께 설계하면 나중에 혼란을 줄일 수 있습니다.

---

2. Pydantic BaseModel 사용

김개발 씨가 TypedDict로 상태를 정의하고 나니 박시니어 씨가 다시 찾아왔습니다. "잘 작성했네요.

그런데 입력 값 검증은 어떻게 할 건가요?" 김개발 씨는 당황했습니다. 검증이라니, 그런 건 생각도 못 했습니다.

Pydantic BaseModel은 TypedDict보다 강력한 상태 정의 방법입니다. 마치 공항 보안 검색대처럼, 상태에 들어오는 모든 데이터를 자동으로 검증하고 변환합니다.

잘못된 데이터가 들어오면 즉시 에러를 발생시켜 버그를 사전에 방지할 수 있습니다.

다음 코드를 살펴봅시다.

from pydantic import BaseModel, Field, validator
from langgraph.graph import StateGraph

class AgentState(BaseModel):
    messages: list[str] = Field(default_factory=list)
    user_name: str = Field(..., min_length=1, max_length=50)
    counter: int = Field(default=0, ge=0)  # 0 이상의 정수
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')

    @validator('user_name')
    def validate_name(cls, v):
        if v.strip() != v:
            raise ValueError('이름 앞뒤 공백 제거 필요')
        return v.strip()

# 그래프 생성 시 model_config 사용
graph = StateGraph(AgentState)

김개발 씨의 챗봇이 어느 정도 완성되어 갔습니다. 그런데 테스트 중 이상한 버그를 발견했습니다.

어떤 사용자가 이름 필드에 빈 문자열을 입력했더니 챗봇이 이상하게 동작했습니다. counter 필드에 음수가 들어간 경우도 있었습니다.

박시니어 씨가 로그를 보더니 말했습니다. "입력 검증이 안 되어 있네요.

Pydantic을 사용해보는 건 어때요?" Pydantic이란 무엇일까요? 쉽게 비유하자면, Pydantic은 마치 나이트클럽의 입구 경비원과 같습니다.

입장하려는 사람의 신분증을 확인하고, 복장 규정을 체크하고, 연령 제한을 검사합니다. 조건에 맞지 않으면 입장을 거부하죠.

Pydantic도 마찬가지로 상태에 들어오는 모든 데이터를 엄격하게 검증합니다. TypedDict는 타입 힌트만 제공할 뿐 실제로 검증하지는 않습니다.

반면 Pydantic BaseModel은 런타임에 실제로 데이터를 검사하고, 필요하면 자동으로 변환까지 해줍니다. 검증 기능이 없던 시절에는 어땠을까요?

개발자들은 if문을 수도 없이 작성해야 했습니다. "이름이 비어있나요?", "카운터가 음수인가요?", "이메일 형식이 맞나요?" 같은 검사를 일일이 손으로 작성했습니다.

코드가 길어지고, 놓치는 검증도 생겼습니다. 더 큰 문제는 일관성이었습니다.

개발자마다 검증 로직을 다르게 작성하면서, 같은 필드인데도 어디서는 검증하고 어디서는 안 하는 일이 벌어졌습니다. 버그는 언제나 검증을 빼먹은 곳에서 터졌습니다.

바로 이런 문제를 해결하기 위해 Pydantic BaseModel을 사용합니다. Pydantic을 사용하면 검증 로직이 모델 정의 안에 선언적으로 들어갑니다.

Field 함수로 최솟값, 최댓값, 정규표현식 패턴 등을 지정할 수 있습니다. validator 데코레이터로 커스텀 검증 로직도 추가할 수 있습니다.

무엇보다 검증이 자동으로 실행되기 때문에 개발자가 실수로 빼먹을 일이 없습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 5번째 줄을 보면 messages 필드가 빈 리스트를 기본값으로 갖도록 설정되어 있습니다. Field의 default_factory를 사용하면 매번 새로운 리스트 인스턴스가 생성됩니다.

6번째 줄의 user_name은 필수 필드이며, 1자 이상 50자 이하여야 합니다. 점 세 개는 필수 필드를 의미합니다.

7번째 줄의 counter는 기본값이 0이고, ge=0은 "greater than or equal"의 약자로 0 이상의 값만 허용합니다. 8번째 줄의 email은 정규표현식으로 이메일 형식을 검증합니다.

패턴에 맞지 않으면 ValidationError가 발생합니다. 11번째 줄부터는 커스텀 validator입니다.

user_name 필드가 설정될 때 자동으로 실행되어 앞뒤 공백을 제거합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 금융 서비스 챗봇을 개발한다고 가정해봅시다. 사용자로부터 입금액, 계좌번호, 비밀번호 등 민감한 정보를 받습니다.

Pydantic을 사용하면 입금액이 양수인지, 계좌번호가 올바른 형식인지, 비밀번호가 충분히 복잡한지 자동으로 검증할 수 있습니다. 잘못된 데이터가 시스템 깊숙이 들어가기 전에 차단하는 것입니다.

많은 스타트업과 대기업에서 FastAPI와 함께 Pydantic을 사용하는 이유가 바로 이 강력한 검증 기능 때문입니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 과도한 검증을 추가하는 것입니다. 모든 필드에 복잡한 validator를 달면 코드가 느려지고 복잡해집니다.

정말 필요한 검증만 추가하고, 나머지는 애플리케이션 로직에서 처리하는 것이 좋습니다. 따라서 비즈니스 규칙과 데이터 무결성 검증을 적절히 분리해야 합니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. Pydantic으로 상태를 다시 정의한 김개발 씨는 테스트를 돌려봤습니다.

이번에는 잘못된 데이터가 들어오자마자 명확한 에러 메시지가 출력되었습니다. "오, 이거 편한데요!" Pydantic BaseModel을 제대로 이해하면 더 안전하고 신뢰할 수 있는 애플리케이션을 만들 수 있습니다.

여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - Field의 description 파라미터로 각 필드의 용도를 문서화할 수 있습니다.

• Config 클래스를 사용하면 extra='forbid'로 정의되지 않은 필드를 거부할 수 있습니다.
• Pydantic V2에서는 성능이 크게 개선되었으니 최신 버전을 사용하세요.

---

3. 기본 Reducer: 값 덮어쓰기

이제 상태 정의는 완벽해졌습니다. 그런데 김개발 씨는 또 다른 의문이 들었습니다.

"상태를 업데이트하면 기존 값이 어떻게 되는 거지?" 박시니어 씨가 웃으며 답했습니다. "Reducer를 이해해야 할 시간이네요."

기본 Reducer는 상태 업데이트의 가장 단순한 방식입니다. 마치 칠판에 적힌 글씨를 지우고 새로 쓰는 것처럼, 기존 값을 완전히 덮어씁니다.

LangGraph에서 별도 설정이 없으면 모든 필드는 이 방식으로 동작합니다.

다음 코드를 살펴봅시다.

from typing import TypedDict
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    counter: int
    status: str
    last_update: str

def increment_node(state: AgentState) -> AgentState:
    # counter는 완전히 새로운 값으로 교체됨
    return {
        "counter": state["counter"] + 1,
        "status": "processing",  # 이전 status 값은 사라짐
        "last_update": "2024-01-15"
    }

graph = StateGraph(AgentState)
graph.add_node("increment", increment_node)

김개발 씨의 챗봇이 점점 발전하고 있었습니다. 여러 노드를 만들어 상태를 업데이트하는 로직을 작성했습니다.

그런데 이상한 일이 벌어졌습니다. 한 노드에서 counter를 1 증가시켰더니, 동시에 설정한 줄 몰랐던 다른 필드들이 사라져 버렸습니다.

당황한 김개발 씨에게 박시니어 씨가 물었습니다. "Reducer가 어떻게 동작하는지 알고 있나요?" Reducer란 무엇일까요?

쉽게 비유하자면, Reducer는 마치 문서 편집 프로그램의 '저장 모드'와 같습니다. "덮어쓰기"를 선택하면 기존 파일이 완전히 새 내용으로 교체됩니다.

"추가하기"를 선택하면 기존 내용에 새 내용이 붙습니다. Reducer도 마찬가지로 상태를 어떤 방식으로 업데이트할지 결정합니다.

LangGraph에서 기본 Reducer는 "덮어쓰기" 모드입니다. 노드가 반환하는 딕셔너리의 각 키에 대해, 해당 키의 값을 완전히 새 값으로 교체합니다.

명시적인 업데이트 로직이 없던 시절에는 어땠을까요? 초기 상태 관리 라이브러리들은 업데이트 방식이 일관되지 않았습니다.

어떤 필드는 덮어쓰고, 어떤 필드는 병합하고, 개발자가 매번 고민해야 했습니다. 특히 여러 노드가 동시에 같은 상태를 업데이트하면 예측하기 어려운 결과가 나왔습니다.

더 큰 문제는 디버깅이었습니다. 상태가 예상과 다르게 변경되었을 때, 어느 노드에서 무슨 일이 일어났는지 추적하기가 매우 어려웠습니다.

바로 이런 문제를 해결하기 위해 명시적인 Reducer 개념이 도입되었습니다. 기본 Reducer를 사용하면 업데이트 로직이 매우 명확해집니다.

노드가 반환한 값이 그대로 상태에 반영됩니다. 예측 가능하고 이해하기 쉽습니다.

무엇보다 간단한 업데이트에는 별도 설정 없이 바로 사용할 수 있다는 큰 이점이 있습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 9번째 줄부터 15번째 줄까지가 핵심입니다. increment_node 함수는 상태를 받아서 새로운 상태 딕셔너리를 반환합니다.

11번째 줄에서 counter는 기존 값에 1을 더한 새 값으로 완전히 교체됩니다. 12번째 줄의 status도 "processing"이라는 새 값으로 덮어써집니다.

만약 이전에 "idle"이었다면 그 값은 사라집니다. 중요한 점은 반환 딕셔너리에 포함된 키만 업데이트된다는 것입니다.

만약 AgentState에 user_name 필드가 있었는데 반환 딕셔너리에 없다면, user_name은 그대로 유지됩니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 주문 처리 시스템을 개발한다고 가정해봅시다. 주문 상태는 "접수", "처리중", "배송중", "완료" 같은 값을 갖습니다.

각 단계를 처리하는 노드는 status 필드를 다음 상태로 덮어쓰면 됩니다. 간단하고 명확합니다.

많은 워크플로우 엔진과 상태 머신에서 이런 덮어쓰기 방식을 기본으로 사용합니다. 대부분의 경우 이것만으로 충분하기 때문입니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 부분 업데이트를 의도했는데 실수로 필드를 누락하는 것입니다.

예를 들어 counter만 증가시키려고 했는데, 반환 딕셔너리에 다른 필드를 빼먹으면 그 필드들은 이전 값을 유지합니다. 이것이 의도된 동작인지 확인해야 합니다.

따라서 업데이트할 필드와 유지할 필드를 명확히 구분하여 코드를 작성해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

박시니어 씨의 설명을 들은 김개발 씨는 자신의 코드를 다시 보았습니다. "아, 제가 반환 딕셔너리에 일부 필드만 넣어서 나머지가 유지되고 있었군요!" 기본 Reducer를 제대로 이해하면 상태 업데이트 로직을 더 정확하게 제어할 수 있습니다.

여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - 반환 딕셔너리에 명시적으로 포함된 키만 업데이트되므로, 업데이트 범위를 명확히 하세요.

• TypedDict를 사용하면 IDE가 누락된 필드를 경고해주므로 실수를 줄일 수 있습니다.
• 복잡한 업데이트 로직이 필요하다면 커스텀 Reducer를 고려하세요.

---

4. 커스텀 Reducer: Annotated 사용

김개발 씨는 이제 메시지 목록을 관리해야 했습니다. 새 메시지가 올 때마다 기존 메시지는 유지하고 새 것만 추가하고 싶었습니다.

그런데 기본 Reducer는 덮어쓰기만 하니까 곤란했습니다. "어떻게 하죠?" 박시니어 씨가 미소를 지었습니다.

"이제 커스텀 Reducer를 배울 시간이네요."

커스텀 Reducer는 Annotated를 사용하여 특정 필드의 업데이트 방식을 커스터마이징합니다. 마치 맞춤 정장을 제작하는 것처럼, 각 필드마다 원하는 병합 로직을 지정할 수 있습니다.

리스트에 추가하거나, 숫자를 누적하거나, 딕셔너리를 병합하는 등 다양한 패턴이 가능합니다.

다음 코드를 살펴봅시다.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph

def merge_lists(existing: list, new: list) -> list:
    """기존 리스트에 새 항목들을 추가"""
    return existing + new

def max_value(existing: int, new: int) -> int:
    """두 값 중 큰 값을 선택"""
    return max(existing, new)

class AgentState(TypedDict):
    # 커스텀 Reducer로 리스트 병합
    messages: Annotated[list[str], merge_lists]
    # 최댓값 추적
    max_score: Annotated[int, max_value]
    # 기본 Reducer (덮어쓰기)
    status: str

graph = StateGraph(AgentState)

김개발 씨의 챗봇이 복잡해지면서 새로운 요구사항이 생겼습니다. 대화 히스토리를 관리해야 하는데, 새 메시지가 들어올 때마다 기존 메시지들은 보존하면서 끝에 추가해야 했습니다.

또한 사용자 만족도 점수를 추적하는데, 여러 노드에서 계산한 점수 중 최댓값을 유지해야 했습니다. 기본 Reducer만으로는 불가능했습니다.

덮어쓰기만 하니까 기존 데이터가 계속 사라졌습니다. 박시니어 씨가 키보드를 가져가며 말했습니다.

"이럴 때 커스텀 Reducer를 사용하는 겁니다." 커스텀 Reducer란 무엇일까요? 쉽게 비유하자면, 커스텀 Reducer는 마치 레고 블록의 결합 방식을 직접 정의하는 것과 같습니다.

일반 레고는 위아래로만 결합되지만, 테크닉 레고는 옆으로도, 각도를 주어서도 결합할 수 있습니다. 커스텀 Reducer도 마찬가지로 각 필드마다 "어떻게 결합할지" 자유롭게 정의할 수 있습니다.

Annotated는 Python의 타입 힌트를 확장하는 기능입니다. 타입 정보에 추가 메타데이터를 붙일 수 있는데, LangGraph는 이것을 활용하여 Reducer 함수를 지정합니다.

유연한 업데이트 로직이 없던 시절에는 어땠을까요? 개발자들은 매번 노드 함수 안에서 수동으로 병합 로직을 작성해야 했습니다.

"기존 messages를 가져와서 새 메시지를 추가하고..." 같은 코드를 노드마다 반복해서 작성했습니다. 실수하기도 쉽고, 일관성도 떨어졌습니다.

더 큰 문제는 중복 코드였습니다. 같은 병합 로직이 프로젝트 전체에 흩어져 있으면, 나중에 로직을 변경할 때 모든 곳을 찾아 수정해야 했습니다.

바로 이런 문제를 해결하기 위해 커스텀 Reducer가 도입되었습니다. 커스텀 Reducer를 사용하면 병합 로직이 상태 정의에 선언적으로 포함됩니다.

모든 노드는 그냥 새 값을 반환하기만 하면, LangGraph가 자동으로 정의된 Reducer를 적용합니다. 코드가 간결해지고, 일관성이 보장되며, 유지보수가 쉬워집니다.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 4번째 줄부터 6번째 줄을 보면 merge_lists라는 Reducer 함수를 정의하고 있습니다.

이 함수는 기존 리스트와 새 리스트를 받아서 연결한 결과를 반환합니다. 매우 간단하지만 강력합니다.

8번째 줄부터 10번째 줄의 max_value는 두 정수 중 큰 값을 선택하는 Reducer입니다. 여러 노드에서 점수를 업데이트하면 항상 최댓값만 유지됩니다.

14번째 줄이 핵심입니다. messages 필드의 타입은 list[str]이지만, Annotated로 감싸서 두 번째 인자로 merge_lists 함수를 전달합니다.

이제 messages가 업데이트될 때마다 merge_lists가 자동으로 실행됩니다. 16번째 줄의 max_score도 마찬가지로 max_value Reducer를 사용합니다.

18번째 줄의 status는 Annotated가 없으므로 기본 Reducer가 적용됩니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 실시간 채팅 애플리케이션을 개발한다고 가정해봅시다. 여러 사용자가 동시에 메시지를 보내면, 각 메시지를 히스토리에 추가해야 합니다.

커스텀 Reducer로 메시지 리스트를 정의하면, 각 노드는 그냥 새 메시지만 반환하면 됩니다. LangGraph가 알아서 기존 히스토리에 추가합니다.

또 다른 예로, 여러 AI 모델의 예측 점수를 결합하는 앙상블 시스템에서는 max, min, average 같은 다양한 Reducer를 사용하여 점수를 집계할 수 있습니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 Reducer 함수가 사이드 이펙트를 일으키도록 작성하는 것입니다. Reducer는 순수 함수여야 합니다.

입력을 받아 출력을 반환할 뿐, 외부 상태를 변경하거나 API 호출을 해서는 안 됩니다. 그렇지 않으면 예측 불가능한 동작이 발생할 수 있습니다.

따라서 Reducer는 간단하고 예측 가능하게 유지해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

커스텀 Reducer를 적용한 김개발 씨는 테스트를 돌려봤습니다. 이번에는 메시지가 계속 쌓이고, 점수는 최댓값만 유지되었습니다.

"와, 이거 정말 편한데요!" 커스텀 Reducer를 제대로 이해하면 복잡한 상태 관리도 선언적이고 간결하게 처리할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - Reducer 함수는 순수 함수로 작성하세요. 외부 상태를 변경하지 말고 새 값을 반환하세요.

• 자주 사용하는 Reducer는 별도 모듈로 분리하여 재사용하세요.
• 복잡한 Reducer는 단위 테스트를 작성하여 동작을 검증하세요.

---

5. operator.add 메시지 누적

박시니어 씨가 김개발 씨의 코드를 보더니 고개를 끄덕였습니다. "잘 작성했네요.

그런데 메시지 리스트를 병합하는 건 너무 흔한 패턴이라, Python 표준 라이브러리에 이미 있는 함수를 쓸 수도 있어요." 김개발 씨는 눈이 커졌습니다. "정말요?"

operator.add는 Python 내장 모듈의 함수로, 두 값을 더하는 연산을 수행합니다. 리스트에 사용하면 연결 효과를 내므로, 메시지 누적용 Reducer로 완벽합니다.

별도 함수를 정의할 필요 없이 바로 사용할 수 있어 매우 편리합니다.

다음 코드를 살펴봅시다.

from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    # operator.add를 Reducer로 사용
    messages: Annotated[list[str], add]
    # 숫자도 누적 가능
    total_tokens: Annotated[int, add]
    # 리스트 외에 다른 타입도 지원
    tags: Annotated[list[str], add]

def process_message(state: AgentState) -> AgentState:
    return {
        "messages": ["새 메시지"],  # 기존에 추가됨
        "total_tokens": 150,        # 기존 값에 더해짐
        "tags": ["처리완료"]         # 기존 태그에 추가됨
    }

graph = StateGraph(AgentState)
graph.add_node("process", process_message)

김개발 씨는 지금까지 메시지를 병합하기 위해 merge_lists 같은 함수를 직접 작성했습니다. 잘 작동하긴 했지만, 코드를 볼 때마다 "이게 표준 패턴인데 매번 정의해야 하나?" 하는 생각이 들었습니다.

박시니어 씨가 리팩토링을 도와주면서 operator 모듈을 소개했습니다. "Python 표준 라이브러리에 이미 다 있어요." operator.add란 무엇일까요?

쉽게 비유하자면, operator.add는 마치 만능 접착제와 같습니다. 숫자를 붙이면 덧셈을 하고, 문자열을 붙이면 연결하고, 리스트를 붙이면 병합합니다.

타입에 따라 적절한 "더하기" 연산을 자동으로 수행합니다. Python에서 리스트를 더하면 어떻게 될까요?

[1, 2] + [3, 4]는 [1, 2, 3, 4]가 됩니다. 바로 이 연산을 수행하는 것이 operator.add입니다.

a + b와 add(a, b)는 완전히 동일합니다. 표준 연산자를 활용하지 못하던 시절에는 어땠을까요?

개발자들은 리스트 병합, 숫자 누적, 문자열 연결을 위해 각각 별도 함수를 작성했습니다. concat_lists, sum_numbers, join_strings 같은 함수들이 프로젝트마다 넘쳐났습니다.

본질적으로는 모두 "더하기" 연산인데도 말이죠. 더 큰 문제는 일관성이었습니다.

같은 동작을 하는 함수인데도 이름과 구현이 제각각이었습니다. 새 프로젝트에 들어갈 때마다 "여기서는 리스트 병합을 뭐라고 부르지?" 하고 찾아봐야 했습니다.

바로 이런 문제를 해결하기 위해 operator 모듈을 활용합니다. operator.add를 사용하면 코드가 극도로 간결해집니다.

import 한 줄이면 끝입니다. 누구나 아는 표준 라이브러리 함수이므로 다른 개발자가 봐도 즉시 이해합니다.

무엇보다 Python의 내장 연산자 의미를 그대로 활용하므로, 별도 문서를 읽을 필요가 없습니다. 위의 코드를 한 줄씩 살펴보겠습니다.

먼저 2번째 줄을 보면 operator 모듈에서 add를 임포트합니다. 이것만으로 준비 끝입니다.

7번째 줄이 핵심입니다. messages 필드는 Annotated[list[str], add]로 정의되어 있습니다.

이제 노드가 messages를 업데이트하면, 기존 리스트에 새 리스트가 자동으로 연결됩니다. 9번째 줄의 total_tokens도 주목할 만합니다.

정수 타입인데도 add를 Reducer로 사용합니다. 각 노드가 소비한 토큰 수를 반환하면, 자동으로 누적되어 총합이 계산됩니다.

매우 유용합니다. 14번째 줄부터의 process_message 함수를 보세요.

이 함수는 그냥 새 값들을 반환할 뿐입니다. "기존 messages를 가져와서 새 메시지를 추가하고..." 같은 로직이 전혀 없습니다.

LangGraph가 add Reducer를 자동으로 적용하기 때문입니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 문서 처리 파이프라인을 개발한다고 가정해봅시다. 여러 단계를 거치면서 처리 로그가 쌓이고, 각 단계의 처리 시간이 누적되고, 발견된 이슈들이 리스트에 추가됩니다.

operator.add를 사용하면 이 모든 누적 로직을 선언적으로 정의할 수 있습니다. 많은 데이터 처리 파이프라인에서 이런 누적 패턴은 매우 흔합니다.

operator.add는 가장 간단하면서도 강력한 해법입니다. 하지만 주의할 점도 있습니다.

초보 개발자들이 흔히 하는 실수 중 하나는 operator.add가 항상 사용 가능할 거라고 생각하는 것입니다. 하지만 add는 덧셈이 정의된 타입에만 작동합니다.

딕셔너리는 + 연산자가 없으므로 add를 사용할 수 없습니다. 이런 경우 커스텀 Reducer를 작성해야 합니다.

따라서 타입의 특성을 이해하고 적절한 Reducer를 선택해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

operator.add로 코드를 리팩토링한 김개발 씨는 만족스러운 표정을 지었습니다. "코드가 훨씬 깔끔해졌어요!" 직접 작성한 merge_lists 함수들을 모두 지우고, import 한 줄로 교체했습니다.

operator.add를 제대로 이해하면 보일러플레이트 코드를 크게 줄이고 표준적인 패턴을 사용할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - operator 모듈에는 add 외에도 mul, sub 등 다양한 연산자 함수가 있으니 필요에 따라 활용하세요.

• 리스트와 숫자 누적에는 거의 항상 operator.add가 최선의 선택입니다.
• 타입이 + 연산자를 지원하는지 확인한 후 operator.add를 사용하세요.

---

6. MessagesState 사전 구축 상태

김개발 씨의 챗봇이 거의 완성 단계에 접어들었습니다. 그런데 마지막 리뷰에서 박시니어 씨가 웃으며 말했습니다.

"잘 만들었는데, 사실 LangGraph에 메시지 관리용 상태가 이미 만들어져 있어요." 김개발 씨는 헛웃음을 지었습니다. "진작 말씀해주시지..."

MessagesState는 LangGraph에서 제공하는 사전 구축 상태 클래스입니다. 마치 가구를 직접 만들지 않고 이케아에서 사는 것처럼, 채팅 애플리케이션에 필요한 메시지 관리 기능이 모두 갖춰져 있습니다.

메시지 리스트는 물론 자동 누적 로직까지 포함되어 있어 즉시 사용할 수 있습니다.

다음 코드를 살펴봅시다.

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

# MessagesState는 이미 messages 필드가 정의되어 있음
# messages: Annotated[list[BaseMessage], add_messages]

def chatbot_node(state: MessagesState) -> MessagesState:
    # 마지막 사용자 메시지 가져오기
    last_message = state["messages"][-1]

    # AI 응답 생성
    response = f"'{last_message.content}'에 대한 답변입니다."

    return {
        "messages": [AIMessage(content=response)]
    }

# 그래프 생성 - 커스텀 필드 추가 가능
graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot_node)

김개발 씨는 지금까지 열심히 상태를 정의하고, Reducer를 작성하고, 메시지 관리 로직을 구현했습니다. 잘 작동하는 코드였지만, 박시니어 씨의 말에 조금 허탈했습니다.

"이미 만들어진 게 있었다니..." 박시니어 씨가 위로하듯 말했습니다. "아니에요, 지금까지 배운 것들이 모두 MessagesState를 이해하는 데 필요한 기초였어요." MessagesState란 무엇일까요?

쉽게 비유하자면, MessagesState는 마치 스타터 팩과 같습니다. 게임을 시작할 때 기본 장비가 주어지는 것처럼, 채팅 애플리케이션을 만들 때 필요한 기본 상태가 이미 준비되어 있습니다.

메시지를 담을 리스트, 자동 누적 Reducer, 심지어 메시지 타입까지 모두 갖춰져 있습니다. LangGraph 팀은 수많은 개발자가 비슷한 상태 정의를 반복하는 것을 발견했습니다.

거의 모든 챗봇과 AI 에이전트가 메시지 리스트를 관리했고, 같은 패턴을 사용했습니다. 그래서 이것을 표준화하여 MessagesState로 제공하게 되었습니다.

사전 구축 상태가 없던 시절에는 어땠을까요? 모든 개발자가 프로젝트를 시작할 때마다 같은 코드를 작성했습니다.

"class ChatState(TypedDict): messages: Annotated[list, add] ..." 같은 보일러플레이트가 수백 개의 프로젝트에 중복되었습니다. 실수도 많았습니다.

누군가는 Reducer를 빼먹고, 누군가는 잘못된 타입을 사용했습니다. 더 큰 문제는 일관성이었습니다.

각 프로젝트마다 메시지 필드 이름이 달랐습니다. "messages", "chat_history", "conversation" 등 제각각이었습니다.

라이브러리 간 통합도 어려웠습니다. 바로 이런 문제를 해결하기 위해 MessagesState가 도입되었습니다.

MessagesState를 사용하면 프로젝트 시작이 매우 빨라집니다. 상태 정의에 시간을 쓰지 않고 바로 비즈니스 로직 구현에 집중할 수 있습니다.

LangChain의 메시지 타입과 완벽하게 통합되므로, HumanMessage, AIMessage, SystemMessage 등을 바로 사용할 수 있습니다. 무엇보다 LangGraph 생태계의 표준이므로, 다른 개발자나 라이브러리와 협업이 쉬워집니다.

위의 코드를 한 줄씩 살펴보겠습니다. 먼저 1번째 줄에서 MessagesState를 임포트합니다.

이것만으로 메시지 관리 준비가 끝납니다. 별도로 상태를 정의할 필요가 없습니다.

7번째 줄의 chatbot_node 함수를 보면, state 파라미터의 타입이 MessagesState입니다. 이 상태는 이미 messages 필드를 갖고 있으며, add_messages라는 특별한 Reducer가 적용되어 있습니다.

9번째 줄에서 state["messages"]로 메시지 리스트에 접근합니다. 마지막 메시지를 가져와서 내용을 확인합니다.

15번째 줄이 핵심입니다. AIMessage를 생성하여 messages 리스트로 반환합니다.

add_messages Reducer가 자동으로 적용되므로, 이 메시지는 기존 메시지 리스트 끝에 추가됩니다. 19번째 줄을 보면 그래프를 생성할 때 MessagesState를 그대로 사용합니다.

매우 간단합니다. 실제 현업에서는 어떻게 활용할까요?

예를 들어 고객 지원 AI 챗봇을 개발한다고 가정해봅시다. 고객의 질문과 AI의 답변을 주고받으며 대화 히스토리를 관리해야 합니다.

MessagesState를 사용하면 이 모든 기능이 이미 준비되어 있습니다. 노드에서는 그냥 새 메시지만 반환하면 됩니다.

실제로 LangGraph를 사용하는 수많은 프로덕션 챗봇이 MessagesState를 기반으로 구축되어 있습니다. 검증된 패턴을 사용하면 버그도 줄고 개발 속도도 빨라집니다.

물론 커스터마이징도 가능합니다. MessagesState를 상속하여 추가 필드를 정의할 수 있습니다.

예를 들어 사용자 ID, 세션 정보, 컨텍스트 데이터 등을 추가할 수 있습니다. messages 필드는 그대로 유지되므로, 표준 기능은 그대로 사용하면서 프로젝트 특화 기능을 추가할 수 있습니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 MessagesState가 만능이라고 생각하는 것입니다.

메시지 관리에는 완벽하지만, 다른 종류의 상태에는 맞지 않을 수 있습니다. 예를 들어 단순한 데이터 파이프라인이나 워크플로우 엔진에는 커스텀 상태가 더 적합할 수 있습니다.

따라서 프로젝트 특성에 맞는 상태 구조를 선택해야 합니다. 다시 김개발 씨의 이야기로 돌아가 봅시다.

MessagesState를 사용하여 코드를 리팩토링한 김개발 씨는 놀라움을 금치 못했습니다. 수십 줄의 상태 정의 코드가 단 한 줄의 import로 바뀌었습니다.

"이제야 LangGraph의 진짜 편리함을 알겠어요!" 박시니어 씨가 어깨를 두드렸습니다. "하지만 지금까지 배운 게 헛되지 않아요.

MessagesState가 어떻게 작동하는지 이제 완벽히 이해하잖아요." MessagesState를 제대로 이해하면 채팅 애플리케이션을 훨씬 빠르게 개발할 수 있습니다. 여러분도 오늘 배운 내용을 실제 프로젝트에 적용해 보세요.

실전 팁

💡 - MessagesState를 상속하여 커스텀 필드를 추가할 수 있습니다. class CustomState(MessagesState): user_id: str

• add_messages Reducer는 중복 메시지를 자동으로 제거하는 등 지능적인 병합을 수행합니다.
• LangChain 생태계와 통합할 때는 항상 MessagesState 사용을 우선 고려하세요.

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!