langfuse.utils.langfuse_singleton

 1import httpx
 2import threading
 3from typing import Optional
 4
 5
 6from langfuse import Langfuse
 7
 8
 9class LangfuseSingleton:
10    _instance = None
11    _lock = threading.Lock()
12    _langfuse: Optional[Langfuse] = None
13
14    def __new__(cls):
15        if not cls._instance:
16            with cls._lock:
17                if not cls._instance:
18                    cls._instance = super(LangfuseSingleton, cls).__new__(cls)
19        return cls._instance
20
21    def get(
22        self,
23        *,
24        public_key: Optional[str] = None,
25        secret_key: Optional[str] = None,
26        host: Optional[str] = None,
27        release: Optional[str] = None,
28        debug: Optional[bool] = None,
29        threads: Optional[int] = None,
30        flush_at: Optional[int] = None,
31        flush_interval: Optional[int] = None,
32        max_retries: Optional[int] = None,
33        timeout: Optional[int] = None,
34        httpx_client: Optional[httpx.Client] = None,
35        sdk_integration: Optional[str] = None,
36        enabled: Optional[bool] = None,
37        sample_rate: Optional[float] = None,
38    ) -> Langfuse:
39        if self._langfuse:
40            return self._langfuse
41
42        with self._lock:
43            if self._langfuse:
44                return self._langfuse
45
46            langfuse_init_args = {
47                "public_key": public_key,
48                "secret_key": secret_key,
49                "host": host,
50                "release": release,
51                "debug": debug,
52                "threads": threads,
53                "flush_at": flush_at,
54                "flush_interval": flush_interval,
55                "max_retries": max_retries,
56                "timeout": timeout,
57                "httpx_client": httpx_client,
58                "sdk_integration": sdk_integration,
59                "enabled": enabled,
60                "sample_rate": sample_rate,
61            }
62
63            self._langfuse = Langfuse(
64                **{k: v for k, v in langfuse_init_args.items() if v is not None}
65            )
66
67            return self._langfuse
68
69    def reset(self) -> None:
70        with self._lock:
71            if self._langfuse:
72                self._langfuse.flush()
73
74            self._langfuse = None