Team A: Host Agent 구현

분석 대상

a2a-collaborator-agent/agent.py — HostAgent 클래스 (860줄)

HostAgent 클래스 구조

class HostAgent:
    def __init__(self, bearer_token, memory_hook=None,
                 bedrock_model_id="global.anthropic.claude-sonnet-4-20250514-v1:0"):
        # Rate Limiting
        self._bedrock_semaphore = Semaphore(2)      # 동시 2개 API 호출
        self._last_api_call_time = 0
        self._min_delay_between_calls = 1.0         # 최소 1초 간격

        # 원격 에이전트 관리
        self.remote_agent_connections = {}           # name → RemoteAgentConnections
        self.cards = {}                              # name → AgentCard
        self.completed_requests = set()              # 중복 방지

        # 도구: current_time + send_message_tool
        @tool
        async def send_message_tool(agent_name: str, task: str) -> str:
            return await self._send_message_impl(agent_name, task)
        self.tools = [current_time, send_message_tool]

_async_init_components: 원격 에이전트 검색

async def _async_init_components(self, remote_agent_addresses):
    """원격 에이전트 연결 초기화"""
    async with httpx.AsyncClient(timeout=..., http2=True) as client:
        for address in remote_agent_addresses:
            # 1. 헬스체크
            await client.get(f"{address}/health")

            # 2. Agent Card 조회 (재시도 3회)
            card_resolver = A2ACardResolver(client, address)
            for attempt in range(3):
                try:
                    card = await card_resolver.get_agent_card()
                    break
                except:
                    await asyncio.sleep(2 ** attempt)  # 2s, 4s

            # 3. 연결 등록
            remote_connection = RemoteAgentConnections(
                agent_card=card, agent_url=address
            )
            self.remote_agent_connections[card.name] = remote_connection

    # 4. 시스템 프롬프트 업데이트 (발견된 에이전트 정보)
    self.system_prompt = self.get_default_system_prompt()

Semaphore 기반 Rate Limiting

async def stream(self, user_query):
    async with self._bedrock_semaphore:  # ← 동시 2개 제한
        # 최소 대기 시간 보장
        time_since_last = time.time() - self._last_api_call_time
        if time_since_last < self._min_delay_between_calls:
            await asyncio.sleep(self._min_delay_between_calls - time_since_last)

        self._last_api_call_time = time.time()

        # 지수 백오프 재시도
        retry_count = 0
        while retry_count <= 3:
            try:
                async for event in self.agent.stream_async(user_query):
                    yield event["data"]
                break
            except Exception as e:
                if "throttling" in str(e).lower():
                    backoff = 2.0 * (2 ** retry_count)  # 2s, 4s, 8s
                    await asyncio.sleep(backoff)
                    retry_count += 1

Rate Limiting 메커니즘 비교

메커니즘	구현	위치
asyncio.Semaphore(2)	동시 호출 제한	stream()
Min Delay	호출 간 1초 대기	stream()
Exponential Backoff	2s→4s→8s	stream() 재시도
Duplicate Prevention	completed_requests set	_send_message_impl()

BedrockAgentCoreApp 엔트리포인트

app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke(payload, context):
    user_message = payload["prompt"]
    actor_id = payload["actor_id"]
    session_id = context.session_id

    # HostAgent가 없으면 생성 (원격 에이전트 검색 포함)
    if agent is None:
        agent = await HostAgent.create(
            remote_agent_addresses=config['servers'],
            bearer_token=gateway_access_token,
            memory_hook=memory_hook,
        )

    # 스트리밍 응답
    async for chunk in agent.stream(user_query=user_message):
        await response_queue.put(chunk)

토론 질문

1. Semaphore(2)를 Semaphore(5)로 변경하면 어떤 영향이 있나요?

2. completed_requests가 메모리에서만 관리되므로, 컨테이너 재시작 시 어떻게 되나요?

3. _async_init_components가 실패하면 에이전트가 어떻게 동작하나요?