Amazon Bedrock Агентүүд таны AWS инфраструктурд ухаалаг ажилтнууд шиг байдаг - тэдний асуултууд, дараагийн үйл явцыг шийдвэрлэх, Lambda функцийг ашиглан үйл ажиллагаа явуулж болно. Энэ нийтлэлд, би хэд хэдэн AWS Lambdas-ийг оркестрийг хэрхэн үүсгэх Supervisor Agent-ийг хэрхэн үүсгэх гэж үзнэ үү: 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) "DataBase" үйл ажиллагааны бүлэг (3, AG) зовох хэрэгтэй. Багаж AG-д бид ашиглах ламбда функцийг тохируулж байна (4), энэ ламбда DynamoDB тавтай морилно уу (5), хариу авах (6,7) болон хэрэглэгчдэд хариу хандах болно (8). Хамгийн сүүлийн үеийн загварууд Бүх агент нь хэд хэдэн үйл ажиллагааны бүлэгтэй байж болно. Жишээ нь, бид зарим AWS ресурсын талаархи мэдээлэл авахыг хүсч байна. Эцэст нь бүх ECS үйл явцыг жагсаалтад, логик нь өмнөх нэгтэй байдаг. AG-ийг зовохыг тайлбарлахын тулд - бид үүнийг ашиглах хэрэгтэй. Үүнээс дээш жишээ: Бид EKS үйл ажиллагааны бүлэгтэй нэг AG-ийг нэмж байна. Хэрэв та энд үзнэ үү, бүр үйл ажиллагааны бүлэг нь хүсэл хийхын тулд нэгээс дээш lambda функцтэй байж болно. Энэ жишээ нь, энэ нь ямар ч одоогийн K8S кластерийн ресурсуудыг жагсаалтыг болон илрүүлэх юм. Ажлын бүлэг, lambda функц нь ямар ч функцтай байж болно, тэр ч байтугай цаг агаарын мэдээлэл эсвэл нисэх онгоцны билет хандахын тулд сторонний API-аас өгөгдлийг олж авах хэрэгтэй. Яагаад энэ нь одоо ч бас ягаан тухтай бөгөөд бидний супермаркетын агент тохиргоог унтраах болно: AWS консолид Bedrock → Агентүүд → Агент үүсгэхийн тулд Энэ нь нэр татаж, үүсгэх Нэг удаа үүсгэсэн, та хүсэж байгаа бол загварыг өөрчилж болно, эсвэл стандарттай Claude-ийг хадгалж болно. Би Nova Premier 1.0-д өөрчилж болно Агент зориулсан тодорхойлолт, зааврыг нэмнэ. Ажлын бүлэг бид дараагийн шатанд үүсгэх болно Та AWS Supervisor Агент юм. Загвар: AWS инфраструктурыйг анализ хийхэд туслах. Ажлын групп: ec2: list_instances → өөрсдийн жагсаалт жагсаалт жагсаалт Зохиогчийн эрх: AWS API-ийг шууд дуудлагагүй. For EC2: Call ec2__list_instances Үйлчилгээний өмнө үргэлж <thinking> ашиглах. Та AWS Supervisor Агент юм. Загвар: AWS инфраструктурыйг анализ хийхэд туслах. Ажлын групп: ec2: list_instances → өөрсдийн жагсаалт жагсаалт жагсаалт Зохиогчийн эрх: AWS API-ийг шууд дуудлагагүй. For EC2: Call ec2__list_instances Үйлчилгээний өмнө үргэлж <thinking> ашиглах. Сэтгэгдэл: ec2 - үйл ажиллагааны бүлэг нэр list_instances - функц нэр, би өмнө хэлсэн шиг - Та үйл ажиллагааны бүлэг бүр хэд хэдэн функц байж болно болон “Save” дээр клик Өнгөрсөн дээрх "Prepare" бутлууд. Та хадгалах үед Prepare идэвхжүүлэх болно. Ажлын групп руу шилжих → Add нэр - EC2. Action group. invocation - шинэ lambda функцийг үүсгэхийн тулд Агент заавралдаас тодорхойлогдсон хэлбэрээр байх ёстой 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 policy үүсгэхийн тулд: гэхэд JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Одоо бид манай агент руу ирдэг, "Тест" -ийг дарна уу, энэ нь үнэхээр ажиллаж байгаа эсэхийг шалгана уу: Cool! Эхний үйл ажиллагааны бүлэг нь хүссэн дагуу ажилладаг, Cloudwatch метрик жагсаалтанд нэг үйл ажиллагааны бүлэг нэмнэ үү: Ажлын группын нэр - Cloudwatch Функцийн нэр getMetrics, тодорхойлолт, параметрийг нэмнэ үү, учир нь энэ lambda нь метрикуудыг шалгахын тулд инстанц эсвэл intances мэдэх ёстой. Шинэ үйл ажиллагааны бүтэц хэрхэн ашиглахыг хүсэж байгаа эсэхийг санал болгож, "Save" болон "Prepare" -ийг дахин татаж авах Та 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 Бүтээгдэхүүн нэгтгэсэн Үйлчилгээний өмнө үргэлж <thinking> ашиглах. Та AWS Supervisor Агент юм. Загвар: AWS инфраструктурыйг анализ хийхэд туслах. Ажлын групп: ec2: describeInstances → Instances жагсаалт + InstancesIds returns Cloudwatch: getMetrics → instance_ids шаардлагатай Зохиогчийн эрх: AWS API-ийг шууд дуудлагагүй. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Бүтээгдэхүүн нэгтгэсэн Үйлчилгээний өмнө үргэлж <thinking> ашиглах. Одоо бидний 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)}" } Cloudwatch ламбда зөвшөөрөл шинэчлэх, бид ec2 ламбда нь хийж байна: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } Тэгээд туршиж Бид EC2 болон CloudWatch-ийн үйл ажиллагааны бүлэгүүдтэй бөгөөд EC2 инстанцын жагсаалт, CPU-ийн метрикийг олж авахын тулд агент нь тэднийг зочилж болно. Одоо энэ үйл явцыг ухаалаг, үр ашигтай болгохын тулд Супервизор функцийг нэмнэ үү. Агент EC2 болон CloudWatch-ийг тусгаарлагддаг боловч Супервизор энэ логикейг хадгалж байна. Энэ нь анх удаа EC2 функцийг олж авахын тулд бүх инстанцийг олж авах, дараа нь эдгээр инстанци ID-ийг CloudWatch функцийг олж авахын тулд метрикуудыг олж авах, эцэст нь бүх зүйлийг нэг ягаан туяаны үр дүнд нэгтгэх болно. Энэ арга замаар агент нь зөвхөн нэг үйл явцыг - Супермаркет - зовох хэрэгтэй бөгөөд Супермаркет нь позадин бүх үйл явцыг координатаар. Энэ нь цэвэршүүлэх, хурдан, хадгалах илүү хялбар юм. Энэ нь нэр, тодорхойлолт Функцийн нэр, тодорхойлолт Тэгээд agent-ийн зааврыг шинэчлэхын тулд ec2 болон CloudWatch-ийн үйл ажиллагааны бүлэгдэд шууд зовохын тулд: Дараа нь “Save” болон “Prepare” дээр дарна уу. Supervisor 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}" Бидний 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>" ] } ] } Хамгийн сүүлийн үеийн функцийг туршиж байна, энэ нь гайхамшигтай Би supervusor supervisor функц дневник шалгах бөгөөд энэ нь үзнэ үү нэг нь энэ нь ямар ч хэрэгцээтэй харуулж байна, гэхдээ биш - энэ нь илгээх 3000.00ms. түүний стандарт lambda функц timeout, тохируулах боломжийг олгодог. супервизор функц - конфигурац - ерөнхий болон редактировать Timeout параметр , Би 10 секундын хүртэл өөрчлөгдсөн Энэ нь тусалдаг! Та AWS-ийн арилжааны анализийг нэмж энэ функцийг дэлгэрэнгүй боловсруулах боломжтой. , гэх мэт, та зөвхөн AWS ресурсын хязгаарлагддаггүй байх ёстой. Зөвхөн өөр өөр функцтай байх болно. хямд ec2 инстанцийг ажиллуулах
