Multi-Agent & Workflow

대상 독자: AI 시스템을 설계하거나 도입하려는 개발자, 아키텍트, 프로덕트 매니저

1. 개요 및 핵심 개념

1.1 왜 멀티 에이전트인가?

단일 LLM 호출로 처리하기 어려운 작업들이 존재한다.

• 컨텍스트 한계 초과: 수십만 토큰을 요구하는 장문 분석
• 병렬 처리 필요: 독립적인 서브태스크를 동시에 수행
• 전문화 요구: 각기 다른 역할(조사, 작성, 검토, 실행)을 분리
• 긴 추론 체인: 단계별로 검증이 필요한 복잡한 문제

멀티 에이전트는 이 문제를 분할 정복(Divide & Conquer) 방식으로 해결한다.

1.2 핵심 용어 정의

2. 멀티 에이전트 시스템

2.1 에이전트의 구성 요소

┌──────────────────────────────────────┐
│              에이전트                 │
│                                      │
│  ┌─────────┐    ┌─────────────────┐  │
│  │  LLM    │◄──►│  시스템 프롬프트  │  │
│  └────┬────┘    └─────────────────┘  │
│       │                              │
│  ┌────▼────┐    ┌─────────────────┐  │
│  │ 도구 호출 │◄──►│  메모리/컨텍스트  │  │
│  └────┬────┘    └─────────────────┘  │
│       │                              │
│  ┌────▼────────────────────────────┐ │
│  │  외부 도구 (API, DB, 검색, 코드) │ │
│  └─────────────────────────────────┘ │
└──────────────────────────────────────┘

2.2 에이전트 역할 유형

오케스트레이터 에이전트

• 전체 작업 계획 수립
• 서브 에이전트에게 작업 위임
• 결과 수집 및 통합

전문가(Specialist) 에이전트

• 특정 도메인에 최적화된 시스템 프롬프트
• 제한된 도구셋만 보유
• 단일 역할 집중 (예: 코드 작성, 데이터 분석, 법률 검토)

검증(Verifier) 에이전트

• 다른 에이전트의 출력을 검토
• 오류 탐지 및 품질 보증
• 사실 확인 및 일관성 검증

실행(Executor) 에이전트

• 계획을 실제 액션으로 변환
• 코드 실행, API 호출, 파일 조작
• 부작용이 있는 작업 수행

2.3 에이전트 간 통신 방식

동기(Synchronous) 방식

에이전트 A → 요청 → 에이전트 B → 응답 → 에이전트 A

• 순서가 중요할 때 사용
• 결과를 즉시 다음 단계에 반영

비동기(Asynchronous) 방식

에이전트 A → 태스크 큐 → 에이전트 B, C, D (병렬)
                              ↓
                        결과 수집기

• 독립적인 서브태스크 병렬 처리
• 전체 처리 시간 단축

이벤트 기반(Event-Driven) 방식

에이전트 A → 이벤트 발행 → 이벤트 버스 → 구독 에이전트들

• 느슨한 결합(Loose Coupling)
• 동적인 에이전트 추가/제거 가능

3. AI 워크플로우 설계

3.1 워크플로우 유형

순차 워크플로우 (Sequential)

입력 → [에이전트 1] → [에이전트 2] → [에이전트 3] → 출력

적합한 경우:

• 각 단계가 이전 단계 결과에 의존
• 순서가 중요한 파이프라인 (예: 조사 → 초안 작성 → 편집 → 검토)

병렬 워크플로우 (Parallel)

         ┌─→ [에이전트 A] ─┐
입력 ─→ │─→ [에이전트 B] ─│─→ [통합] → 출력
         └─→ [에이전트 C] ─┘

적합한 경우:

• 독립적인 서브태스크 (예: 여러 소스 동시 조사)
• 처리 속도가 중요한 경우

분기 워크플로우 (Conditional Branching)

입력 → [분류기] → 조건 A? → [전문가 A]
                → 조건 B? → [전문가 B]
                → 기본    → [범용 에이전트]

적합한 경우:

• 입력 유형에 따라 다른 처리 필요
• 전문화된 에이전트 라우팅

루프 워크플로우 (Iterative Loop)

입력 → [에이전트] → 품질 검사 → 충분? → 출력
                    ↑              ↓ (불충분)
                    └──────────────┘

적합한 경우:

• 반복적 개선이 필요한 작업
• 명확한 완료 조건이 있는 경우

3.2 워크플로우 설계 원칙

원자성 (Atomicity)

• 각 에이전트 태스크는 단일 책임을 가져야 한다
• 실패 시 명확한 롤백 지점 확보

멱등성 (Idempotency)

• 동일한 입력에 대해 동일한 결과 보장
• 재시도 시 부작용 방지

실패 격리 (Failure Isolation)

• 한 에이전트의 실패가 전체를 멈추지 않도록 설계
• 대체 경로(Fallback) 정의

3.3 상태 관리

워크플로우 실행 중 상태를 어떻게 관리할 것인가는 핵심 설계 결정이다.

# 워크플로우 상태 예시
workflow_state:
  id: "wf_20260309_001"
  status: "in_progress"  # pending | in_progress | completed | failed
  current_step: "step_3_verification"
  
  steps:
    - name: "step_1_research"
      status: "completed"
      output: "..."
      
    - name: "step_2_draft"
      status: "completed"
      output: "..."
      
    - name: "step_3_verification"
      status: "in_progress"
      started_at: "2026-03-09T10:30:00Z"
      
  metadata:
    total_tokens: 45230
    elapsed_seconds: 127
    retry_count: 0

4. 오케스트레이션 패턴

4.1 중앙집중형 오케스트레이션

구조:

                ┌─────────────────┐
                │   오케스트레이터  │
                └────────┬────────┘
           ┌─────────────┼─────────────┐
    ┌──────▼─────┐ ┌─────▼──────┐ ┌───▼────────┐
    │  에이전트 A │ │ 에이전트 B  │ │ 에이전트 C  │
    └────────────┘ └────────────┘ └────────────┘

특징:

• 오케스트레이터가 모든 에이전트를 직접 제어
• 전체 상태를 단일 지점에서 파악 가능
• 병목 현상 가능성, 오케스트레이터 실패 시 전체 중단

구현 예시 (Python Pseudocode):

class Orchestrator:
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewerAgent()
        }
    
    async def run(self, task: str) -> str:
        # 1단계: 조사
        research = await self.agents["researcher"].run(task)
        
        # 2단계: 작성
        draft = await self.agents["writer"].run(
            task=task, 
            context=research
        )
        
        # 3단계: 검토 (품질 기준 충족까지 반복)
        for attempt in range(3):
            review = await self.agents["reviewer"].run(draft)
            if review.approved:
                return draft
            draft = await self.agents["writer"].revise(review.feedback)
        
        return draft

4.2 계층형 오케스트레이션

구조:

           [최상위 오케스트레이터]
           /                    \
  [서브 오케스트레이터 A]    [서브 오케스트레이터 B]
   /            \               /            \
[에이전트 1] [에이전트 2]  [에이전트 3] [에이전트 4]

특징:

• 대규모 복잡한 작업에 적합
• 책임을 계층별로 분리
• 각 계층이 독립적으로 확장 가능

적용 사례:

• 대규모 소프트웨어 개발 자동화
• 복잡한 비즈니스 프로세스 처리
• 다단계 콘텐츠 생산 파이프라인

4.3 피어-투-피어 오케스트레이션

구조:

[에이전트 A] ←──→ [에이전트 B]
     ↕                  ↕
[에이전트 C] ←──→ [에이전트 D]

특징:

• 중앙 조정자 없이 에이전트 간 직접 통신
• 높은 유연성, 분산 처리 가능
• 상태 추적이 복잡, 디버깅 어려움

4.4 이벤트 기반 오케스트레이션

구조:

[에이전트들] → 이벤트 발행 → [이벤트 버스] → 이벤트 소비 → [에이전트들]

구현 예시:

# 이벤트 정의
@dataclass
class ResearchCompleted:
    task_id: str
    results: list[str]
    timestamp: datetime

# 이벤트 핸들러 등록
@event_bus.subscribe(ResearchCompleted)
async def on_research_done(event: ResearchCompleted):
    await writer_agent.start_draft(event.results)

# 이벤트 발행
await event_bus.publish(ResearchCompleted(
    task_id="task_001",
    results=research_results,
    timestamp=datetime.now()
))

4.5 패턴 선택 가이드

5. 주요 프레임워크 및 도구

5.1 LangGraph

개요: LangChain 생태계의 그래프 기반 에이전트 오케스트레이션 프레임워크

핵심 특징:

• 상태 머신(State Machine) 기반 워크플로우
• 사이클(Cycle)과 분기(Branch) 지원
• 내장 체크포인팅 및 인간 개입(Human-in-the-loop)

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    messages: list
    current_step: str
    result: str

# 그래프 정의
workflow = StateGraph(AgentState)
workflow.add_node("researcher", research_node)
workflow.add_node("writer", writing_node)
workflow.add_node("reviewer", review_node)

# 엣지 정의
workflow.add_edge("researcher", "writer")
workflow.add_conditional_edges(
    "reviewer",
    lambda state: "approved" if state["approved"] else "needs_revision",
    {
        "approved": END,
        "needs_revision": "writer"
    }
)

app = workflow.compile()

적합한 경우: 상태 기반 복잡한 워크플로우, 반복적 처리, 조건 분기가 많은 경우

5.2 AutoGen (Microsoft)

개요: 마이크로소프트가 개발한 멀티 에이전트 대화 프레임워크

핵심 특징:

• 에이전트 간 자연어 대화를 통한 협업
• 코드 실행 에이전트 내장
• 그룹 채팅(Group Chat) 패턴 지원

import autogen

# 에이전트 정의
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"model": "claude-sonnet-4-6"}
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "coding"}
)

# 대화 시작
user_proxy.initiate_chat(
    assistant,
    message="Python으로 데이터 분석 스크립트를 작성해줘"
)

적합한 경우: 코드 생성/실행이 포함된 작업, 에이전트 간 자유로운 대화 필요 시

5.3 CrewAI

개요: 역할 기반 에이전트 협업에 특화된 프레임워크

핵심 특징:

• 직관적인 역할/목표 기반 에이전트 정의
• 태스크 위임(Delegation) 내장
• 순차/병렬 프로세스 지원

from crewai import Agent, Task, Crew, Process

# 역할 기반 에이전트 정의
researcher = Agent(
    role="수석 연구원",
    goal="주어진 주제에 대한 심층 조사 수행",
    backstory="10년 경력의 시장 조사 전문가",
    tools=[search_tool, web_scraper]
)

writer = Agent(
    role="콘텐츠 작가",
    goal="연구 결과를 바탕으로 고품질 보고서 작성",
    backstory="기술 문서 전문 작가"
)

# 태스크 정의
research_task = Task(
    description="AI 오케스트레이션 시장 트렌드 조사",
    agent=researcher
)

writing_task = Task(
    description="조사 결과를 바탕으로 5페이지 보고서 작성",
    agent=writer,
    context=[research_task]
)

# 크루 실행
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential
)

result = crew.kickoff()

적합한 경우: 비즈니스 프로세스 자동화, 직관적인 역할 모델이 필요한 경우

5.4 Anthropic Claude API (직접 구현)

개요: Claude API의 도구 사용(Tool Use) 기능을 직접 활용한 커스텀 오케스트레이션

import anthropic

client = anthropic.Anthropic()

# 도구 정의
tools = [
    {
        "name": "search_web",
        "description": "웹에서 정보를 검색합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "handoff_to_agent",
        "description": "다른 전문 에이전트에게 작업을 넘깁니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "agent_name": {"type": "string"},
                "task": {"type": "string"},
                "context": {"type": "string"}
            },
            "required": ["agent_name", "task"]
        }
    }
]

# 에이전트 루프
def run_agent(system_prompt: str, user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system_prompt,
            tools=tools,
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            return response.content[0].text
            
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

5.5 프레임워크 비교

6. 아키텍처 설계 원칙

6.1 최소 권한 원칙 (Principle of Least Privilege)

각 에이전트는 자신의 역할 수행에 필요한 최소한의 도구와 권한만 부여받아야 한다.

# 잘못된 설계
research_agent:
  tools:
    - web_search
    - database_read
    - database_write      # 불필요
    - send_email          # 불필요
    - execute_code        # 불필요

# 올바른 설계
research_agent:
  tools:
    - web_search
    - database_read       # 읽기만 허용

6.2 명시적 핸드오프

에이전트 간 작업 전환은 명시적이고 구조화된 방식으로 이루어져야 한다.

# 핸드오프 스키마 예시
@dataclass
class Handoff:
    from_agent: str
    to_agent: str
    task_description: str
    context: dict
    artifacts: list[str]
    priority: str  # "high" | "normal" | "low"
    timeout_seconds: int

6.3 실패 복원력 (Resilience)

재시도 전략:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_agent(agent, task):
    return await agent.run(task)

서킷 브레이커:

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.state = "closed"  # closed | open | half_open
        
    def call(self, func, *args):
        if self.state == "open":
            raise CircuitOpenError("에이전트 호출 차단 중")
        try:
            result = func(*args)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

6.4 컨텍스트 관리

에이전트 간 컨텍스트 전달 시 필요한 정보만 선택적으로 전달한다.

def summarize_for_handoff(full_context: dict) -> dict:
    """핸드오프용 컨텍스트 압축"""
    return {
        "task_objective": full_context["objective"],
        "key_findings": full_context["findings"][:5],  # 상위 5개만
        "constraints": full_context["constraints"],
        "previous_attempts": len(full_context["history"])
        # 전체 히스토리는 전달하지 않음
    }

7. 보안 및 신뢰 경계

7.1 프롬프트 인젝션 방어

멀티 에이전트 환경에서 외부 데이터(웹 검색 결과, 사용자 입력 등)가 에이전트의 지시를 조작하려는 시도에 주의해야 한다.

방어 전략:

1. 입력 검증: 외부 데이터를 구조화된 형식(JSON, XML)으로만 수용
2. 샌드박스 실행: 에이전트가 실행하는 코드는 격리된 환경에서 실행
3. 출력 검증: 에이전트 출력을 다음 단계 전달 전 스키마 검증

def validate_agent_output(output: str, schema: dict) -> bool:
    """에이전트 출력이 예상 스키마를 따르는지 검증"""
    try:
        parsed = json.loads(output)
        jsonschema.validate(parsed, schema)
        return True
    except (json.JSONDecodeError, jsonschema.ValidationError):
        return False

7.2 에이전트 신뢰 수준 분류

7.3 인간 개입 (Human-in-the-Loop)

불가역적인 작업이나 높은 위험도의 결정은 반드시 인간 승인을 거쳐야 한다.

HIGH_RISK_ACTIONS = [
    "send_email_to_all",
    "delete_database",
    "deploy_to_production",
    "financial_transaction"
]

async def execute_action(action: str, params: dict):
    if action in HIGH_RISK_ACTIONS:
        approval = await request_human_approval(
            action=action,
            params=params,
            timeout=300  # 5분 내 승인 필요
        )
        if not approval.approved:
            raise ActionDeniedError(f"작업 거부됨: {approval.reason}")
    
    return await perform_action(action, params)

8. 모니터링 및 관찰가능성

8.1 추적해야 할 핵심 지표

성능 지표:

• 전체 워크플로우 완료 시간
• 에이전트별 응답 지연(Latency)
• 토큰 사용량 (에이전트별, 워크플로우별)
• 병렬 처리 효율성

품질 지표:

• 태스크 완료율
• 재시도 횟수
• 오류 발생률
• 인간 개입 빈도

비용 지표:

• 워크플로우당 API 비용
• 에이전트별 비용 분배
• 비용 대비 성능(Cost per Quality Score)

8.2 분산 추적 (Distributed Tracing)

from opentelemetry import trace

tracer = trace.get_tracer("multi-agent-system")

async def run_agent_with_tracing(agent_name: str, task: str):
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("task.description", task[:100])
        
        try:
            result = await agents[agent_name].run(task)
            span.set_attribute("result.success", True)
            span.set_attribute("result.tokens", result.token_count)
            return result
        except Exception as e:
            span.set_attribute("result.success", False)
            span.record_exception(e)
            raise

8.3 로깅 구조

{
  "timestamp": "2026-03-09T10:30:00Z",
  "workflow_id": "wf_001",
  "agent_id": "researcher_agent",
  "event_type": "tool_call",
  "tool_name": "web_search",
  "input": {"query": "AI orchestration patterns 2026"},
  "output_summary": "Found 5 relevant articles",
  "tokens_used": 1250,
  "latency_ms": 2340,
  "success": true
}

8.4 알림 설정

alerts:
  - name: "워크플로우 장시간 실행"
    condition: "workflow_duration > 300s"
    severity: "warning"
    
  - name: "에이전트 오류율 급증"
    condition: "error_rate > 10% in last 5min"
    severity: "critical"
    
  - name: "비용 임계값 초과"
    condition: "daily_cost > $100"
    severity: "warning"
    
  - name: "토큰 한계 근접"
    condition: "token_usage > 80% of limit"
    severity: "info"

9. 실전 패턴 및 안티패턴

9.1 권장 패턴

패턴 1: 검증-수정 루프 (Reflection Pattern)

생성 후 자기 검토를 통해 품질을 높인다.

생성 → 자기 검토 → 기준 충족? → 완료
                  → 미충족    → 수정 → 재검토 (최대 N회)

패턴 2: 전문가 앙상블 (Expert Ensemble)

동일 태스크를 여러 전문가 에이전트가 처리 후 결과를 통합한다.

태스크 → [전문가 A] ─┐
       → [전문가 B] ─│→ [통합 에이전트] → 최종 결과
       → [전문가 C] ─┘

패턴 3: 계획-실행 분리 (Plan-Execute Separation)

계획 단계와 실행 단계를 명확히 분리한다.

사용자 입력 → [계획 에이전트] → 승인? → [실행 에이전트들]
                               ↓
                          계획 수정 요청

패턴 4: 체크포인트 (Checkpoint Pattern)

장시간 워크플로우에서 중간 상태를 저장한다.

async def workflow_with_checkpoints(task):
    checkpoint = load_checkpoint(task.id) or {}
    
    if "research" not in checkpoint:
        checkpoint["research"] = await research_agent.run(task)
        save_checkpoint(task.id, checkpoint)
    
    if "draft" not in checkpoint:
        checkpoint["draft"] = await writer_agent.run(checkpoint["research"])
        save_checkpoint(task.id, checkpoint)
    
    return checkpoint["draft"]

9.2 안티패턴

안티패턴 1: 에이전트 스파게티

에이전트 간 의존 관계가 너무 복잡하게 얽혀 이해/디버깅이 불가능한 상태.

해결: 계층 구조를 명확히 하고, 순환 의존성을 제거한다.

안티패턴 2: 과도한 컨텍스트 전달

모든 에이전트에게 전체 히스토리를 전달하여 토큰을 낭비하고 성능을 저하시킨다.

해결: 에이전트에게 필요한 정보만 선택적으로 전달한다.

안티패턴 3: 오케스트레이터 과부하

오케스트레이터가 너무 많은 책임을 지며 병목이 되는 경우.

해결: 오케스트레이터는 조율만 담당하고, 실제 작업은 전문 에이전트에게 위임한다.

안티패턴 4: 무한 루프

종료 조건이 없는 에이전트 루프.

# 잘못된 코드
while not is_perfect(result):
    result = improve(result)

# 올바른 코드
MAX_ITERATIONS = 5
for i in range(MAX_ITERATIONS):
    if is_good_enough(result):
        break
    result = improve(result)

안티패턴 5: 블라인드 트러스트

에이전트 출력을 검증 없이 다음 단계로 전달.

해결: 에이전트 간 경계에서 항상 스키마 검증을 수행한다.

10. 로드맵 및 미래 방향

10.1 현재 트렌드 (2025~2026)

• 컴퓨터 사용(Computer Use): 에이전트가 실제 GUI 환경을 조작하는 능력 확대
• 장기 메모리: 세션을 넘어 지속되는 에이전트 메모리 표준화
• 에이전트 마켓플레이스: 전문화된 에이전트를 조합하는 생태계 형성
• 비용 효율 최적화: 소형 모델과 대형 모델의 하이브리드 오케스트레이션

10.2 에이전트 성숙도 모델

레벨 1: 단일 에이전트
   └→ 단순 도구 사용, 단일 LLM 호출

레벨 2: 기본 멀티 에이전트
   └→ 2~3개 에이전트 순차 파이프라인

레벨 3: 오케스트레이션 시스템
   └→ 조건 분기, 병렬 처리, 재시도 로직

레벨 4: 자율 에이전트 시스템
   └→ 동적 계획 수립, 자기 개선, 장기 기억

레벨 5: 에이전트 네트워크
   └→ 분산 에이전트 협업, 자율적 조정

10.3 도입 로드맵 (권장)

1~2주차: 기반 구축
  ├ 단일 에이전트 + 도구 사용 프로토타입
  └ 로깅 및 모니터링 기본 설정

3~4주차: 워크플로우 구성
  ├ 2~3개 에이전트 파이프라인 구현
  └ 오류 처리 및 재시도 로직 추가

2개월차: 오케스트레이션 고도화
  ├ 조건 분기 및 병렬 처리 도입
  └ 인간 개입 포인트 정의

3개월차~: 프로덕션 준비
  ├ 보안 감사 및 신뢰 경계 설정
  ├ 성능 최적화 및 비용 관리
  └ 팀 교육 및 운영 가이드 수립

참고 자료

• Anthropic Claude API 문서
• LangGraph 공식 문서
• AutoGen 공식 문서
• CrewAI 공식 문서
• Anthropic Research: Building Effective Agents (2024)
• OpenAI: Orchestrating agents: routines and handoffs (2024)

본 문서는 현재 AI 에이전트 생태계의 모범 사례를 기반으로 작성되었습니다. 기술의 빠른 발전으로 내용이 변경될 수 있습니다.