OpenVINO と Qwen-Agent を使用して関数呼び出しエージェントを作成#

import json 

def get_current_weather(location, unit="fahrenheit"): 
    """Get the current weather in a given location""" 
    if "tokyo" in location.lower(): 
        return json.dumps({"location": "Tokyo", "temperature": "10", "unit": "celsius"}) 
    elif "san francisco" in location.lower(): 
        return json.dumps({"location": "San Francisco", "temperature": "72", "unit": "fahrenheit"}) 
    elif "paris" in location.lower(): 
        return json.dumps({"location": "Paris", "temperature": "22", "unit": "celsius"}) 
    else: 
        return json.dumps({"location": location, "temperature": "unknown"})

functions = [ 
    { 
        "name": "get_current_weather", 
        "description": "Get the current weather in a given location", 
        "parameters": { 
            "type": "object", 
            "properties": { 
                "location": { 
                    "type": "string", 
                    "description": "The city and state, e.g. San Francisco, CA", 
                }, 
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}, 
            }, 
            "required": ["location"], 
        }, 
    } 
]

from pathlib import Path 

model_id = "Qwen/Qwen2-7B-Instruct" 
model_path = "Qwen2-7B-Instruct-ov" 

if not Path(model_path).exists():
     !optimum-cli export openvino --model {model_id} --task text-generation-with-past --trust-remote-code --weight-format int4 --ratio 0.72 {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 qwen_agent.llm import get_chat_model 

ov_config = {"PERFORMANCE_HINT": "LATENCY", "NUM_STREAMS": "1", "CACHE_DIR": ""} 
llm_cfg = { 
    "ov_model_dir": model_path, 
    "model_type": "openvino", 
    "device": device.value, 
    "ov_config": ov_config, 
    # (オプション) 生成用の LLM ハイパーパラメーター:
    "generate_cfg": {"top_p": 0.8}, 
} 
llm = get_chat_model(llm_cfg)

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

import urllib.parse 
import json5 
import requests 
from qwen_agent.tools.base import BaseTool, register_tool 

@register_tool("image_generation") 
class ImageGeneration(BaseTool): 
    description = "AI painting (image generation) service, input text description, and return the image URL drawn based on text information." 
    parameters = [{"name": "prompt", "type": "string", "description": "Detailed description of the desired image content, in English", "required": True}] 

    def call(self, params: str, **kwargs) -> str: 
        prompt = json5.loads(params)["prompt"] 
        prompt = urllib.parse.quote(prompt) 
        return json5.dumps({"image_url": f"https://image.pollinations.ai/prompt/{prompt}"}, ensure_ascii=False) 

@register_tool("get_current_weather") 
class GetCurrentWeather(BaseTool): 
    description = "Get the current weather in a given city name." 
    parameters = [{"name": "city_name", "type": "string", "description": "The city and state, e.g. San Francisco, CA", "required": True}] 

    def call(self, params: str, **kwargs) -> str:
        # `params` は LLM エージェントによって生成される引数です 
        city_name = json5.loads(params)["city_name"] 
        key_selection = { 
            "current_condition": [ 
                "temp_C", 
                "FeelsLikeC", 
                "humidity", 
                "weatherDesc", 
                "observation_time", 
            ], 
        } 
        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) 

@register_tool("wikipedia") 
class Wikipedia(BaseTool): 
    description = "A wrapper around Wikipedia.Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects." 
    parameters = [{"name": "query", "type": "string", "description": "Query to look up on wikipedia", "required": True}] 

    def call(self, params: str, **kwargs) -> str:
        # `params` は LLM エージェントによって生成された引数です 
        from langchain.tools import WikipediaQueryRun 
        from langchain_community.utilities import WikipediaAPIWrapper 

        query = json5.loads(params)["query"] 
        wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=2, doc_content_chars_max=1000)) 
        resutlt = wikipedia.run(query) 
        return str(resutlt)

tools = ["image_generation", "get_current_weather", "wikipedia"]

from qwen_agent.agents import Assistant 
from qwen_agent.gui import WebUI 

bot = Assistant(llm=llm_cfg, function_list=tools, name="OpenVINO Agent")

from typing import List 
from qwen_agent.llm.schema import CONTENT, ROLE, USER, Message 
from qwen_agent.gui.utils import convert_history_to_chatbot 
from qwen_agent.gui.gradio import gr, mgr 

class OpenVINOUI(WebUI): 
    def request_cancel(self): 
        self.agent_list[0].llm.ov_model.request.cancel() 

    def clear_history(self): 
        return [] 

    def add_text(self, _input, _chatbot, _history):
        _history.append( 
            { 
                ROLE: USER, 
                CONTENT: [{"text": _input}], 
            } 
        ) 
        _chatbot.append([_input, None]) 
        yield gr.update(interactive=False, value=None), _chatbot, _history 

    def run( 
        self, 
        messages: List[Message] = None, 
        share: bool = False, 
        server_name: str = None, 
        server_port: int = None, 
        **kwargs, 
    ): 
        self.run_kwargs = kwargs 

        with gr.Blocks( 
            theme=gr.themes.Soft(), 
            css=".disclaimer {font-variant-caps: all-small-caps;}", 
        ) as self.demo: 
            gr.Markdown("""<h1><center>OpenVINO Qwen Agent </center></h1>""") 
            history = gr.State([]) 

            with gr.Row(): 
                with gr.Column(scale=4): 
                    chatbot = mgr.Chatbot( 
                        value=convert_history_to_chatbot(messages=messages), 
                        avatar_images=[ 
                            self.user_config, 
                            self.agent_config_list, 
                        ], 
                        height=900, 
                        avatar_image_width=80, 
                        flushing=False, 
                        show_copy_button=True, 
                    ) 
                    with gr.Column(): 
                        input = 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", variant="primary") 
                            stop = gr.Button("Stop") 
                            clear = gr.Button("Clear") 
                with gr.Column(scale=1): 
                    agent_interactive = self.agent_list[0] 
                    capabilities = [key for key in agent_interactive.function_map.keys()] 
                    gr.CheckboxGroup( 
                        label="Tools", 
                        value=capabilities, 
                        choices=capabilities, 
                        interactive=False, 
                    ) 
            with gr.Row(): 
                gr.Examples(self.prompt_suggestions, inputs=[input], label="Click on any example and press the 'Submit' button") 

            input_promise = submit.click( 
                fn=self.add_text, 
                inputs=[input, chatbot, history], 
                outputs=[input, chatbot, history], 
                queue=False, 
            ) 
            input_promise = input_promise.then( 
                self.agent_run, 
                [chatbot, history], 
                [chatbot, history], 
            ) 
            input_promise.then(self.flushed, None, [input]) 
            stop.click( 
                fn=self.request_cancel, 
                inputs=None, 
                outputs=None, 
                cancels=[input_promise], 
                queue=False, 
            ) 
            clear.click(lambda: None, None, chatbot, queue=False).then(self.clear_history, None, history) 

            self.demo.load(None)
 
        self.demo.launch(share=share, server_name=server_name, server_port=server_port) 

chatbot_config = { 
    "prompt.suggestions": [ 
        "Based on current weather in London, show me a picture of Big Ben", 
        "What is OpenVINO ?", 
        "Create an image of pink cat", 
        "What is the weather like in New York now ?", 
        "How many people live in Canada ?", 
    ], 
    "agent.avatar": openvino_logo, 
    "input.placeholder": "Please input your request here", 
} 

demo = OpenVINOUI( 
    bot, 
    chatbot_config=chatbot_config, 
) 

# リモートで起動する場合は、server_name と server_port を指定 
# demo.run(server_name='your server name', server_port='server port in int') 
try: 
    demo.run() 
except Exception: 
    demo.run(share=True)

# demo.demo.close()