[AI Agent 구현하기] LangGraph에 대해서 알아보자 - 기초편

이전 글에서는 LangChain에 대해 알아봤습니다. 이번에는 AI Agent의 실행 흐름을 그래프 형태로 정의하고 관리할 수 있도록 지원하는 프레임워크인 LangGraph에 대해서 알아보겠습니다.

LangChain에 대한 내용이 궁금하다면 아래 글을 참고 부탁드립니다~!

[AI Agent 구현하기] LangChain에 대해서 알아보자

LangGraph란?

LangGraph는 LLM을 기반으로 AI Agent의 워크플로우를 정의하기 위한 오픈소스 프레임워크입니다. Agent를 그래프(Graph) 구조로 표현하여 각 단계(Node)의 실행 순서와 분기 로직을 정의할 수 있고, 각 노드가 공유하는 데이터를 State를 통해 관리합니다.

LangGraph의 핵심 Component

LangGraph는 이름 그대로 에이전트를 그래프(Graph) 형태로 모델링합니다. 그래프는 크게 세 가지 요소로 구성됩니다.

1. State : 그래프 실행 중 공유되는 데이터들의 저장소로, 보통 TypedDict 또는 Pydantic 모델로 정의합니다
2. Node : 실제 작업을 수행하는 Python 함수로, State를 받아 연산을 수행한 뒤 변경된 State를 반환합니다.
3. Edge : 현재 상태를 기반으로 다음에 실행할 노드를 결정합니다.

State, Node, Edge가 실제 LangGraph에서 어떻게 사용되는지는 아래의 코드 예시를 보면서 확인해보면 되겠습니다.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class ExampleState(TypedDict):
    text: str

def node(state: ExampleState) -> ExampleState:
    return {"text": "만나서 반갑습니다."}

# ExampleState라는 State를 공유하는 Graph 정의.
builder = StateGraph(ExampleState)

# Graph에 Node 추가
builder.add_node("node", node)

# Node들을 Edge로 연결
builder.add_edge(START, "node")
builder.add_edge("node", END)
graph = builder.compile()

# ExampleState 초기값을 설정하여 Graph 테스트
graph.invoke({"text": "안녕하세요"})

코드에 사용된 StateGraph는 주요 Graph 클래스로 사용자 정의 State 객체를 매개변수로 받습니다. 그래프는 반드시 compile() 메서드를 통해 컴파일한 후 사용해야 하며, 컴파일 시 그래프 구조 검증, 연결 관계 검증 등이 수행됩니다.

LangGraph 조건부 엣지(conditional edge)

위의 Graph는 Start → Node → END로 간단하게 끝나는 구조입니다. 만약에 A 상황이면 B 노드로 가고, 그렇지 않으면 C 노드로 가도록 그래프를 그려야할 때는 Conditional Edge라는게 사용됩니다.

LangGraph에서는 add_conditional_edges라는 메서드를 사용하여 조건에 따라 특정 노드로 이동시킬 수 있습니다.

graph_builder.add_conditional_edges(
    source="current_node",
    path=routing_function,
    path_map={"route_a": "next_node_1", "route_b": "next_node_2"} 
)

- source : 조건 검사를 시작할 출발 노드.

- path : 다음에 실행할 노드와 관련된 값을 반환하는 라우팅 함수.

- path_map : 라우팅 함수가 반환한 값과 실행할 노드를 매핑하는 딕셔너리 (생략시 라우팅 함수에서 반환한 값을 노드로 취급) 

State Reducer

기본적으로 LangGraph는 노드가 반환한 값을 기존 State에 덮어쓰기(override) 합니다. 이전 ExampleState에서 처음에 있던 "안녕하세요"라는 state값이 "만나서 반갑습니다"로 변경된 이유도 state 값을 덮어쓰기 때문입니다. 하지만 Agent를 구현할 때에는 이전 데이터들을 계속 누적해야 하는 State값이 있을 수 있습니다. 이를 위해 LangGraph에서는 Reducer를 이용합니다.
Reducer는 새로운 상태 값을 기존 상태와 어떤 방식으로 병합할지 정의하는 함수입니다. 대표적으로 대화 이력을 누적하기 위해 add_messages Reducer를 사용할 수 있습니다. 또한 사용자가 직접 Reducer를 정의하여 중복 데이터를 제거하거나, 데이터를 정렬하거나, 특정 규칙에 따라 상태를 병합하는 등 원하는 방식으로 State를 관리할 수 있습니다.

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

class State(TypedDict):
    # Annotated[타입, 리듀서함수] 형식으로 리듀서 지정
    messages: Annotated[list, add_messages]

Tool 실행하기

AI Agent는 일반적으로 LLM이 Tool Calling 요청을 생성하고, Tool 실행 결과를 전달받아 다음 행동을 결정하는 반복적인 루프 구조로 동작합니다. 이를 위해서는 tool_calls를 생성하는 Node와 실제 Tool을 실행하는 Node가 필요합니다.
LangGraph에서는 실제 Tool을 실행하는 Node를 크게 두 가지 방식으로 구현할 수 있습니다.

  1. Tool 실행 Node를 직접 함수 형태로 구현하는 방식
  2. LangGraph에서 제공하는 ToolNode를 사용하는 방식

이번 글에서는 ToolNode를 사용하는 방식에 대해 설명해보겠습니다.

LangGraph에서 제공하는 ToolNode 사용하기

Tool 실행 Node를 직접 구현하는 경우에는 LLM이 생성한 Tool Call을 실행하고, 실행 결과를 State에 반영하는 로직을 작성해야합니다. 직접 구현하는 방식은 Tool 실행 과정을 제어할 수 있다는 장점이 있지만, Tool 호출 처리, 에러 처리, 병렬 실행 등의 로직을 개발자가 직접 구현해야 하는 부담이 있습니다.
그래서 LangGraph는 Tool 실행을 위한 사전 구현(prebuilt) 컴포넌트인 ToolNode를 제공합니다.

from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import MessagesState, StateGraph

@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

builder = StateGraph(MessagesState)
builder.add_node("tools", ToolNode([search, calculator]))
# ... add other nodes and edges
graph = builder.compile()

ToolNode에 실행 가능한 Tool 목록을 전달하면, 가장 최근의 AIMessage에 포함된 tool_calls 정보를 기반으로 해당 Tool을 자동으로 실행합니다.