[LangChain] LangGraph로 Agent 워크플로우 구현하기 — State, Node, Edge, Router

1. LangGraph

1) 개념

• 상태 기반의 그래프 구조로 복잡한 AI Agent 워크플로우를 설계하고 구현하기 위한 프레임워크
• LangChain이 Prompt, Model, Tool, Retriever 등을 연결하는 데 초점이 있다면, LangGraph는 상태 관리, 조건 분기, 반복 실행, 멀티 에이전트 흐름을 명확하게 제어하는 데 적합하다.

 

2) 핵심 구성요소

구성요소	설명
State	그래프 전체에서 공유되는 데이터 저장소
Node	State를 입력받아 작업을 수행하고 변경할 State를 반환하는 함수
Edge	노드 간 실행 순서를 정의하는 연결
Router	조건부 엣지에서 다음 노드를 결정하는 함수
Compile	정의한 그래프를 실행 가능한 객체로 변환하는 단계

 

3) 기본 구조

• 모든 LangGraph는 START에서 시작한다.
• 하나 이상의 노드를 거친 뒤 END에 도달하면 실행이 종료된다.
• 조건부 엣지를 사용하면 분기와 루프를 구현할 수 있다.

START → Node A → Node B → END

 

4) 사용하는 이유

• 복잡한 워크플로우를 단계별로 제어할 수 있다.
• 조건에 따라 다른 노드로 이동하거나 반복 실행할 수 있다.
• 실행 흐름이 노드와 엣지로 나뉘어 있어 디버깅하기 쉽다.
• Tool Calling, RAG, 검증, 재시도, 사람 승인, 멀티 에이전트 같은 패턴을 구현하기 좋다.

 

2. State

1) 개념

• 그래프 전체에서 공유되는 데이터 저장소
• Agent가 작업을 수행하는 과정에서 발생하는 중간 결과와 상태 변화는 State를 통해 관리된다.
• 각 노드는 State를 읽고, 필요한 값만 수정하여 반환한다.
• State는 TypedDict, dataclass, Pydantic BaseModel 등으로 정의할 수 있다.

from typing import TypedDict, List

class State(TypedDict):
    messages: List[str]
    context: str
    step: int

 

2) State 사용 시 주의점

• 노드는 State 전체를 반환하지 않아도 되고 변경된 필드만 반환하면 LangGraph가 기존 State에 반영한다.
• messages처럼 값이 누적되는 필드는 기존 값에 새 값을 더해 반환하거나, reducer를 사용할 수 있다.

return {
    "messages": state["messages"] + [new_message]
}

 

3) Reducer

• 노드가 반환한 값을 기존 State에 어떻게 반영할지 정하는 함수
• reducer를 지정하지 않으면 기본적으로 새 값이 기존 값을 덮어쓴다.
• messages, logs처럼 값을 계속 누적해야 하는 필드는 reducer를 사용한다.

from typing import Annotated, TypedDict
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

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

• add_messages는 기존 메시지를 유지하면서 새 메시지를 추가하는 대표적인 Reducer이다

 

3. Node

• State를 입력받아 특정 작업을 수행하고, 업데이트할 State를 반환하는 함수
• LLM 호출, Tool 실행, RAG 검색, 응답 검증, 데이터 변환 등을 Node로 만들 수 있다.
• 하나의 노드는 하나의 역할만 담당하는 것이 좋다.

def my_node(state: State) -> State:
    result = llm.invoke(state["messages"][-1])
    return {"messages": state["messages"] + [result]}

 

4. Edge와 Router

1) Edge

• 노드 간 실행 순서 정의
• 일반 엣지 : 항상 정해진 다음 노드로 이동한다
• 조건부 엣지 : State를 기준으로 다음 노드를 분기한다

# 일반 엣지
graph.add_edge("node_a", "node_b")

# 조건부 엣지
def route(state: State):
    if state["step"] > 0:
        return "node_b"
    return "node_c"

graph.add_conditional_edges(
    "node_a",
    route,
    {
        "node_b": "node_b",
        "node_c": "node_c"
    }
)

 

2) Router

• 조건부 엣지에서 다음에 실행할 노드를 결정하는 함수
• State를 읽고 다음 노드 이름을 문자열로 반환한다.
• Router는 실제 작업을 수행하는 노드가 아니다.

구분	노드	라우팅 함수
역할	실제 작업 수행	다음 노드 결정
입력값	State	State
반환값	업데이트할 State 딕셔너리	다음 노드 이름 문자열
등록 방법	add_node()	add_conditional_edges()

 

6. Compile

• 그래프를 정의한 뒤 compile()을 호출하면 실행 가능한 런타임 객체가 생성된다.

graph = builder.compile()

• compile 단계에서 checkpointer를 연결하면 상태 복원과 메모리 기능을 사용할 수 있다.

graph = builder.compile(checkpointer=memory)

 

7. LangGraph 구현 흐름

1) 전체 흐름

상태 정의 → StateGraph 생성 → 노드 생성 → 엣지 연결 → 컴파일 → 실행

2) 단계별 설명

• 상태 정의: TypedDict, dataclass, BaseModel 등으로 State 구조를 정의한다.
• StateGraph 생성: State를 기반으로 그래프 builder를 만든다.
• 노드 생성: State를 입력받고 업데이트할 State를 반환하는 함수를 정의한다.
• 엣지 연결: START, 노드, END 사이의 실행 흐름을 연결한다.
• 컴파일: builder.compile()을 호출하여 실행 가능한 그래프로 변환한다.
• 실행: graph.invoke()를 호출하고 최종 State를 반환받는다.

 

8. LangGraph 구현 예시 - 일반 엣지

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

class State(TypedDict):
    question: str
    answer: str

llm = ChatOpenAI(model="gpt-4o-mini")
builder = StateGraph(State)

def chatbot_node(state: State) -> State:
    response = llm.invoke([("user", state["question"])])
    return {"answer": response.content}

builder.add_node("chatbot_node", chatbot_node)
builder.add_edge(START, "chatbot_node")
builder.add_edge("chatbot_node", END)

graph = builder.compile()
result = graph.invoke({"question": "Agent란 무엇인가", "answer": ""})
print(result["answer"])

 

9. LangGraph 구현 예시 - 조건부 엣지

1) 구조

• 질문 유형을 분류한 뒤, 날씨 질문이면 weather_node, 그 외는 chatbot_node로 분기한다

START → classify_node → weather_node 또는 chatbot_node → END

 

2) 전체 코드

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

class State(TypedDict):
    question: str
    question_type: str
    answer: str

llm = ChatOpenAI(model="gpt-4o-mini")

def classify_node(state: State) -> State:
    question = state["question"]
    question_type = "weather" if "날씨" in question or "기온" in question else "chat"
    return {"question_type": question_type}

def weather_node(state: State) -> State:
    response = llm.invoke([
        ("system", "당신은 날씨 질문에 답변하는 assistant입니다."),
        ("user", state["question"])
    ])
    return {"answer": response.content}

def chatbot_node(state: State) -> State:
    response = llm.invoke([
        ("system", "당신은 친절한 AI assistant입니다."),
        ("user", state["question"])
    ])
    return {"answer": response.content}

def route_by_type(state: State) -> Literal["weather_node", "chatbot_node"]:
    return "weather_node" if state["question_type"] == "weather" else "chatbot_node"

builder = StateGraph(State)
builder.add_node("classify_node", classify_node)
builder.add_node("weather_node", weather_node)
builder.add_node("chatbot_node", chatbot_node)

builder.add_edge(START, "classify_node")
builder.add_conditional_edges("classify_node", route_by_type, {
    "weather_node": "weather_node",
    "chatbot_node": "chatbot_node"
})
builder.add_edge("weather_node", END)
builder.add_edge("chatbot_node", END)

graph = builder.compile()
result = graph.invoke({"question": "오늘 서울 날씨 어때?", "question_type": "", "answer": ""})
print(result["answer"])

 

10. 조건부 엣지와 루프

1) 개념

• 조건부 엣지를 사용하면 특정 조건을 만족할 때까지 이전 노드로 돌아가는 루프를 만들 수 있다.
• 답변 검증, 재시도, 검색 결과 보완, Tool 재실행 같은 흐름에 사용할 수 있다.
• 루프를 만들 때는 반드시 종료 조건이 있어야 한다.

generate_node → validate_node → 다시 generate_node 또는 END

 

2) 언제 사용하는가

• 답변이 조건을 충족하지 못하면 다시 생성해야 할 때
• 검색 결과가 부족하면 다시 검색해야 할 때
• 외부 API 호출이 실패하면 재시도해야 할 때
• 사람이 승인할 때까지 대기하는 Human-in-the-loop 패턴을 구현할 때

 

3) 구현 예시

• 답변 길이가 너무 짧으면 다시 생성하는 루프 예시다.

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

class State(TypedDict):
    question: str
    answer: str
    retry_count: int

llm = ChatOpenAI(model="gpt-4o-mini")

def generate_node(state: State) -> State:
    response = llm.invoke([("user", state["question"])])
    return {
        "answer": response.content,
        "retry_count": state["retry_count"] + 1
    }

def validate_node(state: State) -> State:
    return state

def route_validate(state: State):
    # 답변이 너무 짧거나 재시도 횟수가 3회 미만이면 다시 생성
    if len(state["answer"]) < 50 and state["retry_count"] < 3:
        return "generate_node"
    return END

builder = StateGraph(State)
builder.add_node("generate_node", generate_node)
builder.add_node("validate_node", validate_node)

builder.add_edge(START, "generate_node")
builder.add_edge("generate_node", "validate_node")
builder.add_conditional_edges("validate_node", route_validate, {
    "generate_node": "generate_node",
    END: END
})

graph = builder.compile()

result = graph.invoke({
    "question": "LangGraph란 무엇인가?",
    "answer": "",
    "retry_count": 0
})

print(result["answer"])
print(f"재시도 횟수: {result['retry_count']}")

 

4) 주의점

• 종료 조건이 없으면 무한 루프가 발생한다
• retry_count처럼 최대 반복 횟수를 State에 저장하고 종료 조건에 포함시키는 것이 좋다
• 루프가 깊어질수록 LLM 호출 비용이 증가하므로 적절한 상한을 설정