Перейти к содержимому
PD
AI-автоматизация 8 мин чтения

Каждый промпт - это экспорт данных: сделайте egress-слой до запуска AI-фич

Любой вызов LLM отправляет данные клиента третьей стороне. Разбираю gateway, карту редактирования, таблицу политик и тесты, которые делают экспорт осознанным, залогированным и защитимым.

PD

Pavel Duglas

AI Automation & MVP Architect

Большинство команд относится к вызову LLM как к обычному вызову функции. Это не так. Это экспорт данных в компанию, которой вы не управляете, по сети, которая вам не принадлежит, в логи, которые вы не можете прочитать. Когда вы вставляете тело обращения из поддержки в промпт, вы только что отправили имя клиента, его email и, возможно, адрес через границу. Юридически и операционно это ближе к выгрузке файла на SFTP вендора, чем к array.map().

Я делаю AI-автоматизации для клиентов с живыми пользователями, поэтому этот вопрос всплывает в каждом проекте. И вопрос никогда не звучит как “безопасен ли AI”. Вопрос такой: какие поля покинули наш периметр, к какому вендору, по какому договору и смогу ли я доказать это через полгода. Ниже - архитектура, которой я отвечаю на это, не замедляя разработку.

Сценарий, который никто не планирует

Типичная AI-фича растёт так:

  1. Неделя 1: скрипт суммаризует тикеты поддержки. Тело тикета целиком уходит в промпт, потому что так быстрее всего.
  2. Неделя 3: добавили карточку клиента “для контекста”. Теперь наружу уходят email, телефон и тариф.
  3. Неделя 5: подключили observability для промптов, чтобы дебажить. Полные payload’ы теперь лежат у второго вендора.
  4. Неделя 7: кто-то собрал агента, который читает CRM и пишет в Slack. Он отправляет строку целиком, вместе с внутренними заметками менеджера.
  5. Месяц 4: клиент просит схему потоков данных для своего security review. Нарисовать её не может никто.

Злого умысла тут нет. Каждый шаг был разумным локальным решением. Проблема в том, что не было ни одной точки, где данные пересекают границу. А значит, не было места, куда положить правило.

Лечение скучное и работает: сделать границу явной, узкой и единственной.

Один gateway, без исключений

Правило номер один в любом моём проекте: прикладной код никогда не обращается к SDK провайдера напрямую. Он обращается к моему модулю llm. И только этому модулю разрешено импортировать openai, anthropic, google.generativeai и прочих.

Это не архитектурный перфекционизм. Это единственный способ вообще что-то потом проконтролировать. Если у вас четырнадцать точек вызова, у вас четырнадцать шансов забыть про маскирование. Если одна - у вас одно место, куда добавить правило, и одно место, которое надо покрыть тестом.

Проверяю это линтом и grep-ом в CI:

# билд падает, если SDK провайдера импортируют вне src/llm/
grep -rEn "from (openai|anthropic)" src \
  | grep -v "^src/llm/" \
  && { echo "provider SDK imported outside the gateway"; exit 1; }

Пять строк, некрасиво, ловит ошибку каждый раз. На больших кодовых базах вместо этого - ESLint no-restricted-imports с исключением по пути или контракт import-linter в Python.

Классифицируйте поля один раз, на уровне схемы

Нельзя написать правило маскирования для “тела тикета”. Можно написать его для помеченного поля. Поэтому чувствительность я размечаю в модели данных, а не в промпте.

from dataclasses import dataclass, field

# PUBLIC   - можно отправлять куда угодно
# INTERNAL - бизнес-данные, только вендорам с договором
# PII      - идентифицирует человека, токенизировать перед отправкой
# SECRET   - не покидает процесс никогда

@dataclass
class Ticket:
    id: str = field(metadata={"sens": "INTERNAL"})
    subject: str = field(metadata={"sens": "INTERNAL"})
    body: str = field(metadata={"sens": "PII"})       # свободный текст, считаем худшее
    customer_email: str = field(metadata={"sens": "PII"})
    api_key_hint: str = field(metadata={"sens": "SECRET"})

Любое поле со свободным текстом всегда получает PII. Пользователи пишут свой телефон прямо в теле сообщения. Всегда.

Далее gateway принимает структурированные объекты, а не строки. Именно это решение делает возможным всё остальное:

result = llm.run(
    task="ticket_triage",
    inputs={"ticket": ticket},
    model_policy="eu_contracted",
)

Если gateway получает уже собранную строку промпта, он не знает, что внутри, и может только сканировать регулярками. Регулярки - это пожарный датчик, а не стена.

Обратимая токенизация лучше удаления

Наивный подход - вырезать PII. Тогда модель на выходе говорит “свяжитесь с клиентом” вместо “напишите Анне на её адрес”, и ваша автоматизация становится бесполезной для любого сценария с обратной записью.

Что делаю я: заменяю PII на стабильные локальные токены, держу карту соответствия у себя в памяти или в своей БД и восстанавливаю значения на обратном пути.

import re, uuid

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
PHONE = re.compile(r"\+?\d[\d\s().-]{7,}\d")

class Vault:
    def __init__(self):
        self.fwd = {}   # реальное значение -> токен
        self.rev = {}   # токен -> реальное значение

    def token(self, value, kind):
        if value in self.fwd:
            return self.fwd[value]
        t = f"<{kind}_{uuid.uuid4().hex[:6]}>"
        self.fwd[value] = t
        self.rev[t] = value
        return t

    def mask(self, text):
        text = EMAIL.sub(lambda m: self.token(m.group(), "EMAIL"), text)
        text = PHONE.sub(lambda m: self.token(m.group(), "PHONE"), text)
        return text

    def unmask(self, text):
        for t, v in self.rev.items():
            text = text.replace(t, v)
        return text

Модель видит <EMAIL_9f2c11> спрашивает про счёт <INV_44a0>. С токенами она рассуждает совершенно нормально. Ваш код подставляет реальные значения перед отправкой ответа. У вендора настоящих данных не было никогда.

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

Таблица политик вместо “ну вроде можно”

Следующий вопрос от клиента всегда одинаковый: “какую модель тут можно использовать?” Я отвечаю таблицей, закоммиченной в репозиторий, а не мнением в Slack.

policies:
  eu_contracted:
    allow_sensitivity: [PUBLIC, INTERNAL, PII_TOKENIZED]
    providers: [vendor_a_eu]
    retention_days: 0
    training_optout: true
  public_only:
    allow_sensitivity: [PUBLIC]
    providers: [vendor_a, vendor_b, local_ollama]
  local_only:
    allow_sensitivity: [PUBLIC, INTERNAL, PII]
    providers: [local_ollama]

Gateway отклоняет вызов, если максимальная чувствительность payload’а не входит в allow_sensitivity выбранной политики. Не warning. Исключение, которое валит задачу. Fail closed, всегда. Упавшая задача триажа - это тикет в поддержку. Задача триажа, которая молча отправила сырые данные клиентов на непроверенный endpoint, - это уведомление о утечке.

Побочный, но очень полезный эффект: у вас появляется настоящий ответ на security-опросник. Вот файл, вот список вендоров, вот настройка retention по каждому маршруту.

Журнал egress

На каждый исходящий вызов я пишу запись, в которой нет payload’а:

{
  "ts": "2026-02-04T09:12:44Z",
  "task": "ticket_triage",
  "policy": "eu_contracted",
  "provider": "vendor_a_eu",
  "model": "mid-tier-v3",
  "tenant": "acme",
  "fields": ["ticket.subject:INTERNAL", "ticket.body:PII_TOKENIZED"],
  "payload_sha256": "3f9a...",
  "bytes_out": 4182,
  "tokens_masked": 3,
  "trace_id": "01HQ..."
}

Имена полей и метки чувствительности - да. Значения - никогда. Хеш позволяет доказать, что конкретный payload был или не был отправлен, не храня сам payload. А bytes_out в разрезе tenant/день - лучший детектор утечек, который я знаю: когда кто-то правит промпт и тащит туда полную историю заказов, число подскакивает на порядок и алерт срабатывает раньше, чем это заметит клиент.

Журнал держите в своей БД со своим retention. Это тот артефакт, который вы отдаёте аудитору.

Не забудьте про второй и третий выход

API модели - очевидная дверь. Вот те, которые я нахожу на ревью:

  • Инструменты observability для промптов. По умолчанию они хранят полные тела запросов и ответов. Это второй вендор с теми же персональными данными. Либо отправляйте туда только замаскированное, либо поднимайте self-hosted.
  • Эмбеддинги и векторные базы. Эмбеддинг производный от исходного текста и часто лежит в managed-индексе в облаке. Правила политик действуют ровно те же.
  • Трекеры ошибок. Упавший вызов регулярно прикладывает к исключению тело запроса. Вырезайте payload в своём error handler, а не в настройках дашборда.
  • Инструменты агентов. Тула чтения файлов, shell или браузер дают модели способ достать данные, которые вы вообще не размечали. Белые списки путей и доменов - на уровне слоя тулов.
  • Сторонние skill-паки и плагины для агентов. Установка чужого набора скиллов - это установка кода с сетевым доступом в тот же процесс, где лежат ваши креды. Читайте исходники или запускайте в контейнере без примонтированных секретов.

Тесты, которые валят билд

Документы про политики гниют. Тесты - нет. Три штуки, которые я добавляю в каждый проект:

def test_pii_never_reaches_provider(fake_provider):
    ticket = Ticket(body="позвоните мне +7 903 123 45 67, anna@example.com", ...)
    llm.run(task="ticket_triage", inputs={"ticket": ticket},
            model_policy="eu_contracted")
    sent = fake_provider.last_request_text
    assert "anna@example.com" not in sent
    assert "123 45 67" not in sent

def test_policy_violation_fails_closed():
    with pytest.raises(EgressPolicyError):
        llm.run(task="ticket_triage", inputs={"ticket": raw_pii_ticket},
                model_policy="public_only")

def test_every_call_writes_a_ledger_row(fake_provider, ledger):
    llm.run(task="ticket_triage", inputs={"ticket": ticket}, model_policy="eu_contracted")
    assert len(ledger.rows) == 1
    assert "anna@example.com" not in json.dumps(ledger.rows[0])

Fake-провайдер записывает точные байты, которые ушли бы в сеть. Это и есть проверка, которая реально вас защищает, потому что она тестирует границу, а не намерения разработчика.

Сколько это стоит

Полтора дня на новом проекте, если делать сразу. Примерно неделя, если натягивать это на AI-фичу, у которой уже десяток точек вызова. Взамен вы получаете три вещи: проходите security review клиента, не выдумывая документацию на ходу; меняете провайдера за вечер, потому что всё и так идёт через один модуль; отвечаете на вопрос “что именно вы отправили” SQL-запросом, а не догадкой.

Начните с gateway и grep-проверки. Даже без маскирования вообще: одна точка выхода превращает размытый риск в починяемый. Всё остальное прикручивается к ней.

Вопросы и ответы

Не проще ли просто подписать DPA с провайдером и не заморачиваться?

Договор определяет, что вендор обязан делать с данными. Он не отвечает на вопрос, какие поля вы ему отправили и когда. На security review клиента и при разборе инцидента спрашивают именно второе. Плюс DPA не покрывает вторых и третьих вендоров, которые появляются сами собой: observability для промптов, векторную базу, трекер ошибок. Gateway с журналом egress нужен ровно для того, чтобы у вас был свой источник правды, независимый от чужих обещаний.

Токенизация не ломает качество ответов модели?

На практике почти нет, если токены стабильны внутри запроса и содержат тип сущности. Модель нормально рассуждает про `<EMAIL_9f2c11>` и `<PHONE_4a1b02>`, потому что ей важна структура, а не конкретное значение. Деградация появляется в двух случаях: когда вы маскируете то, что нужно для рассуждения (например, название города при расчёте доставки), и когда токены выглядят как случайный мусор без подсказки о типе. Первое лечится точной разметкой чувствительности, второе - форматом токена.

У меня MVP и три пользователя. Это не оверинжиниринг на такой стадии?

Полный набор - да. Но минимум стоит полдня и окупается сразу: один модуль `llm`, через который идут все вызовы, grep-проверка в CI и лог с именами полей без значений. Этого достаточно, чтобы через полгода добавить маскирование и политики за вечер вместо недели рефакторинга. Дорогой становится не архитектура, а её отсутствие в момент, когда у вас двенадцать точек вызова и первый корпоративный клиент с опросником на сорок вопросов.

Похожие статьи