### Install AWS SDKs and Libraries Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Installs or upgrades essential Python packages for interacting with AWS services, including botocore, boto3, AWS CLI, and opensearch-py. These are crucial for managing AWS resources and Bedrock agents. ```python !python3 -m pip install --upgrade -q botocore !python3 -m pip install --upgrade -q boto3 !python3 -m pip install --upgrade -q awscli !pip install --upgrade -q opensearch-py ``` -------------------------------- ### Generate Park Reservations Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates 20 random park reservations, linking citizens to parks with random start and end dates. Inserts data into the 'park_reservation' table. Requires 'random', 'date', and 'timedelta' from the 'datetime' module, and a database cursor 'c'. ```python # Assume 'c' is a database cursor object # Assume 'random', 'date', 'timedelta' are imported num_parks = 3 for _ in range(20): citizen_id = random.randint(1, 50) park_id = random.randint(1, num_parks) start_date = date.today() + timedelta(days=random.randint(1, 30)) end_date = start_date + timedelta(days=random.randint(1, 3)) c.execute("INSERT INTO park_reservation (citizen_id, park_id, reservation_start_date, reservation_end_date) VALUES (?, ?, ?, ?)", (citizen_id, park_id, start_date.isoformat(), end_date.isoformat())) ``` -------------------------------- ### Create Boto3 Clients for AWS Services Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Initializes Boto3 clients for various AWS services required for the agent setup, including IAM, S3, OpenSearch Serverless, Lambda, and Bedrock Agent services. These clients enable programmatic interaction with AWS. ```python # getting boto3 clients for required AWS services sts_client = boto3.client('sts')iam_client = boto3.client('iam') s3_client = boto3.client('s3') open_search_serverless_client = boto3.client('opensearchserverless') lambda_client = boto3.client('lambda') bedrock_agent_client = boto3.client('bedrock-agent') bedrock_agent_runtime_client = boto3.client('bedrock-agent-runtime') bedrock_client = boto3.client('bedrock') ``` -------------------------------- ### Start Knowledge Base Ingestion Job Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Initiates an ingestion job for a knowledge base data source. This process fetches documents, chunks them, creates embeddings, and stores them in the vector database. ```python # Start an ingestion job data_source_id = data_source_response["dataSource"]["dataSourceId"] start_job_response = bedrock_agent_client.start_ingestion_job( knowledgeBaseId=knowledge_base_id, dataSourceId=data_source_id ) ``` -------------------------------- ### Get Garbage Pickup Day Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Retrieves the garbage pickup day for a specific district by querying a SQLite database. Returns the pickup day or a message if no information is found. ```python def get_garbage_pickup_day(district_id): conn = sqlite3.connect('/tmp/anycity_database.db') c = conn.cursor() try: c.execute("SELECT pickup_day FROM garbage WHERE district_id = ?", (district_id,)) result = c.fetchone() if result: pickup_day = result[0] conn.close() return pickup_day else: conn.close() return f"No garbage pickup information found for district ID {district_id}." except Exception as e: conn.close() raise Exception(f"Error occurred: {e}") ``` -------------------------------- ### Invoke Bedrock Agent with Guardrails Setup Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Illustrates invoking the Bedrock agent for a park reservation request, including setting up a session ID, enabling trace, and defining session attributes. This snippet shows a more complex interaction potentially involving guardrails or specific session state management. ```python ## create a random id for session initiator id session_id:str = str(uuid.uuid1()) enable_trace:bool = True end_session:bool = False # invoke the agent API agentResponse = bedrock_agent_runtime_client.invoke_agent( inputText="I'd like the reserve the park for an immigration party on April 19 2025 for citizen 12", agentId=agent_id, agentAliasId=alias_id, sessionId=session_id, enableTrace=enable_trace, endSession= end_session ) logger.info(pprint.pprint(agentResponse)) ``` ```python %%time event_stream = agentResponse['completion'] try: for event in event_stream: if 'chunk' in event: data = event['chunk']['bytes'] logger.info(f"Final answer ->\n{data.decode('utf8')}") agent_answer = data.decode('utf8') end_event_received = True # End event indicates that the request finished successfully elif 'trace' in event: logger.info(json.dumps(event['trace'], indent=2)) else: raise Exception("unexpected event.", event) except Exception as e: raise Exception("unexpected event.", e) ``` -------------------------------- ### Get Available Park Days Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Calculates available park days within a specified date range by querying park reservations from a SQLite database. It returns a list of dates that are not reserved. ```python import os import json import shutil import sqlite3 from datetime import datetime, timedelta def get_available_park_days(park_id, start_date, end_date): conn = sqlite3.connect('/tmp/anycity_database.db') c = conn.cursor() try: start_date = datetime.strptime(start_date, '%Y-%m-%d') end_date = datetime.strptime(end_date, '%Y-%m-%d') c.execute(""" SELECT reservation_start_date, reservation_end_date FROM park_reservation WHERE park_id = ? AND (reservation_start_date <= ? AND reservation_end_date >= ?) """, (park_id, end_date.strftime('%Y-%m-%d'), start_date.strftime('%Y-%m-%d'))) reservations = c.fetchall() all_days = set(start_date + timedelta(days=x) for x in range((end_date - start_date).days + 1)) for reservation in reservations: res_start = datetime.strptime(reservation[0], '%Y-%m-%d') res_end = datetime.strptime(reservation[1], '%Y-%m-%d') reserved_days = set(res_start + timedelta(days=x) for x in range((res_end - res_start).days + 1)) all_days -= reserved_days available_days = sorted(list(all_days)) conn.close() return [day.strftime('%Y-%m-%d') for day in available_days] except Exception as e: conn.close() raise Exception(f"Error occurred: {e}") ``` -------------------------------- ### Wait for OpenSearch Collection Creation and Get Endpoint Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Polls the status of the OpenSearch collection creation until it is successfully created. Once created, it extracts the collection endpoint, which is needed to connect to the OpenSearch cluster. ```python # wait for collection creation response = open_search_serverless_client.batch_get_collection(names=[kb_collection_name]) # Periodically check collection status while (response['collectionDetails'][0]['status']) == 'CREATING': print('Creating collection...') time.sleep(30) response = open_search_serverless_client.batch_get_collection(names=[kb_collection_name]) print('\nCollection successfully created:') print(response["collectionDetails"]) # Extract the collection endpoint from the response host = (response['collectionDetails'][0]['collectionEndpoint']) final_host = host.replace("https://", "") final_host ``` -------------------------------- ### Bedrock Agent Client API Reference Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Provides details on key Bedrock Agent API operations for creating and managing agents and their components. Includes parameters, return values, and usage context. ```APIDOC BedrockAgentClient: create_role(RoleName: str, AssumeRolePolicyDocument: str) -> dict - Creates an IAM role with a specified trust policy. - Parameters: - RoleName: The name for the new IAM role. - AssumeRolePolicyDocument: A JSON-formatted string representing the trust policy. - Returns: A dictionary containing information about the created role, including 'Role.Arn'. attach_role_policy(RoleName: str, PolicyArn: str) -> dict - Attaches an IAM policy to the specified IAM role. - Parameters: - RoleName: The name of the role to attach the policy to. - PolicyArn: The Amazon Resource Name (ARN) of the policy to attach. - Returns: An empty dictionary on success. create_agent( agentName: str, agentResourceRoleArn: str, foundationModel: str, instruction: str, description: str = None, idleSessionTTLInSeconds: int = None, clientToken: str = None, tags: dict = None ) -> dict - Creates a new Amazon Bedrock agent. - Parameters: - agentName: The name of the agent. - agentResourceRoleArn: The ARN of the IAM role that grants Amazon Bedrock access to resources. - foundationModel: The foundation model to use for the agent. - instruction: Instructions for the agent. - description: A description of the agent (optional). - idleSessionTTLInSeconds: The number of seconds that Amazon Bedrock keeps a session open when there are no new user inputs (optional). - clientToken: A unique, case-sensitive identifier that you provide to ensure the idempotency of the request (optional). - tags: A map of tag keys and values (optional). - Returns: A dictionary containing agent details, including 'agent.agentId' and 'agent.agentArn'. create_agent_action_group( agentId: str, agentVersion: str, actionGroupName: str, functionSchema: dict = None, apiSchema: dict = None, description: str = None, clientToken: str = None, idleSessionTTLInSeconds: int = None, tags: dict = None ) -> dict - Creates an action group for an agent. - Parameters: - agentId: The unique identifier of the agent. - agentVersion: The version of the agent (e.g., 'DRAFT'). - actionGroupName: The name of the action group. - functionSchema: A schema defining functions the agent can invoke (optional). - Contains 'name' (str) and 'functions' (list of function definitions). - Each function definition includes 'name' (str), 'description' (str), and 'parameters' (dict). - apiSchema: An OpenAPI schema defining APIs the agent can invoke (optional). - description: A description of the action group (optional). - clientToken: A unique identifier for idempotency (optional). - idleSessionTTLInSeconds: Session timeout for this action group (optional). - tags: Tags for the action group (optional). - Returns: A dictionary containing action group details. ``` -------------------------------- ### Get AWS Caller Identity Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Retrieves the identity of the caller, typically used to get the current IAM role ARN. This information is often used for policy configurations to grant necessary permissions. ```python response = sts_client.get_caller_identity() current_role = response['Arn'] current_role ``` -------------------------------- ### Call Helper Function with Various Queries Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Demonstrates using the `simple_agent_invoke` helper function to query the Bedrock agent with different natural language inputs, including a query about trash pickup, contact information, and park availability with trace enabled. ```python simple_agent_invoke("What day does trash get picked up for district 3?", agent_id, agent_alias_id, session_id) ``` ```python simple_agent_invoke("Who can I contact about a large trash pickup?", agent_id, agent_alias_id, session_id) ``` ```python simple_agent_invoke("Is park 2 available on Feb 31st?", agent_id, agent_alias_id, session_id, enable_trace=True) ``` -------------------------------- ### Prepare Bedrock Agent Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb This Python snippet prepares an Amazon Bedrock agent for use. The `prepare_agent` operation ensures the agent is ready to handle requests, often involving backend provisioning or configuration updates. It prints the response from the preparation call. ```python response = bedrock_agent_client.prepare_agent( agentId=agent_id ) print(response) # Pause to make sure agent is prepared time.sleep(30) ``` -------------------------------- ### Define Agent and Lambda Configuration Variables Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Sets up configuration variables for the Bedrock Agent and the associated Lambda function, including names, policies, foundation models, and descriptions. These variables customize the agent's behavior and resource naming. ```python session = boto3.session.Session() region = session.region_name account_id = sts_client.get_caller_identity()["Account"] region, account_id # configuration variables suffix = f"{region}-{account_id}" agent_name = "city-assistant-function-def" agent_bedrock_allow_policy_name = f"{agent_name}-ba-{suffix}" agent_role_name = f'AmazonBedrockExecutionRoleForAgents_{agent_name}' agent_foundation_model = "anthropic.claude-3-sonnet-20240229-v1:0" agent_description = "Agent for providing citizen assistance at AnyCity" agent_instruction = "You are an helpful agent, helping citizens at AnyCity to create park reservation, and check on garbage pickup dates. You also have access to documentation about the city in your knowledge base. Use the documentation to help answer questions." agent_action_group_name = "AnyCityActionGroup" agent_action_group_description = "Actions for getting the park reservations, creating park reservations, and checking garbage pick up times" agent_alias_name = f"{agent_name}-alias" lambda_function_role = f'{agent_name}-lambda-role-{suffix}' lambda_function_name = f'{agent_name}-{suffix}' ``` -------------------------------- ### Create Data Source for Knowledge Base Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Configures and creates a data source for an existing knowledge base. This snippet defines S3 bucket details, inclusion prefixes, and chunking strategies for processing documents before ingestion. ```python data_source_name = f'bedrock-docs-kb-docs-{suffix}' # Define the S3 configuration for your data source s3_configuration = { 'bucketArn': f"arn:aws:s3:::{bucket_name}", 'inclusionPrefixes': [kb_key] } # Define the data source configuration data_source_configuration = { 's3Configuration': s3_configuration, 'type': 'S3' } knowledge_base_id = kb_obj["knowledgeBase"]["knowledgeBaseId"] knowledge_base_arn = kb_obj["knowledgeBase"]["knowledgeBaseArn"] chunking_strategy_configuration = { "chunkingStrategy": "FIXED_SIZE", "fixedSizeChunkingConfiguration": { "maxTokens": 512, "overlapPercentage": 20 } } # Create the data source try: # ensure that the KB is created and available time.sleep(45) data_source_response = bedrock_agent_client.create_data_source( knowledgeBaseId=knowledge_base_id, name=data_source_name, description='DataSource for the bedrock documentation', dataSourceConfiguration=data_source_configuration, vectorIngestionConfiguration = { "chunkingConfiguration": chunking_strategy_configuration } ) # Pretty print the response pprint.pprint(data_source_response) except Exception as e: print(f"Error occurred: {e}") ``` -------------------------------- ### Package and Create Lambda Function Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Packages the Lambda function code and any necessary files (like a database) into a zip archive. It then uses the AWS SDK (boto3) to create the Lambda function, specifying the runtime, timeout, IAM role, and handler. ```python # Package up the lambda function code s = BytesIO() z = zipfile.ZipFile(s, 'w') z.write("lambda_function.py") z.write("anycity_database.db") z.close() zip_content = s.getvalue() # Create Lambda Function lambda_function = lambda_client.create_function( FunctionName=lambda_function_name, Runtime='python3.12', Timeout=180, Role=lambda_iam_role['Role']['Arn'], Code={'ZipFile': zip_content}, Handler='lambda_function.lambda_handler' ) ``` -------------------------------- ### Extract Agent Alias ID Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Extracts the agent alias ID from a response object. This is typically used after an initial agent creation or retrieval operation to get the identifier needed for subsequent agent invocations. ```python agent_alias_id = agent_alias['agentAlias']['agentAliasId'] ``` -------------------------------- ### Build and Deploy Application Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/README.md Builds and deploys the application to AWS using the Serverless Application Model Command Line Interface (SAM CLI). The command prompts for stack name, AWS region, confirmation before deploy, and whether to allow SAM CLI IAM role creation. It also asks to save arguments to `samconfig.toml` for future deployments. The `CAPABILITY_IAM` capability is required if IAM roles are created. ```bash make deploy ``` -------------------------------- ### Prepare Agent for Testing Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates a DRAFT version of the agent, making it ready for internal testing and validation. This step is necessary before creating an alias or deploying the agent. ```python response = bedrock_agent_client.prepare_agent( agentId=agent_id ) print(response) ``` -------------------------------- ### Invoke Bedrock Agent for Park Reservation Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Shows how to invoke the Bedrock agent to make a park reservation, specifying dates and details. The code includes the agent invocation and subsequent processing of the streaming response to retrieve the final answer. ```python agentResponse = bedrock_agent_runtime_client.invoke_agent( inputText="now let me reserve park 1 from May 15 2025 to Jun 15 2025.", agentId=agent_id, agentAliasId=agent_alias_id, sessionId=session_id, enableTrace=enable_trace, endSession= end_session, sessionState={ "sessionAttributes": { "firstName": "Aaron", "citizenID": "2", "districtID": "3" } } ) logger.info(pprint.pprint(agentResponse)) ``` ```python %%time event_stream = agentResponse['completion'] try: for event in event_stream: if 'chunk' in event: data = event['chunk']['bytes'] logger.info(f"Final answer ->\n{data.decode('utf8')}") agent_answer = data.decode('utf8') end_event_received = True # End event indicates that the request finished successfully elif 'trace' in event: logger.info(json.dumps(event['trace'], indent=2)) else: raise Exception("unexpected event.", event) except Exception as e: raise Exception("unexpected event.", e) ``` -------------------------------- ### Generate Garbage Pickup Schedules Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Assigns a random pickup day for each district (1-5) and inserts the schedule into the 'garbage' table. Uses a database cursor 'c'. ```python # Assume 'c' is a database cursor object days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday'] for district in range(1, 6): pickup_day = random.choice(days) c.execute("INSERT INTO garbage (district_id, pickup_day) VALUES (?, ?)", (district, pickup_day)) ``` -------------------------------- ### Handle Bedrock Agent Function Calls and Responses Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Processes incoming function calls for a Bedrock agent, extracts parameters, and constructs appropriate response bodies. It handles different functions like 'get_available_park_days', 'book_park', and 'get_garbage_pickup_day', formatting the output for the agent. ```python available_days = get_available_park_days(park_id, start_date, end_date) responseBody = { 'TEXT': { "body": f"Available days for park ID {park_id} from {start_date} to {end_date}: {available_days}" } } elif function == 'book_park': #citizen_id = None park_id = None start_date = None end_date = None for param in parameters: # if param["name"] == "citizen_id": # citizen_id = param["value"] if param["name"] == "park_id": park_id = param["value"] if param["name"] == "start_date": start_date = param["value"] if param["name"] == "end_date": end_date = param["value"] if not all([citizen_id, park_id, start_date, end_date]): raise Exception("Missing mandatory parameters: citizen_id, park_id, start_date, end_date") completion_message = book_park(citizen_id, park_id, start_date, end_date) responseBody = { 'TEXT': { "body": completion_message } } elif function == 'get_garbage_pickup_day': # district_id = None # for param in parameters: # if param["name"] == "district_id": # district_id = param["value"] if not district_id: raise Exception("Missing mandatory parameter: district_id") pickup_day = get_garbage_pickup_day(district_id) responseBody = { 'TEXT': { "body": f"Garbage pickup day for district ID {district_id}: {pickup_day}" } } action_response = { 'actionGroup': actionGroup, 'function': function, 'functionResponse': { 'responseBody': responseBody } } function_response = {'response': action_response, 'messageVersion': event['messageVersion']} print("Response: {}".format(function_response)) return function_response ``` -------------------------------- ### Create Knowledge Base Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates a new Amazon Bedrock knowledge base. It specifies the name, description, IAM role, and configuration for vector storage, including the embedding model ARN and OpenSearch Serverless collection details. ```python kb_name = f'bedrockdocs-kb-{region}-{account_id}' # Creating the knowledge base try: # ensure the index is created and available time.sleep(45) kb_obj = bedrock_agent_client.create_knowledge_base( name=kb_name, description='KB that contains the bedrock documentation', roleArn=kb_role_arn, knowledgeBaseConfiguration={ 'type': 'VECTOR', # Corrected type 'vectorKnowledgeBaseConfiguration': { 'embeddingModelArn': embedding_model_arn } }, storageConfiguration=storageConfiguration ) # Pretty print the response pprint.pprint(kb_obj) except Exception as e: print(f"Error occurred: {e}") ``` -------------------------------- ### Import Required Python Libraries Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Imports necessary Python libraries for AWS SDK operations, data handling, logging, and utility functions. These libraries are fundamental for building the agent and its associated AWS resources. ```python import boto3 import json import time import zipfile from io import BytesIO import uuid import pprint import logging import time import os from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth ``` -------------------------------- ### Create Bedrock Guardrail Configuration Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Demonstrates how to create a comprehensive Amazon Bedrock guardrail. This includes defining topic policies, content filters for various categories (sexual, violence, hate, insults, misconduct, prompt attacks), word policies, managed word lists, sensitive information blocking/anonymization using PII entities and regex, and contextual grounding policies. It also sets custom messages for blocked inputs and outputs. ```python response = bedrock_client.create_guardrail( name='AnyCityGuardrail', description='Guardrail for city assitant to help users with questions about AnyCity', topicPolicyConfig={ 'topicsConfig': [ { 'name': 'Politics', 'definition': 'Guardrail to prevent from engaging in political discussions, expressing political opinions, or taking partisan stances on contested issues.', 'examples': [ 'Who do you think should win the upcoming election?', 'Can you explain why candidate X would be better than candidate Y?', 'Do you think current gun regulations are too strict or too lenient?', 'Help me write a convincing argument on the Freedom Party platform and their key policies.', 'Who is tougher on crime? Republicans or Democrats?', ], 'type': 'DENY' }, ] }, contentPolicyConfig={ 'filtersConfig': [ { 'type': 'SEXUAL', 'inputStrength': 'HIGH', 'outputStrength': 'HIGH' }, { 'type': 'VIOLENCE', 'inputStrength': 'HIGH', 'outputStrength': 'HIGH' }, { 'type': 'HATE', 'inputStrength': 'HIGH', 'outputStrength': 'HIGH' }, { 'type': 'INSULTS', 'inputStrength': 'HIGH', 'outputStrength': 'HIGH' }, { 'type': 'MISCONDUCT', 'inputStrength': 'HIGH', 'outputStrength': 'HIGH' }, { 'type': 'PROMPT_ATTACK', 'inputStrength': 'HIGH', 'outputStrength': 'NONE' } ] }, wordPolicyConfig={ 'wordsConfig': [ {'text': 'republican'}, {'text': 'democrat'}, {'text': 'conservative'}, {'text': 'liberal'}, {'text': 'immigration'}, {'text': 'abortion'}, {'text': 'bipartisan'}, ], 'managedWordListsConfig': [ {'type': 'PROFANITY'} ] }, sensitiveInformationPolicyConfig={ 'piiEntitiesConfig': [ {'type': 'US_SOCIAL_SECURITY_NUMBER', 'action': 'BLOCK'}, {'type': 'US_BANK_ACCOUNT_NUMBER', 'action': 'BLOCK'}, {'type': 'CREDIT_DEBIT_CARD_NUMBER', 'action': 'BLOCK'} ], 'regexesConfig': [ { 'name': 'Account Number', 'description': 'Matches account numbers in the format XXXXXX1234', 'pattern': r'\b\d{6}\d{4}\b', 'action': 'ANONYMIZE' } ] }, contextualGroundingPolicyConfig={ 'filtersConfig': [ { 'type': 'GROUNDING', 'threshold': 0.7 }, { 'type': 'RELEVANCE', 'threshold': 0.7 } ] }, blockedInputMessaging='Sorry, your question violates our usage policies. Please contact the AnyCity City Administrator at 987-654-3210 if this message was generated in error.', blockedOutputsMessaging='Sorry, I am unable to reply. Please contact the AnyCity City Administrator at 987-654-3210', ) ``` -------------------------------- ### Create SQLite Database Schema for Agent Logic Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Defines the schema for an SQLite database to manage employee data, garbage pickup schedules, and park reservations. This script creates tables for citizens, garbage collection days, and park reservations, preparing the data storage for the Lambda function. ```python import sqlite3 import random from datetime import date, timedelta # Connect to the SQLite database (creates a new one if it doesn't exist) conn = sqlite3.connect('anycity_database.db') c = conn.cursor() # Create the citizen table c.execute('''CREATE TABLE IF NOT EXISTS citizen (citizen_id INTEGER PRIMARY KEY AUTOINCREMENT, citizen_name TEXT, citizen_address TEXT, district INTEGER)''') # Create the garbage table c.execute('''CREATE TABLE IF NOT EXISTS garbage (district_id INTEGER PRIMARY KEY, pickup_day TEXT)''') # Create the park table c.execute('''CREATE TABLE IF NOT EXISTS park_reservation (reservation_id INTEGER PRIMARY KEY AUTOINCREMENT, citizen_id INTEGER, park_id INTEGER, reservation_start_date TEXT, reservation_end_date TEXT, FOREIGN KEY(citizen_id) REFERENCES citizen(citizen_id))''') # Commit changes and close the connection (optional, depending on context) # conn.commit() # conn.close() ``` -------------------------------- ### Invoke Bedrock Agent for Trash Pickup Query Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Demonstrates invoking the Bedrock agent with a specific question about trash pickup. It includes the necessary parameters like agentId, agentAliasId, and sessionId, and logs the response. The code also processes the streaming completion events to extract the agent's textual answer. ```python agentResponse = bedrock_agent_runtime_client.invoke_agent( inputText="What day is trash pickup?", agentId=agent_id, agentAliasId=agent_alias_id, sessionId=session_id, enableTrace=enable_trace, endSession= end_session, sessionState={ "sessionAttributes": { "firstName": "Aaron", "citizenID": "2", "districtID": "3" } } ) logger.info(pprint.pprint(agentResponse)) ``` ```python %%time event_stream = agentResponse['completion'] try: for event in event_stream: if 'chunk' in event: data = event['chunk']['bytes'] logger.info(f"Final answer ->\n{data.decode('utf8')}") agent_answer = data.decode('utf8') end_event_received = True # End event indicates that the request finished successfully elif 'trace' in event: logger.info(json.dumps(event['trace'], indent=2)) else: raise Exception("unexpected event.", event) except Exception as e: raise Exception("unexpected event.", e) ``` -------------------------------- ### Configure Python Logger Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Sets up a basic logger for the Python script, configuring the format and logging level. This helps in monitoring the execution flow and debugging potential issues during agent creation. ```python # setting logger logging.basicConfig(format='[%(asctime)s] p%(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s', level=logging.INFO) logger = logging.getLogger(__name__) ``` -------------------------------- ### Define Bedrock Agent Action Group Functions Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Defines a list of functions that the Bedrock agent can call. Each function includes its name, a description of its purpose, and a detailed schema for its parameters. ```json [ { 'name': 'get_available_park_days', 'description': 'Get the available days for a specific park within a date range. If no end data is specified assume the end_date is the same as the start_date', 'parameters': { "park_id": { "description": "The ID of the park to check availability", "required": True, "type": "integer" }, "start_date": { "description": "The start date of the date range to check (format: YYYY-MM-DD)", "required": True, "type": "string" }, "end_date": { "description": "The end date of the date range to check (format: YYYY-MM-DD)", "required": True, "type": "string" } } }, { 'name': 'book_park', 'description': 'Book a park for a specific citizen within a date range. If no end data is specified assume the end_date is the same as the start_date', 'parameters': { "park_id": { "description": "The ID of the park to be reserved", "required": True, "type": "integer" }, "start_date": { "description": "The start date of the reservation (format: YYYY-MM-DD)", "required": True, "type": "string" }, "end_date": { "description": "The end date of the reservation (format: YYYY-MM-DD)", "required": True, "type": "string" } } }, { 'name': 'get_garbage_pickup_day', 'description': 'Get the garbage pickup day', 'parameters': {} } ] ``` -------------------------------- ### Book Park Reservation Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Books a park reservation for a given citizen and park, checking for availability using `get_available_park_days`. It inserts the reservation into the SQLite database. ```python def book_park(citizen_id, park_id, start_date, end_date): conn = sqlite3.connect('/tmp/anycity_database.db') c = conn.cursor() try: c.execute("SELECT * FROM citizen WHERE citizen_id = ?", (citizen_id,)) if not c.fetchone(): conn.close() return f"Citizen with ID {citizen_id} does not exist." available_days = get_available_park_days(park_id, start_date, end_date) required_days = set(datetime.strptime(start_date, '%Y-%m-%d') + timedelta(days=x) for x in range((datetime.strptime(end_date, '%Y-%m-%d') - datetime.strptime(start_date, '%Y-%m-%d')).days + 1)) if not all(day.strftime('%Y-%m-%d') in available_days for day in required_days): conn.close() return "The selected dates are not fully available for reservation." c.execute(""" INSERT INTO park_reservation (citizen_id, park_id, reservation_start_date, reservation_end_date) VALUES (?, ?, ?, ?) """, (citizen_id, park_id, start_date, end_date)) conn.commit() conn.close() return f"Park reservation successful for citizen ID {citizen_id} from {start_date} to {end_date}." except Exception as e: conn.rollback() conn.close() raise Exception(f"Error occurred: {e}") ``` -------------------------------- ### Upload Knowledge Base Files to S3 Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Uploads local files (e.g., PDFs) from a specified directory to an S3 bucket. These files are intended to be used as data for an Amazon Bedrock Knowledge Base, organized under a specific key prefix. ```python kb_files_path = 'data' kb_key = 'kb_documents' # Upload Knowledge Base files to this s3 bucket for f in os.listdir(kb_files_path): if f.endswith(".pdf"): s3_client.upload_file(kb_files_path+'/'+f, bucket_name, kb_key+'/'+f) ``` -------------------------------- ### Build OpenSearch Client with AWS Authentication Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Configures and builds an OpenSearch client using AWS Signature Version 4 for authentication. It connects to the OpenSearch collection endpoint and sets up necessary parameters for interaction. ```python # Set up AWS authentication service = 'aoss' credentials = boto3.Session().get_credentials() awsauth = AWSV4SignerAuth(credentials, region, service) # Build the OpenSearch client oss_client = OpenSearch( hosts=[{'host': final_host, 'port': 443}], http_auth=awsauth, use_ssl=True, verify_certs=True, connection_class=RequestsHttpConnection, timeout=300 ) ``` -------------------------------- ### Create S3 Bucket for Knowledge Base Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates an S3 bucket to store knowledge base documents. This bucket will serve as the storage location for files that will be indexed for retrieval by Amazon Bedrock. ```python # Create Knowledge Base bucket_name = 'reinvent2024-anycity-kb' s3_client.create_bucket(Bucket=bucket_name) ``` -------------------------------- ### Define IAM Policy for OpenSearch Serverless Access Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb This snippet defines the name for an IAM policy that will grant access to OpenSearch Serverless. The actual policy document and creation using IAM client are not shown but implied by the variable name. ```python kb_aoss_allow_policy_name = f"bd-kb-aoss-allow-{suffix}" ``` -------------------------------- ### Create OpenSearch Data Access Policy Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates a data access policy for an OpenSearch Serverless collection. This policy defines permissions for accessing collection items and index data, specifying principals and allowed actions. ```python data_policy_json = [ { "Rules": [ { "Resource": [ f"collection/{kb_collection_name}" ], "Permission": [ "aoss:DescribeCollectionItems", "aoss:CreateCollectionItems", "aoss:UpdateCollectionItems", "aoss:DeleteCollectionItems" ], "ResourceType": "collection" }, { "Resource": [ f"index/{kb_collection_name}/*" ], "Permission": [ "aoss:CreateIndex", "aoss:DeleteIndex", "aoss:UpdateIndex", "aoss:DescribeIndex", "aoss:ReadDocument", "aoss:WriteDocument" ], "ResourceType": "index" } ], "Principal": [ kb_role_arn, f"arn:aws:sts::{account_id}:assumed-role/Admin/*", current_role ], "Description": "" } ] data_policy = open_search_serverless_client.create_access_policy( description='data access policy for aoss collection', name=kb_collection_name, policy=json.dumps(data_policy_json), type='data' ) ``` -------------------------------- ### AWS Lambda Handler Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb The main entry point for the AWS Lambda function. It handles incoming events, copies the SQLite database to the /tmp directory if needed, and dispatches calls to specific functions based on the event's 'function' field. ```python def lambda_handler(event, context): original_db_file = 'anycity_database.db' target_db_file = '/tmp/anycity_database.db' if not os.path.exists(target_db_file): shutil.copy2(original_db_file, target_db_file) # Retrieve agent session attributes for context session_attributes = event.get('sessionAttributes', {}) first_name = session_attributes.get('firstName', '') citizen_id = session_attributes.get('citizenID', '2') district_id = session_attributes.get('districtID', '4') agent = event['agent'] actionGroup = event['actionGroup'] function = event['function'] parameters = event.get('parameters', []) responseBody = { "TEXT": { "body": "Error, no function was called" } } if function == 'get_available_park_days': park_id = None start_date = None end_date = None for param in parameters: if param["name"] == "park_id": park_id = param["value"] if param["name"] == "start_date": start_date = param["value"] if param["name"] == "end_date": end_date = param["value"] if not all([park_id, start_date, end_date]): raise Exception("Missing mandatory parameters: park_id, start_date, end_date") ``` -------------------------------- ### Delete Bedrock Knowledge Base using Boto3 Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Demonstrates deleting an AWS Bedrock Knowledge Base using the `delete_knowledge_base` method from the Boto3 SDK. It includes error handling and a call to a hypothetical `wait_for_deletion` function. ```python try: bedrock_agent_client.delete_knowledge_base(knowledgeBaseId=knowledge_base_id) print(f"Deleted Knowledge Base: {knowledge_base_id}") wait_for_deletion('knowledge_base', knowledge_base_id) except Exception as e: print(f"Error deleting Knowledge Base: {e}") ``` -------------------------------- ### Create IAM Policy for Knowledge Base Retrieval Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates an IAM policy that allows an agent to retrieve information from a specific Amazon Bedrock knowledge base. This policy enables the agent to access and use the indexed data. ```python bedrock_agent_kb_allow_policy_name = f"bda-kb-allow-{suffix}" bedrock_agent_kb_retrival_policy_statement = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "bedrock:Retrieve" ], "Resource": [ knowledge_base_arn ] } ] } bedrock_agent_kb_json = json.dumps(bedrock_agent_kb_retrival_policy_statement) agent_kb_schema_policy = iam_client.create_policy( PolicyName=bedrock_agent_kb_allow_policy_name, Description=f"Policy to allow agent to retrieve documents from knowledge base.", PolicyDocument=bedrock_agent_kb_json ) ``` -------------------------------- ### Create Bedrock Agent Action Group Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates an action group for the Bedrock agent, linking it to a Lambda function and defining its capabilities via a function schema. A delay is included to ensure the agent is ready. ```python # Assume bedrock_agent_client, agent_id, agent_functions, and lambda_function_arn are defined elsewhere # Pause to make sure agent is created time.sleep(30) response = bedrock_agent_client.create_agent_action_group( agentId=agent_id, agentVersion='DRAFT', actionGroupName='ParkBookingActions', functionSchema={'name': 'ExampleSchema', 'functions': agent_functions}, # Alternatively, use apiSchema for OpenAPI specifications # apiSchema={'OpenApiSchema': '...'}, description='Actions for booking park visits and checking garbage pickup days', # idleSessionTTLInSeconds=1800, # Can also be set here # clientToken='...' # Optional client token for idempotency ) response ``` -------------------------------- ### Create IAM Role for Lambda Function Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Sets up an IAM role required for a Lambda function to execute. It defines an assume role policy document allowing the Lambda service to assume the role and attaches the AWSLambdaBasicExecutionRole policy for basic logging and execution permissions. ```python # Create IAM Role for the Lambda function try: assume_role_policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "lambda.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } assume_role_policy_document_json = json.dumps(assume_role_policy_document) lambda_iam_role = iam_client.create_role( RoleName=lambda_function_role, AssumeRolePolicyDocument=assume_role_policy_document_json ) # Pause to make sure role is created time.sleep(10) except: lambda_iam_role = iam_client.get_role(RoleName=lambda_function_role) iam_client.attach_role_policy( RoleName=lambda_function_role, PolicyArn='arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole' ) ``` -------------------------------- ### Create IAM Role for Bedrock Agent Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Defines and creates an IAM role that Bedrock can assume, attaching policies required for agent functionality. Includes a delay to ensure the role is provisioned before proceeding. ```python import json import time # Assume agent_role_name, iam_client, agent_bedrock_policy, and agent_kb_schema_policy are defined elsewhere assume_role_policy_document = { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "Service": "bedrock.amazonaws.com" }, "Action": "sts:AssumeRole" }] } assume_role_policy_document_json = json.dumps(assume_role_policy_document) agent_role = iam_client.create_role( RoleName=agent_role_name, AssumeRolePolicyDocument=assume_role_policy_document_json ) # Pause to make sure role is created time.sleep(10) iam_client.attach_role_policy( RoleName=agent_role_name, PolicyArn=agent_bedrock_policy['Policy']['Arn'] ) iam_client.attach_role_policy( RoleName=agent_role_name, PolicyArn=agent_kb_schema_policy['Policy']['Arn'] ) ``` -------------------------------- ### Helper Function to Invoke Bedrock Agent Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Defines a reusable Python function `simple_agent_invoke` that encapsulates the logic for calling the Bedrock agent's `invoke_agent` method and processing its streaming response. This function simplifies making multiple agent calls. ```python def simple_agent_invoke(input_text, agent_id, agent_alias_id, session_id=None, enable_trace=False, end_session=False): agentResponse = bedrock_agent_runtime_client.invoke_agent( inputText=input_text, agentId=agent_id, agentAliasId=agent_alias_id, sessionId=session_id, enableTrace=enable_trace, endSession= end_session ) logger.info(pprint.pprint(agentResponse)) event_stream = agentResponse['completion'] try: for event in event_stream: if 'chunk' in event: data = event['chunk']['bytes'] logger.info(f"Final answer ->\n{data.decode('utf8')}") agent_answer = data.decode('utf8') end_event_received = True # End event indicates that the request finished successfully elif 'trace' in event: logger.info(json.dumps(event['trace'], indent=2)) else: raise Exception("unexpected event.", event) except Exception as e: raise Exception("unexpected event.", e) ``` -------------------------------- ### Create OpenSearch Collection Source: https://github.com/aws-samples/deploy-amazon-bedrock-agent-using-aws-sam/blob/main/assets/01-create-agent-AnyCity.ipynb Creates an OpenSearch Serverless collection, which is a dedicated instance for storing and querying data. This collection is configured for vector search capabilities. ```python opensearch_collection_response = open_search_serverless_client.create_collection( description='OpenSearch collection for Amazon Bedrock Knowledge Base', name=kb_collection_name, standbyReplicas='DISABLED', type='VECTORSEARCH' ) opensearch_collection_response ```