AWS Bedrock 多智能体协作指南¶

# Bedrock Agent 核心组件
agent_config = {
 "agent_name": "multi-agent-coordinator",
 "foundation_model": "anthropic.claude-sonnet-4-20250514",
 "instruction": """You are a task orchestration agent.
 Coordinate specialized agents to complete complex tasks.""",
 "tools": [
 "arn:aws:bedrock:us-east-1:123456789012:agent/agent-id-1", # 搜索 Agent
 "arn:aws:bedrock:us-east-1:123456789012:agent/agent-id-2", # 编码 Agent
 ],
 "action_group": {
 "name": "task_delegation",
 "description": "Delegate tasks to specialized agents"
 }
}

import boto3

bedrock = boto3.client('bedrock-agent', region_name='us-east-1')

# 创建 Agent
response = bedrock.create_agent(
 agentName='research-agent',
 agentResourceRoleArn='arn:aws:iam::123456789012:role/BedrockAgentRole',
 description='Research specialist agent',
 foundationModel='anthropic.claude-sonnet-4-20250514',
 instruction='''You are a research specialist.
 Your role is to gather, analyze, and synthesize information.
 Always cite your sources and verify facts before reporting.''',
)

agent_id = response['agent']['agentId']
print(f"Created agent: {agent_id}")

# 创建 Agent 版本（快照）
bedrock.create_agent_version(
 agentId=agent_id,
 description='Production version 1.0'
)

# 创建 Alias（指向特定版本）
bedrock.create_agent_alias(
 agentId=agent_id,
 agentAliasName='production',
 agentVersion='1'
)

# 通过 Alias 调用（生产推荐）
alias_arn = f"arn:aws:bedrock:{region}:{account}:agent-alias/{agent_id}/production"

import boto3
from concurrent.futures import ThreadPoolExecutor, as_completed

class BedrockMultiAgentOrchestrator:
 def __init__(self, region='us-east-1'):
 self.bedrock = boto3.client('bedrock-agent-runtime', region_name=region)
 self.agents = {
 'search': 'arn:aws:bedrock:...:agent/search-agent-alias',
 'coder': 'arn:aws:bedrock:...:agent/coder-agent-alias',
 'writer': 'arn:aws:bedrock:...:agent/writer-agent-alias',
 }

 def orchestrate(self, task: str) -> dict:
 """Orchestrator 主流程"""

 # Step 1: 任务分解
 subtasks = self._decompose_task(task)

 # Step 2: 并行执行 Worker Agents
 results = self._execute_workers_parallel(subtasks)

 # Step 3: 结果聚合
 final_response = self._aggregate_results(results)

 return {
 'task': task,
 'subtasks': subtasks,
 'worker_results': results,
 'final_response': final_response
 }

 def _decompose_task(self, task: str) -> list:
 """使用 Orchestrator 分解任务"""
 prompt = f"""Decompose this task into independent subtasks:
 Task: {task}

 Return a JSON array of subtasks, each with:

 - "subtask_id": unique identifier
 - "description": what to do
 - "required_agent": search/coder/writer
 """

 response = self.bedrock.invoke_agent(
 agentAliasId='orchestrator-alias',
 agentId='orchestrator-agent-id',
 sessionId='session-123',
 inputText=prompt
 )

 return self._parse_subtasks(response['completion'])

 def _execute_workers_parallel(self, subtasks: list) -> dict:
 """并行执行多个 Worker Agents"""
 results = {}

 with ThreadPoolExecutor(max_workers=5) as executor:
 futures = {
 executor.submit(self._invoke_agent, subtask): subtask 
 for subtask in subtasks
 }

 for future in as_completed(futures):
 subtask = futures[future]
 try:
 results[subtask['subtask_id'] = future.result()
 except Exception as e:
 results[subtask['subtask_id'] = {'error': str(e)}

 return results

 def _invoke_agent(self, subtask: dict) -> dict:
 """调用单个 Worker Agent"""
 agent_alias = self.agents[subtask['required_agent']

 response = self.bedrock.invoke_agent(
 agentAliasId=agent_alias.split('/')[-1],
 agentId=agent_alias.split(':agent/')[1].split('/')[0],
 sessionId=f"session-{subtask['subtask_id']}",
 inputText=subtask['description']
 )

 return {'agent': subtask['required_agent'], 'output': response['completion']}

 def _aggregate_results(self, results: dict) -> str:
 """聚合 Worker 结果生成最终响应"""
 prompt = f"""Aggregate the following worker results into a coherent response:

 Results: {results}

 Provide a clear, well-structured final answer."""

 response = self.bedrock.invoke_agent(
 agentAliasId='orchestrator-alias',
 agentId='orchestrator-agent-id',
 sessionId='session-aggregate',
 inputText=prompt
 )

 return response['completion']

class HierarchicalAgentManager:
 def __init__(self):
 self.strategy_manager = 'arn:aws:bedrock:...:agent/strategy-manager'
 self.domain_managers = {
 'engineering': 'arn:aws:bedrock:...:agent/eng-manager',
 'sales': 'arn:aws:bedrock:...:agent/sales-manager',
 'operations': 'arn:aws:bedrock:...:agent/ops-manager',
 }

 def process_enterprise_request(self, request: str) -> dict:
 """层级处理企业级请求"""

 # Level 2: 战略分解
 strategy_response = self._invoke_agent(
 self.strategy_manager,
 f"""Analyze this enterprise request and create a hierarchical plan:
 Request: {request}

 Return:

 - Strategic objective
 - Domain assignments (engineering/sales/operations)
 - Coordination requirements
 """
 )

 domain_tasks = strategy_response['domain_assignments']

 # Level 1: 领域管理并行执行
 domain_results = {}
 for domain, task in domain_tasks.items():
 domain_results[domain] = self._invoke_agent(
 self.domain_managers[domain],
 task
 )

 # Level 2: 战略整合
 final_response = self._invoke_agent(
 self.strategy_manager,
 f"""Integrate domain results into final strategic response:

 Domain Results: {domain_results}

 Provide unified recommendation with risk assessment.
 """
 )

 return {
 'strategy': strategy_response,
 'domain_results': domain_results,
 'final': final_response
 }

# 使用 Bedrock Converse API 实现 Agent 间通信
class AgentCommunication:
 def __init__(self):
 self.bedrock = boto3.client('bedrock-agent-runtime')

 def agent_to_agent_message(
 self,
 from_agent: str,
 to_agent: str,
 message: str,
 context: dict = None
 ) -> dict:
 """Agent 间直接消息传递"""

 payload = {
 'source_agent': from_agent,
 'target_agent': to_agent,
 'message': message,
 'context': context or {},
 'protocol': 'bedrock-a2a-v1',
 'timestamp': datetime.utcnow().isoformat()
 }

 response = self.bedrock.invoke_agent(
 agentAliasId=to_agent.split('/')[-1],
 agentId=to_agent.split(':agent/')[1].split('/')[0],
 sessionId=f"a2a-{from_agent}-{to_agent}",
 inputText=f"""Process this agent-to-agent message:

 From: {from_agent}
 Message: {message}
 Context: {context}

 Provide a structured response for the sending agent.
 """
 )

 return {
 'status': 'delivered',
 'response': response['completion'],
 'metadata': payload
 }

import boto3
from botocore.config import Config

# Account B: 配置跨账户访问 Account A 的 Agent
class CrossAccountAgentClient:
 def __init__(self, provider_account_id, provider_region='us-east-1'):
 self.provider_account_id = provider_account_id

 # 配置跨账户客户端
 self.bedrock = boto3.client(
 'bedrock-agent-runtime',
 region_name=provider_region,
 config=Config(
 signature_version='v4',
 retries={'max_attempts': 3}
 )
 )

 # Account B 的执行角色（需被 Account A 信任）
 self.execution_role = 'arn:aws:iam::222222222222:role/BedrockCrossAccountRole'

 def invoke_provider_agent(
 self, 
 agent_id: str,
 alias_id: str,
 input_text: str,
 session_id: str = None
 ) -> dict:
 """跨账户调用 Provider 的 Agent"""

 # 使用 AssumeRole 获取临时凭证
 sts = boto3.client('sts')
 assumed = sts.assume_role(
 RoleArn=f'arn:aws:iam::{self.provider_account_id}:role/BedrockCrossAccountInvoke',
 RoleSessionName=f'bedrock-cross-account-{session_id or uuid.uuid4().hex[:8]}'
 )

 # 用临时凭证创建客户端
 client = boto3.client(
 'bedrock-agent-runtime',
 aws_access_key_id=assumed['Credentials']['AccessKeyId'],
 aws_secret_access_key=assumed['Credentials']['SecretAccessKey'],
 aws_session_token=assumed['Credentials']['SessionToken']
 )

 response = client.invoke_agent(
 agentAliasId=alias_id,
 agentId=agent_id,
 sessionId=session_id or str(uuid.uuid4()),
 inputText=input_text
 )

 return response

# Account A: 设置跨账户信任策略
trust_policy = {
 "Version": "2012-10-17",
 "Statement": [
 {
 "Effect": "Allow",
 "Principal": {
 "AWS": "arn:aws:iam::222222222222:root"
 },
 "Action": [
 "bedrock:GetAgent",
 "bedrock:InvokeAgent"
 ],
 "Resource": [
 "arn:aws:bedrock:us-east-1:111111111111:agent/agent-id/*"
 ]
 }
 ]
}

# 定义 Action Group (相当于 MCP Tools)
action_group = {
 "name": "code_analysis_tools",
 "description": "Tools for code analysis and refactoring",
 "apiSchema": {
 "openapi": "3.0.0",
 "info": {"title": "Code Analysis API", "version": "1.0"},
 "paths": {
 "/analyze": {
 "post": {
 "summary": "Analyze code structure",
 "requestBody": {
 "content": {
 "application/json": {
 "schema": {
 "type": "object",
 "properties": {
 "code": {"type": "string"},
 "language": {"type": "string"}
 }
 }
 }
 }
 },
 "responses": {
 "200": {
 "description": "Analysis result",
 "content": {
 "application/json": {
 "schema": {
 "type": "object",
 "properties": {
 "complexity": {"type": "number"},
 "issues": {"type": "array"}
 }
 }
 }
 }
 }
 }
 }
 }
 }
 }
}

# 使用 Bedrock Converse API 进行函数调用
class FunctionCallingAgent:
 def __init__(self):
 self.bedrock = boto3.client('bedrock-agent-runtime')

 def invoke_with_functions(
 self,
 agent_id: str,
 alias_id: str,
 messages: list,
 functions: list
 ) -> dict:
 """带函数调用的 Agent 调用"""

 response = self.bedrock.invoke_agent(
 agentAliasId=alias_id,
 agentId=agent_id,
 sessionId=str(uuid.uuid4()),
 inputText=messages[-1]['content'],
 inferenceConfiguration={
 'maxTokens': 4096,
 'temperature': 0.7,
 'topP': 0.9
 },
 toolConfiguration={
 'functions': {
 'properties': {f['name']: {'description': f['description']} for f in functions}
 }
 }
 )

 # 处理函数调用响应
 if 'functionInvocations' in response:
 for invocation in response['functionInvocations']:
 result = self._execute_function(
 invocation['name'],
 invocation['arguments']
 )
 messages.append({
 'role': 'user',
 'content': f"Function {invocation['name']} returned: {result}"
 })

 return response

 def _execute_function(self, name: str, args: dict) -> dict:
 """执行函数并返回结果"""
 function_map = {
 'search_code': self._search_code,
 'read_file': self._read_file,
 'write_file': self._write_file,
 'run_tests': self._run_tests,
 }

 if name in function_map:
 return function_map[name](**args)
 return {'error': f'Unknown function: {name}'}

class BedrockSessionManager:
 """优化 Session 重用以提升性能"""

 def __init__(self):
 self.sessions = {} # session_id -> metadata
 self.default_ttl = 3600 # 1 hour

 def get_or_create_session(
 self, 
 agent_id: str, 
 user_id: str,
 reset_if_exists: bool = False
 ) -> str:
 """获取或创建 Session"""
 session_key = f"{agent_id}:{user_id}"

 if session_key in self.sessions and not reset_if_exists:
 session_data = self.sessions[session_key]
 if self._is_valid(session_data):
 return session_data['session_id']

 # 创建新 Session
 session_id = str(uuid.uuid4())
 self.sessions[session_key] = {
 'session_id': session_id,
 'agent_id': agent_id,
 'user_id': user_id,
 'created_at': datetime.utcnow(),
 'last_used': datetime.utcnow(),
 'turn_count': 0
 }

 return session_id

 def update_session_activity(self, session_id: str):
 """更新 Session 活跃时间"""
 for session_data in self.sessions.values():
 if session_data['session_id'] == session_id:
 session_data['last_used'] = datetime.utcnow()
 session_data['turn_count'] += 1
 break

from functools import wraps
import threading

class BedrockConcurrencyController:
 """控制 Agent 调用并发，避免限流"""

 def __init__(self, max_concurrent=10, rate_limit=50):
 self.semaphore = threading.Semaphore(max_concurrent)
 self.rate_limiter = TokenBucket(rate_limit, window=60) # 50 req/min
 self._lock = threading.Lock()
 self._active_calls = 0

 def execute_with_control(self, func, *args, **kwargs):
 """带并发控制的执行"""
 with self.semaphore:
 if not self.rate_limiter.consume(1):
 raise BedrockRateLimitError("Rate limit exceeded, retry later")

 with self._lock:
 self._active_calls += 1

 try:
 return func(*args, **kwargs)
 finally:
 with self._lock:
 self._active_calls -= 1

class TokenBucket:
 """简单的 Token Bucket 限流器"""

 def __init__(self, rate: int, window: int = 60):
 self.rate = rate
 self.window = window
 self.tokens = rate
 self.last_refill = time.time()

 def consume(self, tokens: int = 1) -> bool:
 self._refill()
 if self.tokens >= tokens:
 self.tokens -= tokens
 return True
 return False

 def _refill(self):
 now = time.time()
 elapsed = now - self.last_refill
 new_tokens = (elapsed / self.window) * self.rate
 self.tokens = min(self.rate, self.tokens + new_tokens)
 self.last_refill = now

from botocore.exceptions import ClientError
import backoff

class BedrockAgentRetry:
 """Agent 调用的智能重试策略"""

 def __init__(self, max_retries=3):
 self.max_retries = max_retries

 @backoff.on_exception(
 backoff.expo,
 (ClientError,),
 max_tries=3,
 base=2,
 jitter=backoff.full_jitter,
 factor=1.0
 )
 def invoke_with_retry(self, agent_id: str, alias_id: str, input_text: str):
 """带指数退避的重试调用"""
 try:
 response = bedrock.invoke_agent(
 agentAliasId=alias_id,
 agentId=agent_id,
 sessionId=str(uuid.uuid4()),
 inputText=input_text
 )
 return response

 except ClientError as e:
 error_code = e.response['Error']['Code']

 if error_code == 'ThrottlingException':

 # 限流，等待后重试
 raise
 elif error_code == 'ResourceNotFoundException':

 # Agent 不存在，不重试
 raise BedrockNotFoundError(f"Agent {agent_id} not found") from e
 elif error_code == 'AccessDeniedException':

 # 权限问题，不重试
 raise BedrockAccessDeniedError(f"Access denied to agent {agent_id}") from e
 else:
 raise

# 为 Agent 配置安全护栏
bedrock.put_agent_guardrail(
 agentIdentifier=f"arn:aws:bedrock:{region}:{account}:agent/{agent_id}",
 guardrailIdentifier='guardrail-xxxxx',
 guardrailVersion='DRAFT',

 # 或使用正式版本
 # guardrailVersion='1'
)

import boto3

cloudwatch = boto3.client('cloudwatch')

def emit_agent_metrics(agent_id: str, metrics: dict):
 """向 CloudWatch 发送 Agent 指标"""

 metric_data = [
 {
 'MetricName': 'InvocationCount',
 'Dimensions': [
 {'Name': 'AgentId', 'Value': agent_id}
 ],
 'Value': metrics.get('count', 1),
 'Unit': 'Count'
 },
 {
 'MetricName': 'Latency',
 'Dimensions': [
 {'Name': 'AgentId', 'Value': agent_id}
 ],
 'Value': metrics.get('latency_ms', 0),
 'Unit': 'Milliseconds'
 },
 {
 'MetricName': 'ErrorRate',
 'Dimensions': [
 {'Name': 'AgentId', 'Value': agent_id}
 ],
 'Value': metrics.get('error_rate', 0),
 'Unit': 'Percent'
 }
 ]

 cloudwatch.put_metric_data(
 Namespace='AWS/Bedrock/Agent',
 MetricData=metric_data
 )

import logging
import watchtower

# 配置 Agent 日志到 CloudWatch
logger = logging.getLogger('bedrock-agent')
logger.addHandler(watchtower.CloudWatchLogHandler(
 log_group='/aws/bedrock/agents',
 stream_name='prod-agent-01',
 boto3_session=boto3.Session()
))

def log_agent_interaction(
 agent_id: str,
 session_id: str,
 input_text: str,
 output_text: str,
 latency: float,
 error: str = None
):
 """结构化日志记录"""
 logger.info({
 'event': 'agent_interaction',
 'agent_id': agent_id,
 'session_id': session_id,
 'input_length': len(input_text),
 'output_length': len(output_text),
 'latency_ms': latency,
 'error': error
 })

import boto3

# 使用 EventBridge 触发 Agent
events = boto3.client('events')

# 创建 EventBridge 规则
events.put_rule(
 Name='agent-trigger-rule',
 EventPattern={
 "source": ["aws.custom"],
 "detail-type": ["AgentTask"],
 "detail": {
 "priority": ["high", "medium"]
 }
 },
 RoleArn='arn:aws:iam::123456789012:role/EventBridgeRole'
)

# 关联 Lambda 目标
events.put_targets(
 Rule='agent-trigger-rule',
 Targets=[
 {
 'Id': 'AgentLambda',
 'Arn': 'arn:aws:lambda:us-east-1:123456789012:function:AgentInvoker',
 'InputTransformer': {
 'InputPathsMap': {'task': '$.detail.task'},
 'InputTemplate': '{"task": <task>, "priority": <priority>}'
 }
 }
 ]
)