# 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("