LangGraph 기초 문법 - State 업데이트, add_messages, invoke·stream, 조건·반복

LangGraph는 LangChain 생태계에서 에이전트나 RAG 시스템을 그래프(노드/엣지) 로 설계하고 실행할 수 있게 해주는 오케스트레이션 프레임워크다. 직선형 파이프라인만 만드는 게 아니라 분기(conditional), 반복(loop), 병렬(parallel), 스트리밍(stream) 같은 흐름을 “구조로” 표현할 수 있다는 점이 핵심이다.

이번 글은 LangGraph를 처음 볼 때 가장 헷갈리는 기초 문법을 한 번에 정리한다. 특히 State 업데이트, 메시지 누적 리듀서(add_messages), invoke/ainvoke/stream/astream, 그리고 조건 분기/반복까지 “코드가 어디에서 어떻게 이어지는지” 기준으로 정리해본다. 원문 흐름은 랭그래프 기초 문법을 바탕으로 재구성했다.

---

1. 그래프의 상태(State) 업데이트: “return으로 상태를 갱신한다”

LangGraph의 노드는 보통 이런 형태다.

• 입력: state (딕셔너리처럼 생긴 상태)
• 출력: “업데이트할 값”을 담은 딕셔너리

중요한 포인트는 상태를 직접 mutate하는 게 아니라, 반환값으로 업데이트를 전달한다는 점이다.

아래는 가장 기본적인 “메시지에 AI 답변 하나 추가” 예시다.

!pip install langgraph

from langchain_core.messages import AnyMessage, AIMessage, HumanMessage
from typing_extensions import TypedDict
from langgraph.graph import StateGraph


class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int


def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("안녕하세요! 무엇을 도와드릴까요?")
    return {"messages": messages + [new_message], "extra_field": 10}


graph_builder = StateGraph(State)
graph_builder.add_node("node", node)
graph_builder.set_entry_point("node")  # START -> node
graph = graph_builder.compile()

result = graph.invoke({"messages": [HumanMessage("안녕")]})
print(result["messages"])

핵심은 딱 하나다.

• 기존 messages를 가져와서
• 새 AIMessage를 붙이고
• 그 리스트를 다시 return 한다

---

2. 대화 메시지 누적 업데이트: add_messages 리듀서가 깔끔한 이유

대화형 에이전트에서 가장 자주 하는 일이 “messages를 계속 누적”하는 건데, 이걸 매번 messages + [new]로 쓰면 코드가 금방 지저분해진다.

여기서 등장하는 게 add_messages다.

• add_messages는
• 기존 messages에 새 messages를 병합(merge) 하는 리듀서 역할을 한다.

리듀서를 붙이면, 노드는 “메시지 하나만 return”해도 LangGraph가 알아서 누적해 준다.

from langchain_core.messages import AnyMessage, AIMessage
from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages


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


def node(state: State):
    new_message = AIMessage("안녕하세요! 무엇을 도와드릴까요?")
    return {"messages": new_message, "extra_field": 10}


graph_builder = StateGraph(State)
graph_builder.add_node("node", node)
graph_builder.set_entry_point("node")
graph = graph_builder.compile()

이 방식의 장점은:

• 노드가 “내가 추가하고 싶은 메시지”만 반환하면 되고
• 누적/병합 로직은 상태 스키마에서 통제되며
• 여러 노드가 동시에 메시지를 내는 경우도 예측 가능해진다는 점이다

---

3. invoke / ainvoke / stream / astream: 실행 모드 네 가지

LangGraph를 “앱처럼” 쓸 때 결국 필요한 건 실행 방식이다.

• invoke: 동기 실행. 결과가 나올 때까지 기다린다.
• ainvoke: 비동기 실행. 여러 요청을 동시에 처리할 때 유리하다.
• stream: 중간 결과를 실시간으로 받는다.
• astream: 비동기 스트리밍.

3-1. invoke

input_message = {"role": "user", "content": "안녕하세요."}
result = graph.invoke({"messages": [input_message]})
print(result)

3-2. ainvoke

result = await graph.ainvoke({"messages": [input_message]})
print(result)

3-3. stream (values / updates / messages)

stream_mode는 상황에 따라 느낌이 확 달라진다.

• values: 각 단계의 “전체 상태”
• updates: 각 단계의 “변경분”
• messages: 메시지 단위 스트리밍

for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
    print(chunk)

for chunk in graph.stream({"messages": [input_message]}, stream_mode="updates"):
    print(chunk)

for chunk_msg, metadata in graph.stream({"messages": [input_message]}, stream_mode="messages"):
    print(chunk_msg.content)
    print(metadata["langgraph_node"])

---

4. 노드와 엣지 연결: 순차, 시퀀스, 병렬

4-1. 노드와 엣지 순차 연결

from typing_extensions import TypedDict
from langgraph.graph import START, StateGraph


class State(TypedDict):
    value_1: str
    value_2: int


def step_1(state: State):
    return {"value_1": state["value_1"]}


def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} b"}


def step_3(state: State):
    return {"value_2": 10}


graph_builder = StateGraph(State)
graph_builder.add_node(step_1)
graph_builder.add_node(step_2)
graph_builder.add_node(step_3)

graph_builder.add_edge(START, "step_1")
graph_builder.add_edge("step_1", "step_2")
graph_builder.add_edge("step_2", "step_3")

graph = graph_builder.compile()
print(graph.invoke({"value_1": "apple"}))

4-2. add_sequence로 한 번에 연결

graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")
graph = graph_builder.compile()
print(graph.invoke({"value_1": "c"}))

4-3. 병렬(Parallel) 연결

원문에서 병렬 연결 구조를 그림으로 보면 이해가 빠르다.

병렬 누적은 보통 operator.add 같은 리듀서를 이용해 “결과를 리스트로 누적”하는 패턴을 쓴다.

---

5. 조건 분기와 반복: 루프를 만들 때 어디가 핵심인가

조건부 엣지(conditional edges)는 “상태를 보고 다음 노드를 결정”하게 해준다. 루프는 “조건이 맞으면 다시 돌아가 반복”하게 한다.

원문 예시처럼 그림으로 보면 구조가 잡힌다.

그리고 반복이 너무 깊어지면 GraphRecursionError가 날 수 있으니, recursion_limit은 디버깅할 때 특히 중요하다.

---

마치며

• LangGraph의 핵심은 “노드/엣지/상태”를 코드로 정의해, 분기·반복·병렬 같은 워크플로우를 구조적으로 만드는 것이다.
• 메시지 누적은 add_messages 같은 리듀서를 붙이면 코드가 훨씬 단단해진다.
• invoke/ainvoke/stream/astream은 운영·디버깅·UX에서 체감 차이가 큰 옵션이라, 초반에 같이 익혀 두는 게 좋다.

---

참고

• 랭그래프 기초 문법