LangGraph
에이전트•도구•로직을 묶는 동적 pipeline 설계 도구
- LangChain보다 한 단계 낮다 (추상화 단계가)
: create_agent도 내부적으로 LangGraph 위에서 작동
-> 자유도가 높다 (여러 단계로 분리 된 워크플로우, 코드 기반 분기 (LLM 판단 외), 상태 누적•복원•되돌리기)
동적 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())노드가 반환한 값은 기본적으로 덮어쓰기지만, 대화 메시지처럼 누적해야는 경우도 有
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의 입출력 계약: "무조건 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,
어떤 노드 다음에 어떤 노드로 가는지 -> 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)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_bytesState 값을 보고 다음 노드를 동적으로 선택
분기 함수(routing function) : add_conditional_edges(시작노드, 분기함수, {반환값: 노드이름})
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"},
)룰로 정의하기 어려운 경우 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"}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())- 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())기본적으로 그래프는 매 호출이 독립이라 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())get_state_history 로 과거 checkpoint 를 불러와 그 시점부터 다시 실행 할 수 있다.
# 과거 state 리스트
history = list(
graph.get_state_history(config)
)
# 특정 시점 picked
target = history[3]
# 그 시점 state 로 다시 실행
graph.invoke(
None,
config=target.config,사용 시나리오
- 디버깅 — 실패 시점 직전부터 재실행하며 원인 추적
- 실험 — 동일 분기에서 다른 결정 을 시도
- 감사 (audit) — 어떤 상태에서 어떤 결정이 났는지 추적
- 회복 — 잘못된 분기를 되돌려 다시 진행
그래프가 영속 저장 되어 있어야 가능 (checkpointer 필수).
interrupt_before / interrupt_after 로 특정 노드 전후에 실행을 멈춰 사람의 개입을 받는다.
# tools 노드 — 도구 실행 담당
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([...]))
# ... edges 생략
# 'tools' 노드 직전에 실행 멈춤
graph = builder.compile(
checkpointer=InMemorySaver(),
interrupt_before=["tools"],프롬프트가 커지면 Lost-in-the-Middle. 해결 = 프롬프트를 나눠 여러 에이전트가 분담 = MAS
MAS 안의 각 LLM 노드가 같은 모델일 필요는 없다. Task 복잡도에 맞춰 비싼/저렴한 모델을 섞는다.
# 조율·판단 — 비싼 LLM
supervisor_model = ChatOpenAI(
model="gpt-5"
)
# 단순 분류·요약 — 저렴 LLM
router_model = ChatOpenAI(
model="gpt-5-mini"
)
# 워커 (도구 호출) — 중간
worker_model = ChatOpenAI(
model="gpt-5-mini"
)Supervisor — 고수준 라이브러리
--> langgraph-supervisor 패키지로 Supervisor 그래프 를 한 줄에 생성. 워커는 M2 의 create_agent 그대로 재사용
from langgraph_supervisor import create_supervisor
# 워커 — name 필수
search_agent = create_agent(model, tools=[web_search],
name="search_expert", system_prompt="검색 전문가.")
calc_agent = create_agent(model, tools=[calculator],
name="calc_expert", system_prompt="계산 전문가.")
# Supervisor 그래프
app = create_supervisor(
agents=[search_agent, calc_agent],
model=supervisor_model,
system_prompt="조율자. 적절한 워커에 위임.",
).compile()직접 짜면 세부 제어가 가능하다
ef supervisor(state):
d = sup_llm.invoke(state["messages"])
return {"next": d.next, "messages": [...]}
def worker_a(state):
out = agent_a.invoke({"messages": state["messages"]})
return {"messages": out["messages"]}
g = StateGraph(State)
g.add_node("sup", supervisor)
g.add_node("a", worker_a); g.add_node("b", worker_b)
g.add_edge(START, "sup")
g.add_conditional_edges("sup", lambda s: s["next"],
{"a": "a", "b": "b", "done": END})
g.add_edge("a", "sup"); g.add_edge("b", "sup")
app = g.compile()그래프 객체를 다른 그래프의 노드로 삽입. 여기서는 상위·하위가 같은 State 를 공유 하는 가장 단순한 형태.
class State(TypedDict): # 공통
messages: Annotated[list, add_messages]
topic: str
search_graph = build_search(State)
analysis_graph = build_analysis(State)
top = StateGraph(State)
top.add_node("search", search_graph)
top.add_node("analysis", analysis_graph)
top.add_edge(START, "search")
top.add_edge("search", "analysis")
top.add_edge("analysis", END)
app = top.compile()원칙 — 한 StateGraph 인스턴스 = 한 State 스키마. 예외 없음.
--> 다른 State 가 필요하면 별도 그래프로 만들고 subgraph 로 부착한다.
여러 State = 여러 subgraph 의 합