前回の記事では、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 が計算をし、データはあなたのシステムに留まります。会話記録に残るのは、質問とコードと答え。分析の再現可能な記録でありながら、その分析が扱ったデータは一切含まれていません。