가이드2026년 3월 28일
멀티에이전트 시스템 설계: 여러 AI가 협업하는 아키텍처 패턴
## 멀티에이전트가 필요한 이유
단일 LLM은 컨텍스트 한계, 전문성 한계, 병렬 처리 불가라는 세 가지 벽에 막힙니다.
```mermaid
flowchart LR
Problem[복잡한 태스크] --> Single[단일 LLM]
Single --> Limit1[컨텍스트 한계<br/>128K~200K 토큰]
Single --> Limit2[전문성 부족<br/>모든 것을 잘 할 수 없음]
Single --> Limit3[순차 처리<br/>병렬화 불가]
Problem --> Multi[멀티에이전트]
Multi --> Sol1[태스크 분할<br/>각자 작은 컨텍스트]
Multi --> Sol2[역할 특화<br/>전문 에이전트]
Multi --> Sol3[병렬 실행<br/>속도 향상]
```
---
## 핵심 멀티에이전트 패턴
### 패턴 1: 오케스트레이터-워커
가장 흔하고 안정적인 패턴. 중앙 에이전트가 태스크를 분배합니다.
```mermaid
flowchart TD
User[사용자] --> Orch[오케스트레이터<br/>Claude Opus]
Orch --> W1[리서치 에이전트<br/>웹 검색 전담]
Orch --> W2[분석 에이전트<br/>데이터 처리 전담]
Orch --> W3[작성 에이전트<br/>문서 생성 전담]
W1 --> Orch
W2 --> Orch
W3 --> Orch
Orch --> Result[최종 결과]
```
```python
import anthropic
client = anthropic.Anthropic()
def orchestrator(task: str) -> str:
# 중앙 오케스트레이터: 태스크를 분석하고 워커에게 위임
response = client.messages.create(
model="claude-opus-4-5-20251001", # 강력한 모델로 조율
max_tokens=2048,
system='''당신은 태스크 조율자입니다.
복잡한 태스크를 분석하고 다음 형식으로 서브태스크를 정의하세요:
{"subtasks": [{"agent": "research|analysis|writing", "task": "..."}]}''',
messages=[{"role": "user", "content": task}]
)
import json
plan = json.loads(response.content[0].text)
results = {}
for subtask in plan["subtasks"]:
agent_type = subtask["agent"]
results[agent_type] = worker(agent_type, subtask["task"])
# 결과 통합
return synthesize(task, results)
def worker(agent_type: str, task: str) -> str:
# 특화된 워커 에이전트
system_prompts = {
"research": "당신은 정보 검색 전문가입니다. 정확한 사실만 제공하세요.",
"analysis": "당신은 데이터 분석 전문가입니다. 수치와 패턴을 분석하세요.",
"writing": "당신은 콘텐츠 작성 전문가입니다. 명확하고 간결하게 작성하세요.",
}
response = client.messages.create(
model="claude-haiku-4-5-20251001", # 워커는 빠른 모델
max_tokens=1024,
system=system_prompts[agent_type],
messages=[{"role": "user", "content": task}]
)
return response.content[0].text
```
---
### 패턴 2: 파이프라인 (체인)
출력이 다음 에이전트의 입력이 되는 순차 처리:
```mermaid
flowchart LR
Raw[원본 데이터] --> A1[수집 에이전트<br/>웹 스크래핑]
A1 --> A2[정제 에이전트<br/>노이즈 제거]
A2 --> A3[분석 에이전트<br/>인사이트 추출]
A3 --> A4[리포트 에이전트<br/>문서 생성]
A4 --> Output[최종 리포트]
```
```python
from dataclasses import dataclass
from typing import Callable
@dataclass
class PipelineStage:
name: str
agent: Callable
model: str = "claude-haiku-4-5-20251001"
def run_pipeline(input_data: str, stages: list[PipelineStage]) -> str:
current = input_data
for stage in stages:
print(f"[{stage.name}] 처리 중...")
current = stage.agent(current, stage.model)
return current
# 뉴스 처리 파이프라인 예시
pipeline = [
PipelineStage("수집", collect_articles),
PipelineStage("필터링", filter_relevant, model="claude-haiku-4-5-20251001"),
PipelineStage("요약", summarize_articles, model="claude-sonnet-4-6-20251001"),
PipelineStage("번역", translate_to_korean, model="claude-haiku-4-5-20251001"),
]
result = run_pipeline("AI 최신 뉴스", pipeline)
```
---
### 패턴 3: 검토자-수정자
한 에이전트가 작성하고, 다른 에이전트가 검토하는 품질 보증 패턴:
```python
def draft_and_review(task: str, max_iterations: int = 3) -> str:
draft = None
feedback = None
for i in range(max_iterations):
# 1. 초안 작성
draft_response = client.messages.create(
model="claude-sonnet-4-6-20251001",
max_tokens=2048,
system="당신은 전문 작가입니다. 요청받은 내용을 작성하세요.",
messages=[
{"role": "user", "content": task},
*([{"role": "assistant", "content": draft},
{"role": "user", "content": f"피드백 반영해서 수정:
{feedback}"}]
if draft and feedback else [])
]
)
draft = draft_response.content[0].text
# 2. 품질 검토
review_response = client.messages.create(
model="claude-opus-4-5-20251001", # 검토는 더 강한 모델
max_tokens=1024,
system="당신은 엄격한 편집자입니다. 개선점을 구체적으로 지적하세요. 충분히 좋으면 'APPROVED'라고만 답하세요.",
messages=[{"role": "user", "content": f"원본 요청: {task}
초안:
{draft}"}]
)
feedback = review_response.content[0].text
if "APPROVED" in feedback:
print(f"✓ {i+1}번 만에 승인됨")
break
return draft
```
---
## CrewAI로 역할 기반 에이전트 팀 구성
CrewAI는 역할(Role), 목표(Goal), 배경(Backstory)으로 에이전트를 정의합니다:
```python
from crewai import Agent, Task, Crew
# 에이전트 정의
researcher = Agent(
role="AI 리서처",
goal="최신 AI 동향을 정확하게 조사한다",
backstory="5년 경력의 AI 연구원. 논문과 기술 블로그를 빠르게 파악하는 능력이 있다.",
tools=[web_search_tool, arxiv_tool],
llm="claude-haiku-4-5-20251001",
verbose=True
)
analyst = Agent(
role="데이터 분석가",
goal="수집된 정보에서 핵심 인사이트를 도출한다",
backstory="통계와 데이터 분석 전문가. 숫자에서 의미를 찾는다.",
llm="claude-sonnet-4-6-20251001",
)
writer = Agent(
role="콘텐츠 라이터",
goal="분석 결과를 읽기 쉬운 한국어 보고서로 작성한다",
backstory="IT 전문 기자 출신. 복잡한 기술 내용을 쉽게 설명한다.",
llm="claude-sonnet-4-6-20251001",
)
# 태스크 정의
research_task = Task(
description="이번 주 발표된 AI 모델 중 주목할 만한 것 5개를 조사하세요",
expected_output="모델명, 주요 특징, 성능 지표가 포함된 조사 결과",
agent=researcher
)
analysis_task = Task(
description="조사 결과를 바탕으로 한국 개발자에게 가장 중요한 트렌드 3가지를 분석하세요",
expected_output="트렌드별 중요도와 활용 가능성 분석",
agent=analyst,
context=[research_task] # 이전 태스크 결과를 컨텍스트로 사용
)
write_task = Task(
description="분석 결과를 2,000자 내외의 한국어 뉴스레터로 작성하세요",
expected_output="마크다운 형식의 뉴스레터",
agent=writer,
context=[research_task, analysis_task]
)
# 크루 실행
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, write_task],
verbose=True
)
result = crew.kickoff()
```
---
## 비용과 레이턴시 관리
멀티에이전트의 가장 큰 단점은 비용과 시간입니다:
```mermaid
flowchart LR
Single[단일 GPT-4o 호출<br/>1초, $0.02] --> Multi[멀티에이전트 5개 호출<br/>5-15초, $0.10]
```
**최적화 전략:**
1. **오케스트레이터는 강한 모델, 워커는 약한 모델**
- 오케스트레이터: Claude Opus / GPT-4o
- 워커: Claude Haiku / GPT-4o-mini (비용 10배 절감)
2. **병렬 실행 가능한 태스크는 동시 실행**
```python
import asyncio
async def run_parallel_agents(tasks):
return await asyncio.gather(*[agent.run(t) for t in tasks])
```
3. **중간 결과 캐싱**
- 동일 입력에 대한 에이전트 결과를 Redis에 저장
---
## 결론멀티에이전트 시스템 설계 원칙:
1. **단순하게 시작**: 단일 에이전트로 먼저 해결 시도, 한계에 부딪히면 분리
2. **역할 명확히**: 각 에이전트의 역할, 입력, 출력을 명확히 정의
3. **실패 처리**: 에이전트가 실패했을 때의 폴백 로직 필수
4. **비용 모니터링**: 에이전트 호출당 토큰/비용 추적
5. **사람 확인 포인트**: 불가역적 작업 전 사람 승인 단계
복잡할수록 관리하기 어렵습니다. 3~5개 에이전트로 시작해서 점진적으로 확장하세요.