app.py

import gradio as gr
from huggingface_hub import InferenceClient
import logging
import json
import os

# Initialize the InferenceClient with the model ID from Hugging Face
client = InferenceClient(model="HuggingFaceH4/zephyr-7b-beta")

# Set up logging
logging.basicConfig(
    filename='chatbot_log.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
)

API_KEYS_FILE = 'api_keys.json'

def load_api_keys():
    """Load API keys from the storage."""
    if os.path.exists(API_KEYS_FILE):
        with open(API_KEYS_FILE, 'r') as f:
            return json.load(f)
    return {}

def authenticate(api_key: str):
    """Authenticates the API key by checking against stored keys."""
    api_keys = load_api_keys()
    return api_key in api_keys.values()

def log_conversation(user_message, bot_response):
    """Logs the conversation between the user and the AI."""
    logging.info(f"User: {user_message}")
    logging.info(f"Bot: {bot_response}")

def respond(
    api_key: str,
    message: str,
    history: list[tuple[str, str]],
    system_message: str,
    max_tokens: int,
    temperature: float,
    top_p: float,
    stop_sequence: str,
    stream_response: bool,
):
    """Generates a response from the AI model based on the user's message and chat history."""

    # Authenticate the API key
    if not authenticate(api_key):
        yield "Invalid API key. Access denied."
        return
    
    # Prepare the conversation history for the API call
    messages = [{"role": "system", "content": system_message}]
    
    for user_input, assistant_response in history:
        if user_input:
            messages.append({"role": "user", "content": user_input})
        if assistant_response:
            messages.append({"role": "assistant", "content": assistant_response})

    # Add the latest user message to the conversation
    messages.append({"role": "user", "content": message})

    # Initialize an empty response
    response = ""

    try:
        if stream_response:
            # Generate a response from the model with streaming
            for message in client.chat_completion(
                messages=messages,
                max_tokens=max_tokens,
                stream=True,
                temperature=temperature,
                top_p=top_p,
                stop=stop_sequence,
            ):
                token = message.choices[0].delta.get("content", "")
                response += token
                yield response
        else:
            # Generate a complete response without streaming
            result = client.chat_completion(
                messages=messages,
                max_tokens=max_tokens,
                stream=False,
                temperature=temperature,
                top_p=top_p,
                stop=stop_sequence,
            )
            response = result.choices[0].message.get("content", "")
            log_conversation(message, response)
            yield response

    except Exception as e:
        error_message = f"An error occurred: {str(e)})"
        logging.error(error_message)
        yield error_message

# Define the ChatInterface with additional input components for user customization
demo = gr.ChatInterface(
    fn=respond,
    additional_inputs=[
        gr.Textbox(value="", label="API Key", lines=1, type="password"),
        gr.Textbox(value="You are a friendly Chatbot.", label="System Message", lines=2),
        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max New Tokens"),
        gr.Slider(minimum=0.1, maximum=2.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)"),
        gr.Textbox(value="", label="Stop Sequence (optional)", lines=1),
        gr.Checkbox(label="Stream Response", value=True),
    ],
    title="AI Chatbot Interface",
    description="Interact with an AI chatbot powered by Hugging Face's Zephyr-7B model. Customize the chatbot's behavior and response generation settings.",
    theme="default",
    allow_flagging="never",
)

# Launch the Gradio interface
if __name__ == "__main__":
    logging.info("Launching the Gradio interface...")
    demo.launch()
    logging.info("Gradio interface launched successfully.")