File size: 7,300 Bytes
9eb915e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 |
"""
utils/api_handler.py - API-Kommunikationsmodul für den Dr. Franz Psychochatbot
Dieses Modul verwaltet die Kommunikation mit der HuggingFace Inference API:
- Senden von Anfragen
- Verarbeiten von Antworten
- Fehlerbehandlung und Wiederholungslogik
"""
import requests
import time
import random
import json
from typing import Dict, Any, Optional, List, Union
# Importieren der Konfiguration
import config
class ApiHandler:
"""Klasse zur Verwaltung der API-Kommunikation"""
def __init__(self, api_token: str = config.API_TOKEN, model_id: str = config.MODEL_ID):
"""
Initialisiert den API-Handler
Args:
api_token: Der API-Token für HuggingFace
model_id: Die ID des zu verwendenden Modells
"""
self.api_token = api_token
self.model_id = model_id
self.api_url = f"{config.API_BASE_URL}{model_id}"
self.headers = {"Authorization": f"Bearer {api_token}"}
self.max_retries = 3
self.retry_delay = 2 # Sekunden
def generate_text(self, prompt: str, params: Optional[Dict[str, Any]] = None) -> str:
"""
Generiert Text über die HuggingFace Inference API
Args:
prompt: Der Prompt für die Textgenerierung
params: Optional, Parameter für die Textgenerierung
Returns:
Der generierte Text
"""
# Standardparameter verwenden, wenn keine angegeben sind
if params is None:
params = {
"max_new_tokens": config.MAX_NEW_TOKENS,
"do_sample": True,
"temperature": config.TEMPERATURE,
"top_p": config.TOP_P,
"top_k": config.TOP_K,
"repetition_penalty": config.REPETITION_PENALTY
}
payload = {
"inputs": prompt,
"parameters": params
}
return self._send_api_request(payload)
def _send_api_request(self, payload: Dict[str, Any]) -> str:
"""
Sendet eine Anfrage an die API mit Wiederholungslogik
Args:
payload: Die Anfragedaten
Returns:
Die generierte Antwort
"""
for attempt in range(self.max_retries):
try:
response = requests.post(
self.api_url,
headers=self.headers,
json=payload,
timeout=config.API_TIMEOUT
)
# Prüfen auf erfolgreiche Antwort
if response.status_code == 200:
return self._process_successful_response(response.json())
# Behandlung von Fehlercodes
if response.status_code == 401:
if config.DEBUG_MODE:
print("API-Fehler: Ungültiger API-Token")
return self._get_fallback_response("authentication_error")
if response.status_code == 503 or response.status_code == 429:
# Server überlastet oder Rate-Limit erreicht, erneuter Versuch
if config.DEBUG_MODE:
print(f"API-Fehler: Server überlastet (Versuch {attempt+1}/{self.max_retries})")
time.sleep(self.retry_delay * (attempt + 1)) # Exponentielles Backoff
continue
# Andere Fehler
if config.DEBUG_MODE:
print(f"API-Fehler: Statuscode {response.status_code}")
return self._get_fallback_response("general_error")
except requests.exceptions.Timeout:
if config.DEBUG_MODE:
print(f"API-Timeout (Versuch {attempt+1}/{self.max_retries})")
if attempt < self.max_retries - 1:
time.sleep(self.retry_delay * (attempt + 1))
else:
return self._get_fallback_response("timeout")
except Exception as e:
if config.DEBUG_MODE:
print(f"API-Anfragefehler: {e}")
return self._get_fallback_response("general_error")
# Alle Versuche fehlgeschlagen
return self._get_fallback_response("max_retries")
def _process_successful_response(self, response_data: Union[List[Dict[str, Any]], Dict[str, Any]]) -> str:
"""
Verarbeitet eine erfolgreiche API-Antwort
Args:
response_data: Die Antwortdaten von der API
Returns:
Der extrahierte generierte Text
"""
try:
if isinstance(response_data, list) and len(response_data) > 0 and "generated_text" in response_data[0]:
full_response = response_data[0]["generated_text"]
# Extrahieren der Antwort nach "Dr. Franz:"
if "Dr. Franz:" in full_response:
reply = full_response.split("Dr. Franz:")[-1].strip()
return reply
else:
return full_response
elif isinstance(response_data, dict) and "generated_text" in response_data:
return response_data["generated_text"]
else:
if config.DEBUG_MODE:
print(f"Unerwartetes API-Antwortformat: {response_data}")
return self._get_fallback_response("unexpected_format")
except Exception as e:
if config.DEBUG_MODE:
print(f"Fehler bei der Verarbeitung der API-Antwort: {e}")
return self._get_fallback_response("processing_error")
def _get_fallback_response(self, error_type: str) -> str:
"""
Gibt eine Fallback-Antwort basierend auf dem Fehlertyp zurück
Args:
error_type: Der Typ des aufgetretenen Fehlers
Returns:
Eine passende Fallback-Antwort
"""
# Spezifische Antworten je nach Fehlertyp
error_responses = {
"authentication_error": [
"Es scheint ein Problem mit meiner Verbindung zu geben. Erzählen Sie mir mehr über Ihre Gedanken.",
"Ich muss kurz nachdenken. Was meinen Sie genau mit Ihrer letzten Aussage?"
],
"timeout": [
"Ihre Aussage ist komplex und bedarf tieferer Analyse. Könnten Sie Ihren Gedanken weiter ausführen?",
"Interessant, dass Sie das so formulieren. Ich brauche einen Moment, um das zu verarbeiten."
],
"max_retries": [
"Ihre Gedankengänge sind bemerkenswert verschlungen. Lassen Sie uns einen Schritt zurückgehen.",
"Ich spüre Widerstand in unserer Kommunikation. Vielleicht sollten wir einen anderen Ansatz versuchen."
]
}
# Spezifische Antwort für den Fehlertyp oder allgemeine Fallback-Antwort
if error_type in error_responses:
return random.choice(error_responses[error_type])
else:
return random.choice(config.FALLBACK_RESPONSES)
|