Оркестрация и роутинг моделей

Введение

При разработке систем с несколькими AI-агентами критически важно понимать, как эффективно распределить различные задачи между моделями с разными характеристиками. Использование универсальной модели для всех операций неоптимально с точки зрения стоимости и качества. В этом уроке мы изучим стратегии умного роутинга задач к подходящим моделям в зависимости от требований.

Концепция роутинга моделей

Почему нужен роутинг?

Современный ландшафт LLM предлагает модели с резко различающимися характеристиками:

Примеры трейдофф:

1. Простая классификация (spam vs not-spam) — 7B модель (0.1 токен/сек)
2. Написание кода — Qwen2.5-Coder 32B (специализированная)
3. Глубокий анализ текста — Llama 4 Maverick (максимальная точность)

Экономия ресурсов через роутинг

Без роутинга: 100% задач x 70B модель = 100 затрат
С роутингом:
  - 60% простых задач x 7B = 6 затрат
  - 30% средних задач x 32B = 30 затрат
  - 10% сложных задач x 70B = 10 затрат
  ИТОГО: 46 затрат (уменьшение на 54%)

Стратегии назначения моделей

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

graph LR
    REQ[Входящий запрос] --> CLS{Классификатор}
    CLS -- Простая задача --> S[7B модель]
    CLS -- Код --> M[32B Coder]
    CLS -- Сложный анализ --> L[70B модель]
    S --> RES[Результат]
    M --> RES
    L --> RES

    style REQ fill:#f8fafc,stroke:#e2e8f0
    style CLS fill:#4f46e5,color:#fff,stroke:#4338ca
    style S fill:#059669,color:#fff,stroke:#047857
    style M fill:#2563eb,color:#fff,stroke:#1d4ed8
    style L fill:#dc2626,color:#fff,stroke:#b91c1c
    style RES fill:#f8fafc,stroke:#e2e8f0

Роутинг по роли агента

Каждый агент в системе имеет определённую роль с соответствующей оптимальной моделью:

AGENT_ROLES_TO_MODELS = {
    'researcher': 'llama4:maverick',   # Глубокий анализ
    'coder': 'qwen2.5-coder:32b',      # Специализированный код
    'writer': 'llama4:scout',          # Общий текст
    'reviewer': 'llama4:scout',        # Быстрая проверка
    'translator': 'qwen2.5-32b',       # Многоязычность
}

Роутинг по сложности задачи

Анализируем сложность и выбираем модель:

def estimate_task_complexity(task_description: str) -> str:
    """Оценивает сложность по ключевым словам"""
    task_lower = task_description.lower()

    # Простые задачи
    simple_keywords = ['проверить', 'класс', 'спам', 'формат', 'структура']
    if any(kw in task_lower for kw in simple_keywords):
        return 'simple'  # 7B модель

    # Средние задачи
    medium_keywords = ['написать', 'улучшить', 'рефакторинг', 'отладить']
    if any(kw in task_lower for kw in medium_keywords):
        return 'medium'  # 32B модель

    # Сложные задачи
    complex_keywords = ['анализ', 'стратегия', 'дизайн', 'архитектура', 'исследование']
    if any(kw in task_lower for kw in complex_keywords):
        return 'complex'  # 70B модель

    return 'medium'  # По умолчанию

Роутинг по типу контента

CONTENT_TYPE_ROUTING = {
    'code_generation': 'qwen2.5-coder:32b',
    'code_review': 'qwen2.5-coder:32b',
    'documentation': 'llama4:scout',
    'analysis': 'llama4:maverick',
    'translation': 'qwen2.5-multilang-32b',
    'summarization': 'llama4:scout',
    'brainstorming': 'llama4:maverick',
    'validation': 'llama4:scout',
}

def route_by_content_type(task_type: str) -> str:
    return CONTENT_TYPE_ROUTING.get(task_type, 'llama4:scout')

Каскадный роутинг (Fallback)

Стратегия “try cheap, escalate if needed”:

graph TD
    T[Задача] --> M1[7B модель -- дешёвая]
    M1 --> Q1{Качество OK?}
    Q1 -- Да --> R[Результат]
    Q1 -- Нет --> M2[13B модель -- средняя]
    M2 --> Q2{Качество OK?}
    Q2 -- Да --> R
    Q2 -- Нет --> M3[70B модель -- мощная]
    M3 --> Q3{Качество OK?}
    Q3 -- Да --> R
    Q3 -- Нет --> E[Ошибка -- все попытки исчерпаны]

    style T fill:#f8fafc,stroke:#e2e8f0
    style M1 fill:#059669,color:#fff,stroke:#047857
    style M2 fill:#2563eb,color:#fff,stroke:#1d4ed8
    style M3 fill:#dc2626,color:#fff,stroke:#b91c1c
    style R fill:#059669,color:#fff,stroke:#047857
    style E fill:#dc2626,color:#fff,stroke:#b91c1c

class CascadingRouter:
    """Пытается сначала дешёвую модель, переходит к более мощной при необходимости"""

    ESCALATION_CHAIN = [
        ('llama4:scout', 0.3),     # Первая попытка (стоимость 0.3)
        ('qwen2.5-coder:32b', 0.6),# Вторая попытка (0.6)
        ('llama4:maverick', 1.0),  # Финальная попытка (1.0)
    ]

    def __init__(self):
        self.quality_threshold = 0.7  # Минимальный приемлемый уровень качества

    def route_with_escalation(self, task: str, max_attempts: int = 3) -> dict:
        """Роутирует задачу с возможностью эскалации"""
        for attempt, (model, cost) in enumerate(self.ESCALATION_CHAIN[:max_attempts]):
            print(f"Попытка {attempt + 1}: используем {model}")

            result = self.execute_with_model(task, model)
            quality_score = self.evaluate_quality(result)

            if quality_score >= self.quality_threshold:
                return {
                    'result': result,
                    'model_used': model,
                    'quality_score': quality_score,
                    'cost': cost,
                    'attempts': attempt + 1
                }

        # Если даже сильная модель не помогла
        return {
            'result': None,
            'model_used': 'llama4:maverick',
            'quality_score': 0.0,
            'error': 'Failed after all escalation attempts'
        }

    def evaluate_quality(self, result: str) -> float:
        """Простая оценка качества по различным критериям"""
        # Можно улучшить: использовать отдельную модель для оценки
        if not result or len(result) < 10:
            return 0.2
        if '```' in result and 'error' not in result.lower():
            return 0.9
        return 0.7

    def execute_with_model(self, task: str, model: str) -> str:
        """Выполняет задачу с указанной моделью"""
        # Здесь будет вызов через OpenAI-compatible API
        pass

Реализация на Python

Базовый Router класс

import json
import logging
from typing import Optional, Dict, Any
from enum import Enum

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ModelSize(Enum):
    TINY = "7b"
    SMALL = "13b"
    MEDIUM = "32b"
    LARGE = "70b"

class ModelRouter:
    """Роутер для выбора оптимальной модели на основе типа задачи"""

    MODEL_MAPPING = {
        'simple_validation': 'llama4:scout',
        'general_text': 'llama4:scout',
        'code_generation': 'qwen2.5-coder:32b',
        'deep_analysis': 'llama4:maverick',
    }

    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url
        self.default_model = 'llama4:scout'
        self.request_log = []

    def select_model(self, task_type: str, complexity: Optional[str] = None) -> str:
        """Выбирает модель на основе типа задачи и сложности"""
        selected = self.MODEL_MAPPING.get(task_type, self.default_model)

        logger.info(f"Selected model '{selected}' for task type '{task_type}'")
        return selected

    def log_request(self, task: str, model: str, tokens_used: int, cost: float):
        """Логирует запрос для анализа и отладки"""
        self.request_log.append({
            'task': task[:50] + '...' if len(task) > 50 else task,
            'model': model,
            'tokens': tokens_used,
            'cost': cost
        })

    def get_statistics(self) -> Dict[str, Any]:
        """Возвращает статистику использования моделей"""
        if not self.request_log:
            return {}

        total_tokens = sum(r['tokens'] for r in self.request_log)
        total_cost = sum(r['cost'] for r in self.request_log)
        model_usage = {}

        for request in self.request_log:
            model = request['model']
            if model not in model_usage:
                model_usage[model] = {'count': 0, 'tokens': 0}
            model_usage[model]['count'] += 1
            model_usage[model]['tokens'] += request['tokens']

        return {
            'total_requests': len(self.request_log),
            'total_tokens': total_tokens,
            'total_cost': total_cost,
            'model_usage': model_usage
        }

OpenAI-compatible агент

import asyncio
from openai import AsyncOpenAI

class LlamaServerAgent:
    """Агент, работающий с локальным llama-server через OpenAI API"""

    def __init__(self,
                 agent_id: str,
                 model: str = "llama4:scout",
                 base_url: str = "http://localhost:8000/v1",
                 router: Optional[ModelRouter] = None):
        self.agent_id = agent_id
        self.model = model
        self.client = AsyncOpenAI(
            api_key="not-needed",  # llama-server не требует ключа
            base_url=base_url
        )
        self.router = router or ModelRouter()
        self.messages_history = []

    async def execute(self, task: str, task_type: str = "general_text") -> str:
        """Выполняет задачу с автоматическим выбором модели"""
        selected_model = self.router.select_model(task_type)
        self.model = selected_model

        self.messages_history.append({
            "role": "user",
            "content": task
        })

        try:
            response = await self.client.chat.completions.create(
                model=selected_model,
                messages=self.messages_history,
                temperature=0.7,
                max_tokens=1000
            )

            result = response.choices[0].message.content
            self.messages_history.append({
                "role": "assistant",
                "content": result
            })

            # Логирование
            tokens_used = response.usage.completion_tokens
            cost = self._estimate_cost(tokens_used, selected_model)
            self.router.log_request(task, selected_model, tokens_used, cost)

            logger.info(f"[{self.agent_id}] Executed with {selected_model}: "
                       f"{tokens_used} tokens, ${cost:.4f}")

            return result

        except Exception as e:
            logger.error(f"[{self.agent_id}] Error: {e}")
            raise

    def _estimate_cost(self, tokens: int, model: str) -> float:
        """Оценивает стоимость для локальных моделей (обычно 0, но для трекинга)"""
        # Для локальных моделей стоимость 0, но полезно отслеживать использование
        return 0.0

    def clear_history(self):
        """Очищает историю сообщений"""
        self.messages_history = []

Смена моделей в llama-server

#!/bin/bash
# switch-model.sh - Скрипт для смены модели на запущенном llama-server

MODEL_NAME=${1:-"llama4:maverick"}
OLLAMA_HOST="localhost"
OLLAMA_PORT="11434"

echo "Switching to model: $MODEL_NAME"

# Загружаем модель через Ollama API (router mode llama.cpp: /models/load)
curl -X POST "http://${OLLAMA_HOST}:${OLLAMA_PORT}/api/pull" \
  -H "Content-Type: application/json" \
  -d "{\"name\":\"$MODEL_NAME\"}"

# Проверяем список загруженных моделей
sleep 2
curl -s "http://${OLLAMA_HOST}:${OLLAMA_PORT}/api/tags" | jq '.models[].name'

Практический пример: команда из 3 агентов

Определение ролей и моделей

from dataclasses import dataclass
from typing import List

@dataclass
class AgentProfile:
    """Профиль агента с его ролью и предпочтениями"""
    name: str
    role: str
    primary_model: str
    fallback_model: str
    task_types: List[str]
    max_tokens: int = 2000

# Определяем команду из 3 агентов
AGENT_TEAM = {
    'researcher': AgentProfile(
        name='Researcher',
        role='research_analyst',
        primary_model='llama4:maverick',  # Мощная модель для анализа
        fallback_model='llama4:scout',
        task_types=['analysis', 'research', 'investigation'],
        max_tokens=3000
    ),
    'coder': AgentProfile(
        name='CodeExpert',
        role='software_engineer',
        primary_model='qwen2.5-coder:32b',  # Специализированная для кода
        fallback_model='llama4:scout',
        task_types=['code_generation', 'code_review', 'debugging'],
        max_tokens=2500
    ),
    'reviewer': AgentProfile(
        name='Reviewer',
        role='quality_assurance',
        primary_model='llama4:scout',    # Быстрая проверка
        fallback_model='llama4:scout',
        task_types=['validation', 'review', 'summary'],
        max_tokens=1000
    )
}

Оркестратор команды

from typing import Dict

class AgentOrchestrator:
    """Оркестрирует команду агентов с разными моделями"""

    def __init__(self, agents_config: Dict[str, AgentProfile]):
        self.agents = {}
        self.router = ModelRouter()
        self.execution_log = []

        # Инициализируем агентов
        for agent_id, profile in agents_config.items():
            self.agents[agent_id] = {
                'profile': profile,
                'agent': LlamaServerAgent(
                    agent_id=agent_id,
                    model=profile.primary_model,
                    router=self.router
                ),
                'executions': 0
            }

    async def dispatch_task(self,
                          task: str,
                          agent_id: str,
                          task_type: str) -> Dict[str, Any]:
        """Отправляет задачу конкретному агенту"""
        if agent_id not in self.agents:
            raise ValueError(f"Unknown agent: {agent_id}")

        agent_data = self.agents[agent_id]
        profile = agent_data['profile']
        agent = agent_data['agent']

        logger.info(f"Dispatching task to {agent_id} "
                   f"(role: {profile.role}, type: {task_type})")

        result = await agent.execute(task, task_type)

        # Логируем выполнение
        agent_data['executions'] += 1
        self.execution_log.append({
            'agent': agent_id,
            'task_type': task_type,
            'model': agent.model,
            'success': True
        })

        return {
            'agent_id': agent_id,
            'role': profile.role,
            'model_used': agent.model,
            'result': result,
            'execution_count': agent_data['executions']
        }

    async def process_workflow(self, workflow: List[Dict[str, str]]) -> List[Dict]:
        """Обрабатывает рабочий процесс с несколькими задачами"""
        results = []
        for step in workflow:
            result = await self.dispatch_task(
                task=step['task'],
                agent_id=step['agent'],
                task_type=step.get('type', 'general_text')
            )
            results.append(result)
            # Даём время на обработку
            await asyncio.sleep(0.5)

        return results

    def get_team_statistics(self) -> Dict[str, Any]:
        """Возвращает статистику по команде"""
        stats = {
            'agents': {},
            'router_stats': self.router.get_statistics()
        }

        for agent_id, agent_data in self.agents.items():
            stats['agents'][agent_id] = {
                'role': agent_data['profile'].role,
                'model': agent_data['profile'].primary_model,
                'executions': agent_data['executions']
            }

        return stats

Пример использования

async def main():
    """Основная функция для демонстрации оркестрации"""

    # Инициализируем оркестратор с командой
    orchestrator = AgentOrchestrator(AGENT_TEAM)

    # Определяем рабочий процесс
    workflow = [
        {
            'agent': 'researcher',
            'task': 'Проанализируй тренды в разработке Python за 2024 год',
            'type': 'analysis'
        },
        {
            'agent': 'coder',
            'task': 'Напиши асинхронный вебхук на Python для обработки Github events',
            'type': 'code_generation'
        },
        {
            'agent': 'reviewer',
            'task': 'Проверь приведённый код на наличие основных ошибок',
            'type': 'validation'
        }
    ]

    # Выполняем рабочий процесс
    print("=" * 60)
    print("ЗАПУСК ОРКЕСТРАЦИИ КОМАНДЫ АГЕНТОВ")
    print("=" * 60)

    results = await orchestrator.process_workflow(workflow)

    # Выводим результаты
    for i, result in enumerate(results, 1):
        print(f"\n[ЭТАП {i}] Агент: {result['agent_id']}")
        print(f"Роль: {result['role']}")
        print(f"Модель: {result['model_used']}")
        print(f"Результат (первые 200 символов):")
        print(result['result'][:200] + "...")

    # Статистика
    print("\n" + "=" * 60)
    print("СТАТИСТИКА ВЫПОЛНЕНИЯ")
    print("=" * 60)
    stats = orchestrator.get_team_statistics()
    print(json.dumps(stats, indent=2, ensure_ascii=False))

Настройка llama-server для multi-model

Запуск нескольких экземпляров на разных портах

#!/bin/bash
# start-llama-servers.sh - Запускает несколько llama-server на разных портах

# Один экземпляр Ollama обслуживает все модели.
# Порт задаётся через переменную OLLAMA_HOST.
OLLAMA_HOST=127.0.0.1:11434 nohup ollama serve &
OLLAMA_PID=$!
echo "Ollama server started (PID: $OLLAMA_PID) on port 11434"

# Ждём запуска сервера
sleep 2

# Загружаем модели для каждого агента


echo "All models pulled. Ollama PID: $OLLAMA_PID"

# Для multi-model routing используйте параметр model в запросе:
# curl http://127.0.0.1:11434/api/chat -d '{"model":"llama4:70b","messages":[...]}'
# curl http://127.0.0.1:11434/api/chat -d '{"model":"qwen2.5-coder:32b","messages":[...]}'

Смена моделей на одном экземпляре

#!/bin/bash
# rotate-models.sh - Ротирует модели на одном сервере

MODELS=("llama4:8b" "llama4:13b" "qwen2.5-coder:32b" "llama4:70b")
PORT=${1:-8000}

echo "Starting model rotation on port $PORT..."

for model in "${MODELS[@]}"; do
  echo "Loading model: $model"

  # Загружаем модель через ollama CLI
  ollama pull "$model"

  # Указываем серверу использовать эту модель
  # (это зависит от реализации сервера)

  sleep 10  # Даём время на загрузку
done

echo "Model rotation completed"

nginx load balancer конфиг

Схема инфраструктуры с nginx-балансировкой между серверами моделей:

graph TD
    C[Клиент] --> N[nginx :8080]
    N --> S1[llama-server :8000 -- 70B Researcher]
    N --> S2[llama-server :8001 -- 32B Coder]
    N --> S3[llama-server :8002 -- 7B Reviewer]

    style C fill:#f8fafc,stroke:#e2e8f0
    style N fill:#4f46e5,color:#fff,stroke:#4338ca
    style S1 fill:#dc2626,color:#fff,stroke:#b91c1c
    style S2 fill:#2563eb,color:#fff,stroke:#1d4ed8
    style S3 fill:#059669,color:#fff,stroke:#047857

# /etc/nginx/conf.d/llama-lb.conf
# Балансировка нагрузки для llama-server инстансов

upstream llama_backends {
    server 127.0.0.1:8000 weight=3;  # Researcher (мощный)
    server 127.0.0.1:8001 weight=2;  # Coder
    server 127.0.0.1:8002 weight=1;  # Reviewer
}

server {
    listen 8080;
    server_name localhost;

    location /v1/ {
        proxy_pass http://llama_backends;
        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_set_header X-Forwarded-Proto $scheme;

        # Таймауты для длительных запросов
        proxy_connect_timeout 60s;
        proxy_send_timeout 300s;
        proxy_read_timeout 300s;
    }

    location /health {
        access_log off;
        return 200 "healthy\n";
        add_header Content-Type text/plain;
    }
}

Мониторинг и отладка

Логирование взаимодействий агентов

import logging
from datetime import datetime
import json

class AgentInteractionLogger:
    """Логирует взаимодействия агентов для отладки и анализа"""

    def __init__(self, log_file: str = "agent_interactions.jsonl"):
        self.log_file = log_file
        self.logger = logging.getLogger("agent_interactions")

        # Создаём обработчик для JSON логов
        handler = logging.FileHandler(log_file)
        handler.setFormatter(logging.Formatter('%(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_interaction(self,
                       agent_id: str,
                       task: str,
                       model: str,
                       result: str,
                       tokens: int,
                       duration: float):
        """Логирует одно взаимодействие агента"""

        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'agent_id': agent_id,
            'task': task[:100],  # Первые 100 символов
            'model': model,
            'tokens_used': tokens,
            'duration_seconds': duration,
            'result_length': len(result),
            'result_preview': result[:200]
        }

        self.logger.info(json.dumps(log_entry, ensure_ascii=False))

    def analyze_performance(self) -> Dict[str, Any]:
        """Анализирует логи для выявления паттернов"""

        with open(self.log_file, 'r') as f:
            entries = [json.loads(line) for line in f]

        if not entries:
            return {'error': 'No logs found'}

        analysis = {
            'total_interactions': len(entries),
            'avg_duration': sum(e['duration_seconds'] for e in entries) / len(entries),
            'total_tokens': sum(e['tokens_used'] for e in entries),
            'agents': {},
            'models': {}
        }

        # Агрегируем по агентам
        for entry in entries:
            agent = entry['agent_id']
            if agent not in analysis['agents']:
                analysis['agents'][agent] = {'count': 0, 'tokens': 0}
            analysis['agents'][agent]['count'] += 1
            analysis['agents'][agent]['tokens'] += entry['tokens_used']

            # Агрегируем по моделям
            model = entry['model']
            if model not in analysis['models']:
                analysis['models'][model] = {'count': 0, 'tokens': 0}
            analysis['models'][model]['count'] += 1
            analysis['models'][model]['tokens'] += entry['tokens_used']

        return analysis

Отслеживание стоимости и токенов

class TokenAndCostTracker:
    """Отслеживает использование токенов и стоимость (для облачных API)"""

    # Примерная стоимость токенов (при использовании облачных API)
    COST_PER_1K_TOKENS = {
        'llama4:8b': 0.00075,      # Дешёво
        'llama4:13b': 0.00125,
        'qwen2.5-coder:32b': 0.002,
        'llama4:70b': 0.00375,     # Дорого
    }

    def __init__(self):
        self.tracking_data = {}

    def track_usage(self, agent_id: str, model: str, tokens: int):
        """Отслеживает использование модели"""

        if agent_id not in self.tracking_data:
            self.tracking_data[agent_id] = {
                'total_tokens': 0,
                'total_cost': 0.0,
                'model_switches': 0
            }

        self.tracking_data[agent_id]['total_tokens'] += tokens

        # Для локальных моделей стоимость = 0, но отслеживаем для анализа
        if model in self.COST_PER_1K_TOKENS:
            cost = (tokens / 1000) * self.COST_PER_1K_TOKENS[model]
            self.tracking_data[agent_id]['total_cost'] += cost

    def get_cost_summary(self) -> Dict[str, Any]:
        """Возвращает сводку по стоимости"""

        total_cost = sum(d['total_cost'] for d in self.tracking_data.values())
        total_tokens = sum(d['total_tokens'] for d in self.tracking_data.values())

        return {
            'total_cost': total_cost,
            'total_tokens': total_tokens,
            'cost_per_1k_tokens': (total_cost / (total_tokens / 1000)) if total_tokens > 0 else 0,
            'by_agent': self.tracking_data
        }

    def print_cost_report(self):
        """Выводит красивый отчёт о стоимости"""
        summary = self.get_cost_summary()

        print("\n" + "=" * 60)
        print("ОТЧЁТ ПО ИСПОЛЬЗОВАНИЮ ТОКЕНОВ И СТОИМОСТИ")
        print("=" * 60)
        print(f"Всего токенов: {summary['total_tokens']:,}")
        print(f"Общая стоимость: ${summary['total_cost']:.4f}")
        print(f"Средняя стоимость за 1K токенов: ${summary['cost_per_1k_tokens']:.6f}\n")

        print("По агентам:")
        for agent_id, data in summary['by_agent'].items():
            print(f"  {agent_id}:")
            print(f"    Токенов: {data['total_tokens']:,}")
            print(f"    Стоимость: ${data['total_cost']:.4f}")

Частые проблемы и решения

class CommonFailureModes:
    """Документирует частые проблемы и решения"""

    FAILURES = {
        'model_not_found': {
            'symptom': 'HTTP 404 при запросе к модели',
            'causes': [
                'Модель не загружена на сервер',
                'Неправильное имя модели',
                'llama-server не запущен'
            ],
            'solutions': [
                'ollama pull <model-name>',
                'Проверить наличие модели: ollama list',
                'Перезапустить сервер'
            ]
        },
        'timeout': {
            'symptom': 'Запрос висит или timeout после 30 сек',
            'causes': [
                'Модель требует больше памяти, чем доступно',
                'Слишком много параллельных запросов',
                'Сервер перегружен'
            ],
            'solutions': [
                'Использовать меньшую модель',
                'Добавить параллелизм запросов',
                'Увеличить timeout в клиенте',
                'Добавить больше вычислительных ресурсов'
            ]
        },
        'quality_degradation': {
            'symptom': 'Модель выдаёт низкокачественные ответы',
            'causes': [
                'Параметры temperature слишком высокие',
                'Модель слишком маленькая для задачи',
                'Неправильный prompt'
            ],
            'solutions': [
                'Уменьшить temperature (0.3-0.5)',
                'Использовать каскадный роутинг',
                'Улучшить prompt engineering',
                'Использовать более мощную модель'
            ]
        },
        'memory_overflow': {
            'symptom': 'Процесс убивается (Out of Memory)',
            'causes': [
                '70B модель требует 140+ GB памяти',
                'Накопление истории сообщений',
                'Утечки памяти'
            ],
            'solutions': [
                'Использовать quantized модели (Q4, Q5)',
                'Очищать историю сообщений регулярно',
                'Мониторить использование памяти'
            ]
        }
    }

    @staticmethod
    def diagnose(error_message: str) -> Optional[Dict]:
        """Диагностирует проблему по ошибке"""
        for failure_type, info in CommonFailureModes.FAILURES.items():
            if any(cause_keyword in error_message.lower()
                  for cause_keyword in info['causes']):
                return info
        return None

    @staticmethod
    def print_troubleshooting_guide():
        """Выводит полное руководство по отладке"""
        print("\n" + "=" * 60)
        print("РУКОВОДСТВО ПО УСТРАНЕНИЮ НЕПОЛАДОК")
        print("=" * 60)

        for failure_type, info in CommonFailureModes.FAILURES.items():
            print(f"\n[{failure_type.upper()}]")
            print(f"Симптом: {info['symptom']}")
            print("Причины:")
            for cause in info['causes']:
                print(f"  - {cause}")
            print("Решения:")
            for solution in info['solutions']:
                print(f"  - {solution}")

Итоги

В этом уроке мы изучили:

1. Концепции роутинга: почему использование разных моделей для разных задач экономит ресурсы и улучшает качество

2. 4 стратегии распределения:

   • По ролям агента
   • По сложности задачи
   • По типу контента
   • Каскадный роутинг с fallback

3. Практическую реализацию:

   • Классы для выбора моделей
   • OpenAI-compatible агенты с локальным llama-server
   • Оркестратор для координации команды

4. Инфраструктуру:

   • Запуск нескольких llama-server на разных портах
   • Конфигурация nginx для балансировки
   • Ротация моделей

5. Мониторинг и отладку:

   • Логирование взаимодействий
   • Отслеживание токенов и стоимости
   • Диагностика частых проблем

Интеллектуальный роутинг задач к оптимальным моделям может уменьшить стоимость на 50% при сохранении или улучшении качества результатов.