Spaces:
Sleeping
Sleeping
| # AUTHOR: Adan Saikia | |
| import gradio as gr | |
| from gradio import themes | |
| from openai import OpenAI | |
| import os | |
| import logging | |
| from dotenv import load_dotenv | |
| from datetime import datetime | |
| import os | |
| import json | |
| import copy | |
| load_dotenv(r"C:\Users\devad\OneDrive\Documents\Test\gradioPractice\.env", override=True) | |
| # Set the log level to DEBUG to see raw request/response data | |
| # def log(On:bool):logging.basicConfig(level=logging.DEBUG) if On else None | |
| api_key = os.getenv("GROQ_API_KEY") | |
| if not api_key: | |
| raise ValueError("API key not found. Please check .env file.") | |
| client = OpenAI( | |
| base_url="https://api.groq.com/openai/v1", | |
| api_key=os.getenv("GROQ_API_KEY") | |
| ) | |
| def generate(message, | |
| history, | |
| streaming=False, | |
| contextLen=24, | |
| systemInst="You are a friendly and helpful assistant", | |
| temperature=1.0, | |
| model="openai/gpt-oss-120b", | |
| chats=None): | |
| # def generate(message, history): | |
| if chats is None: | |
| chats = [] | |
| if message == '' or len(message) == 0: | |
| gr.Warning("Message is empty. Please write a prompt.") | |
| yield "", history, gr.update(stop_btn=True, submit_btn=False), chats | |
| messages = [] | |
| systemPrompt = [{"role": "system", "content": systemInst}] | |
| # print(str(history)) | |
| # convert history to OpenAI format | |
| for data in history: | |
| # messages.append({"role": "user", "content": user}) | |
| # messages.append({"role": "assistant", "content": assistant}) | |
| messages.append({"role": data["role"], "content": data["content"][0]["text"]}) | |
| # add current message | |
| messages.append({"role": "user", "content": message}) | |
| history.append({'role': 'user' , 'metadata': None, 'content': [{'text': message, 'type': 'text'}], 'options': None}) | |
| chats, chatHistory = __initNewChats(pastChats=chats, chatHistory=history) | |
| yield "", history, gr.update(stop_btn=True, submit_btn=False), chats | |
| response = client.chat.completions.create( | |
| model=model, | |
| messages=systemPrompt+messages[-contextLen:], | |
| stream=streaming, | |
| temperature=temperature | |
| ) | |
| if streaming: # Hidden BUG: UI should show 'incrementing' message | |
| # history should store only 'final' message | |
| full_message ="" # seperate both, atleast 'exclusively' for streaming-----FIXED. | |
| history.append({'role': 'assistant' , 'metadata': None, 'content': [{'text': full_message, 'type': 'text'}], 'options': None}) # Intentionally append an empty assistant turn | |
| for chunk in response: | |
| # print(type(chunk)) | |
| if chunk.choices[0].delta.content is not None: | |
| # print(f"DEBUG: chunk.choices[0].delta.content = {chunk.choices[0].delta.content}\n") | |
| full_message += chunk.choices[0].delta.content | |
| # print(f"DEBUG: full_message 1 = {full_message}\n") | |
| history[-1] = ({'role': 'assistant' , 'metadata': None, 'content': [{'text': full_message, 'type': 'text'}], 'options': None}) # Now no issues doing so, because we added an empty one initially | |
| yield None, history, gr.update(stop_btn=False, submit_btn=True), chats | |
| # print(f"DEBUG: full_message 2 = {full_message}\n") | |
| messages.append({"role": "assistant", "content": full_message}) # Ensured only the complete message is stored in messages | |
| else: | |
| messages.append({"role": "assistant", "content": response.choices[0].message.content}) | |
| history.append({'role': 'assistant' , 'metadata': None, 'content': [{'text': response.choices[0].message.content, 'type': 'text'}], 'options': None}) | |
| yield None, history, gr.update(stop_btn=False, submit_btn=True), chats | |
| def clearChatStatement(): | |
| gr.Info("Chat Cleared") | |
| chatbot = [] | |
| return chatbot | |
| def regenerateResponse(chatbot): | |
| if len(chatbot) > 0: | |
| chatbot.pop() | |
| if chatbot and 'content' in chatbot[-1]: | |
| # yield1, yield2 = generate(chatbot[-1]['content'][0]['text'], chatbot) # returns first yield after user message | |
| # # second after response | |
| # chatbot = yield1[1] # yield returns tuple of msg content, history and send-stop update obj, we need history | |
| # regeneration accidentally made history format- user, user, assistant. we dont need duplicate user turn | |
| # SAME THING | |
| response = list(generate(chatbot[-1]['content'][0]['text'], chatbot)) | |
| chatbot = response[-1][1] | |
| chatbot.pop(-2) | |
| # print(chatbot) | |
| # responseFinal = {"role": "assistant", "content": returnedChatbot[1][-1]['content'][0]['text']} | |
| # print(f"DEBUG: chatbot = {chatbot}\n\n\nchatbot state = {chatbot_state}\n\n\nresponse = {response}") | |
| # chatbot.append(responseFinal) | |
| return chatbot | |
| def exportChat(chatbot): | |
| if len(chatbot) > 0: | |
| gr.Info(message="Chat Exported. Check 'export' directory.") | |
| messages = [] | |
| messages_json = [] | |
| datetimeCode = "chat_"+datetime.now().strftime("%y-%m-%d_%H-%M") | |
| for data in chatbot: | |
| messages_json.append({"role": data["role"], "content": data["content"][0]["text"]}) | |
| messages.append(f"{data['role']}: {data['content'][0]['text']}") | |
| if not os.path.exists("exports"): | |
| os.mkdir("exports") | |
| with open(fr"exports\{datetimeCode}.json", "w", encoding="utf-8") as f: | |
| json.dump(messages_json,f,indent=4) | |
| with open(fr"exports\{datetimeCode}.txt", "a", encoding="utf-8") as f: | |
| for i in messages: | |
| f.write(f"{i}\n") | |
| def importChat(content): | |
| messages_json = "" | |
| messages_chatbot = [] | |
| try: | |
| # print(f"Debug:\n{content.name}") | |
| with open(content.name, "r") as f: # No worries, it always returns full path | |
| try: | |
| messages_json = json.load(f) | |
| if len(messages_json) == 0: | |
| print("Warning: Empty JSON imported.") | |
| gr.Warning("Warning: Empty JSON found. Unable to import chat. Are you sure you want empty chat?") | |
| except: | |
| print("Error: The imported chat structure is corrupted. Please inspect JSON.") | |
| gr.Error("Error: The imported chat structure is corrupted. Please inspect JSON.") | |
| # except (ValueError, TypeError): | |
| # print("content is not file_name but file_content") // not needed cuz we always ensure "file-name" is recieved via 'content' arg | |
| except FileNotFoundError: | |
| print("Error: The file 'data.json' was not found.") | |
| gr.Warning("Error: The file 'data.json' was not found.") | |
| # print(f"Debug: {messages_json}") | |
| for data in messages_json: | |
| if "role" in data and "content" in data: | |
| messages_chatbot.append({'role':data["role"] , 'metadata': None, 'content': [{'text': data["content"], 'type': 'text'}], 'options': None}) | |
| return messages_chatbot | |
| def __stateToChatbot(chatBtn: gr.Button, pastChats): | |
| chatName = chatBtn.value | |
| print(chatName) | |
| for i in pastChats: | |
| if i['chatName'] == chatName: | |
| chatbot = i['chatHistory'] | |
| return chatbot | |
| with gr.Blocks() as demo: | |
| with gr.Column(): | |
| with gr.Column(): | |
| gr.Row(scale=1) | |
| gr.Markdown("<center><h1>MiniGPT</h1></center>", height=70) | |
| gr.Row(scale=1) | |
| with gr.Row(): | |
| with gr.Column(): | |
| chatbot = gr.Chatbot(editable='user', show_label=False) | |
| msg = gr.TextArea(placeholder="Type a message...", lines=1, show_label=False, submit_btn=True) | |
| # chat = gr.ChatInterface( | |
| # fn=generate, | |
| # chatbot=chatbot, | |
| # additional_inputs= | |
| # [ | |
| # streamCheckbox, | |
| # contextLenSlider, | |
| # tempSlider, | |
| # modelSwitchDropdown, | |
| # sysPromptTextBox | |
| # ], | |
| # save_history=True, | |
| # editable=True | |
| # ) | |
| with gr.Row(): # Took a whie, and lots of trial-error to align well | |
| # gr.Column(scale=0, min_width=200) # such that buttons dont expand to the side panel of # Update: Not required temporarily | |
| with gr.Column(min_width=1000): # chatInterface 'save_history = True' | |
| regenerateBtn = gr.Button(value="Regenerate Response") | |
| with gr.Row(): | |
| exportChatBtn = gr.Button(value="Export Chat") | |
| importChatBtn = gr.UploadButton(label="Import Chat", file_types=[".json"]) | |
| with gr.Sidebar(): | |
| newChatBtn = gr.Button(value="Create new chat") | |
| chats = gr.State([]) | |
| gr.Markdown(value="<center><h4>Past Conversations</h4></center>") | |
| gr.Markdown(value="____________________________________________") | |
| # chat1 = gr.Button("HULK SMASH", variant="huggingface", size="md") | |
| def chatsGenerator(pastChats: list[dict]): | |
| with gr.Column(): | |
| if pastChats and len(pastChats)>0: | |
| # print(pastChats) | |
| for i in pastChats: | |
| # gr.Button(value=i) | |
| # print(i) | |
| chatBtn = gr.Button(value=i['chatName'], elem_classes='chat-btn') | |
| chatBtn.click(fn=__stateToChatbot, inputs=[chatBtn, chats], outputs=[chatbot]) | |
| def __initNewChats(pastChats: list[dict], chatHistory): | |
| # randnum = random.randint(100000, 999999) | |
| # pastChats.append(f"{randnum}") | |
| # Already ensured chatHistory is not empty | |
| # Because it is always used inside 'generate()' after user message. | |
| chatName = chatHistory[0]['content'][0]['text'] | |
| # print(chatName) | |
| for idx, i in enumerate(pastChats): | |
| if i['chatName'] == chatName: | |
| # gr.Warning("That chat already exists.") | |
| return pastChats, chatHistory | |
| elif i['chatName'] == 'New Chat': | |
| pastChats[idx] = {'chatName': chatName, 'chatHistory': chatHistory} | |
| return pastChats, chatHistory | |
| pastChats.append({'chatName': chatName, 'chatHistory': chatHistory}) | |
| return pastChats, chatHistory | |
| def __openNewChat(pastChats: list[dict]): | |
| pastChats.append({'chatName': 'New Chat','chatHistory': []}) | |
| return pastChats, [] | |
| gr.Row(scale=0, min_height=40) | |
| with gr.Column(): | |
| gr.Markdown("<center><h3>------------------ Settings ---------------------</h3><center>") | |
| sysPromptTextBox = gr.Textbox(label="System Prompt", | |
| value="You are an helpful assistant.") | |
| with gr.Row(): | |
| modelSwitchDropdown = gr.Dropdown(label="Available Models:", scale=1, | |
| choices=[ | |
| "llama-3.1-8b-instant", | |
| "llama-3.3-70b-versatile", | |
| "openai/gpt-oss-20b", | |
| "openai/gpt-oss-120b", | |
| "meta-llama/llama-4-scout-17b-16e-instruct", | |
| "moonshotai/kimi-k2-instruct-0905", | |
| "qwen/qwen3-32b" | |
| ], | |
| value="openai/gpt-oss-120b") | |
| tempSlider = gr.Slider(label="Temperature", scale=3, | |
| minimum=0, maximum=2.0, value=1.0, step=0.1) | |
| with gr.Row(): | |
| streamCheckbox = gr.Checkbox(label="Streaming Output", value=False, scale=1) | |
| contextLenSlider = gr.Slider(label="Context Length(no. of messages)",scale=3, | |
| minimum=6, maximum=48, value=24, step=2) | |
| clearChatBtn = gr.Button(value="Reset Chat") | |
| msg.submit(fn=generate, | |
| inputs=[ | |
| msg, | |
| chatbot, | |
| streamCheckbox, | |
| contextLenSlider, | |
| sysPromptTextBox, | |
| tempSlider, | |
| modelSwitchDropdown, | |
| chats], outputs=[msg,chatbot, msg, chats]) | |
| exportChatBtn.click(fn=exportChat, inputs=[chatbot]) | |
| importChatBtn.upload(fn=importChat, inputs=[importChatBtn], outputs=[chatbot]) | |
| clearChatBtn.click(fn=clearChatStatement, outputs=[chatbot]) | |
| regenerateBtn.click(fn=regenerateResponse, inputs=[chatbot], outputs=[chatbot]) | |
| newChatBtn.click(fn=__openNewChat, inputs=[chats], outputs=[chats, chatbot]) | |
| demo.launch(theme=themes.Citrus(), share=True, | |
| css=""" | |
| .chat-btn { | |
| display: block; | |
| width: 100%; | |
| text-align: left; | |
| padding: 6px 10px; | |
| border-radius: 6px; | |
| background-color: transparent; | |
| border: 1px solid rgba(255, 255, 255, 0.15); | |
| color: rgba(255, 255, 255, 0.9); /* FIXES invisible text */ | |
| font-size: 13px; | |
| transition: all 0.15s ease; | |
| cursor: pointer; | |
| white-space: nowrap; | |
| overflow: hidden; | |
| text-overflow: ellipsis; | |
| } | |
| .chat-btn:hover { | |
| border-color: rgba(255, 255, 255, 0.35); | |
| background-color: rgba(255, 255, 255, 0.05); | |
| } | |
| .chat-btn:active { | |
| background-color: rgba(255, 255, 255, 0.1); | |
| } | |
| """) |