Local LLMOps: Инфраструктура для локальных LLM
llmopslocal-llminfrastructuremonitoringautomation
Что такое LLMOps?
LLMOps — это набор практик и инструментов для управления жизненным циклом LLM-приложений от разработки до продакшена. Когда мы говорим о локальных LLM, добавляются специфические требования: управление GPU, версионирование моделей, мониторинг инференса.
Архитектура Local LLMOps Platform
┌─────────────────────────────────────────────────────────────┐
│ Local LLMOps Platform │
├─────────────┬──────────────┬──────────────┬─────────────────┤
│ Model Mgmt │ Training │ Serving │ Monitoring │
│ & Version │ & Fine-tune │ & Routing │ & Logging │
├─────────────┼──────────────┼──────────────┼─────────────────┤
│ DVC │ LoRA/QLoRA │ vLLM/Ollama │ Prometheus │
│ Model Card │ PEFT │ Triton │ Grafana │
│ Registry │ RLHF │ TGI │ LangSmith │
└─────────────┴──────────────┴──────────────┴─────────────────┘
Управление моделями
Model Registry
import json
import hashlib
import yaml
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
from datetime import datetime
@dataclass
class ModelMetadata:
"""Метаданные модели"""
name: str
variant: str
size: str # "1.7B", "7B", "13B", "70B"
format: str # "GGUF", "Safetensors", "PyTorch"
quantization: str # "Q4_K_M", "Q8_0", "FP16"
family: str # "Llama", "Mistral", "Qwen", "Gemma"
license: str
author: str
base_model: Optional[str]
training_data: Optional[str]
benchmarks: Dict[str, float]
parameters: int
context_length: int
vocabulary_size: int
created_at: str
tags: List[str]
def to_dict(self):
return asdict(self)
def model_hash(self) -> str:
"""Хеш модели для версионирования"""
content = json.dumps(self.to_dict(), sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
class ModelRegistry:
"""Registry для управления моделями"""
def __init__(self, registry_dir: str = "~/.llmops/models"):
self.registry_dir = Path(registry_dir)
self.registry_dir.mkdir(parents=True, exist_ok=True)
self.index_path = self.registry_dir / "index.json"
self._index = self._load_index()
def _load_index(self) -> Dict:
"""Загрузка индекса моделей"""
if self.index_path.exists():
with open(self.index_path) as f:
return json.load(f)
return {"models": {}, "tags": {}}
def register_model(
self,
metadata: ModelMetadata,
model_path: str
) -> str:
"""Регистрация новой модели"""
model_id = f"{metadata.name}:{metadata.variant}"
self._index["models"][model_id] = {
"metadata": metadata.to_dict(),
"path": model_path,
"hash": metadata.model_hash(),
"registered_at": datetime.now().isoformat(),
"status": "active"
}
# Обновление тегов
for tag in metadata.tags:
if tag not in self._index["tags"]:
self._index["tags"][tag] = []
if model_id not in self._index["tags"][tag]:
self._index["tags"][tag].append(model_id)
self._save_index()
return model_id
def search(
self,
family: str = None,
size: str = None,
format: str = None,
quantization: str = None,
tag: str = None
) -> List[Dict]:
"""Поиск моделей по критериям"""
results = []
for model_id, info in self._index["models"].items():
meta = info["metadata"]
if family and meta["family"] != family:
continue
if size and meta["size"] != size:
continue
if format and meta["format"] != format:
continue
if quantization and meta["quantization"] != quantization:
continue
if tag and tag not in meta["tags"]:
continue
results.append({
"id": model_id,
"metadata": meta,
"path": info["path"]
})
return results
def get_model(self, model_id: str) -> Optional[Dict]:
"""Получение информации о модели"""
return self._index["models"].get(model_id)
def list_by_tag(self, tag: str) -> List[str]:
"""Список моделей по тегу"""
return self._index["tags"].get(tag, [])
def update_benchmarks(self, model_id: str, benchmarks: Dict[str, float]):
"""Обновление бенчмарков"""
if model_id in self._index["models"]:
self._index["models"][model_id]["metadata"]["benchmarks"] = benchmarks
self._save_index()
def _save_index(self):
"""Сохранение индекса"""
with open(self.index_path, 'w') as f:
json.dump(self._index, f, indent=2)
# Использование
registry = ModelRegistry()
metadata = ModelMetadata(
name="llama",
variant="v3.1-7b-instruct",
size="7B",
format="GGUF",
quantization="Q4_K_M",
family="Llama",
license="Apache-2.0",
author="Meta",
base_model="meta-llama/Llama-3.1-7B-Instruct",
benchmarks={
"mmlu": 0.82,
"hellaswag": 0.91,
"truthfulqa": 0.68,
"gsm8k": 0.85
},
parameters=7_000_000_000,
context_length=131072,
vocabulary_size=128256,
created_at=datetime.now().isoformat(),
tags=["instruct", "chat", "general-purpose", "multilingual"]
)
model_id = registry.register_model(metadata, "/models/llama-3.1-7b-q4_k_m.gguf")
print(f"Registered: {model_id}")
# Поиск
results = registry.search(family="Llama", size="7B", quantization="Q4_K_M")
Model Cards
---
name: Qwen2.5-7B-Instruct
variant: q4_k_m
family: Qwen
size: 7B
format: GGUF
quantization: Q4_K_M
license: Apache-2.0
author: Alibaba
base_model: Qwen/Qwen2.5-7B-Instruct
---
# Qwen2.5-7B-Instruct (Q4_K_M GGUF)
## Описание
Модель Qwen2.5-7B-Instruct — это улучшенная версия Qwen2 с лучшей
многозадачностью, кодированием и рассуждением.
## Характеристики
- **Параметры:** 7 миллиардов
- **Контекст:** 32K токенов
- **Вокбул:** 152K токенов
- **Языки:** 29+ языков
## Бенчмарки
| Benchmark | Score | Description |
|-----------|-------|-------------|
| MMLU | 82.4 | Знания и рассуждение |
| MMLU-Pro | 45.2 | Продвинутое рассуждение |
| GSM8K | 85.3 | Математика |
| HumanEval | 78.5 | Кодирование |
| SQuAD | 89.1 | QA |
## Требования к GPU
| Quantization | VRAM (inference) | VRAM (batch=4) |
|--------------|------------------|----------------|
| Q4_K_M | ~5 GB | ~8 GB |
| Q6_K | ~7 GB | ~11 GB |
| Q8_0 | ~9 GB | ~14 GB |
| FP16 | ~14 GB | ~20 GB |
## Использование
```bash
# Ollama
ollama run qwen2.5:7b
# llama.cpp
./main -m qwen2.5-7b-instruct-q4_k_m.gguf -p "Hello, world!"
# vLLM
python -m vllm.entrypoints.api_server \
--model qwen2.5-7b-instruct \
--quantization gguf
Лицензия
Apache-2.0 — свободное использование с указанием авторства
## Автоматизация пайплайна
### CI/CD для LLM
```yaml
# .github/workflows/llm-pipeline.yml
name: LLM Pipeline
on:
push:
paths:
- 'models/**'
- 'data/**'
- 'configs/**'
pull_request:
paths:
- 'models/**'
- 'data/**'
- 'configs/**'
schedule:
- cron: '0 0 * * 1' # Еженедельный пересмотр
jobs:
validate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Validate model files
run: |
for model in models/*.gguf; do
./llama-model-info "$model" || exit 1
done
- name: Check model registry
run: python scripts/check_registry.py
test-inference:
runs-on: gpu-runner
needs: validate
steps:
- uses: actions/checkout@v4
- name: Run inference tests
run: |
python scripts/test_inference.py \
--model models/llama-3.1-7b-q4_k_m.gguf \
--test-suite tests/inference/
- name: Benchmark
run: |
python scripts/benchmark.py \
--model models/llama-3.1-7b-q4_k_m.gguf \
--output artifacts/benchmark-results.json
evaluate:
runs-on: gpu-runner
needs: test-inference
steps:
- uses: actions/checkout@v4
- name: Run evaluation suite
run: |
python scripts/evaluate.py \
--model llama-3.1-7b-q4_k_m \
--benchmarks mmlu,gsm8k,humaneval \
--output artifacts/eval-results.json
- name: Compare with baseline
run: python scripts/compare_benchmarks.py \
--new artifacts/eval-results.json \
--baseline artifacts/baseline.json
deploy:
runs-on: ubuntu-latest
needs: evaluate
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Update model registry
run: python scripts/update_registry.py \
--model llama-3.1-7b-q4_k_m \
--results artifacts/eval-results.json
- name: Notify
run: python scripts/notify.py \
--channel #llm-updates \
--message "Model llama-3.1-7b-q4_k_m updated"
Автоматическое тестирование моделей
import subprocess
import json
from pathlib import Path
from typing import List, Dict
import pytest
class ModelTestSuite:
"""Suite для тестирования моделей"""
def __init__(self, model_path: str):
self.model_path = Path(model_path)
self.results = {}
def test_model_exists(self):
"""Проверка существования файла модели"""
assert self.model_path.exists(), f"Model not found: {self.model_path}"
def test_model_format(self):
"""Проверка формата модели"""
result = subprocess.run(
["./llama-model-info", str(self.model_path)],
capture_output=True, text=True
)
assert result.returncode == 0, f"Invalid model format: {result.stderr}"
info = json.loads(result.stdout)
assert info["format"] in ["GGUF", "Safetensors", "PyTorch"]
def test_inference(self, prompt: str = "The quick brown fox", max_tokens: int = 10):
"""Тест инференса"""
result = subprocess.run(
[
"./main",
"-m", str(self.model_path),
"-n", str(max_tokens),
"-p", prompt,
"--temp", "0.7"
],
capture_output=True, text=True, timeout=60
)
assert result.returncode == 0, f"Inference failed: {result.stderr}"
assert len(result.stdout) > len(prompt), "Model produced no output"
def test_batch_inference(self, prompts: List[str], batch_size: int = 4):
"""Тест батч-инференса"""
results = []
for i in range(0, len(prompts), batch_size):
batch = prompts[i:i + batch_size]
# Тестирование батча
for prompt in batch:
self.test_inference(prompt, max_tokens=5)
def test_consistency(self, prompt: str, iterations: int = 5):
"""Тест консистентности"""
outputs = []
for _ in range(iterations):
self.test_inference(prompt, max_tokens=20)
# В реальном коде здесь был бы сбор outputs
# Проверка что outputs не слишком различаются
# (можно использовать embedding similarity)
def test_resource_usage(self, max_vram: int = 8, max_ram: int = 16):
"""Тест использования ресурсов"""
# Запуск с мониторингом ресурсов
# Проверка что VRAM и RAM в пределах лимитов
pass
def run_all(self) -> Dict:
"""Запуск всех тестов"""
tests = [
("exists", self.test_model_exists),
("format", self.test_model_format),
("inference", lambda: self.test_inference()),
("consistency", lambda: self.test_consistency("Test prompt")),
]
results = {}
for name, test_fn in tests:
try:
test_fn()
results[name] = {"status": "passed", "error": None}
except Exception as e:
results[name] = {"status": "failed", "error": str(e)}
self.results = results
return results
Мониторинг локальных LLM
Prometheus Metrics
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import psutil
import os
class LLMMetrics:
"""Метрики для LLM инференса"""
# Счётчики
request_counter = Counter(
'llm_requests_total',
'Total LLM requests',
['model', 'endpoint', 'status']
)
request_latency = Histogram(
'llm_request_latency_seconds',
'LLM request latency',
['model', 'endpoint'],
buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0)
)
token_counter = Counter(
'llm_tokens_total',
'Total tokens processed',
['model', 'direction'] # input/output
)
# Гейджи
gpu_memory_used = Gauge(
'gpu_memory_used_bytes',
'GPU memory used in bytes',
['gpu_id']
)
gpu_memory_total = Gauge(
'gpu_memory_total_bytes',
'GPU memory total in bytes',
['gpu_id']
)
gpu_utilization = Gauge(
'gpu_utilization_percent',
'GPU utilization percentage',
['gpu_id']
)
cpu_memory_used = Gauge(
'cpu_memory_used_bytes',
'CPU memory used in bytes'
)
queue_size = Gauge(
'llm_queue_size',
'Number of requests in queue'
)
active_requests = Gauge(
'llm_active_requests',
'Number of active requests'
)