iwantcoding.com
🔥 Daily 👥 Rooms 🏆 Top Log in Sign up

Callbacks

By · Reviewed & last updated · Part of LangChain Tutorial

Callbacks are how LangChain exposes the inner life of a chain: which LLM ran, with what prompt, how many tokens, what tools fired, what errors. They are the right hook for logging, latency tracking, prompt capture, and cost accounting. Implement BaseCallbackHandler, attach it per-call or globally.

Custom callback for token cost + latency tracking

EXAMPLE
from time import perf_counter
from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class CostTracker(BaseCallbackHandler):
    # Rough per-model prices in USD / 1K tokens (June 2026 — verify before billing!)
    PRICES = {
        'gpt-4o-mini': (0.00015, 0.0006),
        'gpt-4o':       (0.005,   0.015),
    }

    def __init__(self):
        self.total_in = 0
        self.total_out = 0
        self.total_cost = 0.0
        self._start = {}

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

    def on_llm_end(self, response, *, run_id, **kwargs):
        elapsed = perf_counter() - self._start.pop(run_id, perf_counter())
        usage = (response.llm_output or {}).get('token_usage', {})
        model = (response.llm_output or {}).get('model_name', '')
        in_tok  = usage.get('prompt_tokens', 0)
        out_tok = usage.get('completion_tokens', 0)
        p_in, p_out = self.PRICES.get(model, (0, 0))
        cost = (in_tok * p_in + out_tok * p_out) / 1000
        self.total_in  += in_tok
        self.total_out += out_tok
        self.total_cost += cost
        print(f'[llm] model={model} in={in_tok} out={out_tok} '
              f'cost=${cost:.5f} {elapsed*1000:.0f}ms')

    def on_tool_start(self, serialized, input_str, **kwargs):
        print(f'[tool] {serialized.get("name")} input={input_str[:60]}')

    def on_chain_error(self, error, **kwargs):
        print(f'[chain] ERROR {type(error).__name__}: {error}')

tracker = CostTracker()
llm = ChatOpenAI(model='gpt-4o-mini', temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ('system', 'Answer in one sentence.'),
    ('human', '{q}'),
])
chain = prompt | llm

for q in ['What is Lambda?', 'Define DNS.', 'Two-sentence intro to MoE.']:
    chain.invoke({'q': q}, config={'callbacks': [tracker]})

print(f'\nTotal: in={tracker.total_in} out={tracker.total_out} '
      f'cost=${tracker.total_cost:.4f}')

Why it matters

Attach callbacks via config={"callbacks": [...]} per-call when the handler holds request-scoped state (cost for THIS user, trace ID for THIS request). Use a global handler only for cross-cutting concerns like Prometheus metrics, where mixing requests in one handler instance is the desired behaviour.

Tip: Tweak the snippet with Try it Yourself », then sit the quiz at the bottom of the page.

Example

Example
from langchain_core.callbacks import BaseCallbackHandler
class Trace(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kw): print('→', prompts[0][:60])
    def on_llm_end(self, response, **kw): print('done')
chain.invoke({'q': 'hi'}, config={'callbacks': [Trace()]})
Try it Yourself »

Discussion

Loading…