Amazon Bedrock AgentCore Multi-Agent

Amazon Bedrock AgentCore 多智能体

Overview

概述

Build sophisticated multi-agent systems using AgentCore's Agent-to-Agent (A2A) protocol. Implement supervisor-worker patterns where a routing agent delegates to specialized domain experts, maintaining context across the agent hierarchy.

Purpose: Orchestrate multiple AI agents for complex, multi-domain tasks

Pattern: Workflow-based (3 orchestration patterns)

Key Principles (validated by AWS December 2025):

Supervisor-Worker Architecture - Routing agent delegates to specialists
A2A Protocol - Standard communication between agents
Context Preservation - Conversation history shared across agents
Failure Isolation - Single agent failure doesn't crash system
Modular Design - Each agent focused on specific domain
Framework Interoperability - Works across different agent frameworks

Quality Targets:

Routing accuracy: ≥ 95%
Context preservation: 100%
Failure isolation: Complete
Latency overhead: < 200ms per hop

借助AgentCore的Agent-to-Agent（A2A）协议构建复杂的多智能体系统。实现主管-工作者模式，由路由智能体将任务委托给专业领域的智能体，并在智能体层级间保持上下文信息。

用途：为复杂的跨领域任务编排多个AI智能体

模式：基于工作流（3种编排模式）

核心原则（经AWS 2025年12月验证）：

主管-工作者架构 - 路由智能体将任务委托给专业智能体
A2A协议 - 智能体间的标准通信协议
上下文保留 - 智能体间共享对话历史
故障隔离 - 单个智能体故障不会导致整个系统崩溃
模块化设计 - 每个智能体专注于特定领域
框架互操作性 - 兼容不同的智能体框架

质量目标:

路由准确率：≥ 95%
上下文保留率：100%
故障隔离：完全隔离
延迟开销：每跳<200ms

When to Use

适用场景

Use bedrock-agentcore-multi-agent when:

Task requires multiple specialized domains
Single agent becomes too complex
Need different models for different tasks
Want fault-isolated agent architecture
Building enterprise customer service
Implementing complex workflows

When NOT to Use:

Simple single-domain tasks
Cost-sensitive applications (multi-hop adds cost)
Low-latency requirements (< 1s total)

在以下场景使用bedrock-agentcore-multi-agent：

任务涉及多个专业领域
单个智能体过于复杂
不同任务需要不同模型
需要故障隔离的智能体架构
构建企业级客户服务系统
实现复杂工作流

不适用场景:

简单的单领域任务
对成本敏感的应用（多跳会增加成本）
低延迟要求（总延迟<1秒）

Prerequisites

前提条件

Required

必需条件

Multiple AgentCore agents deployed
Supervisor agent with routing capability
A2A protocol enabled
IAM permissions for cross-agent calls

已部署多个AgentCore智能体
具备路由能力的主管智能体
已启用A2A协议
跨智能体调用的IAM权限

Architecture Patterns

架构模式

Pattern 1: Hub-and-Spoke (Supervisor)

模式1：中心辐射型（主管模式）

                    ┌─────────────────┐
                    │   User Query    │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │    (Router)     │
                    └────────┬────────┘
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
    ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
    │   Orders    │   │   Returns   │   │   Support   │
    │   Agent     │   │   Agent     │   │   Agent     │
    └─────────────┘   └─────────────┘   └─────────────┘

                    ┌─────────────────┐
                    │   User Query    │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Supervisor    │
                    │    (Router)     │
                    └────────┬────────┘
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
    ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
    │   Orders    │   │   Returns   │   │   Support   │
    │   Agent     │   │   Agent     │   │   Agent     │
    └─────────────┘   └─────────────┘   └─────────────┘

Pattern 2: Sequential Pipeline

模式2：顺序流水线型

    ┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐
    │ Intake  │ ──▶ │ Analyze │ ──▶ │ Resolve │ ──▶ │  Close  │
    │ Agent   │     │ Agent   │     │ Agent   │     │ Agent   │
    └─────────┘     └─────────┘     └─────────┘     └─────────┘

    ┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐
    │ Intake  │ ──▶ │ Analyze │ ──▶ │ Resolve │ ──▶ │  Close  │
    │ Agent   │     │ Agent   │     │ Agent   │     │ Agent   │
    └─────────┘     └─────────┘     └─────────┘     └─────────┘

Pattern 3: Hierarchical

模式3：层级型

                    ┌─────────────────┐
                    │   Executive     │
                    │   Supervisor    │
                    └────────┬────────┘
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
    ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
    │   Sales     │   │   Support   │   │   Billing   │
    │ Supervisor  │   │ Supervisor  │   │ Supervisor  │
    └──────┬──────┘   └──────┬──────┘   └──────┬──────┘
           │                 │                 │
     ┌─────┴─────┐     ┌─────┴─────┐     ┌─────┴─────┐
     │           │     │           │     │           │
     ▼           ▼     ▼           ▼     ▼           ▼
   ┌───┐       ┌───┐ ┌───┐       ┌───┐ ┌───┐       ┌───┐
   │B2B│       │B2C│ │L1 │       │L2 │ │Pay│       │Inv│
   └───┘       └───┘ └───┘       └───┘ └───┘       └───┘

                    ┌─────────────────┐
                    │   Executive     │
                    │   Supervisor    │
                    └────────┬────────┘
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
    ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
    │   Sales     │   │   Support   │   │   Billing   │
    │ Supervisor  │   │ Supervisor  │   │ Supervisor  │
    └──────┬──────┘   └──────┬──────┘   └──────┬──────┘
           │                 │                 │
     ┌─────┴─────┐     ┌─────┴─────┐     ┌─────┴─────┐
     │           │     │           │     │           │
     ▼           ▼     ▼           ▼     ▼           ▼
   ┌───┐       ┌───┐ ┌───┐       ┌───┐ ┌───┐       ┌───┐
   │B2B│       │B2C│ │L1 │       │L2 │ │Pay│       │Inv│
   └───┘       └───┘ └───┘       └───┘ └───┘       └───┘

Operations

操作指南

Operation 1: Create Supervisor Agent

操作1：创建主管智能体

Time: 15-30 minutes Automation: 80% Purpose: Build the routing/orchestration agent

Supervisor Agent Code:

python

undefined

耗时：15-30分钟 自动化程度：80% 用途：构建路由/编排智能体

主管智能体代码:

python

undefined

supervisor_agent.py

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent import boto3 import json

app = BedrockAgentCoreApp() client = boto3.client('bedrock-agentcore')

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent import boto3 import json

app = BedrockAgentCoreApp() client = boto3.client('bedrock-agentcore')

Define collaborator agents

COLLABORATORS = { 'orders': { 'arn': 'arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/orders-agent', 'description': 'Handles order status, tracking, and modifications', 'triggers': ['order', 'tracking', 'delivery', 'shipment', 'status'] }, 'returns': { 'arn': 'arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/returns-agent', 'description': 'Processes returns, refunds, and exchanges', 'triggers': ['return', 'refund', 'exchange', 'damaged', 'wrong item'] }, 'support': { 'arn': 'arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/support-agent', 'description': 'Technical support and product questions', 'triggers': ['help', 'problem', 'issue', 'how to', 'broken', 'not working'] }, 'billing': { 'arn': 'arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/billing-agent', 'description': 'Payment issues, invoices, and account billing', 'triggers': ['payment', 'charge', 'invoice', 'bill', 'subscription'] } }

Routing agent

router = Agent( model="anthropic.claude-sonnet-4-20250514-v1:0", system_prompt=f"""You are a customer service supervisor. Your job is to:

Understand the customer's intent
Route to the appropriate specialist
Handle escalations and complex multi-domain issues

Available specialists: {json.dumps({k: v['description'] for k, v in COLLABORATORS.items()}, indent=2)}

Respond with JSON: {{"route": "orders|returns|support|billing|self", "reason": "...", "context": "..."}}

Use "self" only for greetings or if truly unclear. """ )

@app.entrypoint def invoke(payload): user_message = payload.get('prompt', '') session_id = payload.get('session_id', 'default') conversation_history = payload.get('history', [])

# Step 1: Route the request
routing_prompt = f"""

Customer message: {user_message}

Previous context: {json.dumps(conversation_history[-3:]) if conversation_history else 'None'}

Determine the appropriate specialist and respond with routing JSON. """

routing_response = router(routing_prompt)

try:
    routing = json.loads(routing_response.message)
except json.JSONDecodeError:
    # Fallback to keyword matching
    routing = keyword_route(user_message)

route = routing.get('route', 'self')

# Step 2: Handle routing
if route == 'self':
    # Handle directly
    return {
        'response': "Hello! I'm here to help. What can I assist you with today?",
        'routed_to': None
    }

# Step 3: Delegate to specialist
collaborator = COLLABORATORS.get(route)
if not collaborator:
    return {'response': "I apologize, let me connect you with support.", 'routed_to': 'support'}

# Invoke collaborator with context
try:
    response = client.invoke_agent_runtime(
        agentRuntimeArn=collaborator['arn'],
        runtimeSessionId=f"{session_id}-{route}",
        payload={
            'prompt': user_message,
            'context': routing.get('context', ''),
            'supervisor_notes': routing.get('reason', ''),
            'history': conversation_history
        }
    )

    return {
        'response': response['payload'].get('response', ''),
        'routed_to': route,
        'routing_reason': routing.get('reason', '')
    }

except Exception as e:
    # Fallback: handle ourselves or escalate
    app.logger.error(f"Collaborator failed: {e}")
    return {
        'response': "I apologize for the inconvenience. Let me help you directly.",
        'routed_to': None,
        'error': str(e)
    }

def keyword_route(message): """Fallback keyword-based routing""" message_lower = message.lower()

for route, config in COLLABORATORS.items():
    if any(trigger in message_lower for trigger in config['triggers']):
        return {'route': route, 'reason': 'keyword_match'}

return {'route': 'self', 'reason': 'no_match'}

if name == "main": app.run()

---

router = Agent( model="anthropic.claude-sonnet-4-20250514-v1:0", system_prompt=f"""You are a customer service supervisor. Your job is to:

Understand the customer's intent
Route to the appropriate specialist
Handle escalations and complex multi-domain issues

Available specialists: {json.dumps({k: v['description'] for k, v in COLLABORATORS.items()}, indent=2)}

Respond with JSON: {{"route": "orders|returns|support|billing|self", "reason": "...", "context": "..."}}

Use "self" only for greetings or if truly unclear. """ )

@app.entrypoint def invoke(payload): user_message = payload.get('prompt', '') session_id = payload.get('session_id', 'default') conversation_history = payload.get('history', [])

# Step 1: Route the request
routing_prompt = f"""

Customer message: {user_message}

Previous context: {json.dumps(conversation_history[-3:]) if conversation_history else 'None'}

Determine the appropriate specialist and respond with routing JSON. """

routing_response = router(routing_prompt)

try:
    routing = json.loads(routing_response.message)
except json.JSONDecodeError:
    # Fallback to keyword matching
    routing = keyword_route(user_message)

route = routing.get('route', 'self')

# Step 2: Handle routing
if route == 'self':
    # Handle directly
    return {
        'response': "Hello! I'm here to help. What can I assist you with today?",
        'routed_to': None
    }

# Step 3: Delegate to specialist
collaborator = COLLABORATORS.get(route)
if not collaborator:
    return {'response': "I apologize, let me connect you with support.", 'routed_to': 'support'}

# Invoke collaborator with context
try:
    response = client.invoke_agent_runtime(
        agentRuntimeArn=collaborator['arn'],
        runtimeSessionId=f"{session_id}-{route}",
        payload={
            'prompt': user_message,
            'context': routing.get('context', ''),
            'supervisor_notes': routing.get('reason', ''),
            'history': conversation_history
        }
    )

    return {
        'response': response['payload'].get('response', ''),
        'routed_to': route,
        'routing_reason': routing.get('reason', '')
    }

except Exception as e:
    # Fallback: handle ourselves or escalate
    app.logger.error(f"Collaborator failed: {e}")
    return {
        'response': "I apologize for the inconvenience. Let me help you directly.",
        'routed_to': None,
        'error': str(e)
    }

def keyword_route(message): """Fallback keyword-based routing""" message_lower = message.lower()

for route, config in COLLABORATORS.items():
    if any(trigger in message_lower for trigger in config['triggers']):
        return {'route': route, 'reason': 'keyword_match'}

return {'route': 'self', 'reason': 'no_match'}

if name == "main": app.run()

---

Operation 2: Create Specialist Agents

操作2：创建专业智能体

Time: 10-20 minutes per agent Automation: 85% Purpose: Build domain-specific worker agents

Orders Agent:

python

undefined

耗时：每个智能体10-20分钟 自动化程度：85% 用途：构建特定领域的工作者智能体

订单智能体:

python

undefined

orders_agent.py

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent

app = BedrockAgentCoreApp()

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent

app = BedrockAgentCoreApp()

Orders specialist

agent = Agent( model="anthropic.claude-sonnet-4-20250514-v1:0", system_prompt="""You are an orders specialist. You handle:

Order status inquiries
Delivery tracking
Order modifications
Shipping questions

You have access to these tools:

get_order_status(order_id) - Get order details
track_shipment(tracking_number) - Track delivery
modify_order(order_id, changes) - Update order

Be helpful, accurate, and efficient. If a request is outside your domain (returns, billing, technical support), indicate that clearly. """ )

@app.entrypoint def invoke(payload): user_message = payload.get('prompt', '') context = payload.get('context', '') supervisor_notes = payload.get('supervisor_notes', '') history = payload.get('history', [])

# Build context-aware prompt
prompt = user_message
if context:
    prompt = f"Context from supervisor: {context}\n\nCustomer request: {user_message}"

result = agent(prompt)

# Check if we need to hand back to supervisor
needs_handoff = check_domain_boundary(result.message)

return {
    'response': result.message,
    'needs_handoff': needs_handoff,
    'handoff_reason': needs_handoff if needs_handoff else None
}

def check_domain_boundary(response): """Check if response indicates need for different specialist""" handoff_indicators = [ ('return', 'returns'), ('refund', 'returns'), ('billing', 'billing'), ('payment', 'billing'), ('technical', 'support') ]

response_lower = response.lower()
for indicator, domain in handoff_indicators:
    if f"need to contact {indicator}" in response_lower or \
       f"transfer to {indicator}" in response_lower:
        return domain

return None

if name == "main": app.run()


**Returns Agent**:
```python

agent = Agent( model="anthropic.claude-sonnet-4-20250514-v1:0", system_prompt="""You are an orders specialist. You handle:

Order status inquiries
Delivery tracking
Order modifications
Shipping questions

You have access to these tools:

get_order_status(order_id) - Get order details
track_shipment(tracking_number) - Track delivery
modify_order(order_id, changes) - Update order

Be helpful, accurate, and efficient. If a request is outside your domain (returns, billing, technical support), indicate that clearly. """ )

@app.entrypoint def invoke(payload): user_message = payload.get('prompt', '') context = payload.get('context', '') supervisor_notes = payload.get('supervisor_notes', '') history = payload.get('history', [])

# Build context-aware prompt
prompt = user_message
if context:
    prompt = f"Context from supervisor: {context}\n\nCustomer request: {user_message}"

result = agent(prompt)

# Check if we need to hand back to supervisor
needs_handoff = check_domain_boundary(result.message)

return {
    'response': result.message,
    'needs_handoff': needs_handoff,
    'handoff_reason': needs_handoff if needs_handoff else None
}

def check_domain_boundary(response): """Check if response indicates need for different specialist""" handoff_indicators = [ ('return', 'returns'), ('refund', 'returns'), ('billing', 'billing'), ('payment', 'billing'), ('technical', 'support') ]

response_lower = response.lower()
for indicator, domain in handoff_indicators:
    if f"need to contact {indicator}" in response_lower or \
       f"transfer to {indicator}" in response_lower:
        return domain

return None

if name == "main": app.run()


**退货智能体**:
```python

returns_agent.py

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent

app = BedrockAgentCoreApp()

agent = Agent( model="anthropic.claude-3-haiku-20240307-v1:0", # Faster model for simple tasks system_prompt="""You are a returns specialist. You handle:

Return requests
Refund processing
Exchanges
Damaged item claims

Tools available:

initiate_return(order_id, reason) - Start return process
check_return_eligibility(order_id) - Verify return policy
process_refund(order_id) - Issue refund
create_exchange(order_id, new_item) - Process exchange

Be empathetic and solution-oriented. Follow return policy strictly. """ )

@app.entrypoint def invoke(payload): result = agent(payload.get('prompt', '')) return {'response': result.message}

if name == "main": app.run()

---

from bedrock_agentcore import BedrockAgentCoreApp from strands import Agent

app = BedrockAgentCoreApp()

agent = Agent( model="anthropic.claude-3-haiku-20240307-v1:0", # Faster model for simple tasks system_prompt="""You are a returns specialist. You handle:

Return requests
Refund processing
Exchanges
Damaged item claims

Tools available:

initiate_return(order_id, reason) - Start return process
check_return_eligibility(order_id) - Verify return policy
process_refund(order_id) - Issue refund
create_exchange(order_id, new_item) - Process exchange

Be empathetic and solution-oriented. Follow return policy strictly. """ )

@app.entrypoint def invoke(payload): result = agent(payload.get('prompt', '')) return {'response': result.message}

if name == "main": app.run()

---

Operation 3: Configure A2A Protocol

操作3：配置A2A协议

Time: 10-15 minutes Automation: 90% Purpose: Enable secure agent-to-agent communication

Enable A2A for Agents:

python

import boto3

control = boto3.client('bedrock-agentcore-control')

耗时：10-15分钟 自动化程度：90% 用途：启用安全的智能体间通信

为智能体启用A2A:

python

import boto3

control = boto3.client('bedrock-agentcore-control')

Configure supervisor to call collaborators

response = control.update_agent_runtime( agentRuntimeId='supervisor-agent', collaborationConfig={ 'enabled': True, 'collaborators': [ { 'agentRuntimeArn': 'arn:...:agent-runtime/orders-agent', 'alias': 'orders', 'description': 'Order management specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/returns-agent', 'alias': 'returns', 'description': 'Returns and refunds specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/support-agent', 'alias': 'support', 'description': 'Technical support specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/billing-agent', 'alias': 'billing', 'description': 'Billing and payments specialist' } ], 'routingMode': 'SUPERVISOR_WITH_ROUTING', 'contextSharing': { 'shareConversationHistory': True, 'maxHistoryTurns': 10 } } )


**IAM Policy for A2A**:
```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "bedrock-agentcore:InvokeAgentRuntime",
      "Resource": [
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/orders-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/returns-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/support-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/billing-agent"
      ],
      "Condition": {
        "StringEquals": {
          "bedrock-agentcore:CallerAgentRuntimeArn": "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/supervisor-agent"
        }
      }
    }
  ]
}

response = control.update_agent_runtime( agentRuntimeId='supervisor-agent', collaborationConfig={ 'enabled': True, 'collaborators': [ { 'agentRuntimeArn': 'arn:...:agent-runtime/orders-agent', 'alias': 'orders', 'description': 'Order management specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/returns-agent', 'alias': 'returns', 'description': 'Returns and refunds specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/support-agent', 'alias': 'support', 'description': 'Technical support specialist' }, { 'agentRuntimeArn': 'arn:...:agent-runtime/billing-agent', 'alias': 'billing', 'description': 'Billing and payments specialist' } ], 'routingMode': 'SUPERVISOR_WITH_ROUTING', 'contextSharing': { 'shareConversationHistory': True, 'maxHistoryTurns': 10 } } )


**A2A的IAM策略**:
```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "bedrock-agentcore:InvokeAgentRuntime",
      "Resource": [
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/orders-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/returns-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/support-agent",
        "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/billing-agent"
      ],
      "Condition": {
        "StringEquals": {
          "bedrock-agentcore:CallerAgentRuntimeArn": "arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/supervisor-agent"
        }
      }
    }
  ]
}

Operation 4: Implement Handoff Patterns

操作4：实现交接模式

Time: 15-30 minutes Automation: 70% Purpose: Handle smooth transitions between agents

Context-Preserving Handoff:

python

class AgentHandoff:
    """Manages handoffs between agents"""

    def __init__(self, client, supervisor_arn):
        self.client = client
        self.supervisor_arn = supervisor_arn
        self.context_store = {}

    def initiate_handoff(self, from_agent, to_agent, session_id, context):
        """Hand off from one agent to another"""

        # Store context for receiving agent
        handoff_context = {
            'from_agent': from_agent,
            'reason': context.get('reason', ''),
            'summary': context.get('summary', ''),
            'customer_sentiment': context.get('sentiment', 'neutral'),
            'conversation_history': context.get('history', []),
            'pending_actions': context.get('pending_actions', [])
        }

        self.context_store[session_id] = handoff_context

        # Notify receiving agent
        response = self.client.invoke_agent_runtime(
            agentRuntimeArn=to_agent,
            runtimeSessionId=session_id,
            payload={
                'type': 'HANDOFF_RECEIVE',
                'context': handoff_context,
                'prompt': f"Customer transferred from {from_agent}. Context: {handoff_context['summary']}"
            }
        )

        return response

    def escalate_to_supervisor(self, agent_arn, session_id, reason):
        """Escalate back to supervisor"""

        return self.client.invoke_agent_runtime(
            agentRuntimeArn=self.supervisor_arn,
            runtimeSessionId=session_id,
            payload={
                'type': 'ESCALATION',
                'from_agent': agent_arn,
                'reason': reason,
                'needs_human': reason.get('needs_human', False)
            }
        )

Multi-Turn Handoff Example:

python

undefined

耗时：15-30分钟 自动化程度：70% 用途：处理智能体间的平滑过渡

保留上下文的交接:

python

class AgentHandoff:
    """Manages handoffs between agents"""

    def __init__(self, client, supervisor_arn):
        self.client = client
        self.supervisor_arn = supervisor_arn
        self.context_store = {}

    def initiate_handoff(self, from_agent, to_agent, session_id, context):
        """Hand off from one agent to another"""

        # Store context for receiving agent
        handoff_context = {
            'from_agent': from_agent,
            'reason': context.get('reason', ''),
            'summary': context.get('summary', ''),
            'customer_sentiment': context.get('sentiment', 'neutral'),
            'conversation_history': context.get('history', []),
            'pending_actions': context.get('pending_actions', [])
        }

        self.context_store[session_id] = handoff_context

        # Notify receiving agent
        response = self.client.invoke_agent_runtime(
            agentRuntimeArn=to_agent,
            runtimeSessionId=session_id,
            payload={
                'type': 'HANDOFF_RECEIVE',
                'context': handoff_context,
                'prompt': f"Customer transferred from {from_agent}. Context: {handoff_context['summary']}"
            }
        )

        return response

    def escalate_to_supervisor(self, agent_arn, session_id, reason):
        """Escalate back to supervisor"""

        return self.client.invoke_agent_runtime(
            agentRuntimeArn=self.supervisor_arn,
            runtimeSessionId=session_id,
            payload={
                'type': 'ESCALATION',
                'from_agent': agent_arn,
                'reason': reason,
                'needs_human': reason.get('needs_human', False)
            }
        )

多轮交接示例:

python

undefined

Conversation flow with handoffs

Turn 1: User -> Supervisor

"I want to return an order and also change my payment method"

Supervisor detects multi-domain request

Routes to returns first (primary intent)

Turn 2: Returns Agent handles return

Initiates return process

Turn 3: Returns Agent hands off to Billing

With context: "Return initiated for order #123, customer also needs payment update"

Turn 4: Billing Agent receives handoff

"I see you've started a return. Let me help you update your payment method."

Handles payment update

Turn 5: Billing -> Supervisor

"Both requests handled. Return initiated, payment updated."

---

---

Operation 5: Multi-Agent Monitoring

操作5：多智能体监控

Time: 15-20 minutes Automation: 85% Purpose: Monitor the multi-agent system

Trace Multi-Agent Calls:

python

import boto3
import json

cloudwatch = boto3.client('cloudwatch')
logs = boto3.client('logs')

耗时：15-20分钟 自动化程度：85% 用途：监控多智能体系统

追踪多智能体调用:

python

import boto3
import json

cloudwatch = boto3.client('cloudwatch')
logs = boto3.client('logs')

CloudWatch Metrics for Multi-Agent

cloudwatch.put_metric_data( Namespace='AgentCore/MultiAgent', MetricData=[ { 'MetricName': 'RoutingDecisions', 'Dimensions': [ {'Name': 'SupervisorAgent', 'Value': 'customer-service-supervisor'}, {'Name': 'TargetAgent', 'Value': 'orders-agent'} ], 'Value': 1, 'Unit': 'Count' }, { 'MetricName': 'HandoffLatency', 'Dimensions': [ {'Name': 'FromAgent', 'Value': 'returns-agent'}, {'Name': 'ToAgent', 'Value': 'billing-agent'} ], 'Value': 150, # milliseconds 'Unit': 'Milliseconds' } ] )

Log Insights Query for Multi-Agent Traces


**Dashboard for Multi-Agent System**:
```python
dashboard_body = {
    "widgets": [
        {
            "type": "metric",
            "properties": {
                "title": "Routing Distribution",
                "metrics": [
                    ["AgentCore/MultiAgent", "RoutingDecisions", "TargetAgent", "orders-agent"],
                    [".", ".", ".", "returns-agent"],
                    [".", ".", ".", "support-agent"],
                    [".", ".", ".", "billing-agent"]
                ],
                "period": 3600,
                "stat": "Sum",
                "view": "pie"
            }
        },
        {
            "type": "metric",
            "properties": {
                "title": "Handoff Latency",
                "metrics": [
                    ["AgentCore/MultiAgent", "HandoffLatency"]
                ],
                "period": 300,
                "stat": "p99"
            }
        },
        {
            "type": "metric",
            "properties": {
                "title": "Escalation Rate",
                "metrics": [
                    ["AgentCore/MultiAgent", "Escalations", "Reason", "out_of_scope"],
                    [".", ".", ".", "customer_request"],
                    [".", ".", ".", "agent_failure"]
                ],
                "period": 3600,
                "stat": "Sum"
            }
        }
    ]
}

cloudwatch.put_dashboard(
    DashboardName='MultiAgentOrchestration',
    DashboardBody=json.dumps(dashboard_body)
)


**多智能体系统仪表盘**:
```python
dashboard_body = {
    "widgets": [
        {
            "type": "metric",
            "properties": {
                "title": "Routing Distribution",
                "metrics": [
                    ["AgentCore/MultiAgent", "RoutingDecisions", "TargetAgent", "orders-agent"],
                    [".", ".", ".", "returns-agent"],
                    [".", ".", ".", "support-agent"],
                    [".", ".", ".", "billing-agent"]
                ],
                "period": 3600,
                "stat": "Sum",
                "view": "pie"
            }
        },
        {
            "type": "metric",
            "properties": {
                "title": "Handoff Latency",
                "metrics": [
                    ["AgentCore/MultiAgent", "HandoffLatency"]
                ],
                "period": 300,
                "stat": "p99"
            }
        },
        {
            "type": "metric",
            "properties": {
                "title": "Escalation Rate",
                "metrics": [
                    ["AgentCore/MultiAgent", "Escalations", "Reason", "out_of_scope"],
                    [".", ".", ".", "customer_request"],
                    [".", ".", ".", "agent_failure"]
                ],
                "period": 3600,
                "stat": "Sum"
            }
        }
    ]
}

cloudwatch.put_dashboard(
    DashboardName='MultiAgentOrchestration',
    DashboardBody=json.dumps(dashboard_body)
)

Best Practices

最佳实践

1. Clear Domain Boundaries

1. 明确领域边界

python

undefined

python

undefined

Good: Clear separation

DOMAINS = { 'orders': ['status', 'tracking', 'modify', 'cancel'], 'returns': ['return', 'refund', 'exchange', 'damaged'], 'billing': ['payment', 'invoice', 'subscription'] }

Bad: Overlapping domains

'orders': ['status', 'refund'] # Refund overlaps with returns

undefined

undefined

2. Graceful Degradation

2. 优雅降级

python

async def invoke_with_fallback(primary_agent, fallback_agent, payload):
    """Try primary, fall back to backup"""
    try:
        return await invoke_agent(primary_agent, payload)
    except Exception:
        return await invoke_agent(fallback_agent, payload)

python

async def invoke_with_fallback(primary_agent, fallback_agent, payload):
    """Try primary, fall back to backup"""
    try:
        return await invoke_agent(primary_agent, payload)
    except Exception:
        return await invoke_agent(fallback_agent, payload)

3. Context Compression

3. 上下文压缩

python

def compress_history(history, max_turns=10):
    """Keep relevant context, compress old turns"""
    if len(history) <= max_turns:
        return history

    # Keep first turn (initial context) and recent turns
    return [history[0]] + history[-(max_turns-1):]

python

def compress_history(history, max_turns=10):
    """Keep relevant context, compress old turns"""
    if len(history) <= max_turns:
        return history

    # Keep first turn (initial context) and recent turns
    return [history[0]] + history[-(max_turns-1):]

Related Skills

References

参考资料

```
references/routing-strategies.md
```
- Advanced routing patterns
```
references/context-management.md
```
- Cross-agent context handling
```
references/failure-handling.md
```
- Error recovery patterns

Sources

—

bedrock-agentcore-multi-agent

Original

Translation

Amazon Bedrock AgentCore Multi-Agent

Amazon Bedrock AgentCore 多智能体

Overview

概述

When to Use

适用场景

Prerequisites

前提条件

Required

必需条件

Recommended

推荐条件

Architecture Patterns

架构模式

Pattern 1: Hub-and-Spoke (Supervisor)

模式1：中心辐射型（主管模式）

Pattern 2: Sequential Pipeline

模式2：顺序流水线型

Pattern 3: Hierarchical

模式3：层级型

Operations

操作指南

Operation 1: Create Supervisor Agent

操作1：创建主管智能体

supervisor_agent.py

supervisor_agent.py

Define collaborator agents

Define collaborator agents

Routing agent

Routing agent

Operation 2: Create Specialist Agents

操作2：创建专业智能体

orders_agent.py

orders_agent.py

Orders specialist

Orders specialist

returns_agent.py

returns_agent.py

Operation 3: Configure A2A Protocol

操作3：配置A2A协议

Configure supervisor to call collaborators

Configure supervisor to call collaborators

Operation 4: Implement Handoff Patterns

操作4：实现交接模式

Conversation flow with handoffs

Conversation flow with handoffs

Turn 1: User -> Supervisor

Turn 1: User -> Supervisor

"I want to return an order and also change my payment method"

"I want to return an order and also change my payment method"

Supervisor detects multi-domain request

Supervisor detects multi-domain request

Routes to returns first (primary intent)

Routes to returns first (primary intent)

Turn 2: Returns Agent handles return

Turn 2: Returns Agent handles return

Initiates return process

Initiates return process

Turn 3: Returns Agent hands off to Billing

Turn 3: Returns Agent hands off to Billing

With context: "Return initiated for order #123, customer also needs payment update"

With context: "Return initiated for order #123, customer also needs payment update"

Turn 4: Billing Agent receives handoff

Turn 4: Billing Agent receives handoff

"I see you've started a return. Let me help you update your payment method."

"I see you've started a return. Let me help you update your payment method."

Handles payment update

Handles payment update

Turn 5: Billing -> Supervisor

Turn 5: Billing -> Supervisor

"Both requests handled. Return initiated, payment updated."

"Both requests handled. Return initiated, payment updated."

Operation 5: Multi-Agent Monitoring

操作5：多智能体监控

CloudWatch Metrics for Multi-Agent

CloudWatch Metrics for Multi-Agent

Log Insights Query for Multi-Agent Traces