Spaces:
Running
Running
import gradio as gr | |
import os | |
import spaces | |
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer | |
from threading import Thread | |
# Set an environment variable | |
HF_TOKEN = os.environ.get("HF_TOKEN", None) | |
DESCRIPTION = ''' | |
<div> | |
<h1 style="text-align: center;">Mistral 7B Instruct v0.3</h1> | |
</div> | |
''' | |
# Load the tokenizer and model | |
tokenizer = AutoTokenizer.from_pretrained("mistralai/Mistral-7B-Instruct-v0.3") | |
model = AutoModelForCausalLM.from_pretrained("mistralai/Mistral-7B-Instruct-v0.3", device_map="auto") | |
terminators = [ | |
tokenizer.eos_token_id, | |
tokenizer.convert_tokens_to_ids("<|eot_id|>") | |
] | |
def chat_mistral7b_v0dot3(message: str, | |
history: list, | |
temperature: float, | |
max_new_tokens: int | |
) -> str: | |
""" | |
Generate a streaming response using the mistralai/Mistral-7B-Instruct-v0.3 model. | |
Args: | |
message (str): The input message. | |
history (list): The conversation history used by ChatInterface. | |
temperature (float): The temperature for generating the response. | |
max_new_tokens (int): The maximum number of new tokens to generate. | |
Returns: | |
str: The generated response. | |
""" | |
conversation = [] | |
for user, assistant in history: | |
conversation.extend([{"role": "user", "content": user}, {"role": "assistant", "content": assistant}]) | |
conversation.append({"role": "user", "content": message}) | |
input_ids = tokenizer.apply_chat_template(conversation, return_tensors="pt").to(model.device) | |
streamer = TextIteratorStreamer(tokenizer, timeout=10.0, skip_prompt=True, skip_special_tokens=True) | |
generate_kwargs = dict( | |
input_ids= input_ids, | |
streamer=streamer, | |
max_new_tokens=max_new_tokens, | |
do_sample=True, | |
temperature=temperature, | |
eos_token_id=terminators, | |
) | |
# This will enforce greedy generation (do_sample=False) when the temperature is passed 0, avoiding the crash. | |
if temperature == 0: | |
generate_kwargs['do_sample'] = False | |
t = Thread(target=model.generate, kwargs=generate_kwargs) | |
t.start() | |
outputs = [] | |
for text in streamer: | |
outputs.append(text) | |
yield "".join(outputs) | |
with gr.Blocks() as demo: | |
gr.Interface( | |
fn=chat_mistral7b_v0dot3, | |
inputs=[gr.Textbox(), gr.Textbox(), gr.Number(), gr.Number()], | |
outputs=[gr.Textbox()] | |
) | |
if __name__ == "__main__": | |
demo.launch() |