この記事は筆者のブログ LynxLine からの転載です。英語版はこちら。

前回の記事では、AI エージェントが Python を安全に実行できる場所を用意しました。これで、エージェントが書いたコードをどこで実行するかには答えが出ました。今回はその続編として、同じくらい重要だとわかった問いに答えます。モデルは実際のところ、どのデータを見る必要があるのか？

たいていの場合、その答えは私たちが渡している量よりずっと少ないのです。

素朴なパターン

エージェントを業務システムにつなぐとき、真っ先に思いつく方法はこうです。

ツールがシステムに問い合わせ、結果セットを JSON で返す。
その行がモデルのコンテキストに入る。
モデルがそれを読んで答える。「上位 5 件は……」「平均は……」。

行が 20 件のデモならこれで動きます。実データでは、3 つの点で破綻します。

セキュリティ。 ツールが返した行はすべて会話記録の一部になります。会話記録は保存され、ログに残り、後のターンで再びコンテキストに入り、モデルのプロバイダーに送られ、テレメトリーに拾われ、ときには評価用に保管されます。システムの外に出る必要がまったくなかったデータが、そのすべての場所にコピーされてしまったわけです。
コストと容量。 フィールドが 10 個ある行は、おおよそ 40〜60 トークンになります。5,000 行なら数十万トークン。コンテキストウィンドウの大半、あるいはすべてを占め、それを持ち回るターンのたびに料金がかかります。
正確さ。 LLM は計算をしません。推定をします。5,000 個の数値の合計や、前四半期比の変化によるグループの順位付けをモデルに頼めば、自信満々で、惜しいけれど間違った答えが返ってきます。同じ分析は pandas なら 3 行で、正確な結果が出ます。

結局モデルは、自分が最も苦手なこと（大量データの算術）を、自分が最も少ししか持っていない資源（コンテキスト）を使ってやらされ、しかもその過程でデータをあちこちにばらまいているのです。

パターン：データはその場に留め、モデルにはハンドルを渡す

発想を逆にします。データを生み出すツールはデータを返しません。データをサーバー側にアーティファクトとして保存し、参照と、モデルがそれに対してコードを書くのに必要なだけの説明を返します。

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

モデルの仕事は、「データを読んで答える」から「データの形を理解し、答えを出すコードを書く」に変わります。こちらはモデルの得意な仕事です。

ストアを使うツールは 3 種類あります。

プロデューサー（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

ここで大事な点が 2 つあります。参照は連番ではなくランダムであること、そして読み出しのたびに所有者を確認することです。参照が他人の会話記録に紛れ込んでも、その人にとっては役に立ちません。また、エラーメッセージがエージェントに復旧方法を伝えている点にも注目してください。

中身を出さず、説明する

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")

（ユーザーの質問は「前四半期比で売上が最も落ちた製品ラインの上位 5 つは？」、数ターン後に「全部グラフにして」です。）

処理に関わった行は 48,210 行。会話記録に届いたのは約 2 KB です。答えが正確なのは、モデルではなく pandas が計算したからです。

ステップやターンをまたぐアーティファクト

参照は短い文字列なので、他の値と同じように会話の中を移動できます。

ツール呼び出しの連鎖。 ある run_python 呼び出しの outputs が、次の呼び出しの inputs になります。複数ステップの分析（クリーニング、結合、集計、順位付け）は参照の連鎖です。途中のテーブルは一度もコンテキストを通りません。
後のターン。 2 ターン目の参照は、9 ターン目でも会話記録に残っています。「じゃあそれを地域別に分けて」は保存済みのアーティファクトを再利用します。再取得もなく、データを貼り直すこともありません。
有効期限も設計の一部。 TTL によってストアの大きさは一定に保たれます。期限切れの参照は、再取得するようエージェントに伝えるエラーを返します。古いハンドルは、古いデータを黙って返すのではなく、はっきりと失敗します。

これはエージェントへの指示に書いておきましょう。質問に答えるために生の行を要求しないこと。データはアーティファクトとして取得し、run_python で計算し、グラフは dataRef で描くこと。

このパターンを保つためのルール

データの中身ではなく形を返す。 スキーマ、行数、null の数、数行のサンプル。サンプル数は設定可能にし、機密性の高いデータセットではゼロにします。
モデルに返すものはすべて小さく保ち、それを強制する。 result のサイズに上限を設け、stdout は切り詰めます。そうしないと、print(df) 一つでテーブル全体がモデルに送り返されます。
参照はそれぞれ一人の所有者に属する。 ランダムな ID を使い、読み出しのたびに所有者を確認します。参照を持っていること自体は権限ではありません。
プロデューサーは元システムのルールを守る。 アーティファクトに入るのは、呼び出し元が取得を許されていたものだけ。認可も行数の制限も同じです。
ストアに制限を設ける。 アーティファクトごとのバイト数上限、TTL、所有者単位での検索。作業用データを一時的に置く場所であって、データウェアハウスではありません。
表示に関わるものはすべて参照を受け取れるようにする。 グラフ、表、エクスポートはストアから読み出します。そうすれば「見せて」の段階でも、行がモデルを通ることはありません。

正直なトレードオフ

このアプローチはデータの露出を減らしますが、なくすわけではありません。サンプル行も集計値もデータです。要素が 1 件しかないグループは、1 件のレコードそのものです。それに、粘り強いプロンプトなら、出力の上限の範囲内で行を表示するようエージェントに頼むこともできます。結果の上限とサンプル数はポリシーとして扱い、本当の境界は元システムの認可に置き続けてください。

また、モデルはもうデータを直接見られません。5 行のサンプルは誤解を招くことがあります。おかしな値、単位の混在、90% が null のカラムなど。要求に応じて要約統計を返す安価な describe_data(dataRef) ツールがあれば、データを丸ごと戻さずに、こうした問題の大半に対処できます。

まとめ

前回のサンドボックスは、エージェントのコードを安全に実行できるようにしました。データアーティファクトは、それを実行する価値のあるものにします。モデルは計画を立ててコードを書き、Python が計算をし、データはあなたのシステムに留まります。会話記録に残るのは、質問とコードと答え。分析の再現可能な記録でありながら、その分析が扱ったデータは一切含まれていません。