🎯 Strands 멀티에이전트

왜 멀티에이전트인가

에이전트 하나에 모든 것을 맡길 때 생기는 세 가지 문제

패턴 지형 — 업계 표준과 Strands

멀티에이전트 협업의 보편 패턴 4가지 — Strands는 전부 내장 구현으로 제공

멀티에이전트 패턴

복잡도·자율성·제어력 3축으로 구분하는 4가지 협업 아키텍처

멀티에이전트 패턴 비교

자율성이 높을수록 제어력이 낮아지는 트레이드오프 — Swarm과 Workflow가 양극단

자율성 vs 구조화 매트릭스

구조화 = 실행 흐름을 사전에 고정하는 정도 · 자율성 = 런타임에 에이전트가 결정하는 정도

패턴 선택 의사결정

첫 질문은 제어를 어디에 둘 것인가 — 예/아니오로 좁혀가는 패턴 선택

Agents-as-Tools

가장 단순한 멀티에이전트 — Supervisor가 하위 에이전트를 도구처럼 부리는 구조

Agents-as-Tools 구조도

하위 에이전트끼리는 서로를 모르는 별형 구조 — 위임과 결과 반환이 전부 Supervisor를 경유

Agents-as-Tools 구현

@tool로 감싼 하위 에이전트를 Supervisor의 tools=에 장착 — str(agent(query))로 텍스트 결과만 반환

핵심 포인트

@tool

함수로 감싼 하위 에이전트를 Supervisor의 도구로 노출

system_prompt

하위 에이전트마다 전문 역할을 별도 지정 — 단일 책임

str(agent(query))

AgentResult에서 텍스트만 추출해 도구 결과로 반환

tools=[...]

Supervisor에 하위 에이전트들을 도구 목록으로 장착

코드

from strands import Agent, tool

@tool
def research_agent(query: str) -> str:
    """리서치 전담 에이전트"""
    return str(Agent(system_prompt="You are a research specialist.")(query))

@tool
def writer_agent(content: str) -> str:
    """작문 전담 에이전트"""
    return str(Agent(system_prompt="You are a technical writer.")(content))

supervisor = Agent(  # 하위 에이전트를 도구로 장착
    system_prompt="Delegate tasks to specialists.",
    tools=[research_agent, writer_agent],
)
result = supervisor("AI 트렌드 리포트를 작성해줘")

Supervisor 위임 흐름

Supervisor가 태스크를 분해하고 하위 에이전트에 순차 위임하는 과정

1. 1
   태스크 수신

   Supervisor가 사용자 요청을 분석

2. 2
   태스크 분해

   하위 작업으로 분할 후 적합한 에이전트 선택

3. 3
   Tool 호출

   하위 에이전트를 tool()로 실행

4. 4
   결과 통합

   각 에이전트 응답을 취합하여 최종 답변 생성

Supervisor와 하위 에이전트 — 컨텍스트·상태 경계

넘어가는 것은 도구 인자와 결과 텍스트뿐 — 격리가 이 패턴의 장점이자 제약

Agents-as-Tools 활용 사례

Supervisor 위임 패턴이 빛나는 시나리오

Swarm

Supervisor 없는 협업 — 스스로 대화를 넘겨받는 에이전트들

Swarm 구조도

리서치 → 코딩 → 리뷰가 공유 컨텍스트를 가운데 두고 도는 자율 핸드오프 순환

Swarm 핸드오프 구현

system_prompt로 전환 시점을 지시하고 entry_point와 max_handoffs·max_iterations로 시작점과 상한을 지정

핵심 포인트

name=

핸드오프 대상을 식별하는 에이전트 이름

system_prompt

언제 다른 에이전트로 넘길지를 프롬프트로 지시

entry_point=

대화를 시작할 진입 에이전트 지정

max_handoffs · max_iterations

전환·반복 횟수 상한 — 폭주 방지

코드

from strands import Agent
from strands.multiagent.swarm import Swarm

sales_agent = Agent(
    name="sales",
    system_prompt="Sales specialist. Transfer to support for technical issues.",
)
support_agent = Agent(
    name="support",
    system_prompt="Support engineer. Transfer to sales for pricing questions.",
)

swarm = Swarm(  # 필요 시 에이전트가 자율적으로 핸드오프
    [sales_agent, support_agent], entry_point=sales_agent,
    max_handoffs=20, max_iterations=20,
)
result = swarm("서버 에러가 계속 발생합니다")

Supervisor vs Swarm

중앙집중 위임과 자율 핸드오프의 구조적 차이

자율 핸드오프 흐름

Swarm 내 에이전트 간 컨텍스트 전달과 전환 메커니즘

1. 1
   초기 에이전트 응답

   entry_point 에이전트가 사용자 메시지 처리 시작

2. 2
   핸드오프 판단

   현재 에이전트가 전환 필요성 자율 결정

3. 3
   컨텍스트 전달

   핸드오프 메시지·원 task·작업 이력·shared_context를 조립해 다음 에이전트에 전달

4. 4
   새 에이전트 활성

   전환된 에이전트가 컨텍스트 기반으로 응답 계속

Swarm의 상태 — 공유 컨텍스트의 실체

핸드오프마다 함께 넘어가는 것과 관리해야 할 것

Graph

노드는 에이전트, 엣지는 관계 — 복잡한 워크플로우의 그래프 표현

Graph 구조도

노드와 엣지로 사전 정의하는 실행 그래프 — 병렬 분기와 합류가 구조에 내장

Graph 구현

GraphBuilder로 노드와 엣지를 정의하는 Graph 패턴

핵심 포인트

add_node(agent, node_id=)

에이전트를 그래프 노드로 등록

add_edge(a, b, condition=)

실행 순서 정의 — condition 함수로 조건부 분기

state.results

조건 함수가 받는 GraphState — 노드 결과를 검사해 bool 반환

set_entry_point / build()

진입 노드 지정 후 그래프 완성·실행

코드

from strands import Agent
from strands.multiagent.graph import GraphBuilder

builder = GraphBuilder()
extract = builder.add_node(Agent(system_prompt="Extract key entities."), node_id="extract")
analyze = builder.add_node(Agent(system_prompt="Analyze relationships."), node_id="analyze")
summarize = builder.add_node(Agent(system_prompt="Generate final summary."), node_id="summarize")

def is_confident(state):  # GraphState를 받아 bool 반환
    node = state.results.get("analyze")
    return node is not None and "confident" in str(node.result)

builder.add_edge(extract, analyze)
builder.add_edge(analyze, summarize, condition=is_confident)
builder.add_edge(analyze, extract, condition=lambda s: not is_confident(s))
builder.set_entry_point("extract")
result = builder.build()("분기별 매출 데이터를 분석해줘")

Graph 3가지 토폴로지

DAG·순환·조건부 — 그래프 토폴로지별 특성과 적용 영역

병렬 실행과 합류

독립 노드의 동시 실행과 결과 합류(Fan-out/Fan-in) 메커니즘

1. 1
   Fan-out

   진입 노드가 독립된 복수 노드를 동시 트리거

2. 2
   병렬 처리

   각 노드가 독립적으로 실행 — 서로 대기하지 않음

3. 3
   Fan-in

   모든 병렬 노드 완료 후 합류 노드에서 결과 집계

4. 4
   후속 처리

   집계된 결과를 다음 노드로 전달하여 최종 출력 생성

Graph의 상태 — 엣지로 흐르는 것과 남는 것

노드 사이에는 출력만 흐르고, 실행 상태는 그래프가 관리하는 구조

Workflow

실행 순서를 보장하고 상태를 디스크에 남기는 프로덕션 지향 패턴

Workflow 구조도

고정 순서의 파이프라인 — 각 태스크 결과가 저장되어 실패 지점을 정확히 파악

1. 1
   수집

   태스크 1 — 결과를 상태 저장소에 기록

2. 2
   분석

   태스크 2 — 이전 태스크 결과를 입력으로 사용

3. 3
   작성

   태스크 3 — 실패하면 이 지점이 상태에 기록

4. 4
   발행

   태스크 4 — 완료 상태 기록으로 종료

Workflow 파이프라인 구현

strands-agents-tools의 workflow 도구로 의존성 기반 태스크 파이프라인 실행

핵심 포인트

workflow

strands-agents-tools가 제공하는 태스크 파이프라인 도구

tools=[workflow]

오케스트레이터에 장착 — 자연어 요청을 도구 호출로 수행

create / start / list / status / delete

workflow 도구가 지원하는 5가지 액션

workflow_id

status 액션으로 파이프라인별 진행 상태 확인

코드

from strands import Agent
from strands_tools import workflow

orchestrator = Agent(
    system_prompt="""You manage data pipelines.
    Use the workflow tool to create and run task pipelines.
    Define tasks with dependencies for execution order.""",
    tools=[workflow],
)

# 워크플로우 생성·실행 — 에이전트가 도구 호출로 수행
result = orchestrator("""
다음 데이터 파이프라인을 실행해줘:
1. ingest → 2. validate → 3. transform → 4. load
각 단계는 이전 단계에 의존합니다.
""")
# 태스크 상태는 디스크에 저장 — status로 workflow_id별 진행 확인

태스크 상태 저장과 추적

태스크 상태를 디스크에 기록해 장애 발생 지점을 파악하는 구조

1. 1
   태스크 실행

   의존성이 해소된 태스크부터 순차 실행

2. 2
   상태 저장

   태스크 완료 시마다 상태를 로컬 디스크에 기록

3. 3
   장애 감지

   타임아웃·예외·외부 실패 시 실행 중단

4. 4
   상태 조회

   status 액션으로 workflow_id별 완료·실패 태스크 확인 후 대응

컨텍스트는 어떻게 넘어가는가 — 패턴별 상태 전달

멀티에이전트 설계의 핵심 질문 — 각 패턴이 컨텍스트와 상태를 넘기는 방식

언제 멈추는가 — 패턴별 종료와 폭주 방지

상한 없는 자율이 부르는 비용 사고 — 패턴마다 다른 종료 조건과 폭주 상한

프로덕션 준비도 비교

장애 복원력·관측 가능성·재현성·수평 확장·디버깅 — 상태를 저장하는 Workflow가 복원력·재현성 우위