Stable-Zephyr-3b と OpenVINO を使用した LLM 搭載チャットボット#

from pathlib import Path 
import sys 

genai_llm_bench = Path("openvino.genai/llm_bench/python") 

if not genai_llm_bench.exists():
     !git clone https://github.com/openvinotoolkit/openvino.genai.git 

sys.path.append(str(genai_llm_bench))

%pip install -q "transformers>=4.38.2" 
%pip install -q --extra-index-url https://download.pytorch.org/whl/cpu -r ./openvino.genai/llm_bench/python/requirements.txt 
%pip install --pre -Uq openvino openvino-tokenizers[transformers] --extra-index-url https://storage.openvinotoolkit.org/simple/wheels/nightly 
%pip install -q "gradio>=4.19"

stateful_model_path = Path("stable-zephyr-3b-stateful/pytorch/dldt/compressed_weights/OV_FP16-4BIT_DEFAULT") 

convert_script = genai_llm_bench / "convert.py" 

if not (stateful_model_path / "openvino_model.xml").exists():
     !python $convert_script --model_id stabilityai/stable-zephyr-3b --precision FP16 --compress_weights 4BIT_DEFAULT --output stable-zephyr-3b-stateful --force_convert

import ipywidgets as widgets 
import openvino as 

ov core = ov.Core() 

device = widgets.Dropdown( 
    options=core.available_devices, 
    value="CPU", 
    description="Device:", 
    disabled=False, 
) 

device

benchmark_script = genai_llm_bench / "benchmark.py" 

!python $benchmark_script -m $stateful_model_path -ic 512 -p "Tell me story about cats" -d $device.value

from optimum.intel.openvino import OVModelForCausalLM 
from transformers import AutoConfig 

ov_model = OVModelForCausalLM.from_pretrained( 
    stateful_model_path, 
    config=AutoConfig.from_pretrained(stateful_model_path, trust_remote_code=True), 
    device=device.value, 
)

import torch 
from threading import Event, Thread 
from uuid import uuid4 
from typing import List, Tuple 
import gradio as gr 
from transformers import ( 
    AutoTokenizer, 
    StoppingCriteria, 
    StoppingCriteriaList, 
    TextIteratorStreamer, 
) 

model_name = "stable-zephyr-3b" 

tok = AutoTokenizer.from_pretrained(stateful_model_path) 

DEFAULT_SYSTEM_PROMPT = """\ 
You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure that your responses are socially unbiased and positive in nature. If a question does not make any sense or is not factually coherent, explain why instead of answering something not correct. If you don't know the answer to a question, please don't share false information.\ 
""" 

model_configuration = { 
    "start_message": f"<|system|>\n {DEFAULT_SYSTEM_PROMPT }<|endoftext|>", 
    "history_template": "<|user|>\n{user}<|endoftext|><|assistant|>\n{assistant}<|endoftext|>", 
    "current_message_template": "<|user|>\n{user}<|endoftext|><|assistant|>\n{assistant}", 
} 
history_template = model_configuration["history_template"] 
current_message_template = model_configuration["current_message_template"] 
start_message = model_configuration["start_message"] 
stop_tokens = model_configuration.get("stop_tokens") 
tokenizer_kwargs = model_configuration.get("tokenizer_kwargs", {}) 

examples = [ 
    ["Hello there! How are you doing?"], 
    ["What is OpenVINO?"], 
    ["Who are you?"], 
    ["Can you explain to me briefly what is Python programming language?"], 
    ["Explain the plot of Cinderella in a sentence."], 
    ["What are some common mistakes to avoid when writing code?"], 
    ["Write a 100-word blog post on “Benefits of Artificial Intelligence and OpenVINO“"], 
] 

max_new_tokens = 256 

class StopOnTokens(StoppingCriteria): 
    def __init__(self, token_ids): 
        self.token_ids = token_ids 

    def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool: 
        for stop_id in self.token_ids: 
            if input_ids[0][-1] == stop_id: 
                return True 
        return False 

if stop_tokens is not None: 
    if isinstance(stop_tokens[0], str): 
        stop_tokens = tok.convert_tokens_to_ids(stop_tokens) 
    stop_tokens = [StopOnTokens(stop_tokens)] 

def default_partial_text_processor(partial_text: str, new_text: str): 
    """ 
    helper for updating partially generated answer, used by default 

    Params: 
        partial_text: text buffer for storing previosly generated text 
        new_text: text update for the current step 
    Returns: 
        updated text string 

    """ 
    partial_text += new_text 
    return partial_text 

text_processor = model_configuration.get("partial_text_processor", default_partial_text_processor) 

def convert_history_to_text(history: List[Tuple[str, str]]): 
    """ 
    function for conversion history stored as list pairs of user and assistant messages to string according to model expected conversation template 
    Params: 
        history: dialogue history 
    Returns: 
        history in text format 
    """ 
    text = start_message + "".join(["".join([history_template.format(num=round, user=item[0], assistant=item[1])]) for round, item in enumerate(history[:-1])]) 
    text += "".join( 
        [ 
            "".join( 
                [ 
                    current_message_template.format( 
                        num=len(history) + 1, 
                        user=history[-1][0], 
                        assistant=history[-1][1], 
                    ) 
                ] 
            ) 
        ] 
    ) 
    return text 

def user(message, history): 
    """ 
    callback function for updating user messages in interface on submit button click 

    Params: 
        message: current message 
        history: conversation history 
    Returns:
        None 
    """ 
    # ユーザーのメッセージを会話履歴に追加 
    return "", history + [[message, ""]] 

def bot(history, temperature, top_p, top_k, repetition_penalty, conversation_id): 
    """ 
    callback function for running chatbot on submit button click 

    Params: 
        history: conversation history 
        temperature: parameter for control the level of creativity in AI-generated text.
                     By adjusting the `temperature`, you can influence the AI model's probability distribution, making the text more focused or diverse. 
        top_p: parameter for control the range of tokens considered by the AI model based on their cumulative probability. 
        top_k: parameter for control the range of tokens considered by the AI model based on their cumulative probability, selecting number of tokens with highest probability. 
        repetition_penalty: parameter for penalizing tokens based on how frequently they occur in the text. 
        conversation_id: unique conversation identifier.
    """ 

    # 現在のシステムメッセージと会話履歴を連結して、モデルの入力メッセージ文字列を構築 
    messages = convert_history_to_text(history) 

    # メッセージの文字列をトークン化 
    input_ids = tok(messages, return_tensors="pt", **tokenizer_kwargs).input_ids 
    if input_ids.shape[1] > 2000: 
        history = [history[-1]] 
        messages = convert_history_to_text(history) 
        input_ids = tok(messages, return_tensors="pt", **tokenizer_kwargs).input_ids 
    streamer = TextIteratorStreamer(tok, timeout=30.0, skip_prompt=True, skip_special_tokens=True) 
    generate_kwargs = dict( 
        input_ids=input_ids, 
        max_new_tokens=max_new_tokens, 
        temperature=temperature, 
        do_sample=temperature > 0.0, 
        top_p=top_p, 
        top_k=top_k, 
        repetition_penalty=repetition_penalty, 
        streamer=streamer, 
    ) 
    if stop_tokens is not None: 
        generate_kwargs["stopping_criteria"] = StoppingCriteriaList(stop_tokens) 

    stream_complete = Event() 

def generate_and_signal_complete(): 
    """ 
    genration function for single thread 
    """ 
    global start_time 
    ov_model.generate(**generate_kwargs) 
    stream_complete.set() 

t1 = Thread(target=generate_and_signal_complete) 
t1.start() 

# 生成されたテキストを保存するために空の文字列を初期化 
partial_text = "" 
for new_text in streamer: 
    partial_text = text_processor(partial_text, new_text) 
    history[-1][1] = partial_text 
    yield history 

def get_uuid(): 
    """ 
    universal unique identifier for thread 
    """ 
    return str(uuid4()) 

with gr.Blocks( 
    theme=gr.themes.Soft(), 
    css=".disclaimer {font-variant-caps: all-small-caps;}", 
) as demo: 
    conversation_id = gr.State(get_uuid) 
    gr.Markdown(f"""<h1><center>OpenVINO {model_name} Chatbot</center></h1>""") 
    chatbot = gr.Chatbot(height=500) 
    with gr.Row(): 
        with gr.Column(): 
            msg = gr.Textbox( 
                label="Chat Message Box", 
                placeholder="Chat Message Box", 
                show_label=False, 
                container=False, 
            ) 
        with gr.Column(): 
            with gr.Row(): 
                submit = gr.Button("Submit") 
                stop = gr.Button("Stop") 
                clear = gr.Button("Clear") 
    with gr.Row(): 
        with gr.Accordion("Advanced Options:", open=False): 
            with gr.Row(): 
                with gr.Column(): 
                    with gr.Row(): 
                        temperature = gr.Slider( 
                            label="Temperature", 
                            value=0.1, 
                            minimum=0.0, 
                            maximum=1.0, 
                            step=0.1, 
                            interactive=True, 
                            info="Higher values produce more diverse outputs", 
                        ) 
                with gr.Column(): 
                    with gr.Row(): 
                        top_p = gr.Slider( 
                            label="Top-p (nucleus sampling)", 
                            value=1.0, 
                            minimum=0.0, 
                            maximum=1, 
                            step=0.01, 
                            interactive=True, 
                            info=( 
                                "Sample from the smallest possible set of tokens whose cumulative probability " 
                                "exceeds top_p.Set to 1 to disable and sample from all tokens."                             ), 
                        ) 
                with gr.Column(): 
                    with gr.Row(): 
                        top_k = gr.Slider( 
                            label="Top-k", 
                            value=50, 
                            minimum=0.0, 
                            maximum=200, 
                            step=1, 
                            interactive=True, 
                            info="Sample from a shortlist of top-k tokens — 0 to disable and sample from all tokens.", 
                        ) 
                with gr.Column(): 
                    with gr.Row(): 
                        repetition_penalty = gr.Slider( 
                            label="Repetition Penalty", 
                            value=1.1, 
                            minimum=1.0, 
                            maximum=2.0, 
                            step=0.1, 
                            interactive=True, 
                            info="Penalize repetition — 1.0 to disable.", 
                        ) 
    gr.Examples(examples, inputs=msg, label="Click on any example and press the 'Submit' button") 

    submit_event = msg.submit( 
        fn=user, 
        inputs=[msg, chatbot], 
        outputs=[msg, chatbot], 
        queue=False, 
    ).then( 
        fn=bot, 
        inputs=[ 
            chatbot, 
            temperature, 
            top_p, 
            top_k, 
            repetition_penalty, 
            conversation_id, 
        ], 
        outputs=chatbot, 
        queue=True, 
    ) 
    submit_click_event = submit.click( 
        fn=user, 
        inputs=[msg, chatbot], 
        outputs=[msg, chatbot], 
        queue=False, 
    ).then( 
        fn=bot, 
        inputs=[ 
            chatbot, 
            temperature, 
            top_p, 
            top_k, 
            repetition_penalty, 
            conversation_id, 
        ], 
        outputs=chatbot, 
        queue=True, 
    ) 
    stop.click( 
        fn=None, 
        inputs=None, 
        outputs=None, 
        cancels=[submit_event, submit_click_event], 
        queue=False, 
    ) 
    clear.click(lambda: None, None, chatbot, queue=False) 

demo.queue(max_size=2) 
# リモートで起動する場合は、server_name と server_port を指定 
# demo.launch(server_name='your server name', server_port='server port in int') 
# プラットフォーム上で起動する際に問題がある場合は、起動メソッドに share=True を渡すことができます: 
# demo.launch(share=True) 
# インターフェイスの公開共有可能なリンクを作成。詳細はドキュメントをご覧ください: https://gradio.app/docs/ 
demo.launch(share=True)