OpenVINO による文法エラー修正#

%pip install -q "torch>=2.1.0" "git+https://github.com/huggingface/optimum-intel.git" "openvino>=2024.0.0" onnx tqdm "gradio>=4.19" "transformers>=4.33.0" --extra-index-url https://download.pytorch.org/whl/cpu 
%pip install -q "nncf>=2.9.0" datasets jiwer

from pathlib import Path 
from transformers import pipeline, AutoTokenizer 
from optimum.intel.openvino import OVModelForSeq2SeqLM, OVModelForSequenceClassification

import re 
import transformers 
from tqdm.notebook import tqdm 

def split_text(text: str) -> list: 
    """ 
    Split a string of text into a list of sentence batches.
    Parameters: 
    text (str): The text to be split into sentence batches. 
    Returns: 
    list: A list of sentence batches. Each sentence batch is a list of sentences.
    """ 
    # 正規表現を使用してテキストを文に分割 
    sentences = re.split(r"(?<=[^A-Z].[.?])+(?=[A-Z])", text) 

    # 文のバッチを保存するリストを初期化 
    sentence_batches = [] 

    # 現在の文のバッチを保存する一時リストを初期化 
    temp_batch = [] 

    # 文章を繰り返し 
    for sentence in sentences:
        # 一時バッチに文を追加 
        temp_batch.append(sentence) 

        # 一時バッチの長さが 2 から 3 文の間、または最後のバッチである場合は、それを文バッチのリストに追加 
        if len(temp_batch) >= 2 and len(temp_batch) <= 3 or sentence == sentences[-1]: 
            sentence_batches.append(temp_batch) 
            temp_batch = [] 

    return sentence_batches 

def correct_text( 
    text: str, 
    checker: transformers.pipelines.Pipeline, 
    corrector: transformers.pipelines.Pipeline, 
    separator: str = " ", 
) -> str: 
   """ 
    Correct the grammar in a string of text using a text-classification and text-generation pipeline.

    Parameters: 
    text (str): The inpur text to be corrected. 
    checker (transformers.pipelines.Pipeline): The text-classification pipeline to use for checking the grammar quality of the text. 
    corrector (transformers.pipelines.Pipeline): The text-generation pipeline to use for correcting the text. 
    separator (str, optional): The separator to use when joining the corrected text into a single string. Default is a space character.
 
    Returns: 
    str: The corrected text.
    """ 
    # テキストを文のバッチに分割 
    sentence_batches = split_text(text) 

    # 修正したテキストを保存するリストを初期化 
    corrected_text = [] 

    # 文のバッチを反復処理 
    for batch in tqdm(sentence_batches, total=len(sentence_batches), desc="correcting text.."):
        # バッチ内の文を 1 つの文字列に結合 
        raw_text = " ".join(batch) 

        # テキスト分類パイプラインを使用してテキストの文法品質をチェック 
        results = checker(raw_text) 

        # テキスト分類の結果が LABEL_1 でないか、スコアが 0.9 未満の LABEL_1 である場合にのみテキストを修正 
        if results[0]["label"] != "LABEL_1" or (results[0]["label"] == "LABEL_1" and results[0]["score"] < 0.9):
            # テキスト生成パイプラインを使用してテキストを修正 
            corrected_batch = corrector(raw_text) 
            corrected_text.append(corrected_batch[0]["generated_text"]) 
        else: 
            corrected_text.append(raw_text) 

    # 修正したテキストを 1 つの文字列に結合 
    corrected_text = separator.join(corrected_text) 

    return corrected_text

default_text = ( 
    "Most of the course is about semantic or content of language but there are also interesting" 
    " topics to be learned from the servicefeatures except statistics in characters in documents.At" 
    " this point, He introduces herself as his native English speaker and goes on to say that if" 
    " you contine to work on social scnce" 
) 

corrected_text = correct_text(default_text, grammar_checker_pipe, grammar_corrector_pipe)

print(f"input text: {default_text}\n") 
print(f"generated text: {corrected_text}")

to_quantize = widgets.Checkbox( 
    value=True, 
    description="Quantization", 
    disabled=False, 
) 

to_quantize

import gradio as gr 
import time 

def correct(text, quantized, progress=gr.Progress(track_tqdm=True)): 
    grammar_corrector = grammar_corrector_pipe_int8 if quantized else grammar_corrector_pipe 

    start_time = time.perf_counter() 
    corrected_text = correct_text(text, grammar_checker_pipe, grammar_corrector) 
    end_time = time.perf_counter() 

    return corrected_text, f"{end_time - start_time:.2f}" 

def create_demo_block(quantized: bool, show_model_type: bool): 
    model_type = (" optimized" if quantized else " original") if show_model_type else "" 
    with gr.Row(): 
        gr.Markdown(f"## Run{model_type} grammar correction pipeline") 
    with gr.Row(): 
        with gr.Column(): 
            input_text = gr.Textbox(label="Text") 
        with gr.Column(): 
            output_text = gr.Textbox(label="Correction") 
            correction_time = gr.Textbox(label="Time (seconds)") 
    with gr.Row(): 
        gr.Examples(examples=[default_text], inputs=[input_text]) 
    with gr.Row(): 
        button = gr.Button(f"Run{model_type}") 
        button.click( 
            correct, 
            inputs=[input_text, gr.Number(quantized, visible=False)], 
            outputs=[output_text, correction_time], 
        ) 

with gr.Blocks() as demo: 
    gr.Markdown("# Interactive demo") 
    quantization_is_present = grammar_corrector_pipe_int8 is not None 
    create_demo_block(quantized=False, show_model_type=quantization_is_present) 
    if quantization_is_present: 
        create_demo_block(quantized=True, show_model_type=True) 

# リモートで起動する場合は、server_name と server_port を指定 
# demo.launch(server_name='your server name', server_port='server port in int') 
# 詳細はドキュメントをご覧ください: https://gradio.app/docs/ 
try: 
    demo.queue().launch(debug=False) 
except Exception: 
    demo.queue().launch(share=True, debug=False)