LangGraph

에이전트•도구•로직을 묶는 동적 pipeline 설계 도구

LangChain보다 한 단계 낮다 (추상화 단계가)
: create_agent도 내부적으로 LangGraph 위에서 작동
-> 자유도가 높다 (여러 단계로 분리 된 워크플로우, 코드 기반 분기 (LLM 판단 외), 상태 누적•복원•되돌리기)

State•Node•Edge

동적 Pipeline 설계 (2-layer)

State 정의 : DB
Graph 정의 : 로직('Node')/흐름('Edge')
Graph-State 연결

노드 간 직접 인자 전달 없음. 모두 State에 쓰고 읽는다

설계 단계

State 정의 : DB
Node(Function) 설정: 기능 정의 (***)
그래프 초기화 : 아무것도 없는 백지 Graph -> StateGraph(State)
노드•엣지 추가
그래프 완성 : compile()
실행 : invoke(), stream()

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

# 1. State 정의 — DB 설계
class State(TypedDict):
  messages: Annotated[list, add]
  count: int

# 2. Node 설정 — 기능 설계
def greet(state: State):  # input으로 State 무조건 넣어줘야함 ! 
  return {"messages": ["안녕하세요!"], "count": 1}  # return 도 무조건 dict 로 ! 

def farewell(state: State):
  return {"messages": ["다음에 또 봐요."], "count": 1}

# 3. 그래프 초기화 — 빈 깡통 (State 와 연결)
builder = StateGraph(State)

# 4. 노드·엣지 추가 — 조립
builder.add_node("greet", greet)
builder.add_node("farewell", farewell)
builder.add_edge(START, "greet")
builder.add_edge("greet", "farewell")
builder.add_edge("farewell", END)

# 5. 컴파일 — 실행 가능 객체화
graph = builder.compile()

Node로 LLM 포함하기

# 1. State 정의
class ChatState(TypedDict):
    question: str
    answer: str

# 2. Node 정의 — 노드 안에서 LLM 호출
def ask_llm(state: ChatState):
    response = model.invoke(state["question"])
    return {"answer": response.content}

# 3~5. 조립 → 컴파일
builder = StateGraph(ChatState)
builder.add_node("chat", ask_llm)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile()

# 6. 실행
print(graph.invoke({"question": "사과는 영어로 무엇인가요? 한 단어로 답해 주세요."})["answer"])

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

State

Reducer

노드가 반환한 값은 기본적으로 덮어쓰기지만, 대화 메시지처럼 누적해야는 경우도 有

Annotated[타입, reducer] : 값을 어떻게 병합할지 지정 ("이 필드는 이렇게 합쳐라")

Node 가 반환한 값은 바로 State에 들어가지 않음 : Reducer 함수를 거쳐 어떻게 합칠지 결정된다.

덮어쓰기 : 일반 변수로 (default)
추가 : Annotated[list, add]
합산 : Annotated[int, add]
메세지 누적: Annotated[list, add_messages]
커스텀 : Annotated[list, lambda o, n: keep_latest_n(o, n, 5)]

from typing import Annotated
from operator import add

# 1. State 정의 — count에 add reducer를 붙인다 (← 위 셀과 유일한 차이)
class AccumulateState(TypedDict):
    count: Annotated[int, add]

# 2. Node 정의 — 위와 똑같이 5를 반환한다
def add_five(state: AccumulateState):
    return {"count": 5}

# 3~5. 조립 → 컴파일
builder = StateGraph(AccumulateState)
builder.add_node("add5", add_five)
builder.add_edge(START, "add5")
builder.add_edge("add5", END)
graph = builder.compile()

# 6. 실행 — 시작값 10에 5를 반환하면? → 더해서 15
print(graph.invoke({"count": 10}))   # {'count': 15}

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

=> MessageState: messages 필드와 누적 동작이 자동

MessagesState: 메세지를 누적하는 가장 흔한 State (LangGraph가 미리 정의해 둔 표준)

상속만 하면 messages 필드와 누적 동작이 자동
ㄴ 핵심 : 내부적으로 'messages: Annotated[list, add_messages]' 사용
-> 별도 머지 로직 없이 메시지가 시간순으로 쌓이는 표준 패턴
: 단순 문자열 X, 메시지 객체 (HumanMessage, AIMessage)

from langgraph.graph import MessagesState

# 1. State — MessagesState 상속 (messages 필드와 누적 동작이 자동 포함)
class BotState(MessagesState):
    pass

# 2. Node — 누적된 messages 전체를 모델에 전달하고, 답변을 messages에 추가
def chatbot(state: BotState):
    return {"messages": [model.invoke(state["messages"])]}

# 3~5. 조립
builder = StateGraph(BotState)
builder.add_node("bot", chatbot)
builder.add_edge(START, "bot")
builder.add_edge("bot", END)
graph = builder.compile()

# 6. 실행 — 입력 메시지와 답변이 messages에 함께 쌓인다
out = graph.invoke({"messages": [{"role": "user", "content": "안녕하세요, 제 이름은 지수입니다."}]})
for m in out["messages"]:
    print(type(m).__name__, ":", m.content)

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

"""
HumanMessage : 안녕하세요, 제 이름은 지수입니다.
AIMessage : 안녕하세요, 지수님! 만나서 반갑습니다. 어떻게 도와드릴까요?
"""

type(m).name : 메시지 종류
m.content : 메시지 내용

Node

기능, 함수 그 자체

Node의 입출력 계약: "무조건 State로부터 입출력"

입력 : State 전체

def my_node(state: State):
    # State 의 변수들에 접근
    msgs = state["messages"]
    count = state["count"]
    ..

출력 : 딕셔너리 (변경할 필드만)

def my_node(state: State):
    ...
    return {
        "messages": [new_message],
        "count": count + 1,

Edge

어떤 노드 다음에 어떤 노드로 가는지 -> add_edge(from, to) 와 START / END 로 정의

from langgraph.graph import START, END
builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

시각화 (Mermaid)

from IPython.display import Image
Image(
    graph.get_graph().draw_mermaid_png()
)
# 또는 파일로 저장
png_bytes = graph.get_graph().draw_mermaid_png()
with open("graph.png", "wb") as f:
    f.write(png_bytes

분기•Cycle

Conditional Edge : 조건 분기

State 값을 보고 다음 노드를 동적으로 선택

분기 함수(routing function) : add_conditional_edges(시작노드, 분기함수, {반환값: 노드이름})

1) 룰 기반 (코드 분기)

State 값을 코드 로직으로 검사 -> 빠르고 결정적이며 비용 X

class State(TypedDict):
    score: int
def route_by_score(state: State):
    if state["score"] >= 80:
        return "pass"
    return "fail"
builder.add_conditional_edges(
    "evaluate",
    route_by_score,
    {"pass": "celebrate",
     "fail": "retry"},
)

2) LLM 기반

룰로 정의하기 어려운 경우 LLM이 다음 노드를 결정하게 함 -> 자연어•모호한 입력 분류 가능

from pydantic import BaseModel
from typing import Literal
class Intent(BaseModel):
    next: Literal["search", "calc", "chat"]
router_llm = model.with_structured_output(Intent)  # 정해진 후보 중 하나만 나오도록 출력 형식을 강제
def route_by_llm(state: State):
    text = state["messages"][-1].content
    return router_llm.invoke(f"분류: {text}").next
builder.add_conditional_edges("classify", route_by_llm,
    {"search": "search_node",
     "calc": "calc_node",
     "chat": "chat_node"}

Cycle(Loop)

Edge 가 이전 노드로 되돌아가면 사이클이 생긴다. Conditional Edge 가 언제 빠져나갈지 를 결정한다.

필요 1) 되돌아가는 Edge 필요 2) 종료 조건 Conditional Edge

class State(TypedDict):
    draft: str; score: int; iterations: int
def reviewer_node(state):       # ← 카운터 증가
    score = score_llm.invoke(state["draft"])
    return {"score": score,
            "iterations": state["iterations"] + 1}
def reviewer_route(state):
    if state["score"] >= 80 or state["iterations"] >= 3:
        return "end"            # 80점 또는 3회면 종료
    return "rewrite"
builder.add_conditional_edges("reviewer", reviewer_route,
    {"rewrite": "writer", "end": END})
# writer  → reviewer  → (다시 writer | END)

병렬

# Step 1. State 정의 — 결과를 '누적'해야 하므로 add reducer 사용
class FanState(TypedDict):
    topic: str
    notes: Annotated[list, add]   # 여러 노드의 결과가 합쳐진다

# Step 2. 동시에 실행될 노드 세 개 (각자 다른 관점)
def pros_node(state): return {"notes": [f"장점 관점: {state['topic']}의 좋은 점"]}
def cons_node(state): return {"notes": [f"단점 관점: {state['topic']}의 아쉬운 점"]}
def cost_node(state): return {"notes": [f"비용 관점: {state['topic']}의 비용"]}

# Step 3. 노드 추가 — 노드 이름="pros"/"cons"/"cost", 함수는 *_node
builder = StateGraph(FanState)
builder.add_node("pros", pros_node)
builder.add_node("cons", cons_node)
builder.add_node("cost", cost_node)

# Step 4. START에서 세 노드로 동시에 분기(fan-out), 모두 END로(join)
for n in ["pros", "cons", "cost"]:
    builder.add_edge(START, n)
    builder.add_edge(n, END)
graph = builder.compile()

# Step 5. 실행 — 세 결과가 notes에 모두 모인다
result = graph.invoke({"topic": "재택근무", "notes": []})
for n in result["notes"]:
    print("-", n)

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

도구•ToolNode

tool 정의 - @tool

from langchain.tools import tool

# 평범한 함수에 @tool을 붙이면 '도구'가 된다. docstring은 모델이 읽는 설명.
@tool
def get_weather(city: str) -> str:
    """도시 이름을 받아 현재 날씨를 알려준다."""
    weather = {"서울": "맑음, 14도", "부산": "흐림, 17도"}
    return weather.get(city, f"{city}의 날씨 정보가 없습니다.")

# 도구도 결국 함수 — 직접 호출하면 그냥 실행된다 (.invoke로 호출)
print(get_weather.invoke("서울"))

모델에 도구 연결 - bind_tools

# 모델은 '어떤 도구를 어떤 인자로' 부를지만 정한다 (아직 실행 X)
model_with_tools = model.bind_tools([get_weather])
ai = model_with_tools.invoke("서울 날씨 어때?")
print("tool_calls:", ai.tool_calls)

ToolNode — 도구 실행 전용 노드 LangGraph가 제공하는 prebuild 노드로, tool_calls를 읽어 도구를 실행하고, 결과를 ToolMessage로 돌려준다.

모델 → ToolNode → 끝

from langgraph.prebuilt import ToolNode

# 1. State — 메시지 누적
class WeatherState(MessagesState):
    pass

# 2. 노드 두 개
def call_model(state: WeatherState):       # (a) 모델이 도구 호출을 결정
    return {"messages": [model_with_tools.invoke(state["messages"])]}

tool_node = ToolNode([get_weather])         # (b) 도구 실행 전용 노드

# 3. 그래프: START → 모델 → 도구실행 → END  (한 번만, 반복 없음)
builder = StateGraph(WeatherState)
builder.add_node("model", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "model")
builder.add_edge("model", "tools")
builder.add_edge("tools", END)
graph = builder.compile()

# 4. 실행 — 모델의 도구 호출 → ToolNode 실행 결과(ToolMessage)까지 확인
out = graph.invoke({"messages": [{"role": "user", "content": "서울 날씨 어때?"}]})
for m in out["messages"]:
    print(type(m).__name__, ":", m.content)

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

Memory

기본적으로 그래프는 매 호출이 독립이라 State 에 저장되어있는 Data는 날라감 -> checkpointer 사용: 그래프의 모든 노드 사이 state 가 영속화됨 (State를 자동으로 저장•복원)

from langgraph.checkpoint.memory import InMemorySaver

# 1. 간단한 챗봇 그래프 (노드 이름="chat", 함수=chat_node)
class MemState(MessagesState):
    pass

def chat_node(state: MemState):
    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(MemState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

# 2. compile에 checkpointer를 넘기면 State가 자동 저장·복원된다
graph = builder.compile(checkpointer=InMemorySaver())

# 3. thread_id로 대화를 구분 — 같은 thread는 이어진다
cfg = {"configurable": {"thread_id": "user-1"}}
graph.invoke({"messages": [{"role": "user", "content": "내 이름은 지수야. 기억해줘."}]}, cfg)
out = graph.invoke({"messages": [{"role": "user", "content": "내 이름이 뭐라고 했지?"}]}, cfg)
print("같은 thread:", out["messages"][-1].content)

# 그래프 구조 확인
Image(graph.get_graph().draw_mermaid_png())

Navigation Menu

LangGraph

State•Node•Edge

State

Reducer

Node

Edge

시각화 (Mermaid)

분기•Cycle

Conditional Edge : 조건 분기

1) 룰 기반 (코드 분기)

2) LLM 기반

Cycle(Loop)

병렬

도구•ToolNode

Memory

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally