File size: 7,341 Bytes
7f5ef51 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
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("<Return>", 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()
|