Dry-run для tool calls: пишем безопасный sandbox для AI-агента на Python

В комментариях к предыдущему материалу мы обсуждали canary-проверки для LLM-провайдеров. Но даже исправный провайдер не решает другую проблему: как проверить, что агент правильно вызывает CRM, почту или платёжный API, не создавая реальные записи и не отправляя сообщения клиентам.

Ниже — практическая схема dry-run слоя на Python: единый реестр инструментов, проверка payload по той же схеме, что используется в проде, подмена side effects и безопасный commit после детерминированных проверок.

У tool-calling агента есть два разных класса ошибок.

Первый — модель выбрала неправильный инструмент или сформировала невалидные аргументы. Это можно поймать до внешнего вызова.

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

Поэтому dry-run — это не параметр dry_run=true, разбросанный по интеграциям. Это отдельный режим исполнения со своим контрактом:

Главное правило: dry-run и production не должны иметь две разные схемы payload. Иначе тестовый контур постепенно начнёт проверять уже несуществующий контракт.

В примере используем Pydantic:

from enum import Enum from typing import Any, Awaitable, Callable from pydantic import BaseModel, EmailStr, Field class Risk(str, Enum): READ = "read" REVERSIBLE_WRITE = "reversible_write" IRREVERSIBLE_WRITE = "irreversible_write" class SendEmailArgs(BaseModel): customer_id: str = Field(min_length=1) to: EmailStr subject: str = Field(min_length=1, max_length=160) body: str = Field(min_length=1, max_length=20_000) class ToolSpec(BaseModel): name: str risk: Risk args_model: type[BaseModel] execute: Callable[[BaseModel, str], Awaitable[dict[str, Any]]] simulate: Callable[[BaseModel, str], Awaitable[dict[str, Any]]] class Config: arbitrary_types_allowed = True

У инструмента две реализации, но один args_model. Любое изменение поля в production автоматически меняет и dry-run валидацию.

Плохая заглушка отвечает {"status": "ok"}. Такой тест проверяет только то, что код не упал. Полезная симуляция должна показать, что именно произошло бы:

async def simulate_email(args: SendEmailArgs, operation_id: str) -> dict: customer = await crm.get_customer(args.customer_id) return { "mode": "dry_run", "operation_id": operation_id, "effect": "send_email", "recipient": str(args.to), "customer_email": customer.email, "recipient_matches_crm": str(args.to) == customer.email, "subject_length": len(args.subject), "body_length": len(args.body), "external_requests": 0, } async def execute_email(args: SendEmailArgs, operation_id: str) -> dict: response = await email_provider.send( to=str(args.to), subject=args.subject, body=args.body, idempotency_key=operation_id, ) return { "mode": "commit", "operation_id": operation_id, "provider_message_id": response.id, }

Dry-run может читать данные из CRM, но не должен записывать их. Он возвращает факты, на которых policy layer примет решение.

LLM не должна передавать commit=true. Иначе модель одновременно предлагает действие и выдаёт себе разрешение. Режим задаётся доверенным оркестратором:

from dataclasses import dataclass from uuid import uuid4 @dataclass(frozen=True) class ExecutionContext: mode: str approved_by: str | None = None class ToolRunner: def __init__(self, tools: dict[str, ToolSpec]): self.tools = tools async def run(self, tool_name, raw_args, context, operation_id=None): spec = self.tools[tool_name] args = spec.args_model.model_validate(raw_args) op_id = operation_id or f"op_{uuid4().hex}" if context.mode == "dry_run": return await spec.simulate(args, op_id) if context.mode != "commit": raise ValueError("unknown execution mode") if spec.risk == Risk.IRREVERSIBLE_WRITE and not context.approved_by: raise PermissionError("explicit approval is required") return await spec.execute(args, op_id)

Operation ID создаётся до внешнего действия. После успешной проверки commit продолжает ту же бизнес-операцию, а не создаёт новую.

Не стоит просить вторую LLM оценить, безопасен ли результат первой. Для критичных правил нужен обычный код.

class PolicyViolation(Exception): pass def validate_email_plan(plan: dict) -> None: if plan["effect"] != "send_email": raise PolicyViolation("unexpected effect") if not plan["recipient_matches_crm"]: raise PolicyViolation("recipient differs from CRM") if plan["body_length"] > 10_000: raise PolicyViolation("message is too long") if plan["external_requests"] != 0: raise PolicyViolation("dry-run produced a side effect")

В реальном проекте сюда добавляются лимиты суммы, allowlist доменов, права пользователя, текущее состояние заказа, версия CRM-записи и допустимое окно времени.

Один и тот же payload должен пройти оба этапа без редактирования моделью между ними.

import hashlib import json def payload_hash(payload: dict) -> str: canonical = json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ) return hashlib.sha256(canonical.encode()).hexdigest() async def safe_send_email(tool_call: dict, actor_id: str) -> dict: operation_id = f"email_{uuid4().hex}" original_hash = payload_hash(tool_call["arguments"]) plan = await runner.run( "send_email", tool_call["arguments"], ExecutionContext(mode="dry_run"), operation_id=operation_id, ) validate_email_plan(plan) if payload_hash(tool_call["arguments"]) != original_hash: raise PolicyViolation("payload changed after validation") return await runner.run( "send_email", tool_call["arguments"], ExecutionContext(mode="commit", approved_by=actor_id), operation_id=operation_id, )

Между validate и commit данные во внешней системе могут измениться. Поэтому для записи полезны version, ETag или compare-and-set. Если CRM-объект уже обновился, план нужно построить заново.

Полностью синтетический контур быстро отстаёт от продукта. Короткий промпт на 500 токенов проходит, а реальные запросы с контекстом на 20 тысяч токенов падают. В тестовом payload поле необязательное, а в production оно уже обязательное.

Практичнее хранить обезличенные профили реальной нагрузки:

Сохранять пользовательский текст целиком обычно не нужно. Достаточно структуры, размеров, набора инструментов, числа сообщений и безопасно обезличенных значений.

Расписание полезно как страховка, но основной сигнал — изменение реального трафика. Можно отслеживать p50 и p95 context length, долю запросов с tool calls, распределение инструментов, размер payload, latency, error rate и долю отклонений policy layer.

Если метрики несколько окон подряд выходят за установленный коридор, набор проб пересобирается из свежих обезличенных запросов. После смены модели, system prompt или схемы инструмента пересборка должна быть принудительной.

Новый профиль сначала запускается в shadow-режиме параллельно со старым. Это помогает не потерять редкий сценарий, который исчез из свежей выборки, но всё ещё встречается в production.

async def test_dry_run_has_no_side_effects(): plan = await runner.run( "send_email", { "customer_id": "customer_42", "to": "test@example.com", "subject": "Test", "body": "Preview only", }, ExecutionContext(mode="dry_run"), ) assert plan["mode"] == "dry_run" assert plan["external_requests"] == 0 assert email_provider.send.call_count == 0

Симуляция не докажет, что реальный провайдер примет запрос, не изменит контракт и не вернёт неожиданный ответ. Она также не устранит гонку между проверкой и записью.

Поэтому первые реальные операции после изменения интеграции лучше пропускать через ограниченный canary: минимальный объём, стабильный idempotency key, усиленный аудит, reconciliation после вызова и rollback только для действительно обратимых операций.

Для платежей, сообщений клиенту, выдачи доступа и удаления одного dry-run недостаточно. Нужны явный commit gate и подтверждение фактического результата.

Хороший dry-run не притворяется, что действие успешно выполнено. Он строит проверяемый план: что агент хочет изменить, с какими аргументами, в каком объекте и при каких ограничениях.

Безопасность появляется не из ещё одного промпта, а из разделения ответственности: LLM предлагает tool call, sandbox моделирует эффект, детерминированный слой проверяет правила, а commit выполняет только уже зафиксированную операцию.

А как у вас устроен dry-run для agent tools: заглушки, отдельный test

tenant, shadow traffic или prepare/commit слой?