第5章 エージェントループ — 解答例

本文: 第5章 エージェントループ

問1

→ 参照: 5.3.2節 / 5.5節 / 1.2.2節 / 第12章

(a) 明示的な終了手段がない

問題点: 終了判定が「ツールを呼ばなかった」という消極的な条件に依存している。このため、(1) モデルが単に相槌を返しただけの場合と、作業を完遂した場合を区別できない、(2) 「解決できなかった」という状態を表現できず、失敗が成功として返る、(3) 返るのが自由文のテキストだけなので、呼び出し側が結果を構造的に扱えない。

修正方針: 5.5節の finish ツールを登録し、status(completed / failed / needs_user_input)・summary・result を必須引数にする。ループ側で finish を捕捉して終了させ、構造化された結果を返す。ツール未使用による終了はフォールバックとして残す(モデルが finish を呼び忘れる場合があるため)。

Python

for tu in tool_uses:
    if tu.name == "finish":
        return FinishResult(**tu.input)      # status / summary / result

TypeScript

for (const tu of toolUses) {
  if (tu.name === 'finish') {
    // Python の FinishResult(**tu.input) はキーワード展開で構築するが、
    // JavaScript に相当物はない。input をそのまま受ける
    return tu.input as FinishResult // status / summary / result
  }
}

さらに、客観的に検証できる完了条件を持てる課題(テストが通る、スキーマ検証を通過する)なら、finish よりもそちらを優先すべきである(5.5節)。

(b) ツールが逐次実行されている

問題点: 1回のレスポンスに複数の tool_use が含まれるのに、for ループで順番に実行している。エージェントの実行時間の大半はI/O待ちであり(1.2.2節)、3つのツールを直列に呼ぶと待ち時間が3倍になる。並列ツール使用というモデル側の能力を活かせていない。

修正方針: 独立したツール呼び出しを並列実行する。tool_result は tool_use_id で紐づくので順序に依存しないが、実装上は元の順序を保っておくのが安全である。

Python

import asyncio

async def execute_all(registry: ToolRegistry, tool_uses) -> list[dict]:
    async def run(tu):
        # 同期ハンドラをスレッドに逃がす。ハンドラが async なら直接 await する
        outcome = await asyncio.to_thread(registry.execute, tu.name, tu.input)
        return {
            "type": "tool_result",
            "tool_use_id": tu.id,
            "content": outcome.content,
            "is_error": outcome.is_error,
        }
    return list(await asyncio.gather(*(run(tu) for tu in tool_uses)))

TypeScript

async function executeAll(
  registry: ToolRegistry,
  toolUses: Anthropic.ToolUseBlock[]
): Promise<Anthropic.ToolResultBlockParam[]> {
  const run = async (
    tu: Anthropic.ToolUseBlock
  ): Promise<Anthropic.ToolResultBlockParam> => {
    // Python は同期ハンドラを asyncio.to_thread でスレッドに逃がすが、
    // JavaScript は単一スレッドで、execute を await すれば同期・非同期の
    // どちらのハンドラも扱える
    const outcome = await registry.execute(
      tu.name,
      tu.input as Record<string, unknown>
    )
    return {
      type: 'tool_result',
      tool_use_id: tu.id,
      content: outcome.content,
      is_error: outcome.isError,
    }
  }
  // Promise.all は入力の順序どおりに結果を返すため、元の順序が保たれる
  return Promise.all(toolUses.map(run))
}

注意: 並列化してよいのは副作用のないツール、または互いに独立なツールに限る。書き込み系が同一リソースに触る場合は順序依存があるため、逐次実行にフォールバックする判定(ツール定義に parallel_safe フラグを持たせる等)を入れる。あわせて、各ツールに個別のタイムアウトを設けること(5.2.3節)。

(c) 各ステップの記録がない

問題点: 20ステップ動いた末に失敗しても、何が起きたか再現できない。エージェントは実行経路が毎回変わるため、ログがないとデバッグが原理的に不可能である(4.3.3節「デバッグが難しい」)。コスト上限(5.4.1節)の判定に必要な usage も記録されていない。

修正方針: ステップ単位の構造化ログを取り、タスク全体を1本のトレースとして追えるようにする(第12章)。

Python

@dataclass
class StepRecord:
    trace_id: str
    iteration: int
    tool_name: str | None
    arguments: dict[str, Any] | None
    is_error: bool
    duration_ms: float
    input_tokens: int
    cache_read_tokens: int
    output_tokens: int
    stop_reason: str | None

TypeScript

interface StepRecord {
  traceId: string
  iteration: number
  toolName: string | null
  arguments: Record<string, unknown> | null
  isError: boolean
  durationMs: number
  inputTokens: number
  cacheReadTokens: number
  outputTokens: number
  stopReason: string | null
}

記録すべき最低限は、LLMに送った入力・返ってきた content 全体・実行したツールと引数・結果と成否・所要時間・トークン使用量である。特に response.content はそのまま保存する(思考ブロックやツール呼び出しの文脈が失われるため、テキストだけ抜き出さない — 3.3.4節)。APIキーや個人情報がログに残らないようマスキングを入れること(第13章)。

---

問2

→ 参照: 5.4.2節

StallDetector は「呼び出しの同一性」しか見ていない。設問の3つは、判定軸を 呼び出し → 結果 → 状態 へ上げることで検知できる。この一般則が本問の要点である。

(a) ツールAとBを交互に呼び続ける

単一の呼び出しではなく呼び出し列の周期性を見る必要がある。直近N回の (ツール名, 引数) を系列として保持し、長さ k(k = 2, 3, 4…)の窓を取ってハッシュ化し、同じ窓が閾値回数以上出現したら停滞と判定する。要するに1-gramでの一致判定を n-gram に拡張する。

Python

def is_cycling(history: list[str], k: int = 2, threshold: int = 3) -> bool:
    """直近 k 件の並びが threshold 回以上繰り返されていれば True。"""
    if len(history) < k * threshold:
        return False
    window = history[-k:]
    return all(history[-k * (i + 1): len(history) - k * i] == window
               for i in range(threshold))

TypeScript

/** 直近 k 件の並びが threshold 回以上繰り返されていれば true。 */
function isCycling(
  history: string[],
  k: number = 2,
  threshold: number = 3
): boolean {
  if (history.length < k * threshold) {
    return false
  }
  const window = history.slice(-k)
  // Python はリスト同士を == で比較できるが、JavaScript の配列は参照比較になる。
  // 要素に現れない NUL 文字で連結し、文字列として突き合わせる
  const key = window.join('\u0000')
  return Array.from({ length: threshold }, (_v, i) =>
    history.slice(history.length - k * (i + 1), history.length - k * i).join('\u0000')
  ).every((slice) => slice === key)
}

(b) クエリを毎回わずかに変えながら同じ内容を検索し続ける

引数の文字列が違うので厳密一致では捕まらない。2段階で対処する。

• 軽い方法(推奨): 引数ではなくツールの実行結果をハッシュ化して比較する。クエリが違っても返ってくる文書集合が同じなら、実質的に何も進んでいない。計算コストがほぼゼロで効果が高い。
• 重い方法: クエリを正規化(小文字化・空白除去・記号除去)したうえで比較する。さらに埋め込み(3.5節)を取り、過去のクエリとのコサイン類似度が閾値(例: 0.95)を超えたら「実質同一」とみなす。精度は上がるが、判定のたびに埋め込みAPIを呼ぶコストがかかる。

(c) ファイルを編集しては元に戻す

呼び出しも結果も毎回異なるため、呼び出し側の観測では検知できない。環境の状態を見るしかない。作業対象ファイル(群)の内容ハッシュを毎ステップ記録し、過去に出現した状態ハッシュに戻ったら循環と判定する。

Python

state_hashes: dict[str, int] = {}

h = hashlib.sha256(workspace_snapshot()).hexdigest()
state_hashes[h] = state_hashes.get(h, 0) + 1
if state_hashes[h] >= 3:
    inject_system_message("同じ状態に3回戻っています。……")

TypeScript

const stateHashes = new Map<string, number>()

const h = createHash('sha256').update(workspaceSnapshot()).digest('hex')
stateHashes.set(h, (stateHashes.get(h) ?? 0) + 1)
if ((stateHashes.get(h) ?? 0) >= 3) {
  injectSystemMessage('同じ状態に3回戻っています。……')
}

Gitで作業させている場合は、ワークツリーのハッシュ(git stash create や git write-tree の出力)をそのまま使えるので実装が容易である。

共通の対処: いずれの検知でも、打ち切るのではなくまずシステム介入メッセージを注入して気づかせるのが5.4.2節の方針である。「同じ状態に戻っています。別のアプローチを検討するか、何が障害になっているかを述べて作業を終了してください」と伝えると、モデルは自力で方針を変えるか、原因を報告して終了する。

---

問3

→ 参照: 5.5節 / 5.4.4節 / 5.7.1節(+ 7.5.3節)

(a) 終了条件の設計

多層で持つ。

種別	条件	位置づけ
正常終了(主)	finish ツールが呼ばれる。status は approved / rejected / needs_more_info の enum、reason に根拠とした規程の条番号を必須	◎ 構造化された結果が得られ、「判断保留」も表現できる
正常終了(客観的検証)	決定論的な規程チェッカー(金額上限・領収書添付・勘定科目の妥当性・期限内申請)がすべて合格を返す	◎ 機械的に検証できる部分はモデルの自己申告より確実。コードで書ける規則はコードで書く
安全弁	ステップ数・累計コスト・実時間の上限	△ 発動したら設計を見直す。到達時は「ここまでで判明したこと」を回収する(5.4.1節)
安全弁	停滞検知(同じ申請を何度も照会し続ける等)	△ 同上
異常終了	回復不能なエラー(申請IDが存在しない、当該申請への権限がない)	△ 再試行が無意味な場合は即座に終了し、人間に回す
人間の判断	一定金額以上、または規程に前例のない案件は自動判定せず人間に委ねる	○ 経費承認は副作用が大きく、HITLが必須

要点は、規程チェックのうち機械化できる部分をLLMに判断させないことである。「上限3万円」「領収書必須」といった規則は決定論的に判定でき、そのほうが速く安く確実である。LLMに任せるのは、規程の解釈が必要なグレーゾーンだけにする。

(b) 「副作用を後回しにする」の適用

このユースケースの副作用は 承認処理の確定(会計システムへの反映・支払処理の起動) と 申請者への差し戻し通知 である。

• 読み取り系を先に全部済ませる: 申請内容の取得、規程の検索、過去の類似申請の照会、予算残高の確認、領収書画像の読み取り。ここまでは何度失敗しても巻き戻しコストがゼロである。
• 判定は「案」として作る: draft_decision(判定案・根拠条文・差し戻し理由を組み立てるだけ、副作用なし)と apply_decision(実際に承認・差し戻しを確定する)を別ツールに分ける。5.7.1節の draft_email / send_email 分離と同じ構造。
• 確定は最後に1回だけ: 全項目の検証が終わってから apply_decision を呼ぶ。途中で誤りが見つかっても、確定していないので巻き戻しが不要。
• 冪等キーを必須にする: apply_decision に申請IDから決定的に導出した冪等キーを持たせ、再試行による二重承認・二重通知を防ぐ(7.5.3節)。
• 確認用引数を必須にする: 申請IDだけでなく申請者名と金額も必須引数にし、不一致ならエラーを返す(ポカヨケ、4.4.5節)。申請の取り違えという最も起きやすい事故を構造的に防げる。

(c) 誤りの累積を防ぐ検証ステップ

1. 金額の計算はツールに委譲する。合計額・税額・按分は calculate ツールで計算させ、モデルに暗算させない(第2章2.2.2節)。金額の誤りは以降のすべての判断を汚染する典型的な累積誤りである。
2. 根拠条文の実在検証。判定の根拠として出力させた条番号を、規程DBに照会して実在するか・引用文が一致するかを機械的に確認する。存在しない条文を根拠にしていたら差し戻して再検討させる。これは対象を規程に限定したハルシネーション検出であり、実装コストが低い割に効果が大きい。
3. 対象の同一性の再確認。処理の最後、apply_decision の直前に申請IDから申請内容をもう一度取得し直し、序盤に読み取った内容(申請者・金額・日付)と一致するかを突き合わせる。序盤の取り違えがそのまま確定するのを防ぐ。
4. 節目での振り返り(5.6節 方式C)。検証項目のチェックリスト(規程適合・予算残高・領収書・重複申請)を全部埋めたかを、確定前に一度確認させる。

まとめ: (a)〜(c) を通じた原則は、機械的に検証できることはコードで検証し、LLMには解釈が必要な部分だけを任せること、そして副作用は最後に1回だけ、冪等に、確認つきで実行することである。

---

問4

→ 参照: 5.4.1節 / 2.2.2節 / 2.4節 / 3.7.2節

(a) プロンプトキャッシュなし

• 入力累計 = 30 ステップ × 12,000 = 360,000 トークン
• 入力コスト = 360,000 / 1,000,000 × $3 = $1.0800
• 出力累計 = 30 × 800 = 24,000 トークン → 24,000 / 1,000,000 × $15 = $0.3600
• 1タスクあたり $1.4400

(b) 固定部分6,000トークンに5分キャッシュ

固定部分を切り出し、残り 12,000 − 6,000 = 6,000トークン/ステップは毎回異なる内容として通常課金される。

• キャッシュ書き込み(1回目): 6,000 × 1.25 = 7,500 トークン相当
• キャッシュ読み出し(2〜30ステップ目の29回): 6,000 × 29 × 0.1 = 17,400 トークン相当
• キャッシュ対象外: 6,000 × 30 = 180,000 トークン
• 実効入力 = 7,500 + 17,400 + 180,000 = 204,900 トークン相当
• 入力コスト = 204,900 / 1,000,000 × $3 = $0.6147
• 出力は変わらず $0.3600
• 1タスクあたり $0.9747(削減率 (1.44 − 0.9747) / 1.44 = 32.3%)

月額(1日500タスク × 30日 = 15,000タスク)

• 15,000 × $0.9747 = $14,620.50 / 月

考察

• 入力側だけで $1.08 → $0.61 と約43%削減され、総額では32.3%削減になる。削減率が総額で薄まるのは、出力コスト($0.36)がキャッシュの恩恵を受けないためである。第2章2.4節の「コストの85%が入力側」という構造は、キャッシュ適用後は入力63%・出力37%へと変わり、次に効く最適化が変わることを意味する。
• 月額$14,620は無視できない金額であり、3.7.2節が言う「プロトタイプの段階でコストモデルを作っておく」ことの重要性が具体的に表れている。
• さらなる削減の方向: ①キャッシュのブレークポイントを会話履歴の末尾へ移し、蓄積したツール結果もキャッシュ対象に含める(2.4節)、②ツール結果を切り詰めて可変部分6,000トークン/ステップを減らす(5.4.3節 第1層)、③要約・整形など単純なステップを軽量モデルや低 effort に落とす(3.3.3節)、④平均30ステップという数字そのものを減らす — ツールの粒度を上げてステップ数を減らせば、コストと同時に誤りの累積(5.4.4節)も改善する。

注記: 5分キャッシュのTTL内に次のステップが走ることを前提にしている。コード修正エージェントは連続してループを回すため通常は成立するが、テスト実行など長時間のツールを挟むとキャッシュが失効しうる。その場合は1時間キャッシュ(書き込み2.0倍)の採用を検討する(2.2.2節: 2回の再利用で損益分岐)。

---

問5

→ 参照: 5.5節 / 5.7節

客観的に検証できる完了条件を持てるのは、コード生成・修正と、条件付きでWebスクレイピングである。

ユースケース	客観的完了条件	内容
コード生成・修正	◎ 持てる	テストスイートが通ること。機械的な判定器が存在し、モデルの自己申告が不要。5.7.2節が「エージェントが最も成功しやすい領域」と述べる理由
Webスクレイピング	○ 持てる	収集対象リストの全件について、必要フィールドがスキーマを満たして埋まっていること。充足率で機械判定できる
データ分析	△ 部分的	「分析が十分か」は主観だが、一部は検証可能
パーソナルアシスタント	✗ 持てない	「良い返信案か」「適切な予定か」に機械的な正解がない
NPC / ゲームAI	✗ 持てない	そもそも「完了」の概念が薄い(1ターン分の行動決定で終わる)

持てないものへの工夫

データ分析(5.7.3節) 出力に数値の根拠となるクエリを必ず含めさせ、アプリケーション側でそのクエリを独立に再実行して、報告された数値と一致するかを照合する。一致しなければ差し戻す。これで「分析の妥当性」は測れなくとも「報告された数値の正しさ」は機械的に検証できる。加えて、検証項目のチェックリスト(対象期間・フィルタ条件・欠損値の扱い・外れ値の確認)を定義し、全項目が埋まったことを完了条件にする。5.7.3節が「分析結果は誤っていてももっともらしく見える」と警告しているとおり、人間が検証できる形で出させることが要点である。

パーソナルアシスタント(5.7.1節) ユーザー確認を完了条件に組み込む。finish ツールに status: needs_user_input を用意し、下書きを提示して承認を得た時点を完了とする。加えて、副作用のある操作については「下書き作成まで」を完了条件とし、送信・確定は人間の操作に委ねる。完了の判定をモデルから人間に移すのが、この領域で最も信頼できる方法である。

NPC / ゲームAI(5.7.5節) そもそも長いループを回さないので、完了条件よりも出力の妥当性検証が課題になる。行動を enum(移動 / 攻撃 / 発話 / 待機)で固定し、返された行動がその状況で実行可能な行動集合に含まれるかをスキーマ検証と状態チェックで判定する。不正なら従来のゲームAI(ステートマシン、ビヘイビアツリー)の既定行動にフォールバックする。「1回の呼び出しで1ターン分」という構造的な上限が、実質的に完了条件の代わりを果たしている。

一般則: 5.7.6節が述べるとおり、客観的な完了条件を持てるかどうかがエージェントの成功率を大きく左右する。持てない場合は、(1) 完了判定を人間に委ねる、(2) 検証可能な部分だけでも機械判定する、(3) 構造的な上限(ターン数、行動集合)で代替する、という3つの方向で信頼性を補う。

---