import json
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Dict, List, Literal, Optional, Sequence, Set, Tuple, Union

def calculate_gpa(grades: Sequence[str], hours: Sequence[int]) -> float:
    grade_to_score = {"A": 4, "B": 3, "C": 2}
    total_score, total_hour = 0, 0
    for grade, hour in zip(grades, hours):
        total_score += grade_to_score[grade] * hour
        total_hour += hour
    return round(total_score / total_hour, 2)

tool_map = {"calculate_gpa": calculate_gpa}

from transformers import AutoModelForCausalLM, AutoTokenizer, TextStreamer
from peft import PeftModel

tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-1.5B-Instruct")
model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2-1.5B-Instruct", 
                                             torch_dtype="auto", device_map="auto")

model = PeftModel.from_pretrained(model, "svjack/Qwen2-1_5B_Function_Call_tiny_lora")
streamer = TextStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)

SLOTS = Sequence[Union[str, Set[str], Dict[str, str]]]


DEFAULT_TOOL_PROMPT = (
    "You have access to the following tools:\n{tool_text}"
    "Use the following format if using a tool:\n"
    "```\n"
    "Action: tool name (one of [{tool_names}]).\n"
    "Action Input: the input to the tool, in a JSON format representing the kwargs "
    """(e.g. ```{{"input": "hello world", "num_beams": 5}}```).\n"""
    "```\n"
)

def default_tool_formatter(tools: List[Dict[str, Any]]) -> str:
    tool_text = ""
    tool_names = []
    for tool in tools:
        param_text = ""
        for name, param in tool["parameters"]["properties"].items():
            required = ", required" if name in tool["parameters"].get("required", []) else ""
            enum = ", should be one of [{}]".format(", ".join(param["enum"])) if param.get("enum", None) else ""
            items = (
                ", where each item should be {}".format(param["items"].get("type", "")) if param.get("items") else ""
            )
            param_text += "  - {name} ({type}{required}): {desc}{enum}{items}\n".format(
                name=name,
                type=param.get("type", ""),
                required=required,
                desc=param.get("description", ""),
                enum=enum,
                items=items,
            )

        tool_text += "> Tool Name: {name}\nTool Description: {desc}\nTool Args:\n{args}\n".format(
            name=tool["name"], desc=tool.get("description", ""), args=param_text
        )
        tool_names.append(tool["name"])

    return DEFAULT_TOOL_PROMPT.format(tool_text=tool_text, tool_names=", ".join(tool_names))

def default_tool_extractor(content: str) -> Union[str, List[Tuple[str, str]]]:
    regex = re.compile(r"Action:\s*([a-zA-Z0-9_]+)\s*Action Input:\s*(.+?)(?=\s*Action:|\s*$)", re.DOTALL)
    action_match: List[Tuple[str, str]] = re.findall(regex, content)
    if not action_match:
        return content

    results = []
    for match in action_match:
        tool_name = match[0].strip()
        tool_input = match[1].strip().strip('"').strip("```")
        try:
            arguments = json.loads(tool_input)
            results.append((tool_name, json.dumps(arguments, ensure_ascii=False)))
        except json.JSONDecodeError:
            return content

    return results

#### Function tool defination
tools = [
        {
            "type": "function",
            "function": {
                "name": "calculate_gpa",
                "description": "Calculate the Grade Point Average (GPA) based on grades and credit hours",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "grades": {"type": "array", "items": {"type": "string"}, "description": "The grades"},
                        "hours": {"type": "array", "items": {"type": "integer"}, "description": "The credit hours"},
                    },
                    "required": ["grades", "hours"],
                },
            },
        }
    ]

tools_input = list(map(lambda x: x["function"], tools))
system_tool_prompt = default_tool_formatter(tools_input)
#print(system_tool_prompt)

def qwen_hf_predict(messages, qw_model = model,
    tokenizer = tokenizer, streamer = streamer,
    do_sample = True,
    top_p = 0.95,
    top_k = 40,
    max_new_tokens = 512,
    max_input_length = 3500,
    temperature = 0.9,
    repetition_penalty = 1.0,
    device = "cuda"):

    encodeds = tokenizer.apply_chat_template(messages, return_tensors="pt",
        add_generation_prompt=True
    )
    model_inputs = encodeds.to(device)

    generated_ids = qw_model.generate(model_inputs, max_new_tokens=max_new_tokens,
                                do_sample=do_sample,
                                  streamer = streamer,
                                  top_p = top_p,
                                  top_k = top_k,
                                  temperature = temperature,
                                  repetition_penalty = repetition_penalty,
                                  )
    out = tokenizer.batch_decode(generated_ids)[0].split("<|im_start|>assistant")[-1].replace("<|im_end|>", "").strip()
    return out

messages = [
    {
        "role" :"system",
        "content": system_tool_prompt
    },
    {"role": "user", "content": "My grades are A, A, B, and C. The credit hours are 3, 4, 3, and 2."}
]

out = qwen_hf_predict(messages)
tool_out = default_tool_extractor(out)
print(tool_out)

name, arguments = tool_out[0][0], json.loads(tool_out[0][1])
tool_result = tool_map[name](**arguments)
print(tool_result)

messages.append(
    {
        "role" :"assistant",
        "content": out
    }
)

messages.append({"role": "tool", "content": json.dumps({"gpa": tool_result}, ensure_ascii=False)})

final_out = qwen_hf_predict(messages)
print(final_out)

Output

Action: calculate_gpa
Action Input: {"grades": ["A", "A", "B", "C"], "hours": [3, 4, 3, 2]}
[('calculate_gpa', '{"grades": ["A", "A", "B", "C"], "hours": [3, 4, 3, 2]}')]
3.42
Your calculated GPA is 3.42.

Inference

python

messages = [
    {
        "role" :"system",
        "content": system_tool_prompt
    },
    {"role": "user", "content": "我的成绩分别是A,A,B,C学分分别是3, 4, 3,和2"}
]

out = qwen_hf_predict(messages)
tool_out = default_tool_extractor(out)
print(tool_out)

name, arguments = tool_out[0][0], json.loads(tool_out[0][1])
tool_result = tool_map[name](**arguments)
print(tool_result)

messages.append(
    {
        "role" :"assistant",
        "content": out
    }
)

messages.append({"role": "tool", "content": json.dumps({"gpa": tool_result}, ensure_ascii=False)})

final_out = qwen_hf_predict(messages)
print(final_out)

Output

Action: calculate_gpa
Action Input: {"grades": ["A", "A", "B", "C"], "hours": [3, 4, 3, 2]}
[('calculate_gpa', '{"grades": ["A", "A", "B", "C"], "hours": [3, 4, 3, 2]}')]
3.42
你的GPA是3.42。

train_2024-06-17-19-49-05

This model is a fine-tuned version of Qwen/Qwen2-7B-Instruct on the glaive_toolcall_zh and the glaive_toolcall_en datasets.

Model description

More information needed

Intended uses & limitations

More information needed

Training and evaluation data

More information needed

Training procedure

Training hyperparameters

The following hyperparameters were used during training:

learning_rate: 5e-05
train_batch_size: 1
eval_batch_size: 8
seed: 42
distributed_type: multi-GPU
num_devices: 2
gradient_accumulation_steps: 8
total_train_batch_size: 16
total_eval_batch_size: 16
optimizer: Adam with betas=(0.9,0.999) and epsilon=1e-08
lr_scheduler_type: cosine
num_epochs: 3.0
mixed_precision_training: Native AMP

Training results

Framework versions

PEFT 0.11.1
Transformers 4.41.2
Pytorch 2.3.1+cu121
Datasets 2.20.0
Tokenizers 0.19.1