Produktives Deployment und Engineering von AI-Agenten

Der Produktionseinsatz umfasst mehrere Aspekte wie Zuverlässigkeit, Skalierbarkeit und Kostenkontrolle.

Engineering-Praktiken gewährleisten einen stabilen Systembetrieb und einfache Wartung.

Architektur für den Produktionseinsatz

Produktionsumgebungen stellen höhere Anforderungen an das System.

Mehrere Aspekte wie Zuverlässigkeit, Skalierbarkeit, Monitoring und Alarmierung müssen berücksichtigt werden.

Bereitstellungsmodi

Einzelmaschinen-Bereitstellung

Geeignet für Entwicklungs- und Testumgebungen.

Einfach und leicht bereitzustellen, kann aber keinen Produktionsverkehr bewältigen.

Verteilte Bereitstellung

Geeignet für Produktionsumgebungen; mehrere Aspekte müssen berücksichtigt werden.

Lastverteilung: Anfragen auf mehrere Instanzen verteilen.

Service-Discovery: Die Liste der Instanzen dynamisch verwalten.

Zustandsverwaltung: Konsistenzprobleme verteilter Zustände behandeln.

Fehlertoleranz: Ein einzelner Ausfallpunkt beeinträchtigt den Gesamtservice nicht.

Containerisierte Bereitstellung

Dockerfile-Beispiel

# Basierend auf Python-3.11-Image
FROM python:3.11-slim

# Arbeitsverzeichnis festlegen
WORKDIR /app

# Abhängigkeiten installieren
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Anwendungscode kopieren
COPY . .

# Nicht-Root-Benutzer erstellen (Sicherheitsaspekt)
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# Port freigeben
EXPOSE 8000

# Startbefehl
CMD ["python", "agent.py"]

Kubernetes Deployment

Kubernetes Deployment-Konfiguration

span style="color: green;">
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
  labels:
    app: ai-agent
spec:
  # Anzahl der Replikate
  replicas: 3
  selector:
    matchLabels:
      app: ai-agent
  template:
    metadata:
      labels:
        app: ai-agent
    spec:
      containers:
      - name: agent
        image: your-registry/ai-agent:latest
        ports:
        - containerPort: 8000
        # Ressourcenlimits
        Ressourcen:
          limits:
            memory: "2Gi"
            cpu: "1"
          requests:
            memory: "1Gi"
            cpu: "0.5"
        # Health-Check
        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
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer

Kostenkontrolle

Die Kosten eines Agent-Systems stammen hauptsächlich aus LLM-API-Aufrufen.

Die Kostenoptimierung ist ein wichtiges Thema für den Produktiveinsatz.

Token-Optimierungsstrategien

Prompt-Komprimierung: Den Prompt vereinfachen und unnötigen Text reduzieren.

Kontextkürzung: Nur die wesentlichen Kontextinhalte behalten.

Antwort-Caching: Antworten für identische oder ähnliche Anfragen zwischenspeichern.

Code-Implementierung

Implementierung eines Kostenoptimierungs-Agenten

import hashlib
import json
from functools import lru_cache

class CostOptimizedAgent:
    """
Kostenoptimierter Agent
API-Aufrufkosten durch Caching und Komprimierung reduzieren
    """
   
    def __init__(self, base_agent, cache, embedder=Keine):
        # Basis-Agent
        self.base_agent = base_agent
        # Cache-Speicher
        self.cache = cache
        # Embedder (für Ähnlichkeits-Cache-Abgleich)
        self.embedder = embedder
   
    def process(self, request):
        """
Anfragen verarbeiten, einschließlich Kostenoptimierungslogik
        """
        # Anfrage-Fingerabdruck generieren
        request_fingerprint = self.generate_fingerprint(request)
       
        # Exakten Cache prüfen
        cached_result = self.cache.get(request_fingerprint)
        if cached_result:
            return {
                **cached_result,
                "from_cache": True
            }
       
        # Wenn ein Embedder vorhanden ist, semantischen Ähnlichkeits-Cache prüfen
        if self.embedder:
            similar = self.find_similar_cached(request)
            if similar:
                return {
                    **similar,
                    "from_cache": True,
                    "similar_to": similar.get("request_id")
                }
       
        # Anfrage ausführen
        result = self.base_agent.process(request)
       
        # Ergebnis cachen
        self.cache.set(
            request_fingerprint,
            result,
            ttl=3600  # Läuft in 1 Stunde ab
        )
       
        return {
            **result,
            "from_cache": False
        }
   
    def generate_fingerprint(self, request):
        """
Anfrage-Fingerabdruck generieren
Für exakten Cache-Abgleich
        """
        content = json.dumps(request, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
   
    def find_similar_cached(self, request, threshold=0.95):
        """
Semantisch ähnliche zwischengespeicherte Ergebnisse finden
Unter Verwendung der Kosinusähnlichkeit von Embedding-Vektoren
        """
        if not self.embedder:
            return None
       
        # Anfrage vektorisieren
        request_vector = self.embedder.embed([str(request)])[0]
       
        # Im Cache nach ähnlichen Anfragen suchen
        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:
    """
Antwort-Cache
Einfache In-Memory-Cache-Implementierung
Redis kann in der Produktion verwendet werden
    """
   
    def __init__(self):
        self.cache = {}
        self.timestamps = {}
   
    def get(self, key):
        """Cache abrufen"""
        if key in self.cache:
            # Prüfen, ob abgelaufen
            if self.is_expired(key):
                self.delete(key)
                return None
            return self.cache[key]
        return None
   
    def set(self, key, value, ttl=3600):
        """Cache setzen"""
        self.cache[key] = value
        self.timestamps[key] = time.time() + ttl
   
    def delete(self, key):
        """Cache löschen"""
        if key in self.cache:
            del self.cache[key]
        if key in self.timestamps:
            del self.timestamps[key]
   
    def is_expired(self, key):
        """Prüfen, ob abgelaufen"""
        if key not in self.timestamps:
            return True
        return time.time() > self.timestamps[key]
   
    def get_all(self):
        """Alle Cache-Einträge abrufen"""
        return list(self.cache.values())


def cosine_similarity(vec1, vec2):
    """Kosinusähnlichkeit berechnen"""
    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 und asynchron

Streaming-Antwort

Streaming-Antworten reduzieren die Wartezeit und verbessern die Benutzererfahrung.

Sie müssen nicht auf die vollständige Antwort warten; Sie können generierten Inhalt schrittweise anzeigen.

Code-Implementierung

Streaming-Antwort-Agent

import asyncio

class StreamingAgent:
    """
Agent, der Streaming-Antworten unterstützt
Während der Generierung zurückgeben, wodurch die Wartezeit reduziert wird
    """
   
    def __init__(self, llm):
        self.llm = llm
   
    async def stream_generate(self, prompt):
        """
Generierung von Streaming-Antworten
Async-Generator-Muster verwenden
:yield: Das generierte Textfragment
        """
        # Asynchrone Generierungsaufgabe starten
        async for chunk in self.llm.stream_generate(prompt):
            yield chunk
   
    async def process_stream(self, request):
        """
Streaming-Anfragen verarbeiten
Async-Generator zurückgeben
        """
        prompt = self.build_prompt(request)
       
        # Generator zurückgeben
        async def generate():
            async for chunk in self.stream_generate(prompt):
                yield chunk
       
        return generate()


# Verwendungsbeispiel
async def main():
    agent = StreamingAgent(llm)
   
    # Streaming-Generierung starten
    async for token in agent.stream_generate("Quantencomputing erklären"):
        # Während der Generierung ausgeben
        print(token, end="", flush=True)
    print()  # Zeilenumbruch


# ausführen
asyncio.run(main())

Asynchrone Aufgabenwarteschlange

Für lang andauernde Aufgaben verwenden Sie asynchrone Warteschlangenverarbeitung.

Geben Sie sofort zurück, nachdem der Benutzer eine Aufgabe übermittelt hat, und führen Sie sie asynchron im Hintergrund aus.

Implementierung einer asynchronen Aufgabenwarteschlange

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

class AsyncAgent:
    """
Asynchroner Agent
Verwenden Sie eine Aufgabenwarteschlange, um lang andauernde Aufgaben zu bearbeiten
    """
   
    def __init__(self, agent, task_queue):
        # Basis-Agent
        self.agent = agent
        # Aufgabenwarteschlange
        self.task_queue = task_queue
        # Ergebnisspeicher
        self.results = {}
        # Hintergrund-Worker-Thread starten
        self.worker_thread = Thread(target=self.process_queue, daemon=True)
        self.worker_thread.start()
   
    async def submit(self, task):
        """
Aufgabe übermitteln
Aufgaben-ID sofort zurückgeben
        """
        # Eine eindeutige Aufgaben-ID generieren
        task_id = str(uuid.uuid4())
       
        # Zur Warteschlange hinzufügen
        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):
        """
Aufgabenergebnisse abrufen
nicht blockierend
        """
        if task_id in self.results:
            return self.results[task_id]
       
        # Aufgabenstatus prüfen
        status = await self.task_queue.get_status(task_id)
       
        if status is None:
            return None  # Aufgabe existiert nicht
       
        if status == "pending" or status == "processing":
            return {
                "status": status,
                "result": None
            }
       
        return None
   
    def process_queue(self):
        """
Hintergrund-Worker-Thread
Eine Aufgabe aus der Warteschlange entnehmen und ausführen
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
       
        while True:
            task = loop.run_until_complete(self.task_queue.dequeue())
           
            if task:
                # Status auf Verarbeitung aktualisieren
                loop.run_until_complete(
                    self.task_queue.update_status(task["id"], "processing")
                )
               
                try:
                    # Aufgabe ausführen
                    result = self.agent.process(task["task"])
                   
                    # Ergebnis speichern
                    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:
    """Einfache Implementierung einer Aufgabenwarteschlange"""
   
    def __init__(self):
        self.queue = Queue()
        self.statuses = {}
   
    async def enqueue(self, task):
        """Aufgabe zur Warteschlange hinzufügen"""
        self.queue.put(task)
        self.statuses[task["id"]] = "pending"
   
    async def dequeue(self):
        Eine Aufgabe abrufen
        if not self.queue.empty():
            return self.queue.get()
        return None
   
    async def get_status(self, task_id):
        Aufgabenstatus abrufen
        return self.statuses.get(task_id)
   
    async def update_status(self, task_id, status):
        Aufgabenstatus aktualisieren
        self.statuses[task_id] = status

Microservices-Design

Teilen Sie das Agent-System in mehrere unabhängige Dienste auf, um Wartbarkeit und Skalierbarkeit zu verbessern.

Service-Partitionierung

Dienste Verantwortlichkeiten
API Gateway Einheitlicher Einstiegspunkt, Authentifizierung, Ratenbegrenzung, Routenweiterleitung.
Agent-Dienst Kernlogik, Schlussfolgerung und Planung.
Tool-Dienst Externe API-Integration, Tool-Aufruf.
Speicherdienst Wissensdatenbank, Zustandsverwaltung, Caching.
Monitoring-Dienst Protokollsammlung, Metriküberwachung, Alarmierung.

Dienstübergreifende Kommunikation

synchrone Kommunikation: HTTP/gRPC, verwendet für Echtzeit-Anfrage-Antwort.

asynchrone Kommunikation: Nachrichtenwarteschlange, verwendet für zeitaufwändige Operationen und Ereignisbenachrichtigungen.

Kapitelzusammenfassung

Dieses Kapitel stellt wichtige Praktiken für Produktionsbereitstellung und Engineering vor.

ProduktionsbereitstellungsarchitekturVon Standalone zu verteilt, schrittweise skalieren.

Containerisierte BereitstellungDocker und Kubernetes verwenden.

KostenkontrolleKosten durch Caching und Token-Optimierung senken.

Streaming und AsynchronitätBenutzererfahrung und Systemdurchsatz verbessern.

Microservice-orientiertes DesignDie Wartbarkeit und Skalierbarkeit des Systems verbessern.

Die Produktionsbereitstellung erfordert eine umfassende Berücksichtigung mehrerer Aspekte wie Zuverlässigkeit, Leistung und Kosten.

Es wird empfohlen, mit einem kleinen Umfang zu beginnen und die Komplexität schrittweise zu erhöhen.

Weitere Erweiterungen