LangGraph Integration

How to instrument LangGraph workflows for per-run cost attribution — the primary use case for Kostrack's trace/span system.

Wrapping a LangGraph run

Wrap graph.invoke() inside a kostrack.trace() block. Every LLM call made by any node in the graph automatically inherits the trace ID:

import kostrack
from kostrack import Anthropic
from langgraph.graph import StateGraph

# Use kostrack client in your nodes
llm = Anthropic(tags={"project": "openmanagr"})

def validate_node(state):
    response = llm.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=256,
        messages=[{"role": "user", "content": state["input"]}],
        kostrack_tags={"span_name": "validate"},
    )
    return {"validated": response.content[0].text}

def classify_node(state):
    response = llm.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        messages=[{"role": "user", "content": state["validated"]}],
        kostrack_tags={"span_name": "classify"},
    )
    return {"classification": response.content[0].text}

graph = StateGraph(dict)
graph.add_node("validate", validate_node)
graph.add_node("classify", classify_node)
graph.add_edge("validate", "classify")
graph.set_entry_point("validate")
app = graph.compile()

# Wrap the run in a trace
with kostrack.trace(
    tags={"project": "openmanagr", "feature": "invoice-pipeline"}
) as t:
    result = app.invoke({"input": "Invoice #1234..."})

print(f"Pipeline cost: ${t.total_cost_usd:.6f}")
print(f"API calls: {t.call_count}")

Multi-model workflows

Mix providers in the same trace — all calls share the same trace_id:

from kostrack import Anthropic, OpenAI

anthropic_llm = Anthropic(tags={"project": "openmanagr"})
openai_llm = OpenAI(tags={"project": "openmanagr"})

with kostrack.trace(tags={"feature": "hybrid-pipeline"}) as t:
    # Extraction with Anthropic
    extracted = anthropic_llm.messages.create(...)
    # Classification with OpenAI (cheaper)
    classified = openai_llm.chat.completions.create(...)

Querying workflow costs

-- Average cost per invoice pipeline run, last 7 days
SELECT AVG(total_cost_usd) AS avg_cost,
       COUNT(*) AS runs,
       MAX(total_cost_usd) AS max_cost
FROM trace_costs
WHERE feature = 'invoice-pipeline'
  AND started_at > NOW() - INTERVAL '7 days';