LangChain

LangChain Cost Tracking: Monitoring Token Usage Across Chains

Pramod Dutta · Jul 1, 2026 · 15 min read

Why your LangChain bill is a mystery until it isn't

You ship a RAG chatbot. It works in the demo. Three weeks later finance forwards you an OpenAI invoice that's four times what you budgeted, and nobody can say why. Was it the retriever pulling in too many documents? A user who ran the same query fifty times? A prompt template that quietly grew a two-thousand-token system message? Without token-level visibility, you're debugging a cost problem with a bank statement instead of a profiler.

This is the single most common production gap in LangChain applications: teams instrument logging, they instrument latency, they even instrument error rates, but token usage — the thing that directly maps to dollars — gets left out until the invoice forces the conversation. The good news is that LangChain already tracks this data internally for every LLM call. It's just a matter of tapping into it correctly, at the right layer, for chains that may call the model once or fifty times per request.

This article walks through the practical mechanics: the built-in get_openai_callback() context manager, why it silently fails for non-OpenAI models, how to build a custom callback handler that works across providers and multi-step chains, and how to turn raw token counts into per-user, per-feature cost attribution you can actually put in a dashboard.

How LangChain tracks tokens under the hood

Every LLM call in LangChain — whether it's a single invoke(), a chain with five sequential steps, or an agent looping through tool calls — passes through LangChain's callback system. Callbacks are hooks that fire at specific lifecycle events: on_llm_start, on_llm_end, on_chain_start, on_tool_end, and so on. The token usage data lives inside on_llm_end, specifically in the LLMResult object's llm_output field.

For OpenAI models, that field looks something like this after a call:

{
    "token_usage": {
        "prompt_tokens": 812,
        "completion_tokens": 143,
        "total_tokens": 955
    },
    "model_name": "gpt-4o-mini"
}

The critical detail is that this is populated per LLM call, not per chain. If your chain calls the model three times (say, a query rewrite step, a retrieval-grading step, and a final answer step), you get three separate on_llm_end events, each with its own token counts. Cost tracking in LangChain is fundamentally about aggregating these events correctly — not double-counting, not missing intermediate calls, and attaching the right metadata so you know which user or feature generated the spend.

This matters more than it sounds. A naive approach — logging tokens only from the final chain output — will systematically undercount every multi-step chain, agent, or chain-of-chains setup, which is most production LangChain code.

The quick way: get_openai_callback()

For simple, synchronous OpenAI-only workflows, LangChain ships a context manager that does the aggregation for you: get_openai_callback(). It wraps a block of code, listens to every on_llm_end event fired inside that block, and sums up tokens and estimated cost.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.callbacks import get_openai_callback

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_template(
    "Summarize the following support ticket in two sentences:\n\n{ticket}"
)

chain = prompt | llm

with get_openai_callback() as cb:
    result = chain.invoke({"ticket": "Customer reports login failures after password reset..."})
    print(result.content)
    print(f"Prompt tokens:     {cb.prompt_tokens}")
    print(f"Completion tokens: {cb.completion_tokens}")
    print(f"Total tokens:      {cb.total_tokens}")
    print(f"Total cost (USD):  ${cb.total_cost:.6f}")

Running this against a chain with a single LLM call gives you exactly what you'd expect: one set of numbers. The interesting part is what happens when the chain does more than one call. If you put a retrieval-augmented chain — one that rewrites the query, retrieves documents, and then generates an answer with two separate model calls — inside the same with block, cb.total_tokens accumulates across both calls automatically:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

rewrite_prompt = ChatPromptTemplate.from_template(
    "Rewrite this question to be a clean, standalone search query: {question}"
)
answer_prompt = ChatPromptTemplate.from_template(
    "Using this context, answer the question.\n\nContext: {context}\n\nQuestion: {question}"
)

rewrite_chain = rewrite_prompt | llm | StrOutputParser()

def fake_retrieve(query: str) -> str:
    return "Password resets require email verification within 15 minutes."

full_chain = (
    {"question": RunnablePassthrough()}
    | RunnablePassthrough.assign(rewritten=lambda x: rewrite_chain.invoke({"question": x["question"]}))
    | RunnablePassthrough.assign(context=lambda x: fake_retrieve(x["rewritten"]))
    | answer_prompt
    | llm
    | StrOutputParser()
)

with get_openai_callback() as cb:
    answer = full_chain.invoke("Why can't I log in?")
    print(answer)
    print(f"Total LLM calls counted in total_tokens: {cb.successful_requests}")
    print(f"Total tokens across BOTH calls: {cb.total_tokens}")
    print(f"Total cost across BOTH calls: ${cb.total_cost:.6f}")

The successful_requests attribute is worth calling out specifically — it tells you how many distinct LLM invocations happened inside the block, which is the fastest way to sanity-check whether your chain is calling the model more times than you think it should. If you expected two calls and successful_requests says four, that's often the first sign a chain is retrying silently or an agent is looping more than intended.

Where get_openai_callback() breaks down

This context manager has three real limitations that show up fast once you leave toy examples.

It only understands OpenAI's pricing table. The cost calculation is hardcoded to a lookup table of OpenAI model prices baked into the LangChain community package. If you're using Anthropic's Claude models, open-weight models through a provider like Together or Groq, or a local model through Ollama, cb.total_cost will silently report 0.0 even though cb.total_tokens might still populate correctly (or not, depending on whether that provider's integration emits the same token_usage shape).
It doesn't survive across async boundaries or background tasks cleanly. If your chain kicks off an async subtask that isn't awaited inside the with block, or if you're using .batch() with high concurrency, callback context can behave inconsistently depending on your LangChain version. It's a convenience wrapper, not a robust production instrument.
It has no concept of "who." It gives you a total for the block of code, but nothing that lets you say "this was user 4821's request" or "this was the summarization feature, not the chat feature." For any real SaaS product, cost without attribution is not actionable — you need to know which customer, tenant, or feature is driving spend.

For a personal script or a one-off cost estimate, get_openai_callback() is genuinely useful and you should reach for it first. For anything running in production behind real users, you need a custom callback handler.

There's a fourth, quieter limitation worth flagging: get_openai_callback() aggregates everything inside the with block into one flat total. If you're evaluating five different prompt variants in a loop to see which is cheapest, or running a batch job across a hundred documents, you don't want one number for the whole run — you want per-iteration numbers so you can compare them. The context manager doesn't reset itself between iterations unless you re-enter it, which means the natural pattern is to open and close it once per unit of work you actually want to measure:

results = []
documents = ["doc one text...", "doc two text...", "doc three text..."]

for doc in documents:
    with get_openai_callback() as cb:
        summary = chain.invoke({"ticket": doc})
        results.append({
            "summary": summary.content,
            "tokens": cb.total_tokens,
            "cost": cb.total_cost,
        })

for r in results:
    print(f"{r['tokens']} tokens, ${r['cost']:.6f}")

This works, but notice how quickly it turns into boilerplate the moment you need anything beyond a flat script — per-document tracking, retries, or parallel execution with .batch() all push you toward a handler you control directly rather than a context manager you have to re-enter carefully.

Building a custom callback handler with real cost attribution

The pattern that scales is to write your own class that inherits from BaseCallbackHandler, override on_llm_end, and push structured records somewhere durable — a database table, a logging pipeline, or an in-memory aggregator for the lifetime of a single request. This works identically for OpenAI, Anthropic, and most other providers because it reads directly from the LLMResult, not from a provider-specific price table.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from datetime import datetime, timezone
import uuid


class TokenUsageTracker(BaseCallbackHandler):
    """Collects token usage per LLM call and attaches request-level metadata."""

    def __init__(self, user_id: str, feature: str):
        self.user_id = user_id
        self.feature = feature
        self.records = []

    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        for generation_batch in response.generations:
            for generation in generation_batch:
                info = getattr(generation, "generation_info", None) or {}
                usage = None

                # OpenAI / most chat models: token_usage on llm_output
                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", "unknown")

                # Anthropic: usage lives on the message metadata
                elif response.llm_output and "usage" in response.llm_output:
                    raw = response.llm_output["usage"]
                    usage = {
                        "prompt_tokens": raw.get("input_tokens", 0),
                        "completion_tokens": raw.get("output_tokens", 0),
                        "total_tokens": raw.get("input_tokens", 0) + raw.get("output_tokens", 0),
                    }
                    model_name = response.llm_output.get("model", "unknown")
                else:
                    continue

                self.records.append({
                    "id": str(uuid.uuid4()),
                    "timestamp": datetime.now(timezone.utc).isoformat(),
                    "user_id": self.user_id,
                    "feature": self.feature,
                    "model": model_name,
                    "prompt_tokens": usage.get("prompt_tokens", 0),
                    "completion_tokens": usage.get("completion_tokens", 0),
                    "total_tokens": usage.get("total_tokens", 0),
                })

    def summary(self) -> dict:
        total = sum(r["total_tokens"] for r in self.records)
        prompt = sum(r["prompt_tokens"] for r in self.records)
        completion = sum(r["completion_tokens"] for r in self.records)
        return {
            "user_id": self.user_id,
            "feature": self.feature,
            "llm_calls": len(self.records),
            "prompt_tokens": prompt,
            "completion_tokens": completion,
            "total_tokens": total,
        }

Usage looks like this — you attach the handler per request, not globally, so each user's request gets its own isolated tracker:

tracker = TokenUsageTracker(user_id="user_4821", feature="ticket_summarizer")

result = chain.invoke(
    {"ticket": "Customer reports login failures..."},
    config={"callbacks": [tracker]},
)

print(tracker.summary())
# {'user_id': 'user_4821', 'feature': 'ticket_summarizer', 'llm_calls': 1,
#  'prompt_tokens': 812, 'completion_tokens': 143, 'total_tokens': 955}

This is provider-agnostic by design — the branch logic reads whatever shape of usage data the model integration exposes rather than assuming OpenAI's schema is the only one. When you add a new provider, you extend the elif chain once, and every chain in your codebase that uses TokenUsageTracker picks up the change automatically.

Keep cost calculation out of the callback handler itself. The handler's job is to record raw token counts accurately; pricing changes constantly and shouldn't require touching your instrumentation code. Instead, maintain a small pricing table you can update independently and apply it after the fact.

PRICING_PER_MILLION_TOKENS = {
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "claude-3-5-sonnet-20241022": {"input": 3.00, "output": 15.00},
    "claude-3-5-haiku-20241022": {"input": 0.80, "output": 4.00},
}


def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    rates = PRICING_PER_MILLION_TOKENS.get(model)
    if not rates:
        return 0.0
    input_cost = (prompt_tokens / 1_000_000) * rates["input"]
    output_cost = (completion_tokens / 1_000_000) * rates["output"]
    return round(input_cost + output_cost, 8)


def enrich_with_cost(records: list[dict]) -> list[dict]:
    for r in records:
        r["cost_usd"] = calculate_cost(
            r["model"], r["prompt_tokens"], r["completion_tokens"]
        )
    return records

Run this over tracker.records after the chain finishes, and you get a per-call cost breakdown you can insert into Postgres, ship to a metrics backend, or aggregate in memory for a request-level total. Because pricing lives in one dictionary instead of scattered across your codebase, updating it when a provider changes prices is a one-line diff, not a hunt through instrumentation code.

Tracking cost across streaming responses

Streaming complicates token counting because on_llm_end still fires exactly once per call — the aggregation problem doesn't change — but you often want incremental visibility while tokens are still arriving, particularly for a "typing" cost meter in a UI. The trick is to separate two concerns: use on_llm_new_token purely for UI feedback (it does not reliably carry usage metadata across all providers) and rely on on_llm_end as the single source of truth for the actual count.

class StreamingCostTracker(BaseCallbackHandler):
    def __init__(self):
        self.token_chunks_seen = 0
        self.final_usage = None

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        # Cheap running counter for UI purposes only — not authoritative.
        self.token_chunks_seen += 1

    def on_llm_end(self, response: LLMResult, **kwargs) -> None:
        if response.llm_output and "token_usage" in response.llm_output:
            self.final_usage = response.llm_output["token_usage"]

    def get_authoritative_usage(self) -> dict:
        return self.final_usage or {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

stream_tracker = StreamingCostTracker()

for chunk in chain.stream(
    {"ticket": "App crashes on startup after latest update"},
    config={"callbacks": [stream_tracker]},
):
    print(chunk.content, end="", flush=True)

print("\n---")
print("Authoritative usage:", stream_tracker.get_authoritative_usage())

Never bill or log cost off token_chunks_seen — text chunks in a stream don't have a fixed correspondence to tokens (a chunk can be a partial token, a whole word, or several tokens depending on the provider's streaming granularity). Only on_llm_end's llm_output is safe to treat as ground truth.

Aggregating across an entire agent run

Agents are where cost tracking earns its keep, because a single user request can trigger anywhere from two to twenty LLM calls depending on how many tool-use loops the agent needs before it reaches a final answer. The custom handler pattern from earlier works unchanged — you attach it once at the top level, and it captures every nested LLM call the agent makes, including the reasoning steps that never make it into the final response.

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool


@tool
def lookup_order_status(order_id: str) -> str:
    """Look up the shipping status of an order by ID."""
    return f"Order {order_id} shipped on 2026-06-28, arriving 2026-07-05."


agent = create_tool_calling_agent(llm, [lookup_order_status], prompt)
executor = AgentExecutor(agent=agent, tools=[lookup_order_status], verbose=False)

agent_tracker = TokenUsageTracker(user_id="user_9012", feature="order_support_agent")

response = executor.invoke(
    {"input": "Where's my order 55231?"},
    config={"callbacks": [agent_tracker]},
)

summary = agent_tracker.summary()
print(f"Agent used {summary['llm_calls']} LLM calls, {summary['total_tokens']} total tokens")

For a tool-calling agent, llm_calls will typically be at least 2 — one call where the model decides to invoke lookup_order_status, and a second call where it formats the tool's result into a final answer. If your agent has a multi-step plan-then-execute pattern, that number climbs further. This is exactly the visibility that a flat, chain-level cost check misses, and it's usually where the "why is this feature so expensive" question gets answered — often the surprise isn't the final answer generation, it's an agent that re-plans three or four times before committing to a tool call.

Turning raw records into a cost dashboard

Once TokenUsageTracker records are flowing out of every request, the last step is persistence and aggregation. A simple approach that scales to real traffic is to write each record to a table and roll it up with scheduled queries rather than trying to compute running totals in application memory.

import sqlite3

def init_usage_table(db_path: str = "llm_usage.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS llm_usage (
            id TEXT PRIMARY KEY,
            timestamp TEXT,
            user_id TEXT,
            feature TEXT,
            model TEXT,
            prompt_tokens INTEGER,
            completion_tokens INTEGER,
            total_tokens INTEGER,
            cost_usd REAL
        )
    """)
    conn.commit()
    return conn


def persist_records(conn, records: list[dict]):
    conn.executemany(
        """INSERT INTO llm_usage
           (id, timestamp, user_id, feature, model, prompt_tokens, completion_tokens, total_tokens, cost_usd)
           VALUES (:id, :timestamp, :user_id, :feature, :model, :prompt_tokens, :completion_tokens, :total_tokens, :cost_usd)""",
        records,
    )
    conn.commit()


# After a request finishes:
conn = init_usage_table()
enriched = enrich_with_cost(tracker.records)
persist_records(conn, enriched)

From there, a query grouped by feature and a date range tells you exactly which part of your product is driving spend, and a query grouped by user_id flags the accounts making unusually expensive requests — useful both for cost control and for spotting abuse of a free tier. Swap SQLite for Postgres or your existing warehouse in production; the shape of the table doesn't need to change.

Common mistakes, and setting budget alerts instead of just dashboards

Attaching the callback at the LLM level instead of the invocation level. If you bind a callback to the ChatOpenAI constructor itself (ChatOpenAI(callbacks=[tracker])), it fires for every invocation of that shared instance across all users, mixing everyone's usage into one tracker. Pass callbacks through the config argument at invoke() time instead, scoped to a single request.
Assuming `on_llm_end` fires once per chain. It fires once per LLM call. A five-step chain with three LLM calls fires it three times. Sum, don't overwrite.
Forgetting cached responses still need accounting. If you add a caching layer (semantic cache or exact-match cache) in front of the LLM, cached hits won't trigger on_llm_end at all — which is correct for cost tracking, since no tokens were spent, but it means your "requests served" and "LLM calls billed" numbers will diverge, and that gap is worth surfacing in your dashboard rather than treating as a bug.
Not tagging metadata early enough. If user_id and feature aren't available at the point you construct the callback, retrofitting attribution later means parsing logs. Thread the identifiers through from the API layer at request start.
Ignoring retries at the transport layer. If you've wrapped your LLM calls with a retry decorator (for rate limits or transient errors), a single logical request might fire on_llm_end two or three times before it succeeds — once per failed attempt that still returned partial usage data, plus once for the final success. Decide up front whether failed attempts should count toward cost (they often do, since the tokens were still processed by the provider) and make that explicit in your aggregation rather than silently summing everything.

A dashboard you have to remember to check is only half the job. Once records are landing in a table, it's worth adding a cheap threshold check that runs alongside your normal request flow rather than as a separate nightly job — catching a cost spike an hour after it starts is far more useful than catching it the next morning.

def check_user_budget(conn, user_id: str, daily_limit_usd: float = 5.00) -> bool:
    cursor = conn.execute(
        """SELECT SUM(cost_usd) FROM llm_usage
           WHERE user_id = ? AND date(timestamp) = date('now')""",
        (user_id,),
    )
    spent = cursor.fetchone()[0] or 0.0
    if spent >= daily_limit_usd:
        return False  # block or throttle further requests
    return True


if not check_user_budget(conn, "user_4821"):
    raise RuntimeError("Daily LLM budget exceeded for this user")

This is deliberately simple — a single SQL aggregation, no external service — because the value comes from wiring it into the request path early, not from sophistication. Once this exists, you can layer in per-feature limits, per-tenant limits for a B2B product, or a global kill-switch for a runaway agent loop, all using the same llm_usage table you're already populating from the callback handler.

Wrapping up

Token usage in LangChain is fully observable — it's not a black box, it's a callback event you can hook into cheaply. get_openai_callback() is fine for quick scripts and OpenAI-only prototypes, but any real product needs a custom BaseCallbackHandler that captures raw usage per call, keeps pricing logic separate and updatable, and tags every record with the user and feature that caused it. Do that once, at the chain and agent level, and the "why is this bill so high" conversation turns into a five-minute SQL query instead of a guessing game.

If you want to go deeper into building production-grade chains, agents, and the observability layer around them — including cost dashboards, tracing, and evaluation pipelines — that's exactly what we cover hands-on in the LangChain Tutorial 2026 course on teachyou.ai.