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](../ch11-**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>}'
            }
        }
    ]
)