LangGraph 2.0 실전 가이드 — 그래프 기반 AI 에이전트, 체크포인팅, MCP 연동, 프로덕션 배포

LangGraph 2.0 기본 에이전트 구조 (Python)
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage

# 1. State 정의
class AgentState(TypedDict):
    messages: list
    next_action: str

# 2. Node 정의
llm = ChatAnthropic(model="claude-opus-4-6")

def call_llm(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def route(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tool_node"
    return END

# 3. Graph 구성
builder = StateGraph(AgentState)
builder.add_node("llm", call_llm)
builder.set_entry_point("llm")
builder.add_conditional_edges("llm", route)

# 4. 컴파일 및 실행
graph = builder.compile()
result = graph.invoke({"messages": [HumanMessage("현재 날씨를 알려줘")]})

PostgreSQL 체크포인터 설정 (프로덕션)
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import asyncpg

# 연결 풀 생성 (프로덕션: min 5, max 20 권장)
async def create_graph_with_checkpointer():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost:5432/langgraph",
        min_size=5,
        max_size=20
    )
    checkpointer = AsyncPostgresSaver(pool)
    await checkpointer.setup()  # 체크포인트 테이블 자동 생성

    graph = builder.compile(checkpointer=checkpointer)
    return graph

# 실행: thread_id로 세션 구분
config = {"configurable": {"thread_id": "user-session-abc123"}}
result = await graph.ainvoke(
    {"messages": [HumanMessage("이전 작업 이어서 진행해줘")]},
    config=config
)

# 이전 상태 조회
state = await graph.aget_state(config)
print(state.values)  # 현재 저장된 상태 확인

Human-in-the-Loop 승인 패턴
from langgraph.types import Command, interrupt

def risky_action(state: AgentState) -> AgentState:
    # 실행 전 사람에게 확인 요청
    user_approval = interrupt({
        "message": f"다음 작업을 실행합니다: {state['next_action']}. 승인하시겠습니까?",
        "action_details": state["next_action"]
    })
    
    if user_approval["approved"]:
        # 승인된 경우 실제 작업 실행
        result = execute_action(state["next_action"])
        return {"messages": state["messages"] + [{"role": "tool", "content": result}]}
    else:
        # 거부된 경우 중단
        return {"messages": state["messages"] + [{"role": "system", "content": "작업이 취소되었습니다."}]}

# 그래프 컴파일 시 중단점 지정
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["risky_action"]  # 이 노드 전에 자동 중단
)

# 에이전트 실행 → 중단점에서 일시 정지
result = await graph.ainvoke(inputs, config)
# result["__interrupt__"] 에 중단 정보 포함

# 사람이 승인 후 재개
await graph.aupdate_state(config, {}, as_node="risky_action")
final = await graph.ainvoke(Command(resume={"approved": True}), config)

LangGraph 2.0 + MCP 서버 연동
from langgraph.mcp import MCPClient
from langgraph.prebuilt import create_react_agent

# HTTP MCP 서버 연결 (프로덕션)
async def build_mcp_agent():
    async with MCPClient(
        url="https://your-mcp-server.com/mcp",
        transport="http"
    ) as mcp_client:
        # MCP 서버에서 툴 목록 자동 로드
        tools = await mcp_client.get_tools()
        
        # ReAct 에이전트에 MCP 툴 장착
        agent = create_react_agent(
            model="claude-opus-4-6",
            tools=tools,
            checkpointer=checkpointer
        )
        
        config = {"configurable": {"thread_id": "mcp-session-001"}}
        result = await agent.ainvoke(
            {"messages": [HumanMessage("GitHub에서 최근 이슈 10개를 요약해줘")]},
            config=config
        )
        return result

# stdio MCP 서버 연결 (로컬 개발)
from langgraph.mcp import StdioMCPClient

mcp = StdioMCPClient(command=["npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"])
tools = await mcp.get_tools()

슈퍼바이저 멀티 에이전트 패턴
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class TeamState(TypedDict):
    task: str
    research_result: str
    code_result: str
    final_answer: str

# 서브 에이전트 노드
def research_agent(state: TeamState) -> TeamState:
    """검색·조사 전문 에이전트"""
    result = research_llm.invoke([HumanMessage(f"조사 태스크: {state['task']}")])
    return {"research_result": result.content}

def code_agent(state: TeamState) -> TeamState:
    """코드 작성 전문 에이전트"""
    result = coding_llm.invoke([
        HumanMessage(f"코드 작성 태스크: {state['task']}\n참고 자료: {state['research_result']}")
    ])
    return {"code_result": result.content}

def supervisor(state: TeamState) -> TeamState:
    """결과 통합 및 최종 답변 생성"""
    result = supervisor_llm.invoke([
        HumanMessage(f"조사 결과: {state['research_result']}\n코드: {state['code_result']}\n이 내용을 통합해서 최종 답변을 작성해줘")
    ])
    return {"final_answer": result.content}

# 그래프 구성
builder = StateGraph(TeamState)
builder.add_node("research", research_agent)
builder.add_node("code", code_agent)
builder.add_node("supervisor", supervisor)

builder.set_entry_point("research")
builder.add_edge("research", "code")  # 조사 → 코딩
builder.add_edge("code", "supervisor")  # 코딩 → 슈퍼바이저
builder.add_edge("supervisor", END)

team_graph = builder.compile(checkpointer=checkpointer)

LangGraph 2.0 가드레일 설정
from langgraph.guardrails import ContentFilter, RateLimiter, AuditLogger

# 콘텐츠 필터 — 유해 콘텐츠 차단
content_filter = ContentFilter(
    blocked_categories=["violence", "hate_speech", "personal_info"],
    on_violation="reject"  # 또는 "sanitize"
)

# 속도 제한 — 사용자당 요청 수 제한
rate_limiter = RateLimiter(
    requests_per_minute=10,
    requests_per_day=100,
    key_fn=lambda state: state["user_id"]
)

# 감사 로그 — 규정 준수 목적
audit_logger = AuditLogger(
    log_level="all",  # "tool_calls", "state_changes", "all"
    storage="postgresql",
    retention_days=90
)

# 가드레일 적용
graph = builder.compile(
    checkpointer=checkpointer,
    guardrails=[content_filter, rate_limiter, audit_logger]
)

# langgraph.json — LangGraph Platform 배포 설정
# {
#   "dependencies": ["."],
#   "graphs": {"agent": "./agent.py:graph"},
#   "env": ".env"
# }