diff --git a/deck/llm.py b/deck/llm.py new file mode 100644 index 0000000..3003448 --- /dev/null +++ b/deck/llm.py @@ -0,0 +1,126 @@ +# deck.llm + +import json + +# 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() + if status: + print("We're online, initing requests lib...") + requests = net.init_requests() +except: + import requests + +class LLMChat: + """ + Class for chat-like LLM interactions + """ + def __init__(self, config_file='llmcfg.json', message_limit=10): + self.config = self.load_config(config_file) + self.config_file = config_file + self.message_limit = message_limit + self.messages = [] + self.setup_provider() + self.reset_context() + + def load_config(self, config_file='llmcfg.json'): + """ + 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(self.config_file, 'w') as f: + json.dump(self.config, f) + return (True, "Configuration saved.") + except OSError as e: + return (False, 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: + return (False, f"Provider {active_name} not found.") + + 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) + return (True, "Provider ready") + + def reset_context(self): + """ + Clears history and re-adds system prompt. + """ + self.messages = [{"role": "system", "content": self.config.get('system_prompt', '')}] + return (True, "Context cleared") + + def get_completion(self, user_input): + """ + Sends request to LLM. + """ + self.messages.append({"role": "user", "content": user_input}) + + if len(self.messages) > (self.message_limit + 1): + self.messages = [self.messages[0]] + self.messages[-self.message_limit:] + + payload = { + "model": self.provider.get('model'), + "messages": self.messages, + "temperature": self.config.get('temperature', 0.7), + "stream": False + } + + url = f"{self.provider.get('base_url').rstrip('/')}/chat/completions" + + 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 (True, bot_content) + except Exception as e: + if len(self.messages) > 0: self.messages.pop() + return (False, f"Error: {e}") + + def list_models(self): + """ + Lists available models for the active provider. + """ + url = f"{self.provider.get('base_url').rstrip('/')}/models" + try: + response = requests.get(url, headers=self.headers) + data = response.json() + response.close() + res = [] + if 'data' in data: + res = [m.get('id') for m in data['data']] + return True, res + except Exception as e: + return False, f"Error: {e}" diff --git a/llmchat.py b/llmchat.py index 405b478..04f0bcb 100644 --- a/llmchat.py +++ b/llmchat.py @@ -4,185 +4,89 @@ # 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 +from deck.chat import DeckChat +from deck.llm import LLMChat CONFIG_FILE = 'llmcfg.json' -class LLMChat: - def __init__(self): - self.config = self.load_config() - self.messages = [] - self.setup_provider() - self.reset_context() +# chat handlers - 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 send_msg(msg, state): + print("Please wait for the answer...") + status, ans = state.get_completion(msg) + return ans, state - 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 set_sys_prompt(prompt, state): + if prompt: + state.config['system_prompt'] = prompt + state.save_config() + state.reset_context() + return 'System prompt updated', state + else: + return f"System: {state.config['system_prompt']}", state - 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 set_temp(temp, state): + if temp: + state.config['temperature'] = float(temp) + state.save_config() + return 'Temperature updated', state + else: + return f"Temperature: {state.config['temperature']}", state - 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 set_prov(prov, state): + if prov and prov in state.config.get('providers', {}): + state.config['active_provider'] = prov + state.setup_provider() + state.save_config() + state.reset_context() + return 'Provider updated', state + else: + return f"Active provider: {state.config['active_provider']}", state - 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 set_model(model, state): + if model: + state.provider['model'] = model + state.save_config() + return 'Model updated', state + else: + return f"Active model: {state.provider.get('model')}", state - 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:] +def list_models(stub, state): + status, mlist = state.list_models() + if status: + resp = f"Model list for {state.config['active_provider']}:" + for m in mlist: + resp += f'\n- {m}' + else: + resp = mlist + return resp, state - payload = { - "model": self.provider.get('model'), - "messages": self.messages, - "temperature": self.config.get('temperature', 0.7) - } +def list_providers(stub, state): + resp = 'Provider list:' + for p in state.config.get('providers', {}): + resp += f'\n - {p}' + return resp, state - 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 display_help(stub, state): + resp = "/clear, /system, /temp, /prov, /model, /modellist, /provlist, /exit" + return resp, state - 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 clear_ctx(stub, state): + state.reset_context() + return 'Context reset', state - 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() +# main initialization +llm = LLMChat(config_file=CONFIG_FILE, message_limit=20) +chat = DeckChat(start_state=llm, chat_prefix='> ', paginate=8) +chat.command('default', send_msg) +chat.command('clear', clear_ctx) +chat.command('system', set_sys_prompt) +chat.command('temp', set_temp) +chat.command('prov', set_prov) +chat.command('model', set_model) +chat.command('modellist', list_models) +chat.command('provlist', list_providers) +chat.command('help', display_help) +chat.command('exit', lambda m,s: ('Exiting', None)) +chat.command('quit', lambda m,s: ('Exiting', None)) +chat.start()