Tracing token usage across CrewAI agent chains

Tracing token usage in CrewAI is not optional when you run multi-agent pipelines in production—cost overruns hide in repeated planner calls, tool reflections, and silent retries. This guide shows how to attribute every token to a specific agent and task using LangChain callbacks and a unified OpenAI-compatible endpoint, so you can see exactly which part of the chain burns your budget.

Step 1: Install dependencies and configure a metered endpoint

CrewAI delegates LLM calls to a LangChain-compatible chat model. The framework itself does not emit token accounting, so we add it at the LLM layer. Install the packages:

pip install crewai langchain-openai

Point the chat model at an OpenAI-compatible gateway. If you use a gateway such as n4n.ai, you get per-token usage metering on the server side and automatic fallback when a provider is rate-limited, which complements the client-side tracing shown below. Set the base URL and key from environment variables:

import os

OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.n4n.ai/v1")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "sk-...")

We use ChatOpenAI from langchain_openai because it exposes clean usage metadata through callbacks and is accepted anywhere CrewAI expects an llm argument.

Step 2: Implement a token-tracking callback

LangChain invokes on_llm_end after each completion, passing an LLMResult that contains llm_output.token_usage. Subclass BaseCallbackHandler and accumulate deltas in a shared dict keyed by agent name. Use a lock if you run crews concurrently.

import threading
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import LLMResult

class AgentTokenTracker(BaseCallbackHandler):
    def __init__(self, agent_name: str, store: dict, lock: threading.Lock):
        self.agent_name = agent_name
        self.store = store
        self.lock = lock
        with self.lock:
            self.store.setdefault(agent_name, {"prompt": 0, "completion": 0, "total": 0})

    def on_llm_end(self, response: LLMResult, **kwargs):
        usage = response.llm_output.get("token_usage", None)
        if not usage:
            return
        with self.lock:
            entry = self.store[self.agent_name]
            entry["prompt"] += usage.get("prompt_tokens", 0)
            entry["completion"] += usage.get("completion_tokens", 0)
            entry["total"] += usage.get("total_tokens", 0)

The store is a plain dict you own. Because CrewAI may instantiate multiple agents, we bind one tracker per agent name.

Step 3: Build per-agent LLM factories

Do not share a single LLM object across agents if you want attribution. Create a factory that attaches a tracker to each agent name and returns a configured ChatOpenAI.

from langchain_openai import ChatOpenAI

def make_llm(agent_name: str, store: dict, lock: threading.Lock, model: str = "gpt-4o-mini"):
    return ChatOpenAI(
        model=model,
        temperature=0.2,
        base_url=OPENAI_BASE_URL,
        api_key=OPENAI_API_KEY,
        callbacks=[AgentTokenTracker(agent_name, store, lock)],
    )

If a task uses tools that internally call an LLM (e.g., a RAG retriever or a code interpreter), pass the same make_llm pattern to the tool’s underlying chain so those tokens are captured under the calling agent’s name.

Step 4: Define agents and tasks in CrewAI

Construct a minimal crew with a planner and an executor. Each gets its own LLM from the factory.

from crewai import Agent, Task, Crew, Process

store = {}
lock = threading.Lock()

planner = Agent(
    role="Planner",
    goal="Break the request into steps",
    backstory="You decompose tasks.",
    llm=make_llm("planner", store, lock),
    verbose=False,
)

executor = Agent(
    role="Executor",
    goal="Execute the steps using tools",
    backstory="You run code and report.",
    llm=make_llm("executor", store, lock),
    verbose=False,
)

task = Task(
    description="Write a Python function that sorts a list of dicts by key.",
    expected_output="A working code snippet.",
    agent=executor,
)

crew = Crew(
    agents=[planner, executor],
    tasks=[task],
    process=Process.sequential,
)

For hierarchical processes, the manager agent also needs its own tracked LLM. Pass manager_llm=make_llm("manager", store, lock) to Crew. Without this, manager tokens are invisible.

Step 5: Run the crew and emit a report

Kick off the crew and print the aggregated store.

result = crew.kickoff()

print("=== Token Usage by Agent ===")
for agent, usage in sorted(store.items()):
    print(f"{agent:10} prompt={usage['prompt']:6} completion={usage['completion']:6} total={usage['total']:6}")

print("\nTotal tokens:", sum(v['total'] for v in store.values()))

This prints per-agent prompt, completion, and total tokens. The totals should match what the gateway meters if you use a unified endpoint.

Handling streaming and tool calls

If you enable streaming=True on ChatOpenAI, on_llm_end still fires with usage, but some providers only return usage on the final chunk. Test with your model. For tool-calling agents, CrewAI may emit multiple LLM calls per task (reason, call tool, summarize). The callback captures each, so the agent total is accurate.

Verifying success

Run the script and confirm three things:

1. The store dict contains an entry for each agent name you passed to make_llm.
2. The total per agent is greater than zero after a non-trivial task.
3. The summed total equals the usage.total_tokens reported by your OpenAI-compatible gateway’s response logs. For n4n.ai, the per-token usage metering on the endpoint provides a server-side cross-check.

If an agent shows zero tokens, its LLM instance was not used—check that you assigned llm= correctly and that the agent actually ran (set verbose=True to inspect).

Step 6: Export usage as JSON for monitoring

In production you rarely print to stdout. Push the store to your metrics pipeline. A minimal JSON dump:

import json

def export_usage(store: dict, path: str = "token_usage.json"):
    with open(path, "w") as f:
        json.dump(store, f, indent=2)
    return path

export_usage(store)

Wire this into a CI job or a cron that runs a representative crew daily. Drift in token counts signals prompt regressions or agent loops.

Advanced: tagging by task

Agent-level attribution is often enough, but you may need per-task breakdown. Wrap the tracker to read kwargs['metadata'] if CrewAI passes it, or manually push a stack context before crew.kickoff():

import contextvars
current_task = contextvars.ContextVar("current_task", default="none")

class TaskAwareTracker(AgentTokenTracker):
    def on_llm_end(self, response, **kwargs):
        task = current_task.get()
        key = f"{self.agent_name}:{task}"
        with self.lock:
            self.store.setdefault(key, {"prompt":0,"completion":0,"total":0})
            entry = self.store[key]
            usage = response.llm_output.get("token_usage", {})
            entry["prompt"] += usage.get("prompt_tokens", 0)
            entry["completion"] += usage.get("completion_tokens", 0)
            entry["total"] += usage.get("total_tokens", 0)

Set current_task.set("task_1") in your orchestration code before each crew step. This isolates the planner’s first-pass cost from its revision cost.

Common pitfalls

• Shared LLM instances: Passing the same ChatOpenAI object to two agents merges their tokens under one tracker. Always factory-per-agent.
• LiteLLM passthrough: CrewAI’s native LLM class uses LiteLLM and does not surface llm_output the same way. The LangChain wrapper is the reliable path for tracing token usage in CrewAI today.
• Cached tokens: Some providers report prompt_tokens including cached hits. If your gateway forwards provider cache-control hints, those tokens are cheaper but still counted. Track them separately if you bill back.

Why this matters

Tracing token usage in CrewAI without attribution leads to guesswork. When a planner loops three times because of a vague prompt, that cost shows up only if you tag the planner’s LLM. The pattern above adds about 40 lines of code and zero meaningful runtime overhead beyond a dict update.

Keep the base_url indirection even after instrumentation is done. A gateway that honors client routing directives and forwards provider cache-control hints will also reduce repeated token burns via prompt caching—another lever for multi-agent cost control.

That is the full loop: instrument, run, verify, export. You now have a repeatable way to hold every agent in a CrewAI chain accountable for the tokens it spends.