实战精选 | 从0到1构建 OpenVINO™ 智能体流水线

 !optimum-cli export openvino --model "Qwen/Qwen2.5-3B-Instruct"--task text-generation-with-past --trust-remote-code --weight-format int4 --group-size 128 --ratio 1.0 --sym “Qwen2.5-3B-Instruct-int4-ov”

tokenizer = AutoTokenizer.from_pretrained(llm_model_path, trust_remote_code=True)
ov_config = {hints.performance_mode(): hints.PerformanceMode.LATENCY, streams.num(): "1", props.cache_dir(): ""}
llm = OVModelForCausalLM.from_pretrained(    llm_model_path,    device=llm_device.value,    ov_config=ov_config,    config=AutoConfig.from_pretrained(llm_model_path, trust_remote_code=True),    trust_remote_code=True,)

def text_completion(prompt: str, stop_words) -> str:    im_end = "<|im_end|>"    if im_end not in stop_words:        stop_words = stop_words + [im_end]    streamer = TextStreamer(tokenizer, timeout=60.0, skip_prompt=True, skip_special_tokens=True)
    stopping_criteria = StoppingCriteriaList([StopSequenceCriteria(stop_words, tokenizer)])    input_ids = torch.tensor([tokenizer.encode(prompt)])    generate_kwargs = dict(        input_ids=input_ids,        streamer=streamer,        stopping_criteria=stopping_criteria,    )    output = llm.generate(**generate_kwargs)    output = output.tolist()[0]    output = tokenizer.decode(output, errors="ignore")    assert output.startswith(prompt)    output = output[len(prompt) :].replace("<|endoftext|>", "").replace(im_end, "")
    for stop_str in stop_words:        idx = output.find(stop_str)        if idx != -1:            output = output[: idx + len(stop_str)]    return output

TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters}"""
PROMPT_REACT = """Answer the following questions as best you can. You have access to the following APIs:
{tools_text}
Use the following format:
Question: the input question you must answerThought: you should always think about what to doAction: the action to take, should be one of [{tools_name_text}]Action Input: the input to the actionObservation: the result of the action... (this Thought/Action/Action Input/Observation can be repeated zero or more times)Thought: I now know the final answerFinal Answer: the final answer to the original input question
Begin!
Question: {query}"""

def parse_latest_tool_call(text):    tool_name, tool_args = "", ""    i = text.rfind("\nAction:")    j = text.rfind("\nAction Input:")    k = text.rfind("\nObservation:")    if 0 <= i < j:  # If the text has `Action` and `Action input`,        if k < j:  # but does not contain `Observation`,            # then it is likely that `Observation` is ommited by the LLM,            # because the output text may have discarded the stop word.            text = text.rstrip() + "\nObservation:"  # Add it back.        k = text.rfind("\nObservation:")        tool_name = text[i + len("\nAction:") : j].strip()        tool_args = text[j + len("\nAction Input:") : k].strip()        text = text[:k]    return tool_name, tool_args, text

tools = [    {        "name_for_human": "get weather",        "name_for_model": "get_weather",        "description_for_model": 'Get the current weather in a given city name."',        "parameters": [            {                "name": "city_name",                "description": "City name",                "required": True,                "schema": {"type": "string"},            }        ],    },    {        "name_for_human": "image generation",        "name_for_model": "image_gen",        "description_for_model": "AI painting (image generation) service, input text description, and return the image URL drawn based on text information.",        "parameters": [            {                "name": "prompt",                "description": "describe the image",                "required": True,                "schema": {"type": "string"},            }        ],    },]

    while True:        output = text_completion(planning_prompt + text, stop_words=["Observation:", "Observation:\n"])        action, action_input, output = parse_latest_tool_call(output)        if action:            observation = call_tool(action, action_input)            output += f"\nObservation: = {observation}\nThought:"            observation = f"{observation}\nThought:"            print(observation)            text += output        else:            text += output            break

query = "get the weather in London, and create a picture of Big Ben based on the weather information"
response, history = llm_with_tool(prompt=query, history=history, list_of_tool_info=tools)