4 – LangGraph로 멀티 에이전트 구성하기

Supervisor · Network · Hierarchical 패턴 & 실전 구현

‘LangChain 시리즈’ 네 번째 글입니다. 이번 편에서는 LangGraph를 활용해 복잡한 멀티 에이전트 워크플로를 설계·구현하는 방법을 **패턴별(Supervisor·Network·Hierarchical)**로 살펴봅니다. 각 패턴의 개념·장단점·코드·운영 팁·성능·보안까지 다루어 어떤 상황에서 어떤 구조를 선택할지 판단할 수 있도록 구성했습니다.

---

1. 멀티 에이전트가 필요한 이유

1. 역할 분리(Separation of Concerns): 전문화된 프롬프트와 Tool 세트가 충돌하지 않습니다.
2. 스케일 아웃(Parallelism): LLM 호출을 병렬화해 레이턴시와 비용을 최적화합니다.
3. 내구성(Resilience): LangGraph 체크포인트 덕분에 일부 노드 실패 시 전체 재시작이 필요 없습니다.
4. 관측 가능성(Observability): 모듈 단위로 토큰·비용·오류를 추적해 튜닝이 쉽습니다.
5. 인간 협업(HITL): 중요한 단계마다 pause() → 승인 → resume() 패턴을 넣어 책임 있는 AI 시스템을 구현합니다.

실무 사례 — “논문 조사 → 수식 계산 → 시각화 → 보고서 작성”을 각기 다른 에이전트로 나눠 병렬·순차적으로 수행했을 때 단일 거대 에이전트 대비 속도 3 배, 비용 40 % 절감이 보고되었습니다.

---

2. 패턴 개요 & 선택 체크리스트

2‑1. 세 가지 대표 패턴

패턴	구조	장점	유의사항
Supervisor	중앙 감독관이 하위 에이전트를 Tool처럼 호출	로직 단순, 디버깅 용이, HITL 삽입 편리	감독관 프롬프트 복잡도↑, 병목 가능성
Network	에이전트 간 자유로운 메시지 교환	창의·탐색 문제에 강점, 상호 피드백으로 품질↑	루프 방지 및 Edge Rule 설계 필요
Hierarchical	상위→하위 트리 + 병렬 작업 + 결과 통합	대용량 데이터 분산, 리소스 이용 최적	Aggregator 품질·토큰 비용 관리 필요

2‑2. 패턴 선택 4단계

1. 문제 성격: 해답 경로가 명확하면 Supervisor, 불확실·창의적이면 Network.
2. 데이터 규모: 수천 문서 이상이라면 Hierarchical로 샤딩.
3. 응답 지연: Latency < 5 s 요구 시 병렬 구조 필수.
4. HITL 요구: 검수·승인 단계가 있으면 Supervisor 또는 Hierarchical 상위 노드에 HITL 배치.

---

3. Supervisor 패턴 — 설계 원칙 & 예제

3‑1. 설계 원칙

• 단일 진실 원천(SSOT): Supervisor만 전체 목표를 알고, 하위 에이전트는 각 단계 목적만 알도록 설계합니다.
• Tool 계약: 입력·출력 스키마를 엄격히 정의해 에러를 최소화합니다.
• Loop Guard: max_iterations, max_execution_time으로 무한 반복을 방지합니다.
• 메모리 전략: Supervisor에만 Thread Memory를 두고, 하위 에이전트는 필요한 최소 메모리만 사용합니다.

3‑2. 예제 코드

# supervisor_workflow.py — 논문 요약 + 나이 계산
from langgraph.graph import StateGraph
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.memory import ConversationBufferMemory

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 하위 에이전트
a_research = create_react_agent(llm, tools=[], system_message="You are a diligent researcher.")
a_calc     = create_react_agent(llm, tools=[], system_message="You are a precise mathematician.")

@tool
def run_research(topic: str) -> str:
    """주제를 조사해 3문장으로 요약합니다."""
    return a_research.invoke({"input": topic})

@tool
def run_calc(expr: str) -> str:
    """수식 결과를 반환합니다."""
    return a_calc.invoke({"input": expr})

supervisor = create_react_agent(
    llm,
    tools=[run_research, run_calc],
    system_message=(
        "You are a supervisor.\n"
        "1) 문제를 research·calculate 단계로 분할한다.\n"
        "2) 툴을 순서대로 호출한다.\n"
        "3) 최종 답을 한국어로만 출력한다."
    ),
    max_iterations=6,
)

memory  = ConversationBufferMemory(k=10, return_messages=True)
workflow = StateGraph(supervisor, memory=memory, storage="sqlite:///sup.db")

if __name__ == "__main__":
    question = "스티브 잡스의 생애를 3줄 요약하고 나이를 계산해 줘"
    print(workflow.invoke({"input": question}))

Tip: storage를 영속 볼륨에 두어 컨테이너 재시작에도 상태를 보존하세요.

3‑3. 실행 로그 (LangSmith)

12:04:01 Supervisor ➡️ run_research("Steve Jobs biography")
12:04:03 Researcher ⬅️ "Steve Jobs는…"
12:04:04 Supervisor ➡️ run_calc("(2011-10-05)-(1955-02-24)")
12:04:05 Calculator ⬅️ "56"
12:04:05 Supervisor ⬅️ 최종 답변 출력

---

4. Network 패턴 — 동적 협업 & 피드백 루프

4‑1. 구조

User ► Researcher ► Critic
        ▲──────────┘

1. Researcher: 초안 작성 → tag:draft.
2. Critic: 문제 지적·개선 → tag:critique.
3. Researcher: 수정·최종안 → tag:final → 사용자.

4‑2. 코드 핵심

graph = StateGraph(memory="sqlite:///net.db")

graph.add_node("research", researcher_agent)
graph.add_node("critic", critic_agent)

graph.set_conditional_edges("research", lambda m: "draft" in m["tags"], target="critic")

graph.set_conditional_edges("critic", lambda m: "critique" in m["tags"], target="research")

graph.stop_when(lambda s: s["round"] >= 3 or "final" in s["tags"])

Loop 방지: 최대 라운드 + final 태그 두 조건으로 안전 장치.

4‑3. 품질 팁

• Temperature 조절: Researcher T = 0.7, Critic T = 0.0.
• 결과 캐시: Critic 결과 Redis TTL 600 s.
• 전역 타임아웃: 전체 실행 60 초 제한.

---

5. Hierarchical 패턴 — 대규모 병렬 처리 & 통합

5‑1. 다이어그램

Manager ─┬ Worker(10) → Summary₁
         ├ Worker(10) → Summary₂
         ⋮                     ⋮
         └ Worker(10) → Summary₁₀
                      ▼
                Aggregator → 1‑Page 요약

5‑2. 코드 스케치

def worker_subgraph(urls):
    retriever  = WebRetriever(urls)
    summarizer = create_react_agent(llm, tools=[], system_message="Summarize in Korean")
    return StateGraph(summarizer).bind_urls(urls)

parallel = graph.add_parallel([
    worker_subgraph(urls[i:i+10]) for i in range(0, 100, 10)
])

graph.add_node("merge", aggregator_agent)
graph.set_edges(parallel, "merge")

• Concurrency 제어: max_concurrency=5.
• Prefetch: URL 다운로드를 asyncio gather로 미리 실행.
• Compression: Worker 출력 500 자 제한.

5‑3. 운영 경험

• 토큰 예상: Worker ≈ 50 k, Aggregator ≈ 5 k.
• 체크포인트: At Node + DuckDB가 속도·안정성 균형.

---

6. 성능·비용·안정성 최적화

항목	전략	효과
LLM 호출	Batch Embedding 96	TPS ↑ 30 %
스토리지	DuckDB + WAL 모드	동시성 ↑ / 잠금 ↓
캐싱	Tool 결과 Redis TTL 300 s	비용 ↓ 15 %
Failover	retry(backoff=exp) + 대체 모델	가용성 확보
모니터링	LangSmith Trace + Prometheus	비용 초과 예방

---

7. 보안·HITL·운영 모범 사례

1. 권한 격리: Worker 컨테이너를 읽기 전용 파일 시스템으로 실행합니다.
2. PII 마스킹: Aggregator 전 단계에서 민감 정보 정규식 제거.
3. HITL 게이트: pause(tag="needs‑approval") + Slack 승인.
4. 관측 태그: with_config(tags=[...])로 노드별 비용 추적.
5. 단위 테스트: FakeListLLM으로 로직 검증 후 실제 모델 교체.