Der Stacktrace zeigt ggfs. solch ein Problem: Das Modell
dolphin-mixtral existiert nicht lokal in Ihrer
Ollama-Installation. Dann muss das natürlich behandelt werden. Der
Fehler kommt allerdings erst beim invoke-Aufruf, nicht bei
der Initialisierung. Das ist ein wichtiges Detail, das Ihre
Fehlerbehandlungsstrategie beeinflusst.
LangChain’s init_chat_model ist lazy – es validiert
nicht sofort, ob das Modell tatsächlich verfügbar ist. Die Funktion
erstellt lediglich ein Client-Objekt, das erst beim ersten API-Call
(hier: invoke) tatsächlich mit dem Modell kommuniziert. Das
bedeutet: Ihr Programm startet erfolgreich, scheitert aber zur Laufzeit
beim ersten echten Request.
Diese verzögerte Validierung hat Vor- und Nachteile. Vorteil: Schnellerer Programmstart, keine unnötigen Netzwerk-Calls während der Initialisierung. Nachteil: Fehler werden erst spät sichtbar, möglicherweise nach aufwändiger Preprocessing-Arbeit.
Für produktionsnahe Systeme brauchen Sie mehrere Verteidigungslinien.
Die einfachste Variante: Fangen Sie die Exception beim Modell-Laden
ab und behandeln Sie sie explizit. Da die eigentliche Validierung erst
beim ersten invoke passiert, können Sie einen Dummy-Test
durchführen.
from langchain_core.prompts import ChatPromptTemplate
from langchain.chat_models import init_chat_model
from ollama._types import ResponseError
import sys
def initialize_model_safely(model_name: str, temperature: float = 0.2):
"""
Initialisiert ein Chat-Modell mit sofortiger Validierung.
Führt einen Test-Request durch, um sicherzustellen, dass das Modell
tatsächlich verfügbar ist. Wirft aussagekräftige Exceptions bei Problemen.
"""
try:
llm = init_chat_model(model_name, temperature=temperature)
# Validierungs-Ping: Minimaler Request zum Testen der Verfügbarkeit
test_prompt = ChatPromptTemplate([("user", "test")])
llm.invoke(test_prompt.invoke({}))
print(f"✓ Modell '{model_name}' erfolgreich geladen und validiert")
return llm
except ResponseError as e:
# Ollama-spezifischer Fehler (404, 500, etc.)
if e.status_code == 404:
print(f"✗ Fehler: Modell '{model_name}' nicht gefunden", file=sys.stderr)
print(f" Verfügbare Modelle prüfen mit: ollama list", file=sys.stderr)
print(f" Modell herunterladen mit: ollama pull {model_name.split(':')[1]}",
file=sys.stderr)
else:
print(f"✗ Ollama-Fehler: {e} (Status: {e.status_code})", file=sys.stderr)
raise
except ConnectionError as e:
# Ollama-Service läuft nicht
print(f"✗ Fehler: Kann nicht mit Ollama verbinden", file=sys.stderr)
print(f" Ist der Ollama-Service gestartet?", file=sys.stderr)
raise
except Exception as e:
# Unerwartete Fehler
print(f"✗ Unerwarteter Fehler beim Laden von '{model_name}': {e}",
file=sys.stderr)
raise
# Verwendung
try:
llm = initialize_model_safely("ollama:dolphin-mixtral", temperature=0.2)
# Ab hier ist garantiert, dass das Modell funktioniert
except Exception:
sys.exit(1)Diese Variante scheitert früh und laut. Der Test-Request kostet minimal Performance (ein paar Millisekunden), gibt Ihnen aber Gewissheit. In Produktionsumgebungen möchten Sie wissen, dass Ihre Modelle verfügbar sind, bevor Sie komplexe Chains aufbauen.
In vielen Szenarien ist die spezifische Modellwahl nicht kritisch. Sie haben eine Präferenz, aber könnten mit einem Fallback leben. Hier macht eine Kaskaden-Strategie Sinn.
from langchain.chat_models import init_chat_model
from typing import List, Tuple, Optional
def initialize_with_fallback(
model_preferences: List[Tuple[str, dict]],
validate: bool = True
) -> Optional[object]:
"""
Versucht Modelle in Reihenfolge der Präferenz zu laden.
Args:
model_preferences: Liste von (model_name, kwargs) Tupeln
validate: Ob jedes Modell mit Test-Request validiert werden soll
Returns:
Das erste erfolgreich geladene Modell, oder None
"""
for model_name, kwargs in model_preferences:
try:
llm = init_chat_model(model_name, **kwargs)
if validate:
# Schneller Validierungs-Check
from langchain_core.messages import HumanMessage
llm.invoke([HumanMessage(content="test")])
print(f"✓ Verwende Modell: {model_name}")
return llm
except Exception as e:
print(f"⚠ Modell '{model_name}' nicht verfügbar: {e}")
continue
return None
# Definition der Fallback-Kette
model_preferences = [
("ollama:dolphin-mixtral", {"temperature": 0.2}),
("ollama:llama2", {"temperature": 0.2}),
("ollama:mistral", {"temperature": 0.2}),
# Wenn gar nichts lokal läuft: OpenAI als letzter Fallback
# ("openai:gpt-3.5-turbo", {"temperature": 0.2, "api_key": "..."})
]
llm = initialize_with_fallback(model_preferences)
if llm is None:
print("✗ Kein Modell verfügbar. Programm wird beendet.")
sys.exit(1)
# Ab hier: llm ist garantiert funktionalDiese Strategie ist besonders nützlich in heterogenen Umgebungen. Entwickler haben vielleicht unterschiedliche lokale Modelle installiert, oder ein Modell ist temporär nicht verfügbar. Der Code degradiert graceful, statt komplett zu scheitern.
In größeren Projekten gehört die Modellkonfiguration nicht in den Code, sondern in Konfigurationsdateien. Hier eine robuste Variante mit expliziter Validierung.
import os
from pathlib import Path
from typing import Optional
import json
class ModelConfig:
"""Kapselt Modellkonfiguration mit Validierung."""
def __init__(self, config_path: str = "model_config.json"):
self.config_path = Path(config_path)
self.config = self._load_config()
def _load_config(self) -> dict:
"""Lädt und validiert Konfiguration."""
if not self.config_path.exists():
# Fallback auf Standardkonfiguration
return {
"primary_model": "ollama:llama2",
"fallback_models": ["ollama:mistral"],
"temperature": 0.2,
"validate_on_init": True
}
with open(self.config_path) as f:
return json.load(f)
def get_llm(self):
"""
Lädt das konfigurierte Modell mit vollständiger Fehlerbehandlung.
"""
models_to_try = [self.config["primary_model"]] + \
self.config.get("fallback_models", [])
temperature = self.config.get("temperature", 0.2)
validate = self.config.get("validate_on_init", True)
for model_name in models_to_try:
try:
llm = init_chat_model(model_name, temperature=temperature)
if validate:
# Test-Request zur Validierung
from langchain_core.messages import HumanMessage
llm.invoke([HumanMessage(content="ping")])
print(f"✓ Aktives Modell: {model_name}")
return llm
except Exception as e:
print(f"⚠ Überspringe {model_name}: {type(e).__name__}")
continue
raise RuntimeError(
f"Kein Modell verfügbar. Geprüfte Modelle: {models_to_try}\n"
f"Stellen Sie sicher, dass Ollama läuft und Modelle installiert sind."
)
# Verwendung
try:
config = ModelConfig()
llm = config.get_llm()
except RuntimeError as e:
print(f"✗ Kritischer Fehler: {e}", file=sys.stderr)
sys.exit(1)Die zugehörige model_config.json könnte so aussehen:
{
"primary_model": "ollama:dolphin-mixtral",
"fallback_models": [
"ollama:llama2",
"ollama:mistral"
],
"temperature": 0.2,
"validate_on_init": true
}Diese Variante trennt Konfiguration von Code und ermöglicht umgebungsspezifische Anpassungen ohne Code-Änderungen. In Docker-Deployments können Sie verschiedene Configs mounten, in Tests andere Modelle verwenden.
Manchmal scheitert ein Request nicht, weil das Modell fehlt, sondern wegen temporärer Probleme (Netzwerk, Überlastung, Memory). Hier hilft ein Retry-Wrapper.
from langchain.chat_models import init_chat_model
from langchain_core.messages import BaseMessage
from typing import List
import time
class ResilientChatModel:
"""
Wrapper um ein Chat-Modell mit Retry-Logik und besserer Fehlerbehandlung.
"""
def __init__(self, model_name: str, max_retries: int = 3, **kwargs):
self.model_name = model_name
self.max_retries = max_retries
self.kwargs = kwargs
self.llm = None
self._initialize()
def _initialize(self):
"""Initialisiert das Modell mit Fehlerbehandlung."""
try:
self.llm = init_chat_model(self.model_name, **self.kwargs)
# Validierungs-Ping
from langchain_core.messages import HumanMessage
self.llm.invoke([HumanMessage(content="test")])
except Exception as e:
raise RuntimeError(
f"Modell '{self.model_name}' konnte nicht initialisiert werden: {e}"
) from e
def invoke(self, messages: List[BaseMessage], **kwargs):
"""
Ruft das Modell mit automatischer Retry-Logik auf.
"""
last_exception = None
for attempt in range(self.max_retries):
try:
return self.llm.invoke(messages, **kwargs)
except ResponseError as e:
# 404 = Modell fehlt -> kein Retry sinnvoll
if e.status_code == 404:
raise
# Andere Fehler: Retry mit Backoff
last_exception = e
wait_time = 2 ** attempt # Exponential Backoff
print(f"⚠ Versuch {attempt + 1}/{self.max_retries} fehlgeschlagen, "
f"warte {wait_time}s...")
time.sleep(wait_time)
except Exception as e:
# Unerwartete Fehler
last_exception = e
break
# Alle Retries aufgebraucht
raise RuntimeError(
f"Request nach {self.max_retries} Versuchen fehlgeschlagen"
) from last_exception
# Verwendung
try:
llm = ResilientChatModel("ollama:llama2", temperature=0.2, max_retries=3)
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate([("user", "Erkläre {topic}")])
prompt = template.invoke({"topic": "Closures"})
response = llm.invoke(prompt.to_messages())
print(response.content)
except RuntimeError as e:
print(f"✗ Finale Fehler: {e}")
sys.exit(1)Die robusteste Lösung: Trennen Sie Model-Initialisierung von Business-Logik. Implementieren Sie einen Health-Check, der beim Programmstart alle kritischen Abhängigkeiten prüft.
import sys
from typing import Dict, Any
class ApplicationHealth:
"""Prüft alle kritischen Komponenten vor dem Programmstart."""
def __init__(self):
self.checks = []
self.results = {}
def add_check(self, name: str, check_fn):
"""Registriert eine Health-Check-Funktion."""
self.checks.append((name, check_fn))
def run_all_checks(self) -> bool:
"""Führt alle Checks aus und gibt True zurück, wenn alles OK ist."""
all_passed = True
print("=== System Health Check ===")
for name, check_fn in self.checks:
try:
check_fn()
self.results[name] = "✓ OK"
print(f" {self.results[name]} {name}")
except Exception as e:
self.results[name] = f"✗ FAILED: {e}"
print(f" {self.results[name]}")
all_passed = False
print("===========================")
return all_passed
def get_report(self) -> Dict[str, Any]:
"""Gibt einen detaillierten Report zurück."""
return self.results
def check_ollama_connection():
"""Prüft, ob Ollama erreichbar ist."""
import requests
response = requests.get("http://localhost:11434/api/tags", timeout=2)
if response.status_code != 200:
raise ConnectionError("Ollama nicht erreichbar")
def check_model_availability(model_name: str):
"""Factory für modellspezifische Checks."""
def check():
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
llm = init_chat_model(model_name, temperature=0)
llm.invoke([HumanMessage(content="test")])
return check
# Health-Check-Routine
health = ApplicationHealth()
health.add_check("Ollama Service", check_ollama_connection)
health.add_check("Model: llama2", check_model_availability("ollama:llama2"))
if not health.run_all_checks():
print("\n✗ System nicht bereit. Bitte Fehler beheben.", file=sys.stderr)
sys.exit(1)
print("\n✓ Alle Checks bestanden, starte Anwendung...\n")
# Ab hier: Alle kritischen Komponenten sind verfügbarDiese Variante eignet sich besonders für containerisierte Deployments mit Kubernetes-Health-Probes oder systemd-Service-Management. Der Service startet erst, wenn alle Abhängigkeiten verfügbar sind.
Entwicklungsumgebung: Strategie 1 (fail-fast) oder 2 (fallback) sind ideal. Sie wollen schnelles Feedback, ob Ihre lokale Setup funktioniert.
CI/CD-Pipeline: Strategie 5 (health-check) ist Pflicht. Die Pipeline muss wissen, ob das Deployment funktionsfähig ist.
Produktionsumgebung: Kombination aus 3 (Config-basiert), 4 (Retry-Logik) und 5 (Health-Checks). Robustheit ist kritisch, und Sie wollen nicht wegen eines fehlenden Modells um 3 Uhr nachts geweckt werden.
Experimentelle Notebooks: Strategie 2 (fallback) mit manueller Fehlerbehandlung. Flexibilität ist wichtiger als Robustheit.
Die zentrale Erkenntnis: Modell-Initialisierung ist ein kritischer Pfad in LLM-Anwendungen. Behandeln Sie sie mit derselben Sorgfalt wie Datenbankverbindungen oder API-Credentials. Lazy Initialization ist praktisch, aber Sie brauchen explizite Validierung an den richtigen Stellen.