Развёртывание и инженерия AI-агентов в продакшене

Развёртывание в продакшене включает несколько аспектов, таких как надёжность, масштабируемость и контроль затрат.

Инженерные практики обеспечивают стабильную работу системы и простоту сопровождения.

Архитектура развёртывания в продакшене

Продакшен-среды предъявляют повышенные требования к системе.

Необходимо учитывать несколько аспектов, таких как надёжность, масштабируемость, мониторинг и оповещения.

Режимы развёртывания

Развёртывание на одной машине

Подходит для сред разработки и тестирования.

Просто и легко развернуть, но не справляется с трафиком уровня продакшена.

Распределённое развёртывание

Подходит для продакшен-сред; необходимо учитывать множество аспектов.

Балансировка нагрузки: Распределение запросов между несколькими экземплярами.

Обнаружение сервисов: Динамическое управление списком экземпляров.

Управление состоянием: Обработка проблем согласованности распределённого состояния.

Отказоустойчивость: Единая точка отказа не влияет на работу сервиса в целом.

Контейнеризованное развёртывание

Пример Dockerfile

# На основе образа Python 3.11
FROM python:3.11-slim

# Задать рабочий каталог
WORKDIR /app

# Установить зависимости
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Скопировать код приложения
COPY . .

# Создать пользователя без прав root (в целях безопасности)
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

# Открыть порт
EXPOSE 8000

# Команда запуска
CMD ["python", "agent.py"]

Развёртывание Kubernetes

Конфигурация развёртывания Kubernetes

span style="color: green;">
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
  labels:
    app: ai-agent
spec:
  # Количество реплик
  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
        # Лимиты ресурсов
        ресурсы:
          limits:
            memory: "2Gi"
            cpu: "1"
          requests:
            memory: "1Gi"
            cpu: "0.5"
        # Проверка работоспособности
        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

Контроль затрат

Затраты на агентную систему в основном связаны с вызовами LLM API.

Оптимизация затрат — важная тема для развёртывания в продакшене.

Стратегии оптимизации токенов

Сжатие промпта: упростите промпт и сократите ненужный текст.

Усечение контекста: сохранять только ключевое содержимое контекста.

Кэширование ответов: кэшировать ответы для идентичных или похожих запросов.

Реализация кода

Реализация агента оптимизации затрат

import hashlib
import json
from functools import lru_cache

class CostOptimizedAgent:
    """
Агент с оптимизацией затрат
Снижайте затраты на вызовы API с помощью кэширования и сжатия
    """
   
    def __init__(self, base_agent, cache, embedder=Отсутствует):
        # Базовый агент
        self.base_agent = base_agent
        # Хранилище кэша
        self.cache = cache
        # Embedder (для сопоставления кэша по сходству)
        self.embedder = embedder
   
    def process(self, request):
        """
Обработка запросов, включая логику оптимизации затрат
        """
        # Сгенерировать отпечаток запроса
        request_fingerprint = self.generate_fingerprint(request)
       
        # Проверить точный кэш
        cached_result = self.cache.get(request_fingerprint)
        if cached_result:
            return {
                **cached_result,
                "from_cache": True
            }
       
        # Если есть embedder, проверить кэш семантического сходства
        if self.embedder:
            similar = self.find_similar_cached(request)
            if similar:
                return {
                    **similar,
                    "from_cache": True,
                    "similar_to": similar.get("request_id")
                }
       
        # Выполнить запрос
        result = self.base_agent.process(request)
       
        # Кэшировать результат
        self.cache.set(
            request_fingerprint,
            result,
            ttl=3600  # Истекает через 1 час
        )
       
        return {
            **result,
            "from_cache": False
        }
   
    def generate_fingerprint(self, request):
        """
Сгенерировать отпечаток запроса
Для точного сопоставления с кэшем
        """
        content = json.dumps(request, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
   
    def find_similar_cached(self, request, threshold=0.95):
        """
Найти семантически похожие кэшированные результаты
Использование косинусного сходства векторов эмбеддингов
        """
        if not self.embedder:
            return None
       
        # Векторизовать запрос
        request_vector = self.embedder.embed([str(request)])[0]
       
        # Поиск похожих запросов в кэше
        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:
    """
Кэш ответов
Простая реализация кэша в памяти
Redis можно использовать в продакшене
    """
   
    def __init__(self):
        self.cache = {}
        self.timestamps = {}
   
    def get(self, key):
        """Получить кэш"""
        if key in self.cache:
            # Проверить, истёк ли срок действия
            if self.is_expired(key):
                self.delete(key)
                return None
            return self.cache[key]
        return None
   
    def set(self, key, value, ttl=3600):
        """Установить кэш"""
        self.cache[key] = value
        self.timestamps[key] = time.time() + ttl
   
    def delete(self, key):
        """Удалить кэш"""
        if key in self.cache:
            del self.cache[key]
        if key in self.timestamps:
            del self.timestamps[key]
   
    def is_expired(self, key):
        """Проверить, истёк ли срок"""
        if key not in self.timestamps:
            return True
        return time.time() > self.timestamps[key]
   
    def get_all(self):
        """Получить все элементы кэша"""
        return list(self.cache.values())


def cosine_similarity(vec1, vec2):
    """Вычислить косинусное сходство"""
    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)

Потоковая передача и асинхронность

Потоковый ответ

Потоковые ответы сокращают время ожидания и улучшают пользовательский опыт.

Не нужно ждать полного ответа; можно постепенно отображать сгенерированный контент.

Реализация кода

Агент потоковых ответов

import asyncio

class StreamingAgent:
    """
Агент, поддерживающий потоковые ответы
Возврат во время генерации, сокращая время ожидания
    """
   
    def __init__(self, llm):
        self.llm = llm
   
    async def stream_generate(self, prompt):
        """
Генерация потокового ответа
Используйте шаблон асинхронного генератора
:yield: Сгенерированный фрагмент текста
        """
        # Запуск задачи асинхронной генерации
        async for chunk in self.llm.stream_generate(prompt):
            yield chunk
   
    async def process_stream(self, request):
        """
Обрабатывает потоковые запросы
Возвращает асинхронный генератор
        """
        prompt = self.build_prompt(request)
       
        # Возврат генератора
        async def generate():
            async for chunk in self.stream_generate(prompt):
                yield chunk
       
        return generate()


# Пример использования
async def main():
    agent = StreamingAgent(llm)
   
    # Начать потоковую генерацию
    async for token in agent.stream_generate("Объясни квантовые вычисления"):
        # Выводить по мере генерации
        print(token, end="", flush=True)
    print()  # новая строка


# запуск
asyncio.run(main())

Асинхронная очередь задач

Для длительных задач используйте асинхронную обработку очередей.

Немедленно возвращайте управление после того, как пользователь отправит задачу, и выполняйте асинхронно в фоне.

Реализация асинхронной очереди задач

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

class AsyncAgent:
    """
Асинхронный агент
Используйте очередь задач для обработки длительных задач
    """
   
    def __init__(self, agent, task_queue):
        # Базовый агент
        self.agent = agent
        # Очередь задач
        self.task_queue = task_queue
        # Хранилище результатов
        self.results = {}
        # Запустить фоновый рабочий поток
        self.worker_thread = Thread(target=self.process_queue, daemon=True)
        self.worker_thread.start()
   
    async def submit(self, task):
        """
Отправить задачу
Вернуть ID задачи немедленно
        """
        # Сгенерировать уникальный ID задачи
        task_id = str(uuid.uuid4())
       
        # Добавить в очередь
        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):
        """
Получить результаты задачи
неблокирующий
        """
        if task_id in self.results:
            return self.results[task_id]
       
        # Проверить статус задачи
        status = await self.task_queue.get_status(task_id)
       
        if status is None:
            return None  # Задача не существует
       
        if status == "pending" or status == "processing":
            return {
                "status": status,
                "result": None
            }
       
        return None
   
    def process_queue(self):
        """
Фоновый рабочий поток
Взять задачу из очереди и выполнить её
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
       
        while True:
            task = loop.run_until_complete(self.task_queue.dequeue())
           
            if task:
                # Обновить статус на processing
                loop.run_until_complete(
                    self.task_queue.update_status(task["id"], "processing")
                )
               
                try:
                    # Выполнить задачу
                    result = self.agent.process(task["task"])
                   
                    # Сохранить результат
                    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:
    """Простая реализация очереди задач"""
   
    def __init__(self):
        self.queue = Queue()
        self.statuses = {}
   
    async def enqueue(self, task):
        """Добавить задачу в очередь"""
        self.queue.put(task)
        self.statuses[task["id"]] = "pending"
   
    async def dequeue(self):
        Получить задачу
        if not self.queue.empty():
            return self.queue.get()
        return None
   
    async def get_status(self, task_id):
        Получить статус задачи
        return self.statuses.get(task_id)
   
    async def update_status(self, task_id, status):
        Обновить статус задачи
        self.statuses[task_id] = status

Проектирование микросервисов

Разделите систему Agent на несколько независимых сервисов, чтобы повысить удобство сопровождения и масштабируемость.

Разделение на сервисы

Сервисы Обязанности
API-шлюз Единая точка входа, ограничение частоты аутентификации, перенаправление маршрутов.
Сервис агента Основная логика, рассуждения и планирование.
Сервис инструментов Интеграция с внешними API, вызов инструментов.
Сервис хранения База знаний, управление состоянием, кэширование.
Сервис мониторинга Сбор логов, мониторинг метрик, оповещения.

Межсервисное взаимодействие

синхронное взаимодействие: HTTP/gRPC, используется для обмена запросами и ответами в реальном времени.

асинхронное взаимодействие: Очередь сообщений, используется для длительных операций и уведомлений о событиях.

Краткое содержание главы

В этой главе представлены ключевые практики развертывания в производственной среде и инжиниринга.

Архитектура развёртывания в продакшенеОт автономного к распределённому, масштабируйтесь постепенно.

Контейнеризованное развёртываниеИспользуйте Docker и Kubernetes.

Контроль расходовСокращайте расходы за счёт кэширования и оптимизации токенов.

Потоковая передача и асинхронностьУлучшайте пользовательский опыт и пропускную способность системы.

Микросервисно-ориентированный дизайнПовышайте сопровождаемость и масштабируемость системы.

Развёртывание в продакшене требует комплексного учёта множества аспектов, таких как надёжность, производительность и стоимость.

Рекомендуется начинать с небольшого масштаба и постепенно повышать сложность.

Другие расширения