Agents for Bedrock

Duration: 60 min

Bedrock Agents enable foundation models to take actions by calling external APIs and Lambda functions. This module covers agent architecture, action groups, orchestration, and building autonomous AI systems.

What are Agents?

An agent is an AI system that can:

1. Understand user requests
2. Break down complex tasks into steps
3. Call external tools/APIs
4. Process results
5. Iterate until the goal is achieved

User Request
    ↓
Agent Reasoning
    ↓
Decide Action
    ↓
Call Tool/API
    ↓
Process Result
    ↓
Continue or Return Answer

Creating an Agent

import boto3
import json

client = boto3.client('bedrock-agent', region_name='us-east-1')

# Create an agent
response = client.create_agent(
    agentName='customer-support-agent',
    agentRoleArn='arn:aws:iam::ACCOUNT:role/BedrockAgentRole',
    description='Handles customer support queries',
    foundationModel='anthropic.claude-3-sonnet-20240229-v1:0',
    instruction="""You are a helpful customer support agent. 
    Use available tools to help customers with their issues.
    Always be polite and try to resolve issues quickly."""
)

agent_id = response['agent']['id']
print(f"Agent ID: {agent_id}")

Action Groups

Action groups define the tools an agent can use:

# Define an action group
action_group_config = {
    'actionGroupName': 'customer-tools',
    'description': 'Tools for customer support',
    'actionGroupExecutor': {
        'lambda': 'arn:aws:lambda:us-east-1:ACCOUNT:function:customer-support-handler'
    },
    'apiSchema': {
        'payload': json.dumps({
            "openapi": "3.0.0",
            "info": {"title": "Customer Support API", "version": "1.0"},
            "paths": {
                "/get-order": {
                    "post": {
                        "summary": "Get order details",
                        "parameters": [
                            {
                                "name": "order_id",
                                "in": "query",
                                "required": True,
                                "schema": {"type": "string"}
                            }
                        ]
                    }
                },
                "/refund": {
                    "post": {
                        "summary": "Process a refund",
                        "parameters": [
                            {
                                "name": "order_id",
                                "in": "query",
                                "required": True,
                                "schema": {"type": "string"}
                            },
                            {
                                "name": "reason",
                                "in": "query",
                                "required": True,
                                "schema": {"type": "string"}
                            }
                        ]
                    }
                }
            }
        })
    }
}

# Add action group to agent
response = client.create_agent_action_group(
    agentId=agent_id,
    agentVersion='DRAFT',
    actionGroupName=action_group_config['actionGroupName'],
    description=action_group_config['description'],
    actionGroupExecutor=action_group_config['actionGroupExecutor'],
    apiSchema=action_group_config['apiSchema']
)

Lambda Handler for Actions

# Lambda function to handle agent actions
import json
import boto3

def lambda_handler(event, context):
    """Handle agent action calls"""
    
    # Parse the action request
    action_group = event['actionGroup']
    api_path = event['apiPath']
    http_method = event['httpMethod']
    parameters = event.get('parameters', [])
    
    # Extract parameters
    params = {}
    for param in parameters:
        params[param['name']] = param['value']
    
    # Route to appropriate handler
    if api_path == '/get-order':
        return get_order(params['order_id'])
    elif api_path == '/refund':
        return process_refund(params['order_id'], params['reason'])
    else:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'Unknown action'})
        }

def get_order(order_id):
    """Fetch order details from database"""
    # In real scenario, query database
    return {
        'statusCode': 200,
        'body': json.dumps({
            'order_id': order_id,
            'status': 'shipped',
            'total': 99.99,
            'items': ['Item 1', 'Item 2']
        })
    }

def process_refund(order_id, reason):
    """Process refund for an order"""
    # In real scenario, update database and payment system
    return {
        'statusCode': 200,
        'body': json.dumps({
            'success': True,
            'refund_id': f'REF-{order_id}',
            'amount': 99.99,
            'message': f'Refund processed for reason: {reason}'
        })
    }

Invoking an Agent

# Invoke the agent
response = client.invoke_agent(
    agentId=agent_id,
    agentAliasId='LFTSTCQMTR',  # Use DRAFT for testing
    sessionId='session-123',
    inputText='I want to return my order #12345'
)

# Process streaming response
for event in response['completion']:
    if 'chunk' in event:
        chunk = event['chunk']
        if 'bytes' in chunk:
            print(chunk['bytes'].decode('utf-8'), end='', flush=True)

Agent Orchestration

# Multi-step agent workflow
def orchestrate_support_request(customer_id, issue_type):
    """Orchestrate a complex support request"""
    
    agent_client = boto3.client('bedrock-agent-runtime', region_name='us-east-1')
    
    # Step 1: Retrieve customer info
    response = agent_client.invoke_agent(
        agentId=agent_id,
        agentAliasId='LFTSTCQMTR',
        sessionId=f'session-{customer_id}',
        inputText=f'Get customer {customer_id} details'
    )
    
    # Step 2: Analyze issue
    response = agent_client.invoke_agent(
        agentId=agent_id,
        agentAliasId='LFTSTCQMTR',
        sessionId=f'session-{customer_id}',
        inputText=f'Analyze this {issue_type} issue and suggest solutions'
    )
    
    # Step 3: Take action
    response = agent_client.invoke_agent(
        agentId=agent_id,
        agentAliasId='LFTSTCQMTR',
        sessionId=f'session-{customer_id}',
        inputText='Apply the best solution'
    )
    
    return response

Agent with Knowledge Base

# Create agent with knowledge base integration
response = client.create_agent(
    agentName='kb-agent',
    agentRoleArn='arn:aws:iam::ACCOUNT:role/BedrockAgentRole',
    foundationModel='anthropic.claude-3-sonnet-20240229-v1:0',
    instruction='Use the knowledge base to answer questions accurately.'
)

agent_id = response['agent']['id']

# Add knowledge base as action group
response = client.create_agent_action_group(
    agentId=agent_id,
    agentVersion='DRAFT',
    actionGroupName='knowledge-base',
    description='Search company knowledge base',
    actionGroupExecutor={
        'knowledgeBaseLambda': 'arn:aws:lambda:us-east-1:ACCOUNT:function:kb-handler'
    }
)

Best Practices

# ✅ Good: Clear action descriptions
action_group = {
    'actionGroupName': 'order-management',
    'description': 'Manage customer orders - retrieve, modify, or cancel',
    'actions': [
        {
            'name': 'get_order_status',
            'description': 'Get the current status of an order by order ID'
        },
        {
            'name': 'cancel_order',
            'description': 'Cancel an order if it hasn\'t shipped yet'
        }
    ]
}

# ✅ Good: Proper error handling in Lambda
def safe_action_handler(event, context):
    try:
        # Process action
        result = perform_action(event)
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

# ✅ Good: Session management for multi-turn conversations
session_id = f'user-{user_id}-{timestamp}'
response = client.invoke_agent(
    agentId=agent_id,
    agentAliasId='LFTSTCQMTR',
    sessionId=session_id,  # Maintains context across turns
    inputText=user_input
)