Узнайте, как построить AI-пайплайн с машинно-проверяемыми доказательствами: gates, cold critic, оракул. Практический пример с кодом на Python.
Представьте: вы запускаете LLM-агента, он пишет код, вы проверяете — и находите баг. Доверие подорвано. Как сделать так, чтобы каждый артефакт был проверен автоматически, а человек контролировал только необратимые действия? В этой статье — метод, который я применил в реальном проекте: система из пяти ролей, gates и оракул для инвариантов.
LLM-агенты быстры, но ненадёжны. Инженерная проблема — не получить результат, а доверять ему. Решение: работа агента считается непроверенной, пока gate не докажет обратное. Координация автоматизирована, последствия — под контролем. Каждый gate существует, чтобы я мог двигаться быстро, не ломая продукт.
Цикл: Strategy → Execution → Critic → Eval → Ops. Роли суждения (Strategy, Critic, Eval) работают на более сильных моделях; исполнительные — на дешёвых. Стоимость следует за сложностью решения.
Над исполнительными агентами — одна роль оркестрации, которой я управляю. Она планирует единицу работы, направляет её в нужную роль и модель, хранит состояние между шагами. Я не вне цикла — я суждение и авторитет системы.
class Orchestrator:
def __init__(self):
self.state = {}
def plan_work(self, task):
# Планирование: выбрать роли и модели
plan = {
"strategy": "gpt-4",
"execution": "gpt-3.5-turbo",
"critic": "claude-3-opus",
"eval": "gpt-4"
}
return plan
def route(self, work_unit):
# Направление работы согласно плану
role = work_unit["role"]
model = self.plan_work(work_unit)[role]
# Вызов модели
result = call_llm(model, work_unit["prompt"])
return resultПеред закрытием изменения оно проверяется Critic на свежей сессии с нулевым контекстом — другая сильная модель без памяти о том, как писался код. Это ловит то, что автор уже перестал замечать.
# Пример: вызов Critic с нулевым контекстом
import openai
def cold_critic_review(code_snippet):
# Нет истории диалога, нет контекста
response = openai.ChatCompletion.create(
model="claude-3-opus",
messages=[
{"role": "system", "content": "Ты — суровый код-ревьюер. Найди все проблемы."},
{"role": "user", "content": f"Проверь этот код:\n{code_snippet}"}
]
)
return response["choices"][0]["message"]["content"]Передачи между ролями автоматизированы, но каждое необратимое действие (деплой билда на устройство, коммит в git) требует явного одобрения человека. Автоматизируй координацию, никогда — необратимое.
# Пример: gate для коммита
def commit_gate(changes, approval):
if not approval:
raise Exception("Commit rejected: human approval required")
# Выполнить коммит
import subprocess
subprocess.run(["git", "add", "-A"])
subprocess.run(["git", "commit", "-m", "Auto-commit"])
print("Commit successful")Ключевая логика реализуется дважды, и две версии фаззятся друг против друга. Там, где они расходятся, одна неверна — золотые метки не нужны.
# Пример: дифференциальный оракул
def oracle(impl_a, impl_b, input_generator):
for _ in range(1000):
inp = next(input_generator)
res_a = impl_a(inp)
res_b = impl_b(inp)
if res_a != res_b:
print(f"Расхождение на входе {inp}: A={res_a}, B={res_b}")
# Одна из реализаций неверна
return False
return TrueКаждая закрытая единица работы оставляет долговечное, машинно-проверяемое доказательство.
Рабочий элемент не может быть закрыт, пока автоматическая проверка не подтвердит наличие доказательства на диске. Цикл физически не может пропустить этот шаг.
# Пример: gate проверки доказательства
def close_work_item(work_id, proof_path):
import os
if not os.path.exists(proof_path):
raise Exception(f"Work {work_id} cannot close: no proof found")
# Дополнительная проверка: машинно-читаемый формат
with open(proof_path, "r") as f:
proof = f.read()
if not validate_proof_format(proof):
raise Exception("Invalid proof format")
print(f"Work {work_id} closed with proof")Скриншоты проверок на устройстве санируются (чувствительные области замазываются), а манифесты происхождения связывают изображения с конкретным git SHA, размерами экрана и методом редактирования.
# Пример: создание манифеста
def create_provenance_manifest(image_path, git_sha, dimensions, redaction_method):
manifest = {
"image": image_path,
"git_sha": git_sha,
"dimensions": dimensions,
"redaction_method": redaction_method
}
# Сохранить как JSON
import json
with open("manifest.json", "w") as f:
json.dump(manifest, f)
return manifestКаждая контрольная точка записывает: scoped git staging (только явные пути), коммит/SHA по всем репозиториям, аудит реестра артефактов, явное одобрение оператора.
# Пример: создание checkpoint
def create_checkpoint(paths, repos, artifacts):
# 1. git staging только указанных путей
import subprocess
subprocess.run(["git", "add"] + paths)
# 2. Получить SHA
sha = subprocess.check_output(["git", "rev-parse", "HEAD"]).strip()
# 3. Аудит артефактов
audit = {}
for artifact in artifacts:
audit[artifact] = verify_artifact(artifact)
# 4. Ожидание одобрения
approval = input("Approve checkpoint? (y/n): ")
if approval != 'y':
raise Exception("Checkpoint rejected")
print(f"Checkpoint created: SHA={sha}, audit={audit}")Независимая роль Eval выдаёт числовую оценку здоровья с дельтой к предыдущему периоду и таксономией отказов. Регрессии попадают в реестр отказов, который ведёт к исправлениям.
# Пример: периодическая оценка
def evaluate_health():
# Собрать метрики
failures = get_failure_count()
pass_rate = get_pass_rate()
score = 100 - failures * 10 + pass_rate * 0.5
delta = score - previous_score
taxonomy = classify_failures()
return {"score": score, "delta": delta, "taxonomy": taxonomy}Ключевая логика продукта проверяется property-based тестами: сгенерированные входы подаются в движок, и набор инвариантов должен выполняться для каждого. Например, детектор должен согласовываться с независимым полным пересканированием, а обнаружение должно быть без побочных эффектов.
# Пример: property-based тест на Python с Hypothesis
from hypothesis import given, strategies as st
def invariant_detector_agreement(detector, full_scan):
# Инвариант: результат детектора совпадает с полным сканированием
pass
@given(st.lists(st.integers()))
def test_invariant(data):
result = process(data)
# Инвариант: результат не должен содержать дубликатов
assert len(result) == len(set(result)), "Duplicate found"
# Инвариант: все элементы результата должны быть из входных данных
assert all(x in data for x in result), "Element not in input"Метод переносится на любую кодовую базу. Дисциплина доказательств — это навык, который тренируется практикой.
Хочешь закрепить знания на практике?
Решай задачи на Algolit — интерактивная платформа для обучения
Начать бесплатно →