2026-04-29 강의 정리

---

목차

1. 문제와 접근
2. 기본 State vs Annotated State
3. Reducer 개념 (Annotated + 함수)
4. operator.add — 리스트 누적
5. add_messages — LangGraph 공식 리듀서
6. 메시지 타입 정리
7. add_messages의 ID 기반 중복 제거
8. 커스텀 리듀서 — keep_max, add_int, merge_dicts
9. RemoveMessage로 메시지 삭제
10. 메시지 필터링 (ToolMessage 제거)
11. 대화 turn 관리 패턴
12. 핵심 개념 총정리
13. 자주 나오는 실수/주의사항
14. 상태 갱신 규칙을 데이터별로 선택하기

---

1. 문제와 접근

상태를 갱신할 때 기존 값을 덮어쓸지 누적할지는 데이터의 의미에 따라 달라진다. LangGraph reducer를 사용해 메시지, 카운터, 최대값처럼 서로 다른 상태의 갱신 규칙을 살펴본다. 업데이트가 반복되어도 의도한 결과가 유지되는지가 중요한 확인 기준이다.

기본 State는 새 값이 이전 값을 덮어쓰지만,
Reducer를 사용하면 “누적”, “최댓값 유지”, “딕셔너리 병합” 등 다양한 업데이트 전략을 쓸 수 있다.

import os, operator
from typing import TypedDict, Annotated, Optional, get_type_hints
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

llm = ChatOpenAI(model="gpt-4o-mini")

---

2. 기본 State vs Annotated State

기본 State (덮어쓰기)

class SimpleState(TypedDict):
    msgs: list

def node_a(state): return {'msgs': ['A 추가']}  # 이전 값 덮어씀
def node_b(state): return {'msgs': ['B 추가']}  # 이전 값 덮어씀

# node_a → node_b 실행 시 최종 msgs = ['B 추가'] (A 사라짐)

Annotated State (리듀서 적용)

from typing import Annotated

class AccumState(TypedDict):
    msgs: Annotated[list, operator.add]   # 새 값을 기존 값에 누적

def node_a(state): return {'msgs': ['A 추가']}
def node_b(state): return {'msgs': ['B 추가']}

# node_a → node_b 실행 시 최종 msgs = ['A 추가', 'B 추가'] (누적됨)

Annotated[타입, 리듀서함수] 형태로 작성
리듀서 함수는 (현재값, 새값) → 병합된값 을 반환해야 한다

---

3. Reducer 개념 (Annotated + 함수)

def my_reducer(cur, new):
    return (cur or []) + new

class AnnotatedState(TypedDict):
    items: Annotated[list, my_reducer]

Annotated의 역할 확인

from typing import get_type_hints

hints = get_type_hints(AnnotatedState, include_extras=True)
hints
# → {'items': typing.Annotated[list, <function my_reducer ...>]}

Annotated 두 번째 인자가 LangGraph에 의해 리듀서로 사용됨

---

4. operator.add — 리스트 누적

파이썬 내장 operator.add는 리스트에서 + 연산자처럼 작동한다.

operator.add([1], [2])   # → [1, 2]
operator.add(1, 2)       # → 3

실습 — 과목별 점수 누적

class ScoreState(TypedDict):
    scores: Annotated[list, operator.add]

def add_kor(state):  return {'scores': [80]}
def add_eng(state):  return {'scores': [90]}
def add_math(state): return {'scores': [70]}

b = StateGraph(ScoreState)
b.add_node('add_kor',  add_kor)
b.add_node('add_eng',  add_eng)
b.add_node('add_math', add_math)
b.add_edge(START, 'add_kor')
b.add_edge('add_kor',  'add_eng')
b.add_edge('add_eng',  'add_math')
b.add_edge('add_math', END)
app = b.compile()

app.invoke({'scores': []})
# → {'scores': [80, 90, 70]}

---

5. add_messages — LangGraph 공식 리듀서

LangGraph가 메시지 리스트 관리를 위해 제공하는 리듀서.

from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

def user_says(state):
    return {'messages': [HumanMessage(content='안녕')]}

def bot_says(state):
    return {'messages': [AIMessage(content='반갑습니다')]}

b = StateGraph(ChatState)
b.add_node('user_says', user_says)
b.add_node('bot_says',  bot_says)
b.add_edge(START, 'user_says')
b.add_edge('user_says', 'bot_says')
b.add_edge('bot_says', END)
app = b.compile()

result = app.invoke({'messages': []})
# → {'messages': [HumanMessage('안녕'), AIMessage('반갑습니다')]}

operator.add 와 add_messages 의 차이:
operator.add는 단순 리스트 연결
add_messages는 메시지 객체의 ID를 기준으로 중복을 처리함

---

6. 메시지 타입 정리

sys_m  = SystemMessage(content='당신은 컴퓨터 전문가입니다')
hum_m  = HumanMessage(content='랭그래프?')
ai_m   = AIMessage(content='그래프 기반 라이브러리')
tool_m = ToolMessage(content='검색결과....', tool_call_id='call_abc123')

for m in [sys_m, hum_m, ai_m, tool_m]:
    print(f"type={m.type} | class={type(m).__name__} | content={m.content}")

# type=system | class=SystemMessage | content=당신은 컴퓨터 전문가입니다
# type=human  | class=HumanMessage  | content=랭그래프?
# type=ai     | class=AIMessage     | content=그래프 기반 라이브러리
# type=tool   | class=ToolMessage   | content=검색결과....

AIMessage with tool_calls

ai_with_tools = AIMessage(
    content="",
    tool_calls=[
        {'name': 'search', 'args': {'q': '랭그래프'}, 'id': 'call_001'},
        {'name': 'calc',   'args': {'x': 7, 'y': 3},  'id': 'call_002'}
    ]
)

len(ai_with_tools.tool_calls)   # → 2

for tc in ai_with_tools.tool_calls:
    print(f"  - {tc['name']} ({tc['args']}), id={tc['id']}")

---

7. add_messages의 ID 기반 중복 제거

메시지 ID는 같은 내용을 찾는 기준과 다르다. 기존 메시지를 수정하려는 경우와 새 메시지를 추가하려는 경우에 어떤 ID를 전달할지 정해 두어야 의도하지 않은 중복이나 덮어쓰기를 구분할 수 있다.

같은 ID를 가진 메시지가 들어오면 기존 메시지를 새 메시지로 교체한다.

m1 = AIMessage(content='첫 응답',    id='msg_x')
m2 = AIMessage(content='두번째 응답', id='msg_x')   # 같은 ID

merged = add_messages([m1], [m2])
# → [AIMessage(content='두번째 응답', id='msg_x')]   # m1이 m2로 교체됨

이 동작을 이용해 메시지를 수정하거나 업데이트할 수 있다
operator.add 는 단순 연결이라 같은 ID 메시지가 두 번 쌓인다

---

8. 커스텀 리듀서 — keep_max, add_int, merge_dicts

커스텀 reducer는 빈 초기 상태와 반복 업데이트를 함께 확인한다. 합산·최댓값·딕셔너리 병합은 같은 입력을 다시 받았을 때 동작이 다르므로, 재실행이 생기는 경로에서 원하는 갱신 규칙인지 검토한다.

keep_max — 최댓값 유지

def keep_max(cur, new):   # cur=10, new=30
    return max(cur or 0, new)   # → 30

class MaxState(TypedDict):
    score: Annotated[int, keep_max]

def round1(state): return {'score': 50}
def round2(state): return {'score': 80}
def round3(state): return {'score': 70}

# 실행 결과: score = 80 (최댓값 유지)

add_int — 정수 누적합

def add_int(cur, new):
    return (cur or 0) + new

class SumState(TypedDict):
    total: Annotated[int, add_int]

# s1(10) → s2(20) → s3(30) 실행 시 total = 60

merge_dicts — 딕셔너리 병합

def merge_dicts(cur, new):
    return {**(cur or {}), **(new or {})}

class MetaState(TypedDict):
    meta: Annotated[dict, merge_dicts]

def fetch_user(state):    return {'meta': {'user_id':   'u123'}}
def fetch_session(state): return {'meta': {'session_id': 's456'}}
def fetch_locale(state):  return {'meta': {'locale':     'ko-KR'}}

# 최종: {'user_id': 'u123', 'session_id': 's456', 'locale': 'ko-KR'}

같은 키가 있으면 나중에 실행된 노드의 값으로 덮어쓴다

복합 State 예시

class GameState(TypedDict):
    events: Annotated[list, operator.add]   # 이벤트 로그 누적
    best:   Annotated[int, keep_max]        # 최고 점수 유지
    stats:  Annotated[dict, merge_dicts]    # 통계 병합

def play_round1(state):
    return {"events": ["round1"], "best": 30, "stats": {"plays": 1}}

def play_round2(state):
    return {"events": ["round2"], "best": 50, "stats": {"wins": 1}}

# 최종:
# events = ['round1', 'round2']
# best   = 50
# stats  = {'plays': 1, 'wins': 1}

---

9. RemoveMessage로 메시지 삭제

메시지가 너무 많이 쌓이면 오래된 메시지를 삭제한다.

from langgraph.graph.message import RemoveMessage

N = 3   # 최근 N개만 유지

class TrimState(TypedDict):
    messages: Annotated[list, add_messages]

def add_many(state):
    new_msgs = [HumanMessage(content=f"msg-{i}") for i in range(10)]
    return {'messages': new_msgs}

def trim_node(state):
    msgs      = state['messages']
    to_remove = msgs[:-N]               # 뒤에서 N개 빼고 나머지 삭제 대상
    return {'messages': [RemoveMessage(id=m.id) for m in to_remove]}

b = StateGraph(TrimState)
b.add_node('add_many', add_many)
b.add_node('trim_node', trim_node)
b.add_edge(START, 'add_many')
b.add_edge('add_many', 'trim_node')
b.add_edge('trim_node', END)
app = b.compile()

result = app.invoke({'messages': []})
len(result['messages'])   # → 3  (msg-7, msg-8, msg-9 만 남음)

RemoveMessage(id=…) 를 add_messages 리듀서에 전달하면
해당 ID의 메시지가 삭제된다

슬라이싱 방향 주의

msgs[:-3]    # 마지막 3개를 제외한 앞부분 (삭제 대상)
msgs[-3:]    # 마지막 3개 (남길 것)

---

10. 메시지 필터링 (ToolMessage 제거)

LLM에게 ToolMessage를 보내지 않고 싶을 때 필터링한다.

def drop_tool(msgs):
    return [m for m in msgs if not isinstance(m, ToolMessage)]

def only_human(msgs):
    return [m for m in msgs if isinstance(m, HumanMessage)]

필터링 적용 노드

class FilterState(TypedDict):
    messages: Annotated[list, add_messages]
    reply   : str

def seed(state):
    return {"messages": [
        SystemMessage(content="당신은 친절한 도우미"),
        HumanMessage(content="LangGraph는 무엇인가요?"),
        AIMessage(content="확인해보겠습니다."),
        ToolMessage(content="(검색 결과 raw)", tool_call_id="tx"),
    ]}

def respond(state):
    filtered = drop_tool(state["messages"])   # ToolMessage 제거
    response = llm.invoke(filtered)
    return {"reply": response.content}

---

11. 대화 turn 관리 패턴

Annotated[list, add_messages]를 이용해 다회 대화를 유지한다.

class TurnState(TypedDict):
    messages: Annotated[list, add_messages]

def chat_turn(state):
    response = llm.invoke(state["messages"])
    return {'messages': [response]}

b = StateGraph(TurnState)
b.add_node('chat_turn', chat_turn)
b.add_edge(START, 'chat_turn')
b.add_edge('chat_turn', END)
turn_app = b.compile()

history = [SystemMessage(content='당신은 짧게 대답하는 도우미입니다')]
for user_text in ['안녕!', '내 이름은 이상호야', '내 이름이 뭐였지?']:
    out     = turn_app.invoke({'messages': history + [HumanMessage(content=user_text)]})
    history = out['messages']
    print(f"USER: {user_text}")
    print(f"BOT:  {history[-1].content}")

# USER: 안녕!
# BOT:  안녕하세요! 어떻게 도와드릴까요?
# USER: 내 이름은 이상호야
# BOT:  만나서 반가워요, 이상호님!
# USER: 내 이름이 뭐였지?
# BOT:  이상호님이신데요.

---

12. 핵심 개념 총정리

리듀서 선택 기준

상황	리듀서	예시
메시지 누적 (LLM 대화)	add_messages	ChatState
단순 리스트 누적	operator.add	이벤트 로그
최댓값만 유지	keep_max	최고 점수
숫자 합산	add_int	총점
딕셔너리 병합	merge_dicts	설정, 메타데이터

State 업데이트 방식 비교

[기본 State]
  노드 반환 {'key': 새값}
  → 해당 키를 새값으로 덮어씀

[Annotated + Reducer]
  노드 반환 {'key': 새값}
  → reducer(현재값, 새값) 의 결과로 업데이트

---

13. 자주 나오는 실수/주의사항

실수 1: Annotated 없이 리스트 누적 시도

# 틀린 코드 → 매번 덮어써짐
class State(TypedDict):
    msgs: list

def node_a(state): return {'msgs': ['A']}
def node_b(state): return {'msgs': ['B']}
# 최종: msgs = ['B']  (A가 덮어써짐)

# 올바른 코드
class State(TypedDict):
    msgs: Annotated[list, operator.add]
# 최종: msgs = ['A', 'B']

실수 2: operator.add와 add_messages 혼동

# operator.add → 단순 리스트 연결, ID 중복 처리 없음
# add_messages  → ID 기반 중복 제거, RemoveMessage 처리

# 대화 메시지 관리는 반드시 add_messages 사용
class WrongState(TypedDict):
    messages: Annotated[list, operator.add]   # 메시지 수정/삭제 불가

class CorrectState(TypedDict):
    messages: Annotated[list, add_messages]   # ID 기반 관리 가능

실수 3: keep_max에서 초기값 처리 빠뜨리기

# 틀린 코드 → cur가 None일 때 max(None, 50) → TypeError
def keep_max(cur, new):
    return max(cur, new)

# 올바른 코드 → cur or 0 으로 None 처리
def keep_max(cur, new):
    return max(cur or 0, new)

실수 4: RemoveMessage 사용 후 ID 조회

# RemoveMessage를 보낼 때 반드시 실제 메시지 ID 사용
# 메시지 객체의 .id 속성이 있음
to_remove = msgs[:-N]
return {'messages': [RemoveMessage(id=m.id) for m in to_remove]}
# id=m.id 가 아니라 id='abc' 같은 임의 문자열을 주면 삭제 안 됨

실수 5: merge_dicts에서 같은 키 값 유지 착각

# merge_dicts는 나중 값이 앞 값을 덮어씀
def merge_dicts(cur, new):
    return {**(cur or {}), **(new or {})}

# fetch_user  → meta = {'user_id': 'u123'}
# fetch_user2 → meta = {'user_id': 'u456'}  ← 같은 키
# 최종: meta = {'user_id': 'u456'}  (u123 덮어써짐)

---

14. 상태 갱신 규칙을 데이터별로 선택하기

누적이 필요한 데이터와 최신 값만 필요한 데이터를 같은 reducer로 다루지 않는다. 특히 전체 상태를 다시 반환하는 노드와 누적 reducer가 결합하면 기존 값이 다시 적용될 수 있으므로 노드는 변경할 항목을 명확히 반환한다.

메시지를 제거하거나 요약할 때는 도구 호출과 결과의 연결도 확인한다. 기록 길이만 줄이고 호출 관계를 끊으면 다음 모델 입력을 해석하기 어려워질 수 있어, 보존할 단위를 먼저 정하는 편이 낫다.