software

input()을 넘어서: LangGraph로 프로덕션 준비된 인간‑인‑루프 AI 에이전트 구축

발행: (2025년 12월 21일 오후 02:26 GMT+9)
17 min read
원문: Dev.to

Source: Dev.to

번역할 텍스트가 제공되지 않았습니다. 번역이 필요한 본문을 알려주시면 한국어로 번역해 드리겠습니다.

소개

간단한 챗봇이나 CLI 도구를 만든 적이 있다면, 아마도 파이썬의 믿음직한 input() 함수를 사용했을 것입니다. 빠른 스크립트에는 아주 잘 맞습니다: 질문을 하고, 답변을 기다리고, 끝이죠. 하지만 더 정교한 무언가를 만들어야 할 때는 어떻게 될까요? AI 에이전트가 작업 흐름 중간에 멈추고, 인간의 승인을 기다려야 한다면 (몇 시간 혹은 며칠 동안) 그리고 정확히 중단된 지점부터 다시 시작해야 한다면?

바로 그때 input() 은 제 역할을 못합니다.

input()의 문제점

솔직히 말하자면: **input()**은 동기식 차단 함수입니다. 누군가가 무언가를 입력하고 Enter를 누를 때까지 프로그램 전체를 멈춥니다. 실제 운영 환경에서 이것이 의미하는 바는 다음과 같습니다:

Blocking input() in a production workflow

예를 들어, 금융 거래를 위한 AI 승인 시스템을 만든다고 가정해 보세요. 매니저가 금요일 오후 4시에 검토가 필요한 $50,000 거래에 대한 알림을 받습니다. 주말을 떠나려는 상황이죠. input()을 사용하면 프로세스는 그저… 멈춰서 대기하게 됩니다.

LangGraph를 사용하면 워크플로우가 일시 중지되고 모든 리소스를 해제한 뒤 인내심 있게 기다립니다. 매니저가 월요일 아침에 승인하면, 작업이 매끄럽게 재개됩니다.

잠깐, 이거 전에 본 적 있나요? BizTalk 연결

Microsoft BizTalk Server를 사용해 본 적이 있다면 익숙하게 느껴질 수 있습니다. LangGraph의 체크포인팅 시스템은 개념적으로 BizTalk의 탈수/재수화 메커니즘과 유사합니다—그럴 만한 이유가 있죠. 두 시스템 모두 동일한 근본적인 문제를 해결합니다: 오래 실행되는 워크플로우를 자원을 낭비하지 않고 어떻게 일시 중지할 것인가?

BizTalk의 접근 방식: 탈수와 재수화

BizTalk Server에서는 오케스트레이션(워크플로우)이 메시지, 타임아웃, 혹은 사람의 승인을 기다려야 할 때, BizTalk은 이를 메모리에 계속 유지하지 않습니다. 대신 다음과 같이 처리합니다.

  1. 오케스트레이션 인스턴스의 전체 상태를 MessageBox 데이터베이스에 직렬화하여 탈수합니다.
  2. 메모리에서 인스턴스를 제거해 서버 자원을 해제합니다.
  3. 트리거가 발생하면(메시지 수신, 타임아웃 도달) 데이터베이스에서 상태를 로드해 재수화합니다.
  4. 중단된 지점부터 정확히 이어서 실행합니다.

BizTalk 탈수 vs. LangGraph 체크포인팅

BizTalk dehydration vs. LangGraph checkpointing

What I Learned from BizTalk

BizTalk Server와 작업해 본 경험으로, dehydration/rehydration 패턴이 엔터프라이즈 워크플로에 필수적이라는 것을 말씀드릴 수 있습니다. 왜 중요한지 살펴보겠습니다:

  • Purchase Order Approval – 주문이 들어오면 검증을 거친 뒤 관리자의 승인을 기다립니다. BizTalk에서는 해당 오케스트레이션이 데이터베이스에 dehydrate됩니다. 관리자가 몇 시간, 며칠, 혹은 몇 주 후에 승인하면, 오케스트레이션이 rehydrate되어 처리를 계속합니다.
  • Long‑Running Transactions – 며칠, 몇 주, 심지어 몇 달에 걸쳐 진행되는 다단계 비즈니스 프로세스(예: 보험 청구, 계약 승인, 규제 워크플로)는 메모리에 머물 수 없습니다. BizTalk은 상태를 SQL Server에 저장하고, 상관 집합(correlation sets)을 추적해 들어오는 메시지를 올바른 오케스트레이션 인스턴스와 매핑합니다. 외부 승인이나 제3자 응답을 기다리며 몇 주 동안 dehydrate된 오케스트레이션을 본 적이 있습니다.
  • Server Restarts – BizTalk Server가 충돌하거나 재시작되더라도, 모든 dehydrate된 오케스트레이션은 데이터베이스에 영구 저장되므로 살아남습니다. 서버가 복구되면 자동으로 재개됩니다.

LangGraph는 이 검증된 패턴을 AI 워크플로에 적용합니다. XML 메시지와 상관 집합 대신 AI 에이전트 상태와 thread_id를 사용하고, MessageBox 데이터베이스 대신 PostgreSQL이나 SQLite를 활용합니다. 하지만 핵심 개념—상태를 지속하고, 리소스를 해제하며, 나중에 다시 시작하는 것—은 동일합니다.

Fun Fact: BizTalk의 MessageBox 데이터베이스는 본질적으로 거대한 상태 머신입니다. 각 오케스트레이션 인스턴스의 상태가 상관 속성과 함께 저장되어, BizTalk이 들어오는 메시지를 올바른 대기 중인 오케스트레이션으로 라우팅할 수 있게 합니다. LangGraph의 체크포인터도 AI 에이전트 워크플로에 대해 같은 역할을 수행합니다—thread_id가 바로 여러분의 상관 집합입니다!

Source: …

인간‑인‑루프의 세 기둥

프로덕션 수준의 인간‑인‑루프 시스템을 구축하려면 세 가지 핵심 요소가 함께 작동해야 합니다. 각각을 살펴보겠습니다.

1. 체크포인팅: 에이전트의 메모리

에이전트가 일시 중지하고 다시 시작하려면 메모리가 필요합니다. 단순한 메모리가 아니라 지속적이고 신뢰할 수 있는 메모리로, 충돌, 재시작, 심지어 다른 서버로 이동해도 유지됩니다.

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 인‑메모리 체크포인터 (개발에 적합)
checkpointer = MemorySaver()

# 프로덕션에서는 PostgresSaver 또는 SQLiteSaver 사용
graph = workflow.compile(checkpointer=checkpointer)

핵심 인사이트: 체크포인터가 없으면 인터럽트가 작동하지 않습니다. 여기서 말하는 체크포인터는 전체 실행 상태—변수, 컨텍스트, 진행 상황—를 저장합니다. 그래서 몇 시간 혹은 며칠 후에 다시 돌아와도 아무것도 손실되지 않죠.

비디오 게임을 저장하는 것과 비슷합니다. 체크포인트를 만들면 단일 변수만 저장하는 것이 아니라 현재 위치, 인벤토리, 체력, 퀘스트 진행 상황 등 모든 것을 저장하는 것이죠. 다시 시작하면 바로 그 지점에서 이어서 플레이할 수 있습니다.

2. In

(원본 내용이 여기서 끊겼습니다. 필요에 따라 나머지 섹션을 계속 작성하세요.)

인터럽트: 일시 정지 버튼

이제 메모리가 생겼으니 실제로 일시 정지를 할 수 있는 기능이 필요합니다. LangGraph는 이를 위한 두 가지 방법을 제공합니다:

정적 인터럽트 – “항상 여기서 일시 정지”

특정 노드가 항상 인간 검토가 필요하다는 것을 알고 있을 때 사용합니다.

# 노드 실행 **전**에 일시 정지
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_action"]
)

# 노드 실행 **후**에 일시 정지 (검토에 유용)
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_after=["generate_response"]
)

정적 인터럽트는 컴플라이언스 시나리오에 적합합니다.
모든 콘텐츠 생성은 게시 전에 인간 승인이 필요할 수 있고, 모든 데이터베이스 삭제는 두 번째 검토가 필요할 수 있습니다.

동적 인터럽트 – 조건부 일시 정지

런타임에 발생하는 상황에 따라 일시 정지 여부가 달라질 때 사용합니다.

from langgraph.types import interrupt

def process_transaction(state):
    amount = state["transaction_amount"]

    # 고액 거래에만 일시 정지
    if amount > 10_000:
        human_decision = interrupt({
            "question": f"Approve transaction of ${amount}?",
            "transaction_details": state["details"]
        })

        if human_decision.get("approved") != True:
            return {
                "status": "rejected",
                "reason": human_decision.get("reason")
            }

    # 거래 계속 진행
    return {"status": "approved", "processed": True}

이것은 강력합니다. AI가 언제 도움이 필요한지 스마트하게 판단할 수 있습니다:

  • 신뢰도 점수가 낮을 때 → 일시 정지하고 질문.
  • 거래 금액 > $10 K → 일시 정지하고 질문.
  • 그 외 경우 → 계속 진행.

언제 어떤 것을 사용해야 할까?

유형일반적인 사용 사례
Static컴플라이언스 검토, 최종 승인, “항상 여기서 멈추기” 시나리오
Dynamic고가치 거래, 낮은 신뢰도 예측, 엣지 케이스

명령: 재개 버튼

그래서 워크플로가 일시 중지되고, 사람이 검토한 뒤 결정을 내렸습니다. 이제 어떻게 할까요? 여기서 Command가 등장합니다:

from langgraph.types import Command

# 사람의 입력으로 일시 중지된 워크플로 재개
result = graph.invoke(
    Command(resume={
        "approved": True,
        "notes": "Verified customer identity"
    }),
    config={"configurable": {"thread_id": "transaction-123"}}
)

내부에서 무슨 일이 일어나는가

BEFORE (interrupt에서 워크플로 일시 중지):
    human_input = interrupt({...})   # ← 여기서 대기, 상태 저장

RESUME (사람이 결정 제공):
    graph.invoke(Command(resume={"approved": True}), config)
                ↓
    데이터가 interrupt()로 다시 흐름
                ↓
AFTER:
    human_input = {"approved": True} # ← 이제 사람의 응답이 들어감!

워크플로는 정확히 이 지점부터 계속됩니다. 그래프는 전체를 다시 시작하거나 재생하지 않고, 중단된 지점에서 사람의 입력을 바로 노드에 사용할 수 있게 이어집니다.

시각적 개요

Interrupt flow diagram
Resume flow diagram

모두 합치기

실제 프로덕션 워크플로가 어떻게 보이는지 예시입니다:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph

# 1. Set up checkpointing
checkpointer = MemorySaver()

# 2. Define your workflow with dynamic interrupts
def review_node(state):
    if state["risk_score"] > 0.8:
        decision = interrupt({
            "message": "High‑risk detected. Review required.",
            "data": state["analysis"]
        })
        return {"approved": decision["approved"]}
    return {"approved": True}

# 3. Compile with persistence
graph = workflow.compile(checkpointer=checkpointer)

# 4. Invoke the workflow
config = {"configurable": {"thread_id": "workflow-123"}}
result = graph.invoke(initial_state, config)

# 5. Check if paused
state = graph.get_state(config)
if bool(state.next):
    print("Workflow paused, waiting for human input")

    # Later, when the human responds...
    graph.invoke(
        Command(resume={"approved": True}),
        config
    )

왜 이것이 중요한가

생산 환경의 AI 시스템에서는 무한정 차단될 수 없습니다. 다음과 같은 워크플로가 필요합니다:

  • 인간 판단을 위해 일시 중지하되 자원을 소비하지 않음.
  • 재시작을 견디고 수백 개의 동시 실행을 처리함.
  • 금요일에 일시 중지하고 월요일에 재개하되 컨텍스트를 잃지 않음.

그것은 input()이 설계된 목적은 아니지만, 바로 LangGraph의 human‑in‑the‑loop 시스템이 처리하도록 만들어졌습니다. 차이는 단순히 기술적인 것이 아니라, 장난감과 실제 배포 가능한 도구 사이의 차이입니다.

최종 생각

human‑in‑the‑loop AI를 구축한다는 것은 워크플로에 일시정지 버튼을 추가하는 것만을 의미하지 않습니다. 인간의 의사결정이 비동기적이고 예측할 수 없는 특성을 존중하는 시스템을 만드는 것입니다. 관리자는 여러분의 코드 일정에 맞춰 일하지 않으며, 주말을 쉬고 생각할 시간이 필요합니다. LangGraph의 정적·동적 인터럽트와 Command 재개 메커니즘을 활용하면 이러한 유연성을 제공하면서도 AI 파이프라인을 견고하고 확장 가능하며 프로덕션에 바로 투입할 수 있게 유지할 수 있습니다.

결론

결정을 내리기 전에 다른 사람과 상의가 필요할 수도 있습니다.

LangGraph의 아키텍처는 체크포인팅, 인터럽트, 커맨드라는 세 가지 기둥을 통해 이 현실을 인정합니다. 인간과 함께 작동하는 AI 시스템을 구축할 수 있는 도구를 제공하며, 인간을 무시하고 작동하는 시스템이 아닙니다.

다음에 input()을 사용할 때 스스로에게 물어보세요: 나는 스크립트를 만들고 있는가, 아니면 AI (AI Agents) 시스템을 만들고 있는가? 후자라면 해야 할 일을 이미 알고 있는 것입니다.

행복한 개발 되세요!

감사합니다,

스리니 라마도라이

Back to Blog

관련 글

더 보기 »