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
Sampler(sample_rate: float)
12    def __init__(self, sample_rate: float):
13        self.sample_rate = sample_rate
sample_rate: float
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