AI Agent Error Handling Patterns: Common Errors and Fixes

AI Agent Error Handling Patterns: Common Errors and Fixes

💡 What You Will Learn

AI Agent Error Handling Patterns: Common Errors and Fixes

REFUSAL_PATTERNS = [
    "Sorry, I cannot", "Sorry, I cannot do that", "I cannot", "I'm sorry, but",
    "AI", "", "", ""
]

def handle_refusal(response: str, original_prompt: str, max_retries=2) -> str:
    """Retry"""
    if not any(p in response for p in REFUSAL_PATTERNS):
        return response

    print("[] ")

    prompts = [
        f"AI{original_prompt}",
        f"{original_prompt}",
        f"{original_prompt}"
    ]

    for i, alt_prompt in enumerate(prompts[:max_retries]):
        print(f"[] {i+1}prompt")
        new_response = llm.invoke(alt_prompt)
        if not any(p in new_response for p in REFUSAL_PATTERNS):
            return new_response

    return ""

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")
MAX_TOKENS = 32000
RESERVE_TOKENS = 4000

def truncate_history(messages: list) -> list:
    """ConversationN"""
    total_tokens = 0
    truncated = []

    system_msg = [m for m in messages if m['role'] == 'system']
    non_system = [m for m in messages if m['role'] != 'system']

    for msg in reversed(non_system):
        msg_tokens = len(tokenizer.encode(msg['content']))
        if total_tokens + msg_tokens > MAX_TOKENS - RESERVE_TOKENS:
            break
        truncated.insert(0, msg)
        total_tokens += msg_tokens

    return system_msg + truncated

class LoopDetector:
    def __init__(self, max_steps=10, similarity_threshold=0.85):
        self.max_steps = max_steps
        self.threshold = similarity_threshold
        self.history = []

    def check(self, current_action: str) -> tuple:
        """ (, )"""
        self.history.append(current_action)

        if len(self.history) > self.max_steps:
            return True, f"({self.max_steps})"

        if len(self.history) >= 4:
            recent = self.history[-4:]
            if len(set(recent)) <= 2:
                return True, f": {recent}"

        return False, ""

def safe_tool_call(tool_func, *args, max_retries=3, **kwargs):
    """SecurityTool invocation"""
    for attempt in range(1, max_retries + 1):
        try:
            return tool_func(*args, **kwargs)
        except requests.Timeout:
            print(f"[] {tool_func.__name__} {attempt}")
            if attempt == max_retries:
                return {"error": "service_unavailable", "message": "Retry"}
        except requests.HTTPError as e:
            status = e.response.status_code
            if status == 429:
                time.sleep(2 ** attempt)
                continue
            elif 500 <= status < 600:
                time.sleep(1)
                continue
            else:
                return {"error": "bad_request", "message": str(e)}
    return {"error": "max_retries_exceeded", "message": "Retry"}