Track costs in real-time with Langfuse and enforce budgets

Introduction

When your embedding pipeline starts a million-document run, a single misconfigured routing rule or a silent cache regression can turn a $50 job into a $5,000 invoice before anyone notices — and end-of-month billing reports are far too late to halt the bleeding. Real-time cost visibility, with an in-flight kill switch, is the only way to stop overruns mid-run instead of discovering them in the next finance review.

By the end of this lesson you'll be able to instrument an embedding pipeline with Langfuse trace and span logging, accumulate per-run cost counters in process, and raise a hard stop the moment a configured USD budget is exceeded.

Key Terminology

• Trace: a Langfuse record of a single pipeline run that groups every embedding call as child spans, carrying token counts, model identifiers, latency, and computed cost for downstream analysis.
• Cumulative cost: the running USD total of non-cached embedding spend within a single run, maintained in-process so budget checks happen without round-tripping to Langfuse.
• Budget enforcer: an in-line guard that consults cumulative cost after each embedding call and raises BudgetExceededError once the configured USD ceiling is hit, with a one-shot warning fired at a configurable ratio (default 80%) before the hard stop.

Concepts

Cost tracking and budget enforcement for an embedding pipeline rest on three coupled responsibilities. First, instrumentation: every embedding request — cache hit or live call — must emit a Langfuse span tagged with model, provider, token count, per-request cost, and a cumulative-cost field, so post-hoc analysis can attribute spend to a specific run, document, or provider. Second, real-time enforcement: a budget check runs synchronously after each non-cached call against an in-memory cumulative counter; querying Langfuse's backend on the hot path would add latency and fail open during outages, so the counter lives in the tracer itself. Third, aggregation: structured queries over the persisted span data roll spend up by provider, source document, and time window, surfacing which routes or documents dominate cost and which would benefit most from caching. Cache hits are logged but excluded from cumulative spend, since their incremental cost is zero — this is what makes the cache hit rate the single most actionable lever for reducing pipeline cost.

Code Walkthrough

Langfuse Instrumentation and Budget Enforcement

The tracer and the enforcer are designed to work hand-in-glove: the tracer maintains an in-memory cumulative cost counter that the enforcer reads on every call. Combining them in one snippet shows the full real-time cost-control loop — log a span, update the counter, check the budget, raise if over — without the indirection of two separate listings.

LangfuseEmbeddingTracer logs every embedding request as a Langfuse span (cache hits included) and increments cumulative_cost only for live, non-cached calls. BudgetEnforcer consults that counter after each call, fires a one-shot warning at a configurable ratio (default 80%), and raises BudgetExceededError once the configured USD ceiling is hit so the pipeline can flush state and exit cleanly.

Code snippetpython
1from langfuse import Langfuse
2
3class LangfuseEmbeddingTracer:
4    def __init__(
5        self,
6        langfuse_client: Langfuse,
7    ):
8        self.client = langfuse_client
9        self.cumulative_cost = 0.0
10        self.request_count = 0
11        self.cache_hits = 0
12
13    def start_trace(
14        self,
15        run_id: str,
16        metadata: dict = None,
17    ):
18        self.trace = self.client.trace(
19            name=f"embedding_run_{run_id}",
20            metadata=metadata or {},
21        )
22        self.cumulative_cost = 0.0
23        self.request_count = 0
24        self.cache_hits = 0
25        return self.trace
26
27    def log_embedding(
28        self,
29        text: str,
30        result: EmbeddingResult,
31        from_cache: bool = False,
32    ):
33        self.request_count += 1
34        if from_cache:
35            self.cache_hits += 1
36        else:
37            self.cumulative_cost += (
38                result.cost_usd
39            )
40        self.trace.span(
41            name="embedding_request",
42            input={"text_length": len(text)},
43            output={
44                "model": result.model,
45                "provider": result.provider,
46                "dimensions": len(
47                    result.embedding
48                ),
49            },
50            metadata={
51                "token_count": (
52                    result.token_count
53                ),
54                "cost_usd": result.cost_usd,
55                "from_cache": from_cache,
56                "cumulative_cost": (
57                    self.cumulative_cost
58                ),
59            },
60        )
61
62    def finalize(self) -> dict:
63        summary = {
64            "total_requests": (
65                self.request_count
66            ),
67            "cache_hits": self.cache_hits,
68            "cache_hit_rate": (
69                self.cache_hits
70                / max(self.request_count, 1)
71            ),
72            "total_cost_usd": (
73                self.cumulative_cost
74            ),
75        }
76        self.trace.update(
77            metadata={"run_summary": summary},
78        )
79        self.client.flush()
80        return summary
81
82class BudgetExceededError(Exception):
83    def __init__(
84        self,
85        budget: float,
86        spent: float,
87    ):
88        self.budget = budget
89        self.spent = spent
90        super().__init__(
91            f"Budget exceeded: "
92            f"${spent:.4f} / ${budget:.4f}"
93        )
94
95class BudgetEnforcer:
96    def __init__(
97        self,
98        max_budget_usd: float,
99        warn_threshold: float = 0.8,
100    ):
101        self.max_budget = max_budget_usd
102        self.warn_threshold = warn_threshold
103        self._warned = False
104
105    def check(
106        self,
107        tracer: LangfuseEmbeddingTracer,
108    ) -> None:
109        current = tracer.cumulative_cost
110        if current >= self.max_budget:
111            raise BudgetExceededError(
112                self.max_budget, current
113            )
114        ratio = current / self.max_budget
115        if (
116            ratio >= self.warn_threshold
117            and not self._warned
118        ):
119            logger.warning(
120                "Budget warning: %.1f%% used "
121                "($%.4f / $%.4f)",
122                ratio * 100,
123                current,
124                self.max_budget,
125            )
126            self._warned = True

• Tracer constructor and start_trace: hold the Langfuse client and reset the in-memory counters (cumulative_cost, request_count, cache_hits) at the start of each run; the run ID becomes the trace name so pipeline logs correlate directly with Langfuse traces.
• log_embedding: records every call — cache hit or live — as a span carrying token count, cost, provider, and the from_cache flag. Only live calls increment cumulative_cost, which is what makes cache hit rate the most actionable lever for cost reduction.
• finalize: writes the run summary (totals and cache hit rate) back onto the trace and calls client.flush() so spans aren't lost on process exit.
• BudgetExceededError: carries both the configured limit and actual spend so the pipeline's error handler can log how far over-budget the run went.
• BudgetEnforcer.check: synchronously consults the tracer's in-memory counter after each call — no Langfuse round-trip on the hot path — fires a one-shot warning at the configured ratio, then raises BudgetExceededError the moment spend crosses the ceiling.

Cost Reporting

Cost reports aggregate spending data across pipeline runs, document sources, and providers. These reports feed into capacity planning, budget allocation, and routing optimization decisions.

CostReporter queries Langfuse traces and aggregates costs by provider, model, and document source. The reporter generates structured summaries that can be rendered as dashboards or exported to spreadsheets for finance review.

1class CostReporter:
2    def __init__(
3        self,
4        db_pool: asyncpg.Pool,
5    ):
6        self.db_pool = db_pool
7
8    async def report_by_provider(
9        self,
10        run_id: str,
11    ) -> list[dict]:
12        async with (
13            self.db_pool.acquire() as conn
14        ):
15            rows = await conn.fetch(
16                """
17                SELECT provider,
18                       COUNT(*) as requests,
19                       SUM(token_count)
20                           AS total_tokens,
21                       SUM(cost_usd)
22                           AS total_cost
23                FROM embedding_logs
24                WHERE run_id = $1
25                GROUP BY provider
26                ORDER BY total_cost DESC
27                """,
28                run_id,
29            )
30        return [dict(r) for r in rows]
31
32    async def report_by_source(
33        self,
34        run_id: str,
35    ) -> list[dict]:
36        async with (
37            self.db_pool.acquire() as conn
38        ):
39            rows = await conn.fetch(
40                """
41                SELECT source_doc,
42                       COUNT(*) as chunks,
43                       SUM(cost_usd)
44                           AS total_cost,
45                       SUM(CASE
46                           WHEN from_cache
47                           THEN 1 ELSE 0 END)
48                           AS cache_hits
49                FROM embedding_logs
50                WHERE run_id = $1
51                GROUP BY source_doc
52                ORDER BY total_cost DESC
53                """,
54                run_id,
55            )
56        return [dict(r) for r in rows]
57
58    async def total_cost_summary(
59        self,
60        days: int = 30,
61    ) -> dict:
62        async with (
63            self.db_pool.acquire() as conn
64        ):
65            row = await conn.fetchrow(
66                """
67                SELECT COUNT(DISTINCT run_id)
68                           AS runs,
69                       SUM(cost_usd)
70                           AS total_cost,
71                       AVG(cost_usd)
72                           AS avg_cost_per_req
73                FROM embedding_logs
74                WHERE created_at > NOW()
75                    - INTERVAL '%s days'
76                """ % days,
77            )
78        return dict(row) if row else {}

Do's and Don'ts

Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.

Do's

1. ✓Log every embedding call as a Langfuse span — cache hits included — so cache hit rate and provider mix can be reconstructed from traces alone.
2. ✓Maintain cumulative cost in-process on the tracer and check budgets synchronously after each non-cached call, so enforcement does not depend on Langfuse availability.
3. ✓Fire a one-shot warning at the configured ratio (e.g. 80%) before the hard stop so operators can intervene before BudgetExceededError aborts the run.

Don'ts

1. ✗Don't increment cumulative cost on cache hits — doing so inflates spend metrics and masks the value of the cache.
2. ✗Don't query Langfuse's backend on the hot path to check budgets; in-memory counters are the source of truth during a run.
3. ✗Don't swallow BudgetExceededError; catch it at the pipeline boundary to flush pending writes and log the interruption point, then re-surface it.