🎉 Gradio chatbot on via Deeploy

.gitignore

@@ -0,0 +1,2 @@
+env
+.gradio

app.py

@@ -1,64 +1,229 @@
 import gradio as gr
-from huggingface_hub import InferenceClient
+import requests
 
-"""
-For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-"""
-client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
+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,
-    history: list[tuple[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
+    message: str,
+    history: list,
+    api_url: str,
+    deployment_token: str,
 ):
-    messages = [{"role": "system", "content": system_message}]
-
-    for val in history:
-        if val[0]:
-            messages.append({"role": "user", "content": val[0]})
-        if val[1]:
-            messages.append({"role": "assistant", "content": val[1]})
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        token = message.choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-demo = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
+    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() as demo:
+    with gr.Column(scale=1):
+        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=1000,
+            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()

requirements.txt

@@ -1 +1 @@
-huggingface_hub==0.25.2
+gradio==5.30.0