Analisis Akar Masalah Kegagalan Agen

Taksonomi Kegagalan Agen

Kesalahan Model: Pemanggilan Alat yang Salah

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

Kesalahan Alat: Kegagalan API

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

Kesalahan Data: Validasi Masukan

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"]}')

Kesalahan Logika: Urutan Langkah yang Salah

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

Pencatatan Kesalahan Terstruktur

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'])

Pemantauan Tingkat Kesalahan

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

Men-debug Kesalahan Model dengan Pemutaran Ulang

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

Daftar Periksa Analisis Akar Masalah

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)

Memberi Peringatan atas Kegagalan Kritis

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)

Templat Analisis Pascakejadian

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))

Uji Pemahaman: Analisis Akar Masalah

Ringkasan Penelusuran Kesalahan