Building an Appraiser Assistive Tool using Amazon Nova Pro

from typing import List
# Import specific Boto3 client for AWS Bedrock
from boto3 import client as boto3_client
import time
from botocore.exceptions import ClientError
import random

def analyze_property_images_with_bedrock(
image_paths: List[str],
inference_profile_arn: str,
system_prompt: str
) -> str:
"""
Analyze frames using Amazon Bedrock Converse API.

Args:
images: List of paths to property images
inference_profile_arn: ARN of the Bedrock inference profile
system_prompt: use prompt text from `property_condition_rating.txt` for property condition rating
or use prompt text from `quality_of_construction_rating.txt` for quality rating

Returns:
Letter rating, numeric rating, and the reason for the generated rating
"""
bedrock_runtime = boto3_client(
service_name="bedrock-runtime",
# Extract region from ARN
region_name=inference_profile_arn.split(":")[3]
)

all_ratings = []

# Prepare the messages for conversation
for image_path in image_paths:
messages = []
# Read image as bytes without encoding
with open(image_path, "rb") as image_file:
image_bytes = image_file.read()

messages.append({
"role": "user",
"content": [{
"image": {
"format": "jpeg",
"source": {
# Send raw bytes directly
"bytes": image_bytes
}
}
}]
})

# Use the Converse API with retry logic
response = call_converse_with_retry(
bedrock_runtime,
modelId=inference_profile_arn,
messages=messages,
system=[{
"text": system_prompt
}]
)

# Extract content from response
if ('output' in response and
'message' in response['output'] and
'content' in response['output']['message']):
rating = response['output']['message']['content'][0]['text']
# print(f"{image_path} rating is: {rating}")
all_ratings.append(rating)
else:
print(f"Unexpected response format: {response}")

# Generate final summary using Converse API
try:
# Combine all ratings into a single composite rating
combined_ratings = "\n\n".join(all_ratings)
print("Starting final rating...")
final_response = call_converse_with_retry(
bedrock_runtime,
modelId=inference_profile_arn,
messages=[{
"role": "user",
"content": [{
"text": combined_ratings
}]
}],
system=[{
"text": (
"Provide a overall numeric rating, letter rating, and the reason for the rating. "
"Letter rating C1 is the best while C6 is the worst."
"Numeric rating 1 is the best while 6 is the worst."
"The overall rating should be based on the worst case rating."
)
}]
)

# Extract content from final response
if ('output' in final_response and
'message' in final_response['output'] and
'content' in final_response['output']['message']):
return final_response['output']['message']['content'][0]['text']
else:
print(f"Unexpected final response format: {final_response}")
return "Error: Unexpected response format"

except Exception as e:
print(f"Error generating final summary: {str(e)}")
return "Error generating summary"

def call_converse_with_retry(bedrock_runtime, **kwargs):
"""
Call the Bedrock Converse API with exponential backoff retry logic.

Args:
bedrock_runtime: Bedrock runtime client
**kwargs: Arguments to pass to the converse API

Returns:
API response
"""
max_retries = 5
base_delay = 1 # Starting delay of 1 second

for attempt in range(max_retries):
try:
return bedrock_runtime.converse(**kwargs)
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ThrottlingException':
if attempt == max_retries - 1: # Last attempt
raise # Re-raise the exception if we've exhausted all retries

# Calculate exponential backoff with jitter
delay = (2 ** attempt * base_delay) + random.random()
print(f"ThrottlingException encountered. Retrying in {delay:.2f} seconds...")
time.sleep(delay)
else:
raise # Re-raise if it's not a throttling error
except Exception as e:
# For other exceptions, don't retry
raise