В предыдущей статье мы дали AI-агенту безопасное место для выполнения Python. Это ответило на вопрос, где выполняется написанный агентом код. Это продолжение отвечает на вопрос, который оказывается не менее важным: какие данные модели на самом деле нужно видеть?

Обычно ответ — намного меньше, чем мы ей даём.

Наивный паттерн

Очевидный способ подключить агента к бизнес-системе выглядит так:

  1. Инструмент делает запрос к системе и возвращает набор результатов в виде JSON.
  2. Строки попадают в контекст модели.
  3. Модель их читает и отвечает: «топ-пять это…», «среднее значение…».

Это работает в демо с двадцатью строками. На реальных данных это ломается тремя способами:

  • Безопасность. Каждая строка, возвращённая инструментом, теперь становится частью истории диалога. История сохраняется, логируется, повторно передаётся в следующие реплики, отправляется провайдеру модели, попадает в телеметрию и иногда сохраняется для оценки. Данные, которым никогда не нужно было покидать вашу систему, теперь скопированы во все эти места.
  • Стоимость и ёмкость. Строка с десятью полями — это примерно 40–60 токенов. Пять тысяч строк — это несколько сотен тысяч токенов: большая часть или всё окно контекста, за которое платят на каждой реплике, которая его несёт.
  • Корректность. LLM не вычисляют. Они оценивают. Попросите модель сложить 5000 чисел или ранжировать группы по изменению квартал к кварталу — и получите уверенный ответ, близкий к истине, но неверный. Тот же анализ на трёх строчках pandas даёт точный результат.

В итоге модель делает именно то, что у неё получается хуже всего (арифметику над большими объёмами данных), используя ресурс, которого у неё меньше всего (контекст), и при этом разбрасывает данные повсюду.

Паттерн: данные остаются на месте, модель получает дескриптор

Перевернём подход. Инструменты, производящие данные, не возвращают сами данные. Они сохраняют их на стороне сервера как артефакт и возвращают ссылку плюс ровно столько описания, сколько нужно модели, чтобы написать код для работы с ним:

+--------------------------- MCP server ----------------------------+
|                                                                   |
|  fetch_*  ---> [ artifact store ] <---> run_python (sandbox)      |
|                   ^        ^                 |                    |
|                   |        +---- outputs ----+                    |
|                   +---- render_chart / export                     |
|                                                                   |
+------------------------------|------------------------------------+
                               | only: dataRef, schema, row count,
                               |       tiny sample, small results
                               v
                       +---------------+
                       |  LLM / agent  |
                       +---------------+

Задача модели меняется с «прочитать данные и ответить» на «понять форму данных и написать код, который даст ответ». С этой задачей модель справляется хорошо.

Хранилище используют три вида инструментов:

  • Производители (fetch_*) выполняют запрос с правами вызывающего, сохраняют результат и возвращают описание.
  • Вычислительный инструмент (run_python) принимает ссылки в качестве входных данных. Внутри песочницы они оказываются DataFrame’ами. Любые DataFrame, сохранённые скриптом, становятся новыми артефактами, а модели возвращается только небольшой JSON-результат.
  • Потребители (render_chart, export_csv) берут ссылку и превращают её во что-то, на что смотрит человек. Строки идут из хранилища прямо на экран пользователя, минуя модель.

Минимальная реализация

В примерах, как и в прошлой статье, используются FastMCP и pandas. Они намеренно небольшие.

Хранилище

import secrets
import time
from dataclasses import dataclass, field

import pandas as pd


@dataclass
class Artifact:
    owner: str
    df: pd.DataFrame
    created: float = field(default_factory=time.monotonic)


class ArtifactStore:
    def __init__(self, ttl_seconds: int = 24 * 3600, max_bytes: int = 50_000_000):
        self._items: dict[str, Artifact] = {}
        self._ttl = ttl_seconds
        self._max_bytes = max_bytes

    def put(self, owner: str, df: pd.DataFrame) -> str:
        if df.memory_usage(deep=True).sum() > self._max_bytes:
            raise ValueError("artifact exceeds size cap - narrow the query")
        ref = "data_" + secrets.token_urlsafe(12)
        self._items[ref] = Artifact(owner, df)
        return ref

    def get(self, owner: str, ref: str) -> pd.DataFrame:
        now = time.monotonic()
        self._items = {k: a for k, a in self._items.items() if now - a.created < self._ttl}
        artifact = self._items.get(ref)
        if artifact is None or artifact.owner != owner:
            raise KeyError(f"unknown or expired dataRef {ref} - fetch the data again")
        return artifact.df

Здесь важны две детали. Ссылки случайные, а не последовательные, и каждое чтение проверяет владельца. Ссылка, оказавшаяся в чужой истории диалога, бесполезна для этого чужого человека. Сообщение об ошибке также подсказывает агенту, как восстановиться.

Описывать, а не вываливать всё

import json


def describe(ref: str, df: pd.DataFrame, sample_rows: int = 5) -> dict:
    return {
        "dataRef": ref,
        "rows": len(df),
        "columns": {c: str(t) for c, t in df.dtypes.items()},
        "nulls": {c: int(n) for c, n in df.isna().sum().items() if n},
        "sample": json.loads(df.head(sample_rows).to_json(orient="records", date_format="iso")),
    }

Имена колонок, типы, количество null-значений и горстка строк — вот что нужно человеку, чтобы написать groupby. Модели нужно то же самое.

Инструменты-производители и вычислительный инструмент

from fastmcp import FastMCP, Context

mcp = FastMCP("data-sandbox")
store = ArtifactStore()


def owner_of(ctx: Context) -> str:
    ...  # resolve the authenticated caller from your auth layer


@mcp.tool
async def fetch_sales(since: str, ctx: Context) -> dict:
    """Load sales since a date into a server-side artifact.
    Returns a dataRef with schema and a small sample, NOT the rows.
    Analyse the data with run_python."""
    df = await query_sales(since=since)  # your system of record, under the caller's permissions
    return describe(store.put(owner_of(ctx), df), df)


@mcp.tool
async def run_python(code: str, inputs: dict[str, str], ctx: Context) -> dict:
    """Run Python in the sandbox. `inputs` maps variable names to dataRefs;
    each arrives as a pandas DataFrame. Store DataFrames in `outputs[name]`
    to keep them as new artifacts. Assign a SMALL JSON value to `result`."""
    owner = owner_of(ctx)
    with tempfile.TemporaryDirectory() as run:
        in_dir, out_dir = Path(run, "in"), Path(run, "out")
        in_dir.mkdir()
        out_dir.mkdir()
        for name, ref in inputs.items():
            if not name.isidentifier():
                raise ValueError(f"input name {name!r} must be a Python identifier")
            store.get(owner, ref).to_parquet(in_dir / f"{name}.parquet")

        res = await execute_sandboxed(code, in_dir, out_dir)  # the sandbox from part one

        new = {p.stem: pd.read_parquet(p) for p in out_dir.glob("*.parquet")}
        outputs = {name: describe(store.put(owner, df), df, sample_rows=3) for name, df in new.items()}

    return {"ok": res.ok, "result": cap_json(res.result), "stdout": truncate(res.stdout),
            "error": res.error, "outputs": outputs}

Внутри песочницы короткий пролог загружает каждый in/*.parquet в глобальную переменную и создаёт пустой словарь outputs. Эпилог записывает outputs в out/ и сериализует result. У песочницы по-прежнему нет сети, и она видит только свою рабочую директорию, так что данные не смогут утечь и через скрипт. (Если в вашем рантайме нет pyarrow, подойдёт и CSV. Просто потеряете информацию о типах колонок.)

Как это выглядит в истории диалога

user:  Which five product lines lost the most revenue quarter over quarter?

agent -> fetch_sales(since="2025-01-01")
      <- {dataRef: "data_x7Kq...", rows: 48210, columns: {...}, sample: [5 rows]}

agent -> run_python(inputs={"sales": "data_x7Kq..."}, code="""
           q = sales.assign(q=sales.date.dt.to_period("Q"))
           pivot = q.pivot_table(index="line", columns="q", values="revenue", aggfunc="sum")
           delta = (pivot.iloc[:, -1] - pivot.iloc[:, -2]).sort_values()
           outputs["qoq"] = delta.rename("delta").reset_index()
           result = delta.head(5).round(2).to_dict()
         """)
      <- {result: {"Line A": -182340.5, ...}, outputs: {"qoq": {dataRef: "data_Pm2...", rows: 312}}}

agent: The five biggest drops were ...

... several turns later ...

user:  Chart all of them.
agent -> render_chart(dataRef="data_Pm2...", x="line", y="delta", kind="bar")

В обработке участвовало 48 210 строк. До истории диалога дошло около двух килобайт. Ответ точен, потому что его вычислил pandas, а не модель.

Артефакты между шагами и репликами

Поскольку ссылка — это короткая строка, она перемещается по диалогу так же, как любое другое значение:

  • Цепочки вызовов инструментов. outputs одного вызова run_python становятся inputs следующего. Многошаговый анализ (очистка, соединение, агрегация, ранжирование) — это цепочка ссылок. Промежуточные таблицы никогда не проходят через контекст.
  • Более поздние реплики. Ссылка из реплики 2 всё ещё в истории диалога на реплике 9. «А теперь разбей это по регионам» переиспользует сохранённый артефакт. Ничего не запрашивается заново, данные не вставляются повторно.
  • Истечение срока — часть архитектуры. TTL держит хранилище ограниченным по размеру. Просроченная ссылка возвращает ошибку, которая говорит агенту запросить данные заново, так что устаревший дескриптор падает чисто, а не тихо возвращает старые данные.

Укажите это агенту в инструкциях: никогда не просить сырые строки, чтобы ответить на вопрос; загружать в артефакт, вычислять через run_python и строить графики по dataRef.

Правила, на которых это держится

  • Возвращайте форму данных, а не их содержимое. Схему, количество строк, количество null-значений и несколько примеров строк. Сделайте размер выборки настраиваемым и установите его в ноль для чувствительных наборов данных.
  • Держите всё, что возвращается модели, маленьким, и обеспечивайте это принудительно. Ограничивайте размер result и обрезайте stdout. Иначе один print(df) отправит модели всю таблицу целиком.
  • Каждая ссылка принадлежит одному владельцу. Используйте случайные ID и проверяйте владельца при каждом чтении. Ссылка — это не право доступа.
  • Производители соблюдают правила исходной системы. Артефакт содержит ровно то, что вызывающему было разрешено получить: та же авторизация, те же лимиты по строкам.
  • Ограничивайте хранилище. Лимит байт на артефакт, TTL и поиск в рамках владельца. Оно держит рабочие данные какое-то время. Это не хранилище данных.
  • Дайте каждому отображающему инструменту принимать ссылки. Графики, таблицы и экспорт читают прямо из хранилища, так что шаг «покажи мне» тоже не прогоняет строки через модель.

Честные компромиссы

Этот подход снижает риск утечки. Он не устраняет его полностью. Примеры строк и агрегаты — всё ещё данные. Группа из одного элемента — это, по сути, отдельная запись, и настойчивый промпт всё ещё может попросить агента вывести строки в пределах лимита вывода. Относитесь к лимиту результата и размеру выборки как к настройкам политики, а настоящей границей считайте авторизацию исходной системы.

Модель также больше не видит данные напрямую. Выборка из пяти строк может вводить в заблуждение: странные значения, смешанные единицы измерения, колонка на 90% из null. Дешёвый инструмент describe_data(dataRef), возвращающий сводную статистику по запросу, решает большую часть этих проблем без возврата полного дампа.

Итог

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