Déploiement en production et ingénierie des agents IA

Le déploiement en production implique de multiples aspects tels que la fiabilité, l'évolutivité et la maîtrise des coûts.

Les pratiques d'ingénierie garantissent un fonctionnement stable du système et une facilité de maintenance.

Architecture de déploiement en production

Les environnements de production imposent des exigences plus élevées au système.

Plusieurs aspects tels que la fiabilité, l'évolutivité, la surveillance et l'alerte doivent être pris en compte.

Modes de déploiement

Déploiement sur une seule machine

Adapté aux environnements de développement et de test.

Simple et facile à déployer, mais ne peut pas gérer le trafic de niveau production.

Déploiement distribué

Adapté aux environnements de production ; plusieurs aspects doivent être pris en compte.

Équilibrage de charge: Répartir les requêtes entre plusieurs instances.

Découverte de services: Gérer dynamiquement la liste des instances.

Gestion de l’état: Gérer les problèmes de cohérence d’état distribués.

Tolérance aux pannes: Un point de défaillance unique n’affecte pas l’ensemble du service.

Déploiement conteneurisé

Exemple de Dockerfile

# Basé sur l'image Python 3.11
FROM python:3.11-slim

# Définir le répertoire de travail
WORKDIR /app

# Installer les dépendances
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier le code de l'application
COPY . .

# Créer un utilisateur non-root (considération de sécurité)
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# Exposer le port
EXPOSE 8000

# Commande de démarrage
CMD ["python", "agent.py"]

Déploiement Kubernetes

Configuration de déploiement Kubernetes

span style="color: green;">
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
  labels:
    app⟦⟧ai-agent: ai-agent
spec:
  # Nombre de réplicas
  replicas: 3
  selector:
    matchLabels:
      app⟦⟧ai-agent: ai-agent
  template:
    metadata:
      labels:
        app⟦⟧ai-agent: ai-agent
    spec:
      containers:
      - name: agent
        image: your-registry/ai-agent:latest
        ports:
        - containerPort: 8000
        # Limites de ressources
        ressources:
          limits:
            memory: "2Gi"
            cpu: "1"
          requests:
            memory: "1Gi"
            cpu: "0.5"
        # Vérification de l'état
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: ai-agent-service
spec:
  selector:
    app⟦⟧ai-agent: ai-agent
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

Contrôle des coûts

Le coût d'un système Agent provient principalement des appels d'API LLM.

L'optimisation des coûts est un sujet important pour le déploiement en production.

Stratégies d'optimisation des tokens

Compression de prompt: Simplifiez le prompt et réduisez le texte inutile.

Troncature du contexte: Ne conserver que le contenu clé du contexte.

Mise en cache des réponses: Mettre en cache les réponses pour des requêtes identiques ou similaires.

Implémentation du code

Implémentation d'un agent d'optimisation des coûts

import hashlib
import json
from functools import lru_cache

class CostOptimizedAgent:
    """
Agent optimisé pour les coûts
Réduire les coûts d'appels API grâce à la mise en cache et à la compression
    """
   
    def __init__(self, base_agent, cache, embedder=Aucune):
        # Agent de base
        self.base_agent = base_agent
        # Stockage du cache
        self.cache = cache
        # Embedder (pour la correspondance de cache par similarité)
        self.embedder = embedder
   
    def process(self, request):
        """
Gérer les requêtes, y compris la logique d'optimisation des coûts
        """
        # Générer l'empreinte de la requête
        request_fingerprint = self.generate_fingerprint(request)
       
        # Vérifier le cache exact
        cached_result = self.cache.get(request_fingerprint)
        if cached_result:
            return {
                **cached_result,
                "from_cache": True
            }
       
        # S'il y a un embedder, vérifier le cache de similarité sémantique
        if self.embedder:
            similar = self.find_similar_cached(request)
            if similar:
                return {
                    **similar,
                    "from_cache": True,
                    "similar_to": similar.get("request_id")
                }
       
        # Exécuter la requête
        result = self.base_agent.process(request)
       
        # Mettre le résultat en cache
        self.cache.set(
            request_fingerprint,
            result,
            ttl=3600  # Expire dans 1 heure
        )
       
        return {
            **result,
            "from_cache": False
        }
   
    def generate_fingerprint(self, request):
        """
Générer une empreinte de requête
Pour une correspondance exacte du cache
        """
        content = json.dumps(request, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
   
    def find_similar_cached(self, request, threshold=0.95):
        """
Trouver des résultats en cache sémantiquement similaires
En utilisant la similarité cosinus des vecteurs d'embedding
        """
        if not self.embedder:
            return None
       
        # Vectoriser la requête
        request_vector = self.embedder.embed([str(request)])[0]
       
        # Rechercher des requêtes similaires dans le cache
        best_match = None
        best_similarity = 0
       
        for cached in self.cache.get_all():
            cached_vector = cached["vector"]
            similarity = cosine_similarity(request_vector, cached_vector)
           
            if similarity > best_similarity and similarity >= threshold:
                best_similarity = similarity
                best_match = cached
       
        return best_match


class ResponseCache:
    """
Cache de réponses
Implémentation simple de cache en mémoire
Redis peut être utilisé en production
    """
   
    def __init__(self):
        self.cache = {}
        self.timestamps = {}
   
    def get(self, key):
        """Obtenir le cache"""
        if key in self.cache:
            # Vérifier si expiré
            if self.is_expired(key):
                self.delete(key)
                return None
            return self.cache[key]
        return None
   
    def set(self, key, value, ttl=3600):
        """Définir le cache"""
        self.cache[key] = value
        self.timestamps[key] = time.time() + ttl
   
    def delete(self, key):
        """Supprimer le cache"""
        if key in self.cache:
            del self.cache[key]
        if key in self.timestamps:
            del self.timestamps[key]
   
    def is_expired(self, key):
        """Vérifier si expiré"""
        if key not in self.timestamps:
            return True
        return time.time() > self.timestamps[key]
   
    def get_all(self):
        """Récupérer tous les éléments du cache"""
        return list(self.cache.values())


def cosine_similarity(vec1, vec2):
    """Calculer la similarité cosinus"""
    dot_product = sum(a * b for a, b in zip(vec1, vec2))
    norm1 = math.sqrt(sum(a * a for a in vec1))
    norm2 = math.sqrt(sum(b * b for b in vec2))
    return dot_product / (norm1 * norm2)

Streaming et asynchrone

Réponse en streaming

Les réponses en streaming réduisent le temps d'attente et améliorent l'expérience utilisateur.

Vous n'avez pas besoin d'attendre la réponse complète ; vous pouvez afficher le contenu généré progressivement.

Implémentation du code

Agent de réponse en streaming

import asyncio

class StreamingAgent:
    """
Agent qui prend en charge les réponses en streaming
Retourner pendant la génération, ce qui réduit le temps d'attente
    """
   
    def __init__(self, llm):
        self.llm = llm
   
    async def stream_generate(self, prompt):
        """
Génération de réponse en streaming
Utiliser le modèle de générateur asynchrone
:yield: Le fragment de texte généré
        """
        # Démarrer la tâche de génération asynchrone
        async for chunk in self.llm.stream_generate(prompt):
            yield chunk
   
    async def process_stream(self, request):
        """
Gérer les requêtes de streaming
Retourner un générateur asynchrone
        """
        prompt = self.build_prompt(request)
       
        # Retourner le générateur
        async def generate():
            async for chunk in self.stream_generate(prompt):
                yield chunk
       
        return generate()


# Exemple d'utilisation
async def main():
    agent = StreamingAgent(llm)
   
    # Démarrer la génération en streaming
    async for token in agent.stream_generate("Explain quantum computing"):
        # Afficher pendant la génération
        print(token, end="", flush=True)
    print()  # saut de ligne


# exécuter
asyncio.run(main())

File de tâches asynchrone

Pour les tâches de longue durée, utilisez le traitement par file d'attente asynchrone.

Retournez immédiatement après que l'utilisateur a soumis une tâche, et exécutez de manière asynchrone en arrière-plan.

Implémentation d'une file d'attente de tâches asynchrone

import asyncio
from queue import Queue
from threading import Thread
import uuid

class AsyncAgent:
    """
Agent asynchrone
Utilisez une file d'attente de tâches pour gérer les tâches de longue durée
    """
   
    def __init__(self, agent, task_queue):
        # Agent de base
        self.agent = agent
        # File d'attente de tâches
        self.task_queue = task_queue
        # Stockage des résultats
        self.results = {}
        # Démarrer le thread de travail en arrière-plan
        self.worker_thread = Thread(target=self.process_queue, daemon=True)
        self.worker_thread.start()
   
    async def submit(self, task):
        """
soumettre la tâche
Retourner immédiatement l'ID de la tâche
        """
        # Générer un ID de tâche unique
        task_id = str(uuid.uuid4())
       
        # Ajouter à la file d'attente
        await self.task_queue.enqueue({
            "id": task_id,
            "task": task,
            "status": "pending",
            "created_at": time.time()
        })
       
        return task_id
   
    async def get_result(self, task_id):
        """
Obtenir les résultats de la tâche
non-bloquant
        """
        if task_id in self.results:
            return self.results[task_id]
       
        # Vérifier l'état de la tâche
        status = await self.task_queue.get_status(task_id)
       
        if status is None:
            return None  # La tâche n'existe pas
       
        if status == "pending" or status == "processing":
            return {
                "status": status,
                "result": None
            }
       
        return None
   
    def process_queue(self):
        """
Thread de travail en arrière-plan
Prendre une tâche de la file d'attente et l'exécuter
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
       
        while True:
            task = loop.run_until_complete(self.task_queue.dequeue())
           
            if task:
                # Mettre à jour le statut vers processing
                loop.run_until_complete(
                    self.task_queue.update_status(task["id"], "processing")
                )
               
                try:
                    # Exécuter la tâche
                    result = self.agent.process(task["task"])
                   
                    # Enregistrer le résultat
                    self.results[task["id"]] = {
                        "status": "completed",
                        "result": result,
                        "completed_at": time.time()
                    }
                   
                    loop.run_until_complete(
                        self.task_queue.update_status(task["id"], "completed")
                    )
                except Exception as e:
                    self.results[task["id"]] = {
                        "status": "failed",
                        "error": str(e)
                    }
                   
                    loop.run_until_complete(
                        self.task_queue.update_status(task["id"], "failed")
                    )


class TaskQueue:
    """Implémentation simple d'une file d'attente de tâches"""
   
    def __init__(self):
        self.queue = Queue()
        self.statuses = {}
   
    async def enqueue(self, task):
        """Ajouter une tâche à la file d'attente"""
        self.queue.put(task)
        self.statuses[task["id"]] = "pending"
   
    async def dequeue(self):
        Récupérer une tâche
        if not self.queue.empty():
            return self.queue.get()
        return None
   
    async def get_status(self, task_id):
        Obtenir le statut de la tâche
        return self.statuses.get(task_id)
   
    async def update_status(self, task_id, status):
        Mettre à jour le statut de la tâche
        self.statuses[task_id] = status

Conception de microservices

Diviser le système Agent en plusieurs services indépendants pour améliorer la maintenabilité et la scalabilité.

Partitionnement des services

Services Responsabilités
Passerelle API Point d'entrée unifié, authentification et limitation de débit, transfert de routes.
Service d'agent Logique centrale, raisonnement et planification.
service d'outils Intégration d'API externes, invocation d'outils.
service de stockage Base de connaissances, gestion d'état, mise en cache.
service de supervision Collecte de journaux, surveillance des métriques, alertes.

Communication inter-services

communication synchrone: HTTP/gRPC, utilisée pour les requêtes-réponses en temps réel.

communication asynchrone: file d'attente de messages, utilisée pour les opérations longues et les notifications d'événements.

Résumé du chapitre

Ce chapitre présente les pratiques clés pour le déploiement en production et l'ingénierie.

Architecture de déploiement en productionDu mode autonome au mode distribué, montez en charge progressivement.

Déploiement conteneuriséUtilisez Docker et Kubernetes.

Maîtrise des coûtsRéduisez les coûts grâce à la mise en cache et à l'optimisation des tokens.

Diffusion en continu et asynchroneAméliorez l'expérience utilisateur et le débit du système.

Conception orientée microservicesAméliorez la maintenabilité et l'évolutivité du système.

Le déploiement en production exige une prise en compte globale de multiples aspects tels que la fiabilité, les performances et le coût.

Il est recommandé de commencer à petite échelle et d'augmenter progressivement la complexité.

Autres extensions