LangGraph로 프로덕션 다중 에이전트(Multi-Agent) 시스템 구축하기: 상태 관리와 순환 그래프 워크플로우

단일 LLM 프롬프트나 단순한 Sequential Chain으로 복잡한 비즈니스 로직을 처리하는 시대는 지났습니다. 실무 환경에서는 기획, 검색, 코드 작성, 보안 검증, QA 테스트 등 각 분야에 특화된 여러 AI 에이전트가 협력하여 임무를 수행하는 다중 에이전트(Multi-Agent) 시스템이 필수적으로 요구됩니다.

기존의 단순 DAG(Directed Acyclic Graph) 기반 오케스트레이션 프레임워크는 에이전트 간의 루프(Loop), 피드백 재시도(Self-Correction), 상태 분기(Conditional Branching), 그리고 장기 지속 상태 관리(Persistence Checkpointing) 를 정교하게 다루기 어려웠습니다.

2026년 표준 다중 에이전트 프레임워크인 LangGraph는 그래프 기반의 상태 기계(State Machine) 구조를 도입하여 이러한 한계를 극복했습니다. 이 글에서는 LangGraph로 복합 에이전트 시스템을 설계하고 프로덕션 DB와 연동하는 실전 기술 패턴을 다룹니다.

핵심 요약

• StateGraph 기반의 상태 중앙화: 모든 에이전트 노드는 분산된 독립 상태 대신 중앙의 TypedDict 상태(State)를 참조하고, 리듀서(Reducer) 함수를 통해 상태를 스레드 안전(Thread-safe)하게 업데이트합니다.
• 순환 그래프(Cyclic Workflow) 지원: 에이전트가 검증 단계에서 실패하면 이전 상태(예: 코더 에이전트 노드)로 되돌아가는 조건부 루프(add_conditional_edges)를 손쉽게 선언할 수 있습니다.
• Checkpointer 영속성 메모리: MemorySaver 또는 AsyncPostgresSaver를 붙여 장애 발생 시 스냅샷 복구(Time Travel) 및 인간 개입 인터럽트(Human-in-the-loop)를 지원합니다.
• 팀 오케스트레이션 패턴: Supervisor(지휘자) 패턴과 Peer-to-Peer(상호 협력) 패턴 중 서비스 복잡도에 맞는 토폴로지를 선택해야 합니다.

1. LangGraph 핵심 개념: Nodes, Edges, State

LangGraph는 3가지 핵심 요소로 구성됩니다.

       [Start]
          │
          ▼
   ┌──────────────┐
   │ ResearchNode │
   └──────────────┘
          │
          ▼
   ┌──────────────┐      Failure
   │ CoderNode    │ ◄──────────────┐
   └──────────────┘                │
          │                        │ (Conditional Edge)
          ▼                        │
   ┌──────────────┐                │
   │ ReviewerNode ├────────────────┘
   └──────┬───────┘
          │ Success
          ▼
        [End]

1. State: 그래프 전체를 관통하는 공유 데이터 구조 (TypedDict 또는 Pydantic 모델).
2. Nodes: 특정 작업을 수행하는 파이썬 함수 (에이전트 또는 툴 호출).
3. Edges: 노드 간의 이동 경로를 결정하는 로직 (일반 Edge 및 Conditional Edge).

2. Supervisor 기반 Multi-Agent 실전 파이썬 코드

아래는 연구원(Researcher) 에이전트와 작가(Writer) 에이전트를 지휘자(Supervisor) 노드가 중계하는 프로덕션 레벨의 코드 스니펫입니다.

from typing import Annotated, TypedDict, Literal
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

# 1. 중앙 그래프 상태(State) 정의
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    next_step: str
    final_report: str

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# 2. 개별 에이전트 노드 정의
def researcher_node(state: AgentState):
    print("--- [Researcher Node Working] ---")
    messages = state["messages"]
    prompt = "당신은 전문 리서처입니다. 주제에 대해 핵심 사실을 수집하세요."
    response = llm.invoke([HumanMessage(content=prompt)] + messages)
    return {"messages": [response], "next_step": "writer"}

def writer_node(state: AgentState):
    print("--- [Writer Node Working] ---")
    messages = state["messages"]
    prompt = "당신은 테크 전문 작가입니다. 리서치 결과를 바탕으로 깔끔한 리포트를 작성하세요."
    response = llm.invoke([HumanMessage(content=prompt)] + messages)
    return {"messages": [response], "final_report": response.content, "next_step": "end"}

# 3. Supervisor (지휘자) 라우팅 노드
def supervisor_router(state: AgentState) -> Literal["researcher", "writer", "__end__"]:
    next_step = state.get("next_step", "researcher")
    if next_step == "writer":
        return "writer"
    elif next_step == "end":
        return END
    return "researcher"

# 4. StateGraph 생성 및 워크플로우 구성
workflow = StateGraph(AgentState)

workflow.add_node("researcher", researcher_node)
workflow.add_node("writer", writer_node)

# 조건부 엣지 및 라우팅 추가
workflow.add_edge(START, "researcher")
workflow.add_conditional_edges("researcher", supervisor_router)
workflow.add_conditional_edges("writer", supervisor_router)

# 5. Checkpointer 메모리 및 컴파일
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 6. 실행 예제
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "session-101"}}
    inputs = {"messages": [HumanMessage(content="Cloudflare D1의 장단점을 분석해줘.")]}
    
    for event in app.stream(inputs, config=config):
        for k, v in event.items():
            print(f"Node '{k}' executed.")

3. Human-in-the-loop (인간 승인) 및 Time Travel

프로덕션에서 결제나 릴리스 관련 조작이 이루어질 때는 AI가 바로 진행하지 않고 인간의 승인을 기다리는 Interrupt 기능이 필수입니다.

# 특정 노드 실행 이전 인터럽트 걸기
app_with_human = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["writer"] # writer 노드 실행 직전 일시정지
)

# 1차 실행 (researcher 까지만 실행 후 대기)
app_with_human.invoke(inputs, config=config)

# 인간 승인 후 재개
app_with_human.invoke(None, config=config)

Checkpointer 덕분에 개발자는 이전 특정 스냅샷 상태로 돌아가(Time Travel) 다른 입력값으로 에이전트 경로를 다시 재실행해볼 수도 있습니다.

4. LangGraph vs CrewAI vs AutoGen 비교

결론

LangGraph는 명확하게 정의된 **상태(State)**와 순환 그래프(Cyclic Graph) 구조를 통해, 실험실 수준의 LLM 데모를 견고하고 복원력 높은 프로덕션 enterprise AI 시스템으로 격상시킵니다.