app.py

from huggingface_hub import InferenceClient, HfApi, HfFileSystem
import gradio as gr
import requests
import random
import prompts
import uuid
import json
import re
import os
access_token=os.environ['HF_TOKEN']

fs = HfFileSystem(token=access_token)

loc_folder="chat_history"
loc_file="chat_json"
user_="Space-Share/"
repo_="test3"
clients = [
    {'type':'image','loc':'hf','key':'','name':'black-forest-labs/FLUX.1-dev','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'image','loc':'hf','key':'','name':'ostris/Flex.1-alpha','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'Qwen/Qwen2.5-Coder-32B-Instruct','rank':'op','max_tokens':32768,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'Qwen/Qwen2.5-72B-Instruct','rank':'op','max_tokens':32768,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'Qwen/QwQ-32B-Preview','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'mistralai/Mixtral-8x7B-Instruct-v0.1','rank':'op','max_tokens':40000,'schema':{'bos':'<s>','eos':'</s>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'mistralai/Mixtral-7B-Instruct-v0.2','rank':'op','max_tokens':40000,'schema':{'bos':'<s>','eos':'</s>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'mistralai/Mixtral-7B-Instruct-v0.3','rank':'op','max_tokens':40000,'schema':{'bos':'<s>','eos':'</s>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B','rank':'op','max_tokens':40000,'schema':{'bos':'<s>','eos':'</s>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'deepseek-ai/DeepSeek-R1-Distill-Qwen-32B','rank':'op','max_tokens':40000,'schema':{'bos':'<s>','eos':'</s>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'meta-llama/Llama-3.1-8B-Instruct','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'meta-llama/Llama-3,2-3B-Instruct','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'meta-llama/Llama-3.3-70B-Instruct','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'google/gemma-2-2b-it','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'google/gemma-2-9b-it','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'google/gemma-2-27b-it','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'microsoft/Phi-3-mini-4k-instruct','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
    {'type':'text','loc':'hf','name':'microsoft/Phi-3.5-mini-instruct','rank':'op','max_tokens':16384,'schema':{'bos':'<|im_start|>','eos':'<|im_end|>'},'ppt':'None'},
]
def file_template(inp):
    if "readme.md" in inp.lower():
        template=prompts.README
    else:template="NONE"
    return template
def format_prompt(message, mod, system):
    eos=f"{clients[int(mod)]['schema']['eos']}\n"
    bos=f"{clients[int(mod)]['schema']['bos']}\n"
    prompt=""
    prompt+=bos
    prompt+=system
    prompt+=eos
    prompt+=bos
    prompt += f"[INST] {message} [/INST]"
    prompt+=eos
    prompt+=bos
    return prompt
def generate(prompt,history,mod=2,tok=4000,seed=1,role="ASSISTANT",data=None):
    #print("#####",history,"######")

    gen_images=False
    client=InferenceClient(clients[int(mod)]['name'])
    client_tok=clients[int(mod)]['max_tokens']
    good_seed=[947385642222,7482965345792,8584806344673]

    if not os.path.isdir(loc_folder):os.mkdir(loc_folder)

    if os.path.isfile(f'{loc_folder}/{loc_file}.json'):
        with open(f'{loc_folder}/{loc_file}.json','r') as word_dict:
            lod=json.loads(word_dict.read())
        word_dict.close()    
    else:
        lod=[]
    if role == "MANAGER":
        system_prompt = prompts.PREFIX + prompts.MANAGER.replace("**TIMELINE**",data[4]).replace("**HISTORY**",str(history))
        formatted_prompt = format_prompt(prompt, mod, system_prompt)
    elif role == "PATHMAKER":
        system_prompt = prompts.PREFIX + prompts.PATH_MAKER.replace("**FILE_LIST**",str(data[3])).replace("**CURRENT_OR_NONE**",str(data[4])).replace("**PROMPT**",json.dumps(data[0],indent=4)).replace("**HISTORY**",str(history))
        formatted_prompt = format_prompt(prompt, mod, system_prompt)
    elif role == "CREATE_FILE":
        system_prompt = prompts.PREFIX + prompts.CREATE_FILE.replace("**FILE_LIST**",str(data[3])).replace("**TIMELINE**",data[4]).replace("**FILENAME**",str(data[1])).replace("**TEMPLATE_OR_NONE**",str(data[2]).strip("''").format())
        formatted_prompt = format_prompt(prompt, mod, system_prompt)
    elif role == "SEARCH":
        system_prompt = prompts.SEARCH.replace("**DATA**",data)
        formatted_prompt = format_prompt(f'USER:{prompt}', mod, system_prompt)
    else: system_prompt = "";formatted_prompt = format_prompt(f'USER:{prompt}', mod, system_prompt)
    
    if tok==None:tok=client_tok-len(formatted_prompt)+10
    print("tok",tok)
    generate_kwargs = dict(
        temperature=0.9,
        max_new_tokens=tok, #total tokens - input tokens
        top_p=0.99,
        repetition_penalty=1.0,
        do_sample=True,
        seed=seed,
    )
    output = ""
    if role=="MANAGER":
        print("Running Manager")
        stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=True)
        for response in stream:
            output += response.token.text
        yield output
        yield history
        yield prompt

    elif role=="PATHMAKER":
        print("Runnning ", role)
        stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=True)
        #prompt=f"We just completed role:{role}, now choose the next tool to complete the task:{prompt}, or COMPLETE"
        for response in stream:
            output += response.token.text
        print(output)
        yield output
        yield history
        yield prompt
        
    elif role=="CREATE_FILE":
        print("Running Create File")
        stream = client.text_generation(formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=True)
        for response in stream:
            output += response.token.text
        
        #print(file_content)
        print(output)
        yield 'test1'
        yield data[1]
        yield output

def parse_json(inp):
    print("PARSE INPUT")
    print(inp)
    if type(inp)==type(""):
        print("STRING-------")
        
        lines=""
        if "```" in inp:
            start = inp.find("```json") + 7  
            end = inp.find("```", start) 
            if start >= 0 and end >= 0:
                inp= inp[start:end] 

            print("Extracted Lines")
            print(inp)
        try:
            start_file=inp.find('filename') + 11
            name_file=inp.find(',', start_file)
            out1 = inp[start_file:name_file].strip("''").strip('""').strip("./")
            end_file=inp.find('filecontent') + 13
            out2 = inp[end_file:-1].strip("''").strip('""').strip("{}").format()
            #out1=inp.split('"filename":')[1]
            #out_json=json.loads([inp][0])
            #out1=str(out_json['filename'])
            #out2=str(out_json['filecontent'])
            return out1,out2
        except Exception as e:
            print(e)
            return "None","None"
    if type(inp)==type({}):
        print("DICT-------")
        out1=str(inp['filename'])
        out2=str(inp['filecontent'])
        return out1,out2
def build_space(repo_name,file_name,file_content,access_token=""):
    try:
        repo_path=user_+str(repo_)
        access_token=os.environ['HF_TOKEN']
        if not access_token:
            return [{'role':'assistant','content': 'ENTER A HUGGINGFACE TOKEN'}]
        api=HfApi(endpoint="https://huggingface.co", token=access_token)
        repo_url = api.create_repo(
                repo_id=repo_path,
                repo_type="space",
                space_sdk="gradio",
                exist_ok=True,
                private=False,
            )
        
        local_file_path=str(uuid.uuid4())
        with open(file_name, 'w') as f:
            f.write(str(file_content).strip("''").strip('""').strip("{}").format())
        f.close()
        # Upload a local file to the Space
        commit_message = "Adding file test: "+ str(file_name)
        
        api.upload_file(path_or_fileobj=file_name, path_in_repo=file_name.strip("./"), repo_id=repo_path, repo_type='space', commit_message=commit_message)
        print("File uploaded successfully.")
        # Commit changes
        commit_message += "\nInitial commit to the repository."+ f'{repo_path}/' + f'{file_name}'
        #api.commit_repo(space_id, message=commit_message)
        return [{'role':'assistant','content': commit_message+'\nCommit Success' }]
    except Exception as e:
        print("ERROR ",e)
        return [{'role':'assistant','content': 'There was an Error: ' + str(e)}]

        
def agent(prompt_in,history,mod=2,tok_in=""):
    print(prompt_in)
    print('mod ',mod)
    in_data=[None,None,None,None,None,]
    #in_data[0]=prompt_in['text']
    in_data[0]=prompt_in
    prompt=prompt_in
    fn=""
    com=""
    go=True
    MAX_DATA=int(clients[int(mod)]['max_tokens'])*2
    if not history:history=[{'role':'user','content':prompt_in['text']}]
    while go == True:
        fs = HfFileSystem(token=access_token)

        try:
            #file_list = fs.ls(f'spaces/{user_}{repo_}',detail=False)
            file_list = fs.ls(f'spaces/{user_}{repo_}',detail=False)
        except Exception as e:
            print(e)
            file_list=["NO FILES YET"]
        print('file list\n',file_list)

        
        seed = random.randint(1,9999999999999)
        c=0
        #history = [history[-4:]]
        if len(str(history)) > MAX_DATA*4:
            history = [history[-2:]]
        print('history',history)
        role="PATHMAKER"
        in_data[3]=file_list
        outph= list(generate(prompt,history,mod,2400,seed,role,in_data))[0]
        
        in_data[4]=outph
        print(outph)
        history.extend([{'role':'assistant','content':str(outph)}])
        yield history
        role="MANAGER"
        outp=generate(prompt,history,mod,128,seed,role,in_data)
        outp0=list(outp)[0].split('<|im_end|>')[0]
        #outp0 = re.sub('[^a-zA-Z0-9\s.,?!%()]', '', outpp)
        history.extend([{'role':'assistant','content':str(outp0)}])
        yield history
        for line in outp0.split("\n"):
            if "action:" in line:
                try:
                    com_line = line.split('action:')[1]
                    fn = com_line.split('action_input=')[0]
                    com = com_line.split('action_input=')[1].split('<|im_end|>')[0]
                    #com = com_line.split('action_input=')[1].replace('<|im_end|>','').replace("}","").replace("]","").replace("'","")
                    print(com)
                except Exception as e:
                    pass
                    fn="NONE"
             
                if 'CREATE_FILE' in fn:
                    print('CREATE_FILE called')
                    in_data[1]=com
                    temp1=file_template(com)
                    in_data[2]=temp1
                    in_data[3]=file_list
                    out_o =generate(prompt,history,mod=mod,tok=10000,seed=seed,role="CREATE_FILE",data=in_data)
                    out_w=list(out_o)
                    ret1,ret2 = parse_json(out_w[2].split('<|im_end|>')[0])
                    print('ret1',ret1)
                    print('ret2',ret2)
                    build_out = build_space(repo_,ret1,ret2,access_token=tok_in)
                    if build_out[0]["content"]=="ENTER A HUGGINGFACE TOKEN":
                        yield [{'role':'assistant','content':"ENTER A HUGGINGFACE TOKEN"}]
                        go=False
                        break
                    history+=[{'role':'system','content':f'observation:{build_out}'}]
                    yield history

                elif 'READ_FILE' in fn:
                    try:
                        with open(f'{com}', 'r') as f:
                            file_read = f.read()
                        f.close()
                        #file_read = fs.read_text(f'spaces/{user_}{repo_}/{com.strip("./")}',detail=False)
                    except Exception as e:
                        print(e)
                        file_read="FILE HAS NO CONTENT"
                    print('file list\n',file_read)
                    history+=[{'role':'system','content':f'RETURNED FILE CONTENT: NAME: spaces/{user_}{repo_}/{com} CONTENT:{file_read}'}]
                    yield history

                elif 'DELETE_FILE' in fn:
                    try:
                        fs.rm(f'spaces/{user_}{repo_}/{com.strip("./")}',detail=False)
                        file_del="SUCCESS"
                    except Exception as e:
                        print(e)
                        file_del="FAILED"
                        #print('file list\n',file_read)
                    history+=[{'role':'system','content':f'REMOVING FILE:\n NAME: spaces/{user_}{repo_}/{com} \n STATUS: {file_del}'}]
                    yield history

                    
                elif 'IMAGE' in fn:
                    print('IMAGE called')
                    #out_im=gen_im(prompt,seed)
                    #yield [{'role':'assistant','content': out_im}]
                elif 'SEARCH' in fn:
                    print('SEARCH called')
                elif 'COMPLETE' in fn:
                    print('COMPLETE')
                    go=False
                    break
                elif 'NONE' in fn:
                    print('ERROR ACTION NOT FOUND')
                    history+=[{'role':'system','content':f'observation:The last thing we attempted resulted in an error, check formatting on the tool call'}]
                else:pass;seed = random.randint(1,9999999999999)
                    
with gr.Blocks() as ux:
    with gr.Row():
        with gr.Column():
            gr.HTML("""<center><div style='font-size:xx-large;font-weight:900;'>Chatbo</div><br>
            <div style='font-size:large;font-weight:700;'>This will make changes to your Huggingface File System</div><br>
            <div style='font-size:large;font-weight:900;'>Use at your own risk!</div><br>
            """)
            chatbot=gr.Chatbot(type='messages',show_label=False, show_share_button=False, show_copy_button=True, layout="panel")
            prompt=gr.MultimodalTextbox(label="Prompt",file_count="multiple", file_types=["image"])
            mod_c=gr.Dropdown(choices=[n['name'] for n in clients],value='Qwen/Qwen2.5-Coder-32B-Instruct',type='index')
            tok_in=gr.Textbox(label='HF TOKEN')
            #chat_ux=gr.ChatInterface(fn=agent,chatbot=chatbot,additional_inputs=[mod_c]).load()
            #chat_ux.additional_inputs=[mod_c]
            #chat_ux.load()
            with gr.Row():
                submit_b = gr.Button()
                stop_b = gr.Button("Stop")
                clear = gr.ClearButton([chatbot,prompt])
            with gr.Row(visible=False):
                stt=gr.Textbox()
        with gr.Column():
            gr.HTML()
            #html_view=gr.HTML("""<iframe src='https://huggingface.co/spaces/community-pool/test1/tree/main' height='1000' width='200'>Viewer Space</iframe>""")
    sub_b = submit_b.click(agent, [prompt,chatbot,mod_c,tok_in],chatbot)
    sub_p = prompt.submit(agent, [prompt,chatbot,mod_c,tok_in],chatbot)
    stop_b.click(None,None,None, cancels=[sub_b,sub_p])
ux.queue(default_concurrency_limit=20).launch(max_threads=40)