Amazon Bedrock Multi-Agents Collaboration Workshop

To process video or audio files, upload them to S3 and then invoke the BDA runtime using invoke_data_automation_async.

Workflow:

1. Upload file to S3 (e.g., bda/input/video/).
2. Call invoke_data_automation_async providing the inputConfiguration (S3 URI), outputConfiguration (S3 URI), and the dataAutomationProjectArn.
3. Poll the status using get_data_automation_status with the returned invocationArn until the status is Success, ServiceError, or ClientError.
4. Download the resulting result.json from the BDA output location to prepare it for Knowledge Base ingestion.

response_vid = bda_runtime_client.invoke_data_automation_async(
    inputConfiguration={'s3Uri':  f"s3://{bucket_name}/{file_input}"},
    outputConfiguration={'s3Uri': f"s3://{bucket_name}/{output_name}"},
    dataAutomationProfileArn= f'arn:aws:bedrock:{region}:{account_id}:data-automation-profile/us.data-automation-v1',
    dataAutomationConfiguration={
        'dataAutomationProjectArn':kb_project_arn,
        'stage': 'DEVELOPMENT'
    })

Action groups allow the agent to perform tasks like detecting peaks or redistributing energy quotas. This involves two steps: defining a Lambda function and defining the function metadata.

1. Lambda Function: The Lambda must handle the event and return a response in the format: {'response': {'actionGroup': ..., 'function': ..., 'functionResponse': {'responseBody': {'TEXT': {'body': ...}}}}}.

2. Function Definitions: Define a list of dictionaries (functions_def) specifying the name, description, and parameters (name, type, required status) for each tool.

3. Association: Use agents.add_action_group_with_lambda to link the Lambda, the function definitions, and the agent together.

functions_def = [
    {
        "name": "detect_peak",
        "description": "detect consumption peak during current month",
        "parameters": {
            "customer_id": {
                "description": "The ID of the customer",
                "required": True,
                "type": "string"
            }
        }
    },
    # ... other functions
]

resp = agents.add_action_group_with_lambda(
    agent_name=peak_agent_name,
    lambda_function_name=peak_lambda_name,
    source_code_file="peak_load.py",
    agent_functions=functions_def,
    agent_action_group_name="peak_load_actions",
    agent_action_group_description="Function to get usage, peaks, redistribution for a user",
    dynamo_args=dynamoDB_args
)

To process multimodal content (video/audio) for the agent's Knowledge Base, you must create a BDA project. This project defines how unstructured data is extracted and summarized.

When creating the project via create_data_automation_project, you can configure:

• Video Extraction: Content moderation, text detection, transcript, bounding boxes, and generative fields (summaries, chapters).
• Audio Extraction: Audio content moderation, topic moderation, transcript, and generative fields (summaries, topic summaries).

The project is reusable across multiple files sharing the same configuration.

response = bda_client.create_data_automation_project(
    projectName=project_name,
    projectDescription='BDA video processing project',
    projectStage='DEVELOPMENT',
    standardOutputConfiguration={
        "video": {
            "extraction": {
                "category": {
                    "state": "ENABLED",
                    "types": ["CONTENT_MODERATION", "TEXT_DETECTION", "TRANSCRIPT"]
                },
                "boundingBox": {"state": "ENABLED"}
            },
            "generativeField": {
                "state": "ENABLED",
                "types": ["VIDEO_SUMMARY", "CHAPTER_SUMMARY", "IAB"]
            }
        },
        "audio": {
            "extraction": {
                "category": {
                    "state": "ENABLED", 
                    "types": ["AUDIO_CONTENT_MODERATION", "TOPIC_CONTENT_MODERATION", "TRANSCRIPT"]
                }
            },
            "generativeField": {
                "state": "ENABLED",
                "types": ["AUDIO_SUMMARY", "TOPIC_SUMMARY", "IAB"]
            }
        }
    }
)

When an agent invokes an action group, it calls an AWS Lambda function. The Lambda must parse the event object to identify the requested function and its parameters, then return a response in a specific format compatible with Bedrock Agents.

Required Response Format:

def populate_function_response(event, response_body):
    return {
        'response': {
            'actionGroup': event['actionGroup'], 
            'function': event['function'],
            'functionResponse': {
                'responseBody': {
                    'TEXT': {'body': str(response_body)}
                }
            }
        }
    }

Lambda Logic Pattern:

1. Extract function name from event.get('function').
2. Extract parameters using a helper (e.g., get_named_parameter(event, name)).
3. Execute business logic (e.g., DynamoDB query).
4. Return the formatted response using populate_function_response.

To build a functional Solar Panel agent, you must combine three components:

1. Instructions: A system prompt defining the agent's persona (e.g., 'Solar Energy Assistant'), capabilities, and response style.
2. Knowledge Base (RAG): Provides the agent with technical documentation (e.g., installation and maintenance manuals) via associate_kb_with_agent.
3. Action Group (Lambda): Provides the agent with tools to perform tasks, such as open_ticket or get_ticket_status. This is configured using add_action_group_with_lambda and requires a defined list of agent_functions (name, description, and parameters).

from utils.bedrock_agent_helper import AgentsForAmazonBedrock

agents = AgentsForAmazonBedrock()

# 1. Create Agent
solar_agent = agents.create_agent(
    solar_agent_name,
    agent_description,
    agent_instruction,
    agent_foundation_model,
    kb_arns=[kb_arn]
)

# 2. Associate Knowledge Base
agents.associate_kb_with_agent(
    solar_agent[0], 
    'Access the knowledge base when customers ask about to install and maintain solar panels',
    kb_id
)

# 3. Add Action Group
functions_def = [
    {
        "name": "open_ticket",
        "description": "Create a ticket to get help...",
        "parameters": {
            "customer_id": {"description": "Unique customer identifier", "required": True, "type": "string"},
            "msg": {"description": "The reason...", "required": True, "type": "string"}
        }
    }
]

agents.add_action_group_with_lambda(
    agent_name=solar_agent_name,
    lambda_function_name=solar_lambda_name,
    source_code_file="solar_energy.py",
    agent_functions=functions_def,
    agent_action_group_name="solar_energy_actions",
    agent_action_group_description="Function to open an energy ticket...",
    dynamo_args=dynamoDB_args
)