LangGraph

A graph's cost is most useful per node, which identifies the expensive step and the one that loops.

pip install langgraph langchain-anthropic capsera

Decorate the nodes

from typing import TypedDict

from langchain_anthropic import ChatAnthropic
from langgraph.graph import END, START, StateGraph
import capsera
from capsera import langgraph_node

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

llm = ChatAnthropic(model="claude-sonnet-4-6")


class State(TypedDict):
    question: str
    plan: str
    draft: str


@langgraph_node("planner", team="research")
def plan(state: State) -> State:
    return {**state, "plan": llm.invoke(f"Plan: {state['question']}").content}


@langgraph_node("drafter", team="research")
def draft(state: State) -> State:
    return {**state, "draft": llm.invoke(f"Draft from: {state['plan']}").content}


graph = StateGraph(State)
graph.add_node("plan", plan)
graph.add_node("draft", draft)
graph.add_edge(START, "plan")
graph.add_edge("plan", "draft")
graph.add_edge("draft", END)
app = graph.compile()

app.invoke({"question": "How should we price agent runs?", "plan": "", "draft": ""})

This produces two agents, planner and drafter, each with its own cost. langgraph_node is @agent() plus a record of the framework.

Do not decorate the graph invocation

# This collapses every node's spend onto one agent.
@capsera.agent("research-graph")
def run(question):
    return app.invoke({...})

It works, but it discards the per-node breakdown. Decorate the nodes. For a graph-level total, use the team field.

Do not decorate a shared helper

A common pattern is to route every node's model call through one helper:

def _ask_model(prompt: str) -> str:          # do not decorate
    return llm.invoke(prompt).content


@langgraph_node("planner")
def plan(state):
    return {**state, "plan": _ask_model(f"Plan: {state['question']}")}

Leave _ask_model undecorated. Attribution follows the call stack, so calls inside it are already attributed to whichever node is running.

Decorating it would attribute every call in the graph to _ask_model, collapsing four nodes into one line item and leaving three nodes with no spend.

Loops and retries

A node that runs three times produces three events with the same attribution. A runaway loop is therefore visible as an increase in call count rather than in cost per call.

For conditional edges, decorate each destination node. Attribution follows execution, so whichever path the graph takes is what is recorded.

Inline lambda nodes

A routing node written inline has no function to decorate:

graph.add_node("route", lambda state: {**state, "next": pick(state)})

This is fine if it makes no LLM call. If it does, promote it to a named function and decorate it, or wrap the call in tag().

Async graphs

langgraph_node handles coroutine functions, and contextvars propagate into the tasks LangGraph creates, so parallel branches keep their own attribution:

@langgraph_node("searcher", team="research")
async def search(state: State) -> State:
    result = await llm.ainvoke(state["query"])
    return {**state, "results": result.content}

Two branches running concurrently under asyncio.gather do not mix their attribution, because isolation is per task rather than per thread.