agi_gui.py

"""
Simple chat GUI for the project using customtkinter.

Features:
- Chat history area
- Input field + Send button
- Clear history button
- Status label to show when the agent is thinking
- Uses a background thread to call agent.run so the UI stays responsive

Usage: python agi_gui.py

This file expects the project-local Agent and AgentConfig to exist at
`agi.sophos` and `agi.agent_config`, and `all_tools` at `agi.sophos_tools`.
If `customtkinter` is not installed, the UI will show a helpful error.
"""

import threading
import queue
import time
import traceback
try:
	import customtkinter as ctk
except Exception:
	ctk = None

from datetime import datetime

if ctk:
	ctk.set_appearance_mode("System")
	ctk.set_default_color_theme("blue")


def format_message(role: str, text: str) -> str:
	ts = datetime.now().strftime("%H:%M:%S")
	return f"[{ts}] {role}: {text}\n\n"


class AGIChatApp:
	def __init__(self, root):
		self.root = root
		self.root.title("AGI Chat")
		self.root.geometry("800x600")

		# message queue for thread -> UI communication
		self._q = queue.Queue()

		# top frame: history
		self.history = ctk.CTkTextbox(master=root, width=760, height=420)
		self.history.configure(state="disabled")
		self.history.pack(padx=20, pady=(20, 6))

		# status and controls
		ctrl_frame = ctk.CTkFrame(master=root)
		ctrl_frame.pack(fill="x", padx=20, pady=(0, 6))

		self.status_label = ctk.CTkLabel(master=ctrl_frame, text="Ready")
		self.status_label.pack(side="left", padx=(8, 6))

		clear_btn = ctk.CTkButton(master=ctrl_frame, text="Clear", width=80, command=self.clear_history)
		clear_btn.pack(side="right", padx=(6, 8))

		# bottom frame: entry and send
		bottom = ctk.CTkFrame(master=root)
		bottom.pack(fill="x", padx=20, pady=(0, 20))

		self.entry = ctk.CTkEntry(master=bottom, placeholder_text="Type your message here...")
		self.entry.pack(side="left", fill="x", expand=True, padx=(6, 6), pady=6)
		self.entry.bind("<Return>", self._on_enter)

		send_btn = ctk.CTkButton(master=bottom, text="Send", width=120, command=self.on_send)
		send_btn.pack(side="right", padx=(6, 6), pady=6)

		# Agent will be lazy-initialized on first send to avoid long startup at UI open
		self.agent = None
		self.agent_lock = threading.Lock()

		# periodically poll the queue for results from worker threads
		self.root.after(100, self._poll_queue)

	def _append_history(self, text: str):
		self.history.configure(state="normal")
		self.history.insert("end", text)
		self.history.see("end")
		self.history.configure(state="disabled")

	def clear_history(self):
		self.history.configure(state="normal")
		self.history.delete("0.0", "end")
		self.history.configure(state="disabled")
		self.status_label.configure(text="Cleared")

	def _on_enter(self, event):
		# prevent the default newline insertion
		self.on_send()
		return "break"

	def on_send(self):
		user_text = self.entry.get().strip()
		if not user_text:
			return
		self.entry.delete(0, "end")
		self._append_history(format_message("User", user_text))
		self.status_label.configure(text="Thinking...")

		# start background thread to call agent
		t = threading.Thread(target=self._background_agent_call, args=(user_text,), daemon=True)
		t.start()

	def _init_agent(self):
		# only run import and initialization once, in guarded block
		if self.agent is not None:
			return
		try:
			# local project imports
			from agi.sophos import Agent
			from agi.agent_config import AgentConfig
			from agi.sophos_tools import all_tools

			config = AgentConfig(
				name="Sophos-GUI",
				instructions="You are an AI agent.",
				max_iterations=30,
				max_tokens=1500,
				temperature=0.7,
				verbose=False,
				enable_memory=True,
				memory_limit=5,
			)

			toolbox = all_tools()
			# instantiate Agent
			self.agent = Agent(config=config, tools=toolbox)
			self._append_history(format_message("System", "Agent initialized."))
		except Exception as e:
			tb = traceback.format_exc()
			self._append_history(format_message("Error", f"Failed to initialize agent: {e}\n{tb}"))
			self.status_label.configure(text="Agent init failed")

	def _background_agent_call(self, prompt: str):
		# ensure agent is ready
		with self.agent_lock:
			if self.agent is None:
				self._init_agent()

		if self.agent is None:
			# initialization failed
			self._q.put(("error", "Agent not available"))
			return

		try:
			start = time.time()
			# call agent.run which may be blocking
			resp = self.agent.run(prompt)
			elapsed = time.time() - start
			self._q.put(("response", resp, elapsed))
		except Exception as e:
			tb = traceback.format_exc()
			self._q.put(("error", f"{e}\n{tb}"))

	def _poll_queue(self):
		try:
			while True:
				item = self._q.get_nowait()
				if not item:
					continue
				kind = item[0]
				if kind == "response":
					_, resp, elapsed = item
					self._append_history(format_message("Agent", resp))
					self.status_label.configure(text=f"Done ({elapsed:.1f}s)")
				elif kind == "error":
					_, err = item
					self._append_history(format_message("Error", str(err)))
					self.status_label.configure(text="Error")
				else:
					# unknown message
					self._append_history(format_message("Debug", str(item)))
		except queue.Empty:
			pass
		finally:
			self.root.after(100, self._poll_queue)


def main():
	if ctk is None:
		print("customtkinter is not installed. Install it with: pip install customtkinter")
		# Fallback to a basic tkinter popup that informs the user
		try:
			import tkinter as tk
			from tkinter import messagebox
			root = tk.Tk()
			root.withdraw()
			messagebox.showerror("Missing dependency", "customtkinter is not installed.\nRun: pip install customtkinter")
		except Exception:
			pass
		return

	root = ctk.CTk()
	app = AGIChatApp(root)
	root.mainloop()


if __name__ == "__main__":
	main()