Explain Code
Ask Nova to explain code and use it in an example.
System Prompt Template
You are an expert {programming language} programmer who can read code and explain what its doing in very basic terms.
Follow these instructions:
{enumerated instructions}
User Prompt Template
## Code ##
{code to be explained}
Example
Amazon Nova Pro System Prompt
You are an expert Python programmer who can read code and explain what its doing in very basic terms.
Follow these instructions:
1. Read and understand the code written in ## Code ##. Think step by step
2. Put a 2-3 sentence summary of the code in ## Summary ##
3. Write an example of how to use the code, using code. Put this example in ## Example ## and use ```python ``` to show the example.
Amazon Nova Pro User Prompt
Here's the code:
## Code ##
import time
import requests
import json
import os
import random
import logging
from typing import Dict, Any, Optional, Union
class ApiPoller:
def __init__(
self,
endpoint_url: str,
output_file: str,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 5,
jitter: bool = True,
timeout: int = 10,
headers: Optional[Dict[str, str]] = None
):
self.endpoint_url = endpoint_url
self.output_file = output_file
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.jitter = jitter
self.timeout = timeout
self.headers = headers or {}
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def _calculate_delay(self, attempt: int) -> float:
delay = min(self.base_delay * (2 ** attempt), self.max_delay)
if self.jitter:
delay = delay * (0.5 + random.random())
return delay
def _save_response(self, data: Union[Dict[str, Any], str, bytes]) -> None:
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(os.path.abspath(self.output_file)), exist_ok=True)
# Handle different response types
if isinstance(data, (dict, list)):
with open(self.output_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2)
elif isinstance(data, str):
with open(self.output_file, 'w', encoding='utf-8') as f:
f.write(data)
elif isinstance(data, bytes):
with open(self.output_file, 'wb') as f:
f.write(data)
self.logger.info(f"Response successfully saved to {self.output_file}")
except Exception as e:
self.logger.error(f"Failed to save response: {e}")
raise
def poll(self, method: str = "GET", params: Optional[Dict[str, Any]] = None,
data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
attempt = 0
last_exception = None
while attempt <= self.max_retries:
try:
self.logger.info(f"Making {method} request to {self.endpoint_url} (attempt {attempt + 1}/{self.max_retries + 1})")
response = requests.request(
method=method,
url=self.endpoint_url,
params=params,
json=data,
headers=self.headers,
timeout=self.timeout
)
response.raise_for_status() # Raise an exception for 4XX/5XX responses
result = response.json() if 'application/json' in response.headers.get('Content-Type', '') else response.content
self.logger.info(f"Request successful, status code: {response.status_code}")
self._save_response(result)
return result if isinstance(result, dict) else {"content": result}
except requests.exceptions.RequestException as e:
last_exception = e
attempt += 1
if attempt <= self.max_retries:
delay = self._calculate_delay(attempt - 1)
self.logger.warning(f"Request failed: {e}. Retrying in {delay:.2f} seconds...")
time.sleep(delay)
else:
self.logger.error(f"Maximum retries reached. Last error: {e}")
raise
# This should not be reached but just in case
if last_exception:
raise last_exception
raise RuntimeError("Failed to poll API for unknown reasons")
Amazon Nova Pro Sample Response
Response
Summary
This code defines a class ApiPoller that repeatedly polls an API endpoint with specified parameters and saves the response to a file. It includes retry logic with exponential backoff and optional jitter, logging, and different methods to handle various response types.
Example
# Example usage of the ApiPoller class
# Initialize the ApiPoller with specific parameters
poller = ApiPoller(
endpoint_url="https://api.example.com/data",
output_file="output/response.json",
base_delay=2.0,
max_delay=30.0,
max_retries=3,
jitter=True,
timeout=15,
headers={"Authorization": "Bearer your_token"}
)
# Poll the API with GET method and specific parameters
response = poller.poll(method="GET", params={"key": "value"})
# The response will be saved to "output/response.json" and also returned
print(response)
API Request
import boto3
import json
bedrock_runtime = boto3.client(
service_name="bedrock-runtime",
region_name="us-west-2"
)
response = bedrock_runtime.converse(
modelId='us.amazon.nova-pro-v1:0',
system = [
{
"text": """You are an expert Python programmer who can read code and explain what its doing in very concise, basic terms.
Follow these instructions:
1. Read and understand the code written in ## Code ##.
2. Think step by step and put a 2-3 sentence summary of the code in ## Summary ## in markdown format
3. Write an example of how to use the code, using code. Put this example in ## Example ## and use ```python ``` to show the example."""
}
],
messages = [
{
"role": "user",
"content": [
{
"text": """Here's the code:
## Code ##
import time
import requests
import json
import os
import random
import logging
from typing import Dict, Any, Optional, Union
class ApiPoller:
def __init__(
self,
endpoint_url: str,
output_file: str,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 5,
jitter: bool = True,
timeout: int = 10,
headers: Optional[Dict[str, str]] = None
):
self.endpoint_url = endpoint_url
self.output_file = output_file
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.jitter = jitter
self.timeout = timeout
self.headers = headers or {}
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def _calculate_delay(self, attempt: int) -> float:
delay = min(self.base_delay * (2 ** attempt), self.max_delay)
if self.jitter:
delay = delay * (0.5 + random.random())
return delay
def _save_response(self, data: Union[Dict[str, Any], str, bytes]) -> None:
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(os.path.abspath(self.output_file)), exist_ok=True)
# Handle different response types
if isinstance(data, (dict, list)):
with open(self.output_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2)
elif isinstance(data, str):
with open(self.output_file, 'w', encoding='utf-8') as f:
f.write(data)
elif isinstance(data, bytes):
with open(self.output_file, 'wb') as f:
f.write(data)
self.logger.info(f"Response successfully saved to {self.output_file}")
except Exception as e:
self.logger.error(f"Failed to save response: {e}")
raise
def poll(self, method: str = "GET", params: Optional[Dict[str, Any]] = None,
data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
attempt = 0
last_exception = None
while attempt <= self.max_retries:
try:
self.logger.info(f"Making {method} request to {self.endpoint_url} (attempt {attempt + 1}/{self.max_retries + 1})")
response = requests.request(
method=method,
url=self.endpoint_url,
params=params,
json=data,
headers=self.headers,
timeout=self.timeout
)
response.raise_for_status() # Raise an exception for 4XX/5XX responses
result = response.json() if 'application/json' in response.headers.get('Content-Type', '') else response.content
self.logger.info(f"Request successful, status code: {response.status_code}")
self._save_response(result)
return result if isinstance(result, dict) else {"content": result}
except requests.exceptions.RequestException as e:
last_exception = e
attempt += 1
if attempt <= self.max_retries:
delay = self._calculate_delay(attempt - 1)
self.logger.warning(f"Request failed: {e}. Retrying in {delay:.2f} seconds...")
time.sleep(delay)
else:
self.logger.error(f"Maximum retries reached. Last error: {e}")
raise
# This should not be reached but just in case
if last_exception:
raise last_exception
raise RuntimeError("Failed to poll API for unknown reasons")"""
}
]
}
],
inferenceConfig={
"temperature": 0.1,
"topP": .99,
"maxTokens": 2000
}
)
print(json.dumps(response, indent=2))
aws bedrock-runtime converse \
--model-id "us.amazon.nova-pro-v1:0" \
--system '[
{
"text": "You are an expert Python programmer who can read code and explain what its doing in very basic terms.\nFollow these instructions:\n1. Read and understand the code written in ## Code ##. Think step by step\n2. Put a 2-3 sentence summary of the code in ## Summary ##\n3. Write an example of how to use the code, using code. Put this example in ## Example ## and use ```python ``` to show the example."
}
]' \
--messages '[
{
"role": "user",
"content": [
{
"text": "Here'\''s the code:\n## Code ##\nimport time\nimport requests\nimport json\nimport os\nimport random\nimport logging\nfrom typing import Dict, Any, Optional, Union\n\nclass ApiPoller:\n def __init__(\n self,\n endpoint_url: str,\n output_file: str,\n base_delay: float = 1.0,\n max_delay: float = 60.0,\n max_retries: int = 5,\n jitter: bool = True,\n timeout: int = 10,\n headers: Optional[Dict[str, str]] = None\n ):\n self.endpoint_url = endpoint_url\n self.output_file = output_file\n self.base_delay = base_delay\n self.max_delay = max_delay\n self.max_retries = max_retries\n self.jitter = jitter\n self.timeout = timeout\n self.headers = headers or {}\n \n # Set up logging\n logging.basicConfig(\n level=logging.INFO,\n format='\''%(asctime)s - %(name)s - %(levelname)s - %(message)s'\''\n )\n self.logger = logging.getLogger(__name__)\n \n def _calculate_delay(self, attempt: int) -> float:\n delay = min(self.base_delay * (2 ** attempt), self.max_delay)\n if self.jitter:\n delay = delay * (0.5 + random.random())\n return delay\n \n def _save_response(self, data: Union[Dict[str, Any], str, bytes]) -> None:\n try:\n # Create directory if it doesn'\''t exist\n os.makedirs(os.path.dirname(os.path.abspath(self.output_file)), exist_ok=True)\n \n # Handle different response types\n if isinstance(data, (dict, list)):\n with open(self.output_file, '\''w'\'', encoding='\''utf-8'\'') as f:\n json.dump(data, f, indent=2)\n elif isinstance(data, str):\n with open(self.output_file, '\''w'\'', encoding='\''utf-8'\'') as f:\n f.write(data)\n elif isinstance(data, bytes):\n with open(self.output_file, '\''wb'\'') as f:\n f.write(data)\n \n self.logger.info(f\"Response successfully saved to {self.output_file}\")\n except Exception as e:\n self.logger.error(f\"Failed to save response: {e}\")\n raise\n \n def poll(self, method: str = \"GET\", params: Optional[Dict[str, Any]] = None, \n data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:\n attempt = 0\n last_exception = None\n \n while attempt <= self.max_retries:\n try:\n self.logger.info(f\"Making {method} request to {self.endpoint_url} (attempt {attempt + 1}/{self.max_retries + 1})\")\n \n response = requests.request(\n method=method,\n url=self.endpoint_url,\n params=params,\n json=data,\n headers=self.headers,\n timeout=self.timeout\n )\n \n response.raise_for_status() # Raise an exception for 4XX/5XX responses\n \n result = response.json() if '\''application/json'\'' in response.headers.get('\''Content-Type'\'', '\''\'\'') else response.content\n \n self.logger.info(f\"Request successful, status code: {response.status_code}\")\n self._save_response(result)\n \n return result if isinstance(result, dict) else {\"content\": result}\n \n except requests.exceptions.RequestException as e:\n last_exception = e\n attempt += 1\n \n if attempt <= self.max_retries:\n delay = self._calculate_delay(attempt - 1)\n self.logger.warning(f\"Request failed: {e}. Retrying in {delay:.2f} seconds...\")\n time.sleep(delay)\n else:\n self.logger.error(f\"Maximum retries reached. Last error: {e}\")\n raise\n \n # This should not be reached but just in case\n if last_exception:\n raise last_exception\n raise RuntimeError(\"Failed to poll API for unknown reasons\")"
}
]
}
]' \
--inference-config '{
"temperature": 0.1,
"topP": 0.99,
"maxTokens": 512
}' \
--region us-west-2
{
"modelId": "us.amazon.nova-pro-v1:0",
"system": [
{
"text": "You are an expert Python programmer who can read code and explain what its doing in very basic terms.\nFollow these instructions:\n1. Read and understand the code written in ## Code ##. Think step by step\n2. Put a 2-3 sentence summary of the code in ## Summary ##\n3. Write an example of how to use the code, using code. Put this example in ## Example ## and use ```python ``` to show the example."
}
],
"messages": [
{
"role": "user",
"content": [
{
"text": "Here's the code:\n## Code ##\nimport time\nimport requests\nimport json\nimport os\nimport random\nimport logging\nfrom typing import Dict, Any, Optional, Union\n\nclass ApiPoller:\n def __init__(\n self,\n endpoint_url: str,\n output_file: str,\n base_delay: float = 1.0,\n max_delay: float = 60.0,\n max_retries: int = 5,\n jitter: bool = True,\n timeout: int = 10,\n headers: Optional[Dict[str, str]] = None\n ):\n self.endpoint_url = endpoint_url\n self.output_file = output_file\n self.base_delay = base_delay\n self.max_delay = max_delay\n self.max_retries = max_retries\n self.jitter = jitter\n self.timeout = timeout\n self.headers = headers or {}\n \n # Set up logging\n logging.basicConfig(\n level=logging.INFO,\n format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'\n )\n self.logger = logging.getLogger(__name__)\n \n def _calculate_delay(self, attempt: int) -> float:\n delay = min(self.base_delay * (2 ** attempt), self.max_delay)\n if self.jitter:\n delay = delay * (0.5 + random.random())\n return delay\n \n def _save_response(self, data: Union[Dict[str, Any], str, bytes]) -> None:\n try:\n # Create directory if it doesn't exist\n os.makedirs(os.path.dirname(os.path.abspath(self.output_file)), exist_ok=True)\n \n # Handle different response types\n if isinstance(data, (dict, list)):\n with open(self.output_file, 'w', encoding='utf-8') as f:\n json.dump(data, f, indent=2)\n elif isinstance(data, str):\n with open(self.output_file, 'w', encoding='utf-8') as f:\n f.write(data)\n elif isinstance(data, bytes):\n with open(self.output_file, 'wb') as f:\n f.write(data)\n \n self.logger.info(f\"Response successfully saved to {self.output_file}\")\n except Exception as e:\n self.logger.error(f\"Failed to save response: {e}\")\n raise\n \n def poll(self, method: str = \"GET\", params: Optional[Dict[str, Any]] = None, \n data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:\n attempt = 0\n last_exception = None\n \n while attempt <= self.max_retries:\n try:\n self.logger.info(f\"Making {method} request to {self.endpoint_url} (attempt {attempt + 1}/{self.max_retries + 1})\")\n \n response = requests.request(\n method=method,\n url=self.endpoint_url,\n params=params,\n json=data,\n headers=self.headers,\n timeout=self.timeout\n )\n \n response.raise_for_status() # Raise an exception for 4XX/5XX responses\n \n result = response.json() if 'application/json' in response.headers.get('Content-Type', '') else response.content\n \n self.logger.info(f\"Request successful, status code: {response.status_code}\")\n self._save_response(result)\n \n return result if isinstance(result, dict) else {\"content\": result}\n \n except requests.exceptions.RequestException as e:\n last_exception = e\n attempt += 1\n \n if attempt <= self.max_retries:\n delay = self._calculate_delay(attempt - 1)\n self.logger.warning(f\"Request failed: {e}. Retrying in {delay:.2f} seconds...\")\n time.sleep(delay)\n else:\n self.logger.error(f\"Maximum retries reached. Last error: {e}\")\n raise\n \n # This should not be reached but just in case\n if last_exception:\n raise last_exception\n raise RuntimeError(\"Failed to poll API for unknown reasons\")"
}
]
}
],
"inferenceConfig": {
"temperature": 0.1,
"topP": 0.99,
"maxTokens": 512
}
}