File size: 1,778 Bytes
6993919 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | #!/usr/bin/env python3
"""
Langfuse Smart Sampling — Never exceed free tier, keep local as fallback.
Strategy:
- Sample 20% of normal traces → cloud
- 100% of errors (score < 0.5) → cloud
- 100% of user-facing requests → cloud
- Everything → local ClickHouse (full archive)
- Target: 800-1,200 observations/day (free tier: 1,600/day)
Free tier: 50,000 observations/month
Our target: 30,000/month (60% headroom)
"""
import logging
import os
import random
logger = logging.getLogger("langfuse_sampler")
SAMPLING_RATE = float(os.getenv("LANGFUSE_SAMPLING_RATE", "0.20")) # 20% normal
ERROR_RATE = 1.0 # 100% of errors
USER_RATE = 1.0 # 100% of user-facing
_counters = {
"total": 0,
"sampled": 0,
"errors_captured": 0,
"local_only": 0,
}
def should_send_to_cloud(
is_error: bool = False,
is_user_facing: bool = False,
force: bool = False,
) -> bool:
"""Decide whether to send this trace to Langfuse cloud."""
_counters["total"] += 1
if force:
_counters["sampled"] += 1
return True
# Always capture errors (failed calls, hallucinations, etc.)
if is_error:
_counters["errors_captured"] += 1
_counters["sampled"] += 1
return True
# Always capture user-facing interactions
if is_user_facing:
_counters["sampled"] += 1
return True
# Sample normal background traces
if random.random() < SAMPLING_RATE:
_counters["sampled"] += 1
return True
_counters["local_only"] += 1
return False
def stats() -> dict:
total = max(_counters["total"], 1)
return {
**_counters,
"sampling_rate_pct": round(_counters["sampled"] / total * 100, 1),
"estimated_daily": _counters["sampled"],
}
|