AI 에이전트 블로그
컴퓨터 액세스 시스템

상태 기계와 에이전트: LangGraph로 신뢰할 수 있는 워크플로우 구축하기

agentsarchitecturetutoriallanggraphautonomous-systems

대부분의 에이전트 튜토리얼은 단순한 루프를 보여줍니다: Claude에게 질문하고, 응답을 파싱하고, 도구를 호출하고, 반복합니다. 데모에서는 이것으로 충분합니다. 프로덕션에서는 결정론성, 오류 복구, 인간 승인 게이트, 감사 가능성이 필요합니다.

LangGraph는 에이전트 워크플로우에 상태 기계를 도입합니다. if 문으로 이어붙인 임시방편 루프 대신, 명시적인 그래프를 얻습니다: 이름이 있는 노드(로직 단위), 타입이 있는 키나리(전환), 그리고 전체 실행을 통해 흐르는 공유 상태 스키마.

에이전트에 상태 기계가 필요한 이유

임시방편 루프의 문제점

전형적인 에이전트 루프는 이렇게 생겼습니다:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

도구가 두세 개라면 읽기 쉽습니다. 다섯 개의 도구, 조건부 경로, 인간 승인 단계, 재시도 로직을 추가하면 수백 줄의 얽힌 제어 흐름이 됩니다.

더 깊은 문제는 암묵적 상태입니다. 에이전트가 어떤 단계에 있나요? 어떤 데이터를 수집했나요? 모든 것이 messages에 있습니다 — 강제 스키마 없이 각 노드가 읽고 추가하는 타입 없는 덩어리.

상태 기계가 해결책인 이유

상태 기계는 암묵적인 것을 명시적으로 만듭니다. 정의하는 것은:

  • 노드 — 현재 상태를 받아 한 가지를 하고 상태 업데이트를 반환하는 이산적 로직 단위.
  • 키나리 — 노드 간 전환. 무조건(A → B 항상) 또는 조건부.
  • 상태 — 전체 그래프를 흐르는 타입이 있는 딕셔너리. 각 단계에서 스키마가 검증됩니다.

LangGraph를 쓰지 말아야 할 때

단순한 1단계 작업에는 직접 API 호출이 더 빠르고 명확합니다. 다음의 경우 LangGraph를 사용하세요:

  • 순서대로 실행해야 하는 여러 단계
  • 중간 결과에 기반한 조건부 분기
  • 인간 참여 단계
  • 오류 복구 또는 재시도 로직
  • 감사 가능성 요구사항

LangGraph 기초

상태 정의

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

노드

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"이 주제를 철저히 조사하세요: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"이 메모를 바탕으로 초안을 작성하세요:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

키나리와 조건부 라우팅

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

그래프 구축 및 실행

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "MCP 프로토콜이란 무엇인가?"})
print(result["draft"])

문서 검토 워크플로우 구축

상태 설계

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

노드 구현

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "이 문서에서 정확히 5개의 핵심 용어를 추출하세요. "
                "문자열의 JSON 배열로 반환하세요.\n\n"
                f"문서: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "이 문서의 컴플라이언스 문제를 확인하세요. "
                "찾을 것: 개인정보(이름, 주민등록번호, 이메일, 전화번호), "
                "기밀 표시, 미검증 주장.\n\n"
                "문제 설명의 JSON 배열을 반환하세요. "
                "문제가 없으면 빈 배열 []을 반환하세요.\n\n"
                f"문서: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== 인간 검토 필요 ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "개인정보 삭제 후 검토 및 승인",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"이 문서를 정확히 두 문장으로 요약하세요:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

그래프 조립

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "환자 김철수(주민등록번호: 800101-1234567)는 고혈압을 앓고 있습니다.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

인간 체크포인트와 인터럽트

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("발견된 문제:", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "수동 개인정보 삭제 후 승인", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

영속성과 프로덕션 배포

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

일반적인 패턴과 함정

  • 팬아웃/팬인: Send로 여러 노드를 병렬 실행하고 결과를 병합합니다.
  • 과도한 조건부 라우팅: 라우팅 함수의 분기가 너무 많으면 서브그래프로 분리합니다.
  • 상태 비대화: 대용량 아티팩트 대신 참조(데이터베이스 ID, S3 키)를 저장합니다.
  • 노드 격리 테스트: 노드가 일반 함수이므로 직접 단위 테스트할 수 있습니다.
def test_extract_terms_node():
    state = {
        "document": "인공지능이 현대 사회를 빠르게 변화시키고 있습니다.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph는 임시방편 루프를 명시적이고 디버깅 가능하며 재개 가능한 워크플로우로 대체합니다. 선형 그래프(A → B → C → 끝)로 시작하세요. 분기가 필요할 때 조건부 키나리 하나를 추가하세요. 승인이 필요할 때 인간 체크포인트를 추가하세요. 대부분의 프로덕션 에이전트 워크플로우는 정확히 이 세 가지 패턴만 필요합니다.

관련 기사