OpenVINO を使用して LLM によるチャットボットを作成#

import os 

os.environ["GIT_CLONE_PROTECTION_ACTIVE"] = "false" 

%pip install -Uq pip 
%pip uninstall -q -y optimum optimum-intel 
%pip install --pre -Uq openvino openvino-tokenizers[transformers] --extra-index-url https://storage.openvinotoolkit.org/simple/wheels/nightly 
%pip install -q --extra-index-url https://download.pytorch.org/whl/cpu\ 
"git+https://github.com/huggingface/optimum-intel.git"\ 
"git+https://github.com/openvinotoolkit/nncf.git"\ 
"torch>=2.1"\ 
"datasets"\ 
"accelerate"\ 
"gradio>=4.19"\ 
"onnx" "einops" "transformers_stream_generator" "tiktoken" "transformers>=4.40" "bitsandbytes"

import os 
from pathlib import Path 
import requests 
import shutil 

# モデル設定を取得 

config_shared_path = Path("../../utils/llm_config.py") 
config_dst_path = Path("llm_config.py") 

if not config_dst_path.exists(): 
    if config_shared_path.exists(): 
        try: 
            os.symlink(config_shared_path, config_dst_path) 
        except Exception: 
            shutil.copy(config_shared_path, config_dst_path) 
    else: 
        r = requests.get(url="https://raw.githubusercontent.com/openvinotoolkit/openvino_notebooks/latest/utils/llm_config.py") 
        with open("llm_config.py", "w", encoding="utf-8") as f: 
            f.write(r.text) 
elif not os.path.islink(config_dst_path): 
    print("LLM config will be updated") 
    if config_shared_path.exists(): 
        shutil.copy(config_shared_path, config_dst_path) 
    else: 
        r = requests.get(url="https://raw.githubusercontent.com/openvinotoolkit/openvino_notebooks/latest/utils/llm_config.py") 
        with open("llm_config.py", "w", encoding="utf-8") as f: 
            f.write(r.text)

from llm_config import SUPPORTED_LLM_MODELS 
import ipywidgets as widgets

model_languages = list(SUPPORTED_LLM_MODELS) 

model_language = widgets.Dropdown( 
    options=model_languages, 
    value=model_languages[0], 
    description="Model Language:", 
    disabled=False, 
) 

model_language

model_ids = list(SUPPORTED_LLM_MODELS[model_language.value]) 

model_id = widgets.Dropdown( 
    options=model_ids, 
    value=model_ids[0], 
    description="Model:", 
    disabled=False, 
) 

model_id

model_configuration = SUPPORTED_LLM_MODELS[model_language.value][model_id.value] 
print(f"Selected model {model_id.value}")

import openvino as ov 

core = ov.Core() 

support_devices = core.available_devices 
if "NPU" in support_devices: 
    support_devices.remove("NPU") 

device = widgets.Dropdown( 
    options=support_devices + ["AUTO"], 
    value="CPU", 
    description="Device:", 
    disabled=False, 
) 

device

available_models = [] 
if int4_model_dir.exists(): 
    available_models.append("INT4") 
if int8_model_dir.exists(): 
    available_models.append("INT8") 
if fp16_model_dir.exists(): 
    available_models.append("FP16") 

model_to_run = widgets.Dropdown( 
    options=available_models, 
    value=available_models[0], 
    description="Model to run:", 
    disabled=False, 
) 

model_to_run

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

if model_to_run.value == "INT4": 
    model_dir = int4_model_dir 
elif model_to_run.value == "INT8": 
    model_dir = int8_model_dir 
else: 
    model_dir = fp16_model_dir 
print(f"Loading model from {model_dir}") 

ov_config = {"PERFORMANCE_HINT": "LATENCY", "NUM_STREAMS": "1", "CACHE_DIR": ""} 

    if "GPU" in device.value and "qwen2-7b-instruct" in model_id.value: 
        ov_config["GPU_ENABLE_SDPA_OPTIMIZATION"] = "NO" 

# GPU デバイスでは、モデルは FP16 精度で実行されます。red-pajama-3b-chat モデルでは、これが原因で精度の問題が発生することがありますが、 
# 精度ヒントを "f32" に設定することでこれを回避できます 
if model_id.value == "red-pajama-3b-chat" and "GPU" in core.available_devices and device.value in ["GPU", "AUTO"]: 
    ov_config["INFERENCE_PRECISION_HINT"] = "f32" 

model_name = model_configuration["model_id"] 
tok = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) 

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

tokenizer_kwargs = model_configuration.get("tokenizer_kwargs", {}) 
test_string = "2 + 2 =" 
input_tokens = tok(test_string, return_tensors="pt", **tokenizer_kwargs) 
answer = ov_model.generate(**input_tokens, max_new_tokens=2) 
print(tok.batch_decode(answer, skip_special_tokens=True)[0])

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 = model_configuration["model_id"] 
start_message = model_configuration["start_message"] 
history_template = model_configuration.get("history_template") 
current_message_template = model_configuration.get("current_message_template") 
stop_tokens = model_configuration.get("stop_tokens") 
tokenizer_kwargs = model_configuration.get("tokenizer_kwargs", {}) 

chinese_examples = [ 
    ["你好!"], 
    ["你是谁?"], 
    ["请介绍一下上海"], 
    ["请介绍一下英特尔公司"], 
    ["晚上睡不着怎么办？"], 
    ["给我讲一个年轻人奋斗创业最终取得成功的故事。"], 
    ["给这个故事起一个标题。"], 
] 

english_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“"], 
] 

japanese_examples = [ 
    ["こんにちは！調子はどうですか?"], 
    ["OpenVINO とは何ですか?"], 
    ["あなたは誰ですか?"], 
    ["Python プログラミング言語とは何か簡単に説明してもらえますか?"], 
    ["シンデレラのあらすじを一文で説明してください。"], 
    ["コードを書くときに避けるべきよくある間違いは何ですか?"], 
    ["人工知能と「OpenVINO の利点」について 100 語程度のブログ記事を書いてください。"], 
] 

examples = chinese_examples if (model_language.value == "Chinese") else japanese_examples if (model_language.value == "Japanese") else english_examples 

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_token(history: List[Tuple[str, str]]): 
    """ 
    function for conversion history stored as list pairs of user and assistant messages to tokens according to model expected conversation template 
    Params: 
        history: dialogue history 
    Returns: 
        history in token format 
    """ 
    if pt_model_name == "baichuan2": 
        system_tokens = tok.encode(start_message) 
        history_tokens = [] 
        for old_query, response in history[:-1]: 
            round_tokens = [] 
            round_tokens.append(195) 
            round_tokens.extend(tok.encode(old_query)) 
            round_tokens.append(196) 
            round_tokens.extend(tok.encode(response)) 
            history_tokens = round_tokens + history_tokens input_tokens = system_tokens + history_tokens 
        input_tokens.append(195) 
        input_tokens.extend(tok.encode(history[-1][0])) 
        input_tokens.append(196) 
        input_token = torch.LongTensor([input_tokens]) 
    elif history_template is None: 
        messages = [{"role": "system", "content": start_message}] 
        for idx, (user_msg, model_msg) in enumerate(history): 
            if idx == len(history) - 1 and not model_msg: 
                messages.append({"role": "user", "content": user_msg}) 
                break 
            if user_msg: 
                messages.append({"role": "user", "content": user_msg}) 
            if model_msg: 
                messages.append({"role": "assistant", "content": model_msg}) 

        input_token = tok.apply_chat_template(messages, add_generation_prompt=True, tokenize=True, return_tensors="pt") 
    else: 
        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], 
                        ) 
                    ] 
                ) 
            ] 
        ) 
        input_token = tok(text, return_tensors="pt", **tokenizer_kwargs).input_ids 
    return input_token 

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.
    """ 

    # 現在のシステムメッセージと会話履歴を連結して、モデルの入力メッセージ文字列を構築 
    # メッセージの文字列をトークン化 
    input_ids = convert_history_to_token(history) 
    if input_ids.shape[1] > 2000: 
        history = [history[-1]] 
        input_ids = convert_history_to_token(history) 
    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 request_cancel(): ov_model.request.cancel() 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_id.value} 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=request_cancel, 
        inputs=None, 
        outputs=None, 
        cancels=[submit_event, submit_click_event], 
        queue=False, 
    ) 
    clear.click(lambda: None, None, chatbot, queue=False) 

# リモートで起動する場合は、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()

# gradio インターフェイスを停止するには、このセルのコメントを外して実行します 
# demo.close()