app.py

import json
import os
import shutil
import requests
import threading
import time
import random
import inspect
import logging
import warnings

from requests import HTTPError
from dataclasses import asdict

from pathlib import Path
from typing import Callable, Literal, AsyncGenerator
from gradio.utils import SyncToAsyncIterator, async_iteration

from gradio.components import (
    Button,
    Chatbot,
    IOComponent,
    Markdown,
    State,
    Textbox,
    get_component_instance,
)

from huggingface_hub.utils import (
    BadRequestError,
    build_hf_headers,
    get_session,
    hf_raise_for_status,
)
from huggingface_hub.inference._common import (
    TASKS_EXPECTING_IMAGES,
    ContentT,
    InferenceTimeoutError,
    ModelStatus,
    _b64_encode,
    _b64_to_image,
    _bytes_to_dict,
    _bytes_to_image,
    _bytes_to_list,
    _import_numpy,
    _is_tgi_server,
    _open_as_binary,
    _set_as_non_tgi,
    _stream_text_generation_response,
)
from huggingface_hub.inference._text_generation import (
    TextGenerationParameters,
    TextGenerationRequest,
    TextGenerationResponse,
    TextGenerationStreamResponse,
    raise_text_generation_error,
)
from huggingface_hub.inference._types import (
    ClassificationOutput,
    ConversationalOutput,
    FillMaskOutput,
    ImageSegmentationOutput,
    ObjectDetectionOutput,
    QuestionAnsweringOutput,
    TableQuestionAnsweringOutput,
    TokenClassificationOutput,
)

from gradio.themes import ThemeClass as Theme

import gradio as gr
from gradio.helpers import special_args
from gradio.routes import Request
import anyio

from huggingface_hub import HfApi, Repository, InferenceClient
from utils import force_git_push

from typing import (
    TYPE_CHECKING,
    Any,
    Dict,
    Iterable,
    List,
    Literal,
    Optional,
    Union,
    overload,
)

logger = logging.getLogger(__name__)

HF_TOKEN = os.environ.get("HF_TOKEN", None)
DATASET_REPO_URL = os.getenv("DATASET_REPO_URL")
MODEL_NAME = os.getenv("MODEL_NAME")
ENDPOINT = os.getenv("ENDPOINT")
FORCE_PUSH = os.getenv("FORCE_PUSH")
REVISION = os.getenv("REVISION")
BOT_NAME = "PersianGPT-FT"
PUSH_FREQUENCY = 60 # every minute

HISTORY = ""
PROMPT = ""
NAME = ""
SYSTEM_PROMPT = ""
TEMPERATURE = ""
MAX_NEW_TOKENS = ""
TOP_P = ""
TOP_K = ""
REPETITION_PENALTY = ""

MODEL_REPO_URL = f"https://huggingface.co/{MODEL_NAME}"

if len(ENDPOINT)>0:
    API_URL = f"{ENDPOINT}" #/revision/{REVISION}" if len(REVISION)>0 else f"{ENDPOINT}"
    MODEL_VERSION = f"{ENDPOINT}" if len(REVISION)==0 else REVISION
    print(f'ENDPOINT: {ENDPOINT}')
else:
    API_URL = f"https://api-inference.huggingface.co/models/{MODEL_NAME}/revision/{REVISION}" if len(REVISION)>0 else f"https://api-inference.huggingface.co/models/{MODEL_NAME}"
    model_repo = Repository(local_dir="model", 
                        clone_from=MODEL_REPO_URL, 
                        use_auth_token=HF_TOKEN, 
                        skip_lfs_files=True,
                        git_user="vpcom",
                        revision=REVISION)
    MODEL_VERSION = model_repo.git_head_commit_url() if len(REVISION)==0 else REVISION
    print(f'API URL: {API_URL}')

#print(f'Model Version: {MODEL_VERSION}')

DATA_FILENAME = "data.jsonl"
DATA_FILE = os.path.join("data", DATA_FILENAME)
data_repo = Repository(local_dir="data", 
                  clone_from=DATASET_REPO_URL, 
                  use_auth_token=HF_TOKEN)

stop_sequences = ["<|endoftext|>"]

class InferenceClientUS(InferenceClient):
    def __init__(
        self,
        model: Optional[str] = None,
        token: Union[str, bool, None] = None,
        timeout: Optional[float] = None,
        headers: Optional[Dict[str, str]] = None,
        cookies: Optional[Dict[str, str]] = None,
    ) -> None:
        super().__init__(
            model=model,
            token=token,
            timeout=timeout,
            headers=headers,
            cookies=cookies,
        )
        
    def post(
        self,
        *,
        json: Optional[Union[str, Dict, List]] = None,
        data: Optional[ContentT] = None,
        model: Optional[str] = None,
        task: Optional[str] = None,
        stream: bool = False,
    ) -> Union[bytes, Iterable[bytes]]:
        """
        Make a POST request to the inference server.

        Args:
            json (`Union[str, Dict, List]`, *optional*):
                The JSON data to send in the request body. Defaults to None.
            data (`Union[str, Path, bytes, BinaryIO]`, *optional*):
                The content to send in the request body. It can be raw bytes, a pointer to an opened file, a local file
                path, or a URL to an online resource (image, audio file,...). If both `json` and `data` are passed,
                `data` will take precedence. At least `json` or `data` must be provided. Defaults to None.
            model (`str`, *optional*):
                The model to use for inference. Can be a model ID hosted on the Hugging Face Hub or a URL to a deployed
                Inference Endpoint. Will override the model defined at the instance level. Defaults to None.
            task (`str`, *optional*):
                The task to perform on the inference. Used only to default to a recommended model if `model` is not
                provided. At least `model` or `task` must be provided. Defaults to None.
            stream (`bool`, *optional*):
                Whether to iterate over streaming APIs.

        Returns:
            bytes: The raw bytes returned by the server.

        Raises:
            [`InferenceTimeoutError`]:
                If the model is unavailable or the request times out.
            `HTTPError`:
                If the request fails with an HTTP error status code other than HTTP 503.
        """
        url = self._resolve_url(model, task)

        if data is not None and json is not None:
            warnings.warn("Ignoring `json` as `data` is passed as binary.")

        # Set Accept header if relevant
        headers = self.headers.copy()
        if task in TASKS_EXPECTING_IMAGES and "Accept" not in headers:
            headers["Accept"] = "image/png"

        t0 = time.time()
        timeout = self.timeout
        while True:
            with _open_as_binary(data) as data_as_binary:
                try:
                    response = get_session().post(
                        url,
                        json=json,
                        data=data_as_binary,
                        headers=headers,
                        cookies=self.cookies,
                        timeout=self.timeout,
                        stream=stream,
                    )
                except TimeoutError as error:
                    # Convert any `TimeoutError` to a `InferenceTimeoutError`
                    raise InferenceTimeoutError(f"Inference call timed out: {url}") from error  # type: ignore

            try:
                hf_raise_for_status(response)
                return response.iter_lines() if stream else response.content
            except HTTPError as error:
                if error.response.status_code == 503:
                    # If Model is unavailable, either raise a TimeoutError...
                    if timeout is not None and time.time() - t0 > timeout:
                        raise InferenceTimeoutError(
                            f"Model not loaded on the server: {url}. Please retry with a higher timeout (current:"
                            f" {self.timeout}).",
                            request=error.request,
                            response=error.response,
                        ) from error
                    # ...or wait 1s and retry
                    logger.info(f"Waiting for model to be loaded on the server: {error}")
                    time.sleep(1)
                    if timeout is not None:
                        timeout = max(self.timeout - (time.time() - t0), 1)  # type: ignore
                    continue
                raise

    def text_generation(
        self,
        prompt: str,
        *,
        details: bool = False,
        stream: bool = False,
        model: Optional[str] = None,
        do_sample: bool = False,
        max_new_tokens: int = 20,
        best_of: Optional[int] = None,
        repetition_penalty: Optional[float] = None,
        return_full_text: bool = False,
        seed: Optional[int] = None,
        stop_sequences: Optional[List[str]] = None,
        temperature: Optional[float] = None,
        top_k: Optional[int] = None,
        top_p: Optional[float] = None,
        truncate: Optional[int] = None,
        typical_p: Optional[float] = None,
        watermark: bool = False,
        decoder_input_details: bool = False,
    ) -> Union[str, TextGenerationResponse, Iterable[str], Iterable[TextGenerationStreamResponse]]:
        """
        Given a prompt, generate the following text.

        It is recommended to have Pydantic installed in order to get inputs validated. This is preferable as it allow
        early failures.

        API endpoint is supposed to run with the `text-generation-inference` backend (TGI). This backend is the
        go-to solution to run large language models at scale. However, for some smaller models (e.g. "gpt2") the
        default `transformers` + `api-inference` solution is still in use. Both approaches have very similar APIs, but
        not exactly the same. This method is compatible with both approaches but some parameters are only available for
        `text-generation-inference`. If some parameters are ignored, a warning message is triggered but the process
        continues correctly.

        To learn more about the TGI project, please refer to https://github.com/huggingface/text-generation-inference.

        Args:
            prompt (`str`):
                Input text.
            details (`bool`, *optional*):
                By default, text_generation returns a string. Pass `details=True` if you want a detailed output (tokens,
                probabilities, seed, finish reason, etc.). Only available for models running on with the
                `text-generation-inference` backend.
            stream (`bool`, *optional*):
                By default, text_generation returns the full generated text. Pass `stream=True` if you want a stream of
                tokens to be returned. Only available for models running on with the `text-generation-inference`
                backend.
            model (`str`, *optional*):
                The model to use for inference. Can be a model ID hosted on the Hugging Face Hub or a URL to a deployed
                Inference Endpoint. This parameter overrides the model defined at the instance level. Defaults to None.
            do_sample (`bool`):
                Activate logits sampling
            max_new_tokens (`int`):
                Maximum number of generated tokens
            best_of (`int`):
                Generate best_of sequences and return the one if the highest token logprobs
            repetition_penalty (`float`):
                The parameter for repetition penalty. 1.0 means no penalty. See [this
                paper](https://arxiv.org/pdf/1909.05858.pdf) for more details.
            return_full_text (`bool`):
                Whether to prepend the prompt to the generated text
            seed (`int`):
                Random sampling seed
            stop_sequences (`List[str]`):
                Stop generating tokens if a member of `stop_sequences` is generated
            temperature (`float`):
                The value used to module the logits distribution.
            top_k (`int`):
                The number of highest probability vocabulary tokens to keep for top-k-filtering.
            top_p (`float`):
                If set to < 1, only the smallest set of most probable tokens with probabilities that add up to `top_p` or
                higher are kept for generation.
            truncate (`int`):
                Truncate inputs tokens to the given size
            typical_p (`float`):
                Typical Decoding mass
                See [Typical Decoding for Natural Language Generation](https://arxiv.org/abs/2202.00666) for more information
            watermark (`bool`):
                Watermarking with [A Watermark for Large Language Models](https://arxiv.org/abs/2301.10226)
            decoder_input_details (`bool`):
                Return the decoder input token logprobs and ids. You must set `details=True` as well for it to be taken
                into account. Defaults to `False`.

        Returns:
            `Union[str, TextGenerationResponse, Iterable[str], Iterable[TextGenerationStreamResponse]]`:
            Generated text returned from the server:
            - if `stream=False` and `details=False`, the generated text is returned as a `str` (default)
            - if `stream=True` and `details=False`, the generated text is returned token by token as a `Iterable[str]`
            - if `stream=False` and `details=True`, the generated text is returned with more details as a [`~huggingface_hub.inference._text_generation.TextGenerationResponse`]
            - if `details=True` and `stream=True`, the generated text is returned token by token as a iterable of [`~huggingface_hub.inference._text_generation.TextGenerationStreamResponse`]

        Raises:
            `ValidationError`:
                If input values are not valid. No HTTP call is made to the server.
            [`InferenceTimeoutError`]:
                If the model is unavailable or the request times out.
            `HTTPError`:
                If the request fails with an HTTP error status code other than HTTP 503.

        Example:
        ```py
        >>> from huggingface_hub import InferenceClient
        >>> client = InferenceClient()

        # Case 1: generate text
        >>> client.text_generation("The huggingface_hub library is ", max_new_tokens=12)
        '100% open source and built to be easy to use.'

        # Case 2: iterate over the generated tokens. Useful for large generation.
        >>> for token in client.text_generation("The huggingface_hub library is ", max_new_tokens=12, stream=True):
        ...     print(token)
        100
        %
        open
        source
        and
        built
        to
        be
        easy
        to
        use
        .

        # Case 3: get more details about the generation process.
        >>> client.text_generation("The huggingface_hub library is ", max_new_tokens=12, details=True)
        TextGenerationResponse(
            generated_text='100% open source and built to be easy to use.',
            details=Details(
                finish_reason=<FinishReason.Length: 'length'>,
                generated_tokens=12,
                seed=None,
                prefill=[
                    InputToken(id=487, text='The', logprob=None),
                    InputToken(id=53789, text=' hugging', logprob=-13.171875),
                    (...)
                    InputToken(id=204, text=' ', logprob=-7.0390625)
                ],
                tokens=[
                    Token(id=1425, text='100', logprob=-1.0175781, special=False),
                    Token(id=16, text='%', logprob=-0.0463562, special=False),
                    (...)
                    Token(id=25, text='.', logprob=-0.5703125, special=False)
                ],
                best_of_sequences=None
            )
        )

        # Case 4: iterate over the generated tokens with more details.
        # Last object is more complete, containing the full generated text and the finish reason.
        >>> for details in client.text_generation("The huggingface_hub library is ", max_new_tokens=12, details=True, stream=True):
        ...     print(details)
        ...
        TextGenerationStreamResponse(token=Token(id=1425, text='100', logprob=-1.0175781, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=16, text='%', logprob=-0.0463562, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=1314, text=' open', logprob=-1.3359375, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=3178, text=' source', logprob=-0.28100586, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=273, text=' and', logprob=-0.5961914, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=3426, text=' built', logprob=-1.9423828, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=271, text=' to', logprob=-1.4121094, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=314, text=' be', logprob=-1.5224609, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=1833, text=' easy', logprob=-2.1132812, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=271, text=' to', logprob=-0.08520508, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(id=745, text=' use', logprob=-0.39453125, special=False), generated_text=None, details=None)
        TextGenerationStreamResponse(token=Token(
            id=25,
            text='.',
            logprob=-0.5703125,
            special=False),
            generated_text='100% open source and built to be easy to use.',
            details=StreamDetails(finish_reason=<FinishReason.Length: 'length'>, generated_tokens=12, seed=None)
        )
        ```
        """
        # NOTE: Text-generation integration is taken from the text-generation-inference project. It has more features
        # like input/output validation (if Pydantic is installed). See `_text_generation.py` header for more details.

        if decoder_input_details and not details:
            warnings.warn(
                "`decoder_input_details=True` has been passed to the server but `details=False` is set meaning that"
                " the output from the server will be truncated."
            )
            decoder_input_details = False

        # Validate parameters
        parameters = TextGenerationParameters(
            best_of=best_of,
            details=details,
            do_sample=do_sample,
            max_new_tokens=max_new_tokens,
            repetition_penalty=repetition_penalty,
            return_full_text=return_full_text,
            seed=seed,
            stop=stop_sequences if stop_sequences is not None else [],
            temperature=temperature,
            top_k=top_k,
            top_p=top_p,
            truncate=truncate,
            typical_p=typical_p,
            watermark=watermark,
            decoder_input_details=decoder_input_details,
        )
        request = TextGenerationRequest(inputs=prompt, stream=stream, parameters=parameters)
        payload = asdict(request)


        # add the use_cache option
        print(f"payload:{payload}")
        payload["options"] = {}
        payload["options"]['use_cache'] = False

        # Remove some parameters if not a TGI server
        if not _is_tgi_server(model):
            ignored_parameters = []
            for key in "watermark", "stop", "details", "decoder_input_details":
                if payload["parameters"][key] is not None:
                    ignored_parameters.append(key)
                del payload["parameters"][key]
            if len(ignored_parameters) > 0:
                warnings.warn(
                    "API endpoint/model for text-generation is not served via TGI. Ignoring parameters"
                    f" {ignored_parameters}.",
                    UserWarning,
                )
            if details:
                warnings.warn(
                    "API endpoint/model for text-generation is not served via TGI. Parameter `details=True` will"
                    " be ignored meaning only the generated text will be returned.",
                    UserWarning,
                )
                details = False
            if stream:
                raise ValueError(
                    "API endpoint/model for text-generation is not served via TGI. Cannot return output as a stream."
                    " Please pass `stream=False` as input."
                )

        # Handle errors separately for more precise error messages
        try:
            bytes_output = self.post(json=payload, model=model, task="text-generation", stream=stream)  # type: ignore
        except HTTPError as e:
            if isinstance(e, BadRequestError) and "The following `model_kwargs` are not used by the model" in str(e):
                _set_as_non_tgi(model)
                return self.text_generation(  # type: ignore
                    prompt=prompt,
                    details=details,
                    stream=stream,
                    model=model,
                    do_sample=do_sample,
                    max_new_tokens=max_new_tokens,
                    best_of=best_of,
                    repetition_penalty=repetition_penalty,
                    return_full_text=return_full_text,
                    seed=seed,
                    stop_sequences=stop_sequences,
                    temperature=temperature,
                    top_k=top_k,
                    top_p=top_p,
                    truncate=truncate,
                    typical_p=typical_p,
                    watermark=watermark,
                    decoder_input_details=decoder_input_details,
                )
            raise_text_generation_error(e)

        # Parse output
        if stream:
            return _stream_text_generation_response(bytes_output, details)  # type: ignore

        data = _bytes_to_dict(bytes_output)[0]
        return TextGenerationResponse(**data) if details else data["generated_text"]

client = InferenceClientUS(
    API_URL,
    headers={"Authorization": f"Bearer {HF_TOKEN}"},
)

def asynchronous_push(f_stop):
    if data_repo.is_repo_clean():
        print("Repo currently clean. Ignoring push_to_hub")
        print(data_repo.huggingface_token)
    else:
        data_repo.git_add(auto_lfs_track=True)
        data_repo.git_commit("Auto commit by space")
        if FORCE_PUSH == "yes":
            force_git_push(data_repo)
        else:
            data_repo.git_push()
    if not f_stop.is_set():
        # call again in 60 seconds
        threading.Timer(PUSH_FREQUENCY, asynchronous_push, [f_stop]).start()

f_stop = threading.Event()
asynchronous_push(f_stop)

def format_prompt(message, history, system_prompt):
  prompt = "" # ؛
  if system_prompt:
    prompt += f"{system_prompt}"
  for user_prompt, bot_response in history:
    prompt += f"{user_prompt}"
    prompt += f"{bot_response}"
  prompt += f"""{message}"""
  return prompt.replace('\n','؛').replace('\t','/').replace(' * ','/').replace('\u200c',' ').strip()

def generate(
    prompt, history, name, system_prompt, 
    temperature=0.9, max_new_tokens=100, top_p=0.95, top_k=100,
    repetition_penalty=1.0, seed=42,
):
    global PROMPT
    PROMPT = prompt
    global HISTORY
    HISTORY = history
    global NAME
    NAME = name
    global SYSTEM_PROMPT
    SYSTEM_PROMPT = system_prompt
    global TEMPERATURE
    TEMPERATURE = temperature
    global MAX_NEW_TOKENS
    MAX_NEW_TOKENS = max_new_tokens
    global TOP_P
    TOP_P = top_p
    global TOP_K
    TOP_K = top_k
    global REPETITION_PENALTY
    REPETITION_PENALTY = repetition_penalty
    
    temperature = float(temperature)
    if temperature < 1e-2:
        temperature = 1e-2
    top_p = float(top_p)
    generate_kwargs = dict(
        max_new_tokens=max_new_tokens,
        #output_scores=True, # Not supported as of now
        temperature=None if temperature==1 else temperature,
        top_p=None if top_p==0 else top_p,
        repetition_penalty=None if repetition_penalty==1 else repetition_penalty,
        top_k=None if top_k==0 else top_k,
        stop_sequences=stop_sequences,
        do_sample=True,
        #best_of=2,
        #typical_p=0.9,
        #seed=seed,
    )
    #seed = seed + 1
    history = [] # explicitly set the history as none so no history is important anymore
    formatted_prompt = format_prompt(prompt, history, system_prompt)

    print(f"Formatted Prompt: {formatted_prompt}")
    
    if len(ENDPOINT)>0:
        stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=False, details=True, return_full_text=False)
        stream = [stream] # if stream false
    else:
        stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=False, details=True, return_full_text=False)

    output = "" #f"{prompt}"
    for response in stream:
        #print('stream',response)

        if len(ENDPOINT)>0:
            if isinstance(response, str):
                output += response if response else ""
            else:
                output += response.generated_text if response.generated_text else ""
        else:        
            output += response

        for stop_str in stop_sequences:
            if output.endswith(stop_str):
                output = output[:-len(stop_str)]
                output = output.rstrip()
                yield output
        yield output

    return output

additional_inputs=[
    gr.Textbox(
            "",
            label="Who are you?",
            placeholder="...Please write your name here",
        ),
    gr.Textbox("<|endoftext|>", label="Optional system prompt"), #<|endoftext|>
    gr.Slider(
        label="Temperature",
        value=1.0,
        minimum=0.0,
        maximum=1.0,
        step=0.01,
        interactive=True,
        info="Higher values produce more diverse outputs",
    ),
    gr.Slider(
        label="Max new tokens",
        value=64,
        minimum=0,
        maximum=250,
        step=64,
        interactive=True,
        info="The maximum numbers of new tokens",
    ),
    gr.Slider(
        label="Top-p (nucleus sampling)",
        value=0,
        minimum=0,
        maximum=1,
        step=0.05,
        interactive=True,
        info="Higher values sample more low-probability tokens",
    ),
    gr.Slider(
        label="Top-k",
        value=0,
        minimum=0,
        maximum=1000,
        step=10,
        interactive=True,
        info="Higher values sample more low-probability tokens",
    ),
    gr.Slider(
        label="Repetition penalty",
        value=1.0,
        minimum=1.0001,
        maximum=2.0,
        step=0.05,
        interactive=True,
        info="Penalize repeated tokens",
    ),
    gr.Markdown(
                """
                WARNING: This is the development version, for the stable version check out! [LINK](https://huggingface.co/spaces/DataAnalyticsLab/Sorvad)

                Model Name: Sorvãd (سرواد)

                سرواد. [ س رْ ] (اِ) کلام منظوم و شعر. (برهان ) (غیاث ). شعر پارسی . (تفلیسی ) :
                دگر نخواهم گفتن همی ثنا وغزل
                که رفت یکرهه بازار و قیمت سرواد.
                لبیبی (از لغت فرس ص 108).
                
                
                زهی به عدل تو مرهون عمارت دنیا
                خهی به مدح تومشحون رسایل و سرواد.
                شمس فخری .
                
                - Mojtaba Valipour: Model Design, Pretraining, Data Collection and Programming
                - Ali Ghodsi: Advising
                - Amir Mohammad Marshal Pirgheybi: Data Processing and Cleaning
                """
            )
]

CSS = """
.gradio-container textarea {direction: rtl; white-space: pre-line;}
#component-11 #component-12 {direction: rtl; white-space: pre-line;}
p {direction: rtl; white-space: pre-line;}
"""

class Chatbot(gr.Chatbot):
    def __init__(
        self,
        value: list[list[str | tuple[str] | tuple[str | Path, str] | None]]
        | Callable
        | None = None,
        color_map: dict[str, str] | None = None,
        *,
        label: str | None = None,
        every: float | None = None,
        show_label: bool | None = None,
        container: bool = True,
        scale: int | None = None,
        min_width: int = 160,
        visible: bool = True,
        elem_id: str | None = None,
        elem_classes: list[str] | str | None = None,
        height: int | None = None,
        latex_delimiters: list[dict[str, str | bool]] | None = None,
        rtl: bool = False,
        show_share_button: bool | None = None,
        show_copy_button: bool = False,
        avatar_images: tuple[str | Path | None, str | Path | None] | None = None,
        sanitize_html: bool = True,
        render_markdown: bool = True,
        bubble_full_width: bool = True,
        line_breaks: bool = True,
        layout: Literal["panel", "bubble"] | None = None,
        **kwargs,
    ):
        """
        Parameters:
            value: Default value to show in chatbot. If callable, the function will be called whenever the app loads to set the initial value of the component.
            color_map: This parameter is deprecated.
            label: component name in interface.
            every: If `value` is a callable, run the function 'every' number of seconds while the client connection is open. Has no effect otherwise. Queue must be enabled. The event can be accessed (e.g. to cancel it) via this component's .load_event attribute.
            show_label: if True, will display label.
            container: If True, will place the component in a container - providing some extra padding around the border.
            scale: relative width compared to adjacent Components in a Row. For example, if Component A has scale=2, and Component B has scale=1, A will be twice as wide as B. Should be an integer.
            min_width: minimum pixel width, will wrap if not sufficient screen space to satisfy this value. If a certain scale value results in this Component being narrower than min_width, the min_width parameter will be respected first.
            visible: If False, component will be hidden.
            elem_id: An optional string that is assigned as the id of this component in the HTML DOM. Can be used for targeting CSS styles.
            elem_classes: An optional list of strings that are assigned as the classes of this component in the HTML DOM. Can be used for targeting CSS styles.
            height: height of the component in pixels.
            latex_delimiters: A list of dicts of the form {"left": open delimiter (str), "right": close delimiter (str), "display": whether to display in newline (bool)} that will be used to render LaTeX expressions. If not provided, `latex_delimiters` is set to `[{ "left": "$$", "right": "$$", "display": True }]`, so only expressions enclosed in $$ delimiters will be rendered as LaTeX, and in a new line. Pass in an empty list to disable LaTeX rendering. For more information, see the [KaTeX documentation](https://katex.org/docs/autorender.html).
            rtl: If True, sets the direction of the rendered text to right-to-left. Default is False, which renders text left-to-right.
            show_share_button: If True, will show a share icon in the corner of the component that allows user to share outputs to Hugging Face Spaces Discussions. If False, icon does not appear. If set to None (default behavior), then the icon appears if this Gradio app is launched on Spaces, but not otherwise.
            show_copy_button: If True, will show a copy button for each chatbot message.
            avatar_images: Tuple of two avatar image paths or URLs for user and bot (in that order). Pass None for either the user or bot image to skip. Must be within the working directory of the Gradio app or an external URL.
            sanitize_html: If False, will disable HTML sanitization for chatbot messages. This is not recommended, as it can lead to security vulnerabilities.
            render_markdown: If False, will disable Markdown rendering for chatbot messages.
            bubble_full_width: If False, the chat bubble will fit to the content of the message. If True (default), the chat bubble will be the full width of the component.
            line_breaks: If True (default), will enable Github-flavored Markdown line breaks in chatbot messages. If False, single new lines will be ignored. Only applies if `render_markdown` is True.
            layout: If "panel", will display the chatbot in a llm style layout. If "bubble", will display the chatbot with message bubbles, with the user and bot messages on alterating sides. Will default to "bubble".
        """
        super().__init__(
            value = value,
            color_map = color_map,
            label = label,
            every = every,
            show_label = show_label,
            container = container,
            scale = scale,
            min_width = min_width,
            visible = visible,
            elem_id = elem_id,
            elem_classes = elem_classes,
            height = height,
            latex_delimiters = latex_delimiters,
            rtl = rtl,
            show_share_button = show_share_button,
            show_copy_button = show_copy_button,
            avatar_images = avatar_images,
            sanitize_html = sanitize_html,
            render_markdown = render_markdown,
            bubble_full_width = bubble_full_width,
            line_breaks = line_breaks,
            layout = layout,
            kwargs=kwargs,
        )
        
    def _preprocess_chat_messages(
        self, chat_message: str | dict | None
    ) -> str | tuple[str] | tuple[str, str] | None:
        if chat_message is None:
            return None
        elif isinstance(chat_message, dict):
            if chat_message["alt_text"] is not None:
                return (chat_message["name"], chat_message["alt_text"])
            else:
                return (chat_message["name"],)
        else:  # string
            return chat_message
            
    def preprocess(
        self,
        y: list[list[str | dict | None] | tuple[str | dict | None, str | dict | None]],
    ) -> list[list[str | tuple[str] | tuple[str, str] | None]]:
        if y is None:
            return y
        processed_messages = []
        for message_pair in y:
            if not isinstance(message_pair, (tuple, list)):
                raise TypeError(
                    f"Expected a list of lists or list of tuples. Received: {message_pair}"
                )
            if len(message_pair) != 2:
                raise TypeError(
                    f"Expected a list of lists of length 2 or list of tuples of length 2. Received: {message_pair}"
                )
            processed_messages.append(
                [
                    self._preprocess_chat_messages(message_pair[0]),
                    self._preprocess_chat_messages(message_pair[1]),
                ]
            )
        return processed_messages

    def _postprocess_chat_messages(
        self, chat_message: str | tuple | list | None
    ) -> str | dict | None:
        if chat_message is None:
            return None
        elif isinstance(chat_message, (tuple, list)):
            file_uri = str(chat_message[0])
            if utils.validate_url(file_uri):
                filepath = file_uri
            else:
                filepath = self.make_temp_copy_if_needed(file_uri)

            mime_type = client_utils.get_mimetype(filepath)
            return {
                "name": filepath,
                "mime_type": mime_type,
                "alt_text": chat_message[1] if len(chat_message) > 1 else None,
                "data": None,  # These last two fields are filled in by the frontend
                "is_file": True,
            }
        elif isinstance(chat_message, str):
            chat_message = inspect.cleandoc(chat_message)
            return chat_message
        else:
            raise ValueError(f"Invalid message for Chatbot component: {chat_message}")

    def postprocess(
        self,
        y: list[list[str | tuple[str] | tuple[str, str] | None] | tuple],
    ) -> list[list[str | dict | None]]:
        """
        Parameters:
            y: List of lists representing the message and response pairs. Each message and response should be a string, which may be in Markdown format.  It can also be a tuple whose first element is a string or pathlib.Path filepath or URL to an image/video/audio, and second (optional) element is the alt text, in which case the media file is displayed. It can also be None, in which case that message is not displayed.
        Returns:
            List of lists representing the message and response. Each message and response will be a string of HTML, or a dictionary with media information. Or None if the message is not to be displayed.
        """
        if y is None:
            return []
        processed_messages = []
        for message_pair in y:
            result = ""

            if message_pair[0] is not None:
                result += message_pair[0]
            if message_pair[1] is not None:
                result += message_pair[1]
            if not isinstance(message_pair, (tuple, list)):
                raise TypeError(
                    f"Expected a list of lists or list of tuples. Received: {message_pair}"
                )
            if len(message_pair) != 2:
                raise TypeError(
                    f"Expected a list of lists of length 2 or list of tuples of length 2. Received: {message_pair}"
                )
            result = result.replace('؛','\n').replace('/',' * ').replace('\u200c',' ').strip()
            result = result[:result.rfind('\n')] # filter till the end verse, do not publish not complete verses

            #print('Message Pairs: ',message_pair[0],message_pair[1], result)
            
            processed_messages.append(
                [
                    None,self._postprocess_chat_messages(result)
                    #self._postprocess_chat_messages(message_pair[1])),
                ]
            )
        return processed_messages

chatbot = Chatbot(label="PersianGPT",
                     rtl=True,
                     show_share_button=True,
                     show_copy_button=True,
                     #layout="panel",
                     bubble_full_width = False)

textbox = gr.Textbox(
            label="textbox",
            container=False,
            show_label=False,
            lines=3,
            scale=7,
            placeholder="...Type something here",
            rtl=True,
        )



class ChatInterface(gr.ChatInterface):
    def __init__(
        self,
        fn: Callable,
        *,
        chatbot: Chatbot | None = None,
        textbox: Textbox | None = None,
        additional_inputs: str | IOComponent | list[str | IOComponent] | None = None,
        additional_inputs_accordion_name: str = "Additional Inputs",
        examples: list[str] | None = None,
        cache_examples: bool | None = None,
        title: str | None = None,
        description: str | None = None,
        theme: Theme | str | None = None,
        css: str | None = None,
        analytics_enabled: bool | None = None,
        submit_btn: str | None | Button = "Submit",
        stop_btn: str | None | Button = "Stop",
        retry_btn: str | None | Button = "🔄  Retry",
        undo_btn: str | None | Button = "↩️ Undo",
        clear_btn: str | None | Button = "🗑️  Clear",
        autofocus: bool = True,
    ):
        """
        Parameters:
            fn: the function to wrap the chat interface around. Should accept two parameters: a string input message and list of two-element lists of the form [[user_message, bot_message], ...] representing the chat history, and return a string response. See the Chatbot documentation for more information on the chat history format.
            chatbot: an instance of the gr.Chatbot component to use for the chat interface, if you would like to customize the chatbot properties. If not provided, a default gr.Chatbot component will be created.
            textbox: an instance of the gr.Textbox component to use for the chat interface, if you would like to customize the textbox properties. If not provided, a default gr.Textbox component will be created.
            additional_inputs: an instance or list of instances of gradio components (or their string shortcuts) to use as additional inputs to the chatbot. If components are not already rendered in a surrounding Blocks, then the components will be displayed under the chatbot, in an accordion.
            additional_inputs_accordion_name: the label of the accordion to use for additional inputs, only used if additional_inputs is provided.
            examples: sample inputs for the function; if provided, appear below the chatbot and can be clicked to populate the chatbot input.
            cache_examples: If True, caches examples in the server for fast runtime in examples. The default option in HuggingFace Spaces is True. The default option elsewhere is False.
            title: a title for the interface; if provided, appears above chatbot in large font. Also used as the tab title when opened in a browser window.
            description: a description for the interface; if provided, appears above the chatbot and beneath the title in regular font. Accepts Markdown and HTML content.
            theme: Theme to use, loaded from gradio.themes.
            css: custom css or path to custom css file to use with interface.
            analytics_enabled: Whether to allow basic telemetry. If None, will use GRADIO_ANALYTICS_ENABLED environment variable if defined, or default to True.
            submit_btn: Text to display on the submit button. If None, no button will be displayed. If a Button object, that button will be used.
            stop_btn: Text to display on the stop button, which replaces the submit_btn when the submit_btn or retry_btn is clicked and response is streaming. Clicking on the stop_btn will halt the chatbot response. If set to None, stop button functionality does not appear in the chatbot. If a Button object, that button will be used as the stop button.
            retry_btn: Text to display on the retry button. If None, no button will be displayed. If a Button object, that button will be used.
            undo_btn: Text to display on the delete last button. If None, no button will be displayed. If a Button object, that button will be used.
            clear_btn: Text to display on the clear button. If None, no button will be displayed. If a Button object, that button will be used.
            autofocus: If True, autofocuses to the textbox when the page loads.
        """
        super().__init__(
            fn = fn,
            chatbot = chatbot,
            textbox= textbox,
            additional_inputs = additional_inputs,
            additional_inputs_accordion_name = additional_inputs_accordion_name,
            examples = examples,
            cache_examples = cache_examples,
            title = title,
            description = description,
            theme = theme,
            css = css,
            analytics_enabled = analytics_enabled,
            submit_btn = submit_btn,
            stop_btn = stop_btn,
            retry_btn = retry_btn,
            undo_btn = undo_btn,
            clear_btn = clear_btn,
            autofocus = autofocus,
        )

    async def _stream_fn(
        self,
        message: str,
        history_with_input: list[list[str | None]],
        request: Request,
        *args,
    ) -> AsyncGenerator:
        history = history_with_input[:-1]

        print(f'Message is {message}')
        if len(message)==0:
            message = random.choice(["ا","ب","پ","ت","ث","ج","چ","ح","خ","ل","م","ن","و",
                                    "د","ذ","ر","ز","ژ","س","ش","ص","ض","ط","ظ","ع","غ",
                                    "ف","ق","ه","ی",
                                   ])
        
        inputs, _, _ = special_args(
            self.fn, inputs=[message, history, *args], request=request
        )

        if self.is_async:
            generator = self.fn(*inputs)
        else:
            generator = await anyio.to_thread.run_sync(
                self.fn, *inputs, limiter=self.limiter
            )
            generator = SyncToAsyncIterator(generator, self.limiter)
        try:
            first_response = await async_iteration(generator)
            update = history + [[message, first_response]]
            yield update, update
        except StopIteration:
            update = history + [[message, None]]
            yield update, update
        async for response in generator:
            update = history + [[message, response]]
            yield update, update
        
    async def _submit_fn(
        self,
        message: str,
        history_with_input: list[list[str | None]],
        request: Request,
        *args,
    ) -> tuple[list[list[str | None]], list[list[str | None]]]:
        history = history_with_input[:-1]
        
        inputs, _, _ = special_args(
            self.fn, inputs=[message, history, *args], request=request
        )

        if self.is_async:
            response = await self.fn(*inputs)
        else:
            response = await anyio.to_thread.run_sync(
                self.fn, *inputs, limiter=self.limiter
            )

        history.append([message,response])

        return history, history

chat_interface = ChatInterface(
        generate, 
        chatbot=chatbot,
        textbox=textbox,
        #examples=examples,
        additional_inputs=additional_inputs,
        submit_btn = "Generate",
        stop_btn = None,
        retry_btn = None,
        undo_btn = None,
        clear_btn = None,
        cache_examples=False,
    ) 

def vote(data: gr.LikeData):
    if data.liked:
        print("You upvoted this response: " + data.value)
    else:
        print("You downvoted this response: " + data.value)

    with open(DATA_FILE, "a") as jsonlfile:
        json_data = [
            json.dumps(
                    {
                        "time_stamp": time.time(),
                        "model_version":MODEL_VERSION,
                        "name":NAME,
                        "prompt": PROMPT.replace('\n','؛').replace('\t','/').replace(' * ','/').replace('\u200c',' ').strip(),
                        "system prompt": SYSTEM_PROMPT,
                        "temperature": TEMPERATURE,
                        "max_new_tokens": MAX_NEW_TOKENS,
                        "top_p": TOP_P,
                        "top_k": TOP_K,
                        "repetition_penalty": REPETITION_PENALTY,
                        "response": data.value.replace('\n','؛').replace('\t','/').replace(' * ','/').replace('\u200c',' ').strip(),
                        "label": data.liked,
                    }, ensure_ascii=False
            )
        ]
        print(f'we are writing this: {json_data}')
        jsonlfile.write("\n".join(json_data) + "\n")

    
with gr.Blocks(css=CSS) as demo:
    chatbot.like(vote, None, None)
    chat_interface.render()

demo.queue(concurrency_count=100, api_open=False).launch(show_api=False) #, share=True)