""" ai_single_response.py An executable way to call the model. example: *\gpt2_chatbot> python .\ai_single_response.py --prompt "where is the grocery store?" --time extended-summary: A system and method for interacting with a virtual machine using a series of messages , each message having associated otherwise one or more actions to be taken by the machine. The speaker participates in a chat with a responder , and the response from the responder is returned. """ import argparse import pprint as pp import time import warnings from datetime import datetime from pathlib import Path from cleantext import clean warnings.filterwarnings(action="ignore", message=".*gradient_checkpointing*") from aitextgen import aitextgen def query_gpt_model( folder_path, prompt_msg: str, speaker=None, responder="person beta", kparam=150, temp=0.75, top_p=0.65, verbose=False, use_gpu=False, ): """ query_gpt_model [pass a prompt in to model, get a response. Does NOT "remember" past conversation] Args: folder_path ([type]): [description] prompt_msg (str): [description] speaker ([type], optional): [description]. Defaults to None. responder (str, optional): [description]. Defaults to "person beta". kparam (int, optional): [description]. Defaults to 125. temp (float, optional): [description]. Defaults to 0.75. top_p (float, optional): [description]. Defaults to 0.65. verbose (bool, optional): [description]. Defaults to False. use_gpu (bool, optional): [description]. Defaults to False. Returns: [dict]: [returns a dict with A) just model response as str B) total conversation] """ ai = aitextgen( model="microsoft/DialoGPT-large", #model_folder=folder_path, to_gpu=False, ) print("loaded model") p_list = [] if "natqa" in str(folder_path).lower(): speaker = "person alpha" # manual correction responder = "person beta" if "wow" in str(folder_path).lower(): speaker = "person alpha" # manual correction responder = "person beta" if "peter" in str(folder_path).lower(): speaker = None # manual correction responder = "peter szemraj" if speaker is not None: p_list.append(speaker.lower() + ":" + "\n") # write prompt as the speaker p_list.append(prompt_msg.lower() + "\n") p_list.append("\n") p_list.append(responder.lower() + ":" + "\n") this_prompt = "".join(p_list) if verbose: print("overall prompt:\n") pp.pprint(this_prompt, indent=4) print("\n... generating... \n") this_result = ai.generate( n=1, top_k=kparam, batch_size=512, max_length=128, min_length=16, prompt=this_prompt, temperature=temp, top_p=top_p, do_sample=True, return_as_list=True, use_cache=True, ) if verbose: pp.pprint(this_result) # to see what is going on try: this_result = str(this_result[0]).split("\n") res_out = [clean(ele) for ele in this_result] p_out = [clean(ele) for ele in p_list] if verbose: pp.pprint(res_out) # to see what is going on pp.pprint(p_out) # to see what is going on diff_list = [] name_counter = 0 break_safe = False for resline in res_out: if (responder + ":") in resline: name_counter += 1 break_safe = True # next line a response from bot continue if ":" in resline and name_counter > 0: if break_safe: diff_list.append(resline) break_safe = False else: break if resline in p_out: break_safe = False continue else: diff_list.append(resline) break_safe = False if verbose: print("------------------------diff list: ") pp.pprint(diff_list) # to see what is going on print("---------------------------------") output = ", ".join(diff_list) except: output = "oops, there was an error. try again" p_list.append(output + "\n") p_list.append("\n") model_responses = {"out_text": output, "full_conv": p_list} print("finished!\n") return model_responses # Set up the parsing of command-line arguments def get_parser(): """ get_parser [a helper function for the argparse module] Returns: [argparse.ArgumentParser]: [the argparser relevant for this script] """ parser = argparse.ArgumentParser( description="submit a message and have a 774M parameter GPT model respond" ) parser.add_argument( "--prompt", required=True, # MUST HAVE A PROMPT type=str, help="the message the bot is supposed to respond to. Prompt is said by speaker, answered by responder.", ) parser.add_argument( "--model", required=False, type=str, # "gp2_DDandPeterTexts_774M_73Ksteps", - from GPT-Peter default="GPT2_trivNatQAdailydia_774M_175Ksteps", help="folder - with respect to git directory of your repo that has the model files in it (pytorch.bin + " "config.json). No models? Run the script download_models.py", ) parser.add_argument( "--speaker", required=False, default=None, help="Who the prompt is from (to the bot). Primarily relevant to bots trained on multi-individual chat data", ) parser.add_argument( "--responder", required=False, default="person beta", help="who the responder is. Primarily relevant to bots trained on multi-individual chat data", ) parser.add_argument( "--topk", required=False, type=int, default=150, help="how many responses to sample (positive integer). lower = more random responses", ) parser.add_argument( "--temp", required=False, type=float, default=0.75, help="specify temperature hyperparam (0-1). roughly considered as 'model creativity'", ) parser.add_argument( "--topp", required=False, type=float, default=0.65, help="nucleus sampling frac (0-1). aka: what fraction of possible options are considered?", ) parser.add_argument( "--verbose", default=False, action="store_true", help="pass this argument if you want all the printouts", ) parser.add_argument( "--time", default=False, action="store_true", help="pass this argument if you want to know runtime", ) return parser if __name__ == "__main__": args = get_parser().parse_args() query = args.prompt model_dir = str(args.model) model_loc = Path.cwd() / model_dir spkr = args.speaker rspndr = args.responder k_results = args.topk my_temp = args.temp my_top_p = args.topp want_verbose = args.verbose want_rt = args.time # force-update the speaker+responder params for the generic model case if "dailydialogue" in model_dir.lower(): spkr = "john smith" rspndr = "nancy sellers" # ^ arbitrary people created when parsing Daily Dialogue dataset # # force-update the speaker+responder params # for the generic model case if "natqa" in model_dir.lower(): spkr = "person alpha" rspndr = "person beta" # ^ arbitrary people created when parsing NatQA + TriviaQA + Daily Dialogue datasets st = time.time() resp = query_gpt_model( folder_path=model_loc, prompt_msg=query, speaker=spkr, responder=rspndr, kparam=k_results, temp=my_temp, top_p=my_top_p, verbose=want_verbose, use_gpu=False, ) output = resp["out_text"] pp.pprint(output, indent=4) # pp.pprint(this_result[3].strip(), indent=4) rt = round(time.time() - st, 1) if want_rt: print("took {runtime} seconds to generate. \n".format(runtime=rt)) if want_verbose: print("finished - ", datetime.now()) if want_verbose: p_list = resp["full_conv"] print("A transcript of your chat is as follows: \n") p_list = [item.strip() for item in p_list] pp.pprint(p_list)