LLMエージェントのタイムアウト対策：層ごとの設計と回復メカニズム
LLMエージェントを本番環境に投入して数週間も経つと、必ずといっていいほど「タイムアウト地獄」に遭遇します。単発のLLM呼び出しなら数秒で終わるはずが、ツールを5回・10回と連鎖させていくうちに数分が経過し、HTTPタイムアウトで途中終了、あるいは断片的な処理結果だけが残る——そんな状況は珍しくありません。この記事では、エージェント型LLMのタイムアウト問題を「層ごとに分けて考える」アプローチで整理し、明日から使える実装パターンをまとめます。
エージェント実行がタイムアウトしやすい理由
通常のAPIリクエストと違い、LLMエージェントの実行時間は本質的に「予測不可能」です。主な理由は2つあります。
ツール呼び出しの累積遅延 は最もわかりやすいものです。データベース検索・外部API呼び出し・ファイルI/Oなど、各ツールが1〜3秒かかるとしても、10ステップのエージェントループでは最悪30秒を超えます。さらに、ツールの結果をLLMが解釈して「次のツールを呼ぶ」という判断自体にもLLM推論のレイテンシが加算されます。
LLM応答待機の不確定性 も見逃せません。大規模モデルはトークン数やプロバイダー側の負荷によって応答時間が大きくブレます。ストリーミングを使っていない場合、最初のトークンが返ってくるまでの待機時間（TTFT）が数秒に及ぶことも珍しくなく、これが複数回重なると全体の実行時間は読めなくなります。
結果として、「普段は20秒で終わるタスクが、たまに2分かかる」という不安定な挙動が生じます。この問題に立ち向かうには、タイムアウトを「1つの大きな制限」として扱うのではなく、実行の粒度ごとに設計し直す必要があります。
設定すべきタイムアウトの3層構造
タイムアウトは少なくとも3つの層に分けて設定するのが基本です。
┌─────────────────────────────────────────┐
│ グローバルタイムアウト (例: 120秒) │
│ ┌───────────────────────────────────┐ │
│ │ ステップタイムアウト (例: 30秒/ステップ) │ │
│ │ ┌─────────────────────────────┐ │ │
│ │ │ ツール呼び出しタイムアウト │ │ │
│ │ │ (例: 10秒/ツール) │ │ │
│ │ └─────────────────────────────┘ │ │
│ └───────────────────────────────────┘ │
└─────────────────────────────────────────┘グローバルタイムアウト は、エージェントループ全体に対して設定します。ユーザーへのSLAやインフラコストの観点から決めるもので、超えた場合に即座にエラーを返すのではなく、「ここまで完了した内容」を返す設計が重要です。
ステップタイムアウト は、LLMが1回の推論にかける時間の上限です。モデルの応答が遅延した場合に、無限に待ち続けるのを防ぎます。
ツール呼び出しタイムアウト は、個々のツール（API、DB、ファイル操作など）ごとに設定します。外部APIなら10秒、内部DBなら3秒、といった形でリソースの性質に合わせて調整します。
PythonでLangChainやカスタムエージェントループを組んでいる場合、非同期処理とasyncio.wait_forを組み合わせるのが実践的です。
import asyncio
from typing import Any
async def run_tool_with_timeout(
tool_fn, args: dict, timeout: float = 10.0
) -> dict:
try:
result = await asyncio.wait_for(
tool_fn(**args),
timeout=timeout
)
return {"status": "success", "result": result}
except asyncio.TimeoutError:
return {
"status": "timeout",
"result": None,
"message": f"Tool timed out after {timeout}s"
}
except Exception as e:
return {"status": "error", "result": None, "message": str(e)}
async def agent_loop(task: str, global_timeout: float = 120.0):
start_time = asyncio.get_event_loop().time()
async def _inner():
steps_completed = []
# ... エージェントループの本体 ...
return steps_completed
try:
return await asyncio.wait_for(_inner(), timeout=global_timeout)
except asyncio.TimeoutError:
elapsed = asyncio.get_event_loop().time() - start_time
# 途中まで完了した結果を返す
return {"status": "partial", "elapsed": elapsed}各層でタイムアウトをキャッチし、「失敗」ではなく「部分的な状態」として扱える設計にしておくことが、エラーハンドリングの起点になります。
部分的な実行失敗からの回復：チェックポイントと再開メカニズム
タイムアウトが起きた後に何もできないのは最悪のパターンです。長時間タスクに対しては チェックポイント保存 を組み込んでおくことで、途中再開が可能になります。
チェックポイントに保存すべき最低限の情報は以下の通りです。
from dataclasses import dataclass, asdict
from typing import List, Optional
import json, time
@dataclass
class AgentCheckpoint:
task_id: str
original_task: str
completed_steps: List[dict] # 完了済みステップの結果
pending_steps: List[str] # 未完了ステップの計画
context_summary: str # これまでの文脈サマリー
created_at: float = 0.0
def save(self, store): # store は Redis や DB など
self.created_at = time.time()
store.set(f"checkpoint:{self.task_id}", json.dumps(asdict(self)))
@classmethod
def load(cls, task_id: str, store) -> Optional["AgentCheckpoint"]:
data = store.get(f"checkpoint:{task_id}")
return cls(**json.loads(data)) if data else None「毎ステップ後に保存」するのが最も安全ですが、書き込みコストが気になる場合は「重要度の高いステップ後のみ保存」という判断をLLMに委ねる方法もあります。
再開時には、completed_stepsの結果とcontext_summaryをプロンプトに含めることで、エージェントが「どこまで終わっているか」を把握した上で処理を続けられます。完全な会話履歴を渡すとコンテキスト長を圧迫するため、サマリー形式にしておくのが実用的です。
プロンプト設計にタイムアウト前提の意思決定を組み込む
エラーハンドリングの設計はコードだけでなく、 プロンプト設計 にも反映させる必要があります。タイムアウトを意識したシステムプロンプトには、以下の要素を含めると効果的です。
## 実行ガイドライン
- 各ツール呼び出しの前に、「このステップが失敗した場合の代替手段」を考慮してください
- 結果が取得できなかった場合は "TOOL_UNAVAILABLE" として処理を継続し、
最終的な回答に不確実性を明記してください
- ツール応答が遅い、ステップ数が多いなどタイムアウトが近い兆候がある場合は、
現時点での最善の部分的回答を提示してください
- 「完璧な回答を待つ」より「80%の精度で早く返す」を優先するシナリオでは、
そのトレードオフを明示した上で回答してくださいこのようなガイドラインを組み込むことで、ツールが"status": "timeout"を返したとき、エージェントが「エラーで止まる」のではなく「不確実情報として処理を続ける」という動作を引き出せます。
また、ステップ数に上限を設けて early stopping条件 をプロンプトに明記するのも有効です。「10ステップ以内に結論を出してください」という制約は、エージェントが無駄な探索を続けることを防ぎ、結果的にタイムアウトの発生頻度を下げます。
本番運用での監視とログ：タイムアウトパターンの可視化
タイムアウト問題を「起きた時に対処する」から「予防的に改善する」に移行するには、適切な計測が必要です。少なくとも以下の情報はログに残しておくことをお勧めします。
import logging, time
from typing import Optional
def log_agent_step(
task_id: str,
step_index: int,
tool_name: str,
duration_ms: float,
status: str,
token_count: Optional[int] = None
):
logging.info({
"event": "agent_step",
"task_id": task_id,
"step_index": step_index,
"tool_name": tool_name,
"duration_ms": duration_ms,
"status": status, # success / timeout / error
"token_count": token_count,
"timestamp": time.time()
})このログを集計すると、「どのツールでタイムアウトが多発しているか」「何ステップ目で全体のタイムアウトが起きやすいか」というパターンが見えてきます。特定のツールが頻繁にタイムアウトしているなら、そのツール単体の閾値を見直すか、非同期での並列実行に切り替えるかを検討します。
監視の観点では、以下の3つのメトリクスをダッシュボードに出しておくと問題発見が早くなります。
- P95実行時間 ：タイムアウト閾値の調整根拠になる
- タイムアウト率（全体・ツール別） ：どこがボトルネックかを特定する
- チェックポイントからの再開率 ：部分失敗がどれだけ回復できているかを示す
タイムアウト処理はどうしても「後回し」にされがちな非機能要件ですが、エージェントを本番環境で安定稼働させるには、最初の設計段階から組み込む姿勢が結果的にコストを下げます。層ごとの設計・チェックポイント・プロンプト設計・監視、この4つを揃えることで、予測不可能な実行時間という根本的な問題とうまく付き合えるようになります。
