A langchain callback handler error typically manifests as a raised exception that kills an otherwise healthy LLM run. In most cases the root cause is a bug in your custom CallbackHandler subclass—a missing async method, an unhandled None, or a blocking call—not the model itself.
Step 1: Reproduce the failure with a minimal chain
Do not debug against your full RAG pipeline. Strip the chain down to a fake LLM and the suspect handler so the langchain callback handler error is the only variable.
from langchain.callbacks.base import BaseCallbackHandler
from langchain.llms import FakeListLLM
class BuggyHandler(BaseCallbackHandler):
def on_llm_start(self, *args, **kwargs):
# Intentional bug: expects a kwarg that may be absent
print(kwargs["nonexistent_key"])
llm = FakeListLLM(responses=["ok"])
chain = llm.bind(callbacks=[BuggyHandler()])
try:
chain.invoke("test")
except Exception as e:
print(f"Reproduced langchain callback handler error: {type(e).__name__}: {e}")Run this with python repro.py. If you see the exception printed, you have a clean repro. If not, the error is likely triggered by a specific event (e.g., on_llm_error) or by async execution paths.
Step 2: Enable LangChain’s verbose tracing
LangChain swallows some callback exceptions to protect the main run, which hides the real traceback. Force full visibility:
import langchain
langchain.debug = True # or set LANGCHAIN_VERBOSE=true in envAlternatively, use the ConsoleCallbackHandler alongside your handler to confirm which lifecycle event fires before the crash:
from langchain.callbacks import ConsoleCallbackHandler
chain = llm.bind(callbacks=[BuggyHandler(), ConsoleCallbackHandler()])Verify success
You should now see per-event logs (on_llm_start, on_llm_end) in stderr. The exact event where the langchain callback handler error appears tells you which method to fix.
Step 3: Audit handler method signatures (sync vs async)
The most common class of langchain callback handler error is a signature mismatch. LangChain calls BaseCallbackHandler methods synchronously and AsyncCallbackHandler methods with await. Mixing them raises TypeError or RuntimeError: await wasn't used with future.
from langchain.callbacks.base import AsyncCallbackHandler
class SafeAsyncHandler(AsyncCallbackHandler):
async def on_llm_end(self, response, **kwargs):
# correct: async method for async chain
await log_completion(response)If your chain is built with ainvoke, use AsyncCallbackHandler. If you use invoke, use BaseCallbackHandler. Never call asyncio.run() inside a sync handler—it will crash under an already-running loop.
Step 4: Wrap handler logic in defensive try/except
A callback handler should never abort the user-facing chain unless that is explicit product behavior. Catch, log, and move on:
import logging
class RobustHandler(BaseCallbackHandler):
def on_llm_error(self, error, **kwargs):
try:
metrics.incr("llm_error")
notify_slack(error)
except Exception:
logging.exception("Callback handler side-effect failed")This contains the langchain callback handler error to your logs instead of the call stack. Use structured logging, not print, so you can filter later.
Step 5: Validate side-effect ordering and idempotency
Handlers that write to databases or queues often fail because they assume on_llm_end always follows on_llm_start. Under timeouts or fallback, on_llm_error fires instead. Write handlers that tolerate missing counterparts:
class TraceHandler(BaseCallbackHandler):
def __init__(self):
self._started = {}
def on_llm_start(self, serialized, prompts, *, run_id, **kwargs):
self._started[run_id] = prompts
def on_llm_end(self, response, *, run_id, **kwargs):
prompts = self._started.pop(run_id, [])
store_trace(prompts, response)
def on_llm_error(self, error, *, run_id, **kwargs):
prompts = self._started.pop(run_id, [])
store_trace(prompts, error)Idempotency matters: LangChain may retry a run internally, firing the same handler twice. Guard with run_id dedupe.
Step 6: Isolate provider noise from handler logic
Sometimes the langchain callback handler error is a downstream symptom of a provider timeout that lands in on_llm_error and breaks your reporting code. During debugging, remove provider variance by routing through a stable gateway. A single OpenAI-compatible endpoint that applies automatic fallback when a provider is rate-limited—such as n4n.ai—lets you confirm whether the handler bug reproduces independent of model flakiness.
from langchain_openai import ChatOpenAI
# Point LangChain at a gateway instead of a direct provider
llm = ChatOpenAI(
base_url="https://api.n4n.ai/v1",
api_key="your-key",
model="gpt-4o-mini",
)With fallback handled upstream, you can focus on whether your handler correctly processes both successful and error events.
Step 7: Test handlers against mock events
Do not wait for production traffic. Unit-test handlers by calling their methods directly with LangChain’s event objects:
import pytest
from langchain.schema import LLMResult
def test_robust_handler_no_raise():
h = RobustHandler()
# Simulate an error event with a broken side-effect
with pytest.raises(Exception):
h.on_llm_error(ValueError("x")) # if side-effect also broken, logged not raised
# Assert metrics incremented in mockFor async handlers, use asyncio.run in the test only when no loop is active, or mark the test async def with pytest-asyncio.
Step 8: Verify end-to-end success
Replace the buggy handler in your real chain and run a smoke test:
python -m pytest tests/test_callbacks.py -q
python smoke.py # invokes chain with the fixed handlerSuccess criteria:
- No
langchain callback handler errorin logs. - Expected side effects (DB rows, metrics, traces) present for both happy path and injected error.
- Chain latency unchanged when handler side effects are slow (they should be async or offloaded).
If all three hold, the debugging cycle is closed. The fix is not “suppress the exception”—it is making the handler correct for every lifecycle event LangChain actually emits.