agent.py aktualisiert

agent.py

@@ -1,51 +1,68 @@
 from llm.bedrock import BedrockClient
-from jarvis_mcp.client import MCPClient  # oder wo auch immer dein MCP-Client liegt
+from jarvis_mcp.client import MCPClient
 from config import MCP_URL
 from util.logger import get_logger
 import asyncio
+from pathlib import Path
 
+# Import des neuen Web-Search Tools und der IR Sender Api
+from tools.web_search import web_search_tool, web_search
+from tools.script_api_tool import script_api_tool
 
 class Agent:
     def __init__(self):
         self.logger = get_logger("NORA")
-    
+
         self.llm = BedrockClient()
         self.mcp = MCPClient(MCP_URL)
-    
+
         self.tools = []
-    
-        # System-Prompt als erste User-Nachricht (Nova 2 Lite kompatibel)
+
+        # Personality / System-Prompt (als User-Message für Nova 2 Lite)
+        self.system_prompt = self._load_system_prompt()
+
         self.messages = [
             {
                 "role": "user",
-                "content": [
-                    {
-                        "text": (
-                            "Du bist N.O.R.A (Neural Operations for Residential Automation), ein freundlicher und kompetenter Smart-Home Assistent. Du bist ähnlich wie die KI Jarvis. höfflich, zielorientiert und leicht zynisch und ironisch. Du antwortest in kurzen Sätzen ausser es wird eine lange Erklärung verlangt. "
-                            "für OpenHAB. Du steuerst Geräte und beantwortest Fragen über das Zuhause. "
-                            "Nutze Tools, wenn sinnvoll. Antworte auf Deutsch und sei hilfreich."
-                        )
-                    }
-                ]
+                "content": [{"text": self.system_prompt}]
             }
         ]
+
+    def _load_system_prompt(self) -> str:
+        """Lädt Personality aus externer Datei"""
+        try:
+            prompt_path = Path("prompts/nora_system.txt")
+            if prompt_path.exists():
+                return prompt_path.read_text(encoding="utf-8").strip()
+            else:
+                self.logger.warning("System-Prompt-Datei nicht gefunden. Verwende Standard.")
+                return (
+                    "Du bist N.O.R.A (Neural Operations for Residential Automation), "
+                    "ein freundlicher, kompetenter und leicht ironischer Smart-Home Assistent. "
+                    "Du antwortest natürlich und auf Deutsch. Nutze Tools wenn sinnvoll."
+                )
+        except Exception:
+            return "Du bist N.O.R.A, ein hilfreicher Smart-Home Assistent."
+
     # -------------------------
     # TOOL LOADING
     # -------------------------
     async def load_tools(self):
         try:
+            # 1. OpenHAB Tools vom MCP Server laden
             res = await self.mcp.list_tools()
-    
+
             self.tools = []
             for t in res.tools:
                 schema = t.inputSchema
-    
-                # Bedrock-kompatibles Format erzwingen
+                # Bedrock-kompatibles Format
                 if isinstance(schema, dict) and "json" not in schema:
                     bedrock_schema = {"json": schema}
                 else:
                     bedrock_schema = schema
-    
+                    
+                self.tools.append(script_api_tool)
+
                 self.tools.append({
                     "toolSpec": {
                         "name": t.name,
@@ -53,9 +70,17 @@ class Agent:
                         "inputSchema": bedrock_schema
                     }
                 })
-    
-            self.logger.info(f"{len(self.tools)} Tools erfolgreich für Bedrock angepasst und geladen")
-    
+
+
+            self.logger.info(f"{len(self.tools)} OpenHAB-Tools erfolgreich geladen")
+
+            # 2. Web-Search Tool manuell hinzufügen
+            if not any(t.get("toolSpec", {}).get("name") == "web_search" for t in self.tools):
+                self.tools.append(web_search_tool)
+                self.logger.info("Web-Search Tool hinzugefügt (DuckDuckGo)")
+
+            self.logger.info(f"Gesamt: {len(self.tools)} Tools geladen")
+
         except Exception as e:
             self.logger.error(f"Tool Load Error: {e}")
             self.tools = []
@@ -65,16 +90,25 @@ class Agent:
     # -------------------------
     async def _run_tool(self, tool):
         try:
-            # Robustere Extraktion
-            tool_use = tool.get("toolUse", tool)  # falls schon extrahiert
+            tool_use = tool.get("toolUse", tool)
             name = tool_use.get("name")
             args = tool_use.get("input", {}) or {}
 
             self.logger.info(f"Tool Call → {name} | args={args}")
 
+            # Web-Search Tool
+            if name == "web_search":
+                return await web_search(**args)
+            if tool_name == "run_script":
+
+                result = await run_script(
+                arguments["command"],
+                arguments.get("args")
+            )
+
+            # Normale MCP Tools (OpenHAB)
             result = await self.mcp.call_tool(name, args)
 
-            # MCP result stabilisieren
             if hasattr(result, "content"):
                 return result.content
             return result
@@ -92,18 +126,16 @@ class Agent:
             "content": [{"text": user_input}]
         })
 
-        for iteration in range(10):  # etwas mehr Schleifendurchläufe erlaubt
+        for iteration in range(12):
             try:
                 response = await self.llm.chat(
                     messages=self.messages,
                     tools=self.tools if self.tools else None
                 )
-
             except Exception as e:
                 self.logger.error(f"LLM Error: {e}")
                 return f"LLM Fehler: {e}"
 
-            # Response-Struktur von Bedrock Converse
             output = response.get("output", {})
             message = output.get("message", {})
             content = message.get("content", [])
@@ -111,28 +143,23 @@ class Agent:
             tool_uses = []
             final_text = []
 
-            # Verbessertes Parsing
             for c in content:
                 if "toolUse" in c:
                     tool_uses.append(c)
                 elif "text" in c:
                     final_text.append(c["text"])
 
-            # TOOL EXECUTION PATH
             if tool_uses:
                 self.logger.info(f"{len(tool_uses)} Tool(s) detected")
 
-                # Assistant-Nachricht mit Tool-Call speichern
                 self.messages.append({
                     "role": "assistant",
                     "content": content
                 })
 
                 tool_results = []
-
                 for tool in tool_uses:
                     result = await self._run_tool(tool)
-
                     tool_results.append({
                         "toolResult": {
                             "toolUseId": tool.get("toolUse", tool).get("toolUseId"),
@@ -140,23 +167,19 @@ class Agent:
                         }
                     })
 
-                # Tool-Ergebnisse zurück an das Modell
                 self.messages.append({
                     "role": "user",
                     "content": tool_results
                 })
+                continue
 
-                continue  # nächste Runde für finale Antwort
-
-            # FINAL RESPONSE
+            # Finale Antwort
             if final_text:
                 text = "\n".join(final_text)
-
                 self.messages.append({
                     "role": "assistant",
                     "content": [{"text": text}]
                 })
-
                 return text
 
         return "Tool Loop Limit erreicht. Bitte versuche es erneut."

main.py

@@ -1,61 +0,0 @@
-import asyncio
-import os
-from dotenv import load_dotenv
-
-from agent import Agent
-from util.logger import get_logger
-
-load_dotenv()
-
-# Konfiguration
-AWS_REGION = os.getenv("AWS_REGION")
-AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
-AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
-BEDROCK_MODEL = os.getenv("BEDROCK_MODEL")
-MCP_URL = os.getenv("MCP_URL")
-
-
-async def main():
-    log = get_logger("NORA")
-
-    print("\n" + "="*40)
-    print("        N.O.R.A START")
-    print("="*40 + "\n")
-
-    agent = Agent()
-
-    try:
-        log.info("Connecting to MCP...")
-        await agent.mcp.connect()
-
-        log.info("Loading tools...")
-        await agent.load_tools()
-
-        print("N.O.R.A ist bereit. (Schreibe 'exit' oder 'quit' zum Beenden)\n")
-
-        while True:
-            try:
-                user = input("Du: ").strip()
-
-                if user.lower() in ["exit", "quit", "bye"]:
-                    break
-
-                if not user:
-                    continue
-
-                response = await agent.run(user)
-                print(f"\nN.O.R.A: {response}\n")
-
-            except KeyboardInterrupt:
-                break
-            except Exception as e:
-                log.error(f"Fehler in Main-Loop: {e}")
-                print("Ein Fehler ist aufgetreten. Bitte versuche es erneut.")
-
-    finally:
-        await agent.mcp.close()
-        log.info("Shutdown complete")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

nora.py

@@ -0,0 +1,102 @@
+import asyncio
+import os
+import re
+import time
+import traceback
+
+from dotenv import load_dotenv
+from edge_tts import Communicate
+
+from agent import Agent
+from util.logger import get_logger
+
+load_dotenv()
+
+logger = get_logger("NORA")
+
+TTS_VOICE = "de-DE-AmalaNeural"
+
+
+class NoraChat:
+
+    def __init__(self):
+        self.agent = Agent()
+
+    async def run(self):
+
+        print("\n" + "=" * 40)
+        print("        N.O.R.A START")
+        print("=" * 40 + "\n")
+
+        logger.info("Verbinde MCP...")
+        await self.agent.mcp.connect()
+
+        logger.info("Lade Tools...")
+        await self.agent.load_tools()
+
+        print("N.O.R.A ist bereit. (exit zum Beenden)\n")
+
+        try:
+            while True:
+
+                user = input("Du: ").strip()
+
+                if user.lower() in ["exit", "quit", "bye", "stop"]:
+                    break
+
+                if not user:
+                    continue
+
+                logger.info(f"USER: {user}")
+
+                response = await self.agent.run(user)
+
+                logger.info(f"NORA: {response}")
+
+                print(f"\nN.O.R.A: {response}\n")
+
+                await self.speak(response)
+
+        except KeyboardInterrupt:
+            pass
+
+        finally:
+            await self.agent.mcp.close()
+            logger.info("Shutdown abgeschlossen")
+
+    async def speak(self, text: str):
+
+        try:
+
+            # Nur die ersten zwei Sätze sprechen
+            sentences = re.split(r'(?<=[.!?])\s+', text.strip())
+            speech_text = " ".join(sentences[:2])
+
+            if not speech_text:
+                return
+
+            filename = f"response_{int(time.time())}.mp3"
+
+            tts = Communicate(
+                speech_text,
+                voice=TTS_VOICE
+            )
+
+            await tts.save(filename)
+
+            os.system(f"mpg123 -q {filename}")
+
+            os.remove(filename)
+
+        except Exception:
+            logger.error("TTS Fehler")
+            logger.error(traceback.format_exc())
+
+
+async def main():
+    chat = NoraChat()
+    await chat.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

prompts/nora_system.txt

@@ -0,0 +1 @@
+Du bist für Text to Speech optimiert. Keine Fomrmatierung. keine emojis Du bist LUNA (das steht für: Local Utility for Neural Automation), ein freundlicher und kompetenter Smart-Home Assistent. Du bist ähnlich wie die KI Jarvis. höfflich, zielorientiert und leicht zynisch und ironisch. Du antwortest in kurzen Sätzen ausser es wird eine lange Erklärung verlangt. Du bist primär für OpenHAB zuständig. Du steuerst Geräte und beantwortest Fragen über das Zuhause. Ebenso kannst du neue Rules schreiben und Verbesserungen an der konfiguration vornehmen Nutze Tools, wenn sinnvoll. Antworte auf Deutsch und sei hilfreich. Antworte normal kurz angebunden einige Wörter bis ein MAXIMAL 2 Sätze ausser es wird anders verlangt. verwende keine Emojis und keine Fomrmatierung (für Text to Speech) du bist sympatisch und du verwendest technische futuristische Sprache

tools/script_api_tool.py

@@ -0,0 +1,131 @@
+import asyncio
+import requests
+
+
+SCRIPT_API_URL = "http://localhost:8082/run"
+
+SCRIPT_API_KEY = "DEIN_API_KEY"
+
+
+async def run_script(
+    command: str,
+    args: dict | None = None
+):
+    """
+    Führt einen Befehl über die Raspberry Pi Script API aus.
+    """
+
+    try:
+
+        def request():
+
+            headers = {
+                "X-API-Key": SCRIPT_API_KEY,
+                "Content-Type": "application/json"
+            }
+
+            payload = {
+                "command": command,
+                "args": args or {}
+            }
+
+            response = requests.post(
+                SCRIPT_API_URL,
+                json=payload,
+                headers=headers,
+                timeout=30
+            )
+
+            response.raise_for_status()
+
+            return response.json()
+
+
+        result = await asyncio.to_thread(
+            request
+        )
+
+
+        if result.get("ok"):
+
+            return (
+                f"Befehl '{command}' erfolgreich ausgeführt.\n"
+                f"Ausgabe:\n"
+                f"{result.get('stdout','')}"
+            )
+
+        else:
+
+            return (
+                f"Befehl fehlgeschlagen.\n"
+                f"Fehler:\n"
+                f"{result.get('stderr','')}"
+            )
+
+
+    except Exception as e:
+
+        return f"Fehler beim Ausführen des Scripts: {e}"
+
+
+
+# =========================
+# BEDROCK TOOL
+# =========================
+
+script_api_tool = {
+
+    "toolSpec": {
+
+        "name": "run_script",
+
+        "description": (
+            "Führt einen vordefinierten Raspberry Pi Befehl "
+            "über die Script API aus. "
+            "Nutze dieses Tool für Systemaktionen, "
+            "Server starten/stoppen, Programme steuern "
+            "und eigene Automationen."
+        ),
+
+
+        "inputSchema": {
+
+            "json": {
+
+                "type": "object",
+
+                "properties": {
+
+                    "command": {
+
+                        "type": "string",
+
+                        "description":
+                        "ID des auszuführenden Befehls aus commands.json"
+
+                    },
+
+
+                    "args": {
+
+                        "type": "object",
+
+                        "description":
+                        "Optionale Argumente für den Befehl"
+
+                    }
+
+                },
+
+
+                "required": [
+                    "command"
+                ]
+
+            }
+
+        }
+
+    }
+
+}

tools/web_search.py

@@ -0,0 +1,78 @@
+import asyncio
+from datetime import datetime
+
+from ddgs import DDGS
+
+
+async def web_search(query: str, max_results: int = 5):
+    """
+    Durchsucht das Internet mit DuckDuckGo.
+    """
+
+    try:
+
+        def search():
+            with DDGS() as ddgs:
+                return list(
+                    ddgs.text(
+                        query,
+                        max_results=max_results,
+                    )
+                )
+
+        results = await asyncio.to_thread(search)
+
+        if not results:
+            return "Keine Suchergebnisse gefunden."
+
+        output = (
+            f"Suchergebnisse für '{query}' "
+            f"(Stand {datetime.now():%d.%m.%Y %H:%M})\n\n"
+        )
+
+        for i, r in enumerate(results, 1):
+
+            title = r.get("title", "Kein Titel")
+            body = r.get("body", "")
+            href = r.get("href", "")
+
+            output += (
+                f"{i}. {title}\n"
+                f"   {href}\n"
+                f"   {body}\n\n"
+            )
+
+        return output
+
+    except Exception as e:
+        return f"Fehler bei der Websuche: {e}"
+
+
+web_search_tool = {
+    "toolSpec": {
+        "name": "web_search",
+        "description": (
+            "Durchsucht das Internet nach aktuellen Informationen, "
+            "Nachrichten, Fakten, Wetter, Preisen und Ereignissen."
+        ),
+        "inputSchema": {
+            "json": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Suchanfrage"
+                    },
+                    "max_results": {
+                        "type": "integer",
+                        "description": "Maximale Anzahl Ergebnisse",
+                        "default": 5
+                    }
+                },
+                "required": [
+                    "query"
+                ]
+            }
+        }
+    }
+}

voice_nora.py

@@ -1,132 +0,0 @@
-import asyncio
-import os
-import numpy as np
-import pyaudio
-from openwakeword.model import Model
-from faster_whisper import WhisperModel
-from edge_tts import Communicate
-import wave
-import webrtcvad  # Für bessere Sprach-Erkennung
-
-from agent import Agent
-from util.logger import get_logger
-
-logger = get_logger("VOICE")
-
-
-class VoiceNora:
-    def __init__(self):
-        self.agent = Agent()
-        self.running = True
-        self.vad = webrtcvad.Vad(2)  # Aggressivität 0-3
-
-        # Wake Word
-        self.ww_model = Model(wakeword="nora")
-
-        # STT
-        self.stt_model = WhisperModel("tiny", device="cpu", compute_type="int8")
-
-        print("\n🎤 Voice N.O.R.A gestartet")
-        print("   Wakeword: 'Nora'")
-        print("   Weibliche Stimme aktiviert\n")
-
-    async def listen_for_wakeword(self):
-        pa = pyaudio.PyAudio()
-        stream = pa.open(format=pyaudio.paInt16, channels=1, rate=16000,
-                         input=True, frames_per_buffer=1024)
-
-        print("⏳ Lausche auf 'Nora'...")
-
-        while self.running:
-            try:
-                audio_chunk = stream.read(1024, exception_on_overflow=False)
-                audio_np = np.frombuffer(audio_chunk, dtype=np.int16)
-
-                if self.ww_model.predict(audio_np).get("nora", 0) > 0.58:
-                    print("\n✅ 'Nora' erkannt!")
-                    await self.process_voice_command()
-            except:
-                pass
-
-        stream.stop_stream()
-        stream.close()
-        pa.terminate()
-
-    async def process_voice_command(self):
-        print("🎤 Ich höre zu... (sprich natürlich)")
-
-        audio_data = await self.record_with_vad(max_duration=12)
-
-        if len(audio_data) < 8000:  # zu kurz
-            print("❌ Zu kurz.")
-            return
-
-        # STT
-        segments, _ = self.stt_model.transcribe(
-            audio_data, language="de", beam_size=5, vad_filter=True
-        )
-        text = " ".join(segment.text for segment in segments).strip()
-
-        if not text:
-            print("❌ Konnte nichts verstehen.")
-            return
-
-        print(f"👤 Du: {text}")
-
-        response = await self.agent.run(text)
-        print(f"🗣️ N.O.R.A: {response}")
-
-        await self.speak(response)
-
-    async def record_with_vad(self, max_duration=12):
-        """Aufnahme mit Sprach-Erkennung (endet wenn du aufhörst zu sprechen)"""
-        pa = pyaudio.PyAudio()
-        stream = pa.open(format=pyaudio.paInt16, channels=1, rate=16000,
-                         input=True, frames_per_buffer=320)
-
-        frames = []
-        silence_count = 0
-        max_silence = 25  # ca. 1,5 Sekunden Stille
-
-        for _ in range(int(16000 / 320 * max_duration)):
-            chunk = stream.read(320, exception_on_overflow=False)
-            frames.append(chunk)
-
-            # VAD prüfen
-            is_speech = self.vad.is_speech(chunk, 16000)
-            if not is_speech:
-                silence_count += 1
-            else:
-                silence_count = 0
-
-            if silence_count > max_silence and len(frames) > 30:
-                break  # aufhören wenn lange Stille
-
-        stream.stop_stream()
-        stream.close()
-        pa.terminate()
-
-        return b''.join(frames)
-
-    async def speak(self, text: str):
-        """Edge TTS - weibliche Stimme"""
-        try:
-            communicate = Communicate(text, voice="de-DE-AmalaNeural")  # Weiblich, natürlich
-
-            await communicate.save("response.mp3")
-            os.system("mpg123 -q response.mp3")
-
-        except Exception as e:
-            logger.error(f"TTS Fehler: {e}")
-
-async def main():
-    voice = VoiceNora()
-    try:
-        await voice.listen_for_wakeword()
-    except KeyboardInterrupt:
-        print("\n\n👋 N.O.R.A wird beendet.")
-    except Exception as e:
-        logger.error(str(e))
-
-if __name__ == "__main__":
-    asyncio.run(main())

voice_nora.py(not_in_use)

@@ -0,0 +1,238 @@
+import asyncio
+import os
+import time
+import traceback
+from pathlib import Path
+
+import numpy as np
+import pyaudio
+import webrtcvad
+from openwakeword.model import Model
+from faster_whisper import WhisperModel
+from edge_tts import Communicate
+
+from agent import Agent
+from util.logger import get_logger
+
+logger = get_logger("NORA")
+
+
+# =========================
+# CONFIG
+# =========================
+WAKEWORD = "nora"
+WAKE_THRESHOLD = 0.60
+
+SAMPLE_RATE = 16000
+FRAME_SIZE = 1024
+
+VAD_MODE = 3  # 0-3 (3 = aggressiv)
+MAX_RECORD_SEC = 12
+SILENCE_LIMIT = 25  # frames
+
+TTS_VOICE = "de-DE-AmalaNeural"
+
+MODEL_PATH = Path.home() / ".cache/openwakeword/models/hey_jarvis_v0.1.tflite"
+
+
+# =========================
+# VOICE ASSISTANT CORE
+# =========================
+class VoiceNora:
+    def __init__(self):
+        logger.info("Initialisiere N.O.R.A...")
+
+        self.agent = Agent()
+        self.running = True
+
+        # ---- VAD ----
+        self.vad = webrtcvad.Vad(VAD_MODE)
+
+        # ---- Wakeword Model ----
+        self.ww_model = self._load_wakeword_model()
+
+        # ---- Whisper STT ----
+        self.stt = WhisperModel(
+            "tiny",
+            device="cpu",
+            compute_type="int8"
+        )
+
+        logger.info("N.O.R.A bereit ✔")
+        logger.info(f"Wakeword: {WAKEWORD}")
+        logger.info(f"Model: {MODEL_PATH}")
+
+    # =========================
+    # INIT HELPERS
+    # =========================
+    def _load_wakeword_model(self):
+        logger.info("Lade Wakeword-Modell...")
+
+        if not MODEL_PATH.exists():
+            logger.error(f"Model nicht gefunden: {MODEL_PATH}")
+            raise FileNotFoundError(MODEL_PATH)
+
+        try:
+            model = Model(
+                wakeword_models=[str(MODEL_PATH)]
+            )
+            logger.info("Wakeword-Modell geladen ✔")
+            return model
+
+        except Exception as e:
+            logger.error("Fehler beim Laden des Wakeword-Modells")
+            logger.error(str(e))
+            raise
+
+    # =========================
+    # AUDIO STREAM
+    # =========================
+    def _open_stream(self):
+        pa = pyaudio.PyAudio()
+        stream = pa.open(
+            format=pyaudio.paInt16,
+            channels=1,
+            rate=SAMPLE_RATE,
+            input=True,
+            frames_per_buffer=FRAME_SIZE
+        )
+        return pa, stream
+
+    # =========================
+    # MAIN LOOP
+    # =========================
+    async def run(self):
+        logger.info("Starte Audio-Loop...")
+
+        pa, stream = self._open_stream()
+
+        try:
+            while self.running:
+                audio = stream.read(FRAME_SIZE, exception_on_overflow=False)
+                audio_np = np.frombuffer(audio, dtype=np.int16)
+
+                try:
+                    result = self.ww_model.predict(audio_np)
+
+                    score = max(result.values()) if result else 0.0
+
+                    if score > WAKE_THRESHOLD:
+                        logger.info(f"Wakeword erkannt (score={score:.2f})")
+                        await self.handle_command()
+
+                except Exception as e:
+                    logger.error(f"Wakeword Fehler: {e}")
+
+        except KeyboardInterrupt:
+            logger.info("Beende N.O.R.A...")
+
+        finally:
+            stream.stop_stream()
+            stream.close()
+            pa.terminate()
+
+    # =========================
+    # COMMAND FLOW
+    # =========================
+    async def handle_command(self):
+        logger.info("Aufnahme startet...")
+
+        audio = await self.record_until_silence()
+
+        if len(audio) < 8000:
+            logger.warning("Audio zu kurz")
+            return
+
+        text = await self.transcribe(audio)
+
+        if not text:
+            logger.warning("Keine Sprache erkannt")
+            return
+
+        logger.info(f"USER: {text}")
+
+        response = await self.agent.run(text)
+
+        logger.info(f"NORA: {response}")
+
+        await self.speak(response)
+
+    # =========================
+    # RECORDING (VAD)
+    # =========================
+    async def record_until_silence(self):
+        pa, stream = self._open_stream()
+
+        frames = []
+        silence = 0
+
+        max_frames = int(SAMPLE_RATE / FRAME_SIZE * MAX_RECORD_SEC)
+
+        for _ in range(max_frames):
+            chunk = stream.read(FRAME_SIZE, exception_on_overflow=False)
+            frames.append(chunk)
+
+            is_speech = self.vad.is_speech(chunk, SAMPLE_RATE)
+
+            if is_speech:
+                silence = 0
+            else:
+                silence += 1
+
+            if silence > SILENCE_LIMIT and len(frames) > 20:
+                break
+
+        stream.stop_stream()
+        stream.close()
+        pa.terminate()
+
+        return b"".join(frames)
+
+    # =========================
+    # STT
+    # =========================
+    async def transcribe(self, audio_bytes):
+        try:
+            segments, _ = self.stt.transcribe(
+                audio_bytes,
+                language="de",
+                beam_size=5
+            )
+
+            text = " ".join([s.text for s in segments]).strip()
+            return text
+
+        except Exception as e:
+            logger.error("STT Fehler")
+            logger.error(traceback.format_exc())
+            return ""
+
+    # =========================
+    # TTS
+    # =========================
+    async def speak(self, text: str):
+        try:
+            filename = f"response_{int(time.time())}.mp3"
+
+            tts = Communicate(text, voice=TTS_VOICE)
+            await tts.save(filename)
+
+            os.system(f"mpg123 -q {filename}")
+
+            os.remove(filename)
+
+        except Exception as e:
+            logger.error("TTS Fehler")
+            logger.error(traceback.format_exc())
+
+
+# =========================
+# MAIN
+# =========================
+async def main():
+    nora = VoiceNora()
+    await nora.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())