langfuse.Sampler
1import hashlib 2import logging 3 4 5log = logging.getLogger("langfuse") 6 7 8class Sampler: 9 sample_rate: float 10 11 def __init__(self, sample_rate: float): 12 self.sample_rate = sample_rate 13 14 def sample_event(self, event: dict): 15 # need to get trace_id from a given event 16 # returns true if 17 18 if "type" in event and "body" in event: 19 event_type = event["type"] 20 21 trace_id = None 22 23 if event_type == "trace-create" and "id" in event["body"]: 24 trace_id = event["body"]["id"] 25 elif "trace_id" in event["body"]: 26 trace_id = event["body"]["trace_id"] 27 elif "traceId" in event["body"]: 28 trace_id = event["body"]["traceId"] 29 else: 30 log.error("Unexpected event format: No trace id found in event") 31 return True 32 33 return self.deterministic_sample(trace_id, self.sample_rate) 34 35 else: 36 log.error("Unexpected event format: No trace id found in event") 37 return True 38 39 def deterministic_sample(self, trace_id: str, sample_rate: float): 40 """Determins if an event should be sampled based on the trace_id and sample_rate. Event will be sent to server if True""" 41 log.debug( 42 f"Applying deterministic sampling to trace_id: {trace_id} with rate {sample_rate}" 43 ) 44 45 # Use SHA-256 to hash the trace_id 46 hash_object = hashlib.sha256(trace_id.encode()) 47 # Get the hexadecimal representation of the hash 48 hash_hex = hash_object.hexdigest() 49 50 # Take the first 8 characters of the hex digest and convert to integer 51 hash_int = int(hash_hex[:8], 16) 52 53 # Normalize the integer to a float in the range [0, 1) 54 normalized_hash = hash_int / 0xFFFFFFFF 55 56 result = normalized_hash < sample_rate 57 58 if not result: 59 log.debug( 60 f"event with trace_id: {trace_id} and rate {sample_rate} was sampled and not sent to the server" 61 ) 62 63 return result
log =
<Logger langfuse (WARNING)>
class
Sampler:
9class Sampler: 10 sample_rate: float 11 12 def __init__(self, sample_rate: float): 13 self.sample_rate = sample_rate 14 15 def sample_event(self, event: dict): 16 # need to get trace_id from a given event 17 # returns true if 18 19 if "type" in event and "body" in event: 20 event_type = event["type"] 21 22 trace_id = None 23 24 if event_type == "trace-create" and "id" in event["body"]: 25 trace_id = event["body"]["id"] 26 elif "trace_id" in event["body"]: 27 trace_id = event["body"]["trace_id"] 28 elif "traceId" in event["body"]: 29 trace_id = event["body"]["traceId"] 30 else: 31 log.error("Unexpected event format: No trace id found in event") 32 return True 33 34 return self.deterministic_sample(trace_id, self.sample_rate) 35 36 else: 37 log.error("Unexpected event format: No trace id found in event") 38 return True 39 40 def deterministic_sample(self, trace_id: str, sample_rate: float): 41 """Determins if an event should be sampled based on the trace_id and sample_rate. Event will be sent to server if True""" 42 log.debug( 43 f"Applying deterministic sampling to trace_id: {trace_id} with rate {sample_rate}" 44 ) 45 46 # Use SHA-256 to hash the trace_id 47 hash_object = hashlib.sha256(trace_id.encode()) 48 # Get the hexadecimal representation of the hash 49 hash_hex = hash_object.hexdigest() 50 51 # Take the first 8 characters of the hex digest and convert to integer 52 hash_int = int(hash_hex[:8], 16) 53 54 # Normalize the integer to a float in the range [0, 1) 55 normalized_hash = hash_int / 0xFFFFFFFF 56 57 result = normalized_hash < sample_rate 58 59 if not result: 60 log.debug( 61 f"event with trace_id: {trace_id} and rate {sample_rate} was sampled and not sent to the server" 62 ) 63 64 return result
def
sample_event(self, event: dict):
15 def sample_event(self, event: dict): 16 # need to get trace_id from a given event 17 # returns true if 18 19 if "type" in event and "body" in event: 20 event_type = event["type"] 21 22 trace_id = None 23 24 if event_type == "trace-create" and "id" in event["body"]: 25 trace_id = event["body"]["id"] 26 elif "trace_id" in event["body"]: 27 trace_id = event["body"]["trace_id"] 28 elif "traceId" in event["body"]: 29 trace_id = event["body"]["traceId"] 30 else: 31 log.error("Unexpected event format: No trace id found in event") 32 return True 33 34 return self.deterministic_sample(trace_id, self.sample_rate) 35 36 else: 37 log.error("Unexpected event format: No trace id found in event") 38 return True
def
deterministic_sample(self, trace_id: str, sample_rate: float):
40 def deterministic_sample(self, trace_id: str, sample_rate: float): 41 """Determins if an event should be sampled based on the trace_id and sample_rate. Event will be sent to server if True""" 42 log.debug( 43 f"Applying deterministic sampling to trace_id: {trace_id} with rate {sample_rate}" 44 ) 45 46 # Use SHA-256 to hash the trace_id 47 hash_object = hashlib.sha256(trace_id.encode()) 48 # Get the hexadecimal representation of the hash 49 hash_hex = hash_object.hexdigest() 50 51 # Take the first 8 characters of the hex digest and convert to integer 52 hash_int = int(hash_hex[:8], 16) 53 54 # Normalize the integer to a float in the range [0, 1) 55 normalized_hash = hash_int / 0xFFFFFFFF 56 57 result = normalized_hash < sample_rate 58 59 if not result: 60 log.debug( 61 f"event with trace_id: {trace_id} and rate {sample_rate} was sampled and not sent to the server" 62 ) 63 64 return result
Determins if an event should be sampled based on the trace_id and sample_rate. Event will be sent to server if True