自律エージェントを「動かす」ことと「本番品質で動かし続ける」ことの間には、大きな溝があります。
Claude Code SDK を使って最初のエージェントを作るのは意外と簡単です。でも、それをスケジュール実行で毎日動かし続けると、必ず想定外の壊れ方をします。私は複数のコンテンツ自動化システムを本番で運用する中で、その壊れ方のパターンをほぼ出尽くしたと思っています。
ここではそこから蒸留した 7 つの設計原則と、その実装コードを公開します。「動くエージェント」から「壊れにくいエージェント」への距離を縮める手がかりになれば幸いです。
原則 1：ループ深度の上限を必ず設ける
Claude Code SDK の query() は内部でツール使用ループを自動で処理します。これは便利な反面、ループが終わらないケースを生みやすいです。
最も典型的な無限ループのパターンは「エージェントが同じツールを繰り返し呼び出す」です。エラーが出るたびに同じ修正を試み続け、いつまでも抜け出せない状況です。
import anthropic
def run_agent_with_depth_limit(
prompt: str,
max_turns: int = 20,
model: str = "claude-sonnet-4-6"
) -> str:
"""ループ深度を制限したエージェント実行"""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": prompt}]
turn_count = 0
while turn_count < max_turns:
response = client.messages.create(
model=model,
max_tokens=4096,
tools=TOOLS,
messages=messages
)
turn_count += 1
# ツール使用がなければ完了
if response.stop_reason == "end_turn":
return extract_text(response)
# ツール呼び出しを実行してループ継続
tool_results = execute_tools(response.content)
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
# 上限到達：最後の応答を返す
raise AgentLoopLimitError(f"Agent exceeded {max_turns} turns")max_turns の適切な値はタスクの性質によりますが、私の実装では記事生成が 10〜15、コードレビューが 20〜30 を設定しています。上限に到達したら例外を投げてログに記録し、そのタスクをスキップして次に進む設計にしています。
原則 2：タイムアウトは 2 層で設ける
タイムアウトには「1 回の API 呼び出しのタイムアウト」と「エージェントループ全体のタイムアウト」の 2 層が必要です。API 単体には timeout パラメータで制御できますが、ループ全体は別途実装が必要です。
import asyncio
import signal
from contextlib import contextmanager
@contextmanager
def timeout_context(seconds: int):
"""シグナルベースのタイムアウト（Unix系OS向け）"""
def handler(signum, frame):
raise TimeoutError(f"Agent timed out after {seconds} seconds")
old_handler = signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
async def run_agent_with_timeout(
prompt: str,
timeout_seconds: int = 300 # 5分
) -> str:
"""全体タイムアウト付きエージェント実行"""
try:
with timeout_context(timeout_seconds):
return await run_agent_core(prompt)
except TimeoutError as e:
log_timeout(prompt, timeout_seconds)
raise AgentTimeoutError(str(e))スケジュールタスクでは、1 タスクあたり 5〜10 分のタイムアウトを設定しています。これを超えるタスクは何らかの問題がある（ループ脱出できていない、API が応答しないなど）と判断して強制終了します。
原則 3：ツール呼び出しの重複を検知する
無限ループの 90% は「同じツールを同じ引数で繰り返し呼び出す」パターンです。これを検知するシンプルなガードが効果的です。
from collections import deque
from hashlib import md5
import json
class ToolCallDeduplicator:
def __init__(self, window_size: int = 5):
self.recent_calls = deque(maxlen=window_size)
def is_duplicate(self, tool_name: str, tool_input: dict) -> bool:
"""直近 window_size 件と同じ呼び出しかチェック"""
call_hash = md5(
json.dumps({"name": tool_name, "input": tool_input}, sort_keys=True).encode()
).hexdigest()
if call_hash in self.recent_calls:
return True
self.recent_calls.append(call_hash)
return False
def execute_tools_with_dedup(tool_calls: list, dedup: ToolCallDeduplicator) -> list:
results = []
for call in tool_calls:
if dedup.is_duplicate(call.name, call.input):
# 重複検知：エラーメッセージをフィードバックして方向転換を促す
results.append({
"type": "tool_result",
"tool_use_id": call.id,
"content": "Error: This exact tool call was already attempted. Please try a different approach.",
"is_error": True
})
else:
results.append(execute_single_tool(call))
return resultsこのガードを導入してから、無限ループによるタスク失敗が約 85% 減りました。単純ですが効果は大きいです。
原則 4：コスト予算をエージェントレベルで管理する
API 全体のコスト管理はできていても、「このエージェント実行で使っていいコスト上限」をエージェントレベルで管理しているケースは少ないです。しかし、本番運用では必要です。
from dataclasses import dataclass, field
@dataclass
class CostBudget:
max_dollars: float
spent_dollars: float = 0.0
# Sonnet 4.6 料金（MTok単位）
INPUT_RATE = 3.0 / 1_000_000
OUTPUT_RATE = 15.0 / 1_000_000
THINKING_RATE = 15.0 / 1_000_000
def record_usage(self, usage: anthropic.types.Usage) -> None:
self.spent_dollars += (
usage.input_tokens * self.INPUT_RATE +
usage.output_tokens * self.OUTPUT_RATE
)
@property
def is_exceeded(self) -> bool:
return self.spent_dollars >= self.max_dollars
@property
def remaining(self) -> float:
return max(0, self.max_dollars - self.spent_dollars)
def run_agent_with_budget(prompt: str, budget: CostBudget) -> str:
client = anthropic.Anthropic()
messages = [{"role": "user", "content": prompt}]
while True:
if budget.is_exceeded:
raise BudgetExceededError(
f"Budget exceeded: ${budget.spent_dollars:.4f} / ${budget.max_dollars}"
)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
messages=messages
)
budget.record_usage(response.usage)
if response.stop_reason == "end_turn":
return extract_text(response)
# ループ継続処理...タスクごとの予算は $0.05〜$0.20 に設定しています。これを超えるタスクはスキップしてログに記録します。月末に予算超過ログをレビューして、コストがかかりすぎているタスクを特定するのに役立てています。
原則 5：エラーを「回復可能」と「回復不可能」に分類する
エラーハンドリングで犯しがちな失敗は「すべてのエラーを同じように扱う」ことです。API の一時的な過負荷（529）とプロンプトの問題（400）では対処法が全く違います。
import anthropic
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
class RecoverableError(Exception):
"""リトライで回復できる可能性があるエラー"""
pass
class FatalError(Exception):
"""リトライしても意味がないエラー"""
pass
def classify_api_error(e: anthropic.APIError) -> None:
"""APIエラーを分類して適切な例外を投げる"""
if isinstance(e, anthropic.RateLimitError):
raise RecoverableError(f"Rate limit: {e}")
elif isinstance(e, anthropic.InternalServerError):
raise RecoverableError(f"Server error (529): {e}")
elif isinstance(e, anthropic.BadRequestError):
raise FatalError(f"Bad request (400) - prompt issue: {e}")
elif isinstance(e, anthropic.AuthenticationError):
raise FatalError(f"Auth failed (401) - check API key: {e}")
else:
raise RecoverableError(f"Unknown error: {e}")
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=60),
retry=retry_if_exception_type(RecoverableError)
)
def call_api_with_retry(params: dict) -> anthropic.Message:
client = anthropic.Anthropic()
try:
return client.messages.create(**params)
except anthropic.APIError as e:
classify_api_error(e)FatalError は即座に上位に伝播させてタスクをスキップ、RecoverableError は指数バックオフでリトライします。この分類だけで、不要なリトライによる無駄なコストが大幅に減りました。
原則 6：実行ログを構造化してデバッグ可能にする
スケジュール実行のエージェントは「いつ・何が・なぜ失敗したか」が分かりにくくなります。ログを構造化してフィルタ可能にしておくことが後のデバッグを大幅に楽にします。
import json
import logging
from datetime import datetime, timezone
from typing import Any
def create_structured_logger(task_id: str) -> logging.Logger:
logger = logging.getLogger(task_id)
class StructuredFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_data = {
"ts": datetime.now(timezone.utc).isoformat(),
"task_id": task_id,
"level": record.levelname,
"msg": record.getMessage(),
}
if hasattr(record, "extra"):
log_data.update(record.extra)
return json.dumps(log_data, ensure_ascii=False)
handler = logging.StreamHandler()
handler.setFormatter(StructuredFormatter())
logger.addHandler(handler)
return logger
# 使用例
logger = create_structured_logger("article-generation-2026-05-05")
logger.info("Agent started", extra={
"model": "claude-sonnet-4-6",
"task_type": "article_generation",
"budget_usd": 0.10
})
logger.error("Tool call failed", extra={
"tool": "web_search",
"error": "timeout",
"turn": 7
})JSON 構造化ログにすることで、jq や CloudWatch Logs Insights でのフィルタリングが簡単になります。「先週のコスト超過タスクを一覧する」「特定ツールのエラー率を計算する」といった分析が即座にできます。
原則 7：グレースフルデグレードを設計する
最も重要かもしれない原則です。エージェントが失敗したとき、何もしないのではなく「品質は下がっても何かを返す」設計にします。
コンテンツ生成パイプラインでの実装例を示します。
from enum import Enum
class DegradationLevel(Enum):
FULL = "full" # フル機能（拡張思考 + 高品質プロンプト）
REDUCED = "reduced" # 機能縮小（拡張思考なし + 簡略プロンプト）
MINIMAL = "minimal" # 最小（Haiku + 最短プロンプト）
SKIP = "skip" # スキップ（次回に持ち越し）
def run_with_graceful_degradation(
task: dict,
max_retries: int = 3
) -> dict:
level = DegradationLevel.FULL
for attempt in range(max_retries):
try:
if level == DegradationLevel.FULL:
return run_full_agent(task)
elif level == DegradationLevel.REDUCED:
return run_reduced_agent(task)
elif level == DegradationLevel.MINIMAL:
return run_minimal_agent(task)
else:
return {"status": "skipped", "task_id": task["id"]}
except (AgentLoopLimitError, BudgetExceededError):
# 上位品質モードが失敗したら次のレベルへ降格
if level == DegradationLevel.FULL:
level = DegradationLevel.REDUCED
elif level == DegradationLevel.REDUCED:
level = DegradationLevel.MINIMAL
else:
level = DegradationLevel.SKIP
except FatalError:
# 致命的エラーはスキップ
return {"status": "fatal_error", "task_id": task["id"]}
return {"status": "skipped", "task_id": task["id"]}この設計により、スケジュール実行の成功率が「0か100か」から「品質のグラデーション」に変わりました。フル機能が失敗しても最低限の出力は得られます。
実装チェックリスト
本番デプロイ前に確認すべき項目をまとめます。
ループ深度の上限は設定されているかどうか、全体タイムアウトは設定されているかどうか、ツール呼び出し重複検知は実装されているかどうか、コスト予算はエージェントレベルで管理されているかどうか、エラーを回復可能・回復不可で分類しているかどうか、ログは構造化 JSON で出力されているかどうか、グレースフルデグレードは設計されているかどうかという 7 点です。
全体を振り返って：本番品質は「壊れ方の設計」から
エージェントの本番品質は、正常系の完成度だけでは決まりません。「いつ、どのように壊れるか」を設計することで、壊れても影響を最小化できます。
7 つの原則はどれも個別に実装できます。まずは最も問題になりやすい「ループ深度の上限」と「タイムアウト」の 2 つから始めるのがお勧めです。その 2 つだけで、スケジュール実行の信頼性は大幅に上がります。
自律エージェントを本番で動かし続けることは、一度作って終わりではありません。壊れ方を知り、設計に組み込み、観察し続けることが必要です。それが「本番品質」への道だと感じています。