Alain Lilic
DEEN
Alle Projekte

LivePython-Plattform

Plattform für Sprach- und Chat-Agenten

Problem

Ich wollte Agenten für Telefon, Chat und Messenger, die sich per Konfiguration statt per Code einrichten lassen. Sie sollen auch auf einem eigenen Server laufen können und nur tun dürfen, was erlaubt ist.

Meine Rolle

Ich habe die Plattform entworfen und gebaut: Spezifikation, Architektur, Sicherheitsmodell, Gesprächsabläufe und Prompts, Tests, Deploy auf dem VPS und die Konfiguration der einzelnen Agenten. Ein Teil des Codes entstand in Cloud-Routinen, die Pull Requests bauen und von einer zweiten Rolle prüfen lassen. Deployt wird nur von mir.

Vorgehen

Ein Agent ist ein Ordner unter agents/<name>/: agent.yaml für Persona, Modelle und Wissen, flow.yaml als Gesprächsgraph mit erlaubten Aktionen je Phase, actions.yaml für Folgeschritte wie Mail, Kalender oder Rückruf, dazu ein Ordner mit Wissen. Kein eigener Code pro Agent. Die Pipeline baut auf Pipecat 1.5 und FastAPI. STT, LLM und TTS kommen aus einer Provider-Fabrik, lokal oder aus der Cloud, mit automatischem Wechsel auf ein Ausweichmodell bei einem anderen Anbieter. Das Wissen durchsucht ein hybrides RAG in LanceDB: Volltext mit deutschem Tokenizer plus Vektorsuche, zusammengeführt per RRF, mit lokalen Embeddings.

Für KI-Assistenten gibt es einen eigenen MCP-Server in einem eigenen Container: OAuth 2.1 mit eigenem Login und TOTP, Rechte als Scopes, verändernde Plattform-Aktionen nur in einem Schreibfenster, das ich im Cockpit öffne. Die einzige Ausnahme ist eng begrenzt: eine Mail an mich selbst. Qualität sichere ich mit einer pytest-Suite und goldenen Transkripten, die echte Flows mit festen Modellantworten durchfahren. Im Einsatz ist die Plattform als Web-Chat bei Seniorenbetreuung Lala in Düsseldorf und als Telegram-Bot bei Haus Wessels / Hotel Nike in Bottrop.

Architektur

  1. Kanäle

    Web-Chat-Widget, Sprache per WebRTC im Browser, Telegram als Rückkanal je Agent, mit eigenem Bot. Telefonnetz über Telnyx gebaut, noch nicht live.

  2. FastAPI und Pipecat

    STT, LLM und TTS aus der Provider-Fabrik, LLM-Failover über den LLMSwitcher auf einen zweiten Anbieter.

  3. Agent als Konfiguration

    agents/<name>/ mit agent.yaml, flow.yaml (Gesprächsgraph, erlaubte Aktionen je Phase), actions.yaml und knowledge/.

  4. Hybrides RAG

    LanceDB mit Volltext (deutscher Tokenizer) und Vektor, RRF, lokale Embeddings mit fastembed, Rückfall auf reine Vektorsuche.

  5. Aktionen und Protokoll

    Kalender, Mail, Webhook, Rückruf. Ein Fehler beendet nie das Gespräch. Jedes Gespräch in SQLite mit Transkript, Aktionen und Kosten.

  6. Eigener MCP-Server

    Eigener Container. OAuth 2.1, Login mit TOTP, Scopes, Schreibfenster aus dem Cockpit für verändernde Plattform-Aktionen (Ausnahme: eine Mail an mich selbst), Protokoll je Werkzeugaufruf. Keine Shell, kein exec.

  7. Qualität

    pytest-Suite, goldene Transkripte mit festen Modellantworten, Tests bei jedem Push auf main.

Entscheidung

Fail-fast statt stiller Retries

Am Telefon ist Stille teurer als ein schwächeres Modell. Das OpenAI-SDK wiederholt bei 429 und 5xx still und wartet dabei den Retry-After ab. Groq schickt beim Tageslimit 20-60 Sekunden, das sind Minuten Stille in einem Gespräch. Ich schalte die SDK-Retries ab, warte nur kurze Limits bis 8 Sekunden genau einmal ab und gebe alles andere sofort an den LLMSwitcher von Pipecat weiter, der auf ein Modell bei einem anderen Anbieter wechselt. Verspätete Fehler des schon abgelösten Dienstes ignoriere ich. Sonst schaltet der Switcher zurück auf das kaputte Modell, das habe ich im Log gesehen: Wechsel und 72 ms später zurück.

Code-Auszug

core/providers.py (gekürzt)Seitlich scrollenpython
def _fail_fast(service_cls):
    """OpenAI-kompatible Services ohne stille SDK-Retries: 429/5xx schlafen
    sonst den Retry-After ab (Groq schickt beim Tages-Limit 20-60 s!) - am
    Telefon ist das Minuten-Stille. Kurze Minuten-Limits (s.o.)
    wartet der Transport einmal ab; alles andere reicht sofort hoch, der
    LLMSwitcher failovert in Millisekunden aufs Fallback-Modell."""

    class FailFast(service_cls):
        def create_client(self, *args, **kwargs):
            client = super().create_client(*args, **kwargs).with_options(max_retries=0)
            # ...
            return client
    # ...

class _FailoverIgnoringStaleErrors(ServiceSwitcherStrategyFailover):
    # ...
    async def handle_error(self, error):
        failed = getattr(error, "processor", None)
        if failed is not None and failed is not self._active_service:
            logger.debug(f"Failover: ignoriere verspäteten Fehler von {failed.name}")
            return None
        self.letzter_fehler = str(getattr(error, "error", "") or "")
        return await super().handle_error(error)
SDK-Retries aus, damit der LLMSwitcher sofort auf das Ausweichmodell wechselt. Fehler eines schon abgelösten Dienstes werden ignoriert, sonst pendelt der Switcher zurück.

Ergebnisse

  • 7330Tests grün in der vollen pytest-Suite vor dem Deploy am 18.10.2026
  • 8goldene Transkript-Szenarien, die echte Flows mit festen Modellantworten durchfahren
  • 2Einsätze live: Web-Chat bei Seniorenbetreuung Lala, Telegram-Bot bei Haus Wessels / Hotel Nike
  • 1256Commits seit dem 07.07.2026

Was ich daraus gelernt habe

Ich habe zu sehr in die Breite gebaut. Am 23.09.2026 entstanden 44 Baupakete in drei Wellen, aber keiner der Telefon-Bausteine ist mit echter Telefonie abgenommen. Heute würde ich zuerst den Weg ins Telefonnetz live schalten und in einem echten Einsatz abnehmen, dann erweitern. Die Bausteine, die sich bei Haus Wessels bewährt haben, sind am stärksten an dieses eine Projekt gebunden. Die würde ich früher allgemein schneiden. Und server.py würde ich früher aufteilen, die Datei hat inzwischen rund 4.800 Zeilen.