app.py

import gradio as gr
import random
import requests
import os
from datetime import datetime
import hashlib
import hmac
import json

def sign(key, msg):
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()

def get_signature_key(secret_key, date_stamp, region_name, service_name):
    k_date = sign(('AWS4' + secret_key).encode('utf-8'), date_stamp)
    k_region = sign(k_date, region_name)
    k_service = sign(k_region, service_name)
    k_signing = sign(k_service, 'aws4_request')
    return k_signing

def generate_response(input, history):
    #bot_message = random.choice(["How are you?", "I love you", "I'm very hungry"])

    #Authentication AWS
    access_key = os.environ['AccessKey']
    secret_key = os.environ['SecretKey']
    region = 'eu-west-1' # e.g., 'us-west-2'
    service = 'sagemaker'
    
    #Hit the Sagemaker API and get results
    
    url = "https://runtime.sagemaker.eu-west-1.amazonaws.com/endpoints/jumpstart-dft-meta-textgeneration-llama-2-7b-f/invocations"
    #authorization_token = os.environ['Authorization']
    # headers = {
    #     "Content-Type": "application/json",
    #     "X-Amz-Date": datetime.utcnow().strftime("%Y%m%dT%H%M%SZ"),  # Using ISO-8601 basic format
    #     "X-Amzn-SageMaker-Custom-Attributes": "accept_eula=true",
    #     "Connection": "keep-alive"
    # }
    payload = {
        "inputs": [
            [
                {
                    "role": "system",
                    "content": "Always answer positively"
                },
                {
                    "role": "user",
                    "content": input #"I am going to Paris, what should I see?"
                }
            ]
        ],
        "parameters": {
            "max_new_tokens": 512,
            "top_p": 0.9,
            "temperature": 0.6
        }
    }

    payload_str = json.dumps(payload)

    t = datetime.utcnow()
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')
    date_stamp = t.strftime('%Y%m%d')

    endpoint = "runtime.sagemaker.eu-west-1.amazonaws.com"
    canonical_uri = '/endpoints/jumpstart-dft-meta-textgeneration-llama-2-7b-f/invocations' # Your canonical URI here
    canonical_headers = f'host:{endpoint}\nx-amz-date:{amz_date}\n'
    signed_headers = 'host;x-amz-date'
    
    hashlib_payload = hashlib.sha256(payload_str.encode('utf-8')).hexdigest()
    canonical_request = f'POST\n{canonical_uri}\n\n{canonical_headers}\n{signed_headers}\n{hashlib_payload}'
    
    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = f'{date_stamp}/{region}/{service}/aws4_request'
    string_to_sign = f'{algorithm}\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()}'
    
    signing_key = get_signature_key(secret_key, date_stamp, region, service)
    signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()
    
    authorization_header = f'{algorithm} Credential={access_key}/{credential_scope}, SignedHeaders={signed_headers}, Signature={signature}'
    
    headers = {
        'x-amz-date': amz_date,
        'Authorization': authorization_header,
        'Content-Type': 'application/json',
        "X-Amzn-SageMaker-Custom-Attributes": "accept_eula=true"
    }
    
    #response = requests.post(endpoint, headers=headers, data=payload_str)

    
    response = requests.post(url, json=payload, headers=headers)
    
    if response.status_code == 200:
        response_json = response.json()
        if response_json and 'generation' in response_json[0]:
            content = response_json[0]['generation']['content']
            print("Content:", content)
            bot_message = content
        else:
            print("Content not found in response")
    else:
        print("Request failed:", response.status_code, response.text)


    return bot_message
    
def my_chatbot(input, history):

    history = history or []

    my_history = list(sum(history, ()))

    my_history.append(input)

    my_input = ' '.join(my_history)

    output = generate_response(input, history)

    history.append((input, output))

    return history, history

with gr.Blocks() as demo:

    gr.Markdown("""<h1><center>My Chatbot</center></h1>""")

    chatbot = gr.Chatbot()

    state = gr.State()

    text = gr.Textbox(placeholder="Hello. Ask me a question.")

    submit = gr.Button("SEND")

    submit.click(my_chatbot, inputs=[text, state], outputs=[chatbot, state])

 

demo.launch(debug=True)