app.py

import gradio as gr
import requests

from theme import logo, theme

DEFAULT_DEPLOYMENT_URL = "https://api.app.deeploy.ml/workspaces/708b5808-27af-461a-8ee5-80add68384c7/deployments/a0a5d36d-ede6-4c53-8705-e4a8727bb0b7/predict"

DEFAULT_PROMPTS = [
    ["What are requirements for a high-risk AI system?"],
    ["Can you help me understand AI content moderation guidelines and limitations?"],
]

MAX_TOKENS = 800
TEMPERATURE = 0.7
TOP_P = 0.95

ERROR_401 = "Error: Invalid Deployment token"
ERROR_403 = "Error: No valid permissions for this Deployment token"
ERROR_404 = "Error: Deployment not found. Check the API URL."

indexed_prediction_log_ids = {}


def respond(
    message: str,
    history: list,
    api_url: str,
    deployment_token: str,
):
    formatted_history = []

    if history and isinstance(history[0], list):
        for user_msg, assistant_msg in history:
            if user_msg:
                formatted_history.append(message_from_user(user_msg))
            if assistant_msg:
                formatted_history.append(message_from_assistant(assistant_msg))
    else:
        formatted_history = history

    messages = [message_from_system("Your are a friendly Chatbot.")]
    messages.extend(formatted_history)

    if message:
        messages.append(message_from_user(message))

    headers = get_headers(deployment_token)
    payload = get_prediction_payload(messages)
    predict_url = get_predict_url(api_url)

    response = requests.post(predict_url, json=payload, headers=headers)

    new_history = formatted_history.copy()

    if message:
        new_history.append(message_from_user(message))

    if response.status_code != 201:
        append_error_to_history(new_history, response)
        return new_history

    try:
        response_data = response.json()

        if isinstance(response_data, dict) and "choices" in response_data:
            if (
                len(response_data["choices"]) > 0
                and "message" in response_data["choices"][0]
            ):
                content = response_data["choices"][0]["message"].get("content", "")
                prediction_log_id = response_data["predictionLogIds"][0]
                indexed_prediction_log_ids[len(new_history)] = prediction_log_id
                new_history.append(message_from_assistant(content))
                return new_history
        else:
            new_history.append(
                message_from_assistant(
                    f"Error: Unexpected response format: {response_data}"
                )
            )
            return new_history

    except Exception as error:
        new_history.append(
            message_from_assistant(f"Error parsing API response: {str(error)}")
        )
        return new_history


def evaluate(
    like_data: gr.LikeData,
    api_url: str,
    deployment_token: str,
) -> str | None:
    prediction_log_id = indexed_prediction_log_ids.get(like_data.index)

    headers = get_headers(deployment_token)
    evaluate_url = get_evaluation_url(api_url, prediction_log_id)
    evaluation_payload = get_evaluation_payload(like_data.liked)

    response = requests.post(evaluate_url, json=evaluation_payload, headers=headers)

    if response.status_code != 201:
        error_msg = "Error: Failed to evaluate the prediction, does your token have the right permissions?"
        return error_msg


def get_prediction_payload(messages: list) -> dict:
    return {
        "messages": messages,
        "max_tokens": MAX_TOKENS,
        "temperature": TEMPERATURE,
        "top_p": TOP_P,
    }


def get_evaluation_payload(liked: bool) -> dict:
    if liked:
        return {"agree": True, "comment": "Clicked thumbs up in the chat"}
    else:
        return {
            "agree": False,
            "comment": "Clicked thumbs down in the chat",
            "desiredOutput": {"predictions": ["A new example output"]},
        }


def get_headers(bearer_token: str) -> dict:
    return {
        "Authorization": f"Bearer {bearer_token}",
        "Content-Type": "application/json",
    }


def append_error_to_history(history: list, response: requests.Response) -> None:
    if response.status_code == 401:
        history.append(message_from_assistant(ERROR_401))
    elif response.status_code == 403:
        history.append(message_from_assistant(ERROR_403))
    elif response.status_code == 404:
        history.append(message_from_assistant(ERROR_404))
    else:
        history.append(
            message_from_assistant(
                f"Error: API returned status code {response.status_code}"
            )
        )


def message_from_assistant(message: str) -> dict:
    return {"role": "assistant", "content": message}


def message_from_user(message: str) -> dict:
    return {"role": "user", "content": message}


def message_from_system(message: str) -> dict:
    return {"role": "system", "content": message}


def get_base_url(url: str) -> str:
    if url.endswith("/predict"):
        return url.split("/predict")[0]
    else:
        if url.endswith("/"):
            return url[:-1]
        else:
            return url


def get_predict_url(url: str) -> str:
    return get_base_url(url) + "/predict"


def get_evaluation_url(url: str, prediction_log_id: str) -> str:
    return (
        get_base_url(url)
        + "/predictionLogs/"
        + prediction_log_id
        + "/evaluatePrediction"
    )


with gr.Blocks(theme=theme, mode="light") as demo:
    with gr.Row():
        with gr.Column(scale=1):
            with gr.Row():
                gr.HTML(f"""
                    <div style="display: flex; align-items: center; column-gap: 8px;">
                        {logo}
                        <h1 style="margin: 0;">Deeploy OpenAI</h1>
                    </div>
                """)
            api_url = gr.Textbox(
                value=DEFAULT_DEPLOYMENT_URL, label="Deeploy API URL", type="text"
            )
            deployment_token = gr.Textbox(label="Deployment token", type="password")

        with gr.Column(scale=2):
            chatbot = gr.Chatbot(
                height=600,
                type="messages",
                render_markdown=True,
                show_copy_button=True,
            )
            msg = gr.Textbox(
                label="Message",
                placeholder="Type your message here...",
                show_label=False,
                submit_btn="Send",
            )
            gr.Examples(
                examples=DEFAULT_PROMPTS,
                inputs=[msg],
            )

    msg.submit(
        respond,
        inputs=[msg, chatbot, api_url, deployment_token],
        outputs=chatbot,
    ).then(lambda: "", None, msg, queue=False)

    error_output = gr.Textbox(visible=False)

    chatbot.like(
        evaluate,
        inputs=[api_url, deployment_token],
        outputs=[error_output],
        like_user_message=False,
    ).success(
        lambda msg: gr.Info(msg) if msg else None,
        [error_output],
        None,
    )

if __name__ == "__main__":
    demo.launch()