LangGraph로 프로덕션 AI 에이전트 만들기 실전 튜토리얼

에이전트를 처음 만들 때 대부분 LangChain 체인으로 시작합니다. 그런데 조금만 복잡해지면 문제가 생깁니다. 툴 호출 실패 시 재시도 로직이 없고, 중간에 크래시 나면 처음부터 다시 돌려야 하고, 사람이 개입할 수 있는 승인 단계를 넣으려면 커스텀 미들웨어를 통째로 짜야 하죠. LangGraph는 이 세 가지 문제를 전부 프레임워크 수준에서 해결합니다.

---

핵심 요약

LangGraph는 2026년 현재 AI 에이전트 프레임워크 중에서 사실상 표준으로 굳어진 도구입니다. 월 3,450만 건 다운로드에 Klarna, Uber, LinkedIn, JPMorgan이 실제 프로덕션에서 돌리고 있고, 2025년 10월 v1.0을 찍으면서 API도 안정화됐습니다.

구조적으로 보면 LangGraph의 핵심 아이디어는 단순합니다. 에이전트 로직을 방향 그래프로 표현하는 건데, 노드는 함수이고 엣지는 노드 간 전이 규칙입니다. 기존 LangChain 체인과 결정적으로 다른 점은 사이클을 지원한다는 것으로, 에이전트가 툴 결과를 보고 판단해서 다른 노드로 돌아가는 진짜 루프 동작이 가능합니다.

상태 관리 측면에서는 TypedDict 또는 Pydantic으로 정의한 State 스키마가 모든 노드를 관통하며 흐르고, 각 노드는 이 상태를 읽고 업데이트해서 반환합니다. 여기에 체크포인터가 붙으면 모든 전이 시점마다 상태가 저장되기 때문에 중간에 파드가 재시작돼도 정확히 멈춘 지점부터 재개할 수 있습니다.

Human-in-the-loop 기능도 강력한 이유 중 하나인데, interrupt_before 설정 단 세 줄로 특정 노드 실행 전에 에이전트를 일시 정지시키고 사람의 승인을 받을 수 있습니다. 이메일 발송이나 DB 수정처럼 되돌리기 어려운 작업 전에 필수입니다.

배포 전략은 환경에 따라 다릅니다. 개발 중에는 MemorySaver, 프로덕션에서는 PostgresSaver를 쓰는 게 표준 패턴이고, LangGraph Cloud를 쓰면 관리형 인프라에 자동 스케일링까지 붙습니다. 토큰 비용은 노드마다 LLM 호출이 발생하므로 리서치 에이전트 기준 쿼리당 10~50K 토큰을 잡아야 합니다.

---

실전 1: 환경 세팅

LangGraph 1.2.0은 2026년 5월 11일에 출시됐으며 Python 3.10~3.14를 완전 지원합니다. 프로젝트를 시작하기 전에 가상환경을 반드시 먼저 만들어야 하는데, LangGraph는 버전 간 의존성 충돌이 CI/CD에서 디버깅하기 까다로운 형태로 터지기 때문입니다.

아래 명령어로 가상환경을 만들고 필요한 패키지를 설치합니다. Tavily는 에이전트가 웹 검색을 할 수 있게 해주는 툴인데, 프리 티어도 있어서 테스트용으로 충분합니다.

mkdir langgraph-agent
cd langgraph-agent
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

pip install langgraph==0.3.34 \
  langchain-openai==0.3.12 \
  langchain-community==0.3.19 \
  tavily-python==0.5.0 \
  python-dotenv==1.1.0

설치가 끝나면 python -c "import langgraph; print(langgraph.__version__)" 로 버전 확인을 해봅니다. 에러 없이 버전 번호가 찍히면 준비 완료입니다.

---

실전 2: StateGraph 기본 구조 잡기

LangGraph 에이전트를 만들 때 가장 먼저 설계해야 하는 건 State 스키마입니다. 이게 모든 노드를 관통하는 공유 메모리 역할을 하고, 여기서 어떤 필드를 정의하느냐에 따라 에이전트가 무엇을 기억하고 무엇을 추적할 수 있는지가 결정됩니다.

아래는 리서치 에이전트의 State 정의와 기본 그래프 구조입니다. Annotated[list, add_messages]는 LangGraph에서 메시지 리스트를 누적 방식으로 업데이트하겠다는 선언입니다.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

# State 스키마 정의
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    search_results: list[str]
    iterations: int
    final_answer: str

# LLM 초기화
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 그래프 초기화
graph = StateGraph(AgentState)

State 스키마를 dict 대신 TypedDict로 쓰는 이유는 타입 힌트 덕분에 어떤 노드에서 어떤 필드를 쓰는지 명확해지기 때문입니다. 필드 하나가 빠지거나 타입이 틀리면 런타임이 아닌 컴파일 단계에서 잡을 수 있습니다.

---

실전 3: 노드와 조건부 엣지 연결

노드는 State를 받아서 업데이트된 State를 반환하는 일반 Python 함수입니다. 중요한 건 return 값이 전체 State가 아니라 변경된 필드만 담은 dict여도 된다는 점으로, LangGraph가 알아서 나머지 필드와 병합해줍니다.

아래 코드는 LLM 판단 노드, 웹 검색 노드, 그리고 두 노드 사이를 조건에 따라 라우팅하는 엣지를 연결하는 전체 흐름입니다.

from langchain_community.tools.tavily_search import TavilySearchResults

search_tool = TavilySearchResults(max_results=3)

# LLM 판단 노드: 검색이 필요한지 아닌지 결정
def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 웹 검색 노드: 실제 검색 수행
def search_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = search_tool.invoke(last_message.content)
    search_texts = [r["content"] for r in results]
    return {
        "search_results": search_texts,
        "iterations": state.get("iterations", 0) + 1
    }

# 라우터: LLM 응답 내용에 따라 다음 노드 결정
def should_search(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "search"
    return "end"

# 그래프에 노드 등록
graph.add_node("agent", agent_node)
graph.add_node("search", search_node)

# 엣지 연결
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_search, {
    "search": "search",
    "end": END
})
graph.add_edge("search", "agent")  # 검색 후 다시 판단

app = graph.compile()

add_conditional_edges가 LangGraph의 핵심인데, 라우터 함수 반환값에 따라 다음 노드를 동적으로 선택합니다. 이 사이클 덕분에 에이전트가 "검색 → 판단 → 검색 → 판단"을 목표 달성까지 반복할 수 있습니다.

---

실전 4: 체크포인팅으로 내구성 확보

프로덕션에서 에이전트가 10단계짜리 리서치 작업을 돌리다가 7단계에서 네트워크 오류로 죽었다면 처음부터 다시 시작해야 합니다. 비용도 비용이지만 이미 완료한 작업을 반복하는 건 사용자 경험에서 치명적입니다. LangGraph 체크포인터는 이 문제를 해결합니다.

개발 단계에서는 MemorySaver, 프로덕션에서는 PostgresSaver를 씁니다. 아래는 두 환경을 환경변수 하나로 전환하는 패턴입니다.

import os
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

def get_checkpointer():
    db_url = os.getenv("DATABASE_URL")
    if db_url:
        return PostgresSaver.from_conn_string(db_url)
    return MemorySaver()

checkpointer = get_checkpointer()
app = graph.compile(checkpointer=checkpointer)

# thread_id를 설정하면 같은 대화 흐름을 유지
config = {"configurable": {"thread_id": "user-session-001"}}

# 첫 번째 실행
result = app.invoke(
    {"messages": [HumanMessage(content="2026년 AI 에이전트 프레임워크 트렌드 분석해줘")]},
    config=config
)

# 이어서 대화 (이전 컨텍스트가 자동으로 유지됨)
result2 = app.invoke(
    {"messages": [HumanMessage(content="LangGraph를 중심으로 더 자세히 알려줘")]},
    config=config
)

thread_id가 같으면 LangGraph가 체크포인터에서 이전 상태를 불러와 이어서 실행합니다. 파드 재시작이나 API 타임아웃이 발생해도 정확히 멈춘 시점부터 재개됩니다.

---

실전 5: Human-in-the-Loop 승인 게이트

에이전트가 이메일을 보내거나 데이터베이스를 수정하는 작업을 한다면 반드시 사람 승인 단계가 필요합니다. LangGraph는 interrupt_before 설정으로 특정 노드 실행 전에 자동으로 일시 정지합니다.

아래는 "send_email" 노드 실행 전 반드시 승인을 받도록 하는 패턴입니다. 체크포인터가 있어야 중단된 상태가 저장되므로 위 실전 4의 PostgresSaver 설정이 전제입니다.

def send_email_node(state: AgentState) -> dict:
    # 실제 이메일 전송 로직
    print(f"이메일 전송: {state['messages'][-1].content}")
    return {"messages": [AIMessage(content="이메일 전송 완료")]}

graph.add_node("send_email", send_email_node)
graph.add_edge("agent", "send_email")

# interrupt_before: send_email 노드 실행 전에 일시 정지
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"]
)

# 실행하면 send_email 직전에 멈춤
config = {"configurable": {"thread_id": "email-task-01"}}
app.invoke(
    {"messages": [HumanMessage(content="팀장님께 주간 보고 이메일 보내줘")]},
    config=config
)

# 현재 상태 확인 후 승인 여부 결정
current_state = app.get_state(config)
print("전송 예정 내용:", current_state.values["messages"][-1].content)

# 승인하면 None을 넣어서 재개
app.invoke(None, config=config)

interrupt_before=["send_email"]만 추가하면 그 외에 별도 미들웨어나 큐가 필요 없습니다. 중단된 상태는 체크포인터에 그대로 저장되니 사람이 며칠 뒤에 승인해도 이어서 실행됩니다.

---

실전 6: FastAPI로 배포

에이전트를 실제 서비스에 붙이려면 API 레이어가 필요합니다. LangGraph는 FastAPI와 조합해서 쓰는 게 가장 일반적인 패턴이고, 스트리밍 응답도 SSE(Server-Sent Events)로 쉽게 노출할 수 있습니다.

아래는 에이전트를 POST 엔드포인트로 노출하고 토큰 스트리밍을 지원하는 최소 FastAPI 서버입니다.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json

api = FastAPI()

class ChatRequest(BaseModel):
    message: str
    thread_id: str

@api.post("/chat")
async def chat(req: ChatRequest):
    config = {"configurable": {"thread_id": req.thread_id}}
    
    async def event_stream():
        async for event in app.astream_events(
            {"messages": [HumanMessage(content=req.message)]},
            config=config,
            version="v2"
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"].content
                if chunk:
                    yield f"data: {json.dumps({'content': chunk})}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(event_stream(), media_type="text/event-stream")

# 실행: uvicorn main:api --host 0.0.0.0 --port 8000

astream_events는 노드 단위가 아닌 토큰 단위로 이벤트를 내보내기 때문에 사용자가 응답 생성 과정을 실시간으로 볼 수 있습니다. 프론트엔드에서는 EventSource API로 이 스트림을 받아서 표시하면 됩니다.

---

마무리

LangGraph가 2026년에 에이전트 프레임워크 경쟁에서 이긴 이유는 가장 화려해서가 아니라 프로덕션에서 터지는 문제들을 직접 해결했기 때문입니다. 상태 내구성, 사이클 지원, Human-in-the-loop, 스트리밍이 전부 프레임워크 안에 있고, FastAPI 한 층만 얹으면 실제 서비스로 바로 올라갑니다.

처음엔 그래프 설계가 낯설게 느껴질 수 있지만 State → 노드 → 엣지 세 개념만 잡으면 나머지는 같은 패턴의 반복입니다. 이 튜토리얼에서 만든 리서치 에이전트를 베이스로 노드를 추가하거나 멀티에이전트 Supervisor 패턴으로 확장해 나가면 Klarna나 Uber 수준의 프로덕션 시스템도 같은 구조 위에서 나옵니다.