Как построить test-time compute pipeline для Qwen3-14B: увеличение производительности для кодирования

Зачем вообще париться с pipeline? Потому что ваш Qwen3 — не гений, а ленивый студент

Допустим, вы скачали Qwen3-14B. Запустили. Написали "напиши функцию сортировки". И получили код, который сломает продакшен в пятницу вечером. Знакомо? Модель в 14 миллиардов параметров — это не GPT-5 Turbo. Это умный, но не всегда старательный помощник. Он может решить задачу, а может сгенерировать красивый, но нефункциональный мусор.

Frontier-модели в облаке делают одно волшебство: они не просто генерируют ответ, они его проверяют. Несколько раз. С разными подходами. И выбирают лучший. Это и есть test-time compute (TTC) — трата дополнительных вычислительных ресурсов не на обучение, а на использование модели для повышения качества ответа.

Хорошая новость: этот подход можно украсть. И заставить вашу локальную Qwen3-14B работать на уровне моделей в 10 раз больше. Плохая новость: нужно строить pipeline. Не страшно. Давайте разберем, как один студент с MacBook M3 Pro и 36 ГБ памяти обогнал по LiveCodeBench некоторые продакшен-решения на API.

Из чего состоит наш pipeline? Три кита, которые выдержат любой хлипкий код

Правильный pipeline для кодирования на локальной LLM строится на трех идеях, актуальных даже в 2026 году.

• Множественная генерация с верификацией (Energy-based verification). Мы заставляем модель сгенерировать не один ответ, а несколько (скажем, 5-10). Потом запускаем легковесный "верификатор" (часто это та же модель, но с другим промптом), который оценивает каждый вариант и выбирает лучший. Это похоже на то, как вы перечитываете свое письмо перед отправкой.
• Внешнее исполнение и тестирование (LiveCodeBench-подход). Самый честный способ проверить код — запустить его. Мы автоматически компилируем или интерпретируем сгенерированный код на изолированном стенде (Docker, sandbox) против набора юнит-тестов. Если код проходит — он хорош. Нет — отбраковываем.
• Динамическое управление контекстом и "зацикливанием" (Halting Problem). LLM могут генерировать бесконечный код или уходить в тангенсы. Наш pipeline должен детектировать это и останавливать генерацию, возможно, с переходом на другой метод решения.

Объединяем это в последовательность, которая превращает сырую генерацию в отлаженный результат. Звучит сложно? На практике это 200-300 строк Python-кода.

1 Готовим железо и софт: не пытайтесь сделать это на Colab

Вам нужно место, где модель будет работать стабильно и долго. Идеально — машина с 32+ ГБ ОЗУ и быстрым SSD. Я использовал MacBook Pro с M3 Max и 48 ГБ, но M4 Pro на 64 ГБ справится еще лучше. На Linux с NVIDIA GPU можно выжать больше скорости через CUDA.

Софт:

# Устанавливаем llama.cpp - самый актуальный форк на март 2026
# Он поддерживает Qwen3, IQ2 квантование и энергоэффективный инференс
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp && make clean && LLAMA_CUBLAS=1 make -j
# Для Metal на Mac: LLAMA_METAL=1 make -j

# Python-окружение с нужными библиотеками
python -m venv venv_ttc
source venv_ttc/bin/activate  # или venv_ttc\Scripts\activate на Windows
pip install --upgrade pip
pip install transformers>=4.45.0 torch>=2.4.0 docker fastapi uvicorn
# Для sandbox-тестирования кода
pip install code-eval>=2.0.0

2 Качаем и квантуем модель: выбираем баланс скорость/качество

Берем Qwen3-14B-Coder, а не базовую Qwen3-14B. Она специально доучивалась на кодe. На Hugging Face (на март 2026) актуальная версия — Qwen3-Coder-14B-Instruct. Качаем оригинал в формате safetensors.

Но весит она около 28 ГБ в FP16. Нам нужно квантование. Тут есть выбор:

Я выбрал Q4_K_M — оптимально. Квантуем через llama.cpp:

# Конвертируем в GGUF (используем свежий скрит из llama.cpp)
python llama.cpp/convert.py \
    --outfile qwencoder-14b.q4.gguf \
    --outtype q4_k_m \
    ~/models/Qwen3-Coder-14B-Instruct/

# Или качаем готовую квантованную, если доверяете источнику
# Но лучше сделать самому — контролируете процесс

Подробнее про тонкости квантования читайте в материале про IQ2 и про Qwen3-32B INT4.

3 Пишем ядро pipeline: генерация, верификация, исполнение

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

Создаем файл pipeline.py. Импортируем нужное.

import asyncio
import subprocess
import numpy as np
from typing import List, Dict, Any
import docker  # для изолированного запуска кода
from llama_cpp import Llama  # биндинги для llama.cpp

# Загружаем модель один раз
print("Loading model...")
llm = Llapa(
    model_path="./qwencoder-14b.q4.gguf",
    n_ctx=16384,  # большой контекст для кода
    n_gpu_layers=40,  # все слои на GPU, если есть
    n_threads=12,
    verbose=False
)

Функция генерации N вариантов кода. Ключевой трюк — меняем temperature и top_p для разнообразия.

async def generate_candidates(prompt: str, n: int = 5) -> List[str]:
    """Генерирует n вариантов кода для одного промпта."""
    candidates = []
    # Разные параметры сэмплинга для разнообразия
    settings = [
        {"temperature": 0.2, "top_p": 0.95},
        {"temperature": 0.5, "top_p": 0.9},
        {"temperature": 0.7, "top_p": 0.8},
        {"temperature": 0.3, "top_p": 0.99},
        {"temperature": 0.8, "top_p": 0.7},
    ]
    # Цикл асинхронно, но llama.cpp может блокировать, поэтому используем ThreadPoolExecutor
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        tasks = []
        for i in range(n):
            params = settings[i % len(settings)]
            task = loop.run_in_executor(
                pool,
                llm.create_completion,
                prompt,
                {
                    "max_tokens": 1024,
                    "stop": ["```", "\n\n\n"],
                    **params
                }
            )
            tasks.append(task)
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for res in results:
            if isinstance(res, dict):
                candidates.append(res["choices"][0]["text"])
    return candidates

Теперь energy-based верификатор. Мы используем ту же модель, но с промптом "Оцени качество этого кода по шкале от 1 до 10. Учитывай корректность, читаемость, эффективность." И заставляем модель выдать число.

async def energy_score(code: str, prompt_context: str) -> float:
    """Оценка кода моделью. Возвращает score от 0 до 1."""
    verification_prompt = f"""
Ты — эксперт по код-ревью. Оцени следующий код, который решает задачу: {prompt_context}
Код:
```
{code}
```
Дай числовую оценку от 1 (плохо) до 10 (отлично). Только число.
"""
    response = await loop.run_in_executor(
        None,
        lambda: llm.create_completion(
            verification_prompt,
            max_tokens=10,
            temperature=0.1,
            stop=["\n"]
        )
    )
    try:
        score_text = response["choices"][0]["text"].strip()
        score = float(score_text) / 10.0  # нормализуем до 0-1
        return max(0.0, min(1.0, score))
    except:
        return 0.5  # если модель сошла с ума

А вот самый честный судья — запуск кода. Используем Docker для изоляции. Для простоты берем Python, но можно адаптировать под JS, Go, Rust.

import docker
import tempfile

client = docker.from_env()

def execute_code_test(code: str, test_input: str, expected_output: str) -> bool:
    """Запускает код в контейнере и проверяет вывод."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        f.flush()
        file_path = f.name
    try:
        # Запускаем контейнер с Python, выполняем код, передаем входные данные
        container = client.containers.run(
            'python:3.11-slim',
            f'python {file_path}',
            stdin_open=True,
            detach=True,
            mem_limit='100m',  # ограничиваем память
            network_mode='none',  # без сети
        )
        # Передаем входные данные
        if test_input:
            container.exec_run(f'sh -c \"echo \"{test_input}\" | python {file_path}\"')
        # Ждем результат
        result = container.wait()
        logs = container.logs().decode().strip()
        container.remove(force=True)
        # Простая проверка вывода
        return expected_output in logs
    except Exception as e:
        print(f"Execution failed: {e}")
        return False
    finally:
        os.unlink(file_path)

Это упрощенный пример. В реальности используйте библиотеку code-eval или аналог, который умеет работать с LiveCodeBench задачами.

4 Собираем все вместе: алгоритм выбора лучшего кода

Теперь у нас есть три источника оценки: score от модели, результат исполнения, и можем добавить простые эвристики (длина кода, наличие опасных функций). Комбинируем.

async def select_best_candidate(candidates: List[str], original_prompt: str, test_cases: List[Dict]) -> Dict:
    """Выбирает лучший кандидат на основе комбинированной оценки."""
    scored = []
    for code in candidates:
        # Параллельно получаем energy score и проверяем выполнение тестов
        energy_task = asyncio.create_task(energy_score(code, original_prompt))
        execution_passed = all(
            execute_code_test(code, tc['input'], tc['output'])
            for tc in test_cases
        )
        energy = await energy_task
        
        # Комбинированный score: execution имеет больший вес
        if execution_passed:
            combined = energy * 0.3 + 0.7  # 0.7 за прохождение тестов
        else:
            combined = energy * 0.3
        
        # Штраф за слишком длинный код (возможное зацикливание)
        if len(code) > 5000:
            combined *= 0.5
        
        scored.append({"code": code, "score": combined, "passed": execution_passed})
    
    # Сортируем по score
    scored.sort(key=lambda x: x['score'], reverse=True)
    return scored[0] if scored else None

И главная функция pipeline:

async def ttc_pipeline(problem_statement: str, test_cases: List[Dict]) -> str:
    """Полный test-time compute pipeline."""
    print(f"Processing: {problem_statement[:50]}...")
    # Шаг 1: Генерация кандидатов
    candidates = await generate_candidates(problem_statement, n=5)
    if not candidates:
        return "// Generation failed"
    
    # Шаг 2: Верификация и выбор
    best = await select_best_candidate(candidates, problem_statement, test_cases)
    
    # Шаг 3: Если лучший не прошел тесты, можно попробовать регенерацию с исправлением ошибки
    if best and not best["passed"]:
        # Добавляем в промпт информацию об ошибке и генерируем еще раз
        retry_prompt = f"{problem_statement}\n\nПредыдущая попытка содержала ошибки. Исправь код."
        retry_candidates = await generate_candidates(retry_prompt, n=3)
        if retry_candidates:
            best_retry = await select_best_candidate(retry_candidates, retry_prompt, test_cases)
            if best_retry and best_retry["score"] > best["score"]:
                best = best_retry
    
    return best["code"] if best else "// No suitable solution found"

Где взять тестовые задачи? LiveCodeBench — ваш новый лучший друг

Чтобы оценить, работает ли ваш pipeline, нужны задачи с известными решениями и тестами. LiveCodeBench (актуальный на 2026 год) — это сборник реальных задач с LeetCode, Codeforces и других платформ, адаптированных для оценки LLM.

Скачиваем датасет, берем несколько задач и превращаем их в наш формат:

# Пример задачи из LiveCodeBench
problem = {
    "id": "lc_123",
    "statement": "Напишите функцию, которая проверяет, является ли строка палиндромом.",
    "test_cases": [
        {"input": "'racecar'", "output": "True"},
        {"input": "'hello'", "output": "False"},
        {"input": "'a'", "output": "True"}
    ],
    "reference_solution": "def is_palindrome(s): return s == s[::-1]"
}

Запускаем pipeline на 20-30 таких задачах, сравниваем с запуском одиночной генерации (без TTC). Метрика — процент правильно решенных задач (pass@1). У студента из истории с этим pipeline на Qwen3-14B pass@1 вырос с 62% до 78% на подмножестве LiveCodeBench. Это уровень некоторых 34B моделей в базовой конфигурации.

Оптимизации, без которых pipeline будет ползти как улитка

Генерация 5 вариантов, плюс верификация, плюс исполнение — это медленно. Вот как ускорить.

• Кеширование: Если один и тот же промпт встречается часто (например, стандартные функции), сохраняйте лучший результат в SQLite или Redis.
• Параллельное исполнение тестов: Запускайте тесты для всех кандидатов одновременно, а не последовательно. Используйте asyncio.gather с ограничением количества контейнеров.
• Более легкий верификатор: Для верификации используйте меньшую модель, например, Qwen3-1.8B, или специально дообученный классификатор. Это снизит нагрузку.
• Ранняя остановка (Halting detection): Если модель генерирует явный мусор (например, повторяющиеся строки), обрывайте генерацию досрочно. Добавьте детектор по паттернам.

Совет по настройке llama.cpp для скорости: изучите этот гайд по оптимизации Qwen3.5 27B. Принципы те же: игра с batch_size, flash_attn, компиляция под ваше железо.

Чего не делать? Ошибки, которые сведут на нет все усилия

1. Не запускайте недоверенный код без песочницы. Выполнение кода с subprocess.run прямо на хосте — билет к rm -rf /. Всегда используйте изоляцию: Docker, gVisor, Firecracker.
2. Не игнорируйте таймауты. Устанавливайте жесткие лимиты на время выполнения контейнера (например, 10 секунд). Иначе одна задача с бесконечным циклом заблокирует весь пайплайн.
3. Не экономьте на контексте. Qwen3-14B поддерживает 128к токенов, но для квантованной версии в llama.cpp может быть меньше. Убедитесь, что n_ctx достаточно для промпта + генерации. Иначе получите ошибки, как в случае с Qwen Coder 30B.
4. Не используйте pipeline для всего. Для простых задач вроде "напиши hello world" достаточно одного вызова модели. Включайте TTC только для сложных или критичных задач.

Что в итоге? Вы получаете свою frontier-модель за копейки

Этот pipeline — не просто скрипт. Это философия. Вы признаете, что одна генерация может быть неудачной, и даете модели шанс исправиться. Вы используете вычислительные ресурсы не для обучения гигантской модели, а для того, чтобы выжать максимум из той, что уже есть.

На 10 марта 2026 года такой подход дает Qwen3-14B-Coder возможность конкурировать с более крупными моделями в задачах кодирования. Вы экономите на облачных API (которые к 2026 все еще дороги для больших контекстов) и получаете полный контроль над конфиденциальностью.

Следующий шаг? Добавить в pipeline семплирование по методам (Chain-of-Thought, Plan-and-Solve) и автоматический подбор промптов. Но это уже тема для другой статьи.

Начните с малого. Возьмите одну задачу, напишите pipeline из 3 кандидатов и простой Docker-песочницы. Увидите разницу. А потом уже масштабируйте.

И да, если у вас несколько GPU, можно распределить генерацию кандидатов между ними. Как это сделать — читайте в материале про 3x3090. Удачи.

Подписаться на канал