エージェントの失敗モードの分類

エージェントが独特な形で失敗する理由

失敗モード1：ツールエラー

class ToolError(Exception):
    def __init__(self, tool_name: str, args: dict, error: Exception):
        self.tool_name = tool_name
        self.args = args
        self.original_error = error
        super().__init__(f'Tool {tool_name} failed: {error}')

def safe_tool_call(tool_func, args: dict) -> str:
    try:
        result = tool_func(**args)
        if not result:
            return 'Tool returned empty result. Try a different approach.'
        return str(result)
    except Exception as e:
        raise ToolError(tool_func.__name__, args, e)

失敗モード2：不正な形式の出力

import json

def validate_agent_output(raw_output: str, expected_format: str) -> dict:
    if expected_format == 'json':
        try:
            return json.loads(raw_output)
        except json.JSONDecodeError as e:
            return {
                'valid': False,
                'error': f'Expected JSON but got invalid JSON: {e}',
                'raw': raw_output[:200]
            }
    return {'valid': True, 'data': raw_output}

失敗モード3：推論ループ

from collections import Counter

class LoopDetector:
    def __init__(self, window: int = 5, threshold: int = 3):
        self.recent_actions = []
        self.window = window
        self.threshold = threshold

    def record(self, action: str) -> bool:
        self.recent_actions.append(action)
        if len(self.recent_actions) > self.window:
            self.recent_actions.pop(0)
        counts = Counter(self.recent_actions)
        most_common_count = counts.most_common(1)[0][1] if counts else 0
        return most_common_count >= self.threshold  # True = loop detected

失敗モード4：コンテキストの枯渇

import tiktoken

CONTEXT_LIMIT = 100_000  # tokens
COMPRESS_AT = 80_000     # trigger compression with headroom

enc = tiktoken.encoding_for_model('gpt-4o')

def total_tokens(messages: list) -> int:
    return sum(len(enc.encode(str(m))) for m in messages)

def check_context(messages: list) -> str:
    tokens = total_tokens(messages)
    if tokens > COMPRESS_AT:
        return 'compress'
    if tokens > CONTEXT_LIMIT:
        return 'critical'
    return 'ok'

失敗モード5：外部サービスの利用不能

import asyncio

async def resilient_tool_call(tool_func, args: dict, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            return await tool_func(**args)
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries - 1:
                return f'Service unavailable after {max_retries} attempts. Error: {e}'
            wait = 2 ** attempt  # 1s, 2s, 4s
            await asyncio.sleep(wait)
    return 'Unexpected error in resilient_tool_call'

失敗モード6：目的の誤認

async def confirm_goal_understanding(original_task: str) -> str:
    resp = await client.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role': 'system', 'content': 'Restate the task in your own words. Be specific about what the final deliverable should be.'},
            {'role': 'user', 'content': f'Task: {original_task}'}
        ]
    )
    return resp.choices[0].message.content

# Use the restatement as the first step of the agent
# to catch misunderstandings before any tools are called

失敗分類システムの構築

from enum import Enum
from dataclasses import dataclass

class FailureType(Enum):
    TOOL_ERROR = 'tool_error'
    MALFORMED_OUTPUT = 'malformed_output'
    REASONING_LOOP = 'reasoning_loop'
    CONTEXT_EXHAUSTION = 'context_exhaustion'
    EXTERNAL_SERVICE = 'external_service'
    GOAL_MISUNDERSTANDING = 'goal_misunderstanding'
    MAX_ITERATIONS = 'max_iterations'
    UNKNOWN = 'unknown'

@dataclass
class AgentFailure:
    failure_type: FailureType
    step: int
    tool_name: str | None
    error_message: str
    recoverable: bool

失敗と復旧アクションの対応付け

RECOVERY_ACTIONS = {
    FailureType.TOOL_ERROR:           'retry_with_corrected_args',
    FailureType.MALFORMED_OUTPUT:     'reformat_output',
    FailureType.REASONING_LOOP:       'inject_diversity_prompt',
    FailureType.CONTEXT_EXHAUSTION:   'compress_history',
    FailureType.EXTERNAL_SERVICE:     'use_fallback_tool',
    FailureType.GOAL_MISUNDERSTANDING:'request_clarification',
    FailureType.MAX_ITERATIONS:       'escalate_to_human',
    FailureType.UNKNOWN:              'escalate_to_human'
}

最大反復回数の設定

MAX_ITERATIONS = 15

async def run_agent(task: str) -> str:
    messages = [{'role': 'user', 'content': task}]
    loop_detector = LoopDetector()
    for iteration in range(MAX_ITERATIONS):
        response = await get_agent_action(messages)
        if response.is_final:
            return response.answer
        action_key = f'{response.tool}:{response.args}'
        if loop_detector.record(action_key):
            messages.append({'role': 'system', 'content': 'You are repeating yourself. Try a completely different approach.'})
            continue
        result = await execute_tool(response.tool, response.args)
        messages.append({'role': 'tool', 'content': result})
    return 'Task exceeded maximum iterations. Partial results: ' + get_partial_result(messages)

事後分析のための失敗ログ記録

import json
from dataclasses import asdict

async def log_agent_failure(task_id: str, failure: AgentFailure, history: list, pool):
    async with pool.acquire() as conn:
        await conn.execute('''
            INSERT INTO agent_failures
            (task_id, failure_type, step, tool_name, error_message,
             recoverable, action_history, failed_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
        ''',
            task_id,
            failure.failure_type.value,
            failure.step,
            failure.tool_name,
            failure.error_message,
            failure.recoverable,
            json.dumps(history)
        )

クイックチェック

レッスンのまとめ