Observability with callbacks and LangSmith

Callbacks, tracing every step, token and cost accounting, and reading a trace to find the step that actually failed.

Callbacks

from langchain_core.callbacks import BaseCallbackHandler
import time

class TimerCallback(BaseCallbackHandler):
    def __init__(self):
        self.starts = {}
        self.rows = []

    def on_llm_start(self, serialized, prompts, *, run_id, **kwargs):
        self.starts[run_id] = time.perf_counter()

    def on_llm_end(self, response, *, run_id, **kwargs):
        elapsed = time.perf_counter() - self.starts.pop(run_id, time.perf_counter())
        usage = getattr(response, "llm_output", {}) or {}
        tokens = usage.get("token_usage", {})
        self.rows.append({
            "ms": round(elapsed * 1000, 1),
            "prompt_tokens": tokens.get("prompt_tokens"),
            "completion_tokens": tokens.get("completion_tokens"),
        })

    def on_llm_error(self, error, *, run_id, **kwargs):
        print("llm error:", type(error).__name__, str(error)[:120])

    def on_chain_error(self, error, *, run_id, **kwargs):
        print("chain error:", type(error).__name__, str(error)[:120])

timer = TimerCallback()
result = chain.invoke({"topic": "caching"}, config={"callbacks": [timer]})
print(timer.rows)

# callbacks also work on a batch and on every nested step
other = TimerCallback()
chain.batch([{"topic": "a"}, {"topic": "b"}], config={"callbacks": [other]})
print(sum(r["ms"] for r in other.rows))

• Callbacks receive a run_id for every step. Matching a start and an end by that id is what gives you per-step latency rather than a total.
• Set a global handler with the LANGCHAIN_CALLBACKS_BACKGROUND environment variable for high-volume logging, or keep it synchronous when you need ordering.
• Handle on_llm_error and on_chain_error. A silent exception in a callback is nearly impossible to find later.
• Token counts come from the provider and may be missing or approximate. Reconcile them against your billing data monthly rather than trusting them absolutely.

Tracing with LangSmith

# enable tracing with two environment variables
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=lsv2_...
export LANGCHAIN_PROJECT=support-assistant

# or set it in code before importing anything that reads the environment
python -c "import os; os.environ['LANGCHAIN_TRACING_V2']='true'; import langchain"

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "support-assistant"

from langsmith import traceable, Client

@traceable(run_type="tool", name="lookup_order")
def lookup_order(order_id: str) -> dict:
    return {"order_id": order_id, "status": "in_transit", "eta": "2026-09-21"}

@traceable(name="support_reply")
def support_reply(question: str, user_id: str) -> str:
    order = lookup_order("A-1042") if "A-" in question else None
    context = f"Order status: {order}" if order else ""
    return chain.invoke({"topic": question, "context": context},
                        config={"metadata": {"user_id": user_id,
                                             "release": "2026-09-18"}}).content

print(support_reply("Where is order A-1042?", "u-88"))

client = Client()
runs = list(client.list_runs(project_name="support-assistant", limit=5))
for run in runs:
    print(run.name, run.run_type, round(run.total_tokens or 0), run.error)

Reading a trace to find the failure

• Start at the root and follow the longest branch. Latency usually concentrates in one step, and it is rarely the one you assumed.
• Compare the same query across two versions. Diffing two traces side by side finds a changed prompt or a changed retrieval result faster than reading logs.
• A trace that shows a retriever returning zero documents before the model answers is a retrieval bug, not a generation bug. The nesting order tells you which component to fix.
• Attach metadata at the entry point: user id, release, feature flag. Without it you cannot group traces by the version that produced the bad behaviour.
• Log the final structured output separately from the raw model text. When parsing fails you need to see both.

# cheap local observability that does not require a hosted service
import json
import logging
import time
from contextlib import contextmanager

logger = logging.getLogger("chain")
logging.basicConfig(level=logging.INFO, format="%(message)s")

@contextmanager
def step(name, **fields):
    start = time.perf_counter()
    record = {"step": name, **fields}
    try:
        yield record
        record["ok"] = True
    except Exception as exc:
        record["ok"] = False
        record["error"] = f"{type(exc).__name__}: {str(exc)[:200]}"
        raise
    finally:
        record["ms"] = round((time.perf_counter() - start) * 1000, 1)
        logger.info(json.dumps(record, default=str))

with step("retrieve", k=6) as r:
    docs = retriever.invoke("refund policy")
    r["returned"] = len(docs)

with step("generate", model="gpt-4o-mini") as r:
    answer = chain.invoke({"topic": "refund policy"}).content
    r["chars"] = len(answer)

Structured logging around each step gets you most of the value of a tracing platform in about twenty lines. Adopt a hosted tracer when you need cross-service traces, team-wide search or evaluation datasets — not merely to see step timings.

FAQ

Related

Runnables and LangChain Expression Language in depth Evaluating chains and RAG

Last refreshed 2026-09-18.