Intégration de grands modèles de langage dans les applications de production-Tutoriel Python-php.cn

Dans ce guide pratique, vous apprendrez à créer une solution de déploiement de modèles hautement évolutive avec des LLM intégrés pour vos applications.
Dans vos exemples, nous utiliserons le modèle ChatGPT2 de Hugging Face, mais vous pouvez facilement brancher n'importe quel autre modèle dont ChatGPT4, Claude, etc.
Que vous conceviez une nouvelle application dotée de capacités d'IA ou que vous amélioriez les systèmes d'IA existants, ce guide vous aidera étape par étape à créer une solide intégration LLM.

Comprendre les principes fondamentaux de l'intégration LLM

Avant de commencer à écrire du code, voyons ce qu'il faut pour créer une intégration LLM de production. Les appels d'API ne sont pas la seule chose que vous devez prendre en compte lors de la création d'une intégration LLM prête pour la production, vous devez également prendre en compte des éléments tels que la fiabilité, le coût et la stabilité. Vos applications de production doivent résoudre des problèmes tels que les interruptions de service, les limites de débit et la variabilité des temps de réponse, tout en maîtrisant les coûts.
Voici ce que nous allons construire ensemble :

Un client API robuste qui gère les échecs avec élégance
Un système de mise en cache intelligent pour optimiser les coûts et la rapidité
Un système de gestion des invites approprié
Gestion et surveillance complètes des erreurs
Un système complet de modération de contenu comme exemple de projet

Conditions préalables

Avant de commencer à coder, assurez-vous d'avoir :

Python 3.8 ou plus récent installé sur votre machine
Compte cloud Redis ou installé localement
Connaissances de base en programmation Python
Compréhension de base des API REST
Une clé API Hugging Face (ou toute autre clé de fournisseur LLM)

Vous voulez suivre ? Le code complet est disponible dans votre référentiel GitHub.

Configuration de votre environnement de développement

Commençons par préparer votre environnement de développement. Nous créerons une structure de projet propre et installerons tous les packages nécessaires.

Tout d'abord, créons votre répertoire de projet et configurons un environnement virtuel Python. Ouvrez votre terminal et exécutez :

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

Copier après la connexion

Maintenant, configurons les dépendances de votre projet. Créez un nouveau fichier exigences.txt avec ces packages essentiels :

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

Copier après la connexion

Décomposons pourquoi nous avons besoin de chacun de ces packages :

transformateurs : il s'agit de la puissante bibliothèque de Hugging Face que nous utiliserons pour s'interfacer avec le modèle Qwen2.5-Coder.
huggingface-hub : nous permet de gérer le chargement du modèle et la gestion des versions. redis : pour implémenter la mise en cache des requêtes
pydantic : utilisé pour la validation des données et les paramètres.
ténacité : responsable de votre fonctionnalité de nouvelle tentative pour une fiabilité accrue
python-dotenv : Pour charger les variables d'environnement
fastapi : construit vos points de terminaison d'API avec une petite quantité de code
uvicorn : Utilisé pour exécuter votre application FastAPI avec une grande efficacité
torch : Pour exécuter des modèles de transformateurs et gérer les opérations d'apprentissage automatique
numpy : utilisé pour le calcul numérique.

Installez tous les packages avec la commande :

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

Copier après la connexion

Organisons votre projet avec une structure épurée. Créez ces répertoires et fichiers dans le répertoire de votre projet :

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

Copier après la connexion

Construire le client LLM

Commençons par votre client LLM qui est l'élément le plus important de votre candidature. C'est ici que nous interagirons avec le modèle ChatGPT (ou tout autre LLM que vous préférez). Ajoutez les extraits de code suivants à votre fichier core/llm_client.py :

pip install -r requirements.txt

Copier après la connexion

Dans cette première partie de votre cours LLMClient, nous mettons en place les bases :

Nous utilisons AutoModelForCausalLM et AutoTokenizer de la bibliothèque des transformateurs pour charger votre modèle
Le paramètre device_map="auto" gère automatiquement l'allocation GPU/CPU
Nous utilisons torch.float16 pour optimiser l'utilisation de la mémoire tout en conservant de bonnes performances

Ajoutons maintenant la méthode qui parle à votre modèle :

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

Copier après la connexion

Décomposons ce qui se passe dans cette méthode de complétion :

Ajout de la méthode @retry decorator pour gérer les échecs temporaires.
Utilisation du gestionnaire de contexte torch.no_grad() pour économiser de la mémoire en désactivant les calculs de dégradé.
Suivi de l'utilisation des jetons en entrée et en sortie, ce qui est très important pour le calcul des coûts.
Renvoie un dictionnaire structuré avec les statistiques de réponse et d'utilisation.

Création de votre gestionnaire de réponses LLM

Ensuite, nous devons ajouter le gestionnaire de réponses pour analyser et structurer la sortie brute du LLM. Faites-le dans votre fichier core/response_handler.py avec les extraits de code suivants :

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__)

Copier après la connexion

Ajout d'un système de mise en cache robuste

Créons maintenant votre système de mise en cache pour améliorer les performances des applications et réduire les coûts. Ajoutez les extraits de code suivants à votre fichier cache/redis_manager.py :

 @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

Copier après la connexion

Dans les extraits de code ci-dessus, nous avons créé une classe CacheManager qui gère toutes les opérations de mise en cache avec les éléments suivants :

La méthode _generate_key, qui crée des clés de cache uniques basées sur des invites et des paramètres
get_cached_response qui vérifie si nous avons une réponse en cache pour une invite donnée
cache_response qui stocke les réponses réussies pour une utilisation future

Création d'un gestionnaire d'invites intelligent

Créons votre gestionnaire d'invites qui gérera les invites de votre modèle LLM. Ajoutez le code suivant à votre core/prompt_manager.py :

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

Copier après la connexion

Créez ensuite un exemple de modèle d'invite pour la modération du contenu dans votre fichier prompts/content_moderation.json avec des extraits de code :

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

Copier après la connexion

Votre gestionnaire d'invites pourra désormais charger des modèles d'invites à partir de votre fichier JSON et obtenir également un modèle d'invite formaté.

Configuration d'un gestionnaire de configuration

Pour conserver toutes vos configurations LLM au même endroit et les réutiliser facilement dans votre application, créons des paramètres de configuration. Ajoutez le code ci-dessous à votre fichier config/settings.py :

pip install -r requirements.txt

Copier après la connexion

Implémentation de la limitation de débit

Ensuite, implémentons une limitation de débit pour contrôler la manière dont les utilisateurs accèdent aux ressources de votre application. Pour ce faire, ajoutez le code suivant à votre fichier utils/rate_limiter.py :

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

Copier après la connexion

Dans RateLimiter, nous avons implémenté une méthode check_rate_limit réutilisable qui peut être utilisée dans n'importe quel itinéraire pour gérer la limitation de débit en transmettant simplement la période et le nombre de requêtes autorisées pour chaque utilisateur pendant une période donnée.

Création de vos points de terminaison d'API

Créons maintenant vos points de terminaison API dans le fichier api/routes.py pour intégrer votre LLM dans votre application :

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__)

Copier après la connexion

Ici, nous avons défini un point de terminaison /moderate dans la classe APIRouter, qui est responsable de l'organisation des routes API. Le décorateur @lru_cache est appliqué aux fonctions d'injection de dépendances (get_llm_client, get_response_handler, get_cache_manager et get_prompt_manager) pour garantir que les instances de LLMClient, CacheManager et PromptManager sont mises en cache pour de meilleures performances. La fonction moderate_content, décorée avec @router.post, définit une route POST pour la modération du contenu et utilise le mécanisme Depends de FastAPI pour injecter ces dépendances. À l'intérieur de la fonction, la classe RateLimiter, configurée avec les paramètres de limite de débit à partir des paramètres, applique les limites de requêtes.

Enfin, mettons à jour votre main.py pour tout rassembler :

 @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

Copier après la connexion

Dans le code ci-dessus, nous avons créé une application FastAPI et le routeur en utilisant api.routes sous le préfixe /api/v1. Activation de la journalisation pour afficher des messages d'information avec des horodatages. L'application exécutera localhost:8000 en utilisant Uvicorn, avec le rechargement à chaud activé.

Exécuter votre application

Nous avons maintenant tous les composants en place, commençons à rendre votre application opérationnelle. Tout d'abord, créez un fichier .env dans le répertoire racine de votre projet et ajoutez votre HUGGINGFACE_API_KEY et REDIS_URL :

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

Copier après la connexion

Assurez-vous ensuite que Redis est exécuté sur votre machine. Sur la plupart des systèmes basés sur Unix, vous pouvez le démarrer avec la commande :

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

Copier après la connexion

Vous pouvez maintenant démarrer votre candidature :

pip install -r requirements.txt

Copier après la connexion

votre serveur FastAPI commencera à fonctionner sur http://localhost:8000. La documentation automatique de l'API sera disponible sur http://localhost:8000/docs - c'est très utile pour tester vos points de terminaison !

Integrating Large Language Models in Production Applications

Tester votre API de modération de contenu

Testons votre API nouvellement créée avec une vraie requête. Ouvrez un nouveau terminal et exécutez cette commande curl :

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

Copier après la connexion

Vous devriez voir une réponse comme celle-ci sur votre terminal :

Integrating Large Language Models in Production Applications

Ajout de surveillance et d'analyse

Ajoutons maintenant quelques fonctionnalités de surveillance pour suivre les performances de votre application et la quantité de ressources utilisées. Ajoutez le code suivant à votre fichier utils/monitoring.py :

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__)

Copier après la connexion

La classe UsageMonitor effectuera les opérations suivantes :

Suivi de chaque requête API avec horodatages
Enregistrement de l'utilisation du jeton pour le suivi des coûts
Mesurer les délais de réponse
Stocker tout dans un fichier journal structuré (remplacez-le par une base de données avant de déployer votre application en production)

Ensuite, ajoutez une nouvelle méthode pour calculer les statistiques d'utilisation :

 @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

Copier après la connexion

Mettez à jour votre API pour ajouter les fonctionnalités de surveillance de la classe UsageMonitor :

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

Copier après la connexion

Maintenant, testez votre point de terminaison /stats en exécutant cette commande curl :

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)
        )

Copier après la connexion

La commande ci-dessus vous montrera les statistiques de vos requêtes sur le point de terminaison /modéré, comme indiqué dans la capture d'écran ci-dessous :

Integrating Large Language Models in Production Applications

Conclusion

Tout au long de ce tutoriel, vous avez appris à utiliser un grand modèle de langage dans les applications de production. Vous avez implémenté des fonctionnalités telles que les clients API, la mise en cache, la gestion des invites et la gestion des erreurs. À titre d'exemple de ces concepts, vous avez développé un système de modération de contenu.

Maintenant que vous disposez d'une base solide, vous pouvez améliorer votre système avec :

Réponses en streaming pour les candidatures en temps réel
Tests A/B pour une amélioration immédiate
Une interface Web pour gérer les invites
Affinage du modèle personnalisé
Intégration avec des services de surveillance tiers

Veuillez rappeler que dans les exemples, vous avez utilisé le modèle ChatGPT2, mais vous pouvez adapter ce système pour fonctionner avec n'importe quel fournisseur LLM. Choisissez donc le modèle qui répond à vos exigences et qui respecte votre budget.

N'hésitez pas à me contacter si vous avez des questions ou si vous souhaitez me dire ce que vous construisez avec ce système.

Bon codage ! ?

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!