LangChain

LCEL chains, invoke, batch, and ainvoke are recorded. Where to attach attribution, and the callback handler for call sites you do not own.

Recorded with no integration, because ChatOpenAI and ChatAnthropic construct the provider clients the SDK patches. Verified in the SDK's test suite against real LangChain running against a local mock provider.

pip install langchain langchain-openai capsera

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
import capsera

capsera.init(api_key=os.environ["CAPSERA_API_KEY"])

prompt = ChatPromptTemplate.from_template("Classify this ticket:\n\n{ticket}")
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

Coverage

Call	Recorded
`chain.invoke(...)`	Yes
`await chain.ainvoke(...)`	Yes
`chain.batch([...])`	Yes, one event per underlying call
`await chain.abatch([...])`	Yes
`chain.stream(...)` and `astream(...)`	Yes
`OpenAIEmbeddings().embed_query(...)`	Yes

batch produces one event per provider call rather than one per batch, so the cost of a backfill is the sum of its calls and an outlier is visible.

Attaching attribution

Decorate the function that owns the work:

@capsera.agent("ticket-classifier", team="support", task_type="triage")
def classify(ticket: str) -> str:
    return chain.invoke({"ticket": ticket})

Every call the chain makes, including retriever embeddings, is attributed to ticket-classifier, and caller_file points at your function rather than langchain_core/runnables/base.py.

For a chain invoked from several places with different meanings, use tag() at the call site:

with capsera.tag("nightly-backfill", task_type="batch"):
    chain.batch(tickets)

Call sites you do not own

Inside an agent executor or a retrieval chain, the LLM call happens in framework code and there is no function of yours to decorate. Use the callback handler:

from capsera import LangChainCapseraCallback

chain.invoke(
    {"ticket": ticket},
    config={
        "callbacks": [
            LangChainCapseraCallback(agent="ticket-classifier", team="support")
        ]
    },
)

The handler pushes attribution on on_llm_start and pops it on on_llm_end or on_llm_error, so it applies for the duration of each LLM call and unwinds correctly when a call raises. It defaults task_type to "langchain", which separates framework-driven spend from direct calls.

Attach it to the constructor to cover every invocation of one model:

llm = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=[LangChainCapseraCallback(agent="ticket-classifier")],
)

Retrieval chains

A RAG chain makes two kinds of call: embeddings for retrieval, then a completion. Both are recorded under the same attribution, which is usually correct because the retrieval cost belongs to the answer it produced.

To separate them, tag the retrieval step:

with capsera.tag("retriever", task_type="embedding"):
    docs = retriever.invoke(question)

with capsera.tag("answerer", task_type="generation"):
    answer = chain.invoke({"question": question, "docs": docs})

Embeddings bypass routing and budget enforcement, so a blocking budget will not stop the retrieval half of a RAG pipeline.

Which provider is recorded

The one the chain used. ChatAnthropic records anthropic and ChatOpenAI records openai. ChatOpenAI pointed at Groq records groq, because the SDK reads the underlying client's base URL rather than the LangChain class name. A chain with fallbacks across providers therefore reports which one served each call.