LangGraph

LangGraph Cost Tracking Across a Multi-Node Graph

Pramod Dutta · Jun 20, 2026 · 12 min read

Why cost tracking breaks down the moment you add a graph

Tracking LLM spend is easy when you have one call. You get a response, you read usage.prompt_tokens and usage.completion_tokens, you multiply by the price per token, done. The problem starts the moment you move from a single call to a graph.

A LangGraph application is a state machine. A single user request might route through a planner node, a retriever node, two or three tool-calling nodes, a critic node that checks the output, and a final summarizer node. Each of those nodes can call a different model. Some nodes call the LLM twice because of a retry. Some nodes don't call the LLM at all — they just transform state. If you only look at the total cost of the run, you have no idea which node is actually expensive. You can't tell if the critic node is burning 60% of your budget on a model that's overkill for the job, or if the retriever's re-ranking step is quietly calling GPT-4 class models when a cheaper model would do.

This matters more in production than it sounds. Once you ship a LangGraph agent with loops — a ReAct-style tool loop, a reflection loop, a supervisor routing between sub-agents — the number of LLM calls per user request stops being fixed. It can be 3 calls or 30 calls depending on how many iterations the graph takes to converge. Without per-node, per-run cost visibility, you're flying blind: you can't set alerts, you can't optimize the expensive node without guessing, and you can't answer the question every AI product eventually gets asked — "what does one user session cost us?"

This article walks through a working pattern for tracking token cost per node across a multi-node LangGraph graph, using LangGraph's callback system and its state object together. We'll build it up from the naive approach to a production-shaped version with per-node breakdowns, running totals in state, and a budget guard that can halt a graph before it overspends.

The naive approach and why it falls short

The instinct most people have is to wrap the LLM call and log usage after each invoke. That works for a single node:

from langchain_openai import ChatOpenAI

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

response = llm.invoke("Summarize this document...")
usage = response.response_metadata.get("token_usage", {})
print(usage)
# {'prompt_tokens': 412, 'completion_tokens': 88, 'total_tokens': 500}

The problem is that this only tells you about the node you just wrote. In a graph with eight nodes, you'd have to sprinkle this logging logic into every node function, manually thread the running total through state, and hope nobody forgets to add it when they write node number nine. It's not wrong, it's just not scalable, and it gets worse the moment a node uses a sub-chain or an internal tool call that you didn't explicitly instrument.

What you actually want is a single point of instrumentation that sees every LLM call in the graph, regardless of which node triggered it, and tags each one with which node it came from. That's what LangGraph's callback system is for.

Instrumenting cost with a callback handler

LangChain and LangGraph both support callback handlers that hook into the LLM call lifecycle — on_llm_start, on_llm_end, on_chat_model_start, and so on. Because LangGraph nodes are just Python functions (or Runnables) that call LLMs internally, a callback attached at the graph level sees every call made anywhere in the graph, without you touching individual node code.

Here's a cost-tracking callback handler that captures token usage and cost per call, and tags it with the currently executing node using run_id and metadata:

from langchain_core.callbacks import BaseCallbackHandler
from collections import defaultdict

# Rough per-1K-token pricing table — keep this in a config file in real projects
PRICE_PER_1K = {
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
    "claude-sonnet-4-5": {"input": 0.003, "output": 0.015},
}

class NodeCostTracker(BaseCallbackHandler):
    def __init__(self):
        self.node_costs = defaultdict(lambda: {"tokens": 0, "cost": 0.0, "calls": 0})
        self.total_cost = 0.0

    def on_llm_end(self, response, **kwargs):
        node_name = kwargs.get("metadata", {}).get("langgraph_node", "unknown")
        for generation in response.generations:
            for gen in generation:
                info = gen.message.response_metadata if hasattr(gen, "message") else {}
                usage = info.get("token_usage", {}) or info.get("usage", {})
                model = info.get("model_name", "gpt-4o-mini")

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

                price = PRICE_PER_1K.get(model, PRICE_PER_1K["gpt-4o-mini"])
                cost = (
                    (prompt_tokens / 1000) * price["input"]
                    + (completion_tokens / 1000) * price["output"]
                )

                self.node_costs[node_name]["tokens"] += prompt_tokens + completion_tokens
                self.node_costs[node_name]["cost"] += cost
                self.node_costs[node_name]["calls"] += 1
                self.total_cost += cost

    def report(self):
        for node, data in self.node_costs.items():
            print(f"{node:20s} calls={data['calls']:3d}  tokens={data['tokens']:6d}  cost=${data['cost']:.5f}")
        print(f"{'TOTAL':20s} cost=${self.total_cost:.5f}")

The key detail here is kwargs.get("metadata", {}).get("langgraph_node"). When you run a graph compiled with graph.compile(), LangGraph automatically injects metadata into every callback event that includes which node the call originated from. You don't have to manually tag anything — LangGraph does the bookkeeping for you as long as your callback reads that metadata field.

Wiring the tracker into a compiled graph

Now let's build an actual multi-node graph and attach the tracker. This example is a research-assistant style graph: a planner node decides what to look up, a retriever node fetches (simulated) context, a writer node drafts an answer, and a critic node reviews it and can loop back to the writer.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

class GraphState(TypedDict):
    question: str
    plan: str
    context: str
    draft: str
    critique: str
    revision_count: int

planner_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
writer_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
critic_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def plan_node(state: GraphState) -> dict:
    resp = planner_llm.invoke([
        SystemMessage(content="Break the question into a short research plan."),
        HumanMessage(content=state["question"]),
    ])
    return {"plan": resp.content}

def retrieve_node(state: GraphState) -> dict:
    # In production this hits a vector store; simulated here
    return {"context": f"Relevant context for: {state['plan']}"}

def write_node(state: GraphState) -> dict:
    resp = writer_llm.invoke([
        SystemMessage(content="Write a clear answer using the context provided."),
        HumanMessage(content=f"Question: {state['question']}\nContext: {state['context']}"),
    ])
    return {"draft": resp.content}

def critique_node(state: GraphState) -> dict:
    resp = critic_llm.invoke([
        SystemMessage(content="Reply APPROVE if the draft is good, otherwise give one revision note."),
        HumanMessage(content=state["draft"]),
    ])
    approved = "APPROVE" in resp.content
    return {
        "critique": resp.content,
        "revision_count": state.get("revision_count", 0) + (0 if approved else 1),
    }

def should_revise(state: GraphState) -> str:
    if "APPROVE" in state["critique"] or state["revision_count"] >= 2:
        return "end"
    return "revise"

builder = StateGraph(GraphState)
builder.add_node("plan", plan_node)
builder.add_node("retrieve", retrieve_node)
builder.add_node("write", write_node)
builder.add_node("critique", critique_node)

builder.set_entry_point("plan")
builder.add_edge("plan", "retrieve")
builder.add_edge("retrieve", "write")
builder.add_edge("write", "critique")
builder.add_conditional_edges("critique", should_revise, {"revise": "write", "end": END})

graph = builder.compile()

tracker = NodeCostTracker()
result = graph.invoke(
    {"question": "What are the tradeoffs of using LangGraph vs a plain agent loop?"},
    config={"callbacks": [tracker]},
)

tracker.report()

Notice the write -> critique -> write loop. If the critic rejects the draft, the graph loops back and calls the writer LLM again. This is exactly the scenario where per-call totals matter — the writer node might get invoked once or three times depending on the run, and the cost report will reflect that automatically because every loop iteration fires its own on_llm_end event tagged with langgraph_node: "write".

Storing running cost inside graph state

Attaching a callback per invocation works well for single request-response calls, but many LangGraph deployments are long-running — a chat session, a multi-turn agent, a background job that resumes across checkpoints. For those, it's often more useful to persist the cost directly in the graph's state so it survives checkpointing and can be inspected mid-run, not just after the fact.

You can do this by adding a cost field to your TypedDict state and updating it from within a wrapper around your LLM calls, instead of (or in addition to) an external callback:

from typing import TypedDict

class GraphState(TypedDict):
    question: str
    draft: str
    cost_so_far: float
    token_log: list

def call_llm_and_track(llm, messages, node_name, state: GraphState):
    response = llm.invoke(messages)
    usage = response.response_metadata.get("token_usage", {})
    model = response.response_metadata.get("model_name", "gpt-4o-mini")

    price = PRICE_PER_1K.get(model, PRICE_PER_1K["gpt-4o-mini"])
    call_cost = (
        (usage.get("prompt_tokens", 0) / 1000) * price["input"]
        + (usage.get("completion_tokens", 0) / 1000) * price["output"]
    )

    state["cost_so_far"] = state.get("cost_so_far", 0.0) + call_cost
    state.setdefault("token_log", []).append({
        "node": node_name,
        "model": model,
        "tokens": usage.get("total_tokens", 0),
        "cost": call_cost,
    })
    return response, state

def write_node(state: GraphState) -> dict:
    response, state = call_llm_and_track(
        writer_llm,
        [HumanMessage(content=state["question"])],
        "write",
        state,
    )
    return {"draft": response.content, "cost_so_far": state["cost_so_far"], "token_log": state["token_log"]}

The advantage of storing cost in state rather than only in an external callback object is that it becomes part of whatever LangGraph is persisting via its checkpointer. If you're using MemorySaver, Postgres, or SQLite checkpointing, cost_so_far gets saved and restored automatically across turns. That means a multi-day conversation thread carries its own running total, and you can query "how much has this thread cost us so far" just by reading its checkpoint state, without replaying the whole callback history.

Building a budget guard that halts the graph

Once cost is visible per node, the natural next step is to enforce a budget instead of just observing it. This is important for anything with loops — a critic that keeps rejecting drafts, or a ReAct agent that keeps calling tools without converging, can quietly run up a large bill if nothing stops it.

You can implement this as a conditional edge that checks accumulated cost before allowing another loop iteration:

MAX_COST_PER_RUN = 0.50  # dollars

def should_revise(state: GraphState) -> str:
    if state.get("cost_so_far", 0.0) >= MAX_COST_PER_RUN:
        return "budget_exceeded"
    if "APPROVE" in state["critique"] or state["revision_count"] >= 2:
        return "end"
    return "revise"

def budget_exceeded_node(state: GraphState) -> dict:
    return {
        "draft": state["draft"] + "\n\n[Note: response finalized early due to cost limit]",
    }

builder.add_node("budget_exceeded", budget_exceeded_node)
builder.add_conditional_edges(
    "critique",
    should_revise,
    {"revise": "write", "end": END, "budget_exceeded": "budget_exceeded"},
)
builder.add_edge("budget_exceeded", END)

This pattern generalizes well beyond a single revision loop. Any conditional edge in your graph can check state["cost_so_far"] against a threshold before deciding whether to keep looping, call an expensive tool, or route to a cheaper fallback model. You can even make the threshold dynamic — pass it in as part of the initial state so different customer tiers get different budgets on the same graph definition.

Aggregating cost across a full session, not just a run

A single graph.invoke() call is one turn. A real chat product involves many turns against the same thread. If you're tracking cost only inside the state returned by a single invoke, you'll lose the aggregate the moment a new invoke starts unless you explicitly carry it forward.

The cleanest way to do this is to read the previous checkpoint's cost before starting a new turn, using the graph's own state retrieval:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

thread_config = {"configurable": {"thread_id": "user-42-session-1"}}

def run_turn(question: str):
    # Pull prior cost from the last checkpoint, if one exists
    existing_state = graph.get_state(thread_config)
    prior_cost = existing_state.values.get("cost_so_far", 0.0) if existing_state.values else 0.0

    result = graph.invoke(
        {"question": question, "cost_so_far": prior_cost},
        config=thread_config,
    )
    print(f"Session cost so far: ${result['cost_so_far']:.4f}")
    return result

run_turn("What is a LangGraph checkpointer?")
run_turn("How does that differ from a memory store?")

Because cost_so_far is threaded through as part of state and the checkpointer persists it, each subsequent call to run_turn for the same thread_id picks up where the last one left off. Multiply this pattern across users and you have the raw data for per-user cost dashboards without needing an external observability platform — though for production scale, exporting the token_log entries to a proper store (Postgres table, a metrics pipeline, or an observability tool like LangSmith) is worth doing rather than keeping everything inside LangGraph state indefinitely.

Common pitfalls when tracking cost in a graph

A few mistakes show up repeatedly when teams add cost tracking to an existing LangGraph app:

Assuming one LLM call per node. Nodes that use tool-calling loops, sub-chains, or retries internally can fire multiple LLM calls per node execution. Your tracker needs to sum, not overwrite, per node.
Forgetting streaming responses. If you're using .stream() instead of .invoke(), token usage sometimes only appears on the final chunk. Check for usage metadata on the last streamed event rather than assuming every chunk carries it.
Hardcoding one price table for all providers. Anthropic, OpenAI, and any self-hosted model all report usage fields slightly differently (input_tokens/output_tokens vs prompt_tokens/completion_tokens). Normalize this in one place rather than in every node.
Not accounting for cached tokens. Several providers now offer discounted pricing for cached prompt prefixes. If your traffic reuses long system prompts, a flat per-token price will overstate real cost — check for cache-read token fields in the usage payload.
Losing cost data on graph errors. If a node throws an exception mid-run, make sure your cost log is flushed or persisted before the exception propagates, otherwise you lose visibility into exactly the expensive runs you most want to see.
Conflating latency debugging with cost debugging. They're related but distinct problems — a slow node isn't necessarily an expensive node, and vice versa. Keep separate metrics rather than trying to infer one from the other.

Turning this into a dashboard-ready log

For anything beyond a prototype, you'll want the token_log entries flowing somewhere queryable. A simple approach is to emit a structured log line per LLM call that a log aggregator or a lightweight script can parse later:

import json
import time

def log_call(node_name, model, usage, cost, thread_id):
    entry = {
        "timestamp": time.time(),
        "thread_id": thread_id,
        "node": node_name,
        "model": model,
        "prompt_tokens": usage.get("prompt_tokens", 0),
        "completion_tokens": usage.get("completion_tokens", 0),
        "cost_usd": round(cost, 6),
    }
    print(json.dumps(entry))  # replace with your logging/metrics sink

Call this from inside on_llm_end in the callback handler, or from call_llm_and_track, and you get a per-call, per-node, per-thread audit trail that's trivial to load into a spreadsheet, a Postgres table, or a metrics dashboard. Once you have that data for even a week of production traffic, the expensive nodes and wasteful loops tend to become obvious almost immediately — usually it's a single node calling too large a model for a task a smaller one would handle fine, or a revision loop that rarely needs more than one pass but is configured to allow three.

Wrapping up

Cost tracking in a single-call LLM app is an afterthought. Cost tracking in a multi-node LangGraph application has to be designed in, because the number and identity of LLM calls per run is dynamic — driven by conditional edges, loops, and retries that only reveal themselves at runtime. The pattern that works is layered: a callback handler that taps into LangGraph's automatic node-tagging metadata for real-time visibility, a cost_so_far field inside your typed state so totals survive checkpointing across turns, and a conditional edge that treats budget the same way you'd treat any other stopping condition in the graph. Once those three pieces are in place, you get per-node cost breakdowns, session-level totals across a checkpointed thread, and the ability to hard-stop a runaway loop before it becomes a surprise on the bill.

If you want to go deeper into building graphs with loops, conditional routing, checkpointers, and production patterns like this one, our LangGraph Tutorial course on teachyou.ai walks through the full framework from first graph to a deployed multi-agent system, with cost, latency, and reliability treated as first-class concerns rather than an afterthought.