Strands 멀티에이전트 운영

멀티에이전트 운영 3축

프로덕션에서 멀티에이전트를 지탱하는 세 가지 핵심 축

1. 1
   통신 (Communication)

   에이전트 간 메시지 전달·공유 상태·프로토콜 표준화. Shared Context부터 A2A Protocol까지 계층별 통신 전략이 필요합니다.

2. 2
   에러 복원 (Resilience)

   타임아웃·환각·핸드오프 유실 등 분산 환경 고유의 실패 모드에 대응. Circuit Breaker와 Supervisor Escalation으로 장애를 격리합니다.

3. 3
   안전 보장 (Safety)

   에이전트별 권한 제어·입출력 검증·행동 제한. Cedar 정책과 Bedrock Guardrails로 자율 행동의 범위를 선언적으로 제한합니다.

축별 핵심 과제

메시지 유실·장애 전파·권한 초과 — 축마다 다른 실패 양상과 대응 패턴

프로토타입 vs 프로덕션

단일 에이전트 PoC에서는 보이지 않던 문제의 프로덕션 폭발 지점

통신 3계층 모델

결합도와 확장성에 따라 계층을 선택하는 통신 아키텍처

A2A Protocol 통합

A2AServer가 Strands 에이전트를 JSON-RPC over HTTP로 노출 — Agent Card 자동 게시와 Task 단위 협업 수신

핵심 포인트

A2AServer

Strands 에이전트를 JSON-RPC over HTTP의 A2A 서버로 노출

agent_factory=

요청 컨텍스트별로 에이전트를 새로 생성하는 팩토리 장착

name / description

Agent Card 자동 생성의 원천 — /.well-known/agent-card.json 경로로 게시

serve()

host·port로 서버 시작 — Task 단위 비동기 협업 수신

코드

from strands import Agent
from strands.multiagent.a2a import A2AServer

# 요청 컨텍스트별로 에이전트를 생성하는 팩토리
def create_research_agent(context_id):
    return Agent(
        name="research-agent",
        description="기술 문서를 검색하고 요약합니다",
        model="us.anthropic.claude-sonnet-4-6",
        system_prompt="기술 리서치 전문 에이전트",
        tools=[web_search, summarize],
    )

# Agent Card는 name/description에서 자동 생성·게시된다
server = A2AServer(agent_factory=create_research_agent,
                   host="0.0.0.0", port=8080)
server.serve()

통신 메커니즘 비교

지연 허용도·결합도·확장성 기준으로 메커니즘 선택

에러 처리와 복원력

분산 에이전트 고유의 실패 모드와 단계적 복구 — 장애 격리의 원칙

멀티에이전트 5가지 실패 모드

타임아웃 연쇄·환각 전파·도구 실패·핸드오프 유실·무한 루프 — 격리가 필요한 이유

SDK 내장 안전장치 파라미터

Swarm 생성자·Graph 빌더에 이미 들어 있는 실행 상한 — 직접 구현 전의 1차 방어선

복구 전략 4단계

실패 감지부터 Supervisor Escalation까지 단계적 복구 흐름

1. 1
   Retry (재시도)

   Exponential backoff로 일시적 장애 극복. 모델 호출 스로틀링 재시도는 SDK 기본 내장 — Agent(retry_strategy=ModelRetryStrategy(...)), 기본 max_attempts=6·초기 지연 4초·최대 지연 128초(Python 기준)이며 None을 주면 꺼집니다. is_retryable() 오버라이드로 재시도 대상 예외를 확장할 수 있습니다. 도구 재시도는 AfterToolCallEvent 훅의 retry 플래그로 트리거합니다.

2. 2
   Circuit Breaker (차단)

   연속 N회 실패 시 회로 차단 → 빠른 실패 반환. half-open 상태에서 주기적으로 복구 확인. 장애가 다른 에이전트로 전파되는 것을 방지합니다.

3. 3
   Fallback (대체)

   주 경로 실패 시 대체 에이전트 또는 간소화된 응답 반환. 예: GPT 실패 → Claude로 전환, 검색 실패 → 캐시된 결과 사용.

4. 4
   Supervisor Escalation

   모든 자동 복구 실패 시 Supervisor 에이전트가 개입. 작업을 재분배하거나 사람에게 에스컬레이션. EventBridge → SNS 알림 연동.

Retry + Circuit Breaker 구현

Strands Hooks 기반의 자동 복구 패턴 코드

핵심 포인트

HookProvider

훅 클래스의 베이스 — register_hooks(registry)로 이벤트별 콜백 등록 (필수)

BeforeToolCallEvent

도구 호출 직전 — 회로가 열려 있으면 event.cancel_tool 설정으로 도구만 취소 (예외 전파 없이 루프 지속)

AfterToolCallEvent

도구 호출 직후 — event.exception으로 실패 판별, failures에 횟수·시각 집계 (코드에선 등록만 표시)

hooks=[...]

Agent에 훅 주입 — 이 한 줄로 자동 차단 활성화

코드

import time
from strands import Agent
from strands.hooks import BeforeToolCallEvent, AfterToolCallEvent, HookProvider

class CircuitBreakerHook(HookProvider):
    def __init__(self, threshold=3, reset_timeout=60):
        self.failures, self.th, self.reset = {}, threshold, reset_timeout

    def register_hooks(self, registry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.on_before)
        registry.add_callback(AfterToolCallEvent, self.on_after)  # 실패 집계

    def on_before(self, event):  # 임계 초과면 차단 (reset 경과 후 half-open)
        s = self.failures.get(event.tool_use["name"], {})
        if s.get("count", 0) >= self.th and time.time() - s["last"] < self.reset:
            event.cancel_tool = "circuit open"  # 도구만 취소 — 루프는 지속

agent = Agent(hooks=[CircuitBreakerHook()])

노드 레벨 훅 — 멀티에이전트 이벤트 5종

단일 에이전트 훅 8종에 더해 노드 실행을 가로채는 전용 이벤트 — 노드 단위 승인·차단·계측

핵심 포인트

MultiAgentInitializedEvent

Swarm·Graph 초기화 완료 시 1회 — 오케스트레이션 시작점 계측

NodeCallEvent 쌍

Before/AfterNodeCallEvent — 노드(에이전트) 실행 전후, event.node_id로 어느 노드인지 식별하는 위임 추적의 정본 지점

MultiAgentInvocationEvent

Before/After 쌍 — 멀티에이전트 호출 전체의 전후, 실행 단위 비용·시간 집계

cancel_node

BeforeNodeCallEvent에서 설정 — 해당 노드만 취소. event.interrupt()로 실행 전 사람 승인 대기도 가능

hooks=[...]

Swarm·Graph 생성자에 그대로 주입 — 단일 에이전트 훅과 같은 등록 방식

코드

from strands import Agent
from strands.multiagent import Swarm
from strands.hooks import HookProvider, BeforeNodeCallEvent

class NodeApprovalHook(HookProvider):
    def register_hooks(self, registry, **kwargs):
        registry.add_callback(BeforeNodeCallEvent, self.on_node)

    def on_node(self, event):
        # 배포 노드만 실행 전 검사 — 거부되면 그 노드만 취소
        if event.node_id == "deploy" and not approved(event.node_id):
            event.cancel_node = "승인 거부 — deploy 노드 차단"

# 단일 에이전트와 같은 hooks= — Swarm/Graph 생성자 지원
swarm = Swarm(nodes=[research, write, deploy],
              hooks=[NodeApprovalHook()])

운영 장면 — 장애에서 복구까지

리서치 Swarm의 요약 노드 타임아웃부터 session_manager 재개까지 — 안전장치가 개입하는 순서

Safety 4계층 적용

입력부터 출력까지 실행 경로 전체에 가드 배치 — 한 계층을 통과한 위협은 다음 계층이 포착

Cedar 정책으로 에이전트 권한 제어

선언적 정책 언어로 에이전트의 허용·금지 도구를 코드로 정의

핵심 포인트

Action::"..."

action이 곧 도구 이름 — pip install "strands-agents[cedar]"로 설치

permit

허용 규칙 — action 목록에 있는 도구만 호출 가능

when { context.session.role }

조건부 허용 — context_enricher가 invocation_state의 role을 주입 (호출 측: invocation_state={"role": "coordinator"})

forbid

명시적 금지 — principal은 principal_resolver가 해석 (기본 User::"anonymous", 해석 실패 시 fail-closed)

코드

// 검색·요약 도구는 모두에게 허용
permit (
    principal,
    action in [Action::"web_search", Action::"summarize"],
    resource
);

// 쓰기 도구는 coordinator 역할에만 허용
permit (
    principal,
    action in [Action::"create_ticket", Action::"send_email"],
    resource
) when { context.session.role == "coordinator" };

// 삭제 도구는 명시적 금지
forbid (principal, action == Action::"delete_resource", resource);

Guardrail 적용 위치

에이전트 실행 경로의 어디에 어떤 Guardrail을 배치하는지 매핑