Langfuse Cost Control and Loop Detection: Span Tree Explosion, Evaluation Storm, and Prompt Polling Loops

Langfuse is an open-source LLM observability platform: instrument your agent with @observe() decorators, and every LLM call gets captured as a nested span tree with timing, token counts, and inferred cost. Fourteen thousand GitHub stars and a managed cloud offering make it one of the most widely deployed tracing backends for Python LLM agents.

The observability layer does exactly what it promises. What it does not do — and does not claim to do — is enforce any spend ceiling. Langfuse records what happened. Nothing in the SDK stops the agent from calling the LLM again. If your agent is in a runaway loop, Langfuse will produce a beautifully detailed trace of every iteration until your API credits run out.

There are also four structural patterns specific to Langfuse's architecture that amplify costs beyond the agent loop itself: the @observe() decorator creates span objects that accumulate without bound in memory; the background flush queue silently drops events when full; run_on_dataset() fans out to O(items × agent steps × judge metrics × retries) LLM calls from a single function call; and get_prompt() with caching disabled makes a network round-trip on every agent iteration. Each one is invisible in your Langfuse trace until the invoice lands.

Pattern 1: Span Tree Explosion in Recursive Agents

The @observe() decorator is Langfuse's primary instrumentation primitive. Applied to a function, it creates a new span object on every call and attaches it as a child of the currently active span. For a simple request-response agent, the trace tree is shallow: one root span, one generation child per LLM call, done. For a recursive self-reflection agent — one that calls itself to critique its own output — the span tree grows without bound.

Internally, the Python SDK maintains an active span stack in a ContextVar. Every @observe()-wrapped function pushes a new StatefulSpanClient onto that stack and pops it on exit. The stack itself is lightweight, but the spans accumulate in the background task manager's send queue. When the agent loops 50 times, you have 50 generations queued for flushing — each one carrying its full input and output text as a JSON payload. At 200 generation objects averaging 4 KB each, that is 800 KB of in-process JSON before the first flush completes.

The SDK's default flush queue holds 100 events (max_queue_size in LangfuseTaskManager). A fast recursive agent exceeds this in under a second. Once the queue is full, new events are silently dropped — you see a partial trace in the Langfuse UI, not an error. The cost of those dropped calls is still billed by your LLM provider. The trace just does not show them.

The guard: SpanDepthGuard

The most effective intervention is to limit call depth before the recursive chain produces its Nth span. A depth counter maintained in a ContextVar mirrors the SDK's span stack without touching Langfuse internals:

import contextvars
from langfuse.decorators import observe, langfuse_context
from runguard import guard, BudgetExceededError

_span_depth: contextvars.ContextVar[int] = contextvars.ContextVar(
    "span_depth", default=0
)

MAX_SPAN_DEPTH = 6   # reflexion agents rarely need more than 4
MAX_CALL_BUDGET = 20 # absolute LLM call ceiling for this trace

class SpanDepthGuard:
    def __init__(self, max_depth: int = MAX_SPAN_DEPTH):
        self.max_depth = max_depth
        self._call_count = 0

    def check(self, label: str = "") -> None:
        depth = _span_depth.get()
        if depth >= self.max_depth:
            # Update Langfuse trace with trip reason before raising
            langfuse_context.update_current_trace(
                tags=["runguard_trip", "span_depth_exceeded"],
                metadata={"runguard_trip_reason": "span_depth_exceeded",
                           "runguard_depth": depth,
                           "runguard_label": label}
            )
            raise RecursionLimitError(
                f"SpanDepthGuard tripped at depth {depth} (max {self.max_depth})"
            )
        self._call_count += 1
        if self._call_count > MAX_CALL_BUDGET:
            langfuse_context.update_current_trace(
                tags=["runguard_trip", "call_budget_exceeded"],
                metadata={"runguard_trip_reason": "call_budget_exceeded",
                           "runguard_calls": self._call_count}
            )
            raise BudgetExceededError(
                f"SpanDepthGuard: call budget {MAX_CALL_BUDGET} exceeded"
            )

class RecursionLimitError(Exception):
    pass

def guarded_observe(span_guard: SpanDepthGuard, name: str = None):
    """Decorator that wraps @observe() with span depth enforcement."""
    def decorator(fn):
        @observe(name=name or fn.__name__)
        def wrapper(*args, **kwargs):
            token = _span_depth.set(_span_depth.get() + 1)
            try:
                span_guard.check(label=fn.__name__)
                return fn(*args, **kwargs)
            finally:
                _span_depth.reset(token)
        return wrapper
    return decorator


# Usage: replace @observe() with @guarded_observe(span_guard)
span_guard = SpanDepthGuard(max_depth=6)

@guarded_observe(span_guard, name="reflect")
def reflect(draft: str, iteration: int) -> str:
    """Self-reflection step — calls LLM to critique and refine."""
    critique = llm_call(f"Critique this draft: {draft}")
    if needs_refinement(critique) and iteration < 10:
        return reflect(draft=critique, iteration=iteration + 1)  # recursive
    return critique

@observe()
def run_agent(task: str) -> str:
    try:
        result = reflect(draft=initial_draft(task), iteration=0)
        return result
    except (RecursionLimitError, BudgetExceededError) as e:
        # Trip is already logged to Langfuse trace via update_current_trace
        return f"[guarded] {e}"

The langfuse_context.update_current_trace() call on trip is important: it writes the guard's verdict into the Langfuse trace itself, so you can filter traces by tag:runguard_trip in the Langfuse UI and see exactly which traces were stopped and at what depth. The observability layer records the circuit break; the circuit break prevents the runaway.

Pattern 2: Async Flush Queue Overflow and Silent Event Loss

Langfuse's Python SDK sends trace events to its backend asynchronously. When you call a function wrapped with @observe(), the resulting span object is placed into an in-memory queue managed by LangfuseTaskManager. A background thread drains that queue and POSTs batches to api.langfuse.com/api/public/ingestion (or your self-hosted endpoint).

The queue has a maximum size. In the Python SDK (as of v3.x), this defaults to 100 events. The background thread drains at whatever rate your network and Langfuse API can sustain — typically 50–200 events per second under normal conditions. A looping agent that fires 10 LLM calls per second will outpace the drain rate within 10 seconds. Once the queue is at capacity, the SDK enqueues with a non-blocking put_nowait(); events that cannot be enqueued are silently dropped. No exception is raised. No log line is emitted by default.

The consequence is that your Langfuse trace appears to show fewer iterations than actually ran. The LLM provider billed for all of them. Teams discover this gap when they reconcile their provider bill against the Langfuse token-count report and find they're missing 30–40% of calls during peak load.

The guard: FlushQueueMonitor

The task manager's queue is accessible at langfuse._task_manager._queue. A pre-submission check that reads qsize() relative to maxsize gives you early warning before events start dropping:

import queue
from langfuse import Langfuse

class FlushQueueMonitor:
    """
    Blocks new LLM submissions when the Langfuse flush queue is near capacity.
    Call check() before each LLM call in your agent loop.
    """
    def __init__(
        self,
        langfuse_client: Langfuse,
        high_watermark: float = 0.80,  # block at 80% full
        critical_watermark: float = 0.95,  # flush sync at 95% full
    ):
        self._q: queue.Queue = langfuse_client._task_manager._queue
        self._high = high_watermark
        self._critical = critical_watermark
        self._langfuse = langfuse_client
        self.drops_prevented = 0

    def check(self) -> None:
        """Raises FlushQueueFullError if queue is at or above high_watermark."""
        maxsize = self._q.maxsize
        if maxsize <= 0:
            return  # unbounded queue — no risk
        fill = self._q.qsize() / maxsize
        if fill >= self._critical:
            # Synchronous flush attempt before blocking
            self._langfuse.flush()
        fill = self._q.qsize() / maxsize
        if fill >= self._high:
            self.drops_prevented += 1
            raise FlushQueueFullError(
                f"Langfuse flush queue at {fill:.0%} capacity "
                f"({self._q.qsize()}/{maxsize}). "
                f"Pausing agent to prevent silent event loss."
            )

class FlushQueueFullError(Exception):
    pass


# Usage
langfuse = Langfuse()
queue_monitor = FlushQueueMonitor(langfuse, high_watermark=0.80)

@observe()
def agent_step(state: dict) -> dict:
    queue_monitor.check()  # raises FlushQueueFullError if queue is near full
    response = llm_call(state["prompt"])
    return {"output": response, "cost": response.usage.total_tokens}

def run_agent_loop(task: str) -> str:
    state = {"prompt": task}
    for i in range(100):
        try:
            state = agent_step(state)
            if is_complete(state):
                break
        except FlushQueueFullError as e:
            # Log the trip as a Langfuse score on the current trace
            langfuse.score(
                name="runguard_flush_queue_trip",
                value=1,
                comment=str(e)
            )
            break  # or: time.sleep(0.5) and retry once
    langfuse.flush()  # ensure all queued events ship before returning
    return state.get("output", "")

The synchronous flush() call at critical_watermark is a one-shot attempt to drain the queue before blocking the agent. If the Langfuse backend is slow (network spike, API rate limit on the ingestion endpoint), the flush call itself may take several seconds — which is acceptable at the 95% mark but would be disruptive at 80%. The two-threshold design handles both cases.

A subtlety: maxsize of 0 in Python's Queue means unbounded. If you initialized Langfuse with a very large or zero-cap queue, the guard is a no-op and drops never happen — which is fine. Check your SDK initialization: the default is 100 unless you passed max_queue_size explicitly.

Pattern 3: Dataset Evaluation Run Fan-out

Langfuse's dataset API is the standard way to run offline evaluations: create a dataset, populate it with test cases, call run_on_dataset() with your agent function, and Langfuse records each run as a trace linked to the dataset item. LLM-as-judge scoring fires additional LLM calls to evaluate each agent response.

The cost arithmetic is multiplicative. A dataset with 200 items, an agent that averages 3 LLM calls per run, and 2 judge metrics each requiring 1 LLM call each produces:

• 200 × 3 = 600 agent LLM calls
• 200 × 2 = 400 judge LLM calls
• 1,000 total calls from one run_on_dataset() invocation

Add a retry wrapper (max_retries=2 is common in eval pipelines to handle intermittent failures) and the ceiling doubles to 2,000 calls. Use a stronger judge model (GPT-4.1, Claude Opus) and you multiply the per-call cost by 5–20×. A 200-row dataset eval with GPT-4.1 as the judge can cost $40–120 per run. Teams often run these in notebooks with a single function call and no spend guard.

The fan-out is not a bug — it is the intended behavior. The problem is the absence of a pre-run estimate that would let you make an informed decision before the charges start.

The guard: DatasetEvalGuard

from dataclasses import dataclass
from typing import Callable, Any
from langfuse import Langfuse

# Model cost table (USD per 1M tokens, combined input+output blended estimate)
MODEL_COST_PER_1M = {
    "gpt-4o": 7.50,
    "gpt-4o-mini": 0.30,
    "gpt-4.1": 8.00,
    "gpt-4.1-mini": 0.60,
    "claude-opus-4-7": 75.00,
    "claude-sonnet-4-6": 9.00,
    "claude-haiku-4-5": 1.25,
}
DEFAULT_TOKENS_PER_CALL = 2000  # conservative blended estimate

@dataclass
class EvalPlan:
    dataset_size: int
    agent_calls_per_item: int
    judge_model: str
    judge_metrics: int
    max_retries: int = 1
    agent_model: str = "gpt-4o-mini"

    def estimated_calls(self) -> int:
        agent = self.dataset_size * self.agent_calls_per_item * self.max_retries
        judge = self.dataset_size * self.judge_metrics * self.max_retries
        return agent + judge

    def estimated_cost_usd(self) -> float:
        agent_cost = (
            self.dataset_size * self.agent_calls_per_item * self.max_retries
            * DEFAULT_TOKENS_PER_CALL / 1_000_000
            * MODEL_COST_PER_1M.get(self.agent_model, 10.0)
        )
        judge_cost = (
            self.dataset_size * self.judge_metrics * self.max_retries
            * DEFAULT_TOKENS_PER_CALL / 1_000_000
            * MODEL_COST_PER_1M.get(self.judge_model, 10.0)
        )
        return agent_cost + judge_cost


class DatasetEvalGuard:
    """
    Pre-flight cost estimator for Langfuse dataset evaluation runs.
    Call plan() before run_on_dataset() to get a confirmed or blocked verdict.
    """
    def __init__(
        self,
        langfuse: Langfuse,
        max_calls: int = 500,
        max_cost_usd: float = 20.0,
        sample_if_over: bool = True,
    ):
        self._lf = langfuse
        self.max_calls = max_calls
        self.max_cost_usd = max_cost_usd
        self.sample_if_over = sample_if_over

    def plan(self, eval_plan: EvalPlan) -> dict:
        """
        Returns a dict with keys: allowed (bool), sampled_size (int | None),
        estimated_calls (int), estimated_cost_usd (float), reason (str).
        """
        calls = eval_plan.estimated_calls()
        cost = eval_plan.estimated_cost_usd()
        over_calls = calls > self.max_calls
        over_cost = cost > self.max_cost_usd

        if not over_calls and not over_cost:
            return {
                "allowed": True, "sampled_size": None,
                "estimated_calls": calls, "estimated_cost_usd": cost,
                "reason": "within limits"
            }

        if not self.sample_if_over:
            return {
                "allowed": False, "sampled_size": None,
                "estimated_calls": calls, "estimated_cost_usd": cost,
                "reason": f"exceeds limits (calls={calls}, cost=${cost:.2f})"
            }

        # Compute max dataset size that fits within both ceilings
        max_by_calls = self.max_calls // (
            eval_plan.agent_calls_per_item * eval_plan.max_retries
            + eval_plan.judge_metrics * eval_plan.max_retries
        )
        max_by_cost = int(self.max_cost_usd / (cost / eval_plan.dataset_size))
        sampled_size = max(10, min(max_by_calls, max_by_cost))

        sampled_plan = EvalPlan(
            dataset_size=sampled_size,
            agent_calls_per_item=eval_plan.agent_calls_per_item,
            judge_model=eval_plan.judge_model,
            judge_metrics=eval_plan.judge_metrics,
            max_retries=eval_plan.max_retries,
            agent_model=eval_plan.agent_model,
        )
        return {
            "allowed": True,
            "sampled_size": sampled_size,
            "estimated_calls": sampled_plan.estimated_calls(),
            "estimated_cost_usd": sampled_plan.estimated_cost_usd(),
            "reason": f"sampled {sampled_size}/{eval_plan.dataset_size} items to fit limits"
        }


# Usage
langfuse = Langfuse()
guard = DatasetEvalGuard(langfuse, max_calls=300, max_cost_usd=15.0)

dataset = langfuse.get_dataset("rag-eval-200-items")
items = dataset.items

plan = EvalPlan(
    dataset_size=len(items),
    agent_calls_per_item=3,       # measured from a 5-item pilot run
    judge_model="gpt-4.1",
    judge_metrics=2,
    max_retries=2,
    agent_model="gpt-4o-mini",
)
verdict = guard.plan(plan)
print(f"Eval plan: {verdict}")
# → {'allowed': True, 'sampled_size': 37, 'estimated_calls': 296,
#    'estimated_cost_usd': 14.22, 'reason': 'sampled 37/200 items to fit limits'}

if verdict["sampled_size"]:
    items = items[:verdict["sampled_size"]]

for item in items:
    with item.observe(run_name="guarded-eval-run") as trace_id:
        output = my_agent(item.input)
        langfuse.score(
            trace_id=trace_id,
            name="correctness",
            value=judge_correctness(item.input, output, item.expected_output)
        )

The agent_calls_per_item estimate deserves a pilot run, not a guess. Run your agent on 5 dataset items, count the LLM calls using Langfuse's cost data for those 5 traces, and use the average. An agent that averages 2.4 calls per item should not be estimated at 1 — the 2.4× difference means your 200-item eval costs 480% more than your initial back-of-envelope math.

Pattern 4: Prompt Version Polling Without Caching

Langfuse's prompt management feature lets you version and deploy prompts through the Langfuse UI. Agents fetch the current version at runtime via langfuse.get_prompt(name). This is genuinely useful for prompt engineering workflows — you can iterate on prompts without redeploying code.

The SDK caches fetched prompts client-side. The cache TTL is configured per-call via cache_ttl_seconds (default: 60 seconds in v3.x). When cache_ttl_seconds=0 is passed — which developers use during prompt iteration to see changes immediately — every get_prompt() call makes an HTTP GET request to api.langfuse.com/api/public/v2/prompts/{name}. In a tight production loop running at 10 iterations per second, that is 10 HTTP requests per second to the Langfuse API, each adding 50–200 ms of network latency to the agent's hot path.

At low iteration rates this is harmless. At production scale — multi-tenant systems handling concurrent agent sessions — it becomes a self-inflicted DDoS against the Langfuse API. The X-RateLimit-Remaining header drops, 429 responses start arriving, and your retry logic triggers another wave of requests. Every retry is a small cost ($0.000N per HTTP call) but at scale the cumulative cost and latency penalty are significant.

The subtler problem: developers set cache_ttl_seconds=0 during development and forget to restore it before deploying. The production system then operates without caching indefinitely.

The guard: PromptCacheGuard

import time
import threading
from langfuse import Langfuse
from langfuse.model import TextPromptClient

class PromptCacheGuard:
    """
    Enforces a minimum prompt cache TTL and monitors fetch rate.
    Wraps langfuse.get_prompt() to prevent accidental cache bypass.
    """
    MIN_TTL_SECONDS = 30          # enforce floor: never allow 0 in production
    MAX_FETCH_RATE_PER_MIN = 10   # alert threshold (not a hard block)

    def __init__(self, langfuse: Langfuse, min_ttl: int = MIN_TTL_SECONDS):
        self._lf = langfuse
        self._min_ttl = min_ttl
        self._fetch_log: list[float] = []
        self._lock = threading.Lock()

    def get_prompt(
        self,
        name: str,
        version: int | None = None,
        cache_ttl_seconds: int = 60,
        label: str = "production",
    ) -> TextPromptClient:
        """Drop-in replacement for langfuse.get_prompt()."""
        # Enforce minimum TTL
        effective_ttl = max(self._min_ttl, cache_ttl_seconds)
        if cache_ttl_seconds < self._min_ttl:
            import warnings
            warnings.warn(
                f"PromptCacheGuard: cache_ttl_seconds={cache_ttl_seconds} is below "
                f"minimum {self._min_ttl}. Using {effective_ttl}s.",
                stacklevel=2
            )

        # Rate monitoring (does not block, but logs for alerting)
        now = time.monotonic()
        with self._lock:
            self._fetch_log = [t for t in self._fetch_log if now - t < 60]
            self._fetch_log.append(now)
            rate = len(self._fetch_log)

        if rate > self.MAX_FETCH_RATE_PER_MIN:
            import logging
            logging.getLogger("runguard.langfuse").warning(
                "PromptCacheGuard: %d prompt fetches in last 60s for '%s'. "
                "Consider raising cache_ttl_seconds or caching prompts at module level.",
                rate, name
            )

        return self._lf.get_prompt(
            name,
            version=version,
            cache_ttl_seconds=effective_ttl,
            label=label,
        )


# Module-level singleton — prompts are fetched once per process start
# and refreshed every 5 minutes. Use this instead of per-call get_prompt().
_langfuse = Langfuse()
_prompt_guard = PromptCacheGuard(_langfuse, min_ttl=30)

# Fetch once at module import; SDK refreshes from cache every 300s
_agent_prompt = _prompt_guard.get_prompt("agent-system-prompt", cache_ttl_seconds=300)

@observe()
def agent_step(user_message: str) -> str:
    # compile() returns the rendered prompt string from the cached version
    system = _agent_prompt.compile()
    response = llm_call(system_prompt=system, user=user_message)
    _agent_prompt.link(langfuse_context.get_current_observation_id())
    return response.content

The prompt.link(observation_id) call on the last line is specific to Langfuse's prompt tracking feature — it records which prompt version was used in each generation span, enabling the "prompt usage" analytics view in the UI. Without it, Langfuse cannot tell you which prompt version produced which responses. The guard preserves this linkage while enforcing the cache floor.

For the development workflow where you genuinely need to see prompt changes immediately: set LANGFUSE_PROMPT_CACHE_TTL=0 in your dev environment and add a guard check that raises in production if cache_ttl_seconds < 30. The development experience stays fast; the production path cannot accidentally go uncached.

Putting It Together: A Guarded Langfuse Agent

Each guard operates on a different resource — span depth, flush queue fill, evaluation call budget, and prompt fetch rate. They compose cleanly because they share no state:

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse()

# Initialize guards
span_guard = SpanDepthGuard(max_depth=6)
queue_monitor = FlushQueueMonitor(langfuse, high_watermark=0.80)
prompt_guard = PromptCacheGuard(langfuse, min_ttl=30)
eval_guard = DatasetEvalGuard(langfuse, max_calls=300, max_cost_usd=15.0)

# Cached prompt — refreshed every 5 minutes, linked per generation
agent_prompt = prompt_guard.get_prompt("agent-v2", cache_ttl_seconds=300)

@guarded_observe(span_guard, name="agent_turn")
def agent_turn(messages: list[dict], depth: int = 0) -> str:
    queue_monitor.check()  # block if flush queue near capacity

    system = agent_prompt.compile()
    response = llm_call(system_prompt=system, messages=messages)
    agent_prompt.link(langfuse_context.get_current_observation_id())

    tool_calls = response.tool_calls or []
    if tool_calls:
        results = [run_tool(tc) for tc in tool_calls]
        messages = messages + tool_results_to_messages(tool_calls, results)
        return agent_turn(messages, depth=depth + 1)  # recursive — guarded

    return response.content

@observe()
def handle_request(user_message: str) -> str:
    try:
        return agent_turn([{"role": "user", "content": user_message}])
    except (RecursionLimitError, BudgetExceededError, FlushQueueFullError) as e:
        langfuse_context.update_current_trace(
            tags=["runguard_trip"],
            metadata={"runguard_error": str(e)}
        )
        return f"[request stopped by RunGuard: {type(e).__name__}]"
    finally:
        # Ensure spans are shipped even when the agent trips early
        langfuse.flush()

The finally: langfuse.flush() at the request boundary is load-bearing. When a guard trips and raises an exception, Python unwinds the call stack and the SDK's normal flush triggers — but only if your framework gives the background thread enough time to complete. In serverless environments (Lambda, Cloud Run) that terminate the process immediately after returning, the in-flight flush thread is killed before it ships. An explicit synchronous flush at the request boundary guarantees the trip is recorded in Langfuse before the process exits.

What RunGuard Adds to a Langfuse-Instrumented Stack

Layer	Langfuse responsibility	RunGuard responsibility
Observability	Capture every LLM call as a span with timing, tokens, cost	—
Span depth	Display nested span tree in trace viewer	Trip agent when depth exceeds ceiling; tag trip in trace
Flush reliability	Drain queue asynchronously (may drop if full)	Block new submissions before queue reaches capacity
Dataset eval	Record each run as a linked trace item	Pre-compute call estimate; sample dataset if over ceiling
Prompt fetch	Cache prompts client-side with configurable TTL	Enforce minimum TTL; alert on high fetch rate
Cost ceiling	Report cost after the fact from token counts	Enforce ceiling before charges accrue

Langfuse's tracing documentation describes the SDK as an observability tool, not a policy engine. That is the right framing. Policy — "this agent must stop before spending $20" — belongs in a separate layer that can trip the agent at runtime, not after the fact in a dashboard.

Failure Modes This Post Does Not Cover

Four patterns were enough for one post. Langfuse's architecture has others:

• Ingestion rate limits — Langfuse Cloud's ingestion API rate-limits at the project level. High-concurrency agents can exhaust the limit for all users in the project.
• Score loop — An automated scoring pipeline that scores each generation using an LLM judge, and then uses the score to decide whether to run the agent again, can recurse if the score threshold is never met.
• Self-hosted backend overload — Langfuse self-hosted (Docker Compose or Kubernetes) has a Postgres write path that degrades under sustained high ingestion rates. Agents writing 1,000+ spans per minute to a single-node Postgres can induce latency spikes that slow the flush thread, compounding the queue-fill risk from Pattern 2.
• Large span payload cost — Langfuse charges for data ingested on its cloud plan. Agents logging full conversation histories (multi-turn, multi-tool) as span I/O can generate 50–200 KB of span payload per trace. At scale, the Langfuse bill grows independently of your LLM provider bill.