LangGraph 기초 문법 – 그래프 구성부터 반복 실행까지

1. 그래프의 기본 구성 요소

LangGraph에서 그래프를 만들기 위한 최소 구성 요소는 세 가지입니다.

• State: 그래프 전체에서 공유되는 데이터 구조 (TypedDict로 정의)
• Node: 상태를 받아서 새 상태를 반환하는 함수
• Edge: 노드와 노드를 연결하는 흐름

가장 단순한 예시를 보면:

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()

set_entry_point("node")는 그래프의 시작 노드를 지정하는 역할로,
START → "node" 엣지를 자동으로 추가하는 것과 같습니다.

실행은 graph.invoke()로 합니다.

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

---

2. 대화 메시지 관리 – add_messages

2.1 add_messages란?

기본 State에서 메시지를 관리하면 매번 전체 리스트를 직접 이어 붙여야 합니다.
이를 자동으로 처리해 주는 것이 add_messages 리듀서(Reducer) 입니다.

from typing_extensions import Annotated
from langgraph.graph.message import add_messages

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

Annotated[list[AnyMessage], add_messages]로 선언하면,
노드가 새 메시지만 반환해도 기존 메시지 뒤에 자동으로 추가됩니다.

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

2.2 invoke / ainvoke / stream

그래프를 실행하는 방법은 크게 세 가지입니다.

메서드	설명
invoke	한 번의 요청을 동기적으로 처리, 최종 결과만 반환
ainvoke	비동기 처리용 (await 필요)
stream	각 단계의 결과를 실시간으로 스트리밍

stream_mode 옵션

stream에는 세 가지 모드가 있습니다.

# stream_mode="values" : 각 단계 실행 후 전체 상태 반환 (기본값: "updates")
for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

# stream_mode="updates" : 각 단계에서 변경된 부분만 반환
for chunk in graph.stream({"messages": [input_message]}, stream_mode="updates"):
    for node, value in chunk.items():
        print(node, value['messages'].content)

# stream_mode="messages" : 메시지 단위로 스트리밍
for chunk_msg, metadata in graph.stream({"messages": [input_message]}, stream_mode="messages"):
    print(chunk_msg.content)
    print(metadata["langgraph_node"])

pretty_print()는 메시지나 객체를 보기 좋게(Pretty) 출력해 주는 메서드입니다.

---

3. 기본 선형 실행

여러 노드를 순서대로 연결하는 가장 기본적인 패턴입니다.

from langgraph.graph import START, StateGraph

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

def step_1(state): return {"value_1": state["value_1"]}
def step_2(state): return {"value_1": f"{state['value_1']} b"}
def step_3(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")  # START → 1
graph_builder.add_edge("step_1", "step_2")  # 1 → 2
graph_builder.add_edge("step_2", "step_3")  # 2 → 3

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

---

4. 순서를 한번에 추가 – add_sequence

위처럼 엣지를 하나씩 추가하는 대신, 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()
graph.invoke({"value_1": "c"})

add_sequence는 리스트 순서대로 노드 간 엣지를 자동으로 연결해 줍니다.

---

5. 팬아웃 실행 (병렬 실행)

하나의 노드에서 여러 노드를 동시에 실행하는 패턴입니다.

여러 노드의 결과를 하나의 상태로 합쳐야 하므로, operator.add 리듀서를 사용합니다.

import operator
from typing import Annotated
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]  # 업데이트마다 리스트에 추가

def a(state): return {"aggregate": ["A"]}
def b(state): return {"aggregate": ["B"]}
def c(state): return {"aggregate": ["C"]}
def d(state): return {"aggregate": ["D"]}

graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)

graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")  # a → b (병렬)
graph_builder.add_edge("a", "c")  # a → c (병렬)
graph_builder.add_edge("b", "d")  # b → d (합류)
graph_builder.add_edge("c", "d")  # c → d (합류)
graph_builder.add_edge("d", END)

operator.add는 리스트를 이어 붙이는 역할로,
여러 병렬 노드의 결과를 하나의 리스트에 누적할 때 사용합니다.

---

6. 조건부 팬아웃 엣지

조건에 따라 서로 다른 노드 조합으로 분기하고 싶을 때는
add_conditional_edges에 라우팅 함수와 가능한 노드 목록을 함께 전달합니다.

from typing import Sequence

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    which: str  # 어떤 경로로 분기할지 결정하는 필드

# "cd" 이면 c, d 실행 / 그 외는 b, c 실행
def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

intermediates = ["b", "c", "d"]
graph_builder.add_conditional_edges(
    "a",
    route_bc_or_cd,
    intermediates,
)

라우팅 함수가 리스트를 반환하면 해당 노드들이 병렬로 실행됩니다.

---

7. 조건과 반복

그래프에서 이전 노드로 엣지를 연결하면 반복(루프)을 구현할 수 있습니다.

def route(state: State):
    if len(state["aggregate"]) < 7:
        return "b"  # 계속 반복
    else:
        return END  # 종료

graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "a")  # b → a로 되돌아감 (루프)
graph = graph_builder.compile()

주의: GraphRecursionError

무한 루프를 방지하기 위해 LangGraph는 기본적으로 최대 재귀 횟수를 제한합니다.
제한을 초과하면 GraphRecursionError가 발생합니다.

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, config={"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

config={"recursion_limit": N}으로 최대 반복 횟수를 직접 지정할 수 있습니다.

---

8. 조건에 따른 반복 처리

반복 안에 병렬 실행을 조합하면 더 복잡한 흐름도 만들 수 있습니다.

from typing import Literal

def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "c")
graph_builder.add_edge("b", "d")
graph_builder.add_edge(["c", "d"], "a")  # c, d 모두 끝나면 a로 돌아감
graph = graph_builder.compile()

add_edge(["c", "d"], "a")처럼 리스트를 출발점으로 넣으면,
두 노드 모두 완료된 후에 다음 노드로 이동합니다.

---

9. 사용자 입력에 따른 반복 실행 흐름

사용자의 입력 내용에 따라 대화를 반복하거나 종료하는 패턴입니다.

from langgraph.graph.message import add_messages

class State(TypedDict):
    human_messages: Annotated[list[HumanMessage], add_messages]
    ai_messages: Annotated[list[AIMessage], add_messages]
    retry_num: int

def chatbot(state: State):
    retry_num = state["retry_num"]
    user_input = input(f"(현재 {retry_num}번째 대화) 사용자 입력: ")
    ai_message = AIMessage(f"{retry_num}번째 응답입니다!")
    return {
        "human_messages": [HumanMessage(content=user_input)],
        "ai_messages": [ai_message]
    }

def retry(state: State):
    return {"retry_num": state["retry_num"] + 1}

라우팅 함수에서 마지막 사용자 메시지 내용을 보고 분기합니다.

def route(state: State):
    if "반복" in state["human_messages"][-1].content:
        return "retry"
    else:
        return END

graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", route)
graph_builder.add_edge("retry", "chatbot")
graph = graph_builder.compile()

사용자가 "반복"이라는 단어를 입력하면 계속 대화를 이어가고,
그렇지 않으면 그래프가 종료됩니다.

[ 오늘의 정리 ] – LangGraph 기초 문법 포인트

• StateGraph는 State(TypedDict) + Node(함수) + Edge(연결)의 세 요소로 구성됩니다.
• add_messages 리듀서를 사용하면 메시지를 직접 이어 붙이지 않아도 자동으로 누적됩니다.
• 실행 방식은 invoke(동기) / ainvoke(비동기) / stream(스트리밍)으로 나뉘며,
  stream은 values, updates, messages 세 가지 모드를 지원합니다.
• 팬아웃(Fan-out): 하나의 노드에서 여러 노드로 분기해 병렬 실행하는 패턴입니다.
• add_conditional_edges: 조건에 따라 단일 또는 복수의 노드로 분기할 수 있습니다.
• 루프: 이전 노드로 엣지를 연결하면 반복이 가능하며, recursion_limit으로 최대 횟수를 제어합니다.