AIエージェントのライブラリを触ると、少ないコードで検索や操作が動きます。一方、失敗したときに「なぜ同じツールを繰り返したか」「どこで止めるか」が見えにくくなります。基本構造は、モデルが行動を選び、アプリが検証して実行し、その観測をモデルへ戻すループです[1][2]。
この記事では、APIキーも外部パッケージも不要な最小実装を作ります。実モデルの代わりに決定的な偽モデルを使うため、そのまま実行してテストできます。完成版はサンプルコードからも参照できます。
実装する境界を先に決める
モデルに任せるのは「次に何をしたいか」の提案までです。実行権限はアプリケーション側に残します。
| 責務 | モデル | アプリケーション |
|---|---|---|
| 次の行動を提案 | ○ | — |
| ツールの許可判定 | — | ○ |
| 引数の型・業務検証 | — | ○ |
| APIやDBの実行 | — | ○ |
| 最大ステップで停止 | — | ○ |
| 最終回答の作成 | ○ | 形式検証 |
この分離が、AIエージェントの7パターンで説明した「自律性は権限」という考え方の実装です。モデルがdelete_customerと出力しても、許可リストになければ実行されません。
判断の型とモデルの境界を作る
最初に、モデルの出力を「ツールを呼ぶ」か「最終回答を返す」かの2種類に絞ります。実APIではネイティブのTool CallingやJSON Schemaをこの型へ変換します。
from dataclasses import dataclass
from typing import Any, Protocol
@dataclass(frozen=True)
class Decision:
kind: str
tool_name: str | None = None
arguments: dict[str, Any] | None = None
answer: str | None = None
class Model(Protocol):
def decide(self, messages: list[dict[str, Any]]) -> Decision: ...
モデルAPI固有のレスポンスをこの境界より内側へ漏らさないのがポイントです。提供者を変えても、エージェントループとツールのテストは再利用できます。フィールドの必須条件や未知キーの拒否は構造化出力パターンと同じ方針で検証します。
ツール実行ループを書く
次が本体です。ツールは名前から関数へ引く許可リストにし、max_stepsを必須にします。ツールの引数不備は観測としてモデルへ返しますが、未許可ツールは権限エラーとして即停止します。
from typing import Callable
Tool = Callable[..., str]
def run_agent(
task: str,
model: Model,
tools: dict[str, Tool],
*,
max_steps: int = 6,
) -> str:
messages = [{"role": "user", "content": task}]
for step in range(1, max_steps + 1):
decision = model.decide(messages)
if decision.kind == "final" and decision.answer:
return decision.answer
if decision.kind != "tool" or not decision.tool_name:
raise ValueError(f"invalid decision at step {step}: {decision}")
tool = tools.get(decision.tool_name)
if tool is None:
raise PermissionError(f"tool is not allowed: {decision.tool_name}")
arguments = decision.arguments or {}
try:
observation = tool(**arguments)
except (TypeError, ValueError) as exc:
observation = f"tool_error: {type(exc).__name__}: {exc}"
messages.extend([
{
"role": "assistant",
"tool_call": {
"name": decision.tool_name,
"arguments": arguments,
},
},
{
"role": "tool",
"name": decision.tool_name,
"content": observation,
},
])
raise RuntimeError(f"agent exceeded max_steps={max_steps}")
観測は文字列に統一していますが、本番ではstatus、data、error_code、retryableを持つ構造にすると分岐しやすくなります。巨大な検索結果をそのまま履歴へ積まず、上限件数と最大文字数をツール側で決めます。
API不要の偽モデルで動作確認する
在庫検索を1回呼び、その結果を最終回答にする偽モデルを作ります。LLMの出力は非決定的でも、制御ループのテストは決定的にできます。
class ScriptedModel:
def __init__(self) -> None:
self.turn = 0
def decide(self, messages: list[dict[str, Any]]) -> Decision:
self.turn += 1
if self.turn == 1:
return Decision(
kind="tool",
tool_name="lookup_inventory",
arguments={"sku": "A-101"},
)
return Decision(
kind="final",
answer=f"在庫照会の結果: {messages[-1]['content']}",
)
def lookup_inventory(sku: str) -> str:
inventory = {"A-101": 12, "B-205": 0}
if sku not in inventory:
raise ValueError("unknown sku")
return f"{sku} は {inventory[sku]} 個"
answer = run_agent(
"A-101の在庫を確認して",
ScriptedModel(),
{"lookup_inventory": lookup_inventory},
)
assert answer == "在庫照会の結果: A-101 は 12 個"
print(answer)
実行結果は在庫照会の結果: A-101 は 12 個です。偽モデルを差し替えて、未許可ツール、誤った引数、同一ツールの反復、最大ステップ超過もテストします。
実モデルへつなぐときの実装順序
- 提供者のTool Calling応答を
Decisionへ変換するアダプターを書く kind、ツール名、引数をスキーマ検証する- 読み取り専用ツール1個で、正解タスク20件を通す
- ツール選択精度、引数正解率、完了率、平均ステップ数を記録する
- 書き込み系は確認画面、冪等性キー、監査ログを加えてから許可する
MCPはモデルを外部ツールやデータへ接続する共通化に使えます[3]。ただし、接続方式が標準化されても権限設計はアプリケーションの責任です。MCPサーバーが公開する全ツールを、そのまま全ユーザーへ見せないようにします。
本番化で足すガードレール
- 予算: ステップ数だけでなく、累積トークン、経過時間、外部API費用で止める
- 権限: ユーザーとタスクに応じてツールを絞り、既定は拒否にする
- 副作用: 書き込み操作はdry-run、確認、冪等性、補償処理を用意する
- 観測: ツール名、引数のハッシュ、所要時間、結果件数、エラー分類を残す
- 入力防御: 検索文書内の命令をツール指示として扱わない
- 人手介入: 金額、公開、削除、権限変更は明示的な承認点を置く
内部の長い推論文をログへ保存するより、「どの入力で、どのツールを、どの検証後に実行し、何が返ったか」を構造化して残す方が、監査と再現に役立ちます。
まとめ
- エージェントの最小形は、判断→許可確認→実行→観測のループ
- モデルは行動を提案し、実行権限と停止条件はコード側に置く
- API固有レスポンスを
Decisionへ変換し、制御ループから分離する - 偽モデルを使えば、エージェント制御をAPI費用なしで決定的にテストできる
- 本番では予算、冪等性、監査ログ、人手承認を追加する
- マルチエージェント化は、単一ループの評価と制御ができてから行う
次の題材として、検索ツール自体の品質はRAGの精度改善ガイドで、複数の検索経路を使い分ける設計はGraphRAG入門で確認できます。
参考文献・一次情報
- [1]
- [2]
- [3]