tessl/pypi-langchain

Building applications with LLMs through composability

—

Pending

Overview

Eval results

Files

Error Handling Patterns

Name: tessl/pypi-langchain
Author: tessl

Robust error handling is essential for building reliable LLM applications. LangChain provides structured error handling mechanisms for tools, validation, and retry patterns.

Core Concepts

Error handling in LangChain works at multiple levels:

ToolException: Structured errors that communicate with the LLM
Validation Errors: Input validation and schema enforcement
Retry Patterns: Automatic retry with exponential backoff
Error Recovery: LLM-driven error recovery and correction

ToolException Usage

Basic ToolException

ToolException is the primary mechanism for communicating tool errors to the LLM:

from langchain.tools import ToolException

raise ToolException("Error message for the LLM")

The key insight: When a tool raises a ToolException, the error message is sent back to the LLM in a ToolMessage, allowing the LLM to understand what went wrong and potentially retry with different parameters.

Basic Example:

from langchain.tools import tool, ToolException

@tool
def divide_numbers(a: float, b: float) -> float:
    """Divide two numbers.

    Args:
        a: Numerator
        b: Denominator

    Returns:
        Result of a / b
    """
    if b == 0:
        raise ToolException(
            "Cannot divide by zero. Please provide a non-zero denominator."
        )
    return a / b

How it works:

from langchain.agents import create_agent
from langchain.messages import HumanMessage

agent = create_agent(
    model="openai:gpt-4o",
    tools=[divide_numbers]
)

# When the LLM tries to divide by zero:
# 1. Tool raises ToolException
# 2. Error message sent to LLM in ToolMessage
# 3. LLM sees the error and can try again with valid parameters
result = agent.invoke({
    "messages": [HumanMessage(content="What is 10 divided by 0?")]
})

# The LLM will see the error and respond appropriately

Error Messages for LLM

Write clear, actionable error messages that help the LLM correct its behavior:

from langchain.tools import tool, ToolException

@tool
def fetch_url(url: str) -> str:
    """Fetch content from a URL.

    Args:
        url: URL to fetch

    Returns:
        Page content
    """
    import requests

    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.Timeout:
        raise ToolException(
            f"Request to {url} timed out after 10 seconds. "
            "The server may be slow or unreachable. "
            "Try a different URL or try again later."
        )
    except requests.HTTPError as e:
        raise ToolException(
            f"HTTP error {e.response.status_code} when fetching {url}. "
            f"The URL may be invalid or the server may be down. "
            f"Please verify the URL is correct."
        )
    except requests.RequestException as e:
        raise ToolException(
            f"Failed to fetch {url}: {str(e)}. "
            f"Please check the URL and try again."
        )

Validation Errors

Use ToolException for input validation:

from langchain.tools import tool, ToolException
from datetime import datetime

@tool
def schedule_meeting(
    date: str,
    time: str,
    duration_minutes: int,
    attendees: list[str]
) -> str:
    """Schedule a meeting.

    Args:
        date: Meeting date in YYYY-MM-DD format
        time: Meeting time in HH:MM format
        duration_minutes: Duration in minutes (15-480)
        attendees: List of attendee email addresses
    """
    # Validate date format
    try:
        datetime.strptime(date, '%Y-%m-%d')
    except ValueError:
        raise ToolException(
            f"Invalid date format: '{date}'. "
            f"Please use YYYY-MM-DD format (e.g., 2024-12-31)."
        )

    # Validate time format
    try:
        datetime.strptime(time, '%H:%M')
    except ValueError:
        raise ToolException(
            f"Invalid time format: '{time}'. "
            f"Please use HH:MM format in 24-hour notation (e.g., 14:30 for 2:30 PM)."
        )

    # Validate duration
    if not 15 <= duration_minutes <= 480:
        raise ToolException(
            f"Invalid duration: {duration_minutes} minutes. "
            f"Duration must be between 15 and 480 minutes (8 hours). "
            f"Please provide a valid duration."
        )

    # Validate attendees
    if len(attendees) == 0:
        raise ToolException(
            "At least one attendee is required. "
            "Please provide attendee email addresses."
        )

    # Validate email format
    import re
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    for email in attendees:
        if not re.match(email_pattern, email):
            raise ToolException(
                f"Invalid email address: '{email}'. "
                f"Please provide valid email addresses."
            )

    # Schedule meeting
    meeting_id = f"meeting_{date}_{time}"
    return f"Meeting scheduled successfully. Meeting ID: {meeting_id}"

Contextual Error Messages

Include context in error messages to help the LLM understand what went wrong:

from langchain.tools import tool, ToolException

@tool
def query_database(query: str) -> list[dict]:
    """Execute a SQL query on the database.

    Args:
        query: SQL query to execute (SELECT only)

    Returns:
        Query results
    """
    # Security validation
    if not query.strip().upper().startswith('SELECT'):
        raise ToolException(
            "Only SELECT queries are allowed for safety. "
            "Queries must start with 'SELECT'. "
            f"Your query started with: '{query.split()[0]}'"
        )

    try:
        # Execute query
        results = execute_query(query)

        if len(results) == 0:
            raise ToolException(
                "Query returned no results. "
                "The search criteria may be too restrictive. "
                "Try broadening your search or checking table names."
            )

        return results

    except DatabaseConnectionError:
        raise ToolException(
            "Database connection lost. "
            "Please try again in a moment."
        )
    except QuerySyntaxError as e:
        raise ToolException(
            f"SQL syntax error: {e}. "
            f"Please check your query syntax. "
            f"Common issues: missing quotes, invalid column names, incorrect table names."
        )
    except QueryTimeoutError:
        raise ToolException(
            "Query timed out after 30 seconds. "
            "The query may be too complex or the database may be under heavy load. "
            "Try simplifying the query or adding more specific filters."
        )

def execute_query(query: str) -> list[dict]:
    """Simulate database query."""
    # Implementation
    return []

class DatabaseConnectionError(Exception):
    pass

class QuerySyntaxError(Exception):
    pass

class QueryTimeoutError(Exception):
    pass

Retry Patterns

Manual Retry with ToolException

The LLM automatically retries when it receives a ToolException:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolException
import random

@tool
def flaky_api_call(endpoint: str) -> dict:
    """Call an API that sometimes fails.

    Args:
        endpoint: API endpoint to call

    Returns:
        API response
    """
    # Simulate flaky API
    if random.random() < 0.5:
        raise ToolException(
            f"API call to {endpoint} failed. "
            f"This is a temporary error. Please try again."
        )

    return {"status": "success", "data": "result"}

agent = create_agent(
    model="openai:gpt-4o",
    tools=[flaky_api_call]
)

# LLM will automatically retry if it gets ToolException
result = agent.invoke({
    "messages": [HumanMessage(content="Call the /users endpoint")]
})

Exponential Backoff

Implement exponential backoff for rate-limited APIs:

from langchain.tools import tool, ToolException
import time

@tool
def rate_limited_api(endpoint: str) -> dict:
    """Call a rate-limited API with exponential backoff.

    Args:
        endpoint: API endpoint to call

    Returns:
        API response
    """
    max_retries = 3
    base_delay = 1

    for attempt in range(max_retries):
        try:
            # Make API call
            response = make_api_call(endpoint)
            return response

        except RateLimitError as e:
            if attempt == max_retries - 1:
                # Last attempt failed
                raise ToolException(
                    f"API rate limit exceeded after {max_retries} attempts. "
                    f"Please try again in a few minutes."
                )

            # Calculate backoff delay
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)

            # Don't raise exception yet, will retry

    # Should never reach here
    raise ToolException("Unexpected error in rate_limited_api")

def make_api_call(endpoint: str) -> dict:
    """Simulate API call."""
    return {"data": "result"}

class RateLimitError(Exception):
    pass

Retry with State Tracking

Track retry attempts using tool state:

from langchain.tools import BaseTool, ToolException
from pydantic import BaseModel, Field

class RetryAPIInput(BaseModel):
    endpoint: str = Field(description="API endpoint to call")

class RetryAPITool(BaseTool):
    name: str = "retry_api"
    description: str = "Call an API with automatic retry"
    args_schema: type[BaseModel] = RetryAPIInput

    # Track retries per endpoint
    retry_counts: dict[str, int] = {}
    max_retries: int = 3

    def _run(self, endpoint: str) -> dict:
        """Execute with retry tracking."""
        # Get current retry count
        count = self.retry_counts.get(endpoint, 0)

        try:
            # Attempt API call
            response = make_api_call(endpoint)

            # Success - reset counter
            self.retry_counts[endpoint] = 0
            return response

        except Exception as e:
            # Increment retry count
            count += 1
            self.retry_counts[endpoint] = count

            if count >= self.max_retries:
                # Max retries reached
                self.retry_counts[endpoint] = 0  # Reset for next time
                raise ToolException(
                    f"API call to {endpoint} failed after {self.max_retries} attempts. "
                    f"Last error: {str(e)}. "
                    f"Please check the endpoint and try again later."
                )
            else:
                # Retry available
                raise ToolException(
                    f"API call to {endpoint} failed (attempt {count}/{self.max_retries}). "
                    f"Error: {str(e)}. Retrying..."
                )

    async def _arun(self, endpoint: str) -> dict:
        return self._run(endpoint)

def make_api_call(endpoint: str) -> dict:
    """Simulate API call."""
    return {"data": "result"}

Error Recovery

LLM-Driven Error Correction

Let the LLM correct its own errors:

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolException
import json

@tool
def parse_json_data(json_string: str) -> dict:
    """Parse JSON data.

    Args:
        json_string: JSON string to parse

    Returns:
        Parsed JSON data
    """
    try:
        return json.loads(json_string)
    except json.JSONDecodeError as e:
        raise ToolException(
            f"Invalid JSON: {str(e)}. "
            f"Please check the JSON syntax. "
            f"Common issues: missing quotes, trailing commas, unescaped characters."
        )

@tool
def validate_email(email: str) -> bool:
    """Validate an email address.

    Args:
        email: Email address to validate

    Returns:
        True if valid, raises ToolException if invalid
    """
    import re
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'

    if not re.match(pattern, email):
        raise ToolException(
            f"Invalid email format: '{email}'. "
            f"Email must be in format: user@domain.com"
        )

    return True

agent = create_agent(
    model="openai:gpt-4o",
    tools=[parse_json_data, validate_email],
    system_prompt="When you encounter errors, read the error message carefully and correct your input."
)

# LLM will automatically correct errors based on error messages
result = agent.invoke({
    "messages": [HumanMessage(
        content="Parse this JSON: {name: John, email: invalid-email}"
    )]
})

Best Practices

Write Clear Error Messages

# Bad error message
raise ToolException("Invalid input")

# Good error message
raise ToolException(
    f"Invalid date format: '{date}'. "
    f"Expected YYYY-MM-DD (e.g., 2024-12-31), got: '{date}'"
)

Include Recovery Instructions

# Bad - no guidance
raise ToolException("Query failed")

# Good - tells LLM how to recover
raise ToolException(
    "Query returned no results. "
    "Try broadening your search by: "
    "1. Removing some filter criteria, "
    "2. Using wildcards in text searches, "
    "3. Checking spelling of search terms"
)

Validate Early

@tool
def process_data(data: str, format: str) -> dict:
    """Process data in specified format."""
    # Validate early
    valid_formats = ["json", "xml", "csv"]
    if format not in valid_formats:
        raise ToolException(
            f"Invalid format: '{format}'. "
            f"Supported formats: {', '.join(valid_formats)}"
        )

    # Continue processing
    return process(data, format)

def process(data: str, format: str) -> dict:
    return {}

Common Mistakes

Mistake: Using Generic Exceptions

# Wrong - LLM can't see the error
@tool
def bad_tool(x: int) -> int:
    if x < 0:
        raise ValueError("Negative number")
    return x * 2

# Right - LLM receives error message
@tool
def good_tool(x: int) -> int:
    if x < 0:
        raise ToolException(
            "Cannot process negative numbers. "
            "Please provide a positive number."
        )
    return x * 2

Mistake: Vague Error Messages

# Wrong - not helpful
raise ToolException("Error occurred")

# Right - specific and actionable
raise ToolException(
    "Connection timeout after 30 seconds when connecting to database. "
    "The database may be unavailable. Please try again in a few minutes."
)

Mistake: Not Handling All Error Cases

# Wrong - uncaught exceptions crash the tool
@tool
def incomplete_tool(url: str) -> str:
    response = requests.get(url)  # May raise many exceptions
    return response.text

# Right - handle all error cases
@tool
def complete_tool(url: str) -> str:
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.text
    except requests.Timeout:
        raise ToolException(f"Request to {url} timed out")
    except requests.HTTPError as e:
        raise ToolException(f"HTTP error {e.response.status_code}")
    except requests.RequestException as e:
        raise ToolException(f"Request failed: {str(e)}")

Related Patterns

Async Patterns - Error handling in async operations
Streaming Patterns - Error handling during streaming
Persistence Patterns - Handling persistence errors

Install with Tessl CLI

npx tessl i tessl/pypi-langchain

docs

advanced

core

patterns

reference