LangChain Mastery

Course Content

LangChain Mastery

7 sections · 109 lessons

Write a function to implement a custom error handler for LangChain.

What you need to know

Goals of an error handler

  • Classify — callers need "retry later", "fix your input" or "our bug", not a stack trace.
  • Log once, with context — request id, error type, step.
  • Protect users — never show raw exception text; it can leak internals.
  • Don't hide outages — a fallback that silently returns "Sorry" forever is worse than an alert.

The chain wrapper

Python
import loggingimport openaifrom pydantic import ValidationErrorfrom langchain_core.exceptions import OutputParserExceptionfrom langchain_core.runnables import RunnableLambdalog = logging.getLogger(__name__)ERRORS = [   # (exception types, code, message for the user)    ((ValidationError,), "bad_input", "Please check your request and try again."),    ((OutputParserException,), "bad_format", "I couldn't produce a valid answer. Please try again."),    ((openai.RateLimitError, openai.APITimeoutError), "busy", "The assistant is busy. Try again in a minute."),    ((openai.APIError,), "upstream", "The assistant is temporarily unavailable."),]def with_error_handling(chain):    def run(inputs: dict, config) -> dict:        try:            return {"ok": True, "data": chain.invoke(inputs, config)}        except Exception as exc:            for types, code, message in ERRORS:                if isinstance(exc, types):                    log.warning(code, exc_info=exc, extra={"run_id": str(config.get("run_id"))})                    return {"ok": False, "error": code, "message": message}            raise                          # unknown: a real bug — let it alert    return RunnableLambda(run)
  • RunnableLambda passes config if the function accepts it; forwarding it to chain.invoke keeps callbacks and tracing connected.
  • Unknown exceptions are re-raised, so real bugs reach your error tracker.
  • The order of ERRORS matters: specific before general.

Tool errors inside agents

In an agent, a failing tool should usually become a message the model can react to ("stock API timed out; try again or tell the user"). ToolErrorMiddleware(on_error) lets you choose which exceptions become an error ToolMessage; anything you don't handle still propagates. Or write your own:

Python
from langchain.agents.middleware import wrap_tool_callfrom langchain.messages import ToolMessage@wrap_tool_calldef tool_errors(request, handler):    try:        return handler(request)    except TimeoutError:        return ToolMessage(content="The service timed out. Tell the user to retry shortly.",                           tool_call_id=request.tool_call["id"], status="error")

Argument-validation errors are already turned into error messages by the agent's tool node before the tool runs.

A real-life example

An HR bot's leave-filing flow calls the HR system's API. When the HR API returned 503 during a maintenance window, the agent crashed and employees saw a blank error page.

With the wrap_tool_call handler, a 503 became "The HR system is under maintenance until 6 pm; your request was not filed." The model passed this on and offered to remind the employee later. The chain-level wrapper mapped remaining failures to codes, and the dashboard showed upstream errors spiking at 2 pm — matching the maintenance window — while bad_format stayed flat, so nobody wasted time on prompts.

Follow-up questions to expect

  • "Why return a result object instead of raising?" — Callers, such as an API handler or a batch job, can branch on a code without knowing LangChain's exception classes.
  • "Should the model see raw exception text?" — No; it may contain internal hostnames, SQL or personal data. Send a short, safe description.
  • "Where do retries go relative to this handler?" — Inside it. The handler sees only errors that survived retries and fallbacks.