LangFuse Observability Client (observability/langfuse_client.py)

Singleton async client for LangFuse telemetry, lazily initialized and container‑aware for observability integration.

# Source: https://github.com/mperkins0155/MIRIX | License: Apache-2.0 | Copyright You must retain, in the Source form of any Derivative Works
# Provided via justclaw.it.com; keep this notice with the code.
"""
LangFuse client singleton implementation.

Async-native singleton pattern optimized for containerized deployment.
Each container instance gets its own singleton client.
"""

import asyncio
import atexit
from typing import TYPE_CHECKING, Optional

from mirix.log import get_logger

if TYPE_CHECKING:
    from langfuse import Langfuse

logger = get_logger(__name__)

_init_lock = asyncio.Lock()

_langfuse_client: Optional["Langfuse"] = None
_langfuse_enabled: bool = False
_initialization_attempted: bool = False


async def initialize_langfuse(force: bool = False) -> Optional["Langfuse"]:


... (truncated -- full source via MCP)