dschlueter/chatterbox-tts-cli
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Compare commits

base: dschlueter:c2ceb9a76c30bde94ec2365525978d61dec40829
Branches Tags
dschlueter:main
..
compare: dschlueter:8e34186c1afcb073bbb5a2ab18f8475ba0cfd2f3
Branches Tags
dschlueter:main

No commits in common. "c2ceb9a76c30bde94ec2365525978d61dec40829" and "8e34186c1afcb073bbb5a2ab18f8475ba0cfd2f3" have entirely different histories.
c2ceb9a76c ... 8e34186c1a

10 changed files with 1 additions and 3004 deletions
Download patch file Download diff file Expand all files Collapse all files

31
.gitignore vendored
View file

@ -1,31 +0,0 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
.eggs/
# Ausgabe-Dateien
*.wav
*.mp3
*.ogg
# Persönliche Daten (Stimmaufnahmen)
my_voice*.wav
voice_*.wav
# Umgebung
.env
*.log
.venv/
env/
# IDE
.vscode/
.idea/
# Claude Code
.claude/
# Ideen
Ideen/

316
BEDIENUNGSANLEITUNG.md
View file

@ -1,316 +0,0 @@
# Bedienungsanleitung: Chatterbox TTS-Assistent
Dieses Programm liest Texte laut vor — ähnlich wie ein Vorlesedienst.
Es wandelt geschriebenen Text in natürlich klingende Sprache um.
---
## Was das Programm braucht
- Einen Computer mit Linux
- Eine installierte Conda-Umgebung namens `chatterbox`
- Eine Grafikkarte (GPU) — macht das Programm deutlich schneller
---
## Automatischer Start im Hintergrund
Der Sprach-Service startet automatisch, sobald du dich am Computer anmeldest.
Du musst nichts weiter tun — er läuft im Hintergrund und wartet auf Anfragen.
Ob der Service läuft, prüfst du so:
```bash
systemctl --user status chatterbox-tts
```
Bei Problemen neu starten:
```bash
systemctl --user restart chatterbox-tts
```
---
## Das Kommandozeilen-Programm starten
Für die direkte Nutzung über das Terminal:
```bash
conda activate chatterbox
cd ~/chatterbox-tts-cli
```
---
## Einen Text vorlesen lassen
### Text aus einer Datei vorlesen
```bash
python chatterbox_cli_v4.py --lang de --input mein_text.txt
```
Ersetze `mein_text.txt` durch den Pfad zu deiner Textdatei.
Die Datei muss im Format **UTF-8** gespeichert sein (das ist der Standard
bei modernen Texteditoren).
### Einen kurzen Text direkt eingeben
```bash
python chatterbox_cli_v4.py --lang de --text "Guten Morgen! Wie geht es Ihnen heute?"
```
---
## Die eigene Stimme verwenden
Wenn du eine Aufnahme deiner Stimme hast (eine WAV-Datei von ca. 10–30 Sekunden),
kann das Programm diese Stimme nachahmen:
```bash
python chatterbox_cli_v4.py --lang de \
--voice meine_stimme.wav \
--input mein_text.txt
```
**Tipp:** Eine Aufnahme von 20 Sekunden reicht aus. Am besten in ruhiger Umgebung
und deutlich sprechen.
---
## Sprache wählen
Das Programm kann in vielen Sprachen vorlesen. Die Sprache wählt man mit `--lang`:
| Befehl | Sprache |
|--------|---------|
| `--lang de` | Deutsch (Standard) |
| `--lang en` | Englisch |
| `--lang fr` | Französisch |
| `--lang es` | Spanisch |
| `--lang it` | Italienisch |
Beispiel auf Englisch:
```bash
python chatterbox_cli_v4.py --lang en --text "Good morning, how are you?"
```
---
## Sprechgeschwindigkeit anpassen
Mit `--speed` kann man einstellen, wie schnell der Text gesprochen wird.
- `1.0` = normale Geschwindigkeit (Standard)
- `0.85` = etwas langsamer — gut für entspanntes Zuhören
- `0.75` = deutlich langsamer
- `1.2` = etwas schneller
```bash
python chatterbox_cli_v4.py --lang de --speed 0.85 --input mein_text.txt
```
**Hinweis:** Die Stimmhöhe bleibt gleich — nur das Tempo ändert sich.
---
## Audio als Datei speichern
Wenn du die Audiodatei behalten möchtest:
```bash
python chatterbox_cli_v4.py --lang de --save --input mein_text.txt
```
Die Datei wird automatisch als `mein_text.de.wav` gespeichert — im selben
Ordner wie die Eingabedatei.
Oder mit eigenem Dateinamen:
```bash
python chatterbox_cli_v4.py --lang de --output ausgabe.wav --input mein_text.txt
```
---
## Nur speichern, nicht abspielen
```bash
python chatterbox_cli_v4.py --lang de --no-play --output ausgabe.wav --input mein_text.txt
```
---
## Aussprache von Eigennamen anpassen
Manche Namen — vor allem aus anderen Sprachen — werden falsch ausgesprochen.
Du kannst das mit einer einfachen Textdatei im JSON-Format korrigieren.
**Beispiel:** Datei `aussprache.json` anlegen:
```json
{
"Seoul": "Söul",
"Macron": "Makron",
"Kubernetes": "Kubernetis"
}
```
Dann so aufrufen:
```bash
python chatterbox_cli_v4.py --lang de \
--pronunciation-dict aussprache.json \
--input nachricht.txt
```
Häufige englische Begriffe wie „GitHub" (→ „Git Hab"), „YouTube" (→ „Jutjub") oder
„Wi-Fi" (→ „Wai Fai") werden bereits automatisch korrekt ausgesprochen.
---
## Den Service aus dem Netzwerk nutzen
Der Service ist im gesamten Heimnetzwerk erreichbar — zum Beispiel vom Handy,
Tablet oder einem anderen Computer.
**Text vorlesen lassen** (aus jedem Gerät im Netzwerk):
```bash
curl -X POST http://COMPUTER-IP:9999/speak \
-H "Content-Type: application/json" \
-d '{"text": "Hallo aus dem Netzwerk", "lang": "de"}'
```
`COMPUTER-IP` ersetzen durch die IP-Adresse dieses Computers (z. B. `192.168.1.42`).
**Aktuelle IP-Adresse herausfinden:**
```bash
hostname -I
```
**Ausgabe pausieren und fortsetzen:**
```bash
curl -X POST http://COMPUTER-IP:9999/pause
curl -X POST http://COMPUTER-IP:9999/resume
```
**Ausgabe stoppen:**
```bash
curl -X POST http://COMPUTER-IP:9999/stop
```
---
## KI-Assistenten lassen vorlesen
Wenn du einen KI-Assistenten auf diesem oder einem anderen Gerät nutzt,
kann er den TTS-Service direkt ansprechen:
### Claude (Claude Code / Claude Desktop)
Claude ist bereits mit dem TTS-Service verbunden. Du kannst Claude einfach bitten,
etwas vorzulesen, zu pausieren oder zu stoppen — er ruft den Service automatisch auf.
Beispiel-Anfragen an Claude:
> „Lies mir bitte diesen Text vor: ..."
> „Pause bitte."
> „Weiter."
### Home Assistant
In der `configuration.yaml` folgendes eintragen:
```yaml
rest_command:
tts_sprechen:
url: "http://COMPUTER-IP:9999/speak"
method: POST
content_type: "application/json"
payload: '{"text": "{{ text }}", "lang": "de"}'
tts_pause:
url: "http://COMPUTER-IP:9999/pause"
method: POST
tts_weiter:
url: "http://COMPUTER-IP:9999/resume"
method: POST
```
Danach in einer Automation verwendbar:
```yaml
service: rest_command.tts_sprechen
data:
text: "Die Waschmaschine ist fertig."
```
### Ollama / LM Studio / Open WebUI
Lokale KI-Modelle (z. B. llama, qwen) können über eine kleine Hilfsklasse mit dem
Service verbunden werden — Details in der `README.md`.
---
## Typischer Arbeitsablauf
1. Text in einem Editor schreiben und als `.txt`-Datei speichern
2. Terminal öffnen, `conda activate chatterbox`
3. Programm aufrufen:
```bash
python chatterbox_cli_v4.py --lang de --voice meine_stimme.wav --input text.txt
```
4. Das Programm beginnt sofort zu sprechen — Satz für Satz
---
## Was das Programm automatisch macht
- **Markdown-Formatierung bereinigen**: `**fett**`, `# Überschrift`, `- Listen` und Links werden vor der Sprachausgabe entfernt
- **Emojis entfernen**: Smileys und Symbole (😊, 🎉) werden still übergangen
- **Abkürzungen buchstabieren**: ARD wird zu „Ah Er De", YMCA zu „Ypsilon Em Tse Ah"
- **Tech-Abkürzungen richtig sprechen**: CPU, GPU, USB, API, JSON u. a. werden lateinisch buchstabiert (nicht auf Deutsch)
- **Zusammengesetzte Wörter mit Abkürzung**: „US-Präsident" wird zu „U Es Präsident"
- **Uhrzeiten vorlesen**: „14:58" wird zu „vierzehn Uhr achtundfünfzig"
- **Jahreszahlen aussprechen**: „2026" wird zu „zweitausendsechsundzwanzig"
- **Einheiten übersetzen**: „25 °C", „100 kWh", „10 m²", „100 W" werden ausgeschrieben
- **Trennzeilen überspringen**: Linien wie „--- Ende ---" werden stillschweigend übersprungen
---
## Wenn etwas nicht klappt
**Kein Ton zu hören:**
```bash
python -c "import sounddevice; print(sounddevice.query_devices())"
```
Dann `--audio-device pulse` oder das passende Gerät angeben.
**Service antwortet nicht:**
```bash
systemctl --user restart chatterbox-tts
# Warte 5 Sekunden, dann:
curl http://localhost:9999/health
```
**„Modell nicht gefunden":**
Beim ersten Start wird das Modell heruntergeladen (~2 GB).
Sicherstellen, dass eine Internetverbindung besteht.
**Programm ist sehr langsam:**
Ohne GPU dauert die Generierung länger als die Wiedergabe — ein Satz
kann 30–60 Sekunden brauchen. Mit GPU (CUDA) dauert es ca. 5–10 Sekunden.
---
## Bekannte Grenzen
- **Betonung einzelner Wörter** lässt sich nicht direkt steuern.
Eine Aufnahme der eigenen Stimme mit natürlicher Betonung kann helfen.
- **Manche Fremdwörter** klingen nicht immer perfekt — mit der Aussprache-Datei lässt sich das korrigieren.
- Das Programm liest alles vor, was in der Datei steht — also auch
Überschriften und Metadaten wie „Schlagzeile:" oder „Stand:".
- Eine laufende Ausgabe kann erst am Ende des aktuellen Satzes unterbrochen
werden, nicht sofort mitten im Wort.

203
CLAUDE.md
View file

@ -1,203 +0,0 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Running the CLI
```bash
conda activate chatterbox
# Deutschen Text aus Datei vorlesen
python chatterbox_cli_v4.py --lang de --input text.txt
# Mit Voice Cloning
python chatterbox_cli_v4.py --lang de --voice my_voice.wav --input text.txt
# Text direkt übergeben (Englisch)
python chatterbox_cli_v4.py --lang en --text "Hello world"
# Nur speichern, kein Playback
python chatterbox_cli_v4.py --lang de --no-play --output ausgabe.wav --input text.txt
# Geschwindigkeit anpassen (pitch-erhaltend, erfordert rubberband-cli)
python chatterbox_cli_v4.py --lang de --speed 0.85 --input text.txt
# Streaming-Modus (experimentell, niedrigere Latenz, kann abgehackt klingen)
python chatterbox_cli_v4.py --lang de --stream --input text.txt
# Aussprache-Wörterbuch (JSON: {"Eigenname": "Lautschrift"})
python chatterbox_cli_v4.py --lang de --pronunciation-dict aussprache.json --input text.txt
# Markdown-Bereinigung deaktivieren (Standard: aktiv)
python chatterbox_cli_v4.py --lang de --no-strip-markdown --input text.txt
# Gemischtsprachiger Text mit Sprachmarkierungen
python chatterbox_cli_v4.py --lang de --text "Das [en]Machine Learning[/en] Modell ist gut."
```
No build step, no test suite, no linter configuration — this is a single-file script.
## Running the HTTP Service
```bash
# Läuft als systemd-User-Service (Autostart beim Login):
systemctl --user status chatterbox-tts
systemctl --user restart chatterbox-tts
journalctl --user -u chatterbox-tts -f
# Manuell starten (Port 9999, LAN-weit erreichbar):
uvicorn tts_service:app --host 0.0.0.0 --port 9999
# Mit Modell-Warmup (Modell beim Start laden, kein Cold-Start beim ersten Request):
TTS_PRELOAD_LANG=de uvicorn tts_service:app --host 0.0.0.0 --port 9999
# Health-Check:
curl http://127.0.0.1:9999/health
```
Endpunkte: `POST /speak`, `POST /stop`, `POST /pause`, `POST /resume`, `GET /health`, `GET /status`, `GET /voices`
## Running the MCP Adapter
```bash
# stdio (Claude Code / Claude Desktop) — bereits in ~/.claude.json konfiguriert:
python mcp_adapter.py --stdio
# HTTP-Transport (Port 8001):
python mcp_adapter.py
# Anderen TTS-Service ansprechen:
TTS_URL=http://192.168.1.10:9999 python mcp_adapter.py --stdio
```
MCP-Tools: `speak`, `stop`, `pause`, `resume`, `get_status`, `list_voices`
## Architecture
### Files
| Datei | Funktion |
|-------|----------|
| `chatterbox_cli_v4.py` | Kern-CLI und alle Hilfsfunktionen; wird von `tts_service.py` importiert |
| `tts_service.py` | FastAPI-Service mit Job-Queue und Worker-Thread |
| `mcp_adapter.py` | MCP-Wrapper über die REST-API |
| `tts_agent.py` | Eigenständiger Konversationsagent (Ollama/OpenAI-kompatibel, max. 10 ReAct-Iterationen) |
### CLI pipeline (`chatterbox_cli_v4.py`)
```
Text input
→ strip_markdown() (Markdown-Syntax entfernen, opt-out via --no-strip-markdown)
→ clean_raw_text() (unsichtbare Zeichen + Emojis entfernen)
→ extract_language_spans() ([en]...[/en]-Markierungen → [(text, lang), ...])
→ split_into_sentences() (pro Span; Abkürzungen + nummerierte Listen korrekt behandelt)
→ preprocess_tts_text() (pro Chunk: Pronunciation-Dict → Einheiten → Zeiten → Jahre → Akronyme)
→ generate_chunk() (TTS mit span-spezifischer language_id)
→ PlaybackWorker (Audio-Ausgabe)
```
Reihenfolge ist kritisch: erst splitten (Satzgrenzen auf Rohtext erkennen), dann normalisieren (Akronym-Punkte würden sonst falsche Satzgrenzen erzeugen).
### Stop/Interrupt/Pause
Zwei modul-globale `threading.Event`-Objekte:
```python
STOP_REQUESTED = threading.Event()
PAUSE_REQUESTED = threading.Event()
request_stop() # setzt STOP_REQUESTED, löscht PAUSE_REQUESTED
clear_stop() # löscht STOP_REQUESTED (vor neuem Job)
stop_requested() # abfragen
request_pause() # setzt PAUSE_REQUESTED
request_resume() # löscht PAUSE_REQUESTED
is_paused() # abfragen
```
`PlaybackWorker._callback()` und beide Synthesize-Funktionen prüfen beide Events an Chunk-Grenzen. Ein laufendes `model.generate()` kann nicht mid-call abgebrochen werden (Python-Thread-Grenzen) — Stop/Pause greifen am nächsten Chunk. Pause hält Audio stumm und blockiert die Chunk-Schleife; Resume setzt sie fort ohne Datenverlust.
### Text preprocessing
#### `strip_markdown(text)`
Entfernt Markdown-Formatierung vor allen weiteren Schritten:
- `**fett**` / `*kursiv*` → Inhalt
- `` `code` `` / ```` ```block``` ```` → Inhalt / entfernt
- `# Überschrift` → Überschrift
- `- Listenpunkt` / `> Blockquote` → Inhalt
- `[Link](URL)` → Link-Text
- `---` (horizontale Linien) → entfernt
Default: aktiv. Deaktivierbar via `--no-strip-markdown`.
#### `clean_raw_text(text)`
- Entfernt unsichtbare Unicode-Zeichen (ZWSP, ZWNJ, BOM, …)
- Entfernt Emojis und nicht-druckbare Sondersymbole (`unicodedata.category` ∈ `{So, Cn, Co}`)
#### `extract_language_spans(text, default_lang)`
Zerlegt Text mit `[xx]...[/xx]`-Markierungen in `[(segment, lang), ...]`-Tupel:
```
"Das [en]Machine Learning[/en] Modell."
→ [("Das", "de"), ("Machine Learning", "en"), ("Modell.", "de")]
```
Ohne Markierungen: `[(text, default_lang)]` — identisches Verhalten wie bisher.
Jedes Segment wird mit der richtigen `language_id` an `generate_chunk()` übergeben.
### Text normalization (`preprocess_tts_text`)
1. Pronunciation dict (vor Akronym-Expansion, damit Eigennamen zuerst greifen)
2. Unit normalization (120 km/h → "120 Kilometer pro Stunde"; auch: °C, °F, W, V, A, kWh, m², m³, m/s, …)
3. Time normalization (14:58 → "vierzehn Uhr achtundfünfzig")
4. Year normalization (2026 → "zweitausendsechsundzwanzig"; bis Billionen)
5. Acronym spelling (ARD → "Ah Er De"; `NON_SPELLED_ACRONYMS` ausgenommen)
`DEFAULT_PRONUNCIATION_DE` enthält eingebaute phonetische Näherungen:
- Chinesische Namen: Xi → "Schi", Xi Jinping → "Schi Jinping"
- Tech-Marken: GitHub → "Git Hab", YouTube → "Jutjub", Wi-Fi → "Wai Fai", iPhone → "Aiphone", LinkedIn → "Linked In"
- KI-Begriffe: ChatGPT, OpenAI, GPT, LLM
`NON_SPELLED_ACRONYMS` (werden NICHT in deutsche Buchstabennamen umgewandelt):
- Internationale Org.: NATO, NASA, UNESCO, OPEC, IAEA, UNICEF
- Tech-Abkürzungen: USB, SSD, RAM, CPU, GPU, URL, API, PDF, LAN, WLAN, HTML, HTTP, HTTPS, JSON, SQL, VPN, SSH, FTP
- Titel: CEO, CFO, CTO, COO
### Text chunking
Drei Modi (CLI-Flags):
- **sentence_mode** (default): `split_into_sentences()` — ein Satz pro TTS-Call, geringste Latenz
- **conversation_mode**: `split_for_conversation()` — erster Chunk klein (`--first-chunk-len`, default 80), Rest bis `--len` (400)
- **plain**: `split_long_text()` — absatzbasiertes Chunking bis `--len`
`split_into_sentences()` behandelt:
- **Abkürzungen** (`_ABBREV_MASK_RE`): z.B., d.h., Dr., Prof., Nr., etc. werden nicht als Satzenden erkannt
- **Nummerierte Listen**: `"1. Punkt\n2. Punkt"` → jeder Listenpunkt wird als eigener Chunk behandelt
- **Überlange Sätze**: `force_split_sentence` sucht bei Überlänge vorwärts zum nächsten Wortende
### Model loading (`load_model`)
- `--lang en` → `ChatterboxTTS` (mono, immer verfügbar)
- Andere Sprachen → `ChatterboxMultilingualTTS` (`HAS_MULTILINGUAL`-Flag bewacht Import; `except (ImportError, ModuleNotFoundError)`)
- `--t3-model v3` (default) oder `v2` wählt den multilingualen T3-Checkpoint
- Modelle werden in `~/.cache/huggingface/` gecacht (~2–3 GB)
- **Kritisch**: `attn_implementation = "eager"` wird beim Import erzwungen — SDPA gibt `None`-Attention-Weights zurück und bricht den `AlignmentStreamAnalyzer`-Hook
### Audio output (`PlaybackWorker`)
- Vor Stream-Start: `sd.query_devices(self.device)` prüft Gerät-Existenz frühzeitig
- `sounddevice.OutputStream` mit Callback bei 48 kHz (PipeWire/PulseAudio-Standard)
- Interner Producer-Thread: Torch-Tensoren → `CALLBACK_BLOCK`-große (2048 Samples) numpy-Arrays
- `--speed != 1.0`: pyrubberband R3-Engine (`--fine`) streckt Zeit ohne Pitch-Änderung, dann Resampling via `torchaudio.functional.resample(chunk, model_sr, 48000)`
- `PlaybackWorker.stop()` schickt `None`-Sentinel in die Queue und jointed den Thread
- Bei `PAUSE_REQUESTED`: Callback gibt Stille aus, Chunk-Schleife wartet
### Two synthesis paths
- **`synthesize_non_streaming`**: generiert jeden Chunk vollständig, füttert fertige Tensoren in `PlaybackWorker`, concateniert alle WAVs für `--save`; unterstützt `[en]...[/en]`-Sprachmarkierungen pro Chunk
- **`synthesize_streaming`**: ruft `model.generate_stream()` mit `chunk_size` auf; jeder Audio-Sub-Chunk geht direkt in `PlaybackWorker`; experimentell
### HTTP Service (`tts_service.py`)
- **Modell-Cache**: `_model_cache: dict[(lang, t3_model), (model, kind, sr)]` — einmal laden, halten; Thread-sicher via `_model_lock`
- **Modell-Warmup**: `TTS_PRELOAD_LANG=de` lädt das Modell beim Service-Start (kein Cold-Start-Delay beim ersten Request)
- **Job-Queue**: `queue.Queue[SpeakJob]` mit einzelnem Worker-Thread; verhindert parallelen GPU/Audio-Zugriff
- **`SpeakRequest.interrupt`**: ruft `request_stop()` + `_drain_queue()` vor dem Einreihen auf
- **Pause/Resume**: `POST /pause` → `request_pause()`, `POST /resume` → `request_resume()`; ohne Datenverlust, Job bleibt in Queue
- **Status**: `_current_job`, `_recent_jobs` (max. 20) via `_state_lock` thread-safe lesbar

411
README.md
View file

@ -1,412 +1,3 @@
# chatterbox-tts-cli
Ein lokaler Text-to-Speech-Assistent auf Basis von
[Chatterbox TTS](https://github.com/resemble-ai/chatterbox) (Resemble AI).
Optimiert für deutsche Sprache; nutzbar als Kommandozeilen-Tool, als lokaler
HTTP-Service und als MCP-Server für KI-Assistenten.
## Features
- **Satz-für-Satz-Ausgabe** — gibt den ersten Satz aus, während die nächsten bereits generiert werden; minimale Latenz
- **Lückenlose Audiowiedergabe** — Callback-basierter OutputStream; keine Unterbrechungen zwischen Sätzen
- **Pause/Resume** — Ausgabe pausieren und fortsetzen ohne Datenverlust (`POST /pause`, `POST /resume`)
- **Geschwindigkeitsanpassung** — pitch-erhaltende Zeitstreckung via pyrubberband (R3-Engine); `--speed 0.5`–`2.0`
- **Voice Cloning** — optionale WAV-Referenz für Akzent und Klang
- **Mehrsprachig** — Deutsch, Englisch und 20+ weitere Sprachen via `ChatterboxMultilingualTTS`
- **Gemischtsprachige Texte** — `[en]...[/en]`-Markierungen für englische Passagen in deutschen Texten
- **Deutsche Textnormalisierung** — Abkürzungen (ARD → „Ah Er De"), Uhrzeiten (14:58 → „vierzehn Uhr achtundfünfzig"), Jahreszahlen bis Milliarden, Einheiten (°C, °F, kWh, m², …), Aussprache-Wörterbuch
- **Markdown-Bereinigung** — entfernt `**fett**`, `# Überschrift`, Links, Code-Blöcke automatisch vor der Synthese
- **HTTP-Service** — FastAPI-Service mit Job-Queue, Stop/Pause/Interrupt, Status-Endpunkt
- **MCP-Adapter** — direkte Integration in Claude Code, Claude Desktop und andere MCP-Hosts
- **Systemd-Autostart** — Service startet automatisch beim Login
---
## Systemvoraussetzungen
- Python 3.11+
- CUDA-GPU empfohlen (RTX 3070 oder besser; CPU möglich, aber langsam)
- Linux mit PipeWire oder PulseAudio
- `rubberband-cli` für Geschwindigkeitsanpassung:
```bash
sudo apt install rubberband-cli
```
---
## Installation
```bash
# 1. Conda-Umgebung
conda create -n chatterbox python=3.11
conda activate chatterbox
# 2. PyTorch mit CUDA (Beispiel CUDA 12.4)
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu124
# 3. Alle Abhängigkeiten
pip install -r requirements.txt
```
Beim ersten Start mit `--lang de` werden Modelle automatisch heruntergeladen (~2–3 GB, `~/.cache/huggingface/`).
---
## Kommandozeilen-CLI
```bash
conda activate chatterbox
# Deutschen Text vorlesen
python chatterbox_cli_v4.py --lang de --input text.txt
# Mit Voice Cloning
python chatterbox_cli_v4.py --lang de --voice stimme.wav --input text.txt
# Text direkt übergeben
python chatterbox_cli_v4.py --lang en --text "Hello, how are you?"
# Langsamer sprechen (pitch bleibt gleich)
python chatterbox_cli_v4.py --lang de --speed 0.85 --input text.txt
# Nur speichern, nicht abspielen
python chatterbox_cli_v4.py --lang de --no-play --output ausgabe.wav --input text.txt
# Aussprache-Wörterbuch
python chatterbox_cli_v4.py --lang de --pronunciation-dict aussprache.json --input text.txt
# Gemischtsprachiger Text
python chatterbox_cli_v4.py --lang de --text \
"Das [en]Machine Learning[/en] Modell kostet ca. 50 €."
# Markdown-Bereinigung deaktivieren
python chatterbox_cli_v4.py --lang de --no-strip-markdown --input text.txt
```
### CLI-Optionen
| Option | Standard | Beschreibung |
|--------|----------|--------------|
| `--text TEXT` | — | Text direkt als Argument |
| `--input DATEI` | — | UTF-8-Textdatei |
| `--lang CODE` | `de` | Sprachcode (de, en, fr, es, …) |
| `--voice DATEI.wav` | — | Referenz-WAV für Voice Cloning (10–30 s) |
| `--speed N` | `1.0` | Wiedergabegeschwindigkeit (0.5–2.0) |
| `--audio-device` | `pulse` | Ausgabegerät (z. B. `pulse`, `default`) |
| `--t3-model` | `v3` | Multilingual-Modell: `v3` oder `v2` |
| `--acronym-mode` | `german` | Akronym-Modus: `german`, `space`, `period_space` |
| `--pronunciation-dict` | — | JSON-Datei mit Aussprache-Substitutionen |
| `--no-strip-markdown` | — | Markdown-Formatierung nicht entfernen |
| `--save` | nein | WAV-Datei speichern |
| `--output DATEI.wav` | — | Ausgabepfad (impliziert `--save`) |
| `--no-play` | — | Nicht live abspielen |
| `--no-sentence-mode` | — | Größere Chunks statt satzweise |
| `--stream` | — | Streaming-Modus (experimentell) |
| `--no-progress` | — | Weniger Konsolenausgabe |
| `--debug-delay N` | `0` | Pause vor jedem Satz (zum Testen) |
| `--stop` | — | Laufende Ausgabe abbrechen |
---
## Gemischtsprachige Texte
Deutsche Texte enthalten oft englische Fachbegriffe, Markennamen oder Zitate.
Mit `[xx]...[/xx]`-Markierungen werden diese Passagen mit der richtigen `language_id`
an das Multilingual-Modell übergeben:
```
Das [en]Machine Learning[/en] Framework kostet ca. 50 €.
Der [en]CEO[/en] sagte: [en]"We are committed to innovation."[/en]
```
Ohne Markierungen verhält sich das System identisch wie bisher.
Häufige englische Tech-Begriffe werden bereits automatisch korrekt ausgesprochen
(eingebaut in `DEFAULT_PRONUNCIATION_DE`):
| Begriff | Aussprache |
|---------|-----------|
| GitHub | „Git Hab" |
| YouTube | „Jutjub" |
| LinkedIn | „Linked In" |
| Wi-Fi | „Wai Fai" |
| iPhone | „Aiphone" |
| ChatGPT | „Tschet Dschie Pie Tie" |
---
## HTTP-Service (`tts_service.py`)
FastAPI-Service mit Job-Queue und Worker-Thread. Startet automatisch via systemd.
```bash
# Manueller Start
uvicorn tts_service:app --host 0.0.0.0 --port 9999
# Mit Modell-Warmup (kein Cold-Start beim ersten Request)
TTS_PRELOAD_LANG=de uvicorn tts_service:app --host 0.0.0.0 --port 9999
# Systemd (Autostart, läuft bereits)
systemctl --user status chatterbox-tts
systemctl --user restart chatterbox-tts
journalctl --user -u chatterbox-tts -f
```
### Endpunkte
| Methode | Pfad | Funktion |
|---------|------|----------|
| `POST` | `/speak` | Text in Queue einreihen |
| `POST` | `/stop` | Ausgabe abbrechen, Queue leeren |
| `POST` | `/pause` | Ausgabe pausieren (ohne Datenverlust) |
| `POST` | `/resume` | Pausierte Ausgabe fortsetzen |
| `GET` | `/health` | Service-Status und Gerät |
| `GET` | `/status` | Aktueller Job, Queue-Länge, letzte Jobs |
| `GET` | `/voices` | Unterstützte Sprachen |
### `/speak` Request-Body
```json
{
"text": "Hallo Welt",
"lang": "de",
"voice": null,
"interrupt": false,
"speed": 1.0,
"t3_model": "v3",
"audio_device": null,
"max_len": 400,
"save_wav": false,
"output_path": null,
"pronunciation_dict": null,
"session_id": null
}
```
```bash
# Beispiel
curl -X POST http://localhost:9999/speak \
-H "Content-Type: application/json" \
-d '{"text": "Hallo Welt", "lang": "de"}'
# Aus dem LAN
curl -X POST http://192.168.x.x:9999/speak \
-H "Content-Type: application/json" \
-d '{"text": "Text aus dem Netzwerk", "lang": "de"}'
# Laufende Ausgabe unterbrechen
curl -X POST http://localhost:9999/speak \
-H "Content-Type: application/json" \
-d '{"text": "Wichtiger Text", "lang": "de", "interrupt": true}'
# Pausieren und fortsetzen
curl -X POST http://localhost:9999/pause
curl -X POST http://localhost:9999/resume
# Stoppen
curl -X POST http://localhost:9999/stop
```
---
## MCP-Adapter (`mcp_adapter.py`)
Dünner Wrapper über die REST-API für MCP-fähige Hosts.
```bash
# stdio-Modus (Claude Code / Claude Desktop)
python mcp_adapter.py --stdio
# HTTP-Modus (andere MCP-Clients, Port 8001)
python mcp_adapter.py
# Anderen TTS-Service ansprechen
TTS_URL=http://192.168.1.10:9999 python mcp_adapter.py --stdio
```
### MCP-Tools
| Tool | Parameter | Funktion |
|------|-----------|----------|
| `speak` | text, lang, voice, interrupt, speed, session_id | Text ausgeben |
| `stop` | — | Ausgabe stoppen und Queue leeren |
| `pause` | — | Ausgabe pausieren (ohne Datenverlust) |
| `resume` | — | Pausierte Ausgabe fortsetzen |
| `get_status` | — | Aktuellen Job und Queue abfragen |
| `list_voices` | — | Unterstützte Sprachen auflisten |
### Claude Code Konfiguration
Bereits eingerichtet via `claude mcp add --scope user`. Zur manuellen Einrichtung:
```bash
claude mcp add --scope user chatterbox-tts \
/home/dschlueter/miniforge3/envs/chatterbox/bin/python \
/home/dschlueter/chatterbox-tts-cli/mcp_adapter.py --stdio
```
### Claude Desktop (`~/.config/claude/claude_desktop_config.json`)
```json
{
"mcpServers": {
"chatterbox-tts": {
"command": "/home/dschlueter/miniforge3/envs/chatterbox/bin/python",
"args": ["/home/dschlueter/chatterbox-tts-cli/mcp_adapter.py", "--stdio"]
}
}
}
```
---
## Integration mit KI-Tools
### Claude Code / Claude Desktop — MCP (fertig eingerichtet)
Claude kann direkt die Tools `speak`, `stop`, `pause`, `resume`, `get_status` und `list_voices` aufrufen.
Kein weiterer Setup nötig.
### Ollama (llama3.2, qwen2.5, mistral-nemo u. a.)
Modelle mit Tool-Support können den REST-Service über Function Calling ansprechen:
```python
import ollama, httpx
tools = [{
"type": "function",
"function": {
"name": "speak",
"description": "Text als Sprache ausgeben",
"parameters": {
"type": "object",
"properties": {
"text": {"type": "string"},
"lang": {"type": "string", "default": "de"},
"speed": {"type": "number", "default": 1.0},
},
"required": ["text"],
},
},
}]
resp = ollama.chat(model="qwen2.5", messages=[{"role": "user", "content": "..."}], tools=tools)
for call in resp.message.tool_calls or []:
if call.function.name == "speak":
httpx.post("http://127.0.0.1:9999/speak", json=call.function.arguments)
```
### TTS Agent (`tts_agent.py`)
Eigenständiger Konversationsagent mit eingebautem Function Calling:
```bash
# Mit Ollama
python tts_agent.py --model qwen2.5
# Mit LM Studio
python tts_agent.py --base-url http://localhost:1234/v1 --model local-model
# Mit OpenAI
OPENAI_API_KEY=sk-... python tts_agent.py --model gpt-4o
# Mit Voice Cloning
python tts_agent.py --model qwen2.5 --voice my_voice.wav
```
### Open WebUI
Im Open-WebUI-Menü unter *Tools* eine neue Python-Klasse anlegen:
```python
import requests
class Tools:
def speak(self, text: str, lang: str = "de") -> str:
"""Text als Sprache ausgeben."""
r = requests.post("http://127.0.0.1:9999/speak",
json={"text": text, "lang": lang}, timeout=10)
return r.json().get("job_id", "error")
def stop(self) -> str:
"""Laufende Sprachausgabe stoppen."""
requests.post("http://127.0.0.1:9999/stop", timeout=5)
return "stopped"
```
### Home Assistant
```yaml
# configuration.yaml
rest_command:
tts_speak:
url: "http://192.168.x.x:9999/speak"
method: POST
content_type: "application/json"
payload: '{"text": "{{ text }}", "lang": "de"}'
tts_stop:
url: "http://192.168.x.x:9999/stop"
method: POST
tts_pause:
url: "http://192.168.x.x:9999/pause"
method: POST
tts_resume:
url: "http://192.168.x.x:9999/resume"
method: POST
```
Aufruf in einer Automation:
```yaml
service: rest_command.tts_speak
data:
text: "Die Waschmaschine ist fertig."
```
### Node-RED / n8n
HTTP-Request-Node direkt auf `POST http://<host>:9999/speak` mit JSON-Body.
Kein weiterer Setup nötig.
---
## Aussprache-Wörterbuch
Für Namen oder Begriffe, die das Modell falsch ausspricht:
```json
{
"Xi Jinping": "Schi Dschinping",
"Putin": "Pjutin",
"Seoul": "Söul",
"Kubernetes": "Kubernetis"
}
```
```bash
python chatterbox_cli_v4.py --lang de \
--pronunciation-dict aussprache.json \
--input nachricht.txt
```
Häufige Begriffe sind bereits eingebaut (GitHub, YouTube, iPhone, Xi Jinping u. a.).
Das eigene Dict wird immer **nach** dem eingebauten angewendet — Überschreibungen sind möglich.
---
## Bekannte Einschränkungen
- **Wortbetonung** lässt sich nicht steuern — kein SSML. Abhilfe: Voice-Referenz mit gewünschter Betonung.
- **Laufendes `model.generate()`** kann nicht mid-call abgebrochen werden (Python-Thread-Grenzen); Stop/Pause greift am nächsten Chunk-Beginn.
- **Sprachmarkierungen `[en]...[/en]`** funktionieren nur mit `ChatterboxMultilingualTTS`; bei `--lang en` (mono) werden sie ignoriert.
- **Streaming-Modus** unterstützt keine Sprachmarkierungen.
---
## Lizenz
MIT — dieses Skript. Das Chatterbox-Modell: MIT-Lizenz (Resemble AI). Modellgewichte: CC BY-NC 4.0.
Ein lokaler Text-to-Speech-Assistent auf Basis von [Chatterbox TTS](https://github.com/resemble-ai/chatterbox) (Resemble AI). Optimiert für deutsche Sprache; nutzbar als Kommandozeilen-Tool, als lokaler HTTP-Service und als MCP-Server für KI-Assistenten.

18
Trump_in_China_kurz.txt
View file

@ -1,18 +0,0 @@
Schlagzeile: Die neue Harmonie - mit Schlagseite
Stand: 15. Mai 2026 - 14:58
​Unterschlagzeile: Ein Besuch in großer Eintracht, so hatte es US-Präsident Trump schon vor Abreise nach China versprochen, und so kam es auch. Doch der Eindruck bleibt, dass die Gegenleistung überschaubar ausfällt.
Eine Analyse von Marie von Mallinckrodt, ARD Peking
Die Blaskapelle der Volksbefreiungsarmee spielt eine Version von "YMCA", eines der Lieblingslieder von US-Präsident Donald Trump. Serviert wird Lobster in Tomatensuppe, Peking-Ente und vieles mehr. Gegessen wird mit goldenem Besteck. Die Stimmung beim pompösen Staatsbankett in der Halle des Volkes ist feierlich und freundlich. Zwei ehemalige Rivalen stoßen auf ihre Annäherung an. Es ist diese Aussage des chinesischen Staats- und Parteichefs Xi Jinping an diesem Abend, die den Wesenskern der Beziehung der beiden Staatsoberhäupter wohl am besten beschreibt: "Die große Wiederbelebung der chinesischen Nation und das Ziel 'Make America great again' können Hand in Hand gehen. Wir können uns gegenseitig zum Erfolg verhelfen."
Zwischenzeile: Beide können zufrieden sein
Sowohl Xi als auch Trump führen eine nationalistische und auf Eigeninteressen bedachte Außenpolitik. Mit dem Unterschied, dass Xi einen langfristigen, strategischen Plan hat, wie er die "Wiederbelebung der Nation" erreichen will und Trump ein eher impulsives Deal-Making betreibt. Xi möchte China gern bis 2049 zur dominierenden, globalen Supermacht machen. So betrachtet fällt die Bilanz des zweitägigen Treffens ganz offenbar für beide nicht schlecht aus, wenn man die beiden wichtigsten Krisenfelder aus ihrer jeweiligen Perspektive betrachtet: Handel und Geopolitik.
--- Ende ---

1294
chatterbox_cli_v4.py
View file

File diff suppressed because it is too large Load diff

160
mcp_adapter.py
View file

@ -1,160 +0,0 @@
#!/usr/bin/env python3
"""
Chatterbox TTS – MCP-Adapter
Setzt einen laufenden tts_service.py voraus (Standard: http://127.0.0.1:9999).
Start (streamable-http, Port 8001 – für beliebige MCP-Clients):
python mcp_adapter.py
Start (stdio – für Claude Code / Claude Desktop):
python mcp_adapter.py --stdio
Claude Code (bereits konfiguriert via `claude mcp add --scope user`):
claude mcp add --scope user chatterbox-tts \
/home/dschlueter/miniforge3/envs/chatterbox/bin/python \
/home/dschlueter/chatterbox-tts-cli/mcp_adapter.py --stdio
Umgebungsvariable TTS_URL überschreibt die Service-Adresse:
TTS_URL=http://192.168.1.10:9999 python mcp_adapter.py --stdio
"""
from __future__ import annotations
import argparse
import os
import httpx
from mcp.server.fastmcp import FastMCP
TTS_URL = os.environ.get("TTS_URL", "http://127.0.0.1:9999").rstrip("/")
mcp = FastMCP(
"Chatterbox TTS",
instructions=(
"Lokaler Text-to-Speech-Service. Liest Texte auf Deutsch und 20+ weiteren "
"Sprachen vor. Unterstützt Voice Cloning, Geschwindigkeitsanpassung und "
"Aussprache-Wörterbücher."
),
port=8001,
)
# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------
def _raise_for_status(r: httpx.Response) -> None:
"""Wirft einen klaren Fehler bei HTTP-4xx/5xx statt rohem httpx-Fehler."""
try:
r.raise_for_status()
except httpx.HTTPStatusError as exc:
raise RuntimeError(
f"TTS-Service antwortet mit HTTP {exc.response.status_code}: {exc.response.text[:200]}"
) from exc
@mcp.tool()
async def speak(
text: str,
lang: str = "de",
voice: str | None = None,
interrupt: bool = False,
speed: float = 1.0,
session_id: str | None = None,
) -> dict:
"""Text als Sprache ausgeben.
Reiht den Text in die Ausgabewarteschlange ein. Das Modell generiert
satzweise und beginnt sofort mit der Wiedergabe.
Args:
text: Auszugebender Text (max. 4000 Zeichen).
lang: Sprachcode, z. B. 'de', 'en', 'fr'. Standard: 'de'.
voice: Optionaler Pfad zu einer WAV-Referenzdatei (10–30s) für
Voice Cloning.
interrupt: True = laufende Ausgabe sofort unterbrechen und diesen
Text vorgezogen abspielen.
speed: Wiedergabegeschwindigkeit (0.5–2.0). Pitch bleibt gleich.
session_id: Optionale Session-ID für Job-Tracking im TTS-Service.
"""
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{TTS_URL}/speak", json={
"text": text,
"lang": lang,
"voice": voice,
"interrupt": interrupt,
"speed": speed,
"session_id": session_id,
})
_raise_for_status(r)
return r.json()
@mcp.tool()
async def stop() -> dict:
"""Laufende Sprachausgabe sofort stoppen und Warteschlange leeren."""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{TTS_URL}/stop")
_raise_for_status(r)
return r.json()
@mcp.tool()
async def pause() -> dict:
"""Laufende Sprachausgabe pausieren (ohne Datenverlust). Mit resume() fortsetzen."""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{TTS_URL}/pause")
_raise_for_status(r)
return r.json()
@mcp.tool()
async def resume() -> dict:
"""Pausierte Sprachausgabe fortsetzen."""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.post(f"{TTS_URL}/resume")
_raise_for_status(r)
return r.json()
@mcp.tool()
async def get_status() -> dict:
"""Aktuellen Ausgabe-Status abfragen.
Gibt zurück: laufender Job (mit Chunk-Fortschritt), Queue-Länge und
die letzten abgeschlossenen Jobs.
"""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(f"{TTS_URL}/status")
_raise_for_status(r)
return r.json()
@mcp.tool()
async def list_voices() -> dict:
"""Unterstützte Sprachen und Hinweise zu Voice Cloning abfragen."""
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(f"{TTS_URL}/voices")
_raise_for_status(r)
return r.json()
# ---------------------------------------------------------------------------
# Einstiegspunkt
# ---------------------------------------------------------------------------
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Chatterbox TTS MCP-Adapter")
parser.add_argument(
"--stdio", action="store_true",
help="stdio-Transport (für Claude Code / Claude Desktop)",
)
parser.add_argument("--host", default="127.0.0.1",
help="Host für streamable-http (Standard: 127.0.0.1)")
parser.add_argument("--port", type=int, default=8001,
help="Port für streamable-http (Standard: 8001)")
args = parser.parse_args()
if args.stdio:
mcp.run() # stdio ist der Default-Transport
else:
mcp.run(transport="streamable-http", host=args.host, port=args.port)

29
requirements.txt
View file

@ -1,29 +0,0 @@
# Chatterbox TTS CLI — Abhängigkeiten
# Getestet mit Python 3.11, CUDA 12.x, Ubuntu 22.04/24.04
#
# PyTorch separat installieren (passende CUDA-Version via pytorch.org):
# pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu124
# --- TTS-Kern ---
chatterbox-tts>=0.1.7
# --- Audio-Ausgabe (Linux/PipeWire/PulseAudio) ---
sounddevice>=0.4.0
# --- Pitch-erhaltende Zeitstreckung (--speed != 1.0) ---
# Systempaket zusätzlich erforderlich: sudo apt install rubberband-cli
pyrubberband>=0.4.0
# --- HTTP-Service (tts_service.py) ---
fastapi>=0.115.0,<2.0
uvicorn[standard]>=0.32.0,<1.0
# --- HTTP-Client (mcp_adapter.py → tts_service.py) ---
httpx>=0.28.0
# --- MCP-Adapter (mcp_adapter.py) ---
mcp>=1.0.0
# --- TTS Agent (tts_agent.py) ---
# OpenAI-SDK als universeller Client für Ollama, LM Studio, OpenAI etc.
openai>=1.0.0

237
tts_agent.py
View file

@ -1,237 +0,0 @@
#!/usr/bin/env python3
"""
TTS Agent — conversational agent with speak/stop/status tools.
Works with any OpenAI-compatible LLM backend:
- Ollama: python tts_agent.py --model qwen2.5
- LM Studio: python tts_agent.py --base-url http://localhost:1234/v1 --model local-model
- OpenAI: OPENAI_API_KEY=sk-... python tts_agent.py --model gpt-4o
The agent automatically calls speak() when it should read text aloud.
Set TTS_URL to override the default TTS service address.
Pi (Inflection AI) is not supported — it has no public API or function calling.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import httpx
TTS_URL = os.environ.get("TTS_URL", "http://127.0.0.1:9999").rstrip("/")
SYSTEM_PROMPT = """Du bist ein hilfreicher Sprachassistent.
Du hast Zugriff auf einen Text-to-Speech-Service mit folgenden Tools:
- speak(text, lang, speed, interrupt): Text vorlesen lassen
- stop(): laufende Ausgabe stoppen
- get_status(): Ausgabe-Status abfragen
Wenn der Nutzer darum bittet, etwas vorzulesen, oder wenn du eine längere Antwort
gibst, die zum Vorlesen geeignet ist, ruf speak() mit dem entsprechenden Text auf.
Antworte auf Deutsch, sofern nicht anders gewünscht."""
TOOLS = [
{
"type": "function",
"function": {
"name": "speak",
"description": (
"Text als Sprache ausgeben. Reiht den Text in die Warteschlange ein "
"und gibt sofort zurück. Ideal für längere Texte oder wenn der Nutzer "
"etwas vorgelesen haben möchte."
),
"parameters": {
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "Auszugebender Text (max. 4000 Zeichen).",
},
"lang": {
"type": "string",
"default": "de",
"description": "Sprachcode: 'de', 'en', 'fr', 'es', … Standard: 'de'.",
},
"speed": {
"type": "number",
"default": 1.0,
"description": "Geschwindigkeit 0.5–2.0. Standard: 1.0.",
},
"interrupt": {
"type": "boolean",
"default": False,
"description": "True = laufende Ausgabe sofort unterbrechen.",
},
},
"required": ["text"],
},
},
},
{
"type": "function",
"function": {
"name": "stop",
"description": "Laufende Sprachausgabe sofort stoppen und Warteschlange leeren.",
"parameters": {"type": "object", "properties": {}},
},
},
{
"type": "function",
"function": {
"name": "get_status",
"description": "Aktuellen Ausgabe-Status abfragen (laufender Job, Queue-Länge).",
"parameters": {"type": "object", "properties": {}},
},
},
]
def _call_tts(name: str, args: dict) -> str:
try:
with httpx.Client(timeout=15) as client:
if name == "speak":
r = client.post(f"{TTS_URL}/speak", json=args)
elif name == "stop":
r = client.post(f"{TTS_URL}/stop")
elif name == "get_status":
r = client.get(f"{TTS_URL}/status")
else:
return json.dumps({"error": f"Unbekanntes Tool: {name}"})
r.raise_for_status()
return r.text
except httpx.ConnectError:
return json.dumps({"error": f"TTS-Service nicht erreichbar: {TTS_URL}"})
except Exception as exc: # noqa: BLE001
return json.dumps({"error": str(exc)})
def run_agent(model: str, base_url: str, system_prompt: str, voice: str | None) -> None:
try:
from openai import OpenAI
except ImportError:
print("Fehler: 'openai' nicht installiert. → pip install openai", file=sys.stderr)
sys.exit(1)
client = OpenAI(
base_url=base_url,
api_key=os.environ.get("OPENAI_API_KEY", "ollama"),
)
if voice:
system_prompt += f"\n\nWenn du speak() aufrufst, übergib immer voice='{voice}'."
messages: list[dict] = [{"role": "system", "content": system_prompt}]
# IMPROVE-7: TTS-Service-Erreichbarkeit früh prüfen
try:
with httpx.Client(timeout=5) as _hc:
_hc.get(f"{TTS_URL}/health")
except Exception as _e:
print(f"Fehler: TTS-Service nicht erreichbar ({TTS_URL}): {_e}", file=sys.stderr)
sys.exit(1)
print(f"Agent gestartet | Modell: {model} | TTS: {TTS_URL}")
print("Beenden mit 'exit' oder Ctrl+C.\n")
while True:
try:
user_input = input("Du: ").strip()
except (KeyboardInterrupt, EOFError):
print("\nTschüss!")
break
if user_input.lower() in ("exit", "quit", "bye", "tschüss"):
print("Tschüss!")
break
if not user_input:
continue
messages.append({"role": "user", "content": user_input})
# BUG-3: Maximale Iterationen verhindern Endlosschleife bei LLMs,
# die wiederholt Tool-Calls ohne abschließende Antwort produzieren.
max_iterations = 10
for _iteration in range(max_iterations):
try:
resp = client.chat.completions.create(
model=model,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
except Exception as exc: # noqa: BLE001
print(f"[LLM-Fehler] {exc}", file=sys.stderr)
break
msg = resp.choices[0].message
# BUG-6: explizites Dict statt model_dump(exclude_unset=True),
# damit 'role' auch bei strict-kompatiblen Backends immer vorhanden ist.
msg_dict: dict = {"role": msg.role, "content": msg.content}
if msg.tool_calls:
msg_dict["tool_calls"] = [tc.model_dump() for tc in msg.tool_calls]
messages.append(msg_dict)
if not msg.tool_calls:
if msg.content:
print(f"\nAssistent: {msg.content}\n")
break
for tc in msg.tool_calls:
args = json.loads(tc.function.arguments)
arg_str = ", ".join(f"{k}={v!r}" for k, v in args.items())
print(f" [{tc.function.name}({arg_str})]")
result = _call_tts(tc.function.name, args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
else:
print(f"[Agent] Maximale Tool-Call-Iterationen ({max_iterations}) erreicht.", file=sys.stderr)
def main() -> None:
p = argparse.ArgumentParser(
description="TTS Agent — LLM mit speak/stop/status Tools",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
p.add_argument(
"--model", default="qwen2.5",
help="Modellname (Standard: qwen2.5). Für OpenAI z. B. 'gpt-4o'.",
)
p.add_argument(
"--base-url", default="http://localhost:11434/v1",
help="OpenAI-kompatibler API-Endpunkt (Standard: Ollama auf Port 11434).",
)
p.add_argument(
"--system-prompt", default=SYSTEM_PROMPT,
help="System-Prompt überschreiben.",
)
p.add_argument(
"--voice", default=None,
help="Pfad zu einer WAV-Referenzdatei für Voice Cloning (wird an speak() weitergegeben).",
)
p.add_argument(
"--lang", default=None,
help="Sprache für speak() überschreiben (z. B. 'en'). Standard: Modell entscheidet.",
)
args = p.parse_args()
system = args.system_prompt
if args.lang:
system += f"\n\nVerwende immer lang='{args.lang}' beim Aufruf von speak()."
run_agent(
model=args.model,
base_url=args.base_url,
system_prompt=system,
voice=args.voice,
)
if __name__ == "__main__":
main()

306
tts_service.py
View file

@ -1,306 +0,0 @@
#!/usr/bin/env python3
"""
Chatterbox TTS – lokaler HTTP-Service
Start:
uvicorn tts_service:app --host 0.0.0.0 --port 9999
Endpunkte:
POST /speak – Text in Warteschlange einreihen
POST /stop – laufende Ausgabe abbrechen, Queue leeren
GET /health – Service-Status
GET /status – aktueller Job + Queue-Länge
GET /voices – unterstützte Sprachen
"""
from __future__ import annotations
import queue
import sys
import threading
import uuid
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional
# CLI-Modul aus demselben Verzeichnis laden
sys.path.insert(0, str(Path(__file__).parent))
import chatterbox_cli_v4 as tts # noqa: E402
import torch
import torchaudio as ta
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
# ---------------------------------------------------------------------------
# Gerät einmalig bestimmen
# ---------------------------------------------------------------------------
_DEVICE = tts.get_device(None)
# ---------------------------------------------------------------------------
# Modell-Cache (lang, t3_model) → (model, model_kind, sr)
# ---------------------------------------------------------------------------
_model_cache: dict[tuple, tuple] = {}
_model_lock = threading.Lock()
def _get_or_load_model(lang: str, t3_model: str) -> tuple:
key = (lang, t3_model)
with _model_lock:
if key not in _model_cache:
_model_cache[key] = tts.load_model(lang, _DEVICE, t3_model=t3_model)
return _model_cache[key]
# Optionaler Warmup: TTS_PRELOAD_LANG=de lädt das Modell beim Service-Start,
# damit der erste Request keine Modell-Ladezeit hat.
_PRELOAD_LANG = __import__("os").environ.get("TTS_PRELOAD_LANG")
if _PRELOAD_LANG:
_preload_t3 = __import__("os").environ.get("TTS_PRELOAD_T3", "v3")
try:
_get_or_load_model(_PRELOAD_LANG, _preload_t3)
print(f"[chatterbox-tts] Modell vorgeladen: lang={_PRELOAD_LANG}, t3={_preload_t3}")
except Exception as _e:
print(f"[chatterbox-tts] Warnung: Warmup fehlgeschlagen: {_e}")
# ---------------------------------------------------------------------------
# Job-Datenmodell
# ---------------------------------------------------------------------------
class JobStatus(str, Enum):
pending = "pending"
running = "running"
done = "done"
cancelled = "cancelled"
error = "error"
@dataclass
class SpeakJob:
id: str
text: str
lang: str
t3_model: str
voice: Optional[str]
speed: float
audio_device: str
max_len: int
save_wav: bool
output_path: Optional[str]
pronunciation_dict: Optional[dict]
session_id: Optional[str]
status: JobStatus = field(default=JobStatus.pending)
text_preview: str = field(default="")
chunks_total: int = 0
chunks_done: int = 0
error: Optional[str] = None
# ---------------------------------------------------------------------------
# Worker-Thread
# ---------------------------------------------------------------------------
_job_queue: queue.Queue[SpeakJob] = queue.Queue()
_current_job: Optional[SpeakJob] = None
_state_lock = threading.Lock()
_recent_jobs: list[SpeakJob] = []
_MAX_RECENT = 20
def _worker() -> None:
global _current_job
while True:
job = _job_queue.get()
with _state_lock:
_current_job = job
job.status = JobStatus.running
tts.clear_stop()
try:
model, model_kind, sr = _get_or_load_model(job.lang, job.t3_model)
raw = tts.clean_raw_text(job.text)
raw_chunks = tts.split_into_sentences(raw, max_len=job.max_len)
chunks = [
tts.preprocess_tts_text(c, lang=job.lang,
pronunciation_dict=job.pronunciation_dict)
for c in raw_chunks
]
chunks = [c for c in chunks if c.strip()]
job.chunks_total = len(chunks)
job.text_preview = job.text[:80]
playback = tts.PlaybackWorker(
sample_rate=sr,
device=job.audio_device or "pulse",
speed=job.speed,
stop_event=tts.STOP_REQUESTED,
)
playback.start()
wavs: list[torch.Tensor] = []
try:
for chunk in chunks:
if tts.stop_requested():
break
wav = tts.generate_chunk(model, model_kind, chunk, job.lang, job.voice)
wavs.append(wav)
playback.put(wav)
job.chunks_done += 1
finally:
playback.stop()
if job.save_wav and job.output_path and wavs:
out = Path(job.output_path)
out.parent.mkdir(parents=True, exist_ok=True)
final = wavs[0] if len(wavs) == 1 else torch.cat(wavs, dim=-1)
ta.save(str(out), final, sr)
job.status = (
JobStatus.cancelled if tts.stop_requested() else JobStatus.done
)
except Exception as exc: # noqa: BLE001
job.status = JobStatus.error
job.error = str(exc)
finally:
with _state_lock:
_current_job = None
_recent_jobs.append(job)
if len(_recent_jobs) > _MAX_RECENT:
_recent_jobs.pop(0)
_job_queue.task_done()
_worker_thread = threading.Thread(target=_worker, daemon=True, name="tts-worker")
_worker_thread.start()
# ---------------------------------------------------------------------------
# API-Modelle
# ---------------------------------------------------------------------------
class SpeakRequest(BaseModel):
text: str = Field(min_length=1, max_length=4000)
lang: str = "de"
voice: Optional[str] = None
interrupt: bool = False
speed: float = Field(default=1.0, ge=0.5, le=2.0)
t3_model: str = "v3"
audio_device: Optional[str] = None
max_len: int = Field(default=400, ge=100, le=1000)
save_wav: bool = False
output_path: Optional[str] = None
session_id: Optional[str] = None
pronunciation_dict: Optional[dict] = None
def _job_to_dict(j: SpeakJob) -> dict:
return {
"id": j.id,
"status": j.status,
"lang": j.lang,
"text_preview": j.text_preview,
"chunks_total": j.chunks_total,
"chunks_done": j.chunks_done,
"error": j.error,
}
def _drain_queue() -> None:
while not _job_queue.empty():
try:
_job_queue.get_nowait()
_job_queue.task_done()
except queue.Empty:
break
# ---------------------------------------------------------------------------
# FastAPI-App
# ---------------------------------------------------------------------------
app = FastAPI(title="Chatterbox TTS Service", version="1.0")
@app.get("/health")
def health():
return {"status": "ok", "device": _DEVICE}
@app.get("/voices")
def voices():
return {
"languages": sorted(tts.SUPPORTED_LANGS),
"note": "Voice cloning via 'voice' field (WAV-Pfad, 10–30s Aufnahme)",
}
@app.post("/speak")
def speak(req: SpeakRequest):
if req.lang not in tts.SUPPORTED_LANGS:
raise HTTPException(status_code=422,
detail=f"Sprache nicht unterstützt: {req.lang}")
if req.voice and not Path(req.voice).exists():
raise HTTPException(status_code=422,
detail=f"Voice-Datei nicht gefunden: {req.voice}")
if req.interrupt:
tts.request_stop()
_drain_queue()
job = SpeakJob(
id=str(uuid.uuid4()),
text=req.text,
lang=req.lang,
t3_model=req.t3_model,
voice=req.voice,
speed=req.speed,
audio_device=req.audio_device,
max_len=req.max_len,
save_wav=req.save_wav,
output_path=req.output_path,
pronunciation_dict=req.pronunciation_dict,
session_id=req.session_id,
)
_job_queue.put(job)
return {
"job_id": job.id,
"status": job.status,
"queue_position": _job_queue.qsize(),
}
@app.post("/stop")
def stop():
tts.request_stop()
_drain_queue()
return {"stopped": True}
@app.post("/pause")
def pause():
tts.request_pause()
return {"paused": True}
@app.post("/resume")
def resume():
tts.request_resume()
return {"resumed": True}
@app.get("/status")
def status():
with _state_lock:
cur = _current_job
recent = list(_recent_jobs)
return {
"current_job": _job_to_dict(cur) if cur else None,
"queue_length": _job_queue.qsize(),
"recent_jobs": [_job_to_dict(j) for j in reversed(recent)],
}