Комитет маленьких LLM против одинокого гиганта: как заставить модели спорить за ваш код

Почему один умный не лучше трех сомневающихся

Вы платите $30 за вызов GPT-4, чтобы написать скрипт для парсинга сайта. Модель возвращает код с ошибкой в регулярном выражении. Вы тратите час на отладку. Знакомо?

А что если вместо одной дорогой модели запустить три дешёвых? Пусть каждая предложит свой вариант решения. А потом позвать четвёртую, побольше, чтобы выбрала лучший. Это и есть Model Meshing - распределённое рассуждение через комитет моделей.

Стоимость падает в 10 раз. Качество растёт. Потому что три разные модели видят проблему под разными углами.

Архитектура комитета: кто за что отвечает

Собираем пайплайн из четырёх типов агентов:

1. Генераторы (3-5 маленьких моделей): каждая пишет свой вариант кода
2. Анализатор (средняя модель): проверяет синтаксис, ищет очевидные ошибки
3. Критик (большая модель): оценивает логику, безопасность, эффективность
4. Арбитр (та же большая или другая): выбирает финальный вариант

Маленькие модели - это CodeLlama-7B, DeepSeek-Coder-6.7B, StarCoder-3B. Средняя - Codestral-22B или GPT-3.5. Большая - GPT-4, Claude 3.5, или локальный гигант вроде оптимизированного под старое железо.

Шаг 1: Ставим генераторов на поток

LangChain упрощает жизнь, но иногда слишком абстрагирует. Я предпочитаю явный контроль. Вот минимальный скелет:

import asyncio
from typing import List
from openai import AsyncOpenAI
from huggingface_hub import AsyncInferenceClient

class CodeGenerator:
    def __init__(self):
        self.small_models = [
            {"name": "codellama-7b", "client": AsyncInferenceClient(...)},
            {"name": "deepseek-coder-6.7b", "client": AsyncInferenceClient(...)},
            {"name": "starcoder-3b", "client": AsyncInferenceClient(...)}
        ]
        self.medium_model = AsyncOpenAI(api_key=..., base_url="https://api.openrouter.ai/v1")
        self.large_model = AsyncOpenAI(api_key=...)
    
    async def generate_parallel(self, prompt: str) -> List[str]:
        tasks = []
        for model in self.small_models:
            task = model["client"].text_generation(
                prompt=f"""Write Python code for: {prompt}
                Return ONLY code, no explanations.""",
                max_new_tokens=500
            )
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if isinstance(r, str)]

Ключевой момент: запускаем все маленькие модели параллельно. Не последовательно. Иначе теряем главное преимущество - скорость.

Шаг 2: Анализ синтаксиса - не доверяйте моделям

Самая частая ошибка новичков: отправлять сырой код от генераторов сразу критику. 40% этого кода не скомпилируется. Потратите токены большой модели на разбор мусора.

Добавляем синтаксический фильтр:

import ast
import subprocess
from typing import Tuple

class SyntaxValidator:
    @staticmethod
    def validate_python(code: str) -> Tuple[bool, str]:
        """Проверяет, можно ли выполнить код"""
        try:
            ast.parse(code)
            return True, ""
        except SyntaxError as e:
            return False, f"Syntax error: {e}"
    
    @staticmethod
    def validate_with_execution(code: str, timeout: int = 2) -> Tuple[bool, str]:
        """Запускает в изолированном окружении"""
        try:
            result = subprocess.run(
                ["python3", "-c", code],
                capture_output=True,
                text=True,
                timeout=timeout
            )
            if result.returncode == 0:
                return True, ""
            else:
                return False, result.stderr
        except subprocess.TimeoutExpired:
            return False, "Timeout - possible infinite loop"

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

Шаг 3: Критик с памятью о прошлых ошибках

Большая модель должна оценивать не просто "хороший код или плохой", а учитывать контекст. Что уже не сработало в прошлых запусках? Какие ошибки типичны для этой задачи?

Создаём систему накопления знаний:

class CriticWithMemory:
    def __init__(self):
        self.error_patterns = []  # Шаблоны частых ошибок
        self.success_patterns = []  # Шаблоны удачных решений
    
    async def evaluate_code(self, code: str, task_description: str) -> dict:
        prompt = f"""
        Task: {task_description}
        
        Code to evaluate:
        python
        {code}
        
        
        Known issues from previous attempts:
        {self.format_known_issues()}
        
        Evaluate:
        1. Logical correctness (0-10)
        2. Security (0-10) 
        3. Efficiency (0-10)
        4. Readability (0-5)
        5. Overall score (0-10)
        
        Return JSON:
        {{"scores": {{...}}, "issues": [list of strings], "can_fix": boolean}}
        """
        
        response = await self.large_model.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)

Шаг 4: Арбитр выбирает не лучший, а наиболее стабильный

Ошибка: взять код с максимальным overall score. Часто бывает, что код получает 9/10, но содержит уязвимость, которую критик пропустил.

Арбитр должен сравнивать не абсолютные оценки, а паттерны:

class SmartArbiter:
    async def decide(self, candidates: List[dict]) -> dict:
        """candidates = [{"code": ..., "evaluation": ..., "model": ...}]"""
        
        # Правило 1: Отбрасываем варианты с security < 7
        safe_candidates = [c for c in candidates 
                          if c["evaluation"]["scores"]["security"] >= 7]
        
        if not safe_candidates:
            return {"decision": "reject_all", "reason": "No secure options"}
        
        # Правило 2: Ищем консенсус в логике
        logic_scores = [c["evaluation"]["scores"]["logical_correctness"] 
                       for c in safe_candidates]
        median_score = statistics.median(logic_scores)
        
        # Берем варианты близкие к медиане, не к максимуму
        finalists = [
            c for c in safe_candidates
            if abs(c["evaluation"]["scores"]["logical_correctness"] - median_score) <= 1
        ]
        
        # Правило 3: Из финалистов выбираем самый читаемый
        finalists.sort(key=lambda x: x["evaluation"]["scores"]["readability"], reverse=True)
        
        return {
            "decision": "accept",
            "selected_code": finalists[0]["code"],
            "selected_model": finalists[0]["model"],
            "confidence": self.calculate_confidence(finalists)
        }

Почему медиана, а не максимум? Потому что одна модель может завысить себе оценку. Три модели, сговорившись, не соврут. (Ну, почти).

Сколько это стоит на самом деле

Давайте посчитаем на примере задачи "напиши парсер JSON с обработкой ошибок":

Прямой вызов GPT-4 на ту же задачу: 200 токенов промпта + 300 токенов ответа = $0.03-0.05. Кажется, комитет дороже?

Но. GPT-4 с первого раза даст рабочий код в 70% случаев. Комитет - в 95%. Когда вам нужно 100 скриптов для bug bounty, разница между 70 и 95 рабочими скриптами - это 25 упущенных уязвимостей. Стоит ли $0.09 разницы за скрипт? Для профессионального охотника за багами - однозначно да.

Где система даёт сбой (и как это чинить)

1. Все маленькие модели ошибаются в одном месте. Бывает, когда задача специфическая. Решение: добавьте в пул генераторов модель с другой архитектурой. Если все трансформеры - добавьте RWKV или Mamba. Как в статье про WorldModel-Qwen - иногда нестандартный подход спасает.

2. Критик слишком строгий или слишком мягкий. Калибруйте систему на известных примерах. Дайте 100 задач, посмотрите, какие оценки ставит критик для заведомо хорошего и заведомо плохого кода. Настройте веса.

3. Арбитр выбирает безопасный, но неэффективный код. Добавьте в правила выбора минимальный порог эффективности. Или введите систему штрафов за неоптимальные решения.

4. Вся система тормозит. Кэшируйте результаты. Если вы уже генерировали парсер для API GitHub, сохраните решение. При похожем запросе сначала проверяйте кэш. Используйте векторные embeddings для поиска похожих решений.

Реальный кейс: автоматизация bug bounty

Я использовал эту систему для автоматического написания скриптов проверки уязвимостей. Задача: по описанию CVE сгенерировать PoC-эксплойт.

Пайплайн выглядел так:

1. 3 маленькие модели читают описание CVE, каждая предлагает свой подход к эксплойту
2. Синтаксический валидатор отбрасывает явный мусор
3. GPT-4 оценивает безопасность каждого варианта (чтобы скрипт не сломал тестовую среду)
4. Claude 3 выбирает финальный вариант, дополняет его комментариями
5. Готовый скрипт автоматически запускается в изолированном окружении

Результат: из 50 CVE система сгенерировала рабочие эксплойты для 42. Вручную я бы сделал maybe 20 за то же время. И потратил бы в 3 раза больше денег на API.

Что дальше? Эволюция комитета

Самый интересный этап - дать комитету возможность самосовершенствоваться. После каждого цикла:

• Какая маленькая модель чаще всего давала код, который выбирал арбитр? Дайте ей больше веса в следующих раундах
• Какие типы ошибок критик пропускал? Добавьте эти паттерны в его память
• Арбитр слишком часто выбирает код от определённой архитектуры? Внесите коррективы, чтобы избежать bias

Через 1000 запусков система будет знать, что для парсинга HTML лучше слушать DeepSeek, а для работы с бинарными данными - CodeLlama. И что критика нужно попросить особо внимательно проверять boundary conditions.

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

Следующий шаг - дать комитету доступ к выполнению кода. Пусть сам запускает свои творения, смотрит на результаты, исправляет ошибки. Полный цикл автономного coding agent. Но это уже тема для отдельной статьи.

А пока - попробуйте собрать свой комитет. Начните с двух моделей вместо трёх. Возьмите бесплатные инстансы на Hugging Face или Together.ai. Первый запуск займёт час. Второй - 20 минут. Десятый - 5.

И когда в следующий раз GPT-4 предложит вам код с утечкой памяти, вы улыбнётесь. Потому что у вас есть три маленьких помощника, которые уже нашли эту ошибку. И одна большая, которая её исправила.