[AI] add_messages

개념

add_messages는 LangGraph에서 제공하는 reducer 함수로, 메시지 리스트를 지능적으로 병합합니다.

 

 

python

from langgraph.graph import add_messages

# 기본 사용법
from typing import Annotated, TypedDict

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

핵심 동작 원리

1. 기본 병합 (누적)

 

 

python

from langgraph.graph import add_messages

# 초기 상태
existing = [
    {"role": "user", "content": "안녕하세요"}
]

# 새 메시지
new = [
    {"role": "assistant", "content": "반갑습니다!"}
]

# add_messages 적용 결과
result = add_messages(existing, new)
# [
#     {"role": "user", "content": "안녕하세요"},
#     {"role": "assistant", "content": "반갑습니다!"}
# ]

2. ID 기반 덮어쓰기

 

 

python

from langchain_core.messages import AIMessage

# 초기 메시지
existing = [
    AIMessage(content="초안입니다", id="msg_1")
]

# 같은 ID로 업데이트
new = [
    AIMessage(content="수정된 내용입니다", id="msg_1")
]

result = add_messages(existing, new)
# [AIMessage(content="수정된 내용입니다", id="msg_1")]
# 기존 메시지가 교체됨

3. 혼합 동작

 

 

python

from langchain_core.messages import HumanMessage, AIMessage

existing = [
    HumanMessage(content="질문", id="msg_1"),
    AIMessage(content="초안 답변", id="msg_2")
]

new = [
    AIMessage(content="수정된 답변", id="msg_2"),  # 덮어쓰기
    HumanMessage(content="추가 질문", id="msg_3")  # 새로 추가
]

result = add_messages(existing, new)
# [
#     HumanMessage(content="질문", id="msg_1"),
#     AIMessage(content="수정된 답변", id="msg_2"),  # 수정됨
#     HumanMessage(content="추가 질문", id="msg_3")  # 추가됨
# ]

사용 케이스

케이스 1: 기본 챗봇

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages, START, END
from langchain_core.messages import HumanMessage, AIMessage

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

def chatbot(state: ChatState):
    user_message = state["messages"][-1].content
    response = f"당신은 '{user_message}'라고 말했습니다."
    
    return {
        "messages": [AIMessage(content=response)]
    }

# 그래프 구성
workflow = StateGraph(ChatState)
workflow.add_node("chat", chatbot)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)

app = workflow.compile()

# 실행
result = app.invoke({
    "messages": [HumanMessage(content="안녕하세요")]
})

print(result["messages"])
# [HumanMessage("안녕하세요"), AIMessage("당신은 '안녕하세요'라고 말했습니다.")]

케이스 2: 스트리밍 응답 수정

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import AIMessage
import uuid

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

def streaming_agent(state: StreamState):
    # 스트리밍 시뮬레이션
    message_id = str(uuid.uuid4())
    
    # 1단계: 부분 응답
    yield {
        "messages": [AIMessage(content="생각 중", id=message_id)]
    }
    
    # 2단계: 업데이트
    yield {
        "messages": [AIMessage(content="생각 중...", id=message_id)]
    }
    
    # 3단계: 최종 응답
    yield {
        "messages": [AIMessage(content="최종 답변입니다!", id=message_id)]
    }

# 같은 ID를 사용하므로 메시지가 계속 업데이트됨

케이스 3: 멀티턴 대화

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages, START, END
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

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

def add_system_message(state: ConversationState):
    return {
        "messages": [SystemMessage(content="당신은 친절한 AI 어시스턴트입니다.")]
    }

def respond(state: ConversationState):
    # 마지막 사용자 메시지 가져오기
    last_user_message = None
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            last_user_message = msg
            break
    
    if last_user_message:
        return {
            "messages": [AIMessage(content=f"답변: {last_user_message.content}")]
        }
    
    return {"messages": []}

# 그래프 구성
workflow = StateGraph(ConversationState)
workflow.add_node("system", add_system_message)
workflow.add_node("respond", respond)
workflow.add_edge(START, "system")
workflow.add_edge("system", "respond")
workflow.add_edge("respond", END)

app = workflow.compile()

# 여러 턴 대화
messages = []
for user_input in ["안녕하세요", "날씨 어때요?", "고마워요"]:
    messages.append(HumanMessage(content=user_input))
    result = app.invoke({"messages": messages})
    messages = result["messages"]
    print(f"User: {user_input}")
    print(f"AI: {messages[-1].content}\n")

케이스 4: 에이전트 체인

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages, START, END
from langchain_core.messages import HumanMessage, AIMessage

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str

def research_agent(state: AgentState):
    """정보 수집 에이전트"""
    return {
        "messages": [AIMessage(
            content="[연구 에이전트] 관련 자료를 수집했습니다.",
            name="researcher"
        )],
        "current_agent": "writer"
    }

def writer_agent(state: AgentState):
    """작성 에이전트"""
    return {
        "messages": [AIMessage(
            content="[작성 에이전트] 보고서를 작성했습니다.",
            name="writer"
        )],
        "current_agent": "reviewer"
    }

def reviewer_agent(state: AgentState):
    """검토 에이전트"""
    return {
        "messages": [AIMessage(
            content="[검토 에이전트] 검토를 완료했습니다.",
            name="reviewer"
        )],
        "current_agent": "done"
    }

def route_agent(state: AgentState):
    agent = state.get("current_agent", "researcher")
    if agent == "done":
        return END
    return agent

# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("researcher", research_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", reviewer_agent)

workflow.add_edge(START, "researcher")
workflow.add_conditional_edges(
    "researcher",
    route_agent,
    {"writer": "writer", END: END}
)
workflow.add_conditional_edges(
    "writer",
    route_agent,
    {"reviewer": "reviewer", END: END}
)
workflow.add_conditional_edges(
    "reviewer",
    route_agent,
    {END: END}
)

app = workflow.compile()

# 실행
result = app.invoke({
    "messages": [HumanMessage(content="보고서를 작성해주세요")],
    "current_agent": "researcher"
})

for msg in result["messages"]:
    print(f"{getattr(msg, 'name', 'user')}: {msg.content}")

케이스 5: 메시지 필터링과 함께 사용

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, RemoveMessage

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

def cleanup_old_messages(state: FilteredState):
    """오래된 메시지 제거"""
    messages = state["messages"]
    
    # 최근 10개만 유지
    if len(messages) > 10:
        # 오래된 메시지 ID 수집
        old_message_ids = [msg.id for msg in messages[:-10]]
        
        # RemoveMessage 사용하여 제거
        return {
            "messages": [RemoveMessage(id=msg_id) for msg_id in old_message_ids]
        }
    
    return {"messages": []}

케이스 6: 조건부 메시지 추가

 

 

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages, START, END
from langchain_core.messages import HumanMessage, AIMessage

class ConditionalState(TypedDict):
    messages: Annotated[list, add_messages]
    include_examples: bool

def responder(state: ConditionalState):
    user_msg = state["messages"][-1].content
    
    messages_to_add = [
        AIMessage(content=f"질문: {user_msg}에 대한 답변입니다.")
    ]
    
    # 조건부로 예시 추가
    if state.get("include_examples", False):
        messages_to_add.append(
            AIMessage(content="예시: 이렇게 사용할 수 있습니다.")
        )
    
    return {"messages": messages_to_add}

workflow = StateGraph(ConditionalState)
workflow.add_node("respond", responder)
workflow.add_edge(START, "respond")
workflow.add_edge("respond", END)

app = workflow.compile()

# 예시 포함
result1 = app.invoke({
    "messages": [HumanMessage(content="도움말")],
    "include_examples": True
})
print(len(result1["messages"]))  # 3개 (user + 2 AI)

# 예시 미포함
result2 = app.invoke({
    "messages": [HumanMessage(content="도움말")],
    "include_examples": False
})
print(len(result2["messages"]))  # 2개 (user + 1 AI)

add_messages vs 일반 리스트 비교

 

 

python

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

# 방법 1: add_messages 사용 (병합)
class StateWithReducer(TypedDict):
    messages: Annotated[list, add_messages]

# 방법 2: 일반 리스트 (덮어쓰기)
class StateWithoutReducer(TypedDict):
    messages: list

# 테스트
def test_node(state):
    return {"messages": [{"role": "ai", "content": "새 메시지"}]}

# StateWithReducer 사용 시:
# 입력: {"messages": [{"role": "user", "content": "안녕"}]}
# 출력: {"messages": [
#           {"role": "user", "content": "안녕"},
#           {"role": "ai", "content": "새 메시지"}
#       ]}

# StateWithoutReducer 사용 시:
# 입력: {"messages": [{"role": "user", "content": "안녕"}]}
# 출력: {"messages": [{"role": "ai", "content": "새 메시지"}]}  # 기존 메시지 사라짐!

주요 장점

1. 자동 병합: 명시적인 병합 코드 불필요
2. ID 기반 업데이트: 스트리밍, 수정 시나리오에 유용
3. 대화 히스토리 유지: 멀티턴 대화에 필수
4. 에러 감소: 실수로 메시지 덮어쓰는 것 방지

주의사항

• LangGraph 전용 기능 (일반 Python 코드에서는 불가)
• 메시지는 LangChain 메시지 타입 권장 (HumanMessage, AIMessage 등)
• 메모리 관리 필요 시 주기적으로 오래된 메시지 제거

 

with Claude