エージェントの失敗に対する根本原因分析
モデルエラー、ツールエラー、データエラー、ロジックエラーを体系的に分類します。
「エージェントの失敗に対する根本原因分析」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントの障害分類
エージェントの障害は、次の4種類に分類できます。
- モデルエラー:LLMが誤ったツールを呼び出す、または不正な出力を生成する
- ツールエラー:外部APIが失敗する、または想定外のデータを返す
- データエラー:不正な入力(形式不正、フィールド不足、想定外の型)がある
- ロジックエラー:ステップ自体は正しいが、順序や前提が誤っている
モデルエラー:誤ったツール呼び出し
モデルエラーは、LLMが誤ったツールを選択したり、不正な引数を渡したり、形式不正なJSONを生成したりしたときに発生します。多くの場合、ツールの説明が不明確であることや、プロンプトが曖昧であることが原因です。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def detect_model_errors(response) -> list:
errors = []
message = response.choices[0].message
if message.tool_calls:
for tc in message.tool_calls:
tool_name = tc.function.name
try:
args = json.loads(tc.function.arguments)
except json.JSONDecodeError as e:
errors.append({
'type': 'model_error',
'subtype': 'malformed_tool_args',
'tool': tool_name,
'raw_args': tc.function.arguments,
'parse_error': str(e)
})
continue
# Validate required arguments
expected_tools = {
'search_web': ['query'],
'send_email': ['to', 'subject', 'body'],
'create_task': ['title']
}
required = expected_tools.get(tool_name, [])
missing = [r for r in required if r not in args]
if missing:
errors.append({
'type': 'model_error',
'subtype': 'missing_required_args',
'tool': tool_name,
'missing': missing
})
return errors
print('Model error detection function defined')ツールエラー:APIの障害
ツールエラーは、外部APIがエラー(4xx、5xx)を返したり、タイムアウトしたり、想定外の形式のデータを返したりしたときに発生します。診断できるよう、正確なエラー、ツールの引数、レスポンスを記録します。
import httpx
from dataclasses import dataclass
from typing import Optional
@dataclass
class ToolError:
tool_name: str
error_type: str
status_code: Optional[int]
message: str
args_used: dict
retry_possible: bool
def classify_tool_error(tool_name: str, args: dict, exception: Exception) -> ToolError:
if isinstance(exception, httpx.TimeoutException):
return ToolError(
tool_name=tool_name,
error_type='timeout',
status_code=None,
message=str(exception),
args_used=args,
retry_possible=True # Retries are appropriate for timeouts
)
elif isinstance(exception, httpx.HTTPStatusError):
status = exception.response.status_code
retry = status >= 500 or status == 429 # Server errors and rate limits are retryable
return ToolError(
tool_name=tool_name,
error_type='http_error',
status_code=status,
message=exception.response.text[:200],
args_used=args,
retry_possible=retry
)
else:
return ToolError(
tool_name=tool_name,
error_type='unexpected_error',
status_code=None,
message=str(exception),
args_used=args,
retry_possible=False
)
print('Tool error classification defined')データエラー:入力の検証
データエラーは、エージェントへの入力に問題がある場合に発生します。たとえば、必須フィールドの不足、データ型の誤り、数値が必要な箇所に文字列が渡される場合などです。エージェントのエントリーポイントで入力を検証し、早期に検出します。
from pydantic import BaseModel, validator, ValidationError
from typing import Optional
class EmailAgentInput(BaseModel):
email_id: str
action: str
user_id: int
priority: Optional[str] = 'normal'
@validator('action')
def action_must_be_valid(cls, v):
valid_actions = ['reply', 'forward', 'archive', 'summarize']
if v not in valid_actions:
raise ValueError(f'action must be one of {valid_actions}, got: {v}')
return v
@validator('email_id')
def email_id_not_empty(cls, v):
if not v.strip():
raise ValueError('email_id cannot be empty')
return v
def validate_agent_input(raw_input: dict) -> tuple:
try:
validated = EmailAgentInput(**raw_input)
return validated, None
except ValidationError as e:
return None, [
{'field': err['loc'][0], 'message': err['msg']}
for err in e.errors()
]
# Test with bad input
valid, errors = validate_agent_input({'email_id': '', 'action': 'delete', 'user_id': 'abc'})
if errors:
print('Data errors found:')
for err in errors:
print(f' {err["field"]}: {err["message"]}')ロジックエラー:ステップ順序の誤り
ロジックエラーは、デバッグが最も難しいエラーです。エージェントが正しい引数で正しいツールを呼び出していても、順序を誤っていたり、必要なステップを飛ばしていたり、前のステップの出力について誤った前提を置いていたりすることがあります。
import logging
logger = logging.getLogger('agent.logic')
class AgentStepGuard:
'''
Enforces that steps execute in the required sequence.
'''
def __init__(self):
self.completed_steps = set()
self.STEP_DEPENDENCIES = {
'extract_action_items': ['read_email'],
'create_trello_card': ['extract_action_items'],
'send_slack_notification': ['create_trello_card']
}
def mark_complete(self, step_name: str):
self.completed_steps.add(step_name)
def can_run(self, step_name: str) -> tuple:
required = self.STEP_DEPENDENCIES.get(step_name, [])
missing = [r for r in required if r not in self.completed_steps]
if missing:
return False, f'Logic error: {step_name} requires {missing} to complete first'
return True, None
def assert_can_run(self, step_name: str):
ok, error = self.can_run(step_name)
if not ok:
logger.error(error)
raise RuntimeError(error)
guard = AgentStepGuard()
# Simulate trying to skip a step
try:
guard.assert_can_run('create_trello_card')
except RuntimeError as e:
print('Caught logic error:', e)
# Correct sequence
guard.mark_complete('read_email')
guard.mark_complete('extract_action_items')
guard.assert_can_run('create_trello_card') # Now allowed
print('Step sequence valid')構造化されたエラーログ
事後分析に必要なコンテキストを含めて、すべてのエラーを記録します。エラーの種類、完全なスタックトレース、ステップの入力、障害発生時点で関連するエージェントの状態などを記録してください。
import logging
import traceback
import json
from datetime import datetime
logger = logging.getLogger('agent.errors')
def log_agent_error(error_type: str, step: str, inputs: dict, exception: Exception, agent_state: dict = None):
error_record = {
'timestamp': datetime.utcnow().isoformat(),
'error_type': error_type,
'step': step,
'exception_type': type(exception).__name__,
'exception_message': str(exception),
'traceback': traceback.format_exc(),
'inputs': inputs,
'agent_state': agent_state or {}
}
logger.error(json.dumps(error_record))
return error_record
# Example usage
try:
raise ValueError('Email ID not found in database')
except Exception as e:
record = log_agent_error(
error_type='data_error',
step='read_email',
inputs={'email_id': 'missing-id-123'},
exception=e,
agent_state={'user_id': 42, 'session_id': 'sess-abc'}
)
print('Error logged:', record['error_type'], '-', record['exception_message'])エラー率を監視する
ステップごと、エラーの種類ごとにエラー率を追跡し、システム全体の問題を特定します。モデルエラーが突然増加した場合は、プロンプトの変更によってツール呼び出しが壊れた可能性があります。ツールエラーが増加した場合は、APIの性能低下を示している可能性があります。
from collections import defaultdict
from datetime import datetime
class ErrorTracker:
def __init__(self):
self.errors = defaultdict(list)
def record(self, error_type: str, step: str):
key = f'{error_type}:{step}'
self.errors[key].append(datetime.utcnow())
def get_rates(self, window_minutes: int = 60) -> dict:
from datetime import timedelta
cutoff = datetime.utcnow() - timedelta(minutes=window_minutes)
rates = {}
for key, timestamps in self.errors.items():
recent = [ts for ts in timestamps if ts >= cutoff]
rates[key] = len(recent)
return dict(sorted(rates.items(), key=lambda x: x[1], reverse=True))
def has_spike(self, error_type: str, step: str, threshold: int = 5, window_minutes: int = 10) -> bool:
key = f'{error_type}:{step}'
rates = self.get_rates(window_minutes)
return rates.get(key, 0) >= threshold
tracker = ErrorTracker()
for _ in range(8):
tracker.record('tool_error', 'web_search')
tracker.record('model_error', 'intent_detection')
print('Error rates:', tracker.get_rates())
print('Spike detected:', tracker.has_spike('tool_error', 'web_search'))リプレイでモデルエラーをデバッグする
モデルエラーが発生したら、同じ入力を使ってLLMの呼び出しを正確に再現します。出力を期待される結果と比較します。修正するには、システムプロンプトにより具体的な指示を追加するか、ツールの説明を改善します。
import openai
import json
client = openai.OpenAI(api_key='sk-...')
def save_failed_call(step_name: str, messages: list, tools: list, actual_response: str, expected_tool: str, filepath: str):
record = {
'step': step_name,
'messages': messages,
'tools': tools,
'actual_response': actual_response,
'expected_tool': expected_tool
}
with open(filepath, 'w') as f:
json.dump(record, f, indent=2)
print(f'Failed call saved to: {filepath}')
def replay_failed_call(filepath: str, improved_system_prompt: str = None) -> str:
with open(filepath) as f:
record = json.load(f)
messages = record['messages']
if improved_system_prompt:
# Replace system prompt
messages = [m if m['role'] != 'system' else {'role': 'system', 'content': improved_system_prompt}
for m in messages]
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
tools=record['tools']
)
return response.choices[0].message
print('Replay debugging functions defined')根本原因分析チェックリスト
エージェントが失敗したら、次のチェックリストを順番に確認します。
- 入力は有効でしたか?(データエラー)
- 外部APIがエラーを返していませんか?(ツールエラー)
- LLMは正しいツールを呼び出しましたか?引数は正しかったですか?(モデルエラー)
- ステップは正しい順序で実行されましたか?(ロジックエラー)
- プロンプトには十分なコンテキストがありましたか?(モデルエラー - コンテキスト)
def diagnose_failure(error_log: dict) -> dict:
diagnosis = {
'error_type': error_log.get('error_type'),
'root_cause': None,
'immediate_fix': None,
'long_term_fix': None
}
if error_log.get('error_type') == 'data_error':
diagnosis['root_cause'] = 'Invalid or missing input data'
diagnosis['immediate_fix'] = 'Return clear error to caller with field-level validation feedback'
diagnosis['long_term_fix'] = 'Add Pydantic validation at agent entry point'
elif error_log.get('error_type') == 'tool_error':
status = error_log.get('status_code')
if status == 429:
diagnosis['root_cause'] = 'Rate limit hit'
diagnosis['immediate_fix'] = 'Retry with exponential backoff'
diagnosis['long_term_fix'] = 'Add rate limiter to tool calls'
elif status and status >= 500:
diagnosis['root_cause'] = 'Upstream service degradation'
diagnosis['immediate_fix'] = 'Retry up to 3 times, then graceful degradation'
diagnosis['long_term_fix'] = 'Add circuit breaker pattern'
elif error_log.get('error_type') == 'model_error':
diagnosis['root_cause'] = 'LLM tool selection failure'
diagnosis['immediate_fix'] = 'Add explicit tool selection validation'
diagnosis['long_term_fix'] = 'Improve tool descriptions; add few-shot examples'
return diagnosis
result = diagnose_failure({'error_type': 'tool_error', 'status_code': 429})
print('Diagnosis:', result)重大な障害に対してアラートを発生させる
すべてのエラーに即時対応が必要とは限りません。エラーを重大度で分類し、それに応じてアラートを振り分けます。重要度の低い処理フローで発生した、すぐに対応する必要のないデータエラーはログに記録し、主要なフローの障害には即時アラートを発生させます。
ERROR_SEVERITY = {
'data_error': 'low', # Bad input: log and return error to caller
'model_error': 'medium', # LLM misbehavior: investigate, may need prompt fix
'tool_error': 'medium', # API failure: may self-recover with retry
'logic_error': 'high' # Sequencing bug: needs code fix immediately
}
def route_alert(error_type: str, step: str, message: str, is_core_flow: bool = False):
severity = ERROR_SEVERITY.get(error_type, 'medium')
if is_core_flow:
severity = 'high'
if severity == 'high':
print(f'[PAGERDUTY] CRITICAL: {error_type} in {step}: {message}')
# Call PagerDuty API here
elif severity == 'medium':
print(f'[SLACK] WARNING: {error_type} in {step}: {message}')
# Call Slack API here
else:
print(f'[LOG] INFO: {error_type} in {step}: {message}')
route_alert('logic_error', 'create_trello_card', 'Prerequisites not met', is_core_flow=True)
route_alert('data_error', 'parse_input', 'Missing optional field', is_core_flow=False)ポストモーテムのテンプレート
重大なエージェント障害が発生したら、ポストモーテムを作成します。優れたポストモーテムには、タイムライン、根本原因、影響、うまくいったこと、問題だったこと、再発防止のためのアクションアイテムが含まれます。
def generate_post_mortem(failure_data: dict) -> str:
return f'''
## Post-Mortem: {failure_data.get("title", "Agent Failure")}
**Date**: {failure_data.get("date")}
**Duration**: {failure_data.get("duration_minutes")} minutes
**Impact**: {failure_data.get("impact")}
### Timeline
{chr(10).join(failure_data.get("timeline", []))}
### Root Cause
{failure_data.get("root_cause")}
### Contributing Factors
{chr(10).join(failure_data.get("contributing_factors", []))}
### Action Items
{chr(10).join([f"- [ ] {item}" for item in failure_data.get("action_items", [])])}
'''.strip()
post_mortem_data = {
'title': 'Email Pipeline Failure - Wrong Trello List',
'date': '2025-01-15',
'duration_minutes': 45,
'impact': '200 emails processed but Trello cards created in wrong list',
'timeline': ['09:00 - Pipeline started', '09:15 - First error logged', '09:45 - Fixed and redeployed'],
'root_cause': 'Logic error: TRELLO_LIST_ID env var defaulted to staging value in production',
'contributing_factors': ['- Missing config validation at startup', '- No alert on wrong list ID'],
'action_items': ['Add config validation on startup', 'Alert if list_id changes unexpectedly']
}
print(generate_post_mortem(post_mortem_data))理解度チェック:根本原因分析
エージェント障害の根本原因分析についての理解度を確認します。
デバッグのまとめ
エージェント障害の体系的な根本原因分析では、4カテゴリの分類法を使用します。モデルエラー(LLMの意思決定の問題)、ツールエラー(APIの障害)、データエラー(不正な入力)、ロジックエラー(処理順序のバグ)です。これに、構造化ログ、エラー率の監視、モデルエラーに対するリプレイデバッグ、重大な障害に対するポストモーテムを組み合わせます。
よくある質問
「エージェントの失敗に対する根本原因分析」レッスンは無料ですか?
はい。「エージェントの失敗に対する根本原因分析」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「エージェントの失敗に対する根本原因分析」で何を学びますか?
モデルエラー、ツールエラー、データエラー、ロジックエラーを体系的に分類します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか?
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「エージェントの失敗に対する根本原因分析」レッスンにはどのくらい時間がかかりますか?
ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか?
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- LangSmith と Langfuse によるトレース分析
- ステップごとのトークン数とコストのプロファイリング
- 遅くてコストの高いステップを特定する
- エージェントの失敗に対する根本原因分析