OpenVINO と LangChain を使用して ReAct エージェントを作成#

from pathlib import Path 

model_id = "mistralai/Mistral-7B-Instruct-v0.3" 
model_path = "Mistral-7B-Instruct-v0.3-ov-int4" 

if not Path(model_path).exists():
     !optimum-cli export openvino --model {model_id} --task text-generation-with-past --trust-remote-code --weight-format int4 {model_path}

import openvino as ov 
import ipywidgets as widgets 

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

from langchain_community.llms.huggingface_pipeline import HuggingFacePipeline 
from transformers.generation.stopping_criteria import StoppingCriteriaList, StoppingCriteria 

class StopSequenceCriteria(StoppingCriteria): 
    """ 
    This class can be used to stop generation whenever a sequence of tokens is encountered.

    Args: 
        stop_sequences (`str` or `List[str]`):
             The sequence (or list of sequences) on which to stop execution. 
        tokenizer: 
             The tokenizer used to decode the model outputs.
     """ 

    def __init__(self, stop_sequences, tokenizer): 
        if isinstance(stop_sequences, str): 
            stop_sequences = [stop_sequences] 
        self.stop_sequences = stop_sequences 
        self.tokenizer = tokenizer 

    def __call__(self, input_ids, scores, **kwargs) -> bool: 
        decoded_output = self.tokenizer.decode(input_ids.tolist()[0]) 
        return any(decoded_output.endswith(stop_sequence) for stop_sequence in self.stop_sequences) 

ov_config = {"PERFORMANCE_HINT": "LATENCY", "NUM_STREAMS": "1", "CACHE_DIR": ""} 
stop_tokens = ["Observation:"] 

ov_llm = HuggingFacePipeline.from_model_id( 
    model_id=model_path, 
    task="text-generation", 
    backend="openvino", 
    model_kwargs={ 
        "device": device.value, 
        "ov_config": ov_config, 
        "trust_remote_code": True, 
    }, 
    pipeline_kwargs={"max_new_tokens": 2048}, 
) 
ov_llm = ov_llm.bind(skip_prompt=True, stop=["Observation:"]) 

tokenizer = ov_llm.pipeline.tokenizer 
ov_llm.pipeline._forward_params["stopping_criteria"] = 
StoppingCriteriaList([StopSequenceCriteria(stop_tokens, tokenizer)])

ov_config = { 
    "KV_CACHE_PRECISION": "u8", 
    "DYNAMIC_QUANTIZATION_GROUP_SIZE": "32", 
    "PERFORMANCE_HINT": "LATENCY", 
    "NUM_STREAMS": "1", 
    "CACHE_DIR": "", 
}

from langchain_community.tools import WikipediaQueryRun 
from langchain_community.utilities import WikipediaAPIWrapper 
from langchain_core.pydantic_v1 import BaseModel, Field 
from langchain_core.callbacks import CallbackManagerForToolRun 
from typing import Optional 

class WikipediaQueryRunWrapper(WikipediaQueryRun): 
    def _run( 
        self, 
        text: str, 
        run_manager: Optional[CallbackManagerForToolRun] = None, 
    ) -> str: 
        """Use the Wikipedia tool.""" 
        return self.api_wrapper.run(text) 

api_wrapper = WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=1000) 

class WikiInputs(BaseModel): 
    """inputs to the wikipedia tool.""" 

    text: str = Field(description="query to look up on wikipedia.") 

wikipedia = WikipediaQueryRunWrapper( 
    description="A wrapper around Wikipedia. Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects. Input should be a search query.", 
    args_schema=WikiInputs, 
    api_wrapper=api_wrapper, 
)

wikipedia.invoke({"text": "OpenVINO"})

import urllib.parse 
import json5 

@tool 
def painting(prompt: str) -> str: 
    """ 
    AI painting (image generation) service, input text description, and return the image URL drawn based on text information.
    """ 
    prompt = urllib.parse.quote(prompt) 
    return json5.dumps({"image_url": f"https://image.pollinations.ai/prompt/{prompt}"}, ensure_ascii=False) 

painting.invoke({"prompt": "a cat"})

@tool 
def weather( 
    city_name: str, 
) -> str: 
    """ 
    Get the current weather for `city_name` 
    """ 

    if not isinstance(city_name, str): 
        raise TypeError("City name must be a string") 

    key_selection = { 
        "current_condition": [ 
            "temp_C", 
            "FeelsLikeC", 
            "humidity", 
            "weatherDesc", 
            "observation_time", 
        ], 
    } 
    import requests 

    resp = requests.get(f"https://wttr.in/{city_name}?format=j1") 
    resp.raise_for_status() 
    resp = resp.json() 
    ret = {k: {_v: resp[k][0][_v] for _v in v} for k, v in key_selection.items()} 

    return str(ret) 

weather.invoke({"city_name": "London"})

tools = [wikipedia, painting, weather] 

agent = StructuredChatAgent.from_llm_and_tools( 
    ov_llm, 
    tools, 
    prefix=PREFIX, 
    suffix=SUFFIX, 
    human_message_template=HUMAN_MESSAGE_TEMPLATE, 
    format_instructions=FORMAT_INSTRUCTIONS, 
) 
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

import gradio as gr 

examples = [ 
    ["Based on current weather in London, show me a picture of Big Ben through its URL"], 
    ["What is OpenVINO ?"], 
    ["Create an image of pink cat and return its URL"], 
    ["How many people live in Canada ?"], 
    ["What is the weather like in New York now ?"], 
] 

def partial_text_processor(partial_text, new_text): 
    """ 
    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 

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): 
    """ 
    callback function for running chatbot on submit button click 

    Params: 
        history: conversation history 

    """ 
    partial_text = "" 

    for new_text in agent_executor.stream( 
        {"input": history[-1][0]}, 
    ): 
        if "output" in new_text.keys(): 
            partial_text = partial_text_processor(partial_text, new_text["output"]) 
            history[-1][1] = partial_text 
            yield history 

def request_cancel(): 
    ov_llm.pipeline.model.request.cancel() 

with gr.Blocks( 
    theme=gr.themes.Soft(), 
    css=".disclaimer {font-variant-caps: all-small-caps;}", 
) as demo: 
    names = [tool.name for tool in tools] 
    gr.Markdown(f"""<h1><center>OpenVINO Agent for {str(names)}</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") 
    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, 
    ], 
    outputs=chatbot, 
    queue=True, 
) 
submit_click_event = submit.click( 
    fn=user, 
    inputs=[msg, chatbot], 
    outputs=[msg, chatbot], 
    queue=False, 
).then( 
    fn=bot, 
    inputs=[ 
        chatbot, 
    ], 
    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()