Самое важное из мира AI — в канале MAX AgentLabОтдельные разборы, инструменты и практические схемы Читайте Agent Lab в Telegram Разборы, кейсы и новости о практических AI-агентах без лишней воды.

Фабрика задач для терминального агента: воспроизводим рекурсивное усложнение

Фабрика задач для терминального агента: воспроизводим рекурсивное усложнение
Temporary fallback cover; replace in editorial pass.

Уровень: продвинутый · Чтение: до 12 минут · Опубликовано 10 октября 2026

Хорошая задача для терминального агента состоит из четырёх частей: формулировки, окружения, эталонного решения и тестов. Все четыре должны друг другу соответствовать. Если писать такие задачи вручную, это дорого. Если генерировать их полностью синтетически, тесты часто проверяют не то, что написано в формулировке, решение тянет пакет, которого нет в образе, а «сложная» задача на деле оказывается просто двусмысленной.

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

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

Что получится в конце

Потребуются Docker, Python 3.10+ и два ваших внешних компонента. Первый — команда расширителя, которая обращается к вашей LLM и правит файлы задачи. Второй — обвязка агента, которая выполняет команды внутри контейнера. Их интерфейс мы зафиксируем, а реализацию оставим вам: конкретные API и модели в статье не предполагаются.

Шаг 1. Структура задачи

Главное правило: тесты не попадают в образ, который видит агент. Их монтируют только на этапе проверки.

tasks/d0/
├── instruction.md      # формулировка для агента
├── Dockerfile          # окружение без тестов и без решения
├── data/               # входные данные задачи
├── solution.sh         # эталонное решение
└── tests/
    └── test_outputs.py # проверка результата, а не способа

Пример исходной задачи (вымышленный, для иллюстрации): «В /app/data/access.log лежит журнал веб-сервера. Запиши в /app/out.txt количество уникальных IP-адресов».

# tasks/d0/Dockerfile
FROM python:3.12-slim
RUN pip install --no-cache-dir pytest \
 && useradd -m runner
WORKDIR /app
COPY data/ /app/data/
RUN chown -R runner:runner /app
USER runner

# tasks/d0/solution.sh
#!/usr/bin/env bash
set -euo pipefail
awk '{print $1}' /app/data/access.log | sort -u | wc -l > /app/out.txt

# tasks/d0/tests/test_outputs.py
from pathlib import Path

def expected():
    lines = Path("/app/data/access.log").read_text().splitlines()
    return len({l.split()[0] for l in lines if l.strip()})

def test_count():
    out = Path("/app/out.txt").read_text().strip()
    assert int(out) == expected()

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

Шаг 2. Запуск в чистом окружении

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

# factory/sandbox.py
import subprocess
from pathlib import Path

LIMITS = ["--network", "none", "--memory", "1g", "--cpus", "1",
          "--pids-limit", "256", "--user", "runner"]

def build(task: Path) -> str:
    tag = f"factory/{task.name}:latest"
    subprocess.run(["docker", "build", "-q", "-t", tag, str(task)],
                   check=True, capture_output=True)
    return tag

def run_check(task: Path, solution: Path | None, tests: Path) -> bool:
    tag = build(task)
    cmd = ["docker", "run", "--rm", *LIMITS,
           "-v", f"{tests.resolve()}:/tests:ro"]
    script = "pytest -q -p no:cacheprovider /tests"
    if solution is not None:
        cmd += ["-v", f"{solution.resolve()}:/sol.sh:ro"]
        script = "bash /sol.sh && " + script
    cmd += [tag, "bash", "-c", script]
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    except subprocess.TimeoutExpired:
        return False
    return r.returncode == 0

Флаг :ro не даёт решению переписать тесты. Флаг -p no:cacheprovider нужен, чтобы pytest не пытался писать кэш в смонтированный только для чтения каталог.

Шаг 3. Валидатор: три обязательные проверки

Задача-потомок child принимается, только если выполнены все условия:

  1. Новое решение проходит новые тесты, причём несколько раз подряд. Повторы отсекают нестабильные тесты.
  2. Без решения новые тесты падают. Если тесты проходят на пустом окружении, они ничего не проверяют.
  3. Решение родителя не проходит новые тесты. Если старое решение справляется, расширение ничего не добавило, и рост сложности фиктивный.
# factory/validate.py
from pathlib import Path
from factory.sandbox import run_check

def validate(parent: Path, child: Path, repeats: int = 3) -> dict:
    tests = child / "tests"
    report = {
        "new_solution_passes": all(
            run_check(child, child / "solution.sh", tests)
            for _ in range(repeats)),
        "empty_fails": not run_check(child, None, tests),
        "parent_solution_fails": not run_check(
            child, parent / "solution.sh", tests),
    }
    report["accepted"] = all(report.values())
    return report

Решение родителя запускается в окружении потомка. Так и задумано: нас интересует, решает ли старый подход новую задачу.

Шаг 4. Расширитель и рекурсия

Расширитель — внешняя команда из переменной EXPANDER_CMD. На вход она получает путь к копии задачи и должна согласованно изменить instruction.md, solution.sh, tests/ и при необходимости Dockerfile. В промпте полезно потребовать ровно одно новое проверяемое требование и запретить менять файлы в data/ без явной причины.

# factory/expand.py
import json, os, shlex, shutil, subprocess, sys
from pathlib import Path
from factory.validate import validate

EXPANDER = shlex.split(os.environ["EXPANDER_CMD"])

def expand_once(parent: Path, child: Path, attempts: int = 3) -> dict | None:
    for i in range(attempts):
        if child.exists():
            shutil.rmtree(child)
        shutil.copytree(parent, child)
        subprocess.run([*EXPANDER, str(child)], check=True, timeout=900)
        report = validate(parent, child)
        report.update(attempt=i + 1, parent=parent.name, child=child.name)
        print(json.dumps(report, ensure_ascii=False), file=sys.stderr)
        if report["accepted"]:
            return report
    shutil.rmtree(child, ignore_errors=True)
    return None

def build_chain(root: Path, max_depth: int):
    parent = root / "d0"
    for d in range(1, max_depth + 1):
        child = root / f"d{d}"
        if expand_once(parent, child) is None:
            print(f"остановка на глубине {d}", file=sys.stderr)
            break
        parent = child

if __name__ == "__main__":
    build_chain(Path("tasks"), int(sys.argv[1]))

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

python -c "from pathlib import Path; from factory.sandbox import run_check as r; \
t=Path('tasks/d0'); print(r(t, t/'solution.sh', t/'tests'), r(t, None, t/'tests'))"
# ожидаемо: True False

export EXPANDER_CMD="./my_expander.sh"   # ваша обёртка над LLM
python -m factory.expand 4 2> expand_log.jsonl

Как может выглядеть цепочка для примера выше (иллюстрация, а не вывод реального прогона): d1 — учитывать только строки с кодом ответа 2xx; d2 — записывать JSON с количеством запросов по каждому IP, отсортированный по убыванию; d3 — поддержать сжатые ротированные журналы access.log.*.gz.

Шаг 5. Измеряем падение успешности

Агент работает в долгоживущем контейнере без тестов. Тесты копируются в контейнер только после того, как агент закончил. Обвязка агента задаётся в AGENT_CMD: она получает имя контейнера и путь к формулировке и выполняет команды через docker exec.

# factory/evaluate.py
import json, os, shlex, subprocess, sys, uuid
from pathlib import Path
from factory.sandbox import LIMITS, build

AGENT = shlex.split(os.environ["AGENT_CMD"])

def sh(*args, **kw):
    return subprocess.run(list(args), capture_output=True, text=True, **kw)

def run_agent(task: Path) -> bool:
    tag, name = build(task), f"eval-{uuid.uuid4().hex[:8]}"
    sh("docker", "run", "-d", "--name", name, *LIMITS, tag,
       "sleep", "infinity", check=True)
    try:
        try:
            sh(*AGENT, name, str(task / "instruction.md"), timeout=1800)
        except subprocess.TimeoutExpired:
            return False
        sh("docker", "cp", str(task / "tests"), f"{name}:/tmp/tests", check=True)
        r = sh("docker", "exec", name, "pytest", "-q",
               "-p", "no:cacheprovider", "/tmp/tests", timeout=600)
        return r.returncode == 0
    finally:
        sh("docker", "rm", "-f", name)

if __name__ == "__main__":
    runs = int(sys.argv[1])
    with open("results.jsonl", "a") as f:
        for task in sorted(Path("tasks").glob("d*")):
            for i in range(runs):
                ok = run_agent(task)
                f.write(json.dumps({"task": task.name, "run": i, "ok": ok}) + "\n")

У сети здесь стоит --network none. Если обвязка вызывает модель с хоста, а внутри контейнера выполняет только команды, агенту сеть не нужна. Если задача честно требует сети, отразите это в формулировке и не скрывайте.

# factory/report.py
import json, math
from collections import defaultdict

def wilson(k, n, z=1.96):
    if n == 0:
        return (0.0, 0.0)
    p = k / n
    d = 1 + z*z/n
    c = (p + z*z/(2*n)) / d
    h = z * math.sqrt(p*(1-p)/n + z*z/(4*n*n)) / d
    return (max(0, c - h), min(1, c + h))

stats = defaultdict(lambda: [0, 0])
for line in open("results.jsonl"):
    r = json.loads(line)
    stats[r["task"]][0] += r["ok"]
    stats[r["task"]][1] += 1

base = None
for task in sorted(stats, key=lambda t: int(t[1:])):
    k, n = stats[task]
    rate, (lo, hi) = k / n, wilson(k, n)
    base = rate if base is None else base
    print(f"{task}: {k}/{n} = {rate:.2f}  95% CI [{lo:.2f}; {hi:.2f}]  "
          f"падение от d0: {base - rate:+.2f}")

export AGENT_CMD="./my_agent_harness.sh"
python -m factory.evaluate 10
python -m factory.report

Как проверить, что конвейер работает

Типовые ошибки

Ограничения подхода

Что дальше

Разобраться с изоляцией и устройством обвязки агента помогут материалы в разделе гайдов, а термины из статьи собраны в глоссарии. Логичное продолжение — хранить expand_log.jsonl рядом с задачами. Тогда для каждой задачи будет видно, с какой попытки её приняли и какие проверки отклонили предыдущие версии.