LangChain

LangChain Callbacks: Hooking Into Every Step of a Chain

Ira Menon · Jul 2, 2026 · 18 min read

Why Your Chain Is a Black Box Without This

You built a chain. It calls a retriever, feeds context into a prompt, hits the LLM, maybe calls a tool, and returns an answer. It works — until it doesn't. A user reports a weird response, and you have no idea what the retriever actually returned, how many tokens the LLM call burned, or whether a tool silently failed halfway through. Without instrumentation, a chain is a black box: input goes in, output comes out, and everything in between is a guess.

This is exactly the problem LangChain's callback system solves. Callbacks are hooks that fire at every meaningful event inside a chain's execution — when an LLM starts generating, when it streams a token, when a tool is invoked, when a chain errors out, when the whole thing finishes. You attach a handler once, and from then on you get a live feed of everything happening inside your application, without touching the core logic of the chain itself.

In this article we'll build real BaseCallbackHandler subclasses from scratch, wire them into chains built with LangChain Expression Language (LCEL), stream tokens to a terminal in real time, and use callbacks for the two things they're actually good for in production: observability and cost tracking. By the end you'll understand not just how callbacks work, but where to plug them in so you're not debugging blind the next time something breaks.

What a Callback Handler Actually Is

At its core, a callback handler is a class that implements a set of on_* methods. LangChain calls these methods automatically as execution moves through your chain. You don't call them yourself — you just define what should happen when they fire.

The base class is BaseCallbackHandler, and it exposes methods for nearly every stage of execution:

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any
from uuid import UUID

class MinimalHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized: dict, prompts: list[str], **kwargs: Any) -> None:
        print(f"LLM starting with {len(prompts)} prompt(s)")

    def on_llm_end(self, response, **kwargs: Any) -> None:
        print("LLM finished")

    def on_llm_error(self, error: BaseException, **kwargs: Any) -> None:
        print(f"LLM errored: {error}")

    def on_chain_start(self, serialized: dict, inputs: dict, **kwargs: Any) -> None:
        print(f"Chain starting with inputs: {inputs}")

    def on_chain_end(self, outputs: dict, **kwargs: Any) -> None:
        print(f"Chain finished with outputs: {outputs}")

    def on_tool_start(self, serialized: dict, input_str: str, **kwargs: Any) -> None:
        print(f"Tool starting with input: {input_str}")

    def on_tool_end(self, output: str, **kwargs: Any) -> None:
        print(f"Tool finished: {output}")

Every one of these methods is optional to override — you only implement the events you care about. If you only want to log LLM starts and ends, you implement two methods and leave the rest alone. LangChain checks which methods exist on your handler and calls the ones that are defined.

Notice the **kwargs in every signature. This is not optional boilerplate — LangChain passes additional context through kwargs (like run_id, parent_run_id, and tags), and the internal calling convention can pass extra arguments depending on the LangChain version and the component emitting the event. Dropping **kwargs is the single most common reason a custom handler throws a TypeError in production.

Attaching Callbacks to a Chain

There are three places you can attach a callback handler, and which one you pick changes its scope.

1. Request-time, via `config` — the handler only fires for this one invocation:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

handler = MinimalHandler()

prompt = ChatPromptTemplate.from_template("Explain {topic} in two sentences.")
model = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | model | StrOutputParser()

result = chain.invoke(
    {"topic": "vector databases"},
    config={"callbacks": [handler]},
)

2. Constructor-time, on the model itself — the handler fires for every call made through that model instance, regardless of which chain uses it:

model = ChatOpenAI(model="gpt-4o-mini", callbacks=[handler])

3. Globally, via environment variable or `set_verbose`/`set_debug` — useful for blanket debugging, not recommended for production because it fires for literally everything, including nested sub-chains you may not care about.

The request-time approach (config={"callbacks": [...]}) is almost always the right default. It propagates the handler down through every step of a chain — retriever, prompt, model, output parser — without leaking into unrelated invocations elsewhere in your app. This propagation is important: if you have a chain composed of sub-chains, a callback attached at the top level will still see events fired deep inside a nested chain, because LangChain passes callbacks down through the execution tree automatically.

Building a Real Handler: Token-Level Streaming

The most common reason people reach for callbacks is streaming tokens to a UI as they're generated, rather than waiting for the full response. Here's a handler that does this properly, with state tracking so you can tell which run produced which tokens:

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any
from uuid import UUID

class StreamToConsoleHandler(BaseCallbackHandler):
    """Streams tokens to stdout as they arrive, prefixed by run id."""

    def __init__(self):
        self.tokens_seen: dict[str, int] = {}

    def on_llm_start(
        self,
        serialized: dict,
        prompts: list[str],
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> None:
        self.tokens_seen[str(run_id)] = 0
        print(f"\n--- generation started (run {str(run_id)[:8]}) ---")

    def on_llm_new_token(
        self,
        token: str,
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> None:
        self.tokens_seen[str(run_id)] = self.tokens_seen.get(str(run_id), 0) + 1
        print(token, end="", flush=True)

    def on_llm_end(
        self,
        response,
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> None:
        count = self.tokens_seen.get(str(run_id), 0)
        print(f"\n--- generation finished ({count} tokens streamed) ---")

To actually see on_llm_new_token fire, the model needs streaming=True — callbacks don't turn streaming on by themselves, they just observe it:

model = ChatOpenAI(model="gpt-4o-mini", streaming=True)
chain = prompt | model | StrOutputParser()

chain.invoke(
    {"topic": "embedding models"},
    config={"callbacks": [StreamToConsoleHandler()]},
)

This is the pattern behind most "typewriter effect" chat UIs you've used. The frontend doesn't wait for the full LLM response — it renders on_llm_new_token events as they arrive over a websocket or server-sent events connection, and the callback handler is the bridge between LangChain's internal streaming loop and whatever transport you're using to push tokens to the browser.

Tracking Cost and Token Usage

The other production-grade use case is cost tracking. Every LLM response carries token usage metadata, and on_llm_end is where you get access to it. Here's a handler that accumulates cost across an entire chain run, including nested calls:

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any

# Rough per-1K-token pricing, update to match your actual model contract
PRICING = {
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
    "gpt-4o": {"input": 0.0025, "output": 0.01},
}

class CostTrackingHandler(BaseCallbackHandler):
    def __init__(self):
        self.total_prompt_tokens = 0
        self.total_completion_tokens = 0
        self.total_cost = 0.0
        self.call_count = 0

    def on_llm_end(self, response, **kwargs: Any) -> None:
        self.call_count += 1
        usage = None

        # llm_output carries token usage on most chat model integrations
        if response.llm_output and "token_usage" in response.llm_output:
            usage = response.llm_output["token_usage"]
            model_name = response.llm_output.get("model_name", "gpt-4o-mini")
        else:
            return

        prompt_tokens = usage.get("prompt_tokens", 0)
        completion_tokens = usage.get("completion_tokens", 0)

        self.total_prompt_tokens += prompt_tokens
        self.total_completion_tokens += completion_tokens

        rates = PRICING.get(model_name, PRICING["gpt-4o-mini"])
        cost = (prompt_tokens / 1000) * rates["input"] + (
            completion_tokens / 1000
        ) * rates["output"]
        self.total_cost += cost

    def summary(self) -> str:
        return (
            f"{self.call_count} LLM call(s) | "
            f"{self.total_prompt_tokens} prompt tokens | "
            f"{self.total_completion_tokens} completion tokens | "
            f"${self.total_cost:.5f} estimated cost"
        )

Use it across a multi-step chain and you get a single running total, even if the chain calls the model three or four times internally (for example, one call to rephrase a query, one to generate the answer, and one to summarize a tool result):

tracker = CostTrackingHandler()

chain.invoke(
    {"topic": "retrieval augmented generation"},
    config={"callbacks": [tracker]},
)

print(tracker.summary())

This is genuinely useful once you're running chains against real traffic. Token usage is easy to reason about for a single prompt-response pair, but it becomes opaque fast once you add retries, retrieval steps, and tool calls — the callback handler is what keeps the total honest without you having to manually thread token counts through every function call.

Scoping Handlers With Tags and Metadata

Once you have more than one chain running in the same process — say, a retrieval chain and a summarization chain sharing the same callback handler for centralized logging — you'll quickly want to know which invocation produced which event. This is what tags and metadata are for. Both can be passed through config alongside callbacks, and both show up in the **kwargs of every hook:

class TaggedLoggingHandler(BaseCallbackHandler):
    def on_chain_start(
        self, serialized: dict, inputs: dict, *, tags: list[str] | None = None,
        metadata: dict | None = None, **kwargs: Any,
    ) -> None:
        tag_str = ",".join(tags) if tags else "untagged"
        user_id = (metadata or {}).get("user_id", "unknown")
        print(f"[{tag_str}] chain start for user={user_id}")

handler = TaggedLoggingHandler()

retrieval_chain.invoke(
    {"query": "what is a token budget"},
    config={
        "callbacks": [handler],
        "tags": ["retrieval"],
        "metadata": {"user_id": "u_1029"},
    },
)

summarization_chain.invoke(
    {"text": "..."},
    config={
        "callbacks": [handler],
        "tags": ["summarization"],
        "metadata": {"user_id": "u_1029"},
    },
)

Both events land in the same TaggedLoggingHandler instance, but the tag tells you which logical part of your pipeline fired it, and the metadata carries whatever business context you care about — a user id, a request id, a feature flag, a tenant name in a multi-tenant SaaS product. This is the pattern that makes a single shared handler viable across an entire application instead of writing a bespoke handler per chain. In practice, most teams end up with one central logging handler that reads tags and metadata to route events to the right place — a specific log index, a specific dashboard panel, a specific customer's usage report — rather than one handler per feature.

You can also filter which events a handler listens to at attachment time by only implementing the hooks you need, but tags solve a different problem: they let the *same* hook distinguish *why* it fired. That distinction matters once a callback handler is shared across a codebase with more than a couple of chains in it, because without tags you're stuck grepping log output to reconstruct which chain produced which line — exactly the kind of manual correlation callbacks are supposed to eliminate in the first place.

Async Callbacks for Async Chains

If your app runs chains with ainvoke or astream — which you should be doing for anything served behind a web framework like FastAPI — your handler needs async versions of the same methods. LangChain provides AsyncCallbackHandler for exactly this:

from langchain_core.callbacks import AsyncCallbackHandler
from typing import Any

class AsyncLoggingHandler(AsyncCallbackHandler):
    async def on_chain_start(
        self, serialized: dict, inputs: dict, **kwargs: Any
    ) -> None:
        print(f"[async] chain start: {inputs}")

    async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        print(token, end="", flush=True)

    async def on_chain_end(self, outputs: dict, **kwargs: Any) -> None:
        print(f"\n[async] chain end: {outputs}")

import asyncio

async def main():
    async for chunk in chain.astream(
        {"topic": "async generators in Python"},
        config={"callbacks": [AsyncLoggingHandler()]},
    ):
        pass  # the handler already prints tokens as they arrive

asyncio.run(main())

A subtle but important detail: if you use a synchronous BaseCallbackHandler inside an async chain, LangChain will run its methods in a thread pool executor to avoid blocking the event loop. That works, but it adds overhead and can reorder log output relative to other async operations. If your chain is async end-to-end, make your handlers async end-to-end too — mixing the two is a common source of subtle timing bugs where log lines appear out of order.

Handling Errors Without Crashing the Chain

Callback handlers should never throw. If on_llm_end raises an exception because you didn't guard against a missing key, that exception can propagate and kill an otherwise-successful chain run — which is a particularly frustrating way to lose a good response over a logging bug. Defensive coding in handlers is not optional:

class SafeLoggingHandler(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs: Any) -> None:
        try:
            usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
            print(f"Tokens used: {usage.get('total_tokens', 'unknown')}")
        except Exception as e:
            # Never let a logging failure take down the chain
            print(f"[callback error, ignored] {e}")

    def on_chain_error(self, error: BaseException, **kwargs: Any) -> None:
        print(f"Chain failed: {type(error).__name__}: {error}")

    def on_tool_error(self, error: BaseException, **kwargs: Any) -> None:
        print(f"Tool failed: {type(error).__name__}: {error}")

Note that on_chain_error and on_tool_error are separate hooks from on_llm_error — a tool failing and an LLM call failing are different events with different handlers, and you'll want to distinguish them in your logs. A tool timing out because a downstream API is slow is a very different incident from the LLM provider returning a rate-limit error, and lumping both into one generic "something broke" log line makes on-call debugging much harder than it needs to be.

Combining Multiple Handlers

You are not limited to one handler per chain. Pass a list, and every handler in it receives every event — this is how you separate concerns cleanly, one handler per responsibility, instead of building a single handler that tries to do logging, cost tracking, and UI streaming all at once:

stream_handler = StreamToConsoleHandler()
cost_handler = CostTrackingHandler()
safe_handler = SafeLoggingHandler()

chain.invoke(
    {"topic": "the transformer attention mechanism"},
    config={"callbacks": [stream_handler, cost_handler, safe_handler]},
)

print(cost_handler.summary())

This composability is one of the most underrated aspects of the callback system. In a real application you'll typically want at minimum three handlers running at once: one that streams to the frontend, one that logs structured events to your observability platform, and one that tracks cost per request so you can attribute spend to a specific user or feature. Keeping them as separate classes means you can add or remove any one of them without touching the others, and you can unit test each in isolation.

Callbacks vs. Tracing Platforms — When to Reach for Which

It's worth being clear about what callbacks are and aren't. Writing your own BaseCallbackHandler gives you full control and zero external dependencies — everything happens in your own process, in code you own and can modify. This is the right choice when you need custom business logic tied to specific events: writing to your own database, triggering an alert when a tool errors more than N times, or computing a cost metric that's specific to your pricing model.

Managed tracing platforms that integrate with LangChain (LangSmith being the most common one, since it's built by the same team) are built on top of this exact same callback mechanism — they register a callback handler under the hood that ships trace data to a hosted dashboard. They're the better choice when you want a full visual trace of nested chain calls, side-by-side prompt comparisons, and dataset-based evaluation without writing any handler code yourself.

The important thing to understand is that these aren't competing systems — a tracing platform's integration is just another callback handler, using the exact same on_llm_start/on_llm_end/on_chain_start interface you've been building by hand in this article. Once you understand the callback contract, you understand how every observability integration for LangChain works under the hood, including ones you didn't write yourself.

Common Pitfalls

A few mistakes come up often enough to call out directly.

Forgetting `kwargs** in a handler method signature. LangChain's internal call sites pass extra keyword arguments (run_id, parent_run_id, tags, metadata) that vary by version and by which component is emitting the event. Without **kwargs your handler throws a TypeError` the moment it's actually exercised.
Attaching callbacks at the model constructor when you meant request-scoped. If you put a handler in ChatOpenAI(callbacks=[handler]), it fires for every single call made through that model object for its entire lifetime — including calls from unrelated chains that happen to reuse the same model instance. This causes handlers to accumulate stale or cross-contaminated state, especially in cost trackers where you don't want to reset totals between requests but also don't want them summed across different users.
Blocking work inside a callback method. A handler that makes a synchronous HTTP call inside on_llm_new_token will slow down every single token of every stream. If a handler needs to do expensive I/O, buffer events and flush them asynchronously, don't do it inline in the hook.
Not distinguishing `run_id` from `parent_run_id`. In a chain with nested LLM calls, every event carries both. Ignoring parent_run_id means you can't tell whether three token-usage events came from one call or three separate one — a mistake that quietly wrecks cost attribution the moment your chain has more than one LLM call in it.
Assuming callbacks catch everything. Callbacks fire around the LangChain-managed lifecycle. Code you run entirely outside a chain invocation — a manual requests.get() call before you ever touch LangChain — will never trigger a callback event, because there's no chain execution happening to hook into.

Scoping Handlers With Tags and Metadata

class TaggedLoggingHandler(BaseCallbackHandler):
    def on_chain_start(
        self, serialized: dict, inputs: dict, *, tags: list[str] | None = None,
        metadata: dict | None = None, **kwargs: Any,
    ) -> None:
        tag_str = ",".join(tags) if tags else "untagged"
        user_id = (metadata or {}).get("user_id", "unknown")
        print(f"[{tag_str}] chain start for user={user_id}")

handler = TaggedLoggingHandler()

retrieval_chain.invoke(
    {"query": "what is a token budget"},
    config={
        "callbacks": [handler],
        "tags": ["retrieval"],
        "metadata": {"user_id": "u_1029"},
    },
)

summarization_chain.invoke(
    {"text": "..."},
    config={
        "callbacks": [handler],
        "tags": ["summarization"],
        "metadata": {"user_id": "u_1029"},
    },
)

Callbacks in Retrieval-Augmented Chains

Callbacks become especially valuable in RAG pipelines, because a RAG chain has more moving parts than a simple prompt-to-model call — a retriever step, a document-formatting step, the LLM call itself — and any one of them can be the actual source of a bad answer. Wiring a handler into on_retriever_start and on_retriever_end lets you inspect exactly what context the model was given, which is often the fastest way to diagnose a "the answer is wrong" bug report:

class RetrievalDebugHandler(BaseCallbackHandler):
    def on_retriever_start(self, serialized: dict, query: str, **kwargs: Any) -> None:
        print(f"Retrieving for query: {query!r}")

    def on_retriever_end(self, documents, **kwargs: Any) -> None:
        print(f"Retrieved {len(documents)} document(s):")
        for i, doc in enumerate(documents):
            preview = doc.page_content[:80].replace("\n", " ")
            print(f"  [{i}] {preview}...")

Attach this alongside your cost tracker and streaming handler, and you get a full picture in one run: what was retrieved, what it cost, and what got streamed back to the user. Without this, diagnosing a bad RAG answer usually means manually re-running the retriever in a notebook to guess at what the chain saw — a slow, disconnected debugging loop that callbacks let you skip entirely, because the actual retrieved documents from the actual failing request are sitting right there in your logs.

This same pattern extends to agents. An agent chain that decides between multiple tools benefits enormously from on_agent_action and on_agent_finish hooks, which tell you exactly which tool the agent chose and why, before it ever executes that tool:

class AgentTraceHandler(BaseCallbackHandler):
    def on_agent_action(self, action, **kwargs: Any) -> None:
        print(f"Agent chose tool: {action.tool} with input: {action.tool_input}")

    def on_agent_finish(self, finish, **kwargs: Any) -> None:
        print(f"Agent finished: {finish.return_values}")

Without this hook, debugging an agent that picked the wrong tool means reading through the raw LLM output line by line looking for the tool-selection reasoning. With it, you get a clean, structured record of every decision the agent made, in the order it made them — which is usually the difference between finding the bug in thirty seconds and losing an afternoon to it.

Wrapping Up

Callbacks turn a LangChain chain from a black box into something you can actually watch run. Once you have a BaseCallbackHandler (or its async twin) wired into your chains via config={"callbacks": [...]}, you get visibility into every LLM call, every tool invocation, every token as it streams, and every error as it happens — without changing a line of your actual chain logic. That separation is the whole point: your business logic stays clean, and your observability lives in handlers you can add, remove, and test independently.

Start small. Add one handler that logs chain start and end events. Once that feels natural, add a cost tracker. Then wire in streaming for your UI. By the time you're running chains in production, you'll have exactly the visibility you need, and none of it will have required touching your prompts or your chain composition.

If you want to go deeper — building production-grade agent pipelines with proper observability, streaming, memory, and tool orchestration from the ground up — that's exactly what we cover hands-on in the LangChain Tutorial 2026 course on teachyou.ai.