Кластеризация LLM: как распределить обработку промптов между разным железом (DGX Spark + Mac Studio)

Проблема: гетерогенное железо и неэффективное использование ресурсов

Представьте ситуацию: у вас есть мощный сервер DGX Spark с несколькими H100/H200 для тяжёлых инференс-задач и Mac Studio с M3 Ultra для повседневной разработки. Каждый работает отдельно, но вы хотите объединить их в единую систему, чтобы:

• Распределять промпты между узлами в зависимости от сложности и приоритета
• Использовать Mac Studio для лёгких запросов (чат, код-ревью)
• Направлять сложные задачи (RAG, длинные контексты) на DGX Spark
• Обеспечить отказоустойчивость — если один узел падает, система продолжает работать
• Максимально загрузить все доступные ресурсы, включая GPU и CPU

Решение: архитектура гетерогенного кластера LLM

Мы построим систему, где:

1. Роутер (Load Balancer) принимает все входящие промпты и решает, куда их направить
2. DGX Spark работает с большими моделями (70B+) через vLLM для максимальной производительности
3. Mac Studio обрабатывает лёгкие модели (7B-13B) через llama.cpp, используя CPU/Neural Engine
4. Мониторинг отслеживает загрузку каждого узла и латенси

Пошаговый план реализации

1 Подготовка узлов: настройка DGX Spark и Mac Studio

Сначала настроим каждый узел отдельно, чтобы они могли работать как независимые инференс-серверы.

На DGX Spark (Ubuntu 22.04):

# Установка vLLM с поддержкой CUDA 12.1
pip install vllm

# Запуск сервера vLLM с моделью 70B
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-70B-Instruct \
    --tensor-parallel-size 4 \
    --gpu-memory-utilization 0.9 \
    --port 8000 \
    --host 0.0.0.0

На Mac Studio (macOS Sonoma):

# Установка llama.cpp с Metal поддержкой
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make clean && LLAMA_METAL=1 make -j

# Конвертация модели в GGUF формат
python convert.py \
    --outfile models/llama-3.2-3b-instruct.Q4_K_M.gguf \
    --outtype q4_K_M

# Запуск сервера llama.cpp
./server -m models/llama-3.2-3b-instruct.Q4_K_M.gguf \
    -c 4096 \
    --host 0.0.0.0 \
    --port 8080 \
    -ngl 99  # Использовать все Neural Engine ядра

2 Создание интеллектуального роутера на Python

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

import asyncio
import aiohttp
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import tiktoken  # Для подсчёта токенов

app = FastAPI(title="LLM Cluster Router")

# Конфигурация узлов
NODES = {
    "dgx_spark": {
        "url": "http://dgx-spark-ip:8000/v1/completions",
        "model": "llama-3.1-70b",
        "max_tokens": 8192,
        "priority": "high"
    },
    "mac_studio": {
        "url": "http://mac-studio-ip:8080/completion",
        "model": "llama-3.2-3b",
        "max_tokens": 4096,
        "priority": "normal"
    }
}

class PromptRequest(BaseModel):
    prompt: str
    max_tokens: Optional[int] = 512
    temperature: Optional[float] = 0.7
    priority: Optional[str] = "normal"
    task_type: Optional[str] = "chat"  # chat, rag, coding, reasoning

def count_tokens(text: str) -> int:
    """Подсчёт токенов для роутинга"""
    encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def select_node(request: PromptRequest) -> str:
    """Интеллектуальный выбор узла"""
    token_count = count_tokens(request.prompt)
    
    # Правила роутинга
    if request.priority == "high":
        return "dgx_spark"
    
    if token_count > 2000:
        return "dgx_spark"
    
    if request.task_type in ["rag", "reasoning", "coding"]:
        return "dgx_spark"
    
    if token_count <= 500 and request.task_type == "chat":
        return "mac_studio"
    
    # По умолчанию — балансировка нагрузки
    return "mac_studio"  # Или можно добавить логику round-robin

@app.post("/generate")
async def generate_text(request: PromptRequest):
    """Основной endpoint для генерации текста"""
    node_name = select_node(request)
    node_config = NODES[node_name]
    
    try:
        async with aiohttp.ClientSession() as session:
            # Адаптация запроса под формат узла
            if node_name == "dgx_spark":
                payload = {
                    "model": node_config["model"],
                    "prompt": request.prompt,
                    "max_tokens": min(request.max_tokens, node_config["max_tokens"]),
                    "temperature": request.temperature
                }
                async with session.post(node_config["url"], json=payload) as resp:
                    result = await resp.json()
                    return {"node": node_name, "response": result["choices"][0]["text"]}
            
            else:  # mac_studio (llama.cpp формат)
                payload = {
                    "prompt": request.prompt,
                    "n_predict": min(request.max_tokens, node_config["max_tokens"]),
                    "temperature": request.temperature
                }
                async with session.post(node_config["url"], json=payload) as resp:
                    result = await resp.json()
                    return {"node": node_name, "response": result["content"]}
                    
    except Exception as e:
        # Fallback на другой узел при ошибке
        fallback_node = "mac_studio" if node_name == "dgx_spark" else "dgx_spark"
        # Повторная попытка на fallback узле
        # ...
        raise HTTPException(status_code=500, detail=f"Node error: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

3 Настройка мониторинга и балансировки нагрузки

Для эффективного роутинга нужен мониторинг состояния узлов:

# monitoring.py
import psutil
import GPUtil
from datetime import datetime
import requests

class NodeMonitor:
    def __init__(self, node_url):
        self.node_url = node_url
        
    def check_health(self):
        """Проверка здоровья узла"""
        try:
            # Для vLLM
            if ":8000" in self.node_url:
                resp = requests.get(f"{self.node_url.replace('/v1/completions', '/health')}", timeout=5)
                return resp.status_code == 200
            # Для llama.cpp
            else:
                resp = requests.get(f"{self.node_url.replace('/completion', '/health')}", timeout=5)
                return resp.status_code == 200
        except:
            return False
    
    def get_metrics(self):
        """Сбор метрик узла"""
        metrics = {
            "timestamp": datetime.now().isoformat(),
            "cpu_percent": psutil.cpu_percent(),
            "memory_percent": psutil.virtual_memory().percent,
            "load_avg": psutil.getloadavg()[0]
        }
        
        # GPU метрики для DGX
        if ":8000" in self.node_url:
            try:
                gpus = GPUtil.getGPUs()
                metrics["gpu_utilization"] = [gpu.load * 100 for gpu in gpus]
                metrics["gpu_memory"] = [gpu.memoryUtil * 100 for gpu in gpus]
            except:
                metrics["gpu_utilization"] = []
                
        return metrics

# Использование в роутере
monitors = {
    "dgx_spark": NodeMonitor("http://dgx-spark-ip:8000"),
    "mac_studio": NodeMonitor("http://mac-studio-ip:8080")
}

# Перед выбором узла проверяем здоровье
healthy_nodes = {}
for name, monitor in monitors.items():
    if monitor.check_health():
        healthy_nodes[name] = monitor.get_metrics()

Нюансы реализации и частые ошибки

1. Проблемы с совместимостью API

vLLM и llama.cpp имеют разные API форматы. Решение — создать адаптер-слой:

class APIAdapter:
    @staticmethod
    def to_vllm_format(request):
        return {
            "prompt": request.prompt,
            "max_tokens": request.max_tokens,
            "temperature": request.temperature,
            "stream": False
        }
    
    @staticmethod
    def to_llamacpp_format(request):
        return {
            "prompt": request.prompt,
            "n_predict": request.max_tokens,
            "temperature": request.temperature,
            "stream": False
        }

2. Сетевая задержка между узлами

Если узлы в разных дата-центрах, латенси может убить производительность. Решения:

• Использовать WireGuard или Tailscale для secure VPN
• Кэшировать частые промпты на роутере
• Балансировать не только по загрузке, но и по сетевой задержке

3. Консистентность ответов

Разные модели на разных узлах дают разные ответы на один промпт. Стратегии:

• Использовать одинаковые модели, но с разными квантованиями
• Добавить post-processing для нормализации ответов
• Для критичных задач всегда использовать DGX Spark

Продвинутые сценарии использования

Каскадная обработка (Chain Processing)

Сложные промпты можно разбивать на этапы, выполняя каждый на оптимальном узле:

async def process_complex_prompt(prompt):
    # Этап 1: Анализ на Mac Studio (быстро)
    analysis_prompt = f"Analyze this request: {prompt}. What type of task is this?"
    analysis = await send_to_node(analysis_prompt, "mac_studio")
    
    # Этап 2: Основная генерация на DGX Spark (качественно)
    if "requires reasoning" in analysis:
        main_result = await send_to_node(prompt, "dgx_spark")
    else:
        main_result = await send_to_node(prompt, "mac_studio")
        
    # Этап 3: Пост-обработка на Mac Studio
    final_prompt = f"Polish this response: {main_result}"
    final_result = await send_to_node(final_prompt, "mac_studio")
    
    return final_result

Динамическое масштабирование

Добавляем возможность подключать новые узлы «на лету»:

@app.post("/register_node")
async def register_node(node_info: NodeInfo):
    """Динамическая регистрация нового узла"""
    NODES[node_info.name] = {
        "url": node_info.url,
        "model": node_info.model,
        "capabilities": node_info.capabilities,
        "registered_at": datetime.now()
    }
    
    # Автоматически добавляем мониторинг
    monitors[node_info.name] = NodeMonitor(node_info.url)
    
    return {"status": "registered", "total_nodes": len(NODES)}

FAQ: Частые вопросы

Заключение

Кластеризация LLM на гетерогенном железе — это не просто техническая задача, а стратегическое решение для оптимизации ресурсов. Объединяя DGX Spark для тяжёлых вычислений и Mac Studio для лёгких задач, вы получаете:

• Экономию до 40% на инфраструктуре (не нужно покупать лишние GPU)
• Увеличение uptime за счёт отказоустойчивой архитектуры
• Гибкость — можно легко добавлять новые типы железа
• Оптимальное качество ответов — каждая задача выполняется на подходящем для неё железе

Начните с простого роутера на Python, постепенно добавляя мониторинг, балансировку и отказоустойчивость. И помните: лучшая архитектура та, которая решает ваши конкретные задачи, а не следует модным трендам.