도구 오류 감지 및 복구

도구는 실패합니다. 이에 대비하십시오.

항상 반환하고, 예외를 발생시키지 마십시오

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.'

도구 오류 처리 패턴

요약