Heim >Backend-Entwicklung >Python-Tutorial >Integration großer Sprachmodelle in Produktionsanwendungen
In diesem praktischen Leitfaden erfahren Sie, wie Sie eine hoch skalierbare Modellbereitstellungslösung mit integrierten LLMs für Ihre Anwendungen erstellen.
In Ihren Beispielen verwenden wir das ChatGPT2-Modell von Hugging Face, Sie können jedoch problemlos jedes andere Modell anschließen, einschließlich ChatGPT4, Claude usw.
Unabhängig davon, ob Sie eine neue Anwendung mit KI-Funktionen entwerfen oder die vorhandenen KI-Systeme verbessern, hilft Ihnen dieser Leitfaden Schritt für Schritt dabei, eine starke LLM-Integration zu schaffen.
Bevor wir mit dem Schreiben von Code beginnen, wollen wir herausfinden, was zum Aufbau einer Produktions-LLM-Integration erforderlich ist. Beim Aufbau einer produktionsreifen LLM-Integration müssen Sie nicht nur API-Aufrufe berücksichtigen, sondern auch Dinge wie Zuverlässigkeit, Kosten und Stabilität. Ihre Produktionsanwendungen müssen Probleme wie Dienstausfälle, Ratenbeschränkungen und Schwankungen in der Reaktionszeit berücksichtigen und gleichzeitig die Kosten unter Kontrolle halten.
Folgendes werden wir gemeinsam aufbauen:
Bevor wir mit dem Codieren beginnen, stellen Sie sicher, dass Sie Folgendes haben:
Möchten Sie mitmachen? Der vollständige Code ist in Ihrem GitHub-Repository verfügbar.
Beginnen wir damit, Ihre Entwicklungsumgebung vorzubereiten. Wir erstellen eine saubere Projektstruktur und installieren alle notwendigen Pakete.
Erstellen wir zunächst Ihr Projektverzeichnis und richten Sie eine virtuelle Python-Umgebung ein. Öffnen Sie Ihr Terminal und führen Sie Folgendes aus:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
Jetzt richten wir Ihre Projektabhängigkeiten ein. Erstellen Sie eine neue „requirements.txt“-Datei mit diesen wesentlichen Paketen:
transformers==4.36.0 huggingface-hub==0.19.4 redis==4.6.0 pydantic==2.5.0 pydantic-settings==2.1.0 tenacity==8.2.3 python-dotenv==1.0.0 fastapi==0.104.1 uvicorn==0.24.0 torch==2.1.0 numpy==1.24.3
Lassen Sie uns zusammenfassen, warum wir jedes dieser Pakete benötigen:
Installieren Sie alle Pakete mit dem Befehl:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
Lassen Sie uns Ihr Projekt mit einer klaren Struktur organisieren. Erstellen Sie diese Verzeichnisse und Dateien in Ihrem Projektverzeichnis:
transformers==4.36.0 huggingface-hub==0.19.4 redis==4.6.0 pydantic==2.5.0 pydantic-settings==2.1.0 tenacity==8.2.3 python-dotenv==1.0.0 fastapi==0.104.1 uvicorn==0.24.0 torch==2.1.0 numpy==1.24.3
Beginnen wir mit Ihrem LLM-Client, der die wichtigste Komponente Ihrer Bewerbung darstellt. Hier interagieren wir mit dem ChatGPT-Modell (oder einem anderen LLM, das Sie bevorzugen). Fügen Sie die folgenden Codeausschnitte zu Ihrer core/llm_client.py-Datei hinzu:
pip install -r requirements.txt
In diesem ersten Teil der yourLLMClient-Klasse legen wir die Grundlage:
Jetzt fügen wir die Methode hinzu, die mit Ihrem Modell kommuniziert:
llm_integration/ ├── core/ │ ├── llm_client.py # your main LLM interaction code │ ├── prompt_manager.py # Handles prompt templates │ └── response_handler.py # Processes LLM responses ├── cache/ │ └── redis_manager.py # Manages your caching system ├── config/ │ └── settings.py # Configuration management ├── api/ │ └── routes.py # API endpoints ├── utils/ │ ├── monitoring.py # Usage tracking │ └── rate_limiter.py # Rate limiting logic ├── requirements.txt └── main.py └── usage_logs.json
Lassen Sie uns aufschlüsseln, was bei dieser Vervollständigungsmethode passiert:
Als nächstes müssen wir den Antworthandler hinzufügen, um die Rohausgabe des LLM zu analysieren und zu strukturieren. Tun Sie dies in Ihrer core/response_handler.py-Datei mit den folgenden Codeausschnitten:
import torch from transformers import AutoModelForCausalLM, AutoTokenizer from tenacity import retry, stop_after_attempt, wait_exponential from typing import Dict, Optional import logging class LLMClient: def __init__(self, model_name: str = "gpt2", timeout: int = 30): try: self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained( model_name, device_map="auto", torch_dtype=torch.float16 ) except Exception as e: logging.error(f"Error loading model: {str(e)}") # Fallback to a simpler model if the specified one fails self.tokenizer = AutoTokenizer.from_pretrained("gpt2") self.model = AutoModelForCausalLM.from_pretrained("gpt2") self.timeout = timeout self.logger = logging.getLogger(__name__)
Jetzt erstellen wir Ihr Caching-System, um die Anwendungsleistung zu verbessern und die Kosten zu senken. Fügen Sie die folgenden Codeausschnitte zu Ihrer Datei „cache/redis_manager.py“ hinzu:
@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) async def complete(self, prompt: str, temperature: float = 0.7, max_tokens: Optional[int] = None) -> Dict: """Get completion from the model with automatic retries""" try: inputs = self.tokenizer(prompt, return_tensors="pt").to( self.model.device ) with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=max_tokens or 100, temperature=temperature, do_sample=True ) response_text = self.tokenizer.decode( outputs[0], skip_special_tokens=True ) # Calculate token usage for monitoring input_tokens = len(inputs.input_ids[0]) output_tokens = len(outputs[0]) - input_tokens return { 'content': response_text, 'usage': { 'prompt_tokens': input_tokens, 'completion_tokens': output_tokens, 'total_tokens': input_tokens + output_tokens }, 'model': "gpt2" } except Exception as e: self.logger.error(f"Error in LLM completion: {str(e)}") raise
In den obigen Codeausschnitten haben wir eine CacheManager-Klasse erstellt, die alle Caching-Vorgänge wie folgt abwickelt:
Lassen Sie uns Ihren Prompt-Manager erstellen, der die Prompts für Ihr LLM-Modell verwaltet. Fügen Sie den folgenden Code zu Ihrer core/prompt_manager.py hinzu:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
Erstellen Sie dann eine Beispiel-Eingabeaufforderungsvorlage für die Inhaltsmoderation in Ihrer prompts/content_moderation.json-Datei mit Codeausschnitten:
transformers==4.36.0 huggingface-hub==0.19.4 redis==4.6.0 pydantic==2.5.0 pydantic-settings==2.1.0 tenacity==8.2.3 python-dotenv==1.0.0 fastapi==0.104.1 uvicorn==0.24.0 torch==2.1.0 numpy==1.24.3
Jetzt kann Ihr Eingabeaufforderungsmanager Eingabeaufforderungsvorlagen aus Ihrer JSON-Datei laden und erhält außerdem eine formatierte Eingabeaufforderungsvorlage.
Um alle Ihre LLM-Konfigurationen an einem Ort zu behalten und sie problemlos in Ihrer Anwendung wiederzuverwenden, erstellen wir Konfigurationseinstellungen. Fügen Sie den folgenden Code zu Ihrer Datei config/settings.py hinzu:
pip install -r requirements.txt
Als nächstes implementieren wir eine Ratenbegrenzung, um zu steuern, wie Benutzer auf die Ressourcen Ihrer Anwendung zugreifen. Fügen Sie dazu den folgenden Code zu Ihrer Datei utils/rate_limiter.py hinzu:
llm_integration/ ├── core/ │ ├── llm_client.py # your main LLM interaction code │ ├── prompt_manager.py # Handles prompt templates │ └── response_handler.py # Processes LLM responses ├── cache/ │ └── redis_manager.py # Manages your caching system ├── config/ │ └── settings.py # Configuration management ├── api/ │ └── routes.py # API endpoints ├── utils/ │ ├── monitoring.py # Usage tracking │ └── rate_limiter.py # Rate limiting logic ├── requirements.txt └── main.py └── usage_logs.json
Im RateLimiter haben wir eine wiederverwendbare check_rate_limit-Methode implementiert, die auf jeder Route verwendet werden kann, um die Ratenbegrenzung zu handhaben, indem einfach der Zeitraum und die Anzahl der für jeden Benutzer zulässigen Anforderungen für einen bestimmten Zeitraum übergeben werden.
Jetzt erstellen wir Ihre API-Endpunkte in der Datei api/routes.py, um Ihr LLM in Ihre Anwendung zu integrieren:
import torch from transformers import AutoModelForCausalLM, AutoTokenizer from tenacity import retry, stop_after_attempt, wait_exponential from typing import Dict, Optional import logging class LLMClient: def __init__(self, model_name: str = "gpt2", timeout: int = 30): try: self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained( model_name, device_map="auto", torch_dtype=torch.float16 ) except Exception as e: logging.error(f"Error loading model: {str(e)}") # Fallback to a simpler model if the specified one fails self.tokenizer = AutoTokenizer.from_pretrained("gpt2") self.model = AutoModelForCausalLM.from_pretrained("gpt2") self.timeout = timeout self.logger = logging.getLogger(__name__)
Hier haben wir einen /moderate-Endpunkt in der APIRouter-Klasse definiert, der für die Organisation von API-Routen verantwortlich ist. Der @lru_cache-Dekorator wird auf Abhängigkeitsinjektionsfunktionen (get_llm_client, get_response_handler, get_cache_manager und get_prompt_manager) angewendet, um sicherzustellen, dass Instanzen von LLMClient, CacheManager und PromptManager für eine bessere Leistung zwischengespeichert werden. Die modere_content-Funktion, dekoriert mit @router.post, definiert eine POST-Route für die Inhaltsmoderation und nutzt den Depends-Mechanismus von FastAPI, um diese Abhängigkeiten einzufügen. Innerhalb der Funktion erzwingt die RateLimiter-Klasse, die mit Ratenbegrenzungseinstellungen aus den Einstellungen konfiguriert ist, Anforderungsbegrenzungen.
Lassen Sie uns abschließend Ihre main.py aktualisieren, um alles zusammenzuführen:
@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) async def complete(self, prompt: str, temperature: float = 0.7, max_tokens: Optional[int] = None) -> Dict: """Get completion from the model with automatic retries""" try: inputs = self.tokenizer(prompt, return_tensors="pt").to( self.model.device ) with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=max_tokens or 100, temperature=temperature, do_sample=True ) response_text = self.tokenizer.decode( outputs[0], skip_special_tokens=True ) # Calculate token usage for monitoring input_tokens = len(inputs.input_ids[0]) output_tokens = len(outputs[0]) - input_tokens return { 'content': response_text, 'usage': { 'prompt_tokens': input_tokens, 'completion_tokens': output_tokens, 'total_tokens': input_tokens + output_tokens }, 'model': "gpt2" } except Exception as e: self.logger.error(f"Error in LLM completion: {str(e)}") raise
Im obigen Code haben wir eine FastAPI-App und den Router mit api.routes unter dem Präfix /api/v1 erstellt. Protokollierung aktiviert, um Informationsmeldungen mit Zeitstempeln anzuzeigen. Die App wird „localhost:8000“ mit Uvicorn und aktiviertem Hot-Reloading ausführen.
Da nun alle Komponenten vorhanden sind, können wir damit beginnen, Ihre Anwendung zum Laufen zu bringen. Erstellen Sie zunächst eine .env-Datei in Ihrem Projektstammverzeichnis und fügen Sie HUGGINGFACE_API_KEY und REDIS_URL hinzu:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
Stellen Sie dann sicher, dass Redis auf Ihrem Computer ausgeführt wird. Auf den meisten Unix-basierten Systemen können Sie es mit dem Befehl:
starten
transformers==4.36.0 huggingface-hub==0.19.4 redis==4.6.0 pydantic==2.5.0 pydantic-settings==2.1.0 tenacity==8.2.3 python-dotenv==1.0.0 fastapi==0.104.1 uvicorn==0.24.0 torch==2.1.0 numpy==1.24.3
Jetzt können Sie mit Ihrer Bewerbung beginnen:
pip install -r requirements.txt
Ihr FastAPI-Server wird unter http://localhost:8000 ausgeführt. Die automatische API-Dokumentation wird unter http://localhost:8000/docs verfügbar sein – das ist super hilfreich zum Testen Ihrer Endpunkte!
Lassen Sie uns Ihre neu erstellte API mit einer echten Anfrage testen. Öffnen Sie ein neues Terminal und führen Sie diesen Curl-Befehl aus:
llm_integration/ ├── core/ │ ├── llm_client.py # your main LLM interaction code │ ├── prompt_manager.py # Handles prompt templates │ └── response_handler.py # Processes LLM responses ├── cache/ │ └── redis_manager.py # Manages your caching system ├── config/ │ └── settings.py # Configuration management ├── api/ │ └── routes.py # API endpoints ├── utils/ │ ├── monitoring.py # Usage tracking │ └── rate_limiter.py # Rate limiting logic ├── requirements.txt └── main.py └── usage_logs.json
Auf Ihrem Terminal sollte eine Antwort wie diese angezeigt werden:
Jetzt fügen wir einige Überwachungsfunktionen hinzu, um zu verfolgen, wie Ihre Anwendung funktioniert und wie viele Ressourcen verwendet werden. Fügen Sie den folgenden Code zu Ihrer Datei utils/monitoring.py hinzu:
import torch from transformers import AutoModelForCausalLM, AutoTokenizer from tenacity import retry, stop_after_attempt, wait_exponential from typing import Dict, Optional import logging class LLMClient: def __init__(self, model_name: str = "gpt2", timeout: int = 30): try: self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained( model_name, device_map="auto", torch_dtype=torch.float16 ) except Exception as e: logging.error(f"Error loading model: {str(e)}") # Fallback to a simpler model if the specified one fails self.tokenizer = AutoTokenizer.from_pretrained("gpt2") self.model = AutoModelForCausalLM.from_pretrained("gpt2") self.timeout = timeout self.logger = logging.getLogger(__name__)
Die UsageMonitor-Klasse führt die folgenden Vorgänge aus:
Fügen Sie als Nächstes eine neue Methode zur Berechnung der Nutzungsstatistiken hinzu:
@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) async def complete(self, prompt: str, temperature: float = 0.7, max_tokens: Optional[int] = None) -> Dict: """Get completion from the model with automatic retries""" try: inputs = self.tokenizer(prompt, return_tensors="pt").to( self.model.device ) with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=max_tokens or 100, temperature=temperature, do_sample=True ) response_text = self.tokenizer.decode( outputs[0], skip_special_tokens=True ) # Calculate token usage for monitoring input_tokens = len(inputs.input_ids[0]) output_tokens = len(outputs[0]) - input_tokens return { 'content': response_text, 'usage': { 'prompt_tokens': input_tokens, 'completion_tokens': output_tokens, 'total_tokens': input_tokens + output_tokens }, 'model': "gpt2" } except Exception as e: self.logger.error(f"Error in LLM completion: {str(e)}") raise
Aktualisieren Sie Ihre API, um die Überwachungsfunktionen aus der UsageMonitor-Klasse hinzuzufügen:
from typing import Dict import logging class ResponseHandler: def __init__(self): self.logger = logging.getLogger(__name__) def parse_moderation_response(self, raw_response: str) -> Dict: """Parse and structure the raw LLM response for moderation""" try: # Default response structure structured_response = { "is_appropriate": True, "confidence_score": 0.0, "reason": None } # Simple keyword-based analysis lower_response = raw_response.lower() # Check for inappropriate content signals if any(word in lower_response for word in ['inappropriate', 'unsafe', 'offensive', 'harmful']): structured_response["is_appropriate"] = False structured_response["confidence_score"] = 0.9 # Extract reason if present if "because" in lower_response: reason_start = lower_response.find("because") structured_response["reason"] = raw_response[reason_start:].split('.')[0].strip() else: structured_response["confidence_score"] = 0.95 return structured_response except Exception as e: self.logger.error(f"Error parsing response: {str(e)}") return { "is_appropriate": True, "confidence_score": 0.5, "reason": "Failed to parse response" } def format_response(self, raw_response: Dict) -> Dict: """Format the final response with parsed content and usage stats""" try: return { "content": self.parse_moderation_response(raw_response["content"]), "usage": raw_response["usage"], "model": raw_response["model"] } except Exception as e: self.logger.error(f"Error formatting response: {str(e)}") raise
Testen Sie jetzt Ihren /stats-Endpunkt, indem Sie diesen Curl-Befehl ausführen:
import redis from typing import Optional, Any import json import hashlib class CacheManager: def __init__(self, redis_url: str, ttl: int = 3600): self.redis = redis.from_url(redis_url) self.ttl = ttl def _generate_key(self, prompt: str, params: dict) -> str: """Generate a unique cache key""" cache_data = { 'prompt': prompt, 'params': params } serialized = json.dumps(cache_data, sort_keys=True) return hashlib.sha256(serialized.encode()).hexdigest() async def get_cached_response(self, prompt: str, params: dict) -> Optional[dict]: """Retrieve cached LLM response""" key = self._generate_key(prompt, params) cached = self.redis.get(key) return json.loads(cached) if cached else None async def cache_response(self, prompt: str, params: dict, response: dict) -> None: """Cache LLM response""" key = self._generate_key(prompt, params) self.redis.setex( key, self.ttl, json.dumps(response) )
Der obige Befehl zeigt Ihnen die Statistiken Ihrer Anfragen auf dem /moderate-Endpunkt an, wie im Screenshot unten gezeigt:
In diesem Tutorial haben Sie gelernt, wie Sie ein großes Sprachmodell in Produktionsanwendungen verwenden. Sie haben Funktionen wie API-Clients, Caching, Prompt-Management und Fehlerbehandlung implementiert. Als Beispiel für diese Konzepte haben Sie ein Content-Moderationssystem entwickelt.
Da Sie nun über eine solide Grundlage verfügen, können Sie Ihr System erweitern mit:
Bitte denken Sie daran, dass Sie in den Beispielen das ChatGPT2-Modell verwendet haben, Sie können dieses System jedoch so anpassen, dass es mit jedem LLM-Anbieter funktioniert. Wählen Sie also das Modell, das Ihren Anforderungen entspricht und innerhalb Ihres Budgets liegt.
Bitte zögern Sie nicht, mich zu kontaktieren, wenn Sie Fragen haben oder mir sagen möchten, was Sie mit diesem System bauen.
Viel Spaß beim Codieren! ?
Das obige ist der detaillierte Inhalt vonIntegration großer Sprachmodelle in Produktionsanwendungen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!