app.py

#from conf import *
#main_path = "/Volumes/TOSHIBA EXT/temp/kbqa_portable_prj"
#main_path = "/Users/svjack/temp/HP_kbqa"
#from conf import *
#main_path = model_path

import logging
import os
os.system("pip install huggingface_hub")
from huggingface_hub import space_info

import sys
from dataclasses import dataclass, field
from typing import Optional

import numpy as np
from datasets import ClassLabel, load_dataset, load_metric

import transformers
import transformers.adapters.composition as ac
from transformers import (
    AdapterConfig,
    AutoConfig,
    AutoModelForTokenClassification,
    AutoTokenizer,
    DataCollatorForTokenClassification,
    HfArgumentParser,
    #MultiLingAdapterArguments,
    PreTrainedTokenizerFast,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import get_last_checkpoint
from transformers.utils import check_min_version
from transformers.utils.versions import require_version


import pandas as pd
import pickle as pkl
from copy import deepcopy
import torch
from scipy.special import softmax
from functools import partial, reduce
import json
from io import StringIO
import re

from transformers import list_adapters, AutoModelWithHeads

from collections import defaultdict

@dataclass
class ModelArguments:
    """
    Arguments pertaining to which model/config/tokenizer we are going to fine-tune from.
    """

    model_name_or_path: str = field(
        metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"}
    )
    config_name: Optional[str] = field(
        default=None, metadata={"help": "Pretrained config name or path if not the same as model_name"}
    )
    tokenizer_name: Optional[str] = field(
        default=None, metadata={"help": "Pretrained tokenizer name or path if not the same as model_name"}
    )
    cache_dir: Optional[str] = field(
        default=None,
        metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"},
    )
    model_revision: str = field(
        default="main",
        metadata={"help": "The specific model version to use (can be a branch name, tag name or commit id)."},
    )
    use_auth_token: bool = field(
        default=False,
        metadata={
            "help": "Will use the token generated when running `transformers-cli login` (necessary to use this script "
            "with private models)."
        },
    )


@dataclass
class DataTrainingArguments:
    """
    Arguments pertaining to what data we are going to input our model for training and eval.
    """

    task_name: Optional[str] = field(default="ner", metadata={"help": "The name of the task (ner, pos...)."})
    dataset_name: Optional[str] = field(
        default=None, metadata={"help": "The name of the dataset to use (via the datasets library)."}
    )
    dataset_config_name: Optional[str] = field(
        default=None, metadata={"help": "The configuration name of the dataset to use (via the datasets library)."}
    )
    train_file: Optional[str] = field(
        default=None, metadata={"help": "The input training data file (a csv or JSON file)."}
    )
    validation_file: Optional[str] = field(
        default=None,
        metadata={"help": "An optional input evaluation data file to evaluate on (a csv or JSON file)."},
    )
    test_file: Optional[str] = field(
        default=None,
        metadata={"help": "An optional input test data file to predict on (a csv or JSON file)."},
    )
    text_column_name: Optional[str] = field(
        default=None, metadata={"help": "The column name of text to input in the file (a csv or JSON file)."}
    )
    label_column_name: Optional[str] = field(
        default=None, metadata={"help": "The column name of label to input in the file (a csv or JSON file)."}
    )
    overwrite_cache: bool = field(
        default=False, metadata={"help": "Overwrite the cached training and evaluation sets"}
    )
    preprocessing_num_workers: Optional[int] = field(
        default=None,
        metadata={"help": "The number of processes to use for the preprocessing."},
    )
    pad_to_max_length: bool = field(
        default=False,
        metadata={
            "help": "Whether to pad all samples to model maximum sentence length. "
            "If False, will pad the samples dynamically when batching to the maximum length in the batch. More "
            "efficient on GPU but very bad for TPU."
        },
    )
    max_train_samples: Optional[int] = field(
        default=None,
        metadata={
            "help": "For debugging purposes or quicker training, truncate the number of training examples to this "
            "value if set."
        },
    )
    max_eval_samples: Optional[int] = field(
        default=None,
        metadata={
            "help": "For debugging purposes or quicker training, truncate the number of evaluation examples to this "
            "value if set."
        },
    )
    max_predict_samples: Optional[int] = field(
        default=None,
        metadata={
            "help": "For debugging purposes or quicker training, truncate the number of prediction examples to this "
            "value if set."
        },
    )
    label_all_tokens: bool = field(
        default=False,
        metadata={
            "help": "Whether to put the label for one word on all tokens of generated by that word or just on the "
            "one (in which case the other tokens will have a padding index)."
        },
    )
    return_entity_level_metrics: bool = field(
        default=False,
        metadata={"help": "Whether to return all the entity levels during evaluation or just the overall ones."},
    )

    def __post_init__(self):
        if self.dataset_name is None and self.train_file is None and self.validation_file is None:
            raise ValueError("Need either a dataset name or a training/validation file.")
        else:
            if self.train_file is not None:
                extension = self.train_file.split(".")[-1]
                assert extension in ["csv", "json"], "`train_file` should be a csv or a json file."
            if self.validation_file is not None:
                extension = self.validation_file.split(".")[-1]
                assert extension in ["csv", "json"], "`validation_file` should be a csv or a json file."
        self.task_name = self.task_name.lower()

import os

#p0 = os.path.join(main_path, "sel_ner/ner_data_args.pkl")
p0 = "sel_ner/ner_data_args.pkl"
assert os.path.exists(p0)
with open(p0, "rb") as f:
    t4 = pkl.load(f)

model_args, data_args, training_args, adapter_args = map(deepcopy, t4)

zh_model = AutoModelWithHeads.from_pretrained("bert-base-chinese")

#config_path = "/Users/svjack/temp/ner_trans/adapter_ner_data/test-sel-ner/sel_ner/adapter_config.json"
#adapter_path = "/Users/svjack/temp/ner_trans/adapter_ner_data/test-sel-ner/sel_ner"
config_path = "sel_ner/adapter_config.json"
adapter_path = "sel_ner"
#config_path = os.path.join(main_path ,"sel_ner/adapter_config.json")
#adapter_path = os.path.join(main_path ,"sel_ner")

config = AdapterConfig.load(config_path)
zh_model.load_adapter(adapter_path, config=config)
zh_model.set_active_adapters(['sel_ner'])

def single_sent_pred(input_text, tokenizer, model):
    input_ = tokenizer(input_text)
    input_ids = input_["input_ids"]
    output = model(torch.Tensor([input_ids]).type(torch.LongTensor))
    output_prob = softmax(output.logits.detach().numpy()[0], axis = -1)
    token_list = tokenizer.convert_ids_to_tokens(input_ids)
    assert len(token_list) == len(output_prob)
    return token_list, output_prob

def single_pred_to_df(token_list, output_prob, label_list):
    assert output_prob.shape[0] == len(token_list) and output_prob.shape[1] == len(label_list)
    pred_label_list = pd.Series(output_prob.argmax(axis = -1)).map(
        lambda idx: label_list[idx]
    ).tolist()
    return pd.concat(list(map(pd.Series, [token_list, pred_label_list])), axis = 1)

def token_l_to_nest_l(token_l, prefix = "##"):
    req = []
    #req.append([])
    #### token_l must startswith [CLS]
    assert token_l[0] == "[CLS]"
    for ele in token_l:
        if not ele.startswith(prefix):
            req.append([ele])
        else:
            req[-1].append(ele)
    return req

def list_window_collect(l, w_size = 1, drop_NONE = False):
    assert len(l) >= w_size
    req = []
    for i in range(len(l)):
        l_slice = l[i: i + w_size]
        l_slice += [None] * (w_size - len(l_slice))
        req.append(l_slice)
    if drop_NONE:
        return list(filter(lambda x: None not in x, req))
    return req

def same_pkt_l(l0, l1):
    l0_size_l = list(map(len, l0))
    assert sum(l0_size_l) == len(l1)
    cum_l0_size = np.cumsum(l0_size_l).tolist()
    slice_l = list_window_collect(cum_l0_size, 2, drop_NONE=True)
    slice_l = [[0 ,slice_l[0][0]]] + slice_l
    slice_df = pd.DataFrame(slice_l)
    return (l0, slice_df.apply(lambda s: l1[s[0]:s[1]], axis = 1).tolist())


def cnt_backtrans_slice(token_list, label_list, prefix = "##",
    token_agg_func = lambda x: x[0] if len(x) == 1 else "".join([x[0]] + list(map(lambda y: y[len("##"):], x[1:]))),
    label_agg_func = lambda x: x[0] if len(x) == 1 else pd.Series(x).value_counts().index.tolist()[0]
                       ):
    token_nest_list = token_l_to_nest_l(token_list, prefix=prefix)
    token_nest_list, label_nest_list = same_pkt_l(token_nest_list, label_list)
    token_list_req = list(map(token_agg_func, token_nest_list))
    label_list_req = list(map(label_agg_func, label_nest_list))
    return (token_list_req, label_list_req)

def from_text_to_final(input_text, tokenizer, model, label_list):
    token_list, output_prob = single_sent_pred(input_text, tokenizer, model)
    token_pred_df = single_pred_to_df(token_list, output_prob, label_list)
    token_list_, label_list_ = token_pred_df[0].tolist(), token_pred_df[1].tolist()
    token_pred_df_reduce = pd.DataFrame(list(zip(*cnt_backtrans_slice(token_list_, label_list_))))
    return token_pred_df_reduce


tokenizer_name_or_path = model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path

tokenizer = AutoTokenizer.from_pretrained(
            tokenizer_name_or_path,
            cache_dir=model_args.cache_dir,
            use_fast=True,
            revision=model_args.model_revision,
            use_auth_token=True if model_args.use_auth_token else None,
)

label_list = ['O-TAG', 'E-TAG', 'T-TAG']

### fix eng with " "
### used when ner_model input with some eng-string fillwith " "
def fill_str(sent ,str_):
    is_en = False
    if re.findall("[a-zA-Z0-9 ]+", str_) and re.findall("[a-zA-Z0-9 ]+", str_)[0] == str_:
        is_en = True
    if not is_en:
        return str_
    find_part = re.findall("([{} ]+)".format(str_), text)
    assert find_part
    find_part = sorted(filter(lambda x: x.replace(" ", "") == str_.replace(" ", "") ,find_part), key = len, reverse = True)[0]
    assert find_part in sent
    return find_part

def for_loop_detect(s, invalid_tag = "O-TAG", sp_token = "123454321"):
    assert type(s) == type(pd.Series())
    char_list = s.iloc[0]
    tag_list = s.iloc[1]
    assert len(char_list) == len(tag_list)
    req = defaultdict(list)
    pre_tag = ""
    for idx, tag in enumerate(tag_list):
        if tag == invalid_tag or tag != pre_tag:
            for k in req.keys():
                if req[k][-1] != invalid_tag:
                    req[k].append(sp_token)
            if tag != pre_tag and tag != invalid_tag:
                char = char_list[idx]
                req[tag].append(char)
        elif tag != invalid_tag:
            char = char_list[idx]
            req[tag].append(char)
        pre_tag = tag
    req = dict(map(lambda t2: (
        t2[0],
        list(
        filter(lambda x: x.strip() ,"".join(t2[1]).split(sp_token))
        )
                         ), req.items()))
    return req

def ner_entity_type_predict_only(question):
    assert type(question) == type("")
    question = question.replace(" ", "")
    ner_df = from_text_to_final(
    " ".join(list(question)),
    tokenizer,
    zh_model,
    label_list
)
    assert ner_df.shape[0] == len(question) + 2
    ### [UNK] filling
    ner_df[0] = ["[CLS]"] + list(question) + ["[SEP]"]
    et_dict = for_loop_detect(ner_df.T.apply(lambda x: x.tolist(), axis = 1))
    return et_dict

import gradio as gr

example_sample = [
    "宁波在哪个省份？",
    "美国的通货是什么？",
]

markdown_exp_size = "##"
lora_repo = "svjack/chatglm3-few-shot"
lora_repo_link = "svjack/chatglm3-few-shot/?input_list_index=3"
emoji_info = space_info(lora_repo).__dict__["cardData"]["emoji"]
space_cnt = 1
task_name = "[---Chinese Question Entity Property decomposition---]"
description = f"{markdown_exp_size} {task_name} few shot prompt in ChatGLM3 Few Shot space repo (click submit to activate) : [{lora_repo_link}](https://huggingface.co/spaces/{lora_repo_link}) {emoji_info}"


demo = gr.Interface(
        fn=ner_entity_type_predict_only,
        inputs="text",
        outputs="json",
        title=f"Chinese Question Entity Property decomposition 🌧️ demonstration",
        description = description,
        examples=example_sample if example_sample else None,
        cache_examples = False
    )

with demo:
    gr.HTML(
    '''
                <div style="justify-content: center; display: flex;">
                <iframe
                src="https://svjack-chatglm3-few-shot-demo.hf.space/?input_list_index=3"
                frameborder="0"
                width="1400"
                height="768"
                ></iframe>
                </div>
    '''
    )

demo.launch(server_name=None, server_port=None)

'''
rep = requests.post(
    url = "http://localhost:8855/extract_et",
    data = {
        "question": "哈利波特的作者是谁？"
    }
)
json.loads(rep.content.decode())

@csrf_exempt
def extract_et(request):
    assert request.method == "POST"
    post_data = request.POST
    question = post_data["question"]
    assert type(question) == type("")
    #question = "宁波在哪个省？"
    #abc = do_search(question)
    et_dict = ner_entity_type_predict_only(question)
    assert type(et_dict) == type({})
    return HttpResponse(json.dumps(et_dict))

if __name__ == "__main__":
    from_text_to_final("宁波在哪个省？",
    tokenizer,
    zh_model,
    label_list
                  )

    from_text_to_final("美国的通货是什么？",
    tokenizer,
    zh_model,
    label_list
                  )
'''