From d131b05e8c12cf74bcdb8d06da4ac7d5cb976f93 Mon Sep 17 00:00:00 2001 From: Luxferre Date: Sat, 20 Dec 2025 18:28:43 +0200 Subject: [PATCH] implemented simple LLM chat --- nethelper.py => deck/net.py | 14 +-- llmchat.py | 188 ++++++++++++++++++++++++++++++++++++ 2 files changed, 195 insertions(+), 7 deletions(-) rename nethelper.py => deck/net.py (88%) create mode 100644 llmchat.py diff --git a/nethelper.py b/deck/net.py similarity index 88% rename from nethelper.py rename to deck/net.py index 19e907b..0b4c5ab 100644 --- a/nethelper.py +++ b/deck/net.py @@ -1,13 +1,13 @@ -# T-Deck network helper library for CircuitPython +# deck.net: T-Deck network helper library for CircuitPython # Depends on some Adafruit Bundle libs # Normal flow (provided Wi-Fi creds are in settings.toml): # -# import nethelper -# status, stext, myip = nethelper.wifi_connect_wait() +# from deck import net +# status, stext, myip = net.wifi_connect_wait() # if status: -# requests = nethelper.init_requests() -# # or socket = nethelper.init_socket() if raw sockets are necessary -# while nethelper.my_ip(): +# requests = net.init_requests() +# # or socket = net.init_socket() if raw sockets are necessary +# while net.my_ip(): # ... # # Created by Luxferre in 2025, released into public domain @@ -39,7 +39,7 @@ def wifi_connect_wait(ssid=None, password=None, tmout=15, retries=3): Usually, you'll want to use this function to connect ''' r = 0 - status = False + status = wifi.radio.connected stext = '' while not wifi.radio.ipv4_address and r < retries: status, stext = wifi_connect(ssid, password) diff --git a/llmchat.py b/llmchat.py new file mode 100644 index 0000000..405b478 --- /dev/null +++ b/llmchat.py @@ -0,0 +1,188 @@ +# LLM chat for T-Deck CicruitPython version +# (also can be run on usual CPython) +# On the T-Deck, requires to be run from tdeck_repl for the +# keyboard to work correctly +# Created by Luxferre in 2025, released into public domain + +import sys, os, json + +# init the tdeck_repl input +try: + from tdeck_repl import input +except: + pass + +# init requests library in a platform-agnostic way +requests = None +try: + from deck import net + print('Waiting for online status...') + status, stext, myip = net.wifi_connect_wait() + print(status, stext, myip) + if status: + print("We're online, initing requests lib...") + requests = net.init_requests() +except: + import requests + +CONFIG_FILE = 'llmcfg.json' + +class LLMChat: + def __init__(self): + self.config = self.load_config() + self.messages = [] + self.setup_provider() + self.reset_context() + + def load_config(self): + """Loads config from JSON file with CircuitPython fallback.""" + try: + with open(CONFIG_FILE, 'r') as f: + return json.load(f) + except Exception as e: + print(f"Error: Could not load {CONFIG_FILE} ({e})") + # Minimal fallback for embedded if file is missing + return { + "system_prompt": "Concise assistant", + "temperature": 0.7, + "active_provider": "local", + "providers": {"pollinations": {"base_url": "https://text.pollinations.ai/openai", "api_key": "", "model": "openai-fast"}} + } + + def save_config(self): + """Saves config. Note: May fail on CircuitPython if USB is connected.""" + try: + with open(CONFIG_FILE, 'w') as f: + json.dump(self.config, f) + print("Configuration saved.") + except OSError as e: + print(f"Save failed: {e}. (Is filesystem read-only?)") + + def setup_provider(self): + """Sets up headers and provider settings.""" + active_name = self.config.get('active_provider') + providers = self.config.get('providers', {}) + + if active_name not in providers: + print(f"Provider {active_name} not found.") + return + + self.provider = providers[active_name] + self.headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.provider.get('api_key', '')}" + } + extra = self.provider.get('extra_headers', {}) + if extra: self.headers.update(extra) + + def reset_context(self): + """Clears history and re-adds system prompt.""" + self.messages = [{"role": "system", "content": self.config.get('system_prompt', '')}] + print("Context cleared.") + + def paginate_output(self, text, lines_per_page=8): + """Splits output for small terminals.""" + lines = text.split('\n') + for i in range(0, len(lines), lines_per_page): + chunk = lines[i:i + lines_per_page] + for line in chunk: print(line) + if i + lines_per_page < len(lines): + input("-- Press Enter --") + + def get_completion(self, user_input): + """Sends request to LLM.""" + self.messages.append({"role": "user", "content": user_input}) + + # Memory safety for embedded: keep only last 10 messages + if len(self.messages) > 11: + self.messages = [self.messages[0]] + self.messages[-10:] + + payload = { + "model": self.provider.get('model'), + "messages": self.messages, + "temperature": self.config.get('temperature', 0.7) + } + + url = f"{self.provider.get('base_url').rstrip('/')}/chat/completions" + print("Please wait for the answer...") + + try: + response = requests.post(url, headers=self.headers, json=payload) + data = response.json() + response.close() # Important for CircuitPython memory + + bot_content = data['choices'][0]['message']['content'] + self.messages.append({"role": "assistant", "content": bot_content}) + return bot_content + except Exception as e: + if len(self.messages) > 0: self.messages.pop() + return f"Error: {e}" + + def list_models(self): + """Lists available models.""" + url = f"{self.provider.get('base_url').rstrip('/')}/models" + try: + response = requests.get(url, headers=self.headers) + data = response.json() + response.close() + print(f"Models for {self.config['active_provider']}:") + if 'data' in data: + for m in data['data']: print(f" - {m.get('id')}") + except Exception as e: + print(f"Error: {e}") + + def run(self): + print(f"LLM CLI | {self.config.get('active_provider')} | {self.provider.get('model')}") + + while True: + try: + user_input = input("\n>> ").strip() + if not user_input: continue + + if user_input.startswith('/'): + parts = user_input.split(maxsplit=1) + cmd, arg = parts[0].lower(), parts[1] if len(parts) > 1 else None + + if cmd in ['/exit', '/quit']: break + elif cmd == '/clear': self.reset_context() + elif cmd == '/system': + if arg: + self.config['system_prompt'] = arg + self.save_config() + self.reset_context() + else: print(f"System: {self.config['system_prompt']}") + elif cmd == '/temp': + if arg: + self.config['temperature'] = float(arg) + self.save_config() + else: print(f"Temp: {self.config['temperature']}") + elif cmd == '/prov': + if arg and arg in self.config.get('providers', {}): + self.config['active_provider'] = arg + self.setup_provider() + self.save_config() + self.reset_context() + else: print(f"Active: {self.config['active_provider']}") + elif cmd == '/model': + if arg: + self.provider['model'] = arg + self.save_config() + else: print(f"Model: {self.provider.get('model')}") + elif cmd == '/modellist': self.list_models() + elif cmd == '/provlist': + for p in self.config.get('providers', {}): print(f" - {p}") + elif cmd == '/help': + print("/clear, /system, /temp, /prov, /model, /modellist, /provlist, /exit") + else: print("Unknown command.") + continue + + ans = self.get_completion(user_input) + print("") + self.paginate_output(ans) + + except KeyboardInterrupt: break + print("\nExiting.") + +# no main, direct init +chat = LLMChat() +chat.run()