import os import asyncio import time import tkinter as tk from tkinter import scrolledtext, messagebox, filedialog import threading from dotenv import load_dotenv import openai # Load environment variables load_dotenv() openai.api_key = os.getenv("OPENAI_API_KEY") ASSISTANT_ID = os.getenv("CODETTE_ASSISTANT_ID", "asst_xxx") # INSERT YOUR ASSISTANT ID HERE class CodetteApp(tk.Tk): def __init__(self): super().__init__() self.title("Codette Universal Reasoning Assistant") self.geometry("900x600") self.configure(bg="#eef6f9") self.resizable(True, True) # Welcome banner banner = tk.Label(self, text="Ask Codette", font=("Helvetica", 21, "bold"), bg="#3e75c3", fg="#fafafa", padx=10, pady=14) banner.pack(fill=tk.X) self._setup_controls() self._setup_output_box() self._setup_input_controls() self.chat_log = [] self.output_box.focus() self.append_chat("Welcome to Codette! 🧠\n(type your question and press Enter or 'Ask')", who="system") self.protocol("WM_DELETE_WINDOW", self.on_exit) def _setup_controls(self): btn_frame = tk.Frame(self, bg="#eef6f9") btn_frame.pack(anchor=tk.NE, pady=7, padx=10) tk.Button(btn_frame, text="Export Chat", command=self.export_chat, font=("Calibri", 11)).pack(side=tk.LEFT, padx=6) tk.Button(btn_frame, text="Clear", command=self.clear_all, font=("Calibri", 11)).pack(side=tk.LEFT, padx=6) tk.Button(btn_frame, text="Exit", command=self.on_exit, font=("Calibri", 11)).pack(side=tk.LEFT, padx=6) def _setup_output_box(self): self.output_frame = tk.Frame(self, bg="#eef6f9") self.output_frame.pack(expand=True, fill=tk.BOTH, padx=14, pady=2) self.output_box = scrolledtext.ScrolledText( self.output_frame, font=("Consolas", 13), bg="#fcfcfc", wrap=tk.WORD, state="disabled", padx=10, pady=8, borderwidth=2, relief=tk.GROOVE) self.output_box.pack(fill=tk.BOTH, expand=True) self.output_box.tag_config('user', foreground='#0d47a1', font=('Arial', 12, 'bold')) self.output_box.tag_config('ai', foreground='#357a38', font=('Arial', 12, 'italic')) self.output_box.tag_config('time', foreground='#ad1457', font=('Arial', 9, 'italic')) self.output_box.tag_config('system', foreground='#808080', font=('Arial', 10, 'italic')) def _setup_input_controls(self): user_frame = tk.Frame(self, bg="#eef6f9") user_frame.pack(side=tk.BOTTOM, fill=tk.X, pady=(1,10), padx=10) self.input_field = tk.Entry(user_frame, font=("Calibri", 15)) self.input_field.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 10), ipady=6) self.input_field.bind("", lambda event: self.handle_ask()) tk.Button(user_frame, text="Ask", font=("Calibri", 13), bg="#357a38", fg="white", command=self.handle_ask).pack(side=tk.LEFT) self.input_field.focus() def append_chat(self, text, who="user", timestamp=None): self.output_box.configure(state='normal') if not timestamp: timestamp = time.strftime('%Y-%m-%d %H:%M:%S') if who == "user": self.output_box.insert(tk.END, f"[{timestamp}] You: ", ('user', 'time')) self.output_box.insert(tk.END, text + "\n", 'user') elif who == "ai": self.output_box.insert(tk.END, f"[{timestamp}] Codette: ", ('ai', 'time')) self.output_box.insert(tk.END, text + "\n\n", 'ai') elif who == "system": self.output_box.insert(tk.END, f"[{timestamp}] SYSTEM: {text}\n", 'system') self.output_box.see(tk.END) self.output_box.configure(state='disabled') self.chat_log.append((timestamp, who, text.strip())) def handle_ask(self): user_query = self.input_field.get().strip() if not user_query: messagebox.showwarning("Input Required", "Please enter your question.") return self.append_chat(user_query, 'user') self.input_field.delete(0, tk.END) self.input_field.focus() threading.Thread(target=self.fetch_codette, args=(user_query,), daemon=True).start() def fetch_codette(self, user_query): try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) resp = loop.run_until_complete(assistant_thread_chat(user_query)) self.append_chat(resp, "ai") except Exception as e: self.append_chat(f"❗️Error: {e}", "system") def clear_all(self): self.output_box.configure(state='normal') self.output_box.delete('1.0', tk.END) self.output_box.configure(state='disabled') self.chat_log = [] self.append_chat("Chat cleared.", "system") self.input_field.focus() def export_chat(self): file_path = filedialog.asksaveasfilename( title="Export Chat", defaultextension=".txt", filetypes=[('Text files', '*.txt')] ) if file_path: with open(file_path, "w", encoding="utf-8") as f: for (timestamp, who, text) in self.chat_log: label = "You:" if who == "user" else "Codette:" if who == "ai" else "SYSTEM:" f.write(f"[{timestamp}] {label} {text}\n") self.append_chat(f"Exported chat log to: {file_path}", "system") def on_exit(self): self.destroy() async def assistant_thread_chat(prompt: str) -> str: try: # 1. Create a thread for this context thread = await asyncio.to_thread(lambda: openai.beta.threads.create()) # 2. Add the user message _ = await asyncio.to_thread( lambda: openai.beta.threads.messages.create( thread_id=thread.id, role="user", content=prompt ) ) # 3. Start a run with your Codette assistant run = await asyncio.to_thread( lambda: openai.beta.threads.runs.create( thread_id=thread.id, assistant_id=ASSISTANT_ID ) ) # 4. Poll until complete while run.status in ['queued', 'in_progress', 'cancelling']: await asyncio.sleep(1) run = await asyncio.to_thread( lambda: openai.beta.threads.runs.retrieve( thread_id=thread.id, run_id=run.id ) ) if run.status == "completed": messages = await asyncio.to_thread( lambda: openai.beta.threads.messages.list(thread_id=thread.id) ) # Get the latest assistant reply for msg in reversed(messages.data): if msg.role == "assistant": return msg.content + " 😊" return "[No assistant response found]" elif run.status == "requires_action": return "[ACTION REQUIRED: Tool/function call not yet implemented.]" else: return f"[ERROR: Run status {run.status}]" except Exception as e: return f"Sorry—Codette encountered an error: {e}" if __name__ == "__main__": app = CodetteApp() app.mainloop()