Assistant
Assistant
Dieses Projekt ist ein vollständig autonomer, lokaler Sprachassistent, der ein hybrides RAG-System (Retrieval-Augmented Generation) mit Echtzeit-Sprachverarbeitung kombiniert. Es wurde speziell für den reinen CPU-Betrieb entwickelt und bietet eine dynamische Unterbrechungsfunktion (Async Interruption).
Die Kernaspekte des Projekts:
Datenbasis & S-Expressions (sexpdata): Die Wissensbasis ist in Lisp-S-Expressions (knowledge.lisp) strukturiert. Der Parser liest diese Daten nativ in Python-Strukturen ein.
Hybrides RAG via TST (Ternary Search Tree): Keywords und Inhaltswörter werden in einem Präfixbaum (TST) indiziert. Bei Anfragen sucht das System blitzschnell relevante Fakten heraus und stellt diese als Kontext bereit.
LLM-Integration (Ollama & Qwen2.5): Die gefilterten Fakten werden an ein lokales Sprachmodell (Qwen2.5) übergeben, um präzise, halluzinationsfreie Antworten zu generieren.
STT (Faster-Whisper): Läuft lokal im CPU-Modus mit int8-Quantisierung für maximale Geschwindigkeit.
TTS (Coqui TTS): Synthetisiert die Antwort ohne GPU-Unterstützung.
Asynchrone Unterbrechung (Interrupt-Handling): Ein Hintergrund-Thread hört das Mikrofon kontinuierlich ab. Sobald der Nutzer eine neue Anweisung spricht, wird die laufende LLM-Generierung gestoppt und die Audioausgabe sofort abgebrochen, um die neue Aufgabe ohne Verzögerung zu verarbeiten.
Präfix-Navigation (Suchalgorithmus)
Wenn Sie in einem TST nach einem Wort oder einem Teil davon suchen, folgt der Algorithmus strikt den mittleren Zeigern (middle), die der Zeichenfolge vom Wortanfang an entsprechen:
- Um Wörter zu finden, die mit „arh“ beginnen, durchläuft der Algorithmus die Hierarchie: a --> r --> h.
- Sobald er den letzten Buchstaben des eingegebenen Präfixes (h) erreicht, befindet er sich an der Wurzel eines Teilbaums, in dem garantiert alle darunterliegenden Knoten – ausnahmslos – Wörter enthalten, die genau mit diesem Präfix beginnen.
Datenerfassung von oben nach unten (Verschiebung zum Anfang)
Nach dem Auffinden des Präfix-Knotens wird die rekursive Funktion zum Sammeln der Ergebnisse (_collect_all) aufgerufen:
def _collect_all(self, node):
if not node:
return []
res = list(node.data) # 1. Zunächst werden die Daten des AKTUELLEN (kürzesten) Wortes herangezogen
res.extend(self._collect_all(node.left))
res.extend(self._collect_all(node.middle)) # 2. Anschließend längere Fortsetzungen
res.extend(self._collect_all(node.right))
return res
Da node.data vor der tiefen Teilbaumtraversierung abgerufen wird, gilt:
- Zunächst erscheinen exakte Übereinstimmungen und kürzere Begriffe, deren Anfang übereinstimmt, in den Ergebnissen.
- Je tiefer sich ein Symbol im Baum befindet (näher am Ende eines langen Wortes), desto später wird es zusammengesetzt.
Ranking nach exakter Zeichenübereinstimmung
In Ihrer RAG-Pipeline wird die Suchanfrage in einzelne Wörter zerlegt, und jedes bereinigte Wort wird im TST ab dem ersten Zeichen gesucht:
- Wenn der Benutzer „Architektur“ ausspricht, gleicht TST die ersten Zeichen a-r-h... sofort mit den Anfängen der Schlüssel in der Wissensdatenbank ab.
- Wörter, die ab dem ersten Zeichen mit der Suchanfrage übereinstimmen, erhalten das maximale Gewicht (Score) und rücken relevante Fakten für Ollama an die Spitze der Liste.
GO!
portaudio & libsndfile
sudo sh -c "apt-get update && apt-get install -y portaudio19-dev libsndfile1"
Ollama & Qwen
curl -fsSL https://ollama.com/install.sh | sh
nohup ollama serve > /dev/null 2>&1 &
ollama pull qwen2.5:3b
requirements.txt
sexpdata
faster-whisper
TTS
sounddevice
soundfile
numpy
requests
knowledge.lisp
(
(:id "k1" :tags ("architektur" "rag") :content "Das System verwendet einen TST-Baum für eine schnelle hybride RAG-Suche.")
(:id "k2" :tags ("status" "prozessor") :content "Alle Whisper- und TTS-Komponenten sind ausschließlich für den Betrieb über die CPU konfiguriert.")
)
main.py
import os
import sys
import time
import queue
import tempfile
import threading
import requests
import sexpdata
import numpy as np
import sounddevice as sd
import soundfile as sf
from typing import List, Dict, Any
from faster_whisper import WhisperModel
from TTS.api import TTS
# ==========================================
# 1. TST (Ternary Search Tree) IMPLEMENTATION
# ==========================================
class TSTNode:
def __init__(self, char: str):
self.char = char
self.left = None
self.middle = None
self.right = None
self.is_end = False
self.data: List[Any] = []
class TernarySearchTree:
def __init__(self):
self.root = None
def insert(self, word: str, item_ref: Any):
self.root = self._insert(self.root, word, 0, item_ref)
def _insert(self, node, word, index, item_ref):
char = word[index]
if node is None:
node = TSTNode(char)
if char < node.char:
node.left = self._insert(node.left, word, index, item_ref)
elif char > node.char:
node.right = self._insert(node.right, word, index, item_ref)
else:
if index + 1 < len(word):
node.middle = self._insert(node.middle, word, index + 1, item_ref)
else:
node.is_end = True
if item_ref not in node.data:
node.data.append(item_ref)
return node
def search(self, word: str) -> List[Any]:
node = self._search(self.root, word, 0)
if not node:
return []
results = list(node.data)
if node.middle:
results.extend(self._collect_all(node.middle))
return results
def _search(self, node, word, index):
if not node:
return None
char = word[index]
if char < node.char:
return self._search(node.left, word, index)
elif char > node.char:
return self._search(node.right, word, index)
else:
if index + 1 == len(word):
return node
return self._search(node.middle, word, index + 1)
def _collect_all(self, node):
if not node:
return []
res = list(node.data)
res.extend(self._collect_all(node.left))
res.extend(self._collect_all(node.middle))
res.extend(self._collect_all(node.right))
return res
# ==========================================
# 2. LISP KNOWLEDGE LOADER
# ==========================================
def load_lisp_knowledge(filepath: str) -> List[Dict]:
if not os.path.exists(filepath):
raise FileNotFoundError(f"Knowledge file not found at: {filepath}")
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
parsed = sexpdata.loads(content)
items = []
for sexp in parsed:
flat_item = {}
iterator = iter(sexp[1:])
for key, val in zip(iterator, iterator):
key_str = str(key).lstrip(':')
if isinstance(val, list):
val = [str(v) for v in val]
else:
val = str(val)
flat_item[key_str] = val
items.append(flat_item)
return items
# ==========================================
# 3. HYBRID RAG PIPELINE
# ==========================================
class LispRAGPipeline:
def __init__(self, lisp_file: str):
self.tst = TernarySearchTree()
self.items = load_lisp_knowledge(lisp_file)
self._build_index()
def _build_index(self):
for item in self.items:
keywords = item.get('tags', [])
content_words = item.get('content', '').lower().split()
for kw in keywords + content_words:
cleaned_kw = ''.join(c for c in kw if c.isalnum())
if cleaned_kw:
self.tst.insert(cleaned_kw, item)
def retrieve_and_rank(self, query: str) -> List[Dict]:
query_words = query.lower().split()
scored_items = {}
for word in query_words:
cleaned = ''.join(c for c in word if c.isalnum())
if not cleaned:
continue
matched_items = self.tst.search(cleaned)
for item in matched_items:
item_id = item.get('id', str(hash(item.get('content', ''))))
if item_id not in scored_items:
scored_items[item_id] = {'item': item, 'score': 0}
scored_items[item_id]['score'] += 1
sorted_results = sorted(scored_items.values(), key=lambda x: x['score'], reverse=True)
return [res['item'] for res in sorted_results]
# ==========================================
# 4. QWEN LLM INTEGRATION (OLLAMA WITH STREAM & INTERRUPT)
# ==========================================
def ask_qwen(query: str, context_items: List[Dict], interrupt_event: threading.Event, model_name: str = "qwen2.5:3b") -> str:
if context_items:
context_str = "\n".join([f"- {item.get('content', '')}" for item in context_items])
prompt = f"Use ONLY the following verified facts to answer the user query:\nFACTS:\n{context_str}\n\nQUERY: {query}\nANSWER:"
else:
prompt = f"QUERY: {query}\nANSWER:"
try:
response = requests.post(
"http://localhost:11434/api/generate",
json={"model": model_name, "prompt": prompt, "stream": True},
stream=True,
timeout=30
)
full_text = ""
for line in response.iter_lines():
if interrupt_event.is_set():
print("\n[🤖 Qwen]: Unterbrechung der Generierung (neuer Befehl eingegangen)...")
response.close()
return ""
if line:
import json
chunk = json.loads(line.decode('utf-8'))
text = chunk.get("response", "")
full_text += text
if chunk.get("done", False):
break
return full_text.strip()
except Exception as e:
return f"Fehler bei der Verbindung zu Ollama: {e}"
# ==========================================
# 5. ASYNC & CPU VOICE INTERFACE
# ==========================================
class VoiceInterface:
def __init__(self, tts_model_name: str = "tts_models/en/ljspeech/vits"):
print("[*] Initialisierung des lokalen Whisper-Modells auf der CPU (int8)...")
self.whisper = WhisperModel("small", device="cpu", compute_type="int8")
print(f"[*] Initialisierung des Coqui TTS-Modells auf der CPU ({tts_model_name})...")
self.tts = TTS(model_name=tts_model_name, progress_bar=False, gpu=False)
self.audio_queue = queue.Queue()
self.interrupt_event = threading.Event()
print("[*] Sprachschnittstelle betriebsbereit!")
def start_listening_thread(self, chunk_duration: int = 5, fs: int = 16000):
def listen_loop():
print("\n[🎤 Kontinuierliches Abhören aktiv... Sprechen Sie jederzeit]")
num_samples = int(chunk_duration * fs)
while True:
audio = sd.rec(num_samples, samplerate=fs, channels=1, dtype='float32')
sd.wait()
if np.max(np.abs(audio)) > 0.03:
self.interrupt_event.set()
self.audio_queue.put(audio)
thread = threading.Thread(target=listen_loop, daemon=True)
thread.start()
def transcribe(self, audio_data: np.ndarray, fs: int = 16000) -> str:
temp_audio_path = tempfile.mktemp(suffix=".wav")
sf.write(temp_audio_path, audio_data, fs)
segments, _ = self.whisper.transcribe(temp_audio_path, beam_size=1, vad_filter=True)
text = " ".join([segment.text for segment in segments]).strip()
if os.path.exists(temp_audio_path):
os.remove(temp_audio_path)
return text
def speak(self, text: str):
if not text or self.interrupt_event.is_set():
return
print("[🔊 Sprachsynthese...]")
temp_audio_path = tempfile.mktemp(suffix=".wav")
self.tts.tts_to_file(text=text, file_path=temp_audio_path)
data, fs = sf.read(temp_audio_path)
print("[🔊 Wiedergabe der Antwort...]")
sd.play(data, fs)
while sd.get_stream().active:
if self.interrupt_event.is_set():
sd.stop()
print("\n[🔊 TTS]: Die Sprachausgabe wurde vom Benutzer unterbrochen.")
break
time.sleep(0.1)
if os.path.exists(temp_audio_path):
os.remove(temp_audio_path)
# ==========================================
# 6. MAIN APPLICATION ENTRY POINT
# ==========================================
if __name__ == "__main__":
knowledge_file = "knowledge.lisp"
print("=== Hybrid Lisp-TST RAG Voice Assistant (CPU + Async Interrupt) ===")
try:
rag = LispRAGPipeline(knowledge_file)
print(f"[+] {len(rag.items)} Wissenselemente aus {knowledge_file} geladen")
except Exception as e:
print(f"[-] Fehler beim Laden der Wissensdatenbank: {e}")
exit(1)
voice = VoiceInterface()
voice.start_listening_thread(chunk_duration=5)
print("\nDas System läuft. Sprechen Sie in das Mikrofon oder drücken Sie Strg+C zum Beenden.")
while True:
try:
audio_data = voice.audio_queue.get()
voice.interrupt_event.clear()
query = voice.transcribe(audio_data)
if not query:
continue
print(f"\n[🗣️ Erkannt]: {query}")
if "exit" in query.lower() or "beenden" in query.lower():
print("Anwendung wird beendet. Auf Wiedersehen!")
break
retrieved_items = rag.retrieve_and_rank(query)
print(f"[🔍 TST-Suche]: {len(retrieved_items)} passende Fakten gefunden.")
answer = ask_qwen(query, retrieved_items, voice.interrupt_event, model_name="qwen2.5:3b")
if answer and not voice.interrupt_event.is_set():
print(f"[🤖 Qwen-Antwort]: {answer}")
voice.speak(answer)
except KeyboardInterrupt:
print("\nVom Benutzer gestoppt.")
break
except Exception as e:
print(f"[-] Fehler in der Hauptschleife: {e}")
Ablauf:
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
python main.py
Benutzer (per Sprache):
"Welche Architektur wird für die RAG-Suche verwendet?"
Whisper (STT): Erkennt die Spracheingabe und wandelt sie in Text um:
"Welche Architektur wird für die RAG-Suche verwendet?"
TST Retrieval: Zerlegt die Anfrage in Wörter ("architektur", "rag", "suche") und sucht nach Übereinstimmungen im Ternary Search Tree. Findet den Eintrag mit :id "k1".
Prompt für Qwen: Erstellt die finale Anfrage an Ollama:
Use ONLY the following verified facts to answer the user query:
FACTS:
- Das System verwendet einen TST-Baum für eine schnelle hybride RAG-Suche.
QUERY: Welche Architektur wird für die RAG-Suche verwendet?
ANSWER:
Antwort des Systems (Konsolenausgabe und Coqui TTS):
[🤖 Qwen Answer]: "Für die RAG-Suche wird eine Architektur auf Basis eines TST-Baums (Ternary Search Tree) verwendet."
Wenn alles funktioniert (Es kann zu Problemen mit dem Mikrofon kommen; falls es nicht klappt, trennen Sie einfach das Audio-Interface und verwenden Sie die Tastatur), machen wir alles auf strukturierte Weise.
sudo -s
vi /etc/systemd/system/assistant.service
[Unit]
Description=Assistant with Qwen (Ollama)
After=network.target ollama.service
Wants=ollama.service
[Service]
Type=simple
User=your_user
WorkingDirectory=/path/to/your/app
ExecStart=/path/to/your/app/.venv/bin/python /path/to/your/app/main.py
Environment=XDG_RUNTIME_DIR=/run/user/1000
Environment=PULSE_SERVER=unix:/run/user/1000/pulse/native
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
systemctl daemon-reload
systemctl enable assistant.service
systemctl start assistant.service
exit
sudo systemctl status assistant.service
journalctl -u assistant.service -f
Und was, wenn wir anhand von Symptomen eine Krankheit – und folglich die Behandlungsmethoden – bestimmen müssen, wobei es eine Unmenge an „Wenns“ gibt!?
Damit die Wissensbasis im S-Expression-Format (Lisp) bedingte Verzweigungen (Wenn → Dann, Sonst) unterstützt, müssen die Lisp-Datenstruktur erweitert, der Wissens-Loader modifiziert und die RAG-Pipeline angepasst werden, um die Bedingungslogik zu interpretieren, bevor der Kontext an das LLM übermittelt wird.
knowledge.lisp
- Wir fügen die Schlüsselwort-Struktur :condition, :then, :else sowie die Verzweigung :nested-conditions (für verschachtelte Bedingungen) hinzu.
(
(:id "k1"
:tags ("architektur" "rag")
:content "Das System verwendet einen TST-Baum für die schnelle hybride RAG-Suche.")
(:id "k2"
:tags ("status" "prozessor")
:content "Alle Whisper- und TTS-Komponenten sind ausschließlich für den CPU-Betrieb konfiguriert.")
(:id "cond_cpu_load"
:tags ("last" "cpu" "leistung")
:condition "hohe_cpu_last"
:then "Whisper auf das Modell 'tiny' umstellen und die TTS-Abtastrate reduzieren."
:else "Das Standard-Whisper-Modell 'small' und die vollständigen VITS-Parameter verwenden.")
(:id "cond_mic_status"
:tags ("mikrofon" "audio" "fehler")
:condition "mikrofon_verfuegbar"
:then "Hintergrund-Audio-Stream mit sd.rec() starten."
:else "In den Texteingabemodus über die Standard-Konsole (STDIN) wechseln.")
(:id "cond_ollama_status"
:tags ("ollama" "llm" "netzwerk")
:condition "ollama_aktiv"
:then "Prompt erstellen und HTTP-POST-Anfrage an http://localhost:11434/api/generate senden."
:else "Lokale Fallback-Antwort basierend auf den gefundenen TST-Fakten ohne LLM-Generierung verwenden.")
;; Verschachtelte Bedingung (Nested Condition)
(:id "cond_nested_audio_pipeline"
:tags ("audio" "unterbrechung" "tts" "sprache")
:condition "benutzer_spricht"
:then (
(:condition "tts_wird_abgespielt"
:then "Aktuelle Sprachausgabe abbrechen (sd.stop()), Qwen-Generierungsstream zurücksetzen und Befehlsaufzeichnung starten."
:else "Warten beenden und den eingehenden Audio-Stream direkt an Whisper weiterleiten.")
)
:else "Normale Wiedergabe fortsetzen oder im VAD-Modus verbleiben.")
;; Regeln mit mehreren Bedingungen (Multi-condition AND)
(:id "cond_hardware_fallback"
:tags ("speicher" "ram" "fallback" "fehler")
:rules (
(:all_of ("hohe_cpu_last" "geringer_ram_speicher")
:action "Coqui TTS vollständig deaktivieren, auf Konsolentextausgabe umstellen und RAG-Kontext auf 2 Fakten reduzieren.")
(:all_of ("hohe_cpu_last" "ram_normal")
:action "Coqui TTS beibehalten, aber die Synthesequalität auf 16kHz reduzieren.")
)
:default "Im normalen Hochleistungsmodus arbeiten.")
;; Kaskadierende Auswahl (Switch / Case)
(:id "cond_ollama_error_handling"
:tags ("netzwerk" "timeout" "ollama" "fehler")
:cases (
(:when "ollama_status_500" :then "Lokalen Ollama-Dienst über `systemctl restart ollama` neu starten.")
(:when "ollama_timeout" :then "RAG-Kontextgröße reduzieren und Anfrage mit timeout=10 wiederholen.")
(:when "ollama_model_not_found" :then "`ollama pull qwen2.5:3b` in einem externen Prozess ausführen.")
)
:else "Unbekannten Netzwerkfehler in den Logs registrieren.")
)
main.py
- Rekursive Verarbeitung von Lisp-S-Expressions, die das Lesen von Zeichenketten, Listen und verschachtelten Konstrukten wie
:condition/:then/:elseermöglicht. - Logik zur Ausführung von Bedingungen (
eval_condition), die den übergebenen Systemstatus prüfen kann (oder die logischen Regeln direkt für das Reasoning an den Qwen-Prompt weiterleitet). - Aktualisierter TST-Indexer, der nicht nur
:contentindexiert, sondern auch die Texte aus den Zweigen:condition,:thenund:else.
import os
import sys
import time
import queue
import tempfile
import threading
import requests
import sexpdata
import numpy as np
import sounddevice as sd
import soundfile as sf
from typing import List, Dict, Any, Union
from faster_whisper import WhisperModel
from TTS.api import TTS
# ==========================================
# 1. TST (Ternary Search Tree) IMPLEMENTATION
# ==========================================
class TSTNode:
def __init__(self, char: str):
self.char = char
self.left = None
self.middle = None
self.right = None
self.is_end = False
self.data: List[Any] = []
class TernarySearchTree:
def __init__(self):
self.root = None
def insert(self, word: str, item_ref: Any):
self.root = self._insert(self.root, word, 0, item_ref)
def _insert(self, node, word, index, item_ref):
char = word[index]
if node is None:
node = TSTNode(char)
if char < node.char:
node.left = self._insert(node.left, word, index, item_ref)
elif char > node.char:
node.right = self._insert(node.right, word, index, item_ref)
else:
if index + 1 < len(word):
node.middle = self._insert(node.middle, word, index + 1, item_ref)
else:
node.is_end = True
if item_ref not in node.data:
node.data.append(item_ref)
return node
def search(self, word: str) -> List[Any]:
node = self._search(self.root, word, 0)
if not node:
return []
results = list(node.data)
if node.middle:
results.extend(self._collect_all(node.middle))
return results
def _search(self, node, word, index):
if not node:
return None
char = word[index]
if char < node.char:
return self._search(node.left, word, index)
elif char > node.char:
return self._search(node.right, word, index)
else:
if index + 1 == len(word):
return node
return self._search(node.middle, word, index + 1)
def _collect_all(self, node):
if not node:
return []
res = list(node.data)
res.extend(self._collect_all(node.left))
res.extend(self._collect_all(node.middle))
res.extend(self._collect_all(node.right))
return res
# ==========================================
# 2. RECURSIVE LISP KNOWLEDGE PARSER
# ==========================================
def _parse_sexp_node(sexp: Any) -> Union[Dict, List, str]:
if isinstance(sexp, list):
if len(sexp) > 0 and isinstance(sexp[0], sexpdata.Symbol) and str(sexp[0]).startswith(':'):
res = {}
iterator = iter(sexp)
for key in iterator:
try:
val = next(iterator)
except StopIteration:
break
key_str = str(key).lstrip(':')
res[key_str] = _parse_sexp_node(val)
return res
else:
return [_parse_sexp_node(item) for item in sexp]
elif isinstance(sexp, sexpdata.Symbol):
return str(sexp).lstrip(':')
else:
return str(sexp)
def load_lisp_knowledge(filepath: str) -> List[Dict]:
if not os.path.exists(filepath):
raise FileNotFoundError(f"Knowledge file not found at: {filepath}")
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
parsed = sexpdata.loads(content)
items = []
for sexp in parsed:
item = _parse_sexp_node(sexp)
if isinstance(item, dict):
items.append(item)
return items
# ==========================================
# 3. ADVANCED HYBRID RAG & RECURSIVE EVALUATOR
# ==========================================
class LispRAGPipeline:
def __init__(self, lisp_file: str):
self.tst = TernarySearchTree()
self.items = load_lisp_knowledge(lisp_file)
self._build_index()
def _extract_words_recursive(self, node: Any) -> List[str]:
words = []
if isinstance(node, str):
words.extend(node.lower().split())
elif isinstance(node, list):
for item in node:
words.extend(self._extract_words_recursive(item))
elif isinstance(node, dict):
for v in node.values():
words.extend(self._extract_words_recursive(v))
return words
def _build_index(self):
for item in self.items:
extracted_words = self._extract_words_recursive(item)
for w in extracted_words:
cleaned = ''.join(c for c in w if c.isalnum())
if cleaned:
self.tst.insert(cleaned, item)
def retrieve_and_rank(self, query: str) -> List[Dict]:
query_words = query.lower().split()
scored_items = {}
for word in query_words:
cleaned = ''.join(c for c in word if c.isalnum())
if not cleaned:
continue
matched_items = self.tst.search(cleaned)
for item in matched_items:
item_id = item.get('id', str(hash(str(item))))
if item_id not in scored_items:
scored_items[item_id] = {'item': item, 'score': 0}
scored_items[item_id]['score'] += 1
sorted_results = sorted(scored_items.values(), key=lambda x: x['score'], reverse=True)
return [res['item'] for res in sorted_results]
def format_item_to_text(self, item: Any, state_context: Dict[str, bool] = None) -> str:
"""Entfaltet rekursiv beliebige Strukturen (IF-THEN-ELSE, RULES, CASES) zu formatiertem Text."""
if isinstance(item, str):
return item
if isinstance(item, list):
return " ".join([self.format_item_to_text(i, state_context) for i in item])
if isinstance(item, dict):
# 1. Bearbeitung IF-THEN-ELSE
if 'condition' in item:
cond = item['condition']
then_b = self.format_item_to_text(item.get('then', ''), state_context)
else_b = self.format_item_to_text(item.get('else', ''), state_context)
if state_context and cond in state_context:
return then_b if state_context[cond] else else_b
return f"WENN ({cond}) DANN [{then_b}] SONST [{else_b}]"
# 2. Bearbeitung RULES (Multi-condition AND)
if 'rules' in item:
rules_txt = []
rules_list = item['rules'] if isinstance(item['rules'], list) else [item['rules']]
for r in rules_list:
if isinstance(r, dict):
all_of = r.get('all_of', [])
conds = " UND ".join(all_of) if isinstance(all_of, list) else str(all_of)
act = self.format_item_to_text(r.get('action', ''), state_context)
rules_txt.append(f"BEI ({conds}) -> {act}")
def_act = self.format_item_to_text(item.get('default', ''), state_context)
return "REGELN: " + "; ".join(rules_txt) + f" (BEI DEFAULT: {def_act})"
# 3. Bearbeitung CASES (Switch/Case)
if 'cases' in item:
cases_txt = []
cases_list = item['cases'] if isinstance(item['cases'], list) else [item['cases']]
for c in cases_list:
if isinstance(c, dict):
when = c.get('when', '')
then_act = self.format_item_to_text(c.get('then', ''), state_context)
cases_txt.append(f"WENN ({when}) -> {then_act}")
else_act = self.format_item_to_text(item.get('else', ''), state_context)
return "OPTIONS: " + " | ".join(cases_txt) + f" (DANN: {else_act})"
# 4. Eine einfache Tatsache
if 'content' in item:
return f"TATSACHE: {item['content']}"
return str(item)
def format_context_for_llm(self, items: List[Dict], state_context: Dict[str, bool] = None) -> str:
formatted = []
for item in items:
text = self.format_item_to_text(item, state_context)
if text:
formatted.append(f"- {text}")
return "\n".join(formatted)
# ==========================================
# 4. QWEN LLM INTEGRATION (OLLAMA)
# ==========================================
def ask_qwen(query: str, context_str: str, interrupt_event: threading.Event, model_name: str = "qwen2.5:3b") -> str:
if context_str:
prompt = (
"Nutze die folgende Wissensbasis und die logischen WENN-DANN-SONST-Regeln, um auf die Anfrage des Nutzers zu antworten.\n"
"Wissensdatenbank und Regeln:\n"
f"{context_str}\n\n"
f"ANFRAGE: {query}\n"
"ANTWORT (kurz und auf den Punkt):"
)
else:
prompt = f"ANFRAGE: {query}\nANTWORT:"
try:
response = requests.post(
"http://localhost:11434/api/generate",
json={"model": model_name, "prompt": prompt, "stream": True},
stream=True,
timeout=30
)
full_text = ""
for line in response.iter_lines():
if interrupt_event.is_set():
print("\n[🤖 Qwen]: Unterbrechung der Generierung...")
response.close()
return ""
if line:
import json
chunk = json.loads(line.decode('utf-8'))
text = chunk.get("response", "")
full_text += text
if chunk.get("done", False):
break
return full_text.strip()
except Exception as e:
return f"Fehler bei der Verbindung zu Ollama: {e}"
# ==========================================
# 5. VOICE INTERFACE
# ==========================================
class VoiceInterface:
def __init__(self, tts_model_name: str = "tts_models/en/ljspeech/vits"):
print("[*] Initialisierung von Whisper auf der CPU (int8)...")
self.whisper = WhisperModel("small", device="cpu", compute_type="int8")
print(f"[*] Initialisierung von Coqui TTS auf CPU ({tts_model_name})...")
self.tts = TTS(model_name=tts_model_name, progress_bar=False, gpu=False)
self.audio_queue = queue.Queue()
self.interrupt_event = threading.Event()
print("[*] Die Sprachschnittstelle ist fertig!")
def start_listening_thread(self, chunk_duration: int = 5, fs: int = 16000):
def listen_loop():
print("\n[🎤 Hintergrund-Abhören aktiv... Sprechen Sie in das Mikrofon.]")
num_samples = int(chunk_duration * fs)
while True:
audio = sd.rec(num_samples, samplerate=fs, channels=1, dtype='float32')
sd.wait()
if np.max(np.abs(audio)) > 0.03:
self.interrupt_event.set()
self.audio_queue.put(audio)
thread = threading.Thread(target=listen_loop, daemon=True)
thread.start()
def transcribe(self, audio_data: np.ndarray, fs: int = 16000) -> str:
temp_audio_path = tempfile.mktemp(suffix=".wav")
sf.write(temp_audio_path, audio_data, fs)
segments, _ = self.whisper.transcribe(temp_audio_path, beam_size=1, vad_filter=True)
text = " ".join([segment.text for segment in segments]).strip()
if os.path.exists(temp_audio_path):
os.remove(temp_audio_path)
return text
def speak(self, text: str):
if not text or self.interrupt_event.is_set():
return
print("[🔊 Sprachsynthese...]")
temp_audio_path = tempfile.mktemp(suffix=".wav")
self.tts.tts_to_file(text=text, file_path=temp_audio_path)
data, fs = sf.read(temp_audio_path)
print("[🔊 Wiedergabe der Antwort...]")
sd.play(data, fs)
while sd.get_stream().active:
if self.interrupt_event.is_set():
sd.stop()
print("\n[🔊 TTS]: Sprachausgabe vom Benutzer unterbrochen.")
break
time.sleep(0.1)
if os.path.exists(temp_audio_path):
os.remove(temp_audio_path)
# ==========================================
# 6. MAIN APPLICATION ENTRY POINT
# ==========================================
if __name__ == "__main__":
knowledge_file = "knowledge.lisp"
# Dynamischer Kontext des Systemzustands
system_state = {
"hohe_cpu_last": False,
"geringer_ram_speicher": False,
"ram_normal": True,
"mikrofon_verfuegbar": True,
"ollama_aktiv": True,
"benutzer_spricht": True,
"tts_wird_abgespielt": False,
"ollama_status_500": False,
"ollama_timeout": False,
"ollama_model_not_found": False
}
print("=== Full Conditional Lisp-TST RAG Voice Assistant ===")
try:
rag = LispRAGPipeline(knowledge_file)
print(f"[+] {len(rag.items)} Blöcke aus der Wissensdatenbank {knowledge_file} geladen")
except Exception as e:
print(f"[-] Fehler beim Laden der Wissensdatenbank: {e}")
sys.exit(1)
voice = VoiceInterface()
voice.start_listening_thread(chunk_duration=5)
print("\nDas System ist betriebsbereit. Sprechen Sie oder drücken Sie Strg+C zum Beenden.")
while True:
try:
audio_data = voice.audio_queue.get()
voice.interrupt_event.clear()
query = voice.transcribe(audio_data)
if not query:
continue
print(f"\n[🗣️ Erkannt]: {query}")
if "exit" in query.lower() or "beenden" in query.lower():
print("Beenden des Programms.")
break
retrieved_items = rag.retrieve_and_rank(query)
context_str = rag.format_context_for_llm(retrieved_items, state_context=system_state)
print(f"[🔍 TST Gefunden]: {len(retrieved_items)} relevante Wissensknoten.")
answer = ask_qwen(query, context_str, voice.interrupt_event, model_name="qwen2.5:3b")
if answer and not voice.interrupt_event.is_set():
print(f"[🤖 Qwen]: {answer}")
voice.speak(answer)
except KeyboardInterrupt:
print("\nStop auf Benutzerbefehl.")
break
except Exception as e:
print(f"[-] Fehler in der Hauptschleife: {e}")
Jetzt funktioniert alles mit beliebig tief verschachtelten Bedingungen
Angenommen, im system_state sind folgende Flags gesetzt:
system_state = {
"benutzer_spricht": True,
"tts_wird_abgespielt": True
}
Und in knowledge.lisp befindet sich ein Knoten mit einer verschachtelten Bedingung (cond_nested_audio_pipeline):
(:id "cond_nested_audio_pipeline"
:tags ("audio" "unterbrechung" "tts" "sprache")
:condition "benutzer_spricht"
:then (
(:condition "tts_wird_abgespielt"
:then "Aktuelle Sprachausgabe abbrechen (sd.stop()), Qwen-Generierungsstream zurücksetzen und Befehlsaufzeichnung starten."
:else "Warten beenden und den eingehenden Audio-Stream direkt an Whisper weiterleiten.")
)
:else "Normale Wiedergabe fortsetzen oder im VAD-Modus verbleiben.")
1. Benutzerfrage (🗣️ Über Whisper erkannt):
«Was soll das System tun, wenn ich direkt während der Sprachausgabe der Antwort zu sprechen beginne?»
2. Verarbeitung in LispRAGPipeline (🔍 Suche und Kontext-Extraktion):
Die TST-Suche findet den Knoten cond_nested_audio_pipeline anhand der Tags ("audio", "unterbrechung", "tts", "sprache").Rekursive Extraktion der Bedingungen (format_item_to_text):
- Äußere Bedingung wird geprüft: "benutzer_spricht" --> True.
- Das System wechselt in den :then-Zweig und trifft auf die verschachtelte Bedingung: "tts_wird_abgespielt".
- Verschachtelte Bedingung wird geprüft: "tts_wird_abgespielt" --> True.
- Der finale Text wird ausgewählt: "Aktuelle Sprachausgabe abbrechen (sd.stop()), Qwen-Generierungsstream zurücksetzen und Befehlsaufzeichnung starten."
Generierter Kontext, der an Qwen gesendet wird:
Verwende die folgende Wissensdatenbank und die logischen WENN-DANN-SONST-Regeln, um die Anfrage des Benutzers zu beantworten.
WISSENSDATENBANK UND REGELN:
- REGEL (Bedingung 'benutzer_spricht' ERFÜLLT -> Verschachtelte Bedingung 'tts_wird_abgespielt' ERFÜLLT): Aktuelle Sprachausgabe abbrechen (sd.stop()), Qwen-Generierungsstream zurücksetzen und Befehlsaufzeichnung starten.
ANFRAGE: Was soll das System tun, wenn ich direkt während der Sprachausgabe der Antwort zu sprechen beginne?
ANTWORT (kurz und prägnant):
Antwort des Modells (🤖 Qwen Answer):
«Wenn Sie während der Sprachausgabe zu sprechen beginnen, bricht das System die Audiowiedergabe über sd.stop() unverzüglich ab, setzt die aktuelle Textgenerierung von Qwen zurück und schaltet auf die Aufzeichnung Ihres neuen Befehls um.»