Amazon Bedrock エージェントは、AWS インフラストラクチャのスマートアシスタントのように、考え、次に何をすべきかを決定し、Lambda 関数を使用してアクションを引き起こすことができます。 この記事では、複数の AWS Lambdas をオーケストラするスーパーバイザー エージェントをどのように構築したかを示します。 List EC2 instances, Fetch their CPU metrics from CloudWatch, Combine both results intelligently — all without the agent ever calling AWS APIs directly. By the end, you’ll understand how Bedrock Agents work, how to use action groups, and how to chain Lambdas through a supervisor function — a clean, scalable pattern for multi-step automation. より良い視覚性と理解のために、エージェントが何であるかを図面と他の例で確認しましょう: ユーザーがBedrockエージェント(1)に電話をかけると、いくつかのタスク、すなわち「どのくらいのテレビが在庫にあるか?」と言えば、エージェントは定義されたプロンプトで、質問が在庫状態をチェックすることに関連している場合、彼らは(2)「データベース」アクショングループ(3, AG)に電話する必要があります。 もう一つ例を調べてみましょう: 各エージェントには複数のアクショングループがある場合があります、たとえば、私たちは、すべてのECSタスクをリストするように、いくつかのAWSリソースについての情報を取得したい、論理は前者と同じです。 さらなる例: EKS アクション グループを含む 1 つの AG を追加しました. ここで見るように、各アクション グループには 1 つ以上の lambda 関数がある場合があります. In this example, it is listing and deleting resources from some existing K8S cluster. アクショングループとランバダ機能は、気象データやフライトチケットの可用性を取得するために、サードパーティのAPIからデータを取得する必要がある場合でも、任意の機能を持っている可能性があります。 今はもう少しはっきりしているといいけど、監督エージェントの設定に戻ろう。 AWS コンソールで Bedrock → Agents → Create agent を開きます。 名前を付け、創る。 一度作成すると、あなたが欲しい場合はモデルを変更したり、デフォルトでクロードを保持することができます. I will change to Nova Premier 1.0 エージェントの説明と指示を追加します. 次のステップで作成するアクショングループ あなたは主な AWS Supervisor エージェントです。 目的:AWSインフラストラクチャの分析を支援します。 アクショングループ: ec2: list_instances → インスタンスリスト + instanceIds を返します。 ルール : 決して AWS API を直接呼び出すことはありません。 For EC2: Call ec2__list_instances 行動の前に「考える」ことを常に使います。 あなたは主な AWS Supervisor エージェントです。 目的:AWSインフラストラクチャの分析を支援します。 アクショングループ: ec2: list_instances → インスタンスリスト + instanceIds を返します。 ルール : 決して AWS API を直接呼び出すことはありません。 For EC2: Call ec2__list_instances 行動の前に「考える」ことを常に使います。 メモ: ec2 - アクショングループ名 list_instances - function name, as I mentioned previously - you can have multiple functions per each action group. list_instances - function name, as I mentioned previously - you can have multiple functions per each action group. そして「Save」をクリックします。 上の「準備」ボタンは、保存すると準備が活発になります。 Scroll down to the actions group → add アクショングループ. invocation - create a new lambda function. アクショングループ. invocation - create a new lambda function. エージェントの指示で定義したものと同じでなければなりません。 list_instances アクション グループの名前と説明を追加し、Create をクリックし、再度「Save」と「Prepare」をクリックします。 あなたの lambda 関数に移動し、Bedrock は名前に EC2 プレフィックスを含む関数を作成し、次のコードを追加します。 import logging from typing import Dict, Any from http import HTTPStatus import boto3 logger = logging.getLogger() logger.setLevel(logging.INFO) ec2_client = boto3.client('ec2') def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """ AWS Lambda handler for processing Bedrock agent requests related to EC2 instances. Supports: - Listing all EC2 instances - Describing a specific instance by ID """ try: action_group = event['actionGroup'] function = event['function'] message_version = event.get('messageVersion', 1) parameters = event.get('parameters', []) response_text = "" if function == "list_instances": # List all EC2 instances instances = ec2_client.describe_instances() instance_list = [] for reservation in instances['Reservations']: for instance in reservation['Instances']: instance_list.append({ 'InstanceId': instance.get('InstanceId'), 'State': instance.get('State', {}).get('Name'), 'InstanceType': instance.get('InstanceType'), 'PrivateIpAddress': instance.get('PrivateIpAddress', 'N/A'), 'PublicIpAddress': instance.get('PublicIpAddress', 'N/A') }) response_text = f"Found {len(instance_list)} EC2 instance(s): {instance_list}" elif function == "describe_instance": # Expect a parameter with the instance ID instance_id_param = next((p for p in parameters if p['name'] == 'instanceId'), None) if not instance_id_param: raise KeyError("Missing required parameter: instanceId") instance_id = instance_id_param['value'] result = ec2_client.describe_instances(InstanceIds=[instance_id]) instance = result['Reservations'][0]['Instances'][0] response_text = ( f"Instance {instance_id} details: " f"State={instance['State']['Name']}, " f"Type={instance['InstanceType']}, " f"Private IP={instance.get('PrivateIpAddress', 'N/A')}, " f"Public IP={instance.get('PublicIpAddress', 'N/A')}" ) else: response_text = f"Unknown function '{function}' requested." # Format Bedrock agent response response_body = { 'TEXT': { 'body': response_text } } action_response = { 'actionGroup': action_group, 'function': function, 'functionResponse': { 'responseBody': response_body } } response = { 'response': action_response, 'messageVersion': message_version } logger.info('Response: %s', response) return response except KeyError as e: logger.error('Missing required field: %s', str(e)) return { 'statusCode': HTTPStatus.BAD_REQUEST, 'body': f'Error: {str(e)}' } except Exception as e: logger.error('Unexpected error: %s', str(e)) return { 'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR, 'body': f'Internal server error: {str(e)}' } 注: 関数の回答は、Bedrock 特有の形式でなければなりませんが、詳細は文書に記載されています。 https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html 機能コードを更新した後 - Configuration → permissions → role name で新しい inline ポリシーを作成します。 JSONについて: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } 今、私たちは私たちのエージェントに戻り、「テスト」をクリックし、テキストを入力して、実際に機能しているかどうかを確認することができます。 クール! 最初のアクショングループは予想通りに動作し、Cloudwatchのメトリクスをリストするためのもう1つのアクショングループを追加します。 アクショングループの名前 - cloudwatch 関数の名前は getMetrics で、この lambda はインスタンスまたは intances を知っておく必要があるため、 記述とパラメータを追加します。 エージェントのプロンプトを更新して、新しいアクショングループを使用する方法を説明し、「保存」と「準備」を再度クリックします。 あなたは主な AWS Supervisor エージェントです。 目的:AWSインフラストラクチャの分析を支援します。 アクショングループ: ec2: describeInstances → returns instance list + instanceIds を返します。 cloudwatch: getMetrics → needs instance_ids ルール : Never call AWS APIs directly. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics 結果を組み合わせる。 行動の前に「考える」ことを常に使います。 あなたは主な AWS Supervisor エージェントです。 目的:AWSインフラストラクチャの分析を支援します。 アクショングループ: ec2: describeInstances → returns instance list + instanceIds を返します。 cloudwatch: getMetrics → needs instance_ids ルール : 決して AWS API を直接呼び出すことはありません。 For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics 結果を組み合わせる。 行動の前に「考える」ことを常に使います。 Cloudwatchの機能コードを更新します。 import boto3 import datetime import logging import json from typing import Dict, Any from http import HTTPStatus logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: try: action_group = event["actionGroup"] function = event["function"] message_version = event.get("messageVersion", 1) parameters = event.get("parameters", []) region = "us-east-1" instance_ids = [] # --- Parse parameters --- for param in parameters: if param.get("name") == "region": region = param.get("value") elif param.get("name") == "instance_ids": raw_value = param.get("value") if isinstance(raw_value, str): # Clean up stringified list from Bedrock agent raw_value = raw_value.strip().replace("[", "").replace("]", "").replace("'", "") instance_ids = [x.strip() for x in raw_value.split(",") if x.strip()] elif isinstance(raw_value, list): instance_ids = raw_value logger.info(f"Parsed instance IDs: {instance_ids}") if not instance_ids: response_text = f"No instance IDs provided for CloudWatch metrics in {region}." else: cloudwatch = boto3.client("cloudwatch", region_name=region) now = datetime.datetime.utcnow() start_time = now - datetime.timedelta(hours=1) metrics_output = [] for instance_id in instance_ids: try: metric = cloudwatch.get_metric_statistics( Namespace="AWS/EC2", MetricName="CPUUtilization", Dimensions=[{"Name": "InstanceId", "Value": instance_id}], StartTime=start_time, EndTime=now, Period=300, Statistics=["Average"] ) datapoints = metric.get("Datapoints", []) if datapoints: datapoints.sort(key=lambda x: x["Timestamp"]) avg_cpu = round(datapoints[-1]["Average"], 2) metrics_output.append(f"{instance_id}: {avg_cpu}% CPU (avg last hour)") else: metrics_output.append(f"{instance_id}: No recent CPU data") except Exception as e: logger.error(f"Error fetching metrics for {instance_id}: {e}") metrics_output.append(f"{instance_id}: Error fetching metrics") response_text = ( f"CPU Utilization (last hour) in {region}:\n" + "\n".join(metrics_output) ) # --- Bedrock Agent response format --- response_body = { "TEXT": { "body": response_text } } action_response = { "actionGroup": action_group, "function": function, "functionResponse": { "responseBody": response_body } } response = { "response": action_response, "messageVersion": message_version } logger.info("Response: %s", response) return response except Exception as e: logger.error(f"Unexpected error: {e}") return { "statusCode": HTTPStatus.INTERNAL_SERVER_ERROR, "body": f"Internal server error: {str(e)}" } そして、ec2 lambdaで行ったように、CloudwatchのLambda権限を更新します。 { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } また試しに♪ EC2 と CloudWatch のアクション グループは、エージェントから呼び出され、EC2 インスタンスとその CPU メトリックのリストを得ることができます。 エージェントが EC2 と CloudWatch の両方を別々に呼び出す代わりに、スーパージャーはその論理を管理します。最初に EC2 関数を呼び出してすべてのインスタンスを取得し、その後インスタンス ID を CloudWatch 関数に転送してメトリクスを取得し、最後にすべてを一つの明確な結果に組み合わせます。 このようにして、エージェントは1つのアクション - スーパーマネージャー - を呼び出さなければなりませんが、スーパーマネージャーは背景のすべてのステップを調整します。 名前と記述を教えてください。 機能名と説明を指定する また、エージェントの指示を更新して、ec2 と CloudWatch のアクション グループへの直接の呼び出しを回避します。 「保存」と「準備」をクリックします。 監督のLambda機能コードを更新し、 NOTE: need to update your EC2 and Cloudwatch functions name in the code below: import boto3 import json import logging import re import ast logger = logging.getLogger() logger.setLevel(logging.INFO) lambda_client = boto3.client("lambda") def lambda_handler(event, context): try: action_group = event["actionGroup"] function = event["function"] parameters = event.get("parameters", []) message_version = event.get("messageVersion", "1.0") # Parse parameters region = "us-east-1" for param in parameters: if param.get("name") == "region": region = param.get("value") # Decide routing if function == "analyzeInfrastructure": logger.info("Supervisor: calling EC2 and CloudWatch") # Step 1: call EC2 Lambda ec2_payload = { "actionGroup": "ec2", "function": "list_instances", "parameters": [{"name": "region", "value": region}], "messageVersion": "1.0" } ec2_response = invoke_lambda("ec2-yeikw", ec2_payload) #### CHANGE TO YOUR EC2 FUNCTION NAME instances = extract_instance_ids(ec2_response) # Step 2: call CloudWatch Lambda (if instances found) if instances: cw_payload = { "actionGroup": "cloudwatch", "function": "getMetrics", "parameters": [ {"name": "region", "value": region}, {"name": "instance_ids", "value": instances} ], "messageVersion": "1.0" } cw_response = invoke_lambda("cloudwatch-ef6ty", cw_payload) #### CHANGE TO YOUR CLOUDWATCH FUNCTION NAME final_text = merge_responses(ec2_response, cw_response) else: final_text = "No instances found to analyze." else: final_text = f"Unknown function: {function}" # Construct Bedrock-style response response = { "messageVersion": message_version, "response": { "actionGroup": action_group, "function": function, "functionResponse": { "responseBody": { "TEXT": {"body": final_text} } } } } logger.info("Supervisor response: %s", response) return response except Exception as e: logger.exception("Error in supervisor") return { "statusCode": 500, "body": f"Supervisor error: {str(e)}" } def invoke_lambda(name, payload): """Helper to call another Lambda and parse response""" response = lambda_client.invoke( FunctionName=name, InvocationType="RequestResponse", Payload=json.dumps(payload), ) result = json.loads(response["Payload"].read()) return result def extract_instance_ids(ec2_response): """Extract instance IDs from EC2 Lambda response""" try: body = ec2_response["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] # Try to extract JSON-like data after "Found X EC2 instance(s):" if "Found" in body and "[" in body and "]" in body: data_part = body.split(":", 1)[1].strip() try: instances = ast.literal_eval(data_part) # safely parse the list return [i["InstanceId"] for i in instances if "InstanceId" in i] except Exception: pass # fallback regex in case of plain text return re.findall(r"i-[0-9a-f]+", body) except Exception as e: logger.error("extract_instance_ids error: %s", e) return [] def merge_responses(ec2_resp, cw_resp): """Combine EC2 and CloudWatch outputs""" ec2_text = ec2_resp["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] cw_text = cw_resp["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] return f"{ec2_text}\n\n{cw_text}" また、監督 lambda 許可を追加して EC2 および Cloudwatch の機能を呼び出します。 { "Version": "2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": [ "arn:aws:lambda:us-east-1:<account_id>:function:ec2-<id>", "arn:aws:lambda:us-east-1:<account_id>:function:cloudwatch-<id>" ] } ] } 機能を再度テストし、驚くほど失敗します。 スーパーバイザーのスーパーバイザー機能ログを確認し、これを見ました。 One it seems it doesn't show anything useful, but doesn't - the hint it 3000.00ms. its default lambda function timeout, lets adjust it. Go to supervisor function - configuration - general and edit Timeout parameter , I changed to 10 seconds 助けてくれました! この機能をさらに拡張するには、AWS 請求分析を追加して、最も高価なリソースまたは最も高価なリソースを見つけることができます。 , など、AWS リソースだけに制限される必要はありません。 あなたが実行している高価な ec2 インスタンス
