WebSocket AI 实时对话：断线重连与心跳机制实现

Es war ein Freitagnachmittag, als unser Produktionssystem plötzlich begann, seltsame Fehler zu werfen: ConnectionError: timeout after 30000ms gefolgt von einer Kaskade weiterer WebSocket connection failed-Meldungen. Die Benutzer beschwerten sich über abgehackte Konversationen, und unser SRE-Team fand heraus, dass die AI-Services von HolySheep AI (Jetzt registrieren) zwar stabil liefen, aber unsere Client-Implementierung nicht robust genug für instabile Netzwerkbedingungen war. In diesem Tutorial zeige ich Ihnen, wie Sie eine professionelle WebSocket-Implementierung mit automatischer断线重连 (Reconnection) und 心跳机制 (Heartbeat) aufbauen.

Warum WebSocket für AI Echtzeit-Dialog?

Traditionelle REST-APIs sind für interaktive AI-Konversationen ungeeignet, da jede Anfrage einen neuen HTTP-Handshake erfordert und Latenzzeiten von 200-500ms entstehen. WebSocket bietet dagegen:

• Permanente Verbindung: Bidirektionale Kommunikation ohne wiederholte Handshakes
• <50ms Latenz: Bei HolySheep AI typisch 35-45ms für Text Streaming
• Effiziente Ressourcennutzung: Nur eine TCP-Verbindung für gesamte Konversation
• Server-Push: AI-Antworten werden tokenweise in Echtzeit übertragen

Die komplette Implementierung

1. WebSocket Manager Klasse

import websocket
import threading
import time
import json
import logging
from typing import Callable, Optional, Dict, Any
from datetime import datetime

class HolySheepWebSocketManager:
    """
    WebSocket Manager für HolySheep AI mit automatischer Reconnection
    und Heartbeat-Mechanismus.
    
    API-Dokumentation: https://docs.holysheep.ai/websocket
    """
    
    def __init__(
        self,
        api_key: str,
        model: str = "gpt-4.1",
        base_url: str = "https://api.holysheep.ai/v1",
        on_message: Optional[Callable[[str], None]] = None,
        on_error: Optional[Callable[[Exception], None]] = None,
        on_connect: Optional[Callable[[], None]] = None,
        on_disconnect: Optional[Callable[[int, str], None]] = None
    ):
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip('/')
        self.websocket_url = f"{self.base_url}/ws/chat"
        
        # Callbacks
        self.on_message = on_message or (lambda x: None)
        self.on_error = on_error or (lambda x: None)
        self.on_connect = on_connect or (lambda: None)
        self.on_disconnect = on_disconnect or (lambda x, y: None)
        
        # Connection State
        self.ws: Optional[websocket.WebSocketApp] = None
        self.is_connected = False
        self.should_reconnect = True
        
        # Heartbeat Configuration
        self.heartbeat_interval = 30  # Sekunden
        self.heartbeat_timeout = 10   # Sekunden für Ping-Antwort
        self.last_pong_time: Optional[datetime] = None
        
        # Reconnection Configuration
        self.max_reconnect_attempts = 10
        self.reconnect_delay = 1      # Start-Verzögerung in Sekunden
        self.max_reconnect_delay = 60 # Max-Verzögerung in Sekunden
        self.current_reconnect_attempt = 0
        
        # Threading
        self.ws_thread: Optional[threading.Thread] = None
        self.heartbeat_thread: Optional[threading.Thread] = None
        
        self.logger = logging.getLogger(__name__)
        
    def connect(self) -> bool:
        """Initialisiert die WebSocket-Verbindung."""
        self.should_reconnect = True
        self.current_reconnect_attempt = 0
        
        try:
            self.ws = websocket.WebSocketApp(
                self.websocket_url,
                header=self._get_headers(),
                on_message=self._on_message,
                on_error=self._on_error,
                on_open=self._on_open,
                on_close=self._on_close,
                on_pong=self._on_pong
            )
            
            self.ws_thread = threading.Thread(
                target=self._run_websocket,
                daemon=True
            )
            self.ws_thread.start()
            
            self.logger.info(f"WebSocket Verbindung zu {self.websocket_url} initiiert")
            return True
            
        except Exception as e:
            self.logger.error(f"Verbindungsfehler: {e}")
            self.on_error(e)
            self._schedule_reconnect()
            return False
    
    def _get_headers(self) -> Dict[str, str]:
        """Generiert Authentifizierungs-Header."""
        return {
            "Authorization": f"Bearer {self.api_key}",
            "X-Model": self.model,
            "X-Client": "HolySheep-Python-SDK/1.0"
        }
    
    def _run_websocket(self):
        """Führt den WebSocket-Event-Loop aus."""
        while self.should_reconnect:
            try:
                self.ws.run_forever(
                    ping_interval=self.heartbeat_interval,
                    ping_timeout=self.heartbeat_timeout,
                    reconnect=0  # Wir handhaben Reconnection manuell
                )
            except Exception as e:
                self.logger.error(f"WebSocket Loop Fehler: {e}")
                
            if self.should_reconnect:
                self._schedule_reconnect()
    
    def _on_open(self, ws):
        """Callback bei erfolgreicher Verbindung."""
        self.is_connected = True
        self.current_reconnect_attempt = 0
        self.last_pong_time = datetime.now()
        self.logger.info("✅ WebSocket Verbindung hergestellt")
        self.on_connect()
        self._start_heartbeat_monitor()
    
    def _on_message(self, ws, message: str):
        """Verarbeitet eingehende Nachrichten."""
        try:
            data = json.loads(message)
            
            # Herzschlag-Antwort
            if data.get("type") == "pong":
                self.last_pong_time = datetime.now()
                self.logger.debug("🫀 Pong empfangen")
                return
            
            # AI-Token
            if data.get("type") == "token" or "content" in data:
                content = data.get("content", data.get("delta", ""))
                self.on_message(content)
                
            # Fehlerbehandlung
            if data.get("type") == "error":
                error_msg = data.get("message", "Unbekannter Fehler")
                self.logger.error(f"Server-Fehler: {error_msg}")
                self.on_error(Exception(error_msg))
                
        except json.JSONDecodeError as e:
            self.logger.warning(f"Ungültiges JSON: {e}, Nachricht: {message}")
    
    def _on_error(self, ws, error):
        """Behandelt WebSocket-Fehler."""
        self.logger.error(f"⚠️ WebSocket Fehler: {error}")
        self.on_error(error)
    
    def _on_close(self, ws, close_status_code: int, close_msg: str):
        """Callback bei Verbindungsende."""
        self.is_connected = False
        self._stop_heartbeat_monitor()
        self.logger.warning(f"🔌 Verbindung geschlossen: {close_status_code} - {close_msg}")
        self.on_disconnect(close_status_code, close_msg)
    
    def _on_pong(self, ws, data: bytes = None):
        """Verarbeitet Pong-Antworten."""
        self.last_pong_time = datetime.now()
        self.logger.debug(f"🫀 Pong empfangen von Server")
    
    def _start_heartbeat_monitor(self):
        """Startet den Heartbeat-Überwachungsthread."""
        self.heartbeat_thread = threading.Thread(
            target=self._heartbeat_loop,
            daemon=True
        )
        self.heartbeat_thread.start()
    
    def _stop_heartbeat_monitor(self):
        """Stoppt den Heartbeat-Thread."""
        if self.heartbeat_thread and self.heartbeat_thread.is_alive():
            self.heartbeat_thread.join(timeout=2)
    
    def _heartbeat_loop(self):
        """Überwachungsschleife für Heartbeat-Status."""
        while self.is_connected and self.should_reconnect:
            try:
                time.sleep(10)  # Alle 10 Sekunden prüfen
                
                if not self.is_connected:
                    break
                    
                if self.last_pong_time:
                    elapsed = (datetime.now() - self.last_pong_time).total_seconds()
                    if elapsed > self.heartbeat_timeout * 3:
                        self.logger.warning(
                            f"⚠️ Keine Heartbeat-Antwort seit {elapsed:.1f}s. "
                            "Verbindung möglicherweise tot."
                        )
                        self._trigger_reconnect()
                        
            except Exception as e:
                self.logger.error(f"Heartbeat Monitor Fehler: {e}")
    
    def _schedule_reconnect(self):
        """Plant einen Reconnection-Versuch mit exponentiellem Backoff."""
        if self.current_reconnect_attempt >= self.max_reconnect_attempts:
            self.logger.error(
                f"Maximale Reconnect-Versuche ({self.max_reconnect_attempts}) erreicht"
            )
            self.should_reconnect = False
            return
        
        # Exponentieller Backoff: 1s, 2s, 4s, 8s, ... max 60s
        delay = min(
            self.reconnect_delay * (2 ** self.current_reconnect_attempt),
            self.max_reconnect_delay
        )
        
        self.current_reconnect_attempt += 1
        self.logger.info(
            f"🔄 Reconnect-Versuch {self.current_reconnect_attempt}/"
            f"{self.max_reconnect_attempts} in {delay:.1f}s"
        )
        
        time.sleep(delay)
        
        if self.should_reconnect and not self.is_connected:
            self.connect()
    
    def _trigger_reconnect(self):
        """Löst sofortige Reconnection aus."""
        if self.ws:
            try:
                self.ws.close()
            except:
                pass
        self._schedule_reconnect()
    
    def send_message(self, content: str, session_id: Optional[str] = None) -> bool:
        """Sendet eine Nachricht an den Server."""
        if not self.is_connected or not self.ws:
            self.logger.error("❌ Nicht verbunden. Nachricht konnte nicht gesendet werden.")
            return False
        
        message = {
            "type": "message",
            "content": content,
            "model": self.model
        }
        
        if session_id:
            message["session_id"] = session_id
            
        try:
            self.ws.send(json.dumps(message))
            self.logger.debug(f"📤 Nachricht gesendet: {content[:50]}...")
            return True
        except Exception as e:
            self.logger.error(f"❌ Sendefehler: {e}")
            self.on_error(e)
            return False
    
    def disconnect(self):
        """Trennt die Verbindung sauber."""
        self.logger.info("🛑 Verbindung wird getrennt...")
        self.should_reconnect = False
        
        if self.ws:
            try:
                self.ws.close()
            except:
                pass
        
        self._stop_heartbeat_monitor()
        self.is_connected = False

from flask import Flask, render_template, request, jsonify, session
import uuid
import logging

app = Flask(__name__)
app.secret_key = 'ihr-sicherer-schluessel'
logging.basicConfig(level=logging.INFO)

============================================================
HOLYSHEEP AI KONFIGURATION
Preise 2026 (USD per 1M Tokens):
- GPT-4.1: $8.00 (85%+ günstiger als OpenAI)
- Claude Sonnet 4.5: $15.00
- Gemini 2.5 Flash: $2.50
- DeepSeek V3.2: $0.42
#
Vorteile: WeChat/Alipay Zahlung, <50ms Latenz,
kostenlose Credits bei Anmeldung
https://www.holysheep.ai/register
============================================================

Ihr HolySheep API Key (NIEMALS hardcodieren in Produktion!)
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_WS_URL = "https://api.holysheep.ai/v1/ws/chat"

from holy_sheep_websocket import HolySheepWebSocketManager

Dictionary zur Verwaltung aktiver Verbindungen
active_connections: dict[str, HolySheepWebSocketManager] = {}

@app.route('/')
def index():
    """Startseite mit Chat-Interface."""
    if 'session_id' not in session:
        session['session_id'] = str(uuid.uuid4())
    return render_template('chat.html', session_id=session['session_id'])

@app.route('/api/connect', methods=['POST'])
def api_connect():
    """Initialisiert eine neue WebSocket-Verbindung."""
    try:
        data = request.get_json() or {}
        session_id = data.get('session_id', session.get('session_id', str(uuid.uuid4())))
        
        if session_id in active_connections:
            active_connections[session_id].disconnect()
        
        collected_response = []
        
        def on_message(content: str):
            collected_response.append(content)
            print(f"Token empfangen: {content}", end='', flush=True)
        
        def on_error(error: Exception):
            print(f"Fehler: {error}")
            
        def on_connect():
            print("✅ Mit HolySheep AI verbunden!")
            
        def on_disconnect(code: int, msg: str):
            print(f"🔌 Getrennt: {code} - {msg}")
        
        ws_manager = HolySheepWebSocketManager(
            api_key=HOLYSHEEP_API_KEY,
            model=data.get('model', 'gpt-4.1'),
            on_message=on_message,
            on_error=on_error,
            on_connect=on_connect,
            on_disconnect=on_disconnect
        )
        
        active_connections[session_id] = ws_manager
        
        if ws_manager.connect():
            return jsonify({
                'status': 'connecting',
                'session_id': session_id,
                'message': 'Verbindung wird hergestellt...'
            }), 200
        else:
            return jsonify({
                'status': 'error',
                'message': 'Verbindung fehlgeschlagen'
            }), 500
            
    except Exception as e:
        logging.error(f"Verbindungsfehler: {e}")
        return jsonify({
            'status': 'error',
            'message': str(e)
        }), 500

@app.route('/api/send', methods=['POST'])
def api_send():
    """Sendet eine Nachricht über die aktive Verbindung."""
    try:
        data = request.get_json()
        session_id = data.get('session_id', session.get('session_id'))
        message = data.get('message', '')
        
        if not session_id or session_id not in active_connections:
            return jsonify({
                'status': 'error',
                'message': 'Keine aktive Verbindung'
            }), 400
        
        ws_manager = active_connections[session_id]
        
        if ws_manager.send_message(message, session_id):
            return jsonify({
                'status': 'sent',
                'message': 'Nachricht gesendet'
            }), 200
        else:
            return jsonify({
                'status': 'error',
                'message': 'Senden fehlgeschlagen'
            }), 500
            
    except Exception as e:
        logging.error(f"Sendefehler: {e}")
        return jsonify({
            'status': 'error',
            'message': str(e)
        }), 500

@app.route('/api/disconnect', methods=['POST'])
def api_disconnect():
    """Trennt eine bestehende Verbindung."""
    try:
        data = request.get_json() or {}
        session_id = data.get('session_id', session.get('session_id'))
        
        if session_id and session_id in active_connections:
            active_connections[session_id].disconnect()
            del active_connections[session_id]
            return jsonify({'status': 'disconnected'}), 200
        else:
            return jsonify({
                'status': 'error',
                'message': 'Keine aktive Verbindung gefunden'
            }), 404
            
    except Exception as e:
        logging.error(f"Trennungsfehler: {e}")
        return jsonify({
            'status': 'error',
            'message': str(e)
        }), 500

if __name__ == '__main__':
    print("=" * 60)
    print("🌟 HolySheep AI WebSocket Chat Server")
    print("=" * 60)
    print("💡 API URL: https://api.holysheep.ai/v1")
    print("💰 Preise 2026: GPT-4.1 $8, DeepSeek V3.2 $0.42 per 1M Tokens")
    print("✅ <50ms Latenz | WeChat/Alipay Unterstützung")
    print("📝 Jetzt registrieren: https://www.holysheep.ai/register")
    print("=" * 60)
    app.run(host='0.0.0.0', port=5000, debug=True, threaded=True)

# FEHLERHAFT: Kein Timeout-Handling
ws = websocket.WebSocketApp(url)
ws.run_forever()  # Blockiert endlos bei Netzwerkproblemen

LÖSUNG: Konfiguriere explizites Timeout
def safe_connect(ws_url: str, api_key: str, timeout: int = 30) -> bool:
    """
    Stellt sichere Verbindung mit Timeout her.
    
    Args:
        ws_url: WebSocket Endpunkt
        api_key: API-Schlüssel
        timeout: Timeout in Sekunden (Standard: 30)
    
    Returns:
        True bei Erfolg, False bei Timeout
    """
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        ws = websocket.WebSocketApp(
            ws_url,
            header=headers
        )
        
        # Timeout für run_forever konfigurieren
        ws.run_forever(
            ping_interval=30,
            ping_timeout=10,
            reconnect=0  # Eigene Reconnection-Logik verwenden
        )
        return True
        
    except websocket.WebSocketTimeoutException:
        print("⏱️ Verbindung timeout nach {}s".format(timeout))
        return False
    except Exception as e:
        print("❌ Verbindungsfehler: {}".format(e))
        return False

# FEHLERHAFT: Falscher Header-Name
headers = {
    "Api-Key": api_key  # ❌ Falsch!
}

LÖSUNG: Korrektes Bearer-Token Format
def validate_and_connect(api_key: str, base_url: str) -> bool:
    """
    Validiert API-Key und stellt Verbindung her.
    
    Raises:
        ValueError: Bei ungültigem oder fehlendem API-Key
        ConnectionError: Bei Verbindungsfehlern
    """
    # Validierung
    if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
        raise ValueError(
            "❌ API-Key fehlt oder ist nicht konfiguriert!\n"
            "📝 Registrieren Sie sich hier: https://www.holysheep.ai/register"
        )
    
    if len(api_key) < 20:
        raise ValueError(
            "❌ API-Key scheint zu kurz zu sein. "
            "Bitte überprüfen Sie Ihren HolySheep AI Key."
        )
    
    # Korrekter Header für HolySheep AI
    ws_url = f"{base_url.rstrip('/')}/ws/chat"
    headers = {
        "Authorization": f"Bearer {api_key}",  # ✅ Richtig!
        "Content-Type": "application/json"
    }
    
    print(f"🔐 Verbinde mit API-Endpunkt: {ws_url}")
    
    try:
        ws = websocket.WebSocketApp(
            ws_url,
            header=headers
        )
        return True
    except Exception as e:
        raise ConnectionError(f"Verbindung fehlgeschlagen: {e}")

# FEHLERHAFT: Keine Reconnection-Logik
def simple_send(ws, message):
    ws.send(message)
    # Verbindung bleibt offen - keine Fehlerbehandlung!

LÖSUNG: Vollständige Reconnection mit Heartbeat
class RobustWebSocketClient:
    def __init__(self, url: str, api_key: str):
        self.url = url
        self.api_key = api_key
        self.ws = None
        self.reconnect_count = 0
        self.max_retries = 5
        
    def _on_close(self, ws, close_status_code: int, close_msg: str):
        """
        Behandelt unerwartete Verbindungstrennungen.
        
        Status Codes:
        - 1000: Normaler Close
        - 1001: Server Going Away
        - 1006: Abnormal Close (hier Problem!)
        - 1008: Policy Violation
        - 1011: Unexpected Error
        """
        print(f"🔌 Verbindung geschlossen: {close_status_code} - {close_msg}")
        
        if close_status_code == 1006:
            print("⚠️ Server hat Verbindung unerwartet geschlossen!")
            print("   Mögliche Ursachen:")
            print("   - API-Key ungültig oder deaktiviert")
            print("   - Request-Limit überschritten")
            print("   - Server-Überlastung")
        
        self._attempt_reconnect()
    
    def _attempt_reconnect(self):
        """Versucht Reconnection mit Backoff."""
        if self.reconnect_count >= self.max_retries:
            print("❌ Maximale Retry-Versuche erreicht!")
            return
            
        self.reconnect_count += 1
        delay = min(2 ** self.reconnect_count, 60)  # Max 60s
        
        print(f"🔄 Reconnect-Versuch {self.reconnect_count}/{self.max_retries} in {delay}s...")
        time.sleep(delay)
        
        self.connect()
    
    def connect(self):
        """Stellt Verbindung mit Heartbeat her."""
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        self.ws = websocket.WebSocketApp(
            self.url,
            header=headers,
            on_close=self._on_close
        )
        
        # Heartbeat: Ping alle 30s, Timeout 10s
        self.ws.run_forever(
            ping_interval=30,
            ping_timeout=10
        )