状態管理と停止条件 - AIエージェントの仕組み

この回のゴール

エージェントが 無限ループ や 予算超過 に陥る原因を理解する
4 つの 停止条件 を使い分けられるようになる
状態トラッキング で「今どこにいるか」を把握する
本番運用で必須の 安全設計 を身につける

1. 暴走の実例

例 1: 無限ループ

Claude が同じツールを繰り返し呼ぶ:

POST https://api.anthropic.com/v1/messages

原因: - ツール結果を 使いこなせていない(理解ミス) - ツールの 戻り値が曖昧 で決定打にならない - system プロンプトが 「検索し続けろ」と示唆 している

例 2: 予算超過

エージェントは 毎反復でトークンを消費 します:

反復 1: 入力 500 tok / 出力 100 tok
反復 2: 入力 700 tok / 出力 150 tok(履歴追加で input 増)
反復 3: 入力 1000 tok / 出力 200 tok
...

10 反復もすると 合計 10,000 トークン 超え。長い調査タスクでは数十万トークンに。

例 3: 答えにならない脱線

RAG エージェントで:

try:
    result = tool_func(**tool_input)
except Exception as e:
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": ...,
        "content": f"Error: {e}",
        "is_error": True,    # 👈 明示的にエラーフラグ
    })

もういい加減にして回答してほしい のに掘り続ける。

2. 4 つの停止条件

① Claude 自身が `end_turn` を返す(理想)

consecutive_errors = 0
for it in range(max_iter):
    if consecutive_errors >= 3:
        # 3 連続エラーなら諦める
        break
    ...
    if any(r["is_error"] for r in tool_results):
        consecutive_errors += 1
    else:
        consecutive_errors = 0

これが 一番自然な停止。ただし常に期待通りに止まるとは限らない。

② 反復回数の上限 (max_iterations)

from anthropic import Anthropic
client = Anthropic()

response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1024,
    system="あなたは日本の歴史に詳しいアシスタントです。",  # 役割を指示
    messages=[
        {"role": "user",      "content": "桃太郎について教えて"},
        {"role": "assistant", "content": "桃太郎は日本の昔話の主人公です。"},
        {"role": "user",      "content": "続きを話して"},
    ],
)

安全弁。必ず設定すべき
目安: 簡単 5、普通 10〜15、複雑 30
大きすぎるとコスト暴発、小さすぎると途中打ち切り

③ トークン予算の上限

response.content                       # [ContentBlock(type="text", text="...")]
response.stop_reason                   # "end_turn" / "max_tokens" / "tool_use" ...
response.usage.input_tokens            # 入力トークン数
response.usage.output_tokens           # 出力トークン数
response.usage.cache_read_input_tokens # キャッシュ読み取り

コスト管理 の王道
例: 1 タスク 50,000 トークンまで → 超えたら打ち切り

④ 時間の上限

with client.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "長い文章を書いて"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

ユーザー待機を避ける
バッチ処理では長めで OK、対話型は 10〜30 秒 くらい

3. 「同じツール連続呼び出し検知」

追加のカスタム条件として:

response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": "<<非常に長い文脈>>",
        "cache_control": {"type": "ephemeral"}  # 👈 キャッシュマーカー
    }],
    messages=[{"role": "user", "content": question}],
)

response.usage.cache_read_input_tokens  # キャッシュから読んだトークン数

4. 状態を「見える化」する

エージェントの状態を トラッキング するための基本項目:

費用 = 入力トークン × 入力単価 + 出力トークン × 出力単価

これを毎反復で更新し、UI に表示 します(進捗バー、コスト計、etc)。

5. デッドロック的な状況とその対処

ツール実行エラー時

try:
    result = tool_func(**tool_input)
except Exception as e:
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": ...,
        "content": f"Error: {e}",
        "is_error": True,    # 👈 明示的にエラーフラグ
    })

Claude は is_error: True を見て リトライ or 別アプローチ を選びます。

連続エラー時の強制終了

consecutive_errors = 0
for it in range(max_iter):
    if consecutive_errors >= 3:
        # 3 連続エラーなら諦める
        break
    ...
    if any(r["is_error"] for r in tool_results):
        consecutive_errors += 1
    else:
        consecutive_errors = 0

6. 安全なエージェントループ(まとめテンプレ)

def safe_agent_loop(
    question, tools, tool_impls,
    max_iter=10, max_tokens_budget=50000, timeout_sec=60,
):
    state = {"iteration": 0, "total_tokens": 0, "status": "running"}
    t0 = time.time()
    messages = [{"role": "user", "content": question}]

    for it in range(max_iter):
        state["iteration"] = it + 1

        if time.time() - t0 > timeout_sec:
            state["status"] = "timeout"; break
        if state["total_tokens"] > max_tokens_budget:
            state["status"] = "budget"; break

        response = client.messages.create(tools=tools, messages=messages, ...)
        state["total_tokens"] += (
            response.usage.input_tokens + response.usage.output_tokens
        )

        if response.stop_reason == "end_turn":
            state["status"] = "done"; break

        # ツール実行(省略)
    else:
        state["status"] = "max_iter"

    return response, state

まとめ

停止条件は 4 種: end_turn / max_iter / トークン予算 / タイムアウト
- 同一ツール連続呼び出し検知 などのカスタム安全弁
エラー時は is_error: True で Claude に明示。連続エラーで強制終了
状態 (state dict) を毎反復で更新し UI に可視化

この回の限界(次への動機)

安全に止まるエージェントは作れた。次は「RAG × 計算 × 時刻」など 複数種類のツールを連携 させた本格タスクを解かせる番。 👉 次回「マルチステップ実践」で、第 5 章の RAG を組み込んだ完成形エージェントを作ります。

参考文献

Anthropic: Building effective agents
AutoGPT / BabyAGI の反省から生まれた「安全なエージェント」設計ドキュメント多数