refactored llmchat
This commit is contained in:
Luxferre
+126
@@ -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}"
|
||||
+73
-169
@@ -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', {})
|
||||
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
|
||||
|
||||
if active_name not in providers:
|
||||
print(f"Provider {active_name} not found.")
|
||||
return
|
||||
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
|
||||
|
||||
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_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 reset_context(self):
|
||||
"""Clears history and re-adds system prompt."""
|
||||
self.messages = [{"role": "system", "content": self.config.get('system_prompt', '')}]
|
||||
print("Context cleared.")
|
||||
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
|
||||
|
||||
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 list_providers(stub, state):
|
||||
resp = 'Provider list:'
|
||||
for p in state.config.get('providers', {}):
|
||||
resp += f'\n - {p}'
|
||||
return resp, state
|
||||
|
||||
def get_completion(self, user_input):
|
||||
"""Sends request to LLM."""
|
||||
self.messages.append({"role": "user", "content": user_input})
|
||||
def display_help(stub, state):
|
||||
resp = "/clear, /system, /temp, /prov, /model, /modellist, /provlist, /exit"
|
||||
return resp, state
|
||||
|
||||
# Memory safety for embedded: keep only last 10 messages
|
||||
if len(self.messages) > 11:
|
||||
self.messages = [self.messages[0]] + self.messages[-10:]
|
||||
def clear_ctx(stub, state):
|
||||
state.reset_context()
|
||||
return 'Context reset', state
|
||||
|
||||
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()
|
||||
# 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()
|
||||
|
||||
Reference in New Issue
Block a user