LangGraph로 생각하기

LangGraph로 생각하기 (Thinking in LangGraph)

LangGraph로 에이전트를 만드는 방법을 어떻게 생각해야 하는지 배워요.

LangGraph로 에이전트를 만들 때, 먼저 에이전트를 **노드(nodes)**라는 개별 단계로 쪼개요. 그다음 각 노드에서의 다양한 결정과 전환을 설명해요. 마지막으로 각 노드가 읽고 쓸 수 있는 공유 **상태(state)**를 통해 노드들을 연결해요.

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

출처: 문서

본문

자동화하려는 프로세스부터 시작하기 (Start with the process you want to automate)

고객 지원 이메일을 처리하는 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로 에이전트를 구현할 때는 보통 같은 다섯 단계를 따르게 돼요.

1단계: 워크플로를 개별 단계로 매핑 (Step 1: Map out your workflow as discrete steps)

프로세스에서 뚜렷한 단계들을 먼저 식별해요. 각 단계는 노드(하나의 특정 일을 하는 함수)가 돼요. 그다음 이 단계들이 서로 어떻게 연결되는지 스케치해요.

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로 갑니다)

2단계: 각 단계가 무엇을 해야 하는지 식별 (Step 2: Identify what each step needs to do)

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

• LLM 단계(LM steps): 텍스트를 이해하고, 분석하고, 생성하거나 추론 결정을 내려야 할 때 사용해요.
• 데이터 단계(Data steps): 외부 소스에서 정보를 검색해야 할 때 사용해요.
• 액션 단계(Action steps): 외부 동작을 수행해야 할 때 사용해요.
• 사용자 입력 단계(User input steps): 인간의 개입이 필요할 때 사용해요.

LLM 단계 (LLM steps)

단계가 텍스트를 이해·분석·생성하거나 추론 결정을 내려야 할 때:

의도 분류 (Classify intent)

• 정적 컨텍스트(프롬프트): 분류 카테고리, 긴급도 정의, 응답 형식
• 동적 컨텍스트(상태로부터): 이메일 콘텐츠, 발신자 정보
• 원하는 결과: 라우팅을 결정하는 구조화된 분류

초안 작성 (Draft reply)

• 정적 컨텍스트(프롬프트): 어조 가이드라인, 회사 정책, 응답 템플릿
• 동적 컨텍스트(상태로부터): 분류 결과, 검색 결과, 고객 이력
• 원하는 결과: 검토 준비가 된 전문 이메일 응답

데이터 단계 (Data steps)

단계가 외부 소스에서 정보를 검색해야 할 때:

문서 검색 (Document search)

• 파라미터: 의도와 주제로 만든 쿼리
• 재시도 전략: 예, 일시적 실패에 지수 백오프(외백off) 사용
• 캐싱: 흔한 쿼리는 캐시해 API 호출을 줄일 수 있어요

고객 이력 조회 (Customer history lookup)

• 파라미터: 상태로부터의 고객 이메일 또는 ID
• 재시도 전략: 예, 하지만 없으면 기본 정보로 폴백
• 캐싱: 예, 신선도와 성능의 균형을 위한 time-to-live 사용

액션 단계 (Action steps)

단계가 외부 동작을 수행해야 할 때:

답장 보내기 (Send reply)

• 노드를 실행할 시점: 승인(사람 또는 자동) 후
• 재시도 전략: 예, 네트워크 문제에 지수 백오프 사용
• 캐시하지 말 것: 각 전송은 고유한 동작

버그 추적 (Bug track)

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

사용자 입력 단계 (User input steps)

단계가 인간의 개입이 필요할 때:

인간 검토 노드 (Human review node)

• 결정을 위한 컨텍스트: 원본 이메일, 초안 응답, 긴급도, 분류
• 기대 입력 형식: 승인 부울 + 선택적 편집된 응답
• 트리거 시점: 높은 긴급도, 복잡한 이슈, 또는 품질 우려

3단계: 상태 설계 (Step 3: Design your state)

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

상태에 무엇을 넣을까 (What belongs in state?)

각 데이터 조각에 대해 스스로에게 물어봐요.

• 상태에 포함 (Include in state): 여러 단계에 걸쳐 유지돼야 하나? 그렇다면 상태에 넣어요.
• 저장하지 말 것 (Don't store): 다른 데이터에서 파생할 수 있나? 그렇다면 상태에 저장하지 말고 필요할 때 계산해요.

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

• 원본 이메일과 발신자 정보 (나중에 재구성할 수 없어요)
• 분류 결과 (여러 이후·다운스트림 노드가 필요해요)
• 검색 결과와 고객 데이터 (다시 가져오는 비용이 커요)
• 초안 응답 (검토를 통과할 때까지 유지돼야 해요)
• 실행 메타데이터 (디버깅과 복구용)

상태는 원시 데이터로, 프롬프트는 필요할 때 포맷 (Keep state raw, format prompts on-demand)

핵심 원칙: 상태에는 포맷된 텍스트가 아니라 원시 데이터를 저장해야 해요. 프롬프트는 노드 안에서 필요할 때 포맷하세요.

이렇게 분리하면:

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

우리의 상태를 정의해 볼게요.

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에서 바로 나온 단일 딕셔너리로 저장돼요.

4단계: 노드 만들기 (Step 4: Build your nodes)

이제 각 단계를 함수로 구현해요. LangGraph에서 노드는 현재 상태를 받아 갱신을 반환하는 일반 Python 함수예요.

오류를 적절히 처리하기 (Handle errors appropriately)

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

Error Type	Who Fixes It	Strategy	When to Use
Transient errors (network issues, rate limits)	System (automatic)	Retry policy	Temporary failures that usually resolve on retry
LLM-recoverable errors (tool failures, parsing issues)	LLM	Store error in state and loop back	LLM can see the error and adjust its approach
User-fixable errors (missing information, unclear instructions)	Human	Pause with interrupt()	Need user input to proceed
Recoverable failure after retries	Developer (declarative)	error_handler	Run a compensation/recovery branch after retry exhaustion
Unexpected errors	Developer	Let them bubble up	Unknown issues that need debugging

일시적 오류 (Transient errors)

네트워크 문제와 rate limit을 자동으로 재시도하는 재시도 정책을 추가해요.

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-복구 가능 (LLM-recoverable)

오류를 상태에 저장하고 루프백해 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"
        )

사용자-수정 가능 (User-fixable)

필요할 때 사용자로부터 정보(계정 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")

예상치 못한 오류 (Unexpected)

디버깅을 위해 그대로 올려보내요(bubble up). 처리할 수 없는 것은 catch하지 마세요.

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

사가/보상 (Saga / compensation)

재시도를 모두 소진한 뒤, 상태를 갱신하고 보상 분기로 라우팅하는 복구 함수를 실행해요.

전체 패턴은 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를 매 add_node마다 반복하지 않고 그래프의 모든 노드에 적용하려면 StateGraph.set_node_defaults(...)를 사용해요. 노드별 값이 여전히 더 우선해요. Fault tolerance를 참고해요.

이메일 에이전트 노드 구현하기 (Implementing our email agent nodes)

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

읽기·분류 노드 (Read and classify nodes)

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 and tracking nodes)

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"
    )

응답 노드 (Response nodes)

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 {}

5단계: 연결하기 (Step 5: Wire it together)

이제 노드들을 동작하는 그래프로 연결해요. 노드들이 자신의 라우팅 결정을 스스로 처리하므로, 필수적인 엣지 몇 개만 있으면 돼요.

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"]] 같은 타입 힌트로 어디로 갈 수 있는지 선언해, 흐름을 명시적이고 추적 가능하게 만들어요.

에이전트 시험해 보기 (Try out your agent)

인간 검토가 필요한 긴급 청구 이슈로 에이전트를 실행해 볼게요.

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!")

Example trace 보기: 이 예시의 공개 LangSmith run을 열어보세요.

그래프는 interrupt()에 닿으면 일시정지하고, 모든 것을 checkpointer에 저장한 뒤 기다려요. 며칠 뒤에도 정확히 멈춘 지점부터 재개할 수 있어요. thread_id는 이 대화의 모든 상태가 함께 보존되도록 보장해요.

요약과 다음 단계 (Summary and next steps)

핵심 통찰 (Key Insights)

이 이메일 에이전트를 만들면서 LangGraph식 사고를 배웠어요.

• 개별 단계로 쪼개기: 각 노드는 한 가지 일을 잘해요. 이런 분해는 스트리밍 진행 업데이트, 일시정지·재개할 수 있는 내구성 있는 실행(durable execution), 그리고 각 단계 사이의 상태를 살펴볼 수 있는 명확한 디버깅을 가능하게 해요.
• 상태는 공유 메모리: 포맷된 텍스트가 아니라 원시 데이터를 저장해요. 그래야 서로 다른 노드가 같은 정보를 다른 방식으로 쓸 수 있어요.
• 노드는 함수: 상태를 받아 작업을 수행하고 갱신을 반환해요. 라우팅 결정이 필요하면 상태 갱신과 다음 목적지를 모두 지정해요.
• 오류는 흐름의 일부: 일시적 실패는 재시도를, LLM-복구 가능 오류는 컨텍스트와 함께 루프백을, 사용자-수정 가능 문제는 입력을 위한 일시정지를, 예상치 못한 오류는 디버깅을 위해 버블업을 받아요.
• 인간 입력은 일급 시민: interrupt() 함수는 실행을 무기한 일시정지하고, 모든 상태를 저장하며, 입력을 주면 정확히 멈췄던 지점에서 재개해요. 노드에서 다른 연산과 결합할 때는 반드시 먼저 와야 해요.
• 그래프 구조는 자연스럽게 드러나요: 필수적인 연결을 정의하면 노드가 자신의 라우팅 로직을 처리해요. 이렇게 하면 제어 흐름이 명시적이고 추적 가능해져서, 현재 노드를 보면 에이전트가 다음에 무엇을 할지 항상 이해할 수 있어요.

고급 고려사항 (Advanced considerations)

노드 세밀함 트레이드오프 (Node granularity trade-offs)

이 섹션은 노드 세밀함 설계의 트레이드오프를 탐구해요. 대부분의 애플리케이션은 이 부분을 건너뛰고 위에 보여준 패턴을 써도 돼요.

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를 단일 노드로 합칠 수도 있어요. 그러면 분류 전에 원시 이메일을 검사하는 능력을 잃고, 그 노드에서 실패하면 두 연산을 모두 반복하게 돼요. 대부분의 애플리케이션에서 노드 분리의 관측 가능성·디버깅 이점이 이 트레이드오프를 상쇄할 만해요.

애플리케이션 수준 우려: 2단계의 캐싱 논의(검색 결과를 캐시할지)는 애플리케이션 수준 결정이지 LangGraph 프레임워크 기능이 아니에요. 노드 함수 안에서 여러분의 특정 요구사항에 따라 캐싱을 구현해요. LangGraph가 이걸 처방하지 않아요.

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

여기서 어디로 갈까 (Where to go from here)

이건 LangGraph로 에이전트를 만드는 사고에 대한 소개였어요. 이 기초를 다음으로 확장할 수 있어요.

• Human-in-the-loop 패턴: 실행 전 도구 승인, 배치 승인 등의 패턴을 배워요.
• 서브그래프: 복잡한 다단계 연산을 위한 서브그래프를 만들어요.
• 스트리밍: 사용자에게 실시간 진행을 보여주는 스트리밍을 추가해요.
• 관측 가능성: 디버깅과 모니터링을 위한 LangSmith 관측 가능성을 추가해요.
• 도구 통합: 웹 검색, DB 쿼리, API 호출을 위한 더 많은 도구를 통합해요.
• 재시도 로직: 실패한 연산에 지수 백오프 재시도 로직을 구현해요.

더 알아보기 (Learn more)

• 인터럽트: human-in-the-loop와 interrupt()를 다뤄요.
• 서브그래프: 복잡한 다단계 연산의 하위 그래프를 다뤄요.
• 스트리밍: 실시간 진행 업데이트를 다뤄요.
• Fault tolerance: 재시도 정책과 오류 처리를 다뤄요.