랭그래프로 사고하기 (Thinking in LangGraph)

랭그래프로 사고하기 (Thinking in LangGraph)

랭그래프(LangGraph)로 에이전트를 만드는 사고법을 배워봐요

LangGraph로 에이전트를 만들 때, 먼저 작업을 노드(node) 라는 개별 단계로 쪼개요. 그리고 각 노드에서 일어나는 여러 결정과 전환을 설명하죠. 마지막으로 각 노드가 읽고 쓸 수 있는 공유 상태(state) 를 통해 노드들을 서로 연결합니다.

이번 실습에서는 LangGraph로 고객 지원 이메일 에이전트를 만드는 사고 과정을 함께 따라가 볼게요.

자동화하고 싶은 프로세스부터 시작해요

고객 지원 이메일을 처리하는 AI 에이전트를 만들어야 한다고 상상해 보세요. 제품팀이 이런 요구사항을 주었어요:

The agent should:

- Read incoming customer emails
- Classify them by urgency and topic
- Search relevant documentation to answer questions
- Draft appropriate responses
- Escalate complex issues to human agents
- Schedule follow-ups when needed

Example scenarios to handle:

1. Simple product question: "How do I reset my password?"
2. Bug report: "The export feature crashes when I select PDF format"
3. Urgent billing issue: "I was charged twice for my subscription!"
4. Feature request: "Can you add dark mode to the mobile app?"
5. Complex technical issue: "Our API integration fails intermittently with 504 errors"

LangGraph로 에이전트를 구현할 때는 보통 같은 다섯 단계를 따르게 돼요.

Step 1: 워크플로를 개별 단계로 나눠요

먼저 프로세스 안에서 뚜렷하게 구분되는 단계를 찾아요. 각 단계는 노드(한 가지 일만 하는 함수)가 되고, 이 단계들이 어떻게 연결되는지 스케치해 봅니다.

flowchart TD
    A[START] --> B[Read Email]
    B --> C[Classify Intent]

    C -.-> D[Doc Search]
    C -.-> E[Bug Track]
    C -.-> F[Human Review]

    D --> G[Draft Reply]
    E --> G
    F --> G

    G -.-> H[Human Review]
    G -.-> I[Send Reply]

    H --> J[END]
    I --> J[END]

    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    class A,B,C,D,E,F,G,H,I,J process

이 다이어그램의 화살표는 가능한 경로를 보여주지만, 실제로 어느 경로를 탈지는 각 노드 내부에서 결정돼요.

이제 워크플로의 구성요소를 파악했으니 각 노드가 무엇을 해야 하는지 살펴볼게요:

• Read Email: 이메일 내용을 추출하고 파싱해요
• Classify Intent: LLM으로 긴급도와 주제를 분류하고, 적절한 동작으로 라우팅해요
• Doc Search: 지식 베이스에서 관련 정보를 조회해요
• Bug Track: 추적 시스템에 이슈를 만들거나 갱신해요
• Draft Reply: 적절한 응답을 생성해요
• Human Review: 승인이나 처리를 위해 인간 에이전트에게 에스컬레이션해요
• Send Reply: 이메일 응답을 발송해요

여기서 어떤 노드는 다음에 어디로 갈지 스스로 결정하고(Classify Intent, Draft Reply, Human Review), 어떤 노드는 항상 같은 다음 단계로 진행한다는 걸 눈여겨보세요(Read Email은 항상 Classify Intent로, Doc Search는 항상 Draft Reply로).

Step 2: 각 단계가 무엇을 해야 하는지 파악해요

그래프의 각 노드에 대해, 그 노드가 어떤 종류의 연산을 나타내는지, 그리고 제대로 동작하려면 어떤 컨텍스트가 필요한지 정해요.

LLM 단계 — 이해, 분석, 텍스트 생성, 추론 결정이 필요할 때 사용해요

데이터 단계 — 외부 소스에서 정보를 가져와야 할 때 사용해요

액션 단계 — 외부 동작을 수행해야 할 때 사용해요

사용자 입력 단계 — 인간의 개입이 필요할 때 사용해요

LLM 단계

단계가 이해, 분석, 텍스트 생성, 또는 추론 결정을 해야 한다면:

Classify intent (의도 분류)

• 정적 컨텍스트(프롬프트): 분류 범주, 긴급도 정의, 응답 형식
• 동적 컨텍스트(상태에서): 이메일 내용, 발신자 정보
• 기대 결과: 라우팅을 결정하는 구조화된 분류

Draft reply (답변 초안)

• 정적 컨텍스트(프롬프트): 톤 가이드라인, 회사 정책, 응답 템플릿
• 동적 컨텍스트(상태에서): 분류 결과, 검색 결과, 고객 이력
• 기대 결과: 검토 준비가 된 전문적인 이메일 응답

데이터 단계

단계가 외부 소스에서 정보를 가져와야 한다면:

Document search (문서 검색)

• 파라미터: 의도와 주제로 만든 쿼리
• 재시도 전략: 예 — 일시적 장애에는 지수 백오프(exponential backoff) 사용
• 캐싱: 공통 쿼리를 캐싱해 API 호출을 줄일 수 있어요

Customer history lookup (고객 이력 조회)

• 파라미터: 상태에서 오는 고객 이메일 또는 ID
• 재시도 전략: 예 — 단, 정보를 구할 수 없으면 기본 정보로 폴백
• 캐싱: 예 — 신선도와 성능의 균형을 위해 TTL(time-to-live) 사용

액션 단계

단계가 외부 동작을 수행해야 한다면:

Send reply (응답 발송)

• 노드 실행 시점: 승인 후(인간 또는 자동)
• 재시도 전략: 예 — 네트워크 문제에는 지수 백오프 사용
• 캐싱 금지: 각 발송은 유일한 동작이에요

Bug track (버그 추적)

• 노드 실행 시점: 의도가 "bug"일 때 항상
• 재시도 전략: 예 — 버그 리포트를 잃지 않는 것이 중요해요
• 반환값: 응답에 포함할 티켓 ID

사용자 입력 단계

단계에 인간의 개입이 필요하다면:

Human review node (인간 검토 노드)

• 결정을 위한 컨텍스트: 원본 이메일, 초안 응답, 긴급도, 분류
• 예상 입력 형식: 승인 불리언 + 선택적으로 수정된 응답
• 트리거 시점: 긴급도 높음, 복잡한 이슈, 또는 품질 우려가 있을 때

Step 3: 상태(state)를 설계해요

상태란 에이전트의 모든 노드가 접근할 수 있는 공유 메모리예요. 에이전트가 프로세스를 진행하면서 배우고 결정한 모든 것을 기록해 두는 공책이라고 생각하면 돼요.

무엇을 상태에 넣어야 할까?

각 데이터 조각에 대해 이런 질문을 스스로 해보세요:

상태에 포함 — 단계 간에 유지되어야 하나요? 그렇다면 상태에 들어가요.

저장하지 않기 — 다른 데이터로부터 유도할 수 있나요? 그렇다면 상태에 저장하는 대신 필요할 때 계산해요.

우리 이메일 에이전트의 경우 다음을 추적해야 해요:

• 원본 이메일과 발신자 정보 (나중에 재구성할 수 없어요)
• 분류 결과 (여러 이후/하류 노드에서 필요해요)
• 검색 결과와 고객 데이터 (다시 가져오기 비용이 커요)
• 초안 응답 (검토 과정을 거치며 유지되어야 해요)
• 실행 메타데이터 (디버깅과 복구용이에요)

상태는 원본 그대로, 프롬프트는 필요할 때 포맷해요

핵심 원칙 하나: 상태에는 원본 데이터(raw data) 를 저장하고, 포맷된 텍스트는 저장하지 않아요. 프롬프트는 노드 안에서 필요할 때 포맷해요.

이렇게 분리하면 어떤 점이 좋은지 볼게요:

• 서로 다른 노드가 같은 데이터를 자기 필요에 맞게 다르게 포맷할 수 있어요
• 상태 스키마를 고치지 않고도 프롬프트 템플릿을 바꿀 수 있어요
• 디버깅이 더 명확해져요 — 각 노드가 정확히 어떤 데이터를 받았는지 보이거든요
• 기존 상태를 깨지 않고 에이전트를 진화시킬 수 있어요

이제 우리 상태를 정의해 봅시다:

from typing import TypedDict, Literal

# Define the structure for email classification
class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    # Raw email data
    email_content: str
    sender_email: str
    email_id: str

    # Classification result
    classification: EmailClassification | None

    # Raw search/API results
    search_results: list[str] | None  # List of raw document chunks
    customer_history: dict | None  # Raw customer data from CRM

    # Generated content
    draft_response: str | None
    messages: list[str] | None

상태에는 원본 데이터만 들어간다는 점을 눈여겨보세요 — 프롬프트 템플릿도, 포맷된 문자열도, 지시사항도 없어요. 분류 출력은 LLM에서 나온 그대로 단일 딕셔너리로 저장돼요.

Step 4: 노드를 만들어요

이제 각 단계를 함수 하나로 구현해요. LangGraph에서 노드는 현재 상태를 받아 그에 대한 업데이트를 반환하는 파이썬 함수일 뿐이에요.

오류는 상황에 맞게 처리해요

서로 다른 오류는 서로 다른 처리 전략이 필요해요:

오류 유형	누가 고치나	전략	언제 쓰나
일시적 오류(네트워크 문제, 속도 제한)	시스템(자동)	재시도 정책	보통 재시도하면 해결되는 일시적 실패
LLM이 복구할 수 있는 오류(도구 실패, 파싱 문제)	LLM	오류를 상태에 저장하고 루프백	LLM이 오류를 보고 접근 방식을 조정할 수 있을 때
사용자가 고칠 수 있는 오류(정보 누락, 불명확한 지시)	인간	interrupt()로 일시 정지	진행하려면 사용자 입력이 필요할 때
재시도 후에도 복구할 수 없는 실패	개발자(선언적)	error_handler	재시도 소진 후 보상/복구 분기 실행
예상치 못한 오류	개발자	그대로 버블링	디버깅이 필요한 알 수 없는 문제

네트워크 문제와 속도 제한을 자동으로 재시도하도록 재시도 정책을 추가해요. timeout=와 함께 쓰면 각 시도에 상한을 둘 수 있어요. 전체 생명주기는 Fault tolerance 문서를 참고하세요.

from langgraph.types import RetryPolicy

workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)

오류를 상태에 저장하고 루프백해서, LLM이 무엇이 잘못됐는지 보고 다시 시도하게 해요:

from langgraph.types import Command

def execute_tool(state: State) -> Command[Literal["agent", "execute_tool"]]:
    try:
        result = run_tool(state['tool_call'])
        return Command(update={"tool_result": result}, goto="agent")
    except ToolError as e:
        # Let the LLM see what went wrong and try again
        return Command(
            update={"tool_result": f"Tool error: {str(e)}"},
            goto="agent"
        )

필요할 때(계정 ID, 주문 번호, 명확한 설명 같은 것) 사용자에게서 정보를 수집하려면 일시 정지해요:

from langgraph.types import Command

def lookup_customer_history(
    state: State
) -> Command[Literal["lookup_customer_history", "draft_response"]]:
    if not state.get('customer_id'):
        user_input = interrupt({
            "message": "Customer ID needed",
            "request": "Please provide the customer's account ID to look up their subscription history"
        })
        return Command(
            update={"customer_id": user_input['customer_id']},
            goto="lookup_customer_history"
        )
    # Now proceed with the lookup
    customer_data = fetch_customer_history(state['customer_id'])
    return Command(update={"customer_history": customer_data}, goto="draft_response")

디버깅을 위해 그대로 버블링하게 두어요. 처리할 수 없는 오류는 잡지 않아요:

def send_reply(state: EmailAgentState):
    try:
        email_service.send(state["draft_response"])
    except Exception:
        raise  # Surface unexpected errors

재시도를 모두 소진한 뒤에는, 상태를 업데이트하고 보상 분기로 라우팅하는 복구 함수를 실행해요. 전체 패턴은 Fault tolerance 문서를 참고하세요.

error_handler는 langgraph>=1.2가 필요해요.

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

workflow.add_node(
    "charge_payment",
    charge_payment,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
    error_handler=payment_error_handler,
)

같은 retry_policy, timeout, error_handler를 모든 노드에 반복해서 넣지 않고 적용하려면 StateGraph.set_node_defaults(...)를 사용해요. 노드별 값이 여전히 우선 적용돼요. 자세한 내용은 Fault tolerance 문서를 참고하세요.

이메일 에이전트 노드 구현하기

각 노드를 단순한 함수로 구현할게요. 기억하세요: 노드는 상태를 받고, 작업을 하고, 업데이트를 반환해요.

Read(읽기)와 classify(분류) 노드

from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command, RetryPolicy
from langchain_openai import ChatOpenAI
from langchain.messages import HumanMessage

llm = ChatOpenAI(model="gpt-5-nano")

def read_email(state: EmailAgentState) -> dict:
    """Extract and parse email content"""
    # In production, this would connect to your email service
    return {
        "messages": [HumanMessage(content=f"Processing email: {state['email_content']}")]
    }

def classify_intent(state: EmailAgentState) -> Command[Literal["search_documentation", "human_review", "draft_response", "bug_tracking"]]:
    """Use LLM to classify email intent and urgency, then route accordingly"""

    # Create structured LLM that returns EmailClassification dict
    structured_llm = llm.with_structured_output(EmailClassification)

    # Format the prompt on-demand, not stored in state
    classification_prompt = f"""
    Analyze this customer email and classify it:

    Email: {state['email_content']}
    From: {state['sender_email']}

    Provide classification including intent, urgency, topic, and summary.
    """

    # Get structured response directly as dict
    classification = structured_llm.invoke(classification_prompt)

    # Determine next node based on classification
    if classification['intent'] == 'billing' or classification['urgency'] == 'critical':
        goto = "human_review"
    elif classification['intent'] in ['question', 'feature']:
        goto = "search_documentation"
    elif classification['intent'] == 'bug':
        goto = "bug_tracking"
    else:
        goto = "draft_response"

    # Store classification as a single dict in state
    return Command(
        update={"classification": classification},
        goto=goto
    )

Search(검색)와 tracking(추적) 노드

def search_documentation(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """Search knowledge base for relevant information"""

    # Build search query from classification
    classification = state.get('classification', {})
    query = f"{classification.get('intent', '')} {classification.get('topic', '')}"

    try:
        # Implement your search logic here
        # Store raw search results, not formatted text
        search_results = [
            "Reset password via Settings > Security > Change Password",
            "Password must be at least 12 characters",
            "Include uppercase, lowercase, numbers, and symbols"
        ]
    except SearchAPIError as e:
        # For recoverable search errors, store error and continue
        search_results = [f"Search temporarily unavailable: {str(e)}"]

    return Command(
        update={"search_results": search_results},  # Store raw results or error
        goto="draft_response"
    )

def bug_tracking(state: EmailAgentState) -> Command[Literal["draft_response"]]:
    """Create or update bug tracking ticket"""

    # Create ticket in your bug tracking system
    ticket_id = "BUG-12345"  # Would be created via API

    return Command(
        update={
            "search_results": [f"Bug ticket {ticket_id} created"],
            "current_step": "bug_tracked"
        },
        goto="draft_response"
    )

응답 노드

def draft_response(state: EmailAgentState) -> Command[Literal["human_review", "send_reply"]]:
    """Generate response using context and route based on quality"""

    classification = state.get('classification', {})

    # Format context from raw state data on-demand
    context_sections = []

    if state.get('search_results'):
        # Format search results for the prompt
        formatted_docs = "\n".join([f"- {doc}" for doc in state['search_results']])
        context_sections.append(f"Relevant documentation:\n{formatted_docs}")

    if state.get('customer_history'):
        # Format customer data for the prompt
        context_sections.append(f"Customer tier: {state['customer_history'].get('tier', 'standard')}")

    # Build the prompt with formatted context
    draft_prompt = f"""
    Draft a response to this customer email:
    {state['email_content']}

    Email intent: {classification.get('intent', 'unknown')}
    Urgency level: {classification.get('urgency', 'medium')}

    {chr(10).join(context_sections)}

    Guidelines:
    - Be professional and helpful
    - Address their specific concern
    - Use the provided documentation when relevant
    """

    response = llm.invoke(draft_prompt)

    # Determine if human review needed based on urgency and intent
    needs_review = (
        classification.get('urgency') in ['high', 'critical'] or
        classification.get('intent') == 'complex'
    )

    # Route to appropriate next node
    goto = "human_review" if needs_review else "send_reply"

    return Command(
        update={"draft_response": response.content},  # Store only the raw response
        goto=goto
    )

def human_review(state: EmailAgentState) -> Command[Literal["send_reply", END]]:
    """Pause for human review using interrupt and route based on decision"""

    classification = state.get('classification', {})

    # interrupt() must come first - any code before it will re-run on resume
    human_decision = interrupt({
        "email_id": state.get('email_id',''),
        "original_email": state.get('email_content',''),
        "draft_response": state.get('draft_response',''),
        "urgency": classification.get('urgency'),
        "intent": classification.get('intent'),
        "action": "Please review and approve/edit this response"
    })

    # Now process the human's decision
    if human_decision.get("approved"):
        return Command(
            update={"draft_response": human_decision.get("edited_response", state.get('draft_response',''))},
            goto="send_reply"
        )
    else:
        # Rejection means human will handle directly
        return Command(update={}, goto=END)

def send_reply(state: EmailAgentState) -> dict:
    """Send the email response"""
    # Integrate with email service
    print(f"Sending reply: {state['draft_response'][:100]}...")
    return {}

Step 5: 모두 연결해요

이제 노드들을 작동하는 그래프 하나로 연결해요. 우리 노드들은 라우팅 결정을 스스로 처리하므로, 필수적인 엣지만 몇 개 정의하면 돼요. interrupt()로 human-in-the-loop을 활성화하려면, 실행 사이에 상태를 저장할 체크포인터(checkpointer)와 함께 컴파일해야 해요:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import RetryPolicy

# Create the graph
workflow = StateGraph(EmailAgentState)

# Add nodes with appropriate error handling
workflow.add_node("read_email", read_email)
workflow.add_node("classify_intent", classify_intent)

# Add retry policy for nodes that might have transient failures
workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3)
)
workflow.add_node("bug_tracking", bug_tracking)
workflow.add_node("draft_response", draft_response)
workflow.add_node("human_review", human_review)
workflow.add_node("send_reply", send_reply)

# Add only the essential edges
workflow.add_edge(START, "read_email")
workflow.add_edge("read_email", "classify_intent")
workflow.add_edge("send_reply", END)

# Compile with checkpointer for persistence, in case run graph with Local_Server --> Please compile without checkpointer
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

그래프 구조를 최소한으로 유지하는 이유는, 라우팅이 노드 안에서 Command 객체를 통해 일어나기 때문이에요. 각 노드는 Command[Literal["node1", "node2"]] 같은 타입 힌트로 갈 수 있는 곳을 선언해서, 흐름을 명시적이고 추적 가능하게 만들어요.

에이전트를 시험해 보기

인간 검토가 필요한 긴급한 결제 이슈로 에이전트를 돌려볼게요:

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class EmailState(TypedDict):
    email_content: str
    response_text: str | None

def human_review_node(state: EmailState):
    interrupt(
        {
            "approved": False,
            "edited_response": state.get("response_text") or "",
        }
    )
    return {"response_text": "placeholder"}

app = (
    StateGraph(EmailState)
    .add_node("human_review", human_review_node)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)

initial_state = {
    "email_content": "I was charged twice for my subscription! This is urgent!",
    "response_text": "Draft response",
}

# Run with a thread_id for persistence
config = {"configurable": {"thread_id": "customer_123"}}
stream = app.stream_events(initial_state, config, version="v3")
_ = stream.output  # drive the stream to completion
# The graph will pause at human_review
print(f"human review interrupt:{stream.interrupts}")

human_response = Command(
    resume={
        "approved": True,
        "edited_response": "We sincerely apologize for the double charge. I've initiated an immediate refund...",
    }
)

# Resume execution
resumed = app.stream_events(human_response, config, version="v3")
final_state = resumed.output
print("Email sent successfully!")

그래프는 interrupt()를 만나면 일시 정지해서 모든 것을 체크포인터에 저장하고 기다려요. 나중에 며칠이 지나서도, 멈췄던 바로 그 지점부터 이어서 재개할 수 있어요. thread_id는 이 대화의 모든 상태가 함께 보존되도록 보장해요.

요약과 다음 단계

핵심 통찰

이 이메일 에이전트를 만들면서 LangGraph식 사고법이 무엇인지 확인했어요:

개별 단계로 쪼개기 — 각 노드는 한 가지 일을 잘 해요. 이렇게 분해하면 스트리밍 진행 업데이트, 일시 정지·재개가 가능한 지속 실행(durable execution), 그리고 단계 사이의 상태를 검사할 수 있는 명확한 디버깅이 가능해져요.

상태는 공유 메모리 — 원본 데이터를 저장하고 포맷된 텍스트는 저장하지 않아요. 이 덕분에 서로 다른 노드가 같은 정보를 다른 방식으로 쓸 수 있어요.

노드는 함수 — 상태를 받아 작업하고 업데이트를 반환해요. 라우팅 결정이 필요할 때는 상태 업데이트와 다음 목적지를 함께 지정해요.

오류도 흐름의 일부 — 일시적 실패는 재시도되고, LLM이 복구할 수 있는 오류는 컨텍스트와 함께 루프백되며, 사용자가 고칠 수 있는 문제는 입력을 위해 일시 정지되고, 예상치 못한 오류는 디버깅을 위해 버블링돼요.

인간 입력은 일급 시민 — interrupt() 함수는 실행을 무기한 일시 정지하고 모든 상태를 저장하며, 입력이 제공되면 멈췄던 바로 그 지점에서 재개해요. 노드에서 다른 연산과 함께 쓸 때는 반드시 먼저 와야 해요.

그래프 구조는 자연스럽게 드러나요 — 필수 연결만 정의하고, 라우팅 로직은 노드가 스스로 처리해요. 이렇게 하면 제어 흐름이 명시적이고 추적 가능해져서, 현재 노드만 봐도 에이전트가 다음에 무엇을 할지 항상 이해할 수 있어요.

고급 고려사항

노드 세분화(granularity)의 트레이드오프

이 절은 노드 세분화 설계의 트레이드오프를 다뤄요. 대부분의 애플리케이션은 이 부분을 건너뛰고 위에 나온 패턴을 그대로 써도 돼요.

Read Email과 Classify Intent를 하나의 노드로 합치면 안 될까? Doc Search와 Draft Reply를 왜 분리했을까? 답은 복원성(resilience)과 관찰 가능성(observability) 사이의 트레이드오프에 있어요.

복원성 고려사항: LangGraph의 지속성 계층은 노드 경계에서 체크포인트를 만들어요. 워크플로가 중단이나 실패 후 재개될 때, 실행이 멈춘 노드의 시작부터 다시 시작해요. 노드가 작을수록 체크포인트가 더 자주 생겨서, 문제가 생겼을 때 반복할 작업이 줄어요. 여러 연산을 하나의 큰 노드로 합치면 끝부분 가까이에서 실패했을 때 그 노드 처음부터 전부 다시 실행해야 해요.

이메일 에이전트에 이 분해를 적용한 이유:

• 외부 서비스 격리: Doc Search와 Bug Track은 외부 API를 호출하므로 별도 노드예요. 검색 서비스가 느리거나 실패해도 LLM 호출과 격리하고 싶었어요. 이 특정 노드에만 다른 노드에 영향 없이 재시도 정책을 추가할 수 있어요.
• 중간 가시성: Classify Intent를 자체 노드로 두면, 행동을 취하기 전에 LLM이 무엇을 결정했는지 검사할 수 있어요. 에이전트가 언제, 왜 인간 검토로 라우팅하는지 정확히 볼 수 있어 디버깅과 모니터링에 유용해요.
• 서로 다른 실패 모드: LLM 호출, DB 조회, 이메일 발송은 재시도 전략이 서로 달라요. 노드를 분리하면 각각을 독립적으로 설정할 수 있어요.
• 재사용성과 테스트: 작은 노드는 격리해서 테스트하고 다른 워크플로에 재사용하기 쉬워요.

다른 유효한 접근도 있어요: Read Email과 Classify Intent를 하나의 노드로 합칠 수 있어요. 그러면 분류 전에 원본 이메일을 검사하는 능력을 잃고, 그 노드에서 실패할 때 두 연산을 모두 반복해야 해요. 대부분의 애플리케이션에서는 별도 노드로 두는 쪽의 관찰 가능성과 디버깅 이점이 그 트레이드오프를 감수할 만해요.

애플리케이션 수준 고려사항: Step 2의 캐싱 논의(검색 결과를 캐싱할지 말지)는 애플리케이션 수준의 결정이지 LangGraph 프레임워크 기능이 아니에요. 캐싱은 구체적인 요구사항에 따라 노드 함수 안에서 직접 구현해요. LangGraph가 이것을 규정하지 않아요.

성능 고려사항: 노드가 많다고 실행이 느려지지는 않아요. LangGraph는 기본적으로 체크포인트를 백그라운드에서 기록하므로(async durability mode), 그래프는 체크포인트 완료를 기다리지 않고 계속 실행돼요. 즉 성능 영향은 최소화하면서 잦은 체크포인트를 얻을 수 있어요. 필요하면 이 동작을 조정할 수 있는데, "exit" 모드는 완료 시에만 체크포인트하고, "sync" 모드는 각 체크포인트가 기록될 때까지 실행을 블록해요.

다음으로 어디로 갈까

이번 글은 LangGraph로 에이전트를 만드는 사고법의 입문이었어요. 이 기반을 다음으로 확장할 수 있어요:

Human-in-the-loop 패턴 — 실행 전 도구 승인, 일괄 승인 같은 패턴 추가하는 법

서브그래프 (Subgraphs) — 복잡한 다단계 연산을 위한 서브그래프 만들기

스트리밍 (Streaming) — 사용자에게 실시간 진행 상황을 보여주는 스트리밍 추가하기

관찰 가능성 (Observability) — 디버깅과 모니터링을 위한 LangSmith 관찰 가능성 추가하기

도구 통합 (Tool Integration) — 웹 검색, DB 쿼리, API 호출을 위한 더 많은 도구 통합하기

재시도 로직 (Retry Logic) — 실패한 연산을 위한 지수 백오프 재시도 로직 구현하기