LangChain callbacks: logging every step of a chain

LangChain callbacks logging chain steps is the most reliable way to understand what your LLM application actually does at runtime. Chains hide multiple model calls, tool invocations, and retrieval steps behind a single invoke() — without callbacks, you’re debugging a black box. This guide walks through building production-ready callback handlers that capture structured logs at every stage.

Why callbacks matter for chain observability

When a chain runs, it executes a directed acyclic graph of components: prompt formatting, model inference, output parsing, tool calls, retriever queries, and more. Each node emits events — on_chain_start, on_llm_start, on_tool_end, on_retriever_end, and their error counterparts. The callback system is the only supported mechanism to tap into these events without modifying library code.

Print statements scattered through your business logic don’t scale. They couple observability to implementation, break in async contexts, and vanish in distributed deployments. A proper callback handler separates concerns: your chain logic stays clean while logging, metrics, tracing, and audit trails live in reusable handlers.

The callback handler interface

LangChain defines BaseCallbackHandler with over 30 methods covering every lifecycle event. Most handlers only need a subset. The interface uses synchronous and asynchronous variants — implement both if your chains run in async contexts.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List, Optional
import json
import time
import uuid

class StructuredLoggingHandler(BaseCallbackHandler):
    def __init__(self, logger, include_inputs: bool = True, include_outputs: bool = True):
        self.logger = logger
        self.include_inputs = include_inputs
        self.include_outputs = include_outputs
        self._run_map: Dict[str, Dict] = {}

    def _log(self, level: str, event: str, run_id: str, parent_run_id: Optional[str], **kwargs):
        payload = {
            "timestamp": time.time(),
            "event": event,
            "run_id": str(run_id),
            "parent_run_id": str(parent_run_id) if parent_run_id else None,
            **kwargs
        }
        getattr(self.logger, level)(json.dumps(payload))

    def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._run_map[str(run_id)] = {"start": time.time(), "type": "chain"}
        self._log("info", "chain_start", run_id, parent_run_id, inputs=inputs if self.include_inputs else None)

    def on_chain_end(self, outputs: Dict[str, Any], *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        duration = time.time() - self._run_map.get(str(run_id), {}).get("start", time.time())
        self._log("info", "chain_end", run_id, parent_run_id, outputs=outputs if self.include_outputs else None, duration_ms=duration * 1000)
        self._run_map.pop(str(run_id), None)

    def on_chain_error(self, error: BaseException, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._log("error", "chain_error", run_id, parent_run_id, error=str(error), error_type=type(error).__name__)
        self._run_map.pop(str(run_id), None)

    # LLM events
    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._run_map[str(run_id)] = {"start": time.time(), "type": "llm", "model": serialized.get("kwargs", {}).get("model_name")}
        self._log("info", "llm_start", run_id, parent_run_id, prompts=prompts if self.include_inputs else None, model=self._run_map[str(run_id)]["model"])

    def on_llm_end(self, response: LLMResult, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        duration = time.time() - self._run_map.get(str(run_id), {}).get("start", time.time())
        usage = response.llm_output.get("token_usage") if response.llm_output else None
        self._log("info", "llm_end", run_id, parent_run_id, generations=[[g.text for g in gen] for gen in response.generations] if self.include_outputs else None, token_usage=usage, duration_ms=duration * 1000)
        self._run_map.pop(str(run_id), None)

    def on_llm_error(self, error: BaseException, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._log("error", "llm_error", run_id, parent_run_id, error=str(error))
        self._run_map.pop(str(run_id), None)

    # Tool events
    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._run_map[str(run_id)] = {"start": time.time(), "type": "tool", "name": serialized.get("name")}
        self._log("info", "tool_start", run_id, parent_run_id, tool_name=serialized.get("name"), input=input_str if self.include_inputs else None)

    def on_tool_end(self, output: str, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        duration = time.time() - self._run_map.get(str(run_id), {}).get("start", time.time())
        self._log("info", "tool_end", run_id, parent_run_id, output=output if self.include_outputs else None, duration_ms=duration * 1000)
        self._run_map.pop(str(run_id), None)

    def on_tool_error(self, error: BaseException, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._log("error", "tool_error", run_id, parent_run_id, error=str(error))
        self._run_map.pop(str(run_id), None)

    # Retriever events
    def on_retriever_start(self, serialized: Dict[str, Any], query: str, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self._run_map[str(run_id)] = {"start": time.time(), "type": "retriever"}
        self._log("info", "retriever_start", run_id, parent_run_id, query=query if self.include_inputs else None)

    def on_retriever_end(self, documents, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        duration = time.time() - self._run_map.get(str(run_id), {}).get("start", time.time())
        self._log("info", "retriever_end", run_id, parent_run_id, doc_count=len(documents), duration_ms=duration * 1000)
        self._run_map.pop(str(run_id), None)

This handler emits structured JSON logs with run IDs, parent run IDs, durations, and optional inputs/outputs. The run_id/parent_run_id pairing lets you reconstruct the full execution tree in your log aggregation system.

Building a structured logging handler

The handler above covers the critical paths. For production, add these concerns:

Sampling: High-volume chains generate massive log volume. Add a sampling rate parameter and drop non-error events probabilistically.

import random

class SampledLoggingHandler(StructuredLoggingHandler):
    def __init__(self, logger, sample_rate: float = 0.1, **kwargs):
        super().__init__(logger, **kwargs)
        self.sample_rate = sample_rate

    def _should_log(self, is_error: bool = False) -> bool:
        if is_error:
            return True
        return random.random() < self.sample_rate

    def _log(self, level: str, event: str, run_id: str, parent_run_id: Optional[str], **kwargs):
        if not self._should_log(level == "error"):
            return
        super()._log(level, event, run_id, parent_run_id, **kwargs)

Redaction: Never log raw prompts or outputs containing PII, secrets, or sensitive context. Add a redaction pass:

import re

SECRET_PATTERNS = [
    (re.compile(r'(api[_-]?key|secret|password|token)["\']?\s*[:=]\s*["\']?([^"\'\s]+)'), r'\1=***REDACTED***'),
    (re.compile(r'Bearer\s+[A-Za-z0-9\-._~+/]+=*'), 'Bearer ***REDACTED***'),
]

def redact(text: str) -> str:
    for pattern, replacement in SECRET_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Apply redact() to every string field before serialization.

Context enrichment: Attach request IDs, user IDs, or feature flags from contextvars so logs correlate with upstream requests.

import contextvars

request_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar("request_id", default=None)

class ContextAwareHandler(StructuredLoggingHandler):
    def _log(self, level: str, event: str, run_id: str, parent_run_id: Optional[str], **kwargs):
        request_id = request_id_var.get()
        if request_id:
            kwargs["request_id"] = request_id
        super()._log(level, event, run_id, parent_run_id, **kwargs)

Set request_id_var.set(request_id) in your API middleware before invoking the chain.

Attaching callbacks at different levels

Callbacks attach at three scopes — global, chain, and invocation. Each serves a different purpose.

Global callbacks apply to every chain execution in the process. Use for infrastructure concerns: metrics emission, audit logging, safety monitoring.

from langchain_core.callbacks import CallbackManager
from langchain_core.globals import set_global_handler

# Register once at startup
set_global_handler(StructuredLoggingHandler(logger))

Chain-level callbacks bind to a specific chain instance. Use for chain-specific enrichment: tagging, custom metadata, per-chain sampling.

from langchain.chains import RetrievalQA
from langchain_community.llms import OpenAI

llm = OpenAI(temperature=0)
chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    callbacks=[StructuredLoggingHandler(logger, include_inputs=False)]  # chain-specific config
)

Invocation-level callbacks pass at call time. Use for request-scoped context: request IDs, user context, A/B test variants.

result = chain.invoke(
    {"query": "What is the refund policy?"},
    config={"callbacks": [ContextAwareHandler(logger)]}
)

Precedence: Invocation callbacks run in addition to chain and global callbacks. All three fire for each event. Avoid duplicate logging by checking run_id in your handler or using a deduplication layer.

Handling async and streaming

Async chains (ainvoke, astream, abatch) require async callback methods. Implement the on_*_start/on_*_end async variants — LangChain calls them when running in an event loop.

class AsyncStructuredLoggingHandler(StructuredLoggingHandler):
    async def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        # Same logic, but async if you need async I/O (e.g., writing to async log sink)
        self.on_chain_start(serialized, inputs, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

    async def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
        self.on_llm_start(serialized, prompts, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

    # ... repeat for other events

For streaming, on_llm_new_token fires for each token. This is high-volume — sample aggressively or aggregate.

def on_llm_new_token(self, token: str, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
    # Aggregate in memory, flush periodically
    self._token_buffers.setdefault(str(run_id), []).append(token)

def on_llm_end(self, response: LLMResult, *, run_id: uuid.UUID, parent_run_id: Optional[uuid.UUID] = None, **kwargs):
    tokens = self._token_buffers.pop(str(run_id), [])
    full_text = "".join(tokens)
    # Log aggregated result
    super().on_llm_end(response, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

When using astream() or astream_log(), the callback receives on_chain_stream events with StreamChunk objects. Handle these if you need per-chunk visibility.

Common pitfalls and tradeoffs

Duplicate events: Global + chain + invocation callbacks fire together. A single LLM call can emit three on_llm_start events. Deduplicate by run_id in your handler or accept duplicates and filter downstream.

Memory leaks: The _run_map in the example grows unbounded if on_*_end/on_*_error never fire (e.g., cancelled tasks). Add a background reaper:

import threading
import time

class ReapingHandler(StructuredLoggingHandler):
    def __init__(self, *args, max_age_seconds: int = 300, **kwargs):
        super().__init__(*args, **kwargs)
        self.max_age_seconds = max_age_seconds
        self._reaper = threading.Thread(target=self._reap, daemon=True)
        self._reaper.start()

    def _reap(self):
        while True:
            time.sleep(60)
            now = time.time()
            stale = [rid for rid, info in self._run_map.items() if now - info["start"] > self.max_age_seconds]
            for rid in stale:
                self._log("warning", "run_abandoned", rid, None, run_info=self._run_map.pop(rid))

Performance overhead: Serializing inputs/outputs to JSON on every event adds latency. For hot paths, use a binary format (msgpack, protobuf) or write to a ring buffer consumed by a separate logging thread.

Exception swallowing: If your callback raises, it can mask the original chain error. Wrap every handler method in try/except and log handler errors separately.

def _safe_log(self, fn, *args, **kwargs):
    try:
        fn(*args, **kwargs)
    except Exception as e:
        # Log to stderr or a separate error sink — don't use self.logger to avoid recursion
        print(f"Callback handler error: {e}", file=sys.stderr)

Serialization failures: When using n4n.ai or similar gateways with automatic fallback, a single logical LLM call may emit multiple on_llm_start/on_llm_end pairs as the gateway retries across providers. Your handler should correlate these via parent_run_id or attach provider metadata from the response headers.

Testing your callback implementation

Unit test handlers in isolation by calling methods directly with constructed arguments.

import pytest
from unittest.mock import MagicMock
import uuid

def test_structured_logging_handler_emits_json():
    logger = MagicMock()
    handler = StructuredLoggingHandler(logger)
    run_id = uuid.uuid4()
    
    handler.on_chain_start({"name": "test_chain"}, {"input": "hello"}, run_id=run_id)
    
    logger.info.assert_called_once()
    call_args = logger.info.call_args[0][0]
    payload = json.loads(call_args)
    assert payload["event"] == "chain_start"
    assert payload["run_id"] == str(run_id)
    assert payload["inputs"] == {"input": "hello"}

def test_handler_captures_duration():
    logger = MagicMock()
    handler = StructuredLoggingHandler(logger)
    run_id = uuid.uuid4()
    
    handler.on_llm_start({"kwargs": {"model_name": "gpt-4"}}, ["prompt"], run_id=run_id)
    time.sleep(0.01)
    handler.on_llm_end(LLMResult(generations=[[Generation(text="response")]], llm_output={"token_usage": {"total_tokens": 10}}), run_id=run_id)
    
    call_args = logger.info.call_args_list[1][0][0]
    payload = json.loads(call_args)
    assert payload["event"] == "llm_end"
    assert payload["duration_ms"] > 5
    assert payload["token_usage"]["total_tokens"] == 10

Integration test with a real chain:

def test_callback_fires_on_real_chain(caplog):
    import logging
    import io
    
    log_stream = io.StringIO()
    logger = logging.getLogger("test")
    logger.setLevel(logging.INFO)
    logger.addHandler(logging.StreamHandler(log_stream))
    
    handler = StructuredLoggingHandler(logger)
    chain = LLMChain(llm=FakeLLM(responses=["answer"]), prompt=PromptTemplate.from_template("{q}"))
    
    chain.invoke({"q": "question"}, config={"callbacks": [handler]})
    
    logs = log_stream.getvalue().strip().split("\n")
    events = [json.loads(line)["event"] for line in logs]
    assert "chain_start" in events
    assert "llm_start" in events
    assert "llm_end" in events
    assert "chain_end" in events

Use FakeLLM from langchain_community.llms.fake for deterministic, fast tests.

Structured output for downstream consumers

If you feed logs to a tracing system (Jaeger, Zipkin, Datadog APM), emit OpenTelemetry spans instead of or alongside JSON logs. The callback events map cleanly to span lifecycle:

Callback event	Span operation
`on_chain_start`	`chain` span start
`on_llm_start`	`llm` span start (child of chain)
`on_tool_start`	`tool` span start
`on_retriever_start`	`retriever` span start
`*_end`	span end with status OK
`*_error`	span end with status ERROR

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

class OTelCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        self._spans: Dict[str, trace.Span] = {}

    def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id=None, **kwargs):
        parent_ctx = trace.set_span_in_context(self._spans.get(str(parent_run_id))) if parent_run_id else None
        span = tracer.start_span(f"chain.{serialized.get('name', 'unknown')}", context=parent_ctx)
        span.set_attribute("langchain.run_id", str(run_id))
        if inputs:
            span.set_attribute("langchain.inputs", json.dumps(inputs, default=str))
        self._spans[str(run_id)] = span

    def on_chain_end(self, outputs, *, run_id, **kwargs):
        span = self._spans.pop(str(run_id), None)
        if span:
            if outputs:
                span.set_attribute("langchain.outputs", json.dumps(outputs, default=str))
            span.end()

    def on_chain_error(self, error, *, run_id, **kwargs):
        span = self._spans.pop(str(run_id), None)
        if span:
            span.record_exception(error)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(error)))
            span.end()

This gives you distributed traces with zero instrumentation in your chain code.

Putting it together

Start with a single global handler that emits structured JSON to stdout. Add redaction, sampling, and context enrichment as pain points emerge. Attach chain-level handlers for team-specific metadata. Use invocation callbacks for request correlation. Test handlers in isolation and in integration.

The callback system is the only supported observability hook in LangChain. Treat it as first-class infrastructure — version it, test it, and monitor its health like any other production component.