File size: 3,497 Bytes
e47481b 694a9b2 e47481b 694a9b2 e47481b df6c855 7bfc491 694a9b2 e47481b 98e87f9 694a9b2 e47481b 741f470 e47481b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
import os
import json
from io import BytesIO
import boto3
import requests
from openai import OpenAI
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
BEDROCK_MODEL_ID = "anthropic.claude-3-5-sonnet-20240620-v1:0"
OPENAI_MODEL_ID = "o4-mini"
OPENAI_MODEL_IMAGE_MODEL_ID = "gpt-4.1-mini"
bedrock_runtime = boto3.client("bedrock-runtime", region_name=os.getenv("AWS_REGION"))
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def invoke_bedrock_model(messages: list[dict]) -> str:
"""Invokes a Bedrock model with the provided messages."""
response = bedrock_runtime.invoke_model(
modelId=BEDROCK_MODEL_ID,
body=json.dumps({"anthropic_version": "bedrock-2023-05-31", "max_tokens": 4096, "messages": messages}),
)["body"].read()
return json.loads(response)["content"][0]["text"]
def invoke_openai_model(messages: list[dict]) -> str:
"""Invokes an OpenAI model with the provided messages."""
response = openai_client.responses.create(
model=OPENAI_MODEL_IMAGE_MODEL_ID,
input=messages,
)
return response.output_text
def get_file(task_id: str) -> BytesIO:
"""Fetches a file associated with a task ID from the default API URL.
Parameters
----------
task_id : str
The ID of the task for which the file is to be fetched.
Returns
-------
BytesIO
A BytesIO object containing the file content.
Raises
------
requests.exceptions.RequestException
If there is an error during the HTTP request.
Exception
For any other unexpected errors that may occur.
"""
url = f"{DEFAULT_API_URL}/files/{task_id}"
try:
response = requests.get(url, timeout=15)
response.raise_for_status()
return BytesIO(response.content)
except requests.exceptions.RequestException as e:
print(f"Error fetching file for task {task_id}: {e}")
raise
except Exception as e:
print(f"An unexpected error occurred fetching file for task {task_id}: {e}")
raise
def s3_upload_file(file_content: BytesIO, bucket_name: str, object_name: str) -> None:
"""Uploads a file to an S3 bucket.
Parameters
----------
file_content : BytesIO
The content of the file to upload.
bucket_name : str
The name of the S3 bucket.
object_name : str
The name of the object in the S3 bucket.
Raises
------
Exception
If there is an error during the upload process.
"""
try:
s3_client = boto3.client("s3")
s3_client.put_object(Bucket=bucket_name, Key=object_name, Body=file_content.getvalue())
except Exception as e:
print(f"Error uploading file to S3: {e}")
raise
def s3_download_file(bucket_name: str, object_name: str) -> BytesIO:
"""Downloads a file from an S3 bucket.
Parameters
----------
bucket_name : str
The name of the S3 bucket.
object_name : str
The name of the object in the S3 bucket.
Returns
-------
BytesIO
A BytesIO object containing the downloaded file content.
Raises
------
Exception
If there is an error during the download process.
"""
try:
s3_client = boto3.client("s3")
response = s3_client.get_object(Bucket=bucket_name, Key=object_name)
return BytesIO(response["Body"].read())
except Exception as e:
print(f"Error downloading file from S3: {e}")
raise
|