LangGraph로 만드는 멀티 에이전트 — Coordinator + Worker 패턴

핵심 요약

단일 에이전트로 풀 수 없는 복잡한 작업은 멀티 에이전트로 역할 분담. LangGraph는 그래프 기반으로 에이전트들의 협력을 코드로 표현. Coordinator(계획·조율) + Worker(실행)의 고전 패턴이 가장 검증됨.

• LangGraph 0.4 (2026-03)
• StateGraph로 상태 공유
• Checkpointer로 영속화
• Human-in-the-loop 1급 시민

1. 기본 — 두 에이전트

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: list[str]
    completed_steps: list[dict]

def coordinator(state):
    # 계획 수립 또는 다음 단계 결정
    if not state.get("plan"):
        plan = llm.invoke(
            f"다음 작업의 step 목록을 만들어:\n{state['messages'][-1].content}"
        )
        return {"plan": parse_plan(plan)}
    if state["plan"]:
        return {"next_action": "worker"}
    return {"next_action": "end"}

def worker(state):
    next_step = state["plan"][0]
    result = execute_step(next_step)
    return {
        "plan": state["plan"][1:],  # 처리된 단계 제거
        "completed_steps": [{"step": next_step, "result": result}]
    }

# Graph
builder = StateGraph(AgentState)
builder.add_node("coordinator", coordinator)
builder.add_node("worker", worker)
builder.set_entry_point("coordinator")
builder.add_conditional_edges(
    "coordinator",
    lambda s: s.get("next_action", "end"),
    {"worker": "worker", "end": END}
)
builder.add_edge("worker", "coordinator")  # 다시 coordinator로

graph = builder.compile()

2. 실전 — 4-에이전트 코드 리뷰

from langgraph.graph import StateGraph

class ReviewState(TypedDict):
    diff: str
    security_findings: list
    perf_findings: list
    style_findings: list
    final_report: str

def security_reviewer(state):
    findings = security_llm.invoke(state["diff"])
    return {"security_findings": findings}

def perf_reviewer(state):
    findings = perf_llm.invoke(state["diff"])
    return {"perf_findings": findings}

def style_reviewer(state):
    findings = style_llm.invoke(state["diff"])
    return {"style_findings": findings}

def synthesizer(state):
    report = synth_llm.invoke({
        "security": state["security_findings"],
        "perf": state["perf_findings"],
        "style": state["style_findings"]
    })
    return {"final_report": report}

builder = StateGraph(ReviewState)
builder.add_node("security", security_reviewer)
builder.add_node("perf", perf_reviewer)
builder.add_node("style", style_reviewer)
builder.add_node("synthesizer", synthesizer)

# 3개 reviewer 병렬 실행 후 synthesizer
builder.set_entry_point("security")  # parallel start
builder.add_edge("security", "synthesizer")
builder.add_edge("perf", "synthesizer")
builder.add_edge("style", "synthesizer")
builder.add_edge("synthesizer", END)
# (LangGraph는 set_entry_point에 list를 받으면 병렬)

3. Checkpointer — 영속화

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
checkpointer.setup()

graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": user_id}}
result = graph.invoke(initial_state, config=config)

# 며칠 후 같은 thread_id로 다시 시작 — 마지막 체크포인트에서 재개
resumed = graph.invoke(None, config=config)

4. Human-in-the-loop

def coordinator(state):
    if needs_approval(state):
        return {"__interrupt__": "사용자 승인 필요"}
    return {...}

# graph 실행 → interrupt 시점에 멈춤
result = graph.invoke(state, config=config)

# 승인 후 재개
graph.update_state(config, {"approved": True})
result = graph.invoke(None, config=config)

금융·보안 워크플로에서 critical 단계 사람 검증 필수.

5. 모델 선택 (역할별)

# 비용 최적화
coordinator = ChatAnthropic(model="claude-opus-4-7")  # 계획·판단
worker = ChatAnthropic(model="claude-sonnet-4-6")  # 실행
style_reviewer = ChatAnthropic(model="claude-haiku-3-5")  # 단순 검사

6. 관측성 — LangSmith

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "..."
os.environ["LANGCHAIN_PROJECT"] = "multi-agent-prod"

# 자동으로 모든 노드·LLM 호출 trace

LangSmith 대시보드에서 단계별 latency·token·cost 추적.

7. 실패 복구

def worker_with_retry(state):
    for attempt in range(3):
        try:
            result = execute_step(state["plan"][0])
            return {"completed_steps": [{...}]}
        except RetryableError as e:
            if attempt == 2:
                return {"error": str(e), "next_action": "human_review"}
            time.sleep(2 ** attempt)

8. 비용·지연 최적화

• 병렬 실행 가능한 노드는 fan-out (parallel start)
• 가장 빈번한 호출은 caching (Anthropic prompt cache)
• 역할별 모델 분리 — 모든 호출에 Opus 쓰지 말 것
• checkpoint를 너무 자주 저장하지 말 것 (배치)

9. 디버깅

# 실행 trace 시각화
from langgraph.checkpoint.memory import MemorySaver
import json

for event in graph.stream(state, config=config):
    print(json.dumps(event, indent=2))
# 각 노드 진입·종료·state 변화 출력

10. 의사결정 — 멀티 에이전트가 정말 필요한가?

다음 조건 중 2개 이상이면 멀티 에이전트 정당화:

• 역할별로 다른 시스템 프롬프트·도구가 필요
• 병렬 처리로 시간 단축이 핵심
• 특정 단계만 큰 모델 필요 (다른 단계는 작은 모델)
• 인간 검증 단계가 명확

아니면 단일 에이전트 + ReAct 패턴이 단순.

11. 한계

• 구현 복잡도 — 단일 에이전트 대비 5배 코드
• 디버깅 어려움 — 어느 노드에서 잘못됐는지
• 비용 — 노드마다 LLM 호출, 빠르게 누적
• 대기 시간 — 노드 간 직렬 실행은 누적 latency

실전 사례

자주 묻는 질문

LangGraph vs CrewAI?LangGraph는 풀 제어·관측성. CrewAI는 단순함. 운영급 시스템은 LangGraph.

LangGraph 학습 부담?StateGraph 개념 이해까지 1~2일. 실제 production 시스템은 1주 정도.

Anthropic Managed Agents와의 차이?Managed는 인프라·세션 자동 관리, 단일 에이전트 위주. LangGraph는 멀티 에이전트·복잡 워크플로 우위.