ツールエラーの検出と復旧

ツールは失敗します。対策を用意しましょう

常に返し、決して raise しない

def safe_dispatch(tool_call):
    try:
        args = json.loads(tool_call.function.arguments)
        return TOOLS[tool_call.function.name](**args)
    except json.JSONDecodeError:
        return {'error': 'Arguments are not valid JSON.'}
    except KeyError:
        return {'error': f'Unknown tool: {tool_call.function.name}'}
    except Exception as e:
        return {'error': f'{type(e).__name__}: {e}'}

構造化されたエラー形式

error = {'ok': False, 'error_type': 'TimeoutError', 'error_message': 'Tavily timed out after 10s', 'retryable': True}
print(error)

リトライ可能なエラーと恒久的なエラーを区別する

if isinstance(e, requests.Timeout):
    return {'ok': False, 'retryable': True, 'error': str(e)}
if isinstance(e, ValueError):
    return {'ok': False, 'retryable': False, 'error': str(e)}

一時的なエラーを自動的にリトライする

from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

@retry(
    wait=wait_exponential(multiplier=1, max=10),
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type((requests.Timeout, requests.ConnectionError))
)
def web_search(query):
    return requests.get('https://api.tavily.com/search', ...).json()

引数の検証

from pydantic import BaseModel, ValidationError

class SearchArgs(BaseModel):
    query: str
    k: int = 5

try:
    args = SearchArgs.model_validate_json(tool_call.function.arguments)
except ValidationError as e:
    return {'error': f'Bad arguments: {e}'}

モデルにエラーを示す

messages.append({
    'role': 'tool',
    'tool_call_id': tc.id,
    'content': json.dumps({'error': 'Argument k must be an integer'})
})
# Next model call: 'Sorry, let me retry with k=5...'

無限エラーループを避ける

recent_calls = []
for tc in msg.tool_calls:
    key = (tc.function.name, tc.function.arguments)
    if recent_calls.count(key) >= 3:
        return 'Agent stuck in retry loop, aborting.'
    recent_calls.append(key)

ツールごとの復旧

def search_with_fallback(query):
    try:
        return tavily_search(query)
    except Exception:
        return bing_search(query)   # secondary provider

すべての呼び出しにタイムアウトを設定

import requests
response = requests.get(url, timeout=10)  # 10s

# For LLM calls:
from openai import OpenAI
client = OpenAI(timeout=30.0)

サーキットブレーカー

import time

class CircuitOpen(Exception):
    pass

def circuit(failure_threshold=5, recovery_timeout=60):
    def decorator(func):
        state = {'failures': 0, 'open_until': 0}
        def wrapper(*args, **kwargs):
            if time.time() < state['open_until']:
                raise CircuitOpen('circuit is open')
            try:
                result = func(*args, **kwargs)
                state['failures'] = 0
                return result
            except Exception:
                state['failures'] += 1
                if state['failures'] >= failure_threshold:
                    state['open_until'] = time.time() + recovery_timeout
                raise
        return wrapper
    return decorator

@circuit(failure_threshold=3, recovery_timeout=1)
def fragile_tool(x):
    if x < 0:
        raise ValueError('bad input')
    return x * 2

for x in [1, -1, -1, -1, -1]:
    try:
        print('ok', fragile_tool(x))
    except CircuitOpen as e:
        print('blocked:', e)
    except ValueError as e:
        print('failed:', e)

コンテキスト付きでエラーをログに記録

グレースフルデグラデーション

if all_search_tools_failed:
    return 'I was unable to search the web right now. Please try again in a few minutes.'

ツールエラーのパターン

まとめ