15. LangGraph로 첫 번째 그래프 만들기
Part IV에서 우리는 도구를 정의하고, LLM에 연결하고, 판단과 실행을 반복하는 에이전트 루프를 완성했습니다. 루프를 도는 것, LLM의 판단을 확인하여 도구를 실행하는 것, 멈추는 것 — 이러한 흐름의 모든 부분을 우리가 직접 파이썬 코드로 작성하였습니다.
이번 장에서는 같은 에이전트를 완전히 다른 방식으로 만들어 봅니다. 흐름을 직접 작성하는 대신, 단계(노드)와 연결 규칙(엣지)을 LangGraph 프레임워크에 등록하여 실행을 맡기는 방식입니다. 동작은 14장과 동일하지만, 만드는 방식이 달라집니다.
이 장에서는 LangGraph의 네 가지 핵심 개념인 StateGraph, 노드, 엣지, State(상태)를 배운 다음, 14장의 에이전트 루프를 LangGraph 그래프로 리팩터링합니다. 이어지는 16장에서는 조건부 라우팅과 미리 만들어진 컴포넌트를 활용하는 방법을 배우고, 17장에서는 에이전트가 중단된 지점에서 재개할 수 있게 해주는 상태 영속화를 다룹니다.
15.1) 왜 그래프인가?
15.1.1) 기존 에이전트 루프의 한계
14장에서 만든 에이전트 루프를 다시 살펴보겠습니다. 오류 처리 등 세부 사항을 제거하고 본질만 남기면 다음과 같은 구조였습니다:
# 14장 에이전트 루프의 핵심 구조 (요약)
messages = [
SystemMessage(content="You are a helpful assistant."),
HumanMessage(content=user_input),
]
for step in range(max_steps):
# LLM에게 다음 행동을 판단하게 합니다
ai_message = llm_with_tools.invoke(messages)
messages.append(ai_message)
# 도구 호출 요청이 없으면 최종 답변을 리턴합니다.
if not ai_message.tool_calls:
return ai_message.content
# 요청된 도구를 실행합니다
for tool_call in ai_message.tool_calls:
selected_tool = tool_map[tool_call["name"]]
tool_message = selected_tool.invoke(tool_call)
messages.append(tool_message)이 코드는 간단한 기능만 구현하였고, 그 외에는 아무것도 다루지 않았습니다. 하지만 실제 서비스 환경에서는 훨씬 더 많은 것이 요구됩니다. 몇 가지 예를 들어 보겠습니다.
- 작업 중단 복구 — 10단계짜리 리서치 작업의 7단계에서 에이전트가 크래시되었다면, 처음부터 다시 시작하지 않고 7단계부터 이어서 진행할 수 있어야 합니다.
- 승인 요청 — 에이전트가 중요한 업무를 처리하기 전에, 잠시 멈추고 사람에게 "실행해도 될까요?"라고 확인을 받을 수 있어야 합니다.
- 실시간 모니터링 — 에이전트가 지금 무엇을 하고 있는지, 어떤 도구를 호출하고 있는지를 사용자에게 보여줄 수 있어야 합니다.
- 시각화와 디버깅 — 에이전트가 어떻게 동작하는지 보여주는 다이어그램을 제공하여, 문제가 생겼을 때 어느 단계에서 잘못되었는지 추적할 수 있어야 합니다.
이런 기능을 직접 구현하는 것이 불가능한 것은 아니지만 쉽지도 않습니다. 예를 들어 작업 중단 복구 하나만 해도, 매 단계의 상태를 직렬화하고, 디스크에 저장하고, 복원하고, 정확한 위치에서 재개하는 코드를 전부 직접 작성해야 합니다. 비즈니스 로직보다 인프라 코드가 더 많아질 수 있습니다.
LangGraph는 이러한 기능을 프레임워크 차원에서 제공하기 위해 만들어졌습니다. 작업 중단 복구, 승인 요청, 모니터링, 시각화 — 이런 것들을 프레임워크가 처리해 줍니다. 그런데 프레임워크의 이런 기능을 이용할 수 있으려면 한 가지 조건이 필요합니다. 프레임워크가 이해할 수 있는 구조로 만들어야 합니다.
14장의 에이전트 루프는 로직 전체를 직접 다루고 있어서, LangGraph 프레임워크와 통합할 수 없습니다. 프레임워크의 기능을 활용하려면, 에이전트를 LangGraph가 이해할 수 있는 구조 — 그래프 — 로 다시 만들어야 합니다. 이것이 이번 장에서 할 일입니다.
15.1.2) LangGraph란
LangGraph는 에이전트의 워크플로우를 그래프로 정의하고 실행하는 오케스트레이션 프레임워크입니다. 여기서 그래프란, 에이전트가 수행하는 각 노드(단계)를 엣지(연결 규칙)로 이어 놓은 구조를 말합니다.
LangGraph에서는 전체 작업을 독립된 노드로 분리하고, 노드 사이를 엣지로 연결합니다. 그러면 LangGraph가 연결된 그래프를 따라가며 각 노드를 실행합니다. 14장의 에이전트 루프를 그래프로 표현하면 다음과 같은 모습이 됩니다:
네모 박스가 노드이고, 화살표가 엣지입니다. 마름모는 조건에 따라 다른 경로로 분기하는 조건부 엣지를 나타냅니다.
14장 에이전트 루프에서는 전체 작업을 for문, if문 등의 코드로 직접 구현했습니다. LangGraph에서는 각 노드의 역할과 업무를 정의하고, 노드 사이를 엣지로 연결하면 됩니다. 즉, 작업을 코드로 작성하는 것에서 구조로 선언하는 것으로 바뀌는 것입니다.
참고로, LangGraph는 여러분이 12~14장에서 배운 것을 대체하지 않습니다. 도구 정의, bind_tools(), tool_calls, ToolMessage — 이 모든 것이 노드 안에서 그대로 사용됩니다.
다음 절에서는 LangGraph의 핵심 구성 요소인 StateGraph, State(상태), 노드, 엣지를 하나씩 알아보겠습니다.
15.2) LangGraph 구성 요소: StateGraph, State, 노드, 엣지
이 절에서는 LangGraph의 네 가지 핵심 구성 요소를 하나씩 살펴봅니다. 먼저 State, 노드, 엣지를 하나로 묶어 그래프로 만들어 주는 StateGraph를 소개하고, 그 안에 들어가는 부품들(State, 노드, 엣지)을 순서대로 알아보겠습니다.
15.2.1) StateGraph
StateGraph는 LangGraph에서 그래프를 만들 때 사용하는 클래스입니다. 그래프가 관리할 State를 지정하고, 노드를 추가하고, 노드들을 엣지로 연결한 다음, 컴파일(compile)하면 실행 가능한 그래프가 만들어집니다.
먼저 사용법을 알아보겠습니다.
StateGraph 인스턴스 생성
StateGraph 생성자를 호출하여 인스턴스를 생성합니다. 이때 그래프가 관리할 State 스키마(클래스 자체)를 파라미터로 전달해야 합니다. 여기서는 LangGraph가 기본으로 제공하는 MessagesState를 사용하고 있습니다. 메시지 리스트를 관리하기 위해 미리 정의된 State인데, 자세한 내용은 15.2.2에서 다룹니다.
from langgraph.graph import StateGraph, MessagesState
builder = StateGraph(MessagesState)노드 추가
add_node()로 노드를 추가합니다. 노드는 현재 State를 받아서 변경할 부분을 반환하는 파이썬 함수입니다. 노드 함수에 대한 자세한 내용은 15.2.3에서 다룹니다.
def say_hello(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
builder.add_node(say_hello) # 노드 이름은 "say_hello"엣지 연결
add_edge(출발 노드, 도착 노드)로 노드를 연결합니다. START와 END는 그래프의 시작점과 종료점을 나타내는 특별한 마커입니다. 엣지는 15.2.4에서 다룹니다.
from langgraph.graph import START, END
builder.add_edge(START, "say_hello") # 그래프 시작 → say_hello 실행
builder.add_edge("say_hello", END) # say_hello 완료 → 그래프 종료컴파일과 실행
compile()을 호출하면 그래프 구조가 검증되고, 실행 가능한 객체가 만들어집니다. 컴파일된 그래프는 invoke()로 실행할 수 있습니다. invoke()에는 State의 초기값을 전달합니다.
graph = builder.compile()
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)이제 이들을 하나로 묶어 간단한 그래프를 만들고 실행해 보겠습니다:
from langgraph.graph import StateGraph, MessagesState, START, END
def say_hello(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
builder = StateGraph(MessagesState)
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
graph = builder.compile()
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)출력:
hello worldinvoke()를 호출하면 그래프는 START → say_hello → END 순서로 실행됩니다. say_hello는 messages를 키로 하는 딕셔너리를 반환하였는데, 이 값은 MessagesState의 messages 리스트에 추가됩니다. 이 동작에 대해서는 15.2.2에서 자세히 알아보겠습니다. 결과적으로 마지막 메시지의 내용을 꺼내면 "hello world"가 출력됩니다.
15.2.2) State: 그래프를 흐르는 데이터
State는 그래프의 모든 노드가 공유하는 데이터입니다. 각 노드는 실행될 때 현재 State를 전달받고, 작업을 수행한 뒤 State에서 변경할 부분을 반환합니다. LangGraph가 이 반환값을 State에 반영한 뒤, 다음 노드에 전달합니다.
State 정의
State는 TypedDict를 상속받아 정의합니다. 에이전트가 다루는 업무에 맞게 필드와 타입을 정해서 만들면 됩니다. 아래는 간단한 State 정의 예시입니다:
from typing_extensions import TypedDict
class AgentState(TypedDict):
messages: list # 메시지 리스트
llm_calls: int # LLM 호출 횟수이제 StateGraph 생성 및 각 노드 함수의 입력 파라미터로 AgentState를 사용하면 됩니다.
리듀서
노드가 값을 반환하면, State의 해당 필드가 업데이트됩니다. 이때 기본 동작은 덮어쓰기입니다. 예를 들어 노드가 {"llm_calls": 3}을 반환하면, llm_calls는 기존 값이 뭐였든 3으로 바뀝니다.
그런데 덮어쓰기가 아니라 추가가 필요한 필드도 있습니다. 예를 들어, messages를 덮어쓰면 어떻게 될까요? 그러면 노드가 새 메시지를 반환할 때마다 기존 대화 기록이 모두 사라지게 됩니다. messages는 덮어쓰기가 아니라 추가가 적합합니다.
LangGraph에서는 필드마다 업데이트를 적용하는 방식을 다르게 지정할 수 있는데, 이때 사용하는 것이 리듀서(reducer) 함수입니다. Annotated의 두 번째 인자로 리듀서를 지정합니다:
from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # 리듀서: 추가
llm_calls: int # 리듀서 없음: 덮어쓰기add_messages는 LangGraph가 제공하는 리듀서 함수로, 새 메시지를 기존 리스트에 추가하는 방식으로 필드를 업데이트합니다. messages 필드에 이 리듀서가 지정되어 있으므로, 노드가 messages에 대한 값을 반환하면 기존 리스트에 추가하게 됩니다. llm_calls에는 리듀서가 없으므로, 반환된 값으로 그대로 덮어쓰기됩니다.
15.2.1에서 say_hello가 반환한 메시지가 messages에 "추가"된 것은 바로 이 리듀서 때문입니다.
MessagesState
LangGraph는 MessagesState라는 미리 정의된 State를 제공합니다. 다음과 같이 정의되어 있습니다:
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]앞에서 배운 것과 동일한 구조입니다. add_messages 리듀서가 설정된 messages 필드를 가지고 있습니다.
추가 필드가 필요하면 서브클래싱하여 사용하면 됩니다:
from langgraph.graph import MessagesState
class AgentState(MessagesState):
llm_calls: int # 기본 덮어쓰기 동작15.2.3) 노드: State를 업데이트하는 함수
노드는 그래프 안에서 하나의 구체적인 작업(Task)을 수행하는 파이썬 함수입니다.
def say_hello(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}노드 함수를 작성할 때 알아야 할 두 가지가 있습니다:
규칙 1: 현재 State를 인자로 받는다. LangGraph는 노드 함수를 실행할 때 현재 State 객체를 파라미터로 전달합니다.
규칙 2: 전체 State가 아니라, 변경할 부분만 반환해야 한다. 노드는 State를 직접 변경하지 않습니다. 변경할 필드만 반환하면, LangGraph가 필드별 업데이트 규칙에 따라 기존 State에 병합합니다.
노드를 StateGraph에 추가할 때는 add_node()를 사용합니다:
builder.add_node(say_hello) # 함수 이름 "say_hello"가 노드 이름이 됩니다
builder.add_node("my_node", my_func) # 이름을 직접 지정할 수도 있습니다15.2.4) 엣지: 노드를 연결하는 규칙
엣지는 "이 노드가 끝나면 다음에 무엇을 실행할지"를 결정합니다. 두 종류가 있습니다.
일반 엣지
다음에 실행할 노드를 고정된 형태로 연결합니다. add_edge(source, target) 형태로 사용합니다. source는 출발 노드, target은 도착 노드입니다.
builder.add_edge(START, "say_hello") # 그래프가 시작되면 say_hello를 실행
builder.add_edge("say_hello", "llm_call") # say_hello가 끝나면 llm_call을 실행
builder.add_edge("llm_call", END) # llm_call이 끝나면 그래프 종료조건부 엣지
State를 보고 다음에 실행할 노드를 동적으로 결정합니다. add_conditional_edges(source, routing_function) 형태로 사용합니다. source는 출발 노드, routing_function은 현재 State를 받아서 다음에 실행할 노드의 이름을 반환하는 함수입니다:
from langgraph.graph import END
def should_continue(state: AgentState):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tool_node" # 도구 호출 요청이 있으면 tool_node로
return END # 없으면 종료
builder.add_conditional_edges("llm_call", should_continue)add_conditional_edges("llm_call", should_continue)는 llm_call 노드가 끝나면 should_continue 함수를 호출하여 다음 노드를 결정하라는 의미입니다. should_continue 함수는 마지막 메시지에 tool_calls가 있으면 "tool_node"로, 없으면 END로 라우팅합니다. 즉, LLM 응답에 도구 사용 요청이 있으면 도구를 실행하는 노드로 이동하고, 없으면 그래프를 종료하게 됩니다.
이제 네 가지 구성 요소를 모두 알았습니다. 다음 절에서는 이 구성 요소들을 사용하여 14장의 에이전트 루프를 LangGraph 그래프로 리팩터링하겠습니다.
15.3) 에이전트 루프를 그래프로 리팩터링하기
이제 14장 에이전트 루프를 LangGraph로 다시 만들어 보겠습니다. 동작은 14장과 동일합니다 — LLM이 판단하고, LLM 요청에 따라 도구를 실행하고, 완료될 때까지 반복합니다. 달라지는 것은 이 흐름을 구성하는 방식뿐입니다.
완성할 그래프의 모습은 다음과 같습니다:
그래프는 llm_call과 tool_node 사이를 반복하다가, LLM이 도구 호출을 멈추면 END로 종료합니다. 하나씩 만들어 보겠습니다.
15.3.1) State 정의
15.2.2에서 배운 MessagesState를 서브클래싱하여 에이전트용 State를 정의합니다. messages 필드는 MessagesState에서 상속받고, LLM 호출 횟수를 추적하는 llm_calls 필드를 추가하겠습니다.
from langgraph.graph import MessagesState
class AgentState(MessagesState):
llm_calls: int # LLM 호출 횟수 (덮어쓰기)messages에는 사용자의 입력(HumanMessage), LLM의 응답(AIMessage), 도구 실행 결과(ToolMessage)가 순서대로 쌓이게 됩니다.
15.3.2) 노드 만들기
먼저 14장에서 만든 도구와 모델을 준비합니다:
from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
@tool
def get_weather(city: str) -> str:
"""도시의 현재 날씨를 가져옵니다."""
fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
return fake_data.get(city, f"No weather data for {city}.")
@tool
def calculate(expression: str) -> str:
"""간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
return str(eval(expression)) # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)이제 두 개의 노드 함수를 작성합니다.
llm_call 노드 — LLM을 호출하고 응답을 리턴합니다:
def llm_call(state: AgentState):
"""LLM을 호출하고 응답을 리턴합니다."""
response = model_with_tools.invoke(state["messages"])
return {
"messages": [response],
"llm_calls": state.get("llm_calls", 0) + 1,
}model_with_tools.invoke()로 LLM을 호출하고, 응답을 messages 키에 담아 리턴하고 있습니다. 이 값은 리듀서에 의해 기존 AgentState의 messages 필드에 추가됩니다. llm_calls는 현재 값에 1을 더한 값을 리턴하고 있으며 기존 값을 덮어쓰게 됩니다.
tool_node 노드 — LLM이 요청한 도구를 실행하여 결과를 리턴합니다.
def tool_node(state: AgentState):
"""LLM이 요청한 도구를 실행합니다."""
last_message = state["messages"][-1]
results = []
for tool_call in last_message.tool_calls:
selected_tool = tool_map[tool_call["name"]]
tool_message = selected_tool.invoke(tool_call)
results.append(tool_message)
return {"messages": results}tool_node는 llm_call 다음에 실행되므로, messages의 마지막 메시지는 LLM이 응답한 AIMessage입니다. 이 메시지의 tool_calls에 LLM이 요청한 도구 호출 정보가 담겨 있습니다. 각 도구를 실행한 결과를 results에 모아 messages 키에 담아 리턴하면, 리듀서에 의해 기존 messages에 추가됩니다.
15.3.3) 조건부 엣지
llm_call 노드가 끝난 뒤, tool_node를 실행할지 END로 종료할지를 결정하는 조건부 엣지를 작성합니다. 15.2.4에서 배운 것과 동일한 패턴입니다:
from typing import Literal
from langgraph.graph import END
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
"""도구를 실행할지 그래프를 종료할지 결정합니다."""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tool_node"
return ENDlast_message.tool_calls가 있다는 것은 LLM이 도구 사용을 요청했다는 의미이므로 tool_node로, 없으면 그래프를 종료하도록 END로 라우팅합니다.
반환 타입 힌트
Literal["tool_node", "__end__"]에서 이 함수가 반환할 수 있는 목적지를 명시해주고 있습니다. 이 타입 힌트가 있어야 그래프를 시각화할 때 조건부 엣지 경로가 정확하게 표시됩니다. 이 타입 힌트가 없어도 동작에는 영향을 주지 않습니다.
"__end__"는END의 실제 문자열 값입니다.Literal안에는 문자열만 사용할 수 있으므로END대신"__end__"를 사용합니다.
15.3.4) 그래프 조립과 실행
지금까지 만든 State, 노드, 조건부 엣지를 StateGraph로 조립하고 컴파일하겠습니다:
from langgraph.graph import StateGraph, START, END
builder = StateGraph(AgentState)
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_call") # 시작 → llm_call
builder.add_conditional_edges("llm_call", should_continue) # llm_call → tool_node 또는 END
builder.add_edge("tool_node", "llm_call") # tool_node → llm_call (루프)
agent = builder.compile()tool_node 다음에 다시 llm_call로 이어지므로 루프가 만들어졌습니다. LLM이 도구 호출을 요청하지 않고 최종 응답을 하게 되면 이 루프를 빠져나오게 됩니다.
실행해 보겠습니다:
from langchain_core.messages import HumanMessage
result = agent.invoke({
"messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
"llm_calls": 0,
})
print(result["messages"][-1].content)
print(f"\n총 LLM 호출 횟수: {result['llm_calls']}")출력:
Current temperature in Cairo: 31°C. Multiplied by 3 = 93.
총 LLM 호출 횟수: 3에이전트가 get_weather("Cairo")를 호출하고, 그 결과를 보고 calculate("31 * 3")을 호출한 뒤, 최종 답변을 만들어 냈습니다. 14장에서 테스트한 것과 동일한 결과입니다.
전체 메시지 기록을 확인하면, messages에 모든 단계가 순서대로 쌓여 있는 것을 볼 수 있습니다:
for message in result["messages"]:
message.pretty_print()출력:
================================ Human Message =================================
Get the temperature in Cairo, then multiply the number by 3.
================================== Ai Message ==================================
Tool Calls:
get_weather (call_DiL9WF)
Args:
city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, sunny
================================== Ai Message ==================================
Tool Calls:
calculate (call_wa6RqWST)
Args:
expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
Current temperature in Cairo: 31°C. Multiplied by 3 = 93.15.3.5) 그래프 시각화
Jupyter 노트북에서는 agent.get_graph().draw_mermaid_png()를 사용하여 그래프의 구조를 이미지로 확인할 수 있습니다.
from IPython.display import Image, display
display(Image(agent.get_graph().draw_mermaid_png()))터미널 환경에서는 PNG 파일로 저장하는 방법을 사용할 수 있습니다.
agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")생성된 이미지:
실선은 일반 엣지, 점선은 조건부 엣지입니다. 이 다이어그램은 코드로부터 자동 생성된 것입니다.
15.3.6) 재귀 제한
14장에서 max_steps로 무한 루프를 방지했던 것처럼, LangGraph에도 안전 장치가 내장되어 있습니다. 그래프 실행 중 노드를 통과할 때마다 카운트가 1씩 증가하며, 이 카운트가 설정된 제한을 초과하면 GraphRecursionError가 발생합니다.
예를 들어 앞의 실행 결과를 보면, get_weather와 calculate 두 개의 도구를 호출하는 과정에서 노드를 다음과 같이 통과했습니다:
llm_call(1) → tool_node(2) → llm_call(3) → tool_node(4) → llm_call(5) → END
총 5번의 노드 통과가 일어났습니다. 만약 recursion_limit을 3으로 설정하면, 3번째 노드 통과 시점에서 제한에 걸려 작업이 중단됩니다:
from langgraph.errors import GraphRecursionError
try:
result = agent.invoke(
{"messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
"llm_calls": 0},
config={"recursion_limit": 3},
)
except GraphRecursionError:
print("에이전트가 재귀 제한에 도달했습니다 — 실행을 중단합니다.")출력:
에이전트가 재귀 제한에 도달했습니다 — 실행을 중단합니다.invoke()에 config={"recursion_limit": 숫자}를 전달하여 제한을 설정할 수 있습니다. 비즈니스 성격과 그래프의 복잡도에 따라 적절한 제한 횟수는 다를 수 있습니다. 충분히 여유로운 값으로 설정한 다음, 테스트를 통해 조정하는 것이 좋습니다.