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

Вы запустили модель на ноутбуке, и она генерировала ответы быстро и точно. Но когда вы помещаете её на сервер и позволяете сотне пользователей одновременно вызывать её, ситуация меняется:Некоторые пользователи ждут ответа десять секунд, у некоторых запросов истекает тайм-аут и появляются ошибки, память GPU заполняется в мгновение ока, а счёт вырастает до болезненного уровня.

Вот проблему, которую должно решить развертывание ИИ:Превратить работающую модель в готовый к использованию сервис。

Узким местом традиционных веб-сервисов обычно являются CPU и база данных.

Узкое место для AI-сервисов — это в основном GPU: достаточно ли памяти, достаточно ли быстро вычисления и как ставятся в очередь одновременные запросы.

Основные задачи развёртывания AI можно свести к трём словам: задержка, пропускная способность и стоимость.

Метрика	Значение	Пользовательский опыт
Задержка	Время с момента отправки пользователем запроса до получения первого символа	Быстро или нет
Пропускная способность	Сколько запросов можно обработать в секунду	Может ли он обслуживать много людей одновременно
Стоимость	Стоимость GPU/сервера для работы сервиса	Дорого или нет

Хорошее развёртывание AI — это поиск баланса между этими тремя: достаточно низкая задержка, достаточно высокая пропускная способность и контролируемая стоимость.

Фреймворк для обслуживания моделей

Чтобы обернуть обученную модель как API-сервис, нужен специализированный фреймворк. В этом разделе представлены три основных варианта: vLLM, TGI и Ollama.

vLLM: высокопроизводительный инференс на основе PagedAttention

vLLM — это движок инференса, разработанный командой UC Berkeley, и его главная особенность — скорость.

Ключевая инновация vLLM — этоPagedAttention— техника эффективного управления памятью GPU, вдохновлённая страничной организацией виртуальной памяти операционных систем.

В традиционных фреймворках инференса KV Cache (кэш ключей-значений) каждого запроса должен занимать непрерывное пространство памяти GPU.

Когда запросы имеют разную длину, память GPU фрагментируется, что приводит к низкой утилизации.

PagedAttention делит память GPU на страницы фиксированного размера. KV Cache каждого запроса может быть разбросан по разным страницам, а таблица страниц фиксирует их расположение.

Это значительно повышает утилизацию памяти GPU и позволяет обслуживать больше запросов одновременно.

Пример

# Установка vLLM
pip install vllm

# Запуск API-сервиса, совместимого с OpenAI, с помощью vLLM
# --model задаёт модель, --host и --port задают адрес прослушивания
# --tensor-parallel-size: количество GPU для параллельного инференса
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192

После запуска сервиса вы можете напрямую вызывать его с помощью OpenAI SDK:

Пример

# Путь к файлу: test_vllm_client.py
from openai import OpenAI

# Подключение к локальному сервису vLLM
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="example-demo-key" # vLLM по умолчанию не требует настоящего API-ключа
)

# Вызов завершения чата
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "Вы — полезный ИИ-ассистент."},
{"role": "user", "content": "Представь учебник по Python в одном предложении."}
],
temperature=0.7,
max_tokens=500,
stream=True # Потоковый вывод
)

print("Ответ ИИ:", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

vLLM также поддерживает написание пользовательских сервисов напрямую с помощью кода на Python:

Пример

# Путь к файлу: custom_vllm_server.py
from vllm import LLM, SamplingParams

# Инициализация модели
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
gpu_memory_utilization=0.9, # Доля использования памяти GPU
tensor_parallel_size=1, # Тензорный параллелизм (количество GPU)
max_model_len=8192, # Максимальная длина контекста
)

# Настройка параметров сэмплирования
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=500,
stop=["</s>"],
)

# Пакетный инференс
prompts = [
"Расскажи о Python",
"Что такое ИИ?",
"Как научиться программировать?",
]

# Сгенерировать ответ
outputs = llm.generate(prompts, sampling_params)

# Вывести результаты
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Запрос: {prompt}")
print(f"Сгенерировано: {generated_text}")
print("-" * 50)

TGI: движок вывода Hugging Face

TGI (Text Generation Inference) — это фреймворк сервиса инференса, выпущенный Hugging Face.

Он обладает полноценной экосистемой, глубокой интеграцией с Hugging Face Hub и поддерживает такие оптимизации, как Flash Attention и динамический батчинг.

Пример

# Запуск TGI с Docker (самый простой способ)
# Примечание: сначала нужно установить Docker и настроить поддержку GPU
docker run -d \
--gpus all \
-p 8080:80 \
-v $PWD/data:/data \
--name tgi-server \
ghcr.io/huggingface/text-generation-inference:latest \
--model-id Qwen/Qwen2.5-7B-Instruct \
--max-input-length 4096 \
--max-total-tokens 8192 \
--max-batch-prefill-tokens 4096

# Или установите напрямую с помощью Python (среда разработки)
# pip install text-generation

Вызов сервиса TGI:

Пример

# Путь к файлу: test_tgi_client.py
from text_generation import Client

# Подключение к сервису TGI
client = Client("http://localhost:8080", timeout=60)

# Вызов без потоковой передачи
response = client.generate(
"Представь руководство по Python",
max_new_tokens=500,
temperature=0.7,
top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# Потоковый вызов
print("Потоковый вывод:")
for chunk in client.generate_stream(
"Напиши Hello World на Python",
max_new_tokens=200,
):
if not chunk.token.special:
print(chunk.token.text, end="", flush=True)
print()

Ollama: от локального демо до развёртывания в продакшене

Ollama известен возможностью "запуска локальной модели в один клик", и многие используют его для разработки и демонстраций.

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

Пример

# Установка Ollama (сервер Linux)
curl -fsSL https://ollama.com/install.sh | sh

# Настройте Ollama для прослушивания на всех сетевых интерфейсах (обратите внимание на брандмауэр в продакшене)
# Отредактируйте файл службы systemd
sudo systemctl edit ollama.service

# Добавьте следующее содержимое:
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# Перезапустите службу
sudo systemctl restart ollama

# Скачайте и запустите модель
ollama pull qwen2.5:7b

Вызовите Ollama через API:

Пример

# Путь к файлу: test_ollama_client.py
import requests
import json

# Адрес API Ollama
base_url = "http://localhost:11434/api"

# Интерфейс чата (потоковая передача)
def chat_stream(model, messages):
url = f"{base_url}/chat"
data = {
"model": model,
"messages": messages,
"stream": True,
"options": {
"temperature": 0.7,
"num_ctx": 8192,
}
}

response = requests.post(url, json=data, stream=True)
full_response = ""

for line in response.iter_lines():
if line:
chunk = json.loads(line)
if "message" in chunk:
content = chunk["message"].get("content", "")
full_response += content
print(content, end="", flush=True)
if chunk.get("done", False):
break

print()
return full_response

# Пример вызова
messages = [
{"role": "system", "content": "Вы — полезный ИИ-ассистент."},
{"role": "user", "content": "Представьте руководство по Python"}
]

print("Ответ ИИ:", end="")
result = chat_stream("qwen2.5:7b", messages)

Сравнение выбора фреймворков

У каждого из трёх фреймворков есть свои особенности. Как выбрать? Это зависит от ваших потребностей.

Фреймворк	Преимущества	Недостатки	Применимые сценарии
vLLM	Самая высокая производительность, максимальная пропускная способность	Относительно сложная конфигурация	Высоконагруженные эксплуатационные среды, ориентированные на производительность
TGI	Экосистема Hugging Face, простое развёртывание	Производительность немного ниже, чем у vLLM	Уже используется инструментарий Hugging Face
Ollama	Минимальное развёртывание, низкие затраты на обслуживание	Не подходит для чрезвычайно высокой одновременной нагрузки	Малые и средние сервисы, быстрый запуск

Ранний этап: начните с Ollama, просто и надёжно. После роста трафика: переходите на vLLM, отдавая приоритет производительности.

Техники оптимизации инференса

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

В этом разделе представлены техники оптимизации инференса, часто используемые в производственных средах.

Непрерывная пакетная обработка

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

Проблема в том, что некоторые запросы короткие (например, генерируют всего 10 символов), а некоторые длинные (генерируют 500 символов).

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

Непрерывная пакетная обработкаОна динамическая: когда один запрос завершается, новые запросы из очереди немедленно добавляются, не дожидаясь завершения всего пакета.

Таким образом, GPU почти всегда загружен полностью, что значительно увеличивает пропускную способность.

И vLLM, и TGI имеют встроенную непрерывную пакетную обработку. Вам не нужно реализовывать её вручную; просто включите её.

Спекулятивное декодирование

Генерация одного токена большой моделью требует полного прямого прохода, что медленно.

Идея спекулятивного декодирования такова:Использовать небольшую модель, чтобы «угадать» следующие несколько токенов, а затем большая модель проверяет их все сразу.。

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

Неправильные догадки отбрасываются и угадываются заново.

Это может повысить общую скорость в 2-3x, сохраняя качество вывода неизменным.

Пример

# Включить спекулятивное декодирование в vLLM
# --speculative-model указывает небольшую модель (черновую модель)
# --num-speculative-tokens указывает, сколько токенов угадывать каждый раз
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--speculative-model Qwen/Qwen2.5-0.5B-Instruct \
--num-speculative-tokens 5 \
--host 0.0.0.0 \
--port 8000

Квантованный инференс: AWQ и GPTQ

Квантование модели сжимает веса модели из 16-битных чисел с плавающей запятой (FP16) в 8-битные или 4-битные целые числа.

Потребление памяти сокращается вдвое или даже меньше, а инференс может работать быстрее.

Два наиболее часто используемых метода квантования — это AWQ и GPTQ.

Метод квантования	Характеристика	Потеря точности	Скорость инференса
AWQ	Квантование весов с учётом активаций	Минимальный	Высокая
GPTQ	Послойно оптимизированное квантование	малый	Высокая
GGUF	Подходит для локального развертывания	Приемлемо	Высокая

Пример

# Запуск квантованной модели AWQ напрямую с vLLM
# Многие квантованные модели можно скачать напрямую с Hugging Face
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Qwen2.5-7B-Instruct-AWQ \
--quantization awq \
--host 0.0.0.0 \
--port 8000

# Или запустите квантованные модели GGUF с Ollama
ollama run qwen2.5:7b-instruct-q4_0

Загрузите AWQ-квантованную модель в Python:

Пример

# Путь к файлу: awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# Загрузить AWQ-квантованную модель
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# Примечание: для запуска этого необходимо установить библиотеку autoawq
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)

# Инференс
messages = [
{"role": "system", "content": "Вы — полезный ИИ-ассистент."},
{"role": "user", "content": "Расскажи о Python"}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.7,
)

response = tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
print(response)

Дистилляция модели

Дистилляция использует большую модель (учителя) для обучения маленькой модели (ученика), позволяя маленькой модели перенять возможности большой модели.

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

Например, использование Qwen2.5-72B в качестве учителя для обучения ученика Qwen2.5-7B может дать результаты, близкие к модели 72B, но со скоростью и стоимостью на уровне 7B.

Преимущества дистилляции обычно таковы:Размер модели сокращается до 1/10, скорость инференса увеличивается в 5–10 раз, при этом сохраняется более 90% производительности.。

Проектирование API-сервиса

Обёртывание модели — это лишь первый шаг; production API также требует множества возможностей: аутентификация, ограничение частоты запросов, потоковые ответы и обработка ошибок.

FastAPI — отличный выбор для этого.

Обёртывание LLM-сервисов с помощью FastAPI

Давайте напишем полноценный API-сервис с аутентификацией, ограничением частоты запросов и потоковыми ответами.

Пример

# Путь к файлу: llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. Инициализируйте приложение и конфигурацию
# ============================================

app = FastAPI(title="EXAMPLE LLM API", version="1.0.0")

# Простая аутентификация по API-ключу (для продакшена рекомендуется более надёжное решение)
VALID_API_KEYS = {
sk-example-123456": {"user": "demo_user", "quota": 1000},
"sk-example-789012": {"user": "pro_user"⟦⟧"quota", "quota": 10000},
}

# Простой ограничитель частоты запросов (реализация в памяти; для продакшена рекомендуется Redis)
class RateLimiter:
def __init__(self):
self.requests⟦⟧defaultdict = defaultdict(list) # ключ -> [метка времени, ...]

async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
"""Проверить, превышен ли лимит запросов"""
now = time.time()
# Очистить устаревшие записи
self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
# Проверить, превышен ли лимит
if len(self.requests[key]) >= max_requests:
return False
# Записать этот запрос
self.requests[key].append(now)
return True

rate_limiter = RateLimiter()

# ============================================
# 2. Модели данных
# ============================================

class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"

class Message(BaseModel):
role: Role = Field(..., description="Роль сообщения")
content: str = Field(..., description="Содержимое сообщения")

class ChatCompletionRequest(BaseModel):
model: str = Field(..., description="Имя модели")
messages: List[Message] = Field(..., description="Список сообщений диалога")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="Температура сэмплирования")
top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Сэмплирование Top-P")
max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="Максимальное количество генерируемых токенов")
stream: Optional[bool] = Field(default=False, description="Использовать ли потоковый вывод")

class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[dict]
usage: dict

# ============================================
# 3. Зависимость аутентификации
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Проверить API-ключ и вернуть информацию о пользователе"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Неверный API-ключ"
)
return VALID_API_KEYS[api_key]

# ============================================
# 4. Симуляция генерации LLM (замените на реальные вызовы модели в реальных проектах)
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
"""Симулировать процесс генерации LLM"""
# Это всего лишь симуляция; в реальном проекте вызывайте настоящую модель
response_text = (
"Привет! Я ИИ-ассистент EXAMPLE."\n\n"
"EXAMPLE предлагает множество обучающих материалов по программированию, включая Python, Java, C++, фронтенд-разработку и многое другое."\n"
"Вы можете посетить https://www.example.com, чтобы узнать больше!"\n\n"
"Чем я могу вам помочь?"
)
# Имитация задержки генерации
await asyncio.sleep(0.5)
return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
"""Имитировать потоковую генерацию LLM"""
response_text = await mock_llm_generate(messages, max_tokens)
# Выводить посимвольно, чтобы имитировать эффект потоковой передачи
for char in response_text:
yield char
await asyncio.sleep(0.03) # Задержка 30ms на символ

# ============================================
# 5. Конечные точки API
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
request: ChatCompletionRequest,
user: dict = Depends(get_current_user)
):
"""Конечная точка завершения чата, совместимая с форматом OpenAI"""

# Проверка ограничения частоты запросов: каждый пользователь может сделать не более 60 запросов в минуту
user_key = user["user"]
if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Превышен лимит запросов"
)

# Сгенерировать ID запроса
request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(datetime.now().timestamp())

if request.stream:
# Потоковый ответ
async def stream_generator():
async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
# Создаём фрагмент в формате SSE
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": created,
"model": request.model,
"choices": [{
"index": 0,
"delta": {"content": token},
"finish_reason": None
}]
}
yield f"data: {__import__('json').dumps(chunk)}\n\n"
# Отправить маркер окончания
yield "data: [DONE]\n\n"

return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)

else:
# Ответ без потоковой передачи
response_text = await mock_llm_generate(request.messages, request.max_tokens)

return ChatCompletionResponse(
id=request_id,
created=created,
model=request.model,
choices=[{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}],
usage={
"prompt_tokens": sum(len(m.content) for m in request.messages),
"completion_tokens": len(response_text),
"total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
}
)

@app.get("/health")
async def health_check():
"""Эндпоинт проверки работоспособности"""
return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
"""Список доступных моделей"""
return {
"object": "list",
"data": [
{
"id": "qwen2.5-7b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "example"
},
{
"id": "llama-3.1-8b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "example"
}
]
}

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

Запустите сервис:

Пример

# Установить зависимости
pip install fastapi uvicorn python-multipart pydantic

# Запустите сервис
python llm_api_server.py

# Проверьте в другом терминале
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-example-123456" \
-d '{
"model": "qwen2.5-7b-instruct",
"messages": [
{"role": "user", "content": "Привет"}
],
"stream": false
}'

Вызов с Python-клиентом:

Пример

# Путь к файлу: client_example.py
from openai import OpenAI

# Подключение к собственному API-сервису
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="sk-example-123456"
)

# Тест без потоковой передачи
print("Вызов без потоковой передачи:")
response = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Расскажи о Python"}],
stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# Тест потоковой передачи
print("Потоковый вызов:")
stream = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Расскажи о Python"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

Преимущества интерфейсов, совместимых с OpenAI

Создание API, совместимого с OpenAI, имеет несколько явных преимуществ:

Пользователям не нужно менять код, достаточно изменить base_url и api_key.

Богатая экосистема — любой SDK, инструмент или приложение с поддержкой OpenAI может напрямую использовать ваш сервис.

Не нужно разрабатывать собственный формат API; дизайн OpenAI уже хорошо продуман.

Аутентификация и ограничение частоты запросов в продакшен-средах

В предыдущих примерах использовалось простое ограничение частоты запросов в памяти; для продакшена рекомендуется использовать Redis для распределённого ограничения частоты запросов.

Пример

# Путь к файлу: redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
"""Распределённый ограничитель скорости на основе Redis"""

def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)

async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int
) -> bool:
"""
Алгоритм ограничения скорости на основе скользящего окна
key: ключ ограничения скорости (например, user_id или api_key)
max_requests: максимальное количество запросов в окне
window_seconds: размер окна (в секундах)
"""
now = time.time()
window_start = now - window_seconds

# Ключ Redis
redis_key = f"rate_limit:{key}"

# Используйте атомарные операции pipeline
async with self.redis.pipeline() as pipe:
# 1. Удалить записи за пределами окна
pipe.zremrangebyscore(redis_key, 0, window_start)
# 2. Подсчитать запросы в текущем окне
pipe.zcard(redis_key)
# 3. Добавить текущий запрос
pipe.zadd(redis_key, {str(now): now})
# 4. Установить время истечения
pipe.expire(redis_key, window_seconds)
# Выполнить
_, current_count, _, _ = await pipe.execute()

return current_count < max_requests

async def close(self):
await self.redis.close()

# Пример использования
async def main():
limiter = RedisRateLimiter("redis://localhost:6379")

# Тест: один и тот же пользователь не более 10 запросов в минуту
user_id = "user_123"
for i in range(15):
allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
print(f"Request {i+1}: {'allowed' if allowed else 'denied'}")

await limiter.close()

if __name__ == "__main__":
import asyncio
asyncio.run(main())

Аутентификация и ограничение скорости — это инфраструктура API; лучше сделать их немного сложнее, чем пожалеть после того, как вас захлестнёт.

Балансировка нагрузки и масштабирование

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

Именно эту задачу должны решать балансировка нагрузки и автоматическое масштабирование.

Проектирование сервисов без состояния

Сначала запомните:AI-сервисы должны проектироваться как сервисы без состояния。

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

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

Состояние сессии (например, история диалога) можно хранить на клиенте (пусть пользователь отправляет его каждый раз) или в Redis.

Обратный прокси Nginx

Nginx — самый распространённый балансировщик нагрузки, распределяющий трафик между несколькими бэкенд-серверами.

Пример

# Путь к файлу: nginx.conf
# Конфигурация Nginx: балансировка нагрузки для нескольких LLM-сервисов

upstream llm_backends {
# Алгоритм наименьшего числа соединений: отправлять запросы на сервер с наименьшим текущим количеством соединений
least_conn;

# Список бэкенд-серверов
server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

# Если у вас несколько моделей GPU, можно распределять трафик в разных пропорциях с помощью параметра weight
# weight=2 означает, что этот сервер получает вдвое больше трафика, чем остальные
}

server {
listen 80;
server_name api.example.com;

# Переадресация API-запросов
location /v1/ {
# Перенаправление на бэкенд
proxy_pass http://llm_backends;

# Настройки тайм-аутов (генерация ИИ может быть медленной)
proxy_connect_timeout 10s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;

# Передача реального IP-адреса клиента
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# Настройки, необходимые для потоковых ответов
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
}

# Эндпоинт проверки работоспособности
location /health {
proxy_pass http://llm_backends/health;
proxy_connect_timeout 5s;
proxy_read_timeout 5s;
}

# Ограничение частоты: максимум 10 запросов в секунду на один IP
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
}

Автоматическое масштабирование Kubernetes HPA

Если вы развёртываете с помощью Kubernetes, вы можете использовать HPA (Horizontal Pod Autoscaler) для автоматического масштабирования вверх и вниз в зависимости от нагрузки.

Пример

# Путь к файлу: llm-deployment.yaml
# Конфигурация развёртывания Kubernetes для LLM-сервиса

# 1. Deployment: развернуть LLM Pod
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-server
namespace: example
spec:
replicas: 3 # Начальное количество реплик
selector:
matchLabels:
app: llm-server
template:
metadata:
labels:
app: llm-server
spec:
containers:
- name: llm-server
image: example/llm-server:latest
ports:
- containerPort: 8000
ресурсы:
requests:
cpu: "4"
memory: "16Gi"
nvidia.com/gpu: 1 # Запрос 1 GPU
limits:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: 1 # Лимит 1 GPU
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300 # Загрузка модели занимает время
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300
periodSeconds: 10
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"

---
# 2. Service: балансировка нагрузки Service
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: example
spec:
selector:
app: llm-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

---
# 3. HPA: автомасштабирование
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-hpa
namespace: example
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-server
minReplicas: 3 # Минимум 3 Pods
maxReplicas: 20 # Максимум 20 Pods
metrics:
# Масштабирование на основе использования CPU
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# Масштабирование на основе использования GPU (требуются пользовательские метрики)
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: 80
# Масштабирование на основе длины очереди (пользовательская метрика)
- type: Pods
pods:
metric:
name: pending_requests
target:
type: AverageValue
averageValue: 10
behavior:
scaleUp:
stabilizationWindowSeconds: 60 # Наблюдать 60 секунд перед масштабированием вверх
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # Наблюдать 5 минут перед масштабированием вниз
policies:
- type: Percent
value: 20
periodSeconds: 120

Управление пулом ресурсов GPU

GPU стоят дорого, поэтому максимизируйте их использование.

Распространённая стратегия: группировать по типу модели, размещая Pod'ы одной и той же модели вместе.

Например:

Пул ресурсов	Модель GPU	Запускаемая модель	Назначение
small-pool	RTX 3090 / L4	модель 7B	Суточный трафик
medium-pool	A10 / L40	модель 13B-70B	Сложные задачи
large-pool	A100 / H100	модель 70B+	Ценные пользователи
spot-pool	Различные типы машин	Офлайн-пакетная обработка	Используйте недорогие Spot-инстансы

A/B-тестирование моделей ИИ

Вы выпустили новую версию модели. Как узнать, лучше ли она старой? Используйте A/B-тестирование.

Суть A/B-тестирования:Разделите трафик на группы, используйте разные модели для разных групп и сравнивайте метрики。

Стратегия разделения трафика

Существует несколько распространённых методов разделения трафика:

Пример

# Путь к файлу: ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
"""Вариант A/B-тестирования"""
name: str # Имя варианта
model_name: str # Имя модели
weight: int # Вес трафика
config: dict # Прочая конфигурация

class ABTestRouter:
"""Маршрутизатор трафика A/B-тестирования"""

def __init__(self, test_name: str, variants: List[Вариант]):
self.test_name = test_name
self.variants = variants
# Вычислить суммарный вес
total_weight = sum(v.weight for v in variants)
# Построить диапазоны сегментов
self.ranges: List[Tuple[int, int, Вариант]] = []
current = 0
for v in variants:
self.ranges.append((current, current + v.weight, v))
current += v.weight
self.total_weight = total_weight

def get_variant_by_user(self, user_id: str) -> Variant:
"""
Стабильное разделение трафика на основе ID пользователя
Один и тот же пользователь всегда назначается в одну и ту же группу, чтобы избежать непоследовательного опыта
"""
# Вычислить оценку пользователя с помощью хеша
hash_input = f"{self.test_name}-{user_id}".encode()
hash_value = hashlib.md5(hash_input).hexdigest()
# Преобразовать в число из диапазона 0-999
score = int(hash_value[:4], 16) % 1000
# Найти соответствующий вариант
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

def get_variant_random(self) -> Variant:
"""Случайное разделение (подходит для сценариев, не требующих закрепления)"""
score = random.randint(0, self.total_weight - 1)
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

# Пример использования
def main():
# Определяем три варианта
variants = [
Вариант(
name="control",
model_name="Qwen2.5-7B-Instruct",
weight=50, # 50% трафика
config={"temperature": 0.7}
),
Вариант(
name="treatment-v1",
model_name="Qwen2.5-14B-Instruct",
weight=30, # 30% трафика
config={"temperature": 0.7}
),
Вариант(
name="treatment-v2",
model_name="Llama-3.1-8B-Instruct",
weight=20, # 20% трафика
config={"temperature": 0.8}
)
]

router = ABTestRouter("model-comparison-v1", variants)

# Разделение тестовых пользователей
test_users = [f"user_{i}" for i in range(10)]
for user_id in test_users:
variant = router.get_variant_by_user(user_id)
print(f"Пользователь {user_id} → Группа {variant.name} → Модель {variant.model_name}")

# Статистика распределения
count: Dict[str, int] = {}
for i in range(10000):
variant = router.get_variant_random()
count[variant.name] = count.get(variant.name, 0) + 1
print("\nСтатистика случайного разбиения (10000 раз):")
for name, cnt in count.items():
print(f" {name}: {cnt} раз ({cnt/100}%)")

if __name__ == "__main__":
main()

Определение и сбор метрик

A/B-тестирование не может просто смотреть на «какая модель отвечает умнее»; необходимо определить измеримые метрики.

Типы метрик	Специфические метрики	Описание
Системные метрики	Задержка, пропускная способность, частота ошибок	Является ли модель быстрой и стабильной
Поведение пользователей	Удовлетворённость пользователей, количество ходов диалога, коэффициент продления	Нравится ли это пользователям
Бизнес-метрики	Конверсия, удержание, выручка	Помогает ли это бизнесу
Метрики качества	Оценки людей, уровень вредности, фактическая точность	Хорошее ли качество вывода

Пример

# Путь к файлу: metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
"""Полные метрики для запросов чата"""
# Основная информация
request_id: str
user_id: str
variant_name: str # Название группы A/B-теста
model_name: str

# Метрики времени
timestamp: str
ttft: float # Время до первого токена
total_time: float # Общее затраченное время

# Метрики генерации
prompt_tokens: int
completion_tokens: int
total_tokens: int

# Бизнес-метрики
user_rating: Optional[int] = None # Оценка пользователя (1-5)
is_error: bool = False
error_message: Optional[str] = None

class MetricsCollector:
"""Сборщик метрик"""

def __init__(self):
# В реальном проекте это должно записываться в Kafka или базу данных
self.buffer = []

def collect(self, metrics: ChatMetrics):
"""Собрать одну метрику"""
self.buffer.append(asdict(metrics))
# Простой вывод; в реальном проекте — запись в хранилище
print(f"[Метрики] {json.dumps(asdict(metrics), ensure_ascii=False)}")

# Когда буфер заполнен, записывайте пакетами
if len(self.buffer) >= 100:
self.flush()

def flush(self):
"""Пакетная запись в хранилище"""
if self.buffer:
# В реальном проекте: запись в ClickHouse, BigQuery и т. д.
print(f"Сброшено {len(self.buffer)} метрик")
self.buffer = []

# Пример использования
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
"""Обработать запрос чата и записать метрики"""
request_id = str(uuid.uuid4())

# Разделить трафик
variant = router.get_variant_by_user(user_id)

# Записать время начала
start_time = time.time()
ttft = 0.0
prompt_tokens = len(prompt)
completion_tokens = 0

try:
# Замените это реальным вызовом модели
# response = call_model(variant.model_name, prompt)
response = "Это ответ ИИ"
completion_tokens = len(response)

# Симулировать TTFT (предположим, что первый токен получен через 0.2 seconds)
ttft = 0.2

# Записать метрики успеха
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
is_error=False
)
collector.collect(metrics)
return response, request_id

except Exception as e:
# Записать метрики ошибок
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=0,
total_tokens=prompt_tokens,
is_error=True,
error_message=str(e)
)
collector.collect(metrics)
raise

# Отзыв пользователя
def submit_user_feedback(request_id: str, rating: int):
"""Пользователь отправляет отзыв с оценкой"""
# В реальном проекте: обновите соответствующую запись в базе данных
print(f"Запрос {request_id} получил оценку {rating}")

Оценка статистической значимости

После получения данных нельзя просто посмотреть на 52% для группы A и 55% для группы B и заключить, что группа B лучше.

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

Пример

# Путь к файлу: ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
control_conv: float,
treatment_conv: float,
control_total: int,
treatment_total: int
) -> Tuple[float, float]:
"""
Вычислить p-значение Z-теста
control_conv: коэффициент конверсии контрольной группы
treatment_conv: коэффициент конверсии экспериментальной группы
control_total: размер выборки контрольной группы
treatment_total: размер выборки экспериментальной группы
"""
# объединённый коэффициент конверсии
p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
# стандартная ошибка
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
# Z-статистика
z_score = (treatment_conv - control_conv) / se_pool
# вычислить p-значение (двусторонний тест)
p_value = 2 * (1 - normal_cdf(abs(z_score)))
return z_score, p_value

def normal_cdf(x: float) -> float:
"""Кумулятивная функция распределения стандартного нормального распределения (приближённо)"""
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
control_conversions: int,
control_total: int,
treatment_conversions: int,
treatment_total: int,
alpha: float = 0.05
):
"""Проанализировать результаты A/B-теста"""
control_conv = control_conversions / control_total
treatment_conv = treatment_conversions / treatment_total

z_score, p_value = calculate_z_test(
control_conv, treatment_conv,
control_total, treatment_total
)

relative_lift = (treatment_conv - control_conv) / control_conv

print("Анализ результатов A/B-теста")
print("=" * 50)
print(f"Контрольная группа: {control_conversions}/{control_total} ({control_conv:.2%})")
print(f"Тестовая группа: {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
print(f"Относительный прирост: {relative_lift:+.2%}")
print("-" * 50)
print(f"Z-статистика: {z_score:.4f}")
print(f"p-значение: {p_value:.4f}")
print("-" * 50)

if p_value < alpha:
print(f"✅ Результат статистически значим (p < {alpha})")
if relative_lift > 0:
print(" Тестовая группа показывает лучший результат!")
else:
print(" Контрольная группа показывает лучший результат!")
else:
print(f"❌ Результат статистически не значим (p >= {alpha})")
print(" Невозможно определить, какая версия лучше; требуется больше данных")

# Пример: предположим, есть результаты A/B-теста
if __name__ == "__main__":
# Сценарий 1: значимое различие
print("Сценарий 1: значимое различие")
analyze_ab_test(
control_conversions=450, # Контрольная группа: 450 удовлетворённых
control_total=1000, # Всего в контрольной группе: 1000 человек
treatment_conversions=520, # Тестовая группа: 520 удовлетворённых
treatment_total=1000 # Всего в тестовой группе: 1000 человек
)

print("\n" + "="*50 + "\n")

# Сценарий 2: значимых различий нет
print("Сценарий 2: значимых различий нет")
analyze_ab_test(
control_conversions=48,
control_total=100,
treatment_conversions=52,
treatment_total=100
)

Статистическая значимость важна. Непроверенное "улучшение" может оказаться просто случайным колебанием.

Отслеживание экспериментов MLflow

Вы обучили десятки версий моделей. Какая из них показывает лучшие результаты? Какая версия развёрнута?

MLflow помогает управлять экспериментами, отслеживать модели и записывать метрики.

Управление экспериментами

Пример

# Путь к файлу: mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# Задайте URI отслеживания MLflow (локальный или удаленный сервер)
mlflow.set_tracking_uri("http://localhost:5000")

# Создать или установить эксперимент
experiment_name = "example-llm-finetuning"
try:
experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# Симулировать эксперимент по тонкой настройке LLM
def run_finetuning_experiment(
model_name: str,
learning_rate: float,
batch_size: int,
num_epochs: int,
lora_rank: int,
dataset_name: str = "example-demo"
):
"""Запустить эксперимент по тонкой настройке и залогировать его в MLflow"""

with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
run_id = run.info.run_id

# Записать параметры
mlflow.log_param("model_name", model_name)
mlflow.log_param("learning_rate", learning_rate)
mlflow.log_param("batch_size", batch_size)
mlflow.log_param("num_epochs", num_epochs)
mlflow.log_param("lora_rank", lora_rank)
mlflow.log_param("dataset_name", dataset_name)

# Симуляция процесса обучения и запись метрик
print(f"Запуск эксперимента по обучению {run_id}...")

for epoch in range(num_epochs):
# Симуляция потерь при обучении (уменьшается с эпохами)
train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

# Записать метрики
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("val_loss", val_loss, step=epoch)
mlflow.log_metric("epoch", epoch, step=epoch)

print(f"Эпоха {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
time.sleep(0.5)

# Итоговые метрики оценки
final_val_loss = 1.2 + np.random.normal(0, 0.1)
final_accuracy = 0.85 + np.random.normal(0, 0.03)
final_f1 = 0.83 + np.random.normal(0, 0.04)

mlflow.log_metric("final_val_loss", final_val_loss)
mlflow.log_metric("final_accuracy", final_accuracy)
mlflow.log_metric("final_f1", final_f1)

# Запись файлов (например, логи обучения, примеры выходных данных)
with open("training_log.txt", "w") as f:
f.write(f"run_id эксперимента: {run_id}"\n")
f.write(f"Модель: {model_name}"\n")
f.write(f"Итоговая точность: {final_accuracy:.4f}"\n")

mlflow.log_artifact("training_log.txt")

# Логирование модели (здесь имитируется с помощью sklearn; в реальных проектах логируйте настоящую LLM)
# В реальных проектах можно использовать mlflow.transformers.log_model
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

mlflow.sklearn.log_model(
model,
"model",
registered_model_name=f"{model_name.replace('/', '-')}-demo"
)

print(f"Эксперимент завершён! run_id: {run_id}")
print(f"Итоговая точность: {final_accuracy:.4f}")

return run_id, final_accuracy

# Запуск нескольких экспериментов
if __name__ == "__main__":
print("Запустите MLflow UI (выполните в другом терминале):")
print(" mlflow ui --port 5000")
print("\nЗапуск экспериментов...\n")

# Эксперимент 1: Базовая конфигурация
run_id1, acc1 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()

# Эксперимент 2: Больший ранг LoRA
run_id2, acc2 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=16
)

print()

# Эксперимент 3: Более высокая скорость обучения
run_id3, acc3 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=5e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()
print("Все эксперименты завершены!")
print(f"Эксперимент 1 (run_id={run_id1}): {acc1:.4f}")
print(f"Эксперимент 2 (run_id={run_id2}): {acc2:.4f}")
print(f"Эксперимент 3 (run_id={run_id3}): {acc3:.4f}")
print("\n"Просмотр результатов: http://localhost:5000")

Реестр моделей

MLflow Model Registry управляет жизненным циклом модели: от Staging (тестирование) до Production.

Пример

# Путь к файлу: mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
model_name: str,
version: int,
description: str = ""
):
"""Перевести модель на стадию Production"""
# Обновить описание версии
client.update_model_version(
name=model_name,
version=version,
description=description
)

# Установить эту версию как Production
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
archive_existing_versions=True # Архивировать старые версии
)
print(f"Модель {model_name} версии {version} была переведена на стадию Production")

def get_production_model(model_name: str):
"""Получить текущую модель Production"""
versions = client.get_latest_versions(model_name, stages=["Production"])
if versions:
return versions[0]
return None

def list_all_models():
"""Вывести список всех зарегистрированных моделей"""
models = client.search_registered_models()
print("Зарегистрированные модели:")
for model in models:
print(f" - {model.name}")
# Список всех версий этой модели
versions = client.search_model_versions(f"name='{model.name}'")
for v in versions:
print(f" Версия {v.version}: {v.current_stage}")

# Пример использования
if __name__ == "__main__":
model_name = "Qwen2.5-7B-Instruct-demo"

print("Список всех моделей:")
list_all_models()

print("\n" + "="*50)
# Предположим, мы хотим перевести версию 1 в продакшен
# promote_model_to_production(
# model_name=model_name,
# version=1,
# description="Модель для продакшен-среды: оптимизированная версия 2024"
# )

print("\nПолучить текущую продакшен-модель: ")
prod_model = get_production_model(model_name)
if prod_model:
print(f" Имя модели: {prod_model.name}")
print(f" Версия: {prod_model.version}")
print(f" Стадия: {prod_model.current_stage}")
print(f" ID запуска: {prod_model.run_id}")
else:
print(" Модели в продакшене пока нет")

Мониторинг и оповещения

После запуска сервиса как вы узнаете, что он работает корректно? Будут ли пользователи сталкиваться с тайм-аутами? Не станет ли модель со временем всё менее точной?

Вам нужна полноценная система мониторинга.

Ключевые метрики: TTFT, TPS, частота ошибок

У ИИ-сервисов есть несколько основных метрик мониторинга:

Метрика	Полное название	Значение	Пример порога срабатывания оповещения
TTFT	Время до первого токена	Время от запроса пользователя до получения первого токена	Оповещение > 3 секунд
TPOT	Время на выходной токен	Среднее время на выходной токен	Оповещение > 100ms
TPS	Токенов в секунду	Сколько токенов генерируется в секунду	Оповещение < 100
Частота ошибок	Частота ошибок	Доля неудачных запросов	Оповещение > 5%
Загрузка GPU	Загрузка GPU	Уровень использования GPU	Оповещение < 30% или > 95%
Использование VRAM	Память GPU	Уровень заполнения GPU VRAM	Оповещение > 90%

Prometheus + Grafana

Prometheus собирает метрики, Grafana отображает дашборды — золотая комбинация для мониторинга.

Пример

# Путь к файлу: metrics_prometheus.py
from prometheus_client import (
Counter, Histogram, Gauge,
start_http_server, generate_latest
)
import time
import random

# ============================================
# Определение метрик
# ============================================

# Counter: счётчик, который только увеличивается
REQUEST_COUNT = Counter(
"llm_requests_total",
"Общее количество запросов LLM",
["model", "variant", "status"] # Метки
)

# Histogram: статистическое распределение (например, задержка)
REQUEST_LATENCY = Histogram(
"llm_request_latency_seconds",
"Задержка запроса LLM",
["model", "variant"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60] # Бакеты
)

TTFT_HISTOGRAM = Histogram(
"llm_ttft_seconds",
"Время до первого токена",
["model", "variant"],
buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Gauge: значение, которое может увеличиваться или уменьшаться
ACTIVE_CONNECTIONS = Gauge(
"llm_active_connections",
"Количество активных подключений",
["model"]
)

GPU_UTILIZATION = Gauge(
"llm_gpu_utilization_percent",
"Использование GPU",
["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
"llm_gpu_memory_used_bytes",
"Использовано памяти GPU",
["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
"llm_tokens_per_second",
"Токенов, сгенерированных в секунду",
["model"]
)

# ============================================
# Симулировать сервис
# ============================================

def process_request(model: str, variant: str):
"""Симулировать обработку запроса и записывать метрики"""
start_time = time.time()

# Активные соединения +1
ACTIVE_CONNECTIONS.labels(model=model).inc()

try:
# Симулировать TTFT (время до первого токена)
ttft = random.uniform(0.05, 0.3)
TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

# Симулировать время обработки
processing_time = random.uniform(0.5, 3.0)
time.sleep(processing_time)

# Симулировать успех
REQUEST_COUNT.labels(
model=model, variant=variant, status="success"
).inc()

# Зафиксировать задержку
latency = time.time() - start_time
REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

# Симулировать TPS
tps = random.uniform(50, 200)
TOKENS_PER_SECOND.labels(model=model).set(tps)

except Exception:
# Имитировать ошибку
REQUEST_COUNT.labels(
model=model, variant=variant, status="error"
).inc()
raise
finally:
# Активные соединения -1
ACTIVE_CONNECTIONS.labels(model=model).dec()

def update_gpu_metrics():
"""Имитировать обновление метрик GPU"""
# Имитировать метрики для GPU 0
gpu0_util = random.uniform(40, 90)
gpu0_mem = random.uniform(10, 20) * 1024**3 # 10-20 GB
GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)

# ============================================
# Запустить сервис
# ============================================

if __name__ == "__main__":
# Запустить эндпоинт метрик Prometheus
start_http_server(8081)
print("Сервер метрик Prometheus запущен на порту 8081")
print("Метрики доступны по адресу http://localhost:8081/metrics")

# Симулировать трафик
models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
variants = ["control", "treatment-v1"]

while True:
model = random.choice(models)
variant = random.choice(variants)
process_request(model, variant)
update_gpu_metrics()
time.sleep(random.uniform(0.1, 0.5))

Файл конфигурации Prometheus:

Пример

# Путь к файлу: prometheus.yml
global:
scrape_interval: 15s

scrape_configs:
# Сбор метрик сервиса LLM
- job_name: "llm-service"
static_configs:
- targets: ["localhost:8081"]
labels:
service: "llm-api"
environment: "production"

# Сбор метрик узла (GPU, CPU, память)
- job_name: "node"
static_configs:
- targets: ["localhost:9100"]

alerting:
alertmanagers:
- static_configs:
- targets: ["localhost:9093"]

# Правила оповещения
rule_files:
- "alerts.yml"

Правила оповещений:

Пример

# Путь к файлу: alerts.yml
groups:
- name: llm_alerts
interval: 30s
rules:
# Оповещение о высокой частоте ошибок
- alert: HighErrorRate
expr: |
sum(rate(llm_requests_total{status="error"}[5m]))
/
sum(rate(llm_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Предупреждение о высокой частоте ошибок"
description: "Частота ошибок превышает 5%, текущее значение: {{ $value | humanizePercentage }}"

# Предупреждение о слишком высоком TTFT
- alert: SlowTTFT
expr: |
histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "TTFT слишком высок"
description: "95-й процентиль TTFT модели {{ $labels.model }} превышает 2 секунды"

# Загрузка GPU слишком низкая (расточительство ресурсов)
- alert: LowGPUUtilization
expr: llm_gpu_utilization_percent < 30
for: 30m
labels:
severity: info
annotations:
summary: "Загрузка GPU слишком низкая"
description: "Загрузка GPU {{ $labels.gpu_id }} — {{ $value }}%, возможно нерациональное использование ресурсов"

# Загрузка GPU слишком высокая (риск узкого места)
- alert: HighGPUUtilization
expr: llm_gpu_utilization_percent > 95
for: 5m
labels:
severity: warning
annotations:
summary: "Загрузка GPU слишком высокая"
description: "Загрузка GPU {{ $labels.gpu_id }} — {{ $value }}%, может стать узким местом"

Мониторинг, специфичный для ИИ: обнаружение дрейфа

Модели ИИ не только «ломаются», но и «дрейфуют»: со временем распределение данных меняется, и качество модели ухудшается.

Пример

# Путь к файлу: drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
"""Статистика распределения"""
mean: float
std: float
min: float
max: float
percentiles: dict

class DataDriftDetector:
"""Детектор дрейфа данных"""

def __init__(self, baseline_data: np.ndarray):
"""Инициализация с базовыми данными"""
self.baseline_stats = self._compute_stats(baseline_data)
self.baseline_data = baseline_data

def _compute_stats(self, data: np.ndarray) -> DistributionStats:
"""Вычисление статистики для данных"""
return DistributionStats(
mean=float(np.mean(data)),
std=float(np.std(data)),
min=float(np.min(data)),
max=float(np.max(data)),
percentiles={
"p10": float(np.percentile(data, 10)),
"p25": float(np.percentile(data, 25)),
"p50": float(np.percentile(data, 50)),
"p75": float(np.percentile(data, 75)),
"p90": float(np.percentile(data, 90))
}
)

def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""
Обнаружение дрейфа
Возвращает: наличие дрейфа, p-значение, статистику
"""
# KS-тест: сравнить, одинаковы ли два распределения
ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

# Вычислить статистику для новых данных
new_stats = self._compute_stats(new_data)

# Вычислить разницу средних (в процентах)
mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

# Определить, произошёл ли дрейф
is_drift = ks_pvalue < threshold

return {
"is_drift": is_drift,
"ks_statistic": float(ks_statistic),
"ks_pvalue": float(ks_pvalue),
"mean_diff_percent": float(mean_diff_pct),
"baseline_stats": self.baseline_stats,
"new_stats": new_stats,
"threshold": threshold
}

# Симуляция: обнаружение дрейфа для длины пользовательского ввода
def simulate_input_length_drift():
# Базовые данные: средняя длина пользовательского ввода 50
np.random.seed(42)
baseline = np.random.normal(loc=50, scale=15, size=1000)
baseline = np.clip(baseline, 5, 200)

detector = DataDriftDetector(baseline)

print("Пример обнаружения дрейфа данных")
print("="*50)
print(fБазовый уровень: среднее={detector.baseline_stats.mean:.2f}, стандартное отклонение={detector.baseline_stats.std:.2f})
print()

# Сценарий 1: Нормальные данные (без дрейфа)
print(Сценарий 1: Нормальные данные)
normal_data = np.random.normal(loc=50, scale=15, size=500)
normal_data = np.clip(normal_data, 5, 200)
result = detector.detect(normal_data)
print(fСреднее новых данных: {result['new_stats'].mean:.2f})
print(f" p-значение KS: {result['ks_pvalue']:.4f}")
print(fДрейф? {'Yes' if result['is_drift'] else 'No'})
print()

# Сценарий 2: Дрейф (входные данные стали длиннее)
print(Сценарий 2: Входные данные стали длиннее)
drift_data = np.random.normal(loc=100, scale=20, size=500)
drift_data = np.clip(drift_data, 5, 300)
result = detector.detect(drift_data)
print(fСреднее новых данных: {result['new_stats'].mean:.2f})
print(f" p-значение KS: {result['ks_pvalue']:.4f}")
print(fДрейф? {'Yes' if result['is_drift'] else 'No'})
print(fСредняя разница: {result['mean_diff_percent']:.1f}%)

if __name__ == "__main__":
simulate_input_length_drift()

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

GPU дороги. Оптимизация затрат — это не «быть дешёвым», а «инженерная способность».

Кэширование запросов: семантическое кэширование

Традиционное кэширование — это точное совпадение: для точно такого же вопроса сразу возвращается предыдущий ответ.

Семантическое кэширование идёт дальше: похожие вопросы также могут повторно использовать предыдущие ответы или промежуточные результаты.

Пример

# Путь к файлу: semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
"""Простая реализация семантического кэша"""

def __init__(self, threshold: float = 0.9):
self.threshold = threshold
self.queries = [] # Хранит векторы вопросов
self.answers = [] # Хранит соответствующие ответы
self.timestamps = [] # Хранит временные метки

def _simulate_embedding(self, text: str) -> np.ndarray:
"""
Имитация эмбеддинга; в реальных проектах используйте OpenAI Embeddings, Sentence-Transformers и т. д.
"""
# Простая имитация хеширования; в реальных проектах используйте настоящую модель эмбеддингов
np.random.seed(hash(text) % (2**32))
return np.random.randn(128) # 128-мерный вектор

def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Вычислить косинусное сходство"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get(self, query: str) -> Optional[str]:
"""Кэш запросов: если найден похожий вопрос, вернуть ответ"""
query_vec = self._simulate_embedding(query)

for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
# Очистка устаревших записей (более 1 часа)
if time.time() - ts > 3600:
continue

similarity = self._cosine_similarity(query_vec, q_vec)
if similarity >= self.threshold:
return answer

return None

def put(self, query: str, answer: str):
"""Сохранить в кэш"""
query_vec = self._simulate_embedding(query)
self.queries.append(query_vec)
self.answers.append(answer)
self.timestamps.append(time.time())

# Простой контроль вместимости: хранить не более 1000 записей
if len(self.queries) > 1000:
self.queries = self.queries[-1000:]
self.answers = self.answers[-1000:]
self.timestamps = self.timestamps[-1000:]

# Пример использования
def main():
cache = SemanticCache(threshold=0.9)

# Первый запрос
query1 = Как прочитать файл CSV в Python?
print(fЗапрос 1: {query1})
answer1 = cache.get(query1)
if answer1:
print(fПопадание в кэш: {answer1})
else:
print(Промах кэша, вызов модели...)
answer1 = Вы можете использовать pandas.read_csv('file.csv') для чтения файлов CSV.
cache.put(query1, answer1)
print(fСохранение в кэше: {answer1})

print()

# Второй раз: похожий вопрос
query2 = Как читать CSV в Python?
print(fЗапрос 2: {query2})
answer2 = cache.get(query2)
if answer2:
print(fПопадание в кэш: {answer2})
else:
print(Промах кэша, вызов модели...)

print()

# Третий раз: совершенно другой вопрос
query3 = Что такое машинное обучение?
print(fЗапрос 3: {query3})
answer3 = cache.get(query3)
if answer3:
print(fПопадание в кэш: {answer3})
else:
print(Промах кэша, вызов модели...)

if __name__ == "__main__":
main()

Маршрутизация моделей: Малая модель как резервная

Не все запросы нуждаются в большой модели. Используйте малую модель для простых вопросов и большую модель для сложных вопросов.

Пример

# Путь к файлу: model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float
max_tokens: int
capabilities: List[str] # Список возможностей

class SmartModelRouter:
"""Интеллектуальный маршрутизатор моделей"""

def __init__(self):
self.models = {
"small": ModelConfig(
name="Qwen2.5-0.5B-Instruct",
cost_per_1k_tokens=0.001,
max_tokens=4096,
capabilities=["chat", "simple_qa", "summarization"]
),
"medium": ModelConfig(
name="Qwen2.5-7B-Instruct",
cost_per_1k_tokens=0.01,
max_tokens=8192,
capabilities=["chat", "qa", "coding", "reasoning"]
),
"large": ModelConfig(
name="Qwen2.5-72B-Instruct",
cost_per_1k_tokens=0.1,
max_tokens=32768,
capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
)
}

def _classify_query(self, query: str) -> str:
"""
Классификация простых вопросов (в реальных проектах можно использовать классификатор)
Возврат: маленький/средний/большой
"""
query_lower = query.lower()

# Ключевые слова для простых вопросов
simple_keywords = [
"Здравствуйте", "здравствуйте", "Спасибо", "привет", "До свидания",
"Погода сегодня", "Который час", "Дата",
"Помоги перевести", "перевести",
"Краткое введение", "Что такое"
]

# Ключевые слова для сложных вопросов
complex_keywords = [
"Код", "Напиши", "Реализуй", "код", "python",
"Анализируй", "Объясни почему", "Выведи", "Докажи",
"Сравни", "Противопоставь", "Различие",
"Подробный", "Углублённый", "Сложный"
]

# Проверить, является ли вопрос простым
for keyword in simple_keywords:
if keyword in query_lower:
return "small"

# Проверить, является ли вопрос сложным
for keyword in complex_keywords:
if keyword in query_lower:
return "large"

# По умолчанию использовать среднюю модель
return "medium"

def route(self, query: str, user_tier: str = "free") -> str:
"""
Решение о маршрутизации
"""
# Сначала классифицировать вопрос
category = self._classify_query(query)

# Платные пользователи могут использовать более качественную модель
if user_tier == "pro":
upgrade_map = {
"small": "small",
"medium": "medium",
"large": "large"
}
category = upgrade_map[category]
elif user_tier == "enterprise":
upgrade_map = {
"small": "medium",
"medium": "large",
"large": "large"
}
category = upgrade_map.get(category, category)

# Вернуть имя модели
return self.models[category].name

def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Оценить стоимость"""
for model in self.models.values():
if model.name == model_name:
return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
return 0.0

# Пример использования
def main():
router = SmartModelRouter()

test_queries = [
"Здравствуйте, я хотел бы узнать больше",
"Что такое Python?",
"Помоги мне написать быструю сортировку на Python",
Предоставьте подробный анализ архитектуры Transformer.
]

print(Пример интеллектуальной маршрутизации)
print("="*50)

for query in test_queries:
model = router.route(query, user_tier="free")
cost = router.get_cost_estimate(model, 100, 200)
print(fЗапрос: {query})
print(fМаршрут к: {model})
print(fОценочная стоимость: ${cost:.4f})
print()

# Разные уровни пользователей
print(Сравнение различных уровней пользователей)
print("-"*50)
запрос = Помоги мне написать быструю сортировку на Python
for tier in ["free", "pro", "enterprise"]:
model = router.route(query, user_tier=tier)
print(f{tier} пользователь: {model})

if __name__ == "__main__":
main()

Спотовые инстансы и пакетная обработка

Используйте On-Demand инстансы для запросов в реальном времени (стабильные, но дорогие), а Spot инстансы — для офлайн-пакетной обработки (дешёвые, но могут быть прерваны).

Краткий обзор нескольких методов оптимизации затрат:

Стратегия	Сценарий	Экономия затрат
Семантический кэш	Много повторяющихся запросов	30-70%
Маршрутизация моделей	Высокая доля простых запросов	50-80%
Квантование модели	Все сценарии	30-50%
Спотовые инстансы	Офлайн-пакетная обработка	60-90%
Автоматическое масштабирование	Большие колебания трафика	30-60%

Обработка сбоев и деградация

В производственных средах не существует понятия «без сбоев» — есть только «как изящно обрабатывать сбои».

Стратегия деградации

Пример

# Путь к файлу: fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"

@dataclass
class FallbackResponse:
content: str
is_fallback: bool
fallback_reason: Optional[str] = None

⟦⟧FallbackHandler: FallbackHandler:
"""Обработчик резервного варианта"""

def __init__(self):
self.error = 0
self.last_error_time = 0
self.circuit_open = False
self.circuit_open_time = 0

# Конфигурация автоматического выключателя
self.error_threshold = 5 Срабатывание после 5 последовательных ошибок
self.circuit_timeout = 30 # Попытка восстановления после 30 секунд разомкнутой цепи

def _call_primary_service(self, query: str) -> str:
"""Имитировать вызов основного сервиса"""
# Замените это реальным вызовом модели
return "Это ответ основного сервиса"

def _call_fallback_service(self, query: str) -> str:
"""Имитировать вызов резервного сервиса (быстрее, но качество ниже)"""
# Это может быть меньшая модель, кэшированный ответ или даже предопределённый шаблон
return "Извините, сервис временно занят. Это упрощённый ответ."

def _call_static_fallback(self, query: str) -> str:
"""Статический резервный ответ (последняя линия защиты)"""
return (
"Мы очень сожалеем, наш сервис в настоящее время испытывает некоторые проблемы.\n"
"Пожалуйста, попробуйте позже. По срочным вопросам обращайтесь по адресу [email protected]."
)

def process_request(self, query: str) -> FallbackResponse:
Обрабатывать запросы с полной логикой резервного варианта

# Проверить, открыт ли автоматический выключатель
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_timeout:
# Истек тайм-аут, попробовать полуоткрытое состояние, пропустить небольшое количество запросов
print("Автоматический выключатель полуоткрыт, попытка восстановления...")
self.circuit_open = False
else:
# Перейти сразу к резервному варианту
print("Автоматический выключатель открыт, переход к резервному варианту напрямую")
return FallbackResponse(
content=self._call_fallback_service(query),
is_fallback=True,
fallback_reason="circuit_open"
)

# 2. Попробовать основной сервис
try:
result = self._call_primary_service(query)
# Успех: сбросить счетчик ошибок
self.error = 0
return FallbackResponse(
content=result,
is_fallback=False
)

except Exception as e:
print(f"Не удалось вызвать основной сервис: {e}")
self.error += 1
self.last_error_time = time.time()

# 3. Проверить, нужно ли размыкание цепи
if self.error_count >= self.error_threshold:
print("Слишком много ошибок, размыкаем цепь")
self.circuit_open = True
self.circuit_open_time = time.time()

# 4. Попробовать резервный сервис
try:
print("Попытка вызвать резервный сервис")
result = self._call_fallback_service(query)
return FallbackResponse(
content=result,
is_fallback=True,
fallback_reason="primary_failed"
)
except Exception as e2:
print(f"Резервный сервис также не сработал: {e2}")

# 5. Финальный статический ответ
return FallbackResponse(
content=self._call_static_fallback(query),
is_fallback=True,
fallback_reason="all_failed"
)

# Пример использования
def main():
handler = FallbackHandler()

print("Пример обработки резервного варианта")
print("="*50)

# Обычный запрос
print("Обычный запрос:")
resp = handler.process_request("Привет")
print(f" Ответ: {resp.content}")
print(f" Это резервный ответ? {resp.is_fallback}")
print()

# Имитация ошибки (в реальных проектах исключения действительно выбрасываются)
# Здесь мы просто демонстрируем логику
print("Сводка по стратегии резервирования:")
print(" 1. Предпочитать основной сервис")
print(" 2. Основной сервис дает сбой → использовать малую модель/кэш")
print(" 3. Последовательные сбои → размыкание цепи, чтобы избежать лавинообразного отказа")
print(" 4. Во время размыкания цепи → быстро вернуть резервный ответ")
print(" 5. Последняя линия защиты → статический ответ")

Канареечный релиз и откат

Не выкатывайте новую версию сразу всем пользователям. Сначала направьте на неё 1% трафика, и если проблем нет, то 5%, 10%, 100%.

Если возникнут проблемы, нужно иметь возможность быстро откатиться. Не ждите, пока пользователи будут долго жаловаться, прежде чем в панике менять конфигурацию.

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