Az Amazon Bedrock ügynökök olyanok, mint az intelligens asszisztensek az AWS infrastruktúrájához – érvelhetnek, eldönthetik, hogy mit tegyenek a következő lépésben, és a Lambda funkciók használatával cselekvéseket indíthatnak el. Ebben a cikkben megmutatom, hogyan építettem felügyelő ügynököt, amely több AWS Lambdát szervez: 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. Ellenőrizzük a diagramot és más példákat, hogy mi az ügynök, a jobb láthatóság és a megértés érdekében: A felhasználó felhívja a Bedrock ügynököt (1) valamilyen feladattal, mondjuk, „mennyi TV van a készletben?”. Az ügynök meghatározott utasítással tudja, hogy ha a kérdés a készlet állapotának ellenőrzésével kapcsolatos, akkor meg kell hívnia (2) az „adatbázis” akciócsoportot (3, AG). Az adatbázis AG-ben egy lambda funkciót határoztunk meg használni (4), és ez a lambda ellenőrzi az állapotot a DynamoDB táblázatban (5), megkapja a választ (6,7) és visszaadja a választ a felhasználónak (8). Nézzünk még egy példát: Minden ügynöknek több cselekvési csoportja lehet, például szeretnénk információt kapni néhány AWS erőforrásról, mint például az összes ECS-feladat felsorolása, a logika ugyanaz, mint az előzőnél. Még több példa: Hozzáadtunk még egy AG-t EKS cselekvési csoportokkal. Mint itt láthatja, minden cselekvési csoportnak több lambda funkciója is lehet, amelyhez kéréseket tehet. Az akciócsoport és a lambda funkció bármilyen funkcionalitással rendelkezhet, még akkor is, ha adatokat kell szereznie egy harmadik fél API-jából az időjárási adatok vagy a repülőjegyek rendelkezésre állása érdekében. Remélem, hogy most egy kicsit világos, és térjünk vissza a felügyelő ügynök beállításához: Az AWS konzolon nyissa meg a Bedrock → Ügynökök → Ügynök létrehozása Adjon nevet és hozzon létre Miután létrehozta, megváltoztathatja a modellt, ha akarja, vagy megtarthatja a Claude-t alapértelmezés szerint. Add hozzá az ügynök leírását és utasításait.A következő lépésben létrehozott cselekvési csoport Ön az AWS fő felügyelő ügynöke. Cél: Segítség az AWS infrastruktúrájának elemzésében. A cselekvési csoportok: ec2: list_instances → returns instance list + instanceIds A szabályok : Soha ne hívja közvetlenül az AWS API-kat. For EC2: Call ec2__list_instances Mindig használd a „gondolkodást” a cselekvés előtt. Ön az AWS fő felügyelő ügynöke. Cél: Segítség az AWS infrastruktúrájának elemzésében. A cselekvési csoportok: ec2: list_instances → visszaadja az instance listát + instanceIds A szabályok : Soha ne hívja közvetlenül az AWS API-kat. For EC2: Call ec2__list_instances Mindig használd a „gondolkodást” a cselekvés előtt. Megjegyzés : ec2 - cselekvési csoport neve list_instances - funkciónév, mint korábban említettem - minden akciócsoporthoz több funkció is tartozhat Kattintson a „Save” És a tetején a „Készítés” gombok.A készítés aktív lesz, ha ment. Csúsztassa le a cselekvési csoportba → Add Cselekvési csoport. felhívás - új lambda funkció létrehozása, ahol azonosnak kell lennie, mint ahogyan az ügynök utasításaiban meghatároztuk list_instances Adja hozzá a cselekvési csoport nevét és leírását, kattintson a létrehozásra, majd ismét a „Mentés” és a „Előkészítés” lehetőségre. Menjen a lambda függvényre, Bedrock létrehozta a függvényt az EC2 előtaggal a nevében, és adja hozzá ezt a kódot: 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)}' } MEGJEGYZÉS: a funkció válaszának Bedrock-specifikus formátumban kell lennie, a részletek megtalálhatók a dokumentációban: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html Miután frissítette a funkciókódot, válassza a Configuration → permissions → role name funkciót, és hozzon létre egy új inline politikát: Mint a JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Most visszatérhetünk az ügynökünkhöz, és kattintson a "Test", írja be a szöveget, hogy ellenőrizze, hogy tényleg működik-e: Az első cselekvési csoport a vártnak megfelelően működik, lehetővé teszi egy további cselekvési csoport hozzáadását a cloudwatch mutatóinak felsorolásához: Az akciócsoport neve - Cloudwatch A funkció neve getMetrics, adjunk hozzá leírást és paramétereket, mivel ennek a lambda-nak ismernie kell a példányt vagy intances-t a paraméterek ellenőrzéséhez. Frissítse az ügynök utasítását, hogy elmagyarázza, hogyan szeretnénk használni az új cselekvési csoportot, majd ismét kattintson a „Ments” és a „Előkészítés” gombra. Ön az AWS fő felügyelő ügynöke. Cél: Segítség az AWS infrastruktúrájának elemzésében. A cselekvési csoportok: ec2: describeInstances → returns instance list + instanceIds cloudwatch: getMetrics → igényel instance_ids A szabályok : Soha ne hívja közvetlenül az AWS API-kat. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Az eredmények kombinálása. Mindig használd a „gondolkodást” a cselekvés előtt. Ön az AWS fő felügyelő ügynöke. Cél: Segítség az AWS infrastruktúrájának elemzésében. A cselekvési csoportok: ec2: describeInstances → returns instance list + instanceIds cloudwatch: getMetrics → igényel instance_ids A szabályok : Soha ne hívja közvetlenül az AWS API-kat. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Az eredmények kombinálása. Mindig használd a „gondolkodást” a cselekvés előtt. Most lehetővé teszi, hogy frissítse a cloudwatch funkciók kódját: 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)}" } És frissítse a cloudwatch lambda engedélyeket, mint az ec2 lambda esetében: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } Próbáld ki újra EC2 és CloudWatch cselekvési csoportjaink vannak, és az ügynök hívhatja őket, hogy megkapja az EC2 példányok és CPU-metrikáik listáját. Ahelyett, hogy az ügynök mind az EC2-t, mind a CloudWatch-ot külön-külön hívná, a felügyelő gondoskodik erről a logikáról.Először az EC2-funkciót hívja, hogy az összes példányt megkapja, majd átadja ezeket a példányazonosítókat a CloudWatch-funkcióra a mutatók megszerzéséhez, és végül mindent egyértelmű eredményre egyesít. Ily módon az ügynöknek csak egy műveletre van szüksége - a felügyelőre -, míg a felügyelő koordinálja az összes lépést a háttérben. Adjon meg egy nevet és leírást Adja meg a funkció nevét és leírását És frissítse az ügynök utasításait, hogy elkerülje az ec2 és a CloudWatch cselekvési csoportok közvetlen hívását: Kattintson a „Megmentés” és a „Előkészítés” gombra. A felügyelő lambda funkciók kódjának frissítése, 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}" Ismét hozzáadjuk a felügyeleti lambda engedélyeket az EC2 és a Cloudwatch funkciókhoz, például: { "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>" ] } ] } Újra teszteljük a funkciót, és meglepő módon nem sikerül Ellenőriztem a Supervisor felügyelő funkció naplóit, és ezt látom Az egyik úgy tűnik, hogy nem mutat semmit hasznos, de nem - a trükk 3000.00ms. az alapértelmezett lambda funkció időtartama, lehetővé teszi, hogy állítsa be. Menj a felügyelő funkció - konfiguráció - általános és szerkessze a Timeout paraméter , megváltoztattam 10 másodperc És ez segített! Ezt a funkciót tovább bővítheti az AWS számlázási elemzés hozzáadásával, hogy megtalálja a legdrágább erőforrásokat vagy a legtöbbet. , és így tovább, és nem kell kizárólag az AWS erőforrásokra korlátozódnia. Drága ec2 példányok futtatása
