Package livekit.plugins.posthog

PostHog plugin for LiveKit Agents ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ AI Observability, Latency Waterfalls, Feature Flags & Dual-Channel Multimodal Tools for PostHog.

Sub-modules

livekit.plugins.posthog.feature_flags

PostHog Feature Flags & Dynamic Prompt Experimentation for LiveKit Agents.

livekit.plugins.posthog.log

livekit.plugins.posthog.models

Data contracts, event models, and configuration types for livekit-plugins-posthog.

livekit.plugins.posthog.observer

livekit.plugins.posthog.tools

livekit.plugins.posthog.version

Functions

def observe(session: Any,
api_key: str | None = None,
host: str | None = None,
session_id: str | None = None,
project_id: str | None = None) ‑> PostHogAgentObserver

Expand source code

def observe(
    session: Any,
    api_key: Optional[str] = None,
    host: Optional[str] = None,
    session_id: Optional[str] = None,
    project_id: Optional[str] = None,
) -> PostHogAgentObserver:
    """
    Convenience helper to initialize a PostHogAgentObserver and immediately attach it
    to an active LiveKit AgentSession or VoicePipelineAgent.
    """
    observer = PostHogAgentObserver(
        api_key=api_key,
        host=host,
        session_id=session_id,
        project_id=project_id,
    )
    observer.attach_to_session(session)
    return observer

Convenience helper to initialize a PostHogAgentObserver and immediately attach it to an active LiveKit AgentSession or VoicePipelineAgent.

Classes

class FunnelStep (name: str, visitors: int, conversion_rate: float)

Expand source code

@dataclass
class FunnelStep:
    """A discrete step in a PostHog conversion funnel."""
    name: str
    visitors: int
    conversion_rate: float

A discrete step in a PostHog conversion funnel.

Instance variables

var conversion_rate : float

The type of the None singleton.

var name : str

The type of the None singleton.

var visitors : int

The type of the None singleton.

class LatencyWaterfallEvent (session_id: str,
stt_duration: float = 0.0,
eou_delay: float = 0.0,
llm_ttft: float = 0.0,
llm_duration: float = 0.0,
tts_ttfb: float = 0.0,
tts_duration: float = 0.0,
total_turn_latency: float = 0.0)

Expand source code

@dataclass
class LatencyWaterfallEvent:
    """Full turn latency waterfall emitted to PostHog for conversational performance analysis."""
    session_id: str
    stt_duration: float = 0.0
    eou_delay: float = 0.0
    llm_ttft: float = 0.0
    llm_duration: float = 0.0
    tts_ttfb: float = 0.0
    tts_duration: float = 0.0
    total_turn_latency: float = 0.0

Full turn latency waterfall emitted to PostHog for conversational performance analysis.

Instance variables

var eou_delay : float

The type of the None singleton.

var llm_duration : float

The type of the None singleton.

var llm_ttft : float

The type of the None singleton.

var session_id : str

The type of the None singleton.

var stt_duration : float

The type of the None singleton.

var total_turn_latency : float

The type of the None singleton.

var tts_duration : float

The type of the None singleton.

var tts_ttfb : float

The type of the None singleton.

class Observer (api_key: str | None = None,
host: str | None = None,
session_id: str | None = None,
project_id: str | None = None)

Expand source code

class PostHogAgentObserver:
    """
    Subscribes to LiveKit Agent metrics and emits standardized PostHog events
    for AI observability ($ai_generation), conversational frustration (interruptions),
    voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: Optional[str] = None,
        session_id: Optional[str] = None,
        project_id: Optional[str] = None,
    ):
        self.api_key = api_key or os.getenv("POSTHOG_API_KEY", "phc_mock_key_prism")
        self.host = host or os.getenv("POSTHOG_HOST", "https://us.i.posthog.com")
        self.session_id = session_id or f"RM_prism_{int(time.time())}"
        self.project_id = project_id or os.getenv("POSTHOG_PROJECT_ID", "default_proj")

        # Initialize PostHog client (flush_at=1 for real-time streaming telemetry)
        try:
            self.client = posthog.Posthog(
                project_api_key=self.api_key,
                host=self.host,
                flush_at=1,
                flush_interval=0.5,
            )
            logger.info("Initialized PostHogAgentObserver with session %s", self.session_id)
        except Exception as e:
            logger.warning("Failed to initialize PostHog client, logging locally: %s", e)
            self.client = None

        # Feature flags helper
        self.flags = PostHogFeatureFlags(client=self.client, default_distinct_id=self.session_id)

    def attach_to_session(self, session: Any):
        """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
        if hasattr(session, "on"):
            session.on("metrics_collected", self.on_metrics_collected)
            logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

    def link_session_replay(
        self,
        livekit_room: str,
        participant_identity: str,
        replay_id: Optional[str] = None,
    ) -> str:
        """
        Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
        Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
        """
        rep_id = replay_id or self.session_id
        replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

        properties = {
            "$session_id": self.session_id,
            "livekit_room": livekit_room,
            "participant_identity": participant_identity,
            "replay_url": replay_url,
            "webrtc_stream_linked": True,
            "timestamp": time.time(),
        }

        logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$session_replay_link",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to emit $session_replay_link event: %s", e)

        return replay_url

    def on_metrics_collected(self, metrics: Any):
        """Dispatches collected metrics into PostHog telemetry."""
        metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

        if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
            self.record_llm_metrics(metrics)
        elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
            self.record_interruption_metrics(metrics)
        elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
            self.record_tts_metrics(metrics)
        elif "STT" in metrics_type:
            self.record_stt_metrics(metrics)
        elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
            self.record_eou_metrics(metrics)
        elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
            self.record_session_usage(metrics)
        else:
            logger.debug("Captured general metric: %s", metrics_type)

    def record_llm_metrics(self, m: Any):
        """
        Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
        """
        input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
        output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
        duration = getattr(m, "duration", getattr(m, "latency", 0.0))
        ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
        model = getattr(m, "model", "gpt-4o-mini")
        cancelled = getattr(m, "cancelled", False)

        tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

        properties = {
            "$session_id": self.session_id,
            "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
            "$ai_model": model,
            "$ai_provider": "openai",
            "$ai_input_tokens": input_tokens,
            "$ai_output_tokens": output_tokens,
            "$ai_latency": round(duration, 4),
            "$ai_time_to_first_token": round(ttft, 4),
            "tokens_per_second": round(tps, 2),
            "cancelled": cancelled,
        }

        logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                    input_tokens, output_tokens, duration)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$ai_generation",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture $ai_generation event: %s", e)

    def record_interruption_metrics(self, m: Any):
        """Records conversational interruption / user frustration event."""
        properties = {
            "$session_id": self.session_id,
            "timestamp": time.time(),
            "speech_duration_before_cut": getattr(m, "duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_interrupted",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_interrupted event: %s", e)

    def record_tts_metrics(self, m: Any):
        """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
        properties = {
            "$session_id": self.session_id,
            "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
            "tts_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "characters_count": getattr(m, "characters_count", 0),
            "tts_model": getattr(m, "label", "tts"),
            "cancelled": getattr(m, "cancelled", False),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                    properties["tts_ttfb"], properties["tts_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_tts_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

    def record_stt_metrics(self, m: Any):
        """Records Speech-to-Text transcription latency metrics."""
        properties = {
            "$session_id": self.session_id,
            "stt_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "stt_model": getattr(m, "label", "stt"),
            "streamed": getattr(m, "streamed", True),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                    properties["stt_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_stt_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

    def record_eou_metrics(self, m: Any):
        """Records End-of-Utterance (endpointing) conversational turn transition delay."""
        properties = {
            "$session_id": self.session_id,
            "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
            "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
            "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                    properties["eou_delay"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_endpointing_delay",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

    def record_session_usage(self, m: Any):
        """Records cumulative session resource and token usage."""
        properties = {
            "$session_id": self.session_id,
            "total_tokens": getattr(m, "total_tokens", 0),
            "input_tokens": getattr(m, "input_tokens", 0),
            "output_tokens": getattr(m, "output_tokens", 0),
            "audio_duration": getattr(m, "audio_duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_session_completed")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_session_completed",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_session_completed event: %s", e)

    def shutdown(self):
        """Flushes remaining events to PostHog."""
        if self.client:
            try:
                self.client.flush()
            except Exception:
                pass

Subscribes to LiveKit Agent metrics and emits standardized PostHog events for AI observability ($ai_generation), conversational frustration (interruptions), voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.

Methods

def attach_to_session(self, session: Any)

Expand source code

def attach_to_session(self, session: Any):
    """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
    if hasattr(session, "on"):
        session.on("metrics_collected", self.on_metrics_collected)
        logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent.

def link_session_replay(self, livekit_room: str, participant_identity: str, replay_id: str | None = None) ‑> str

Expand source code

def link_session_replay(
    self,
    livekit_room: str,
    participant_identity: str,
    replay_id: Optional[str] = None,
) -> str:
    """
    Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
    Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
    """
    rep_id = replay_id or self.session_id
    replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

    properties = {
        "$session_id": self.session_id,
        "livekit_room": livekit_room,
        "participant_identity": participant_identity,
        "replay_url": replay_url,
        "webrtc_stream_linked": True,
        "timestamp": time.time(),
    }

    logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$session_replay_link",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to emit $session_replay_link event: %s", e)

    return replay_url

Cross-links the LiveKit WebRTC voice session with PostHog Session Replay. Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.

def on_metrics_collected(self, metrics: Any)

Expand source code

def on_metrics_collected(self, metrics: Any):
    """Dispatches collected metrics into PostHog telemetry."""
    metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

    if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
        self.record_llm_metrics(metrics)
    elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
        self.record_interruption_metrics(metrics)
    elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
        self.record_tts_metrics(metrics)
    elif "STT" in metrics_type:
        self.record_stt_metrics(metrics)
    elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
        self.record_eou_metrics(metrics)
    elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
        self.record_session_usage(metrics)
    else:
        logger.debug("Captured general metric: %s", metrics_type)

Dispatches collected metrics into PostHog telemetry.

def record_eou_metrics(self, m: Any)

Expand source code

def record_eou_metrics(self, m: Any):
    """Records End-of-Utterance (endpointing) conversational turn transition delay."""
    properties = {
        "$session_id": self.session_id,
        "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
        "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
        "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                properties["eou_delay"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_endpointing_delay",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

Records End-of-Utterance (endpointing) conversational turn transition delay.

def record_interruption_metrics(self, m: Any)

Expand source code

def record_interruption_metrics(self, m: Any):
    """Records conversational interruption / user frustration event."""
    properties = {
        "$session_id": self.session_id,
        "timestamp": time.time(),
        "speech_duration_before_cut": getattr(m, "duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_interrupted",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_interrupted event: %s", e)

Records conversational interruption / user frustration event.

def record_llm_metrics(self, m: Any)

Expand source code

def record_llm_metrics(self, m: Any):
    """
    Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
    """
    input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
    output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
    duration = getattr(m, "duration", getattr(m, "latency", 0.0))
    ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
    model = getattr(m, "model", "gpt-4o-mini")
    cancelled = getattr(m, "cancelled", False)

    tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

    properties = {
        "$session_id": self.session_id,
        "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
        "$ai_model": model,
        "$ai_provider": "openai",
        "$ai_input_tokens": input_tokens,
        "$ai_output_tokens": output_tokens,
        "$ai_latency": round(duration, 4),
        "$ai_time_to_first_token": round(ttft, 4),
        "tokens_per_second": round(tps, 2),
        "cancelled": cancelled,
    }

    logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                input_tokens, output_tokens, duration)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$ai_generation",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture $ai_generation event: %s", e)

Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).

def record_session_usage(self, m: Any)

Expand source code

def record_session_usage(self, m: Any):
    """Records cumulative session resource and token usage."""
    properties = {
        "$session_id": self.session_id,
        "total_tokens": getattr(m, "total_tokens", 0),
        "input_tokens": getattr(m, "input_tokens", 0),
        "output_tokens": getattr(m, "output_tokens", 0),
        "audio_duration": getattr(m, "audio_duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_session_completed")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_session_completed",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_session_completed event: %s", e)

Records cumulative session resource and token usage.

def record_stt_metrics(self, m: Any)

Expand source code

def record_stt_metrics(self, m: Any):
    """Records Speech-to-Text transcription latency metrics."""
    properties = {
        "$session_id": self.session_id,
        "stt_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "stt_model": getattr(m, "label", "stt"),
        "streamed": getattr(m, "streamed", True),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                properties["stt_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_stt_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

Records Speech-to-Text transcription latency metrics.

def record_tts_metrics(self, m: Any)

Expand source code

def record_tts_metrics(self, m: Any):
    """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
    properties = {
        "$session_id": self.session_id,
        "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
        "tts_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "characters_count": getattr(m, "characters_count", 0),
        "tts_model": getattr(m, "label", "tts"),
        "cancelled": getattr(m, "cancelled", False),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                properties["tts_ttfb"], properties["tts_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_tts_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration).

def shutdown(self)

Expand source code

def shutdown(self):
    """Flushes remaining events to PostHog."""
    if self.client:
        try:
            self.client.flush()
        except Exception:
            pass

Flushes remaining events to PostHog.

class AgentObserver (api_key: str | None = None,
host: str | None = None,
session_id: str | None = None,
project_id: str | None = None)

Expand source code

class PostHogAgentObserver:
    """
    Subscribes to LiveKit Agent metrics and emits standardized PostHog events
    for AI observability ($ai_generation), conversational frustration (interruptions),
    voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: Optional[str] = None,
        session_id: Optional[str] = None,
        project_id: Optional[str] = None,
    ):
        self.api_key = api_key or os.getenv("POSTHOG_API_KEY", "phc_mock_key_prism")
        self.host = host or os.getenv("POSTHOG_HOST", "https://us.i.posthog.com")
        self.session_id = session_id or f"RM_prism_{int(time.time())}"
        self.project_id = project_id or os.getenv("POSTHOG_PROJECT_ID", "default_proj")

        # Initialize PostHog client (flush_at=1 for real-time streaming telemetry)
        try:
            self.client = posthog.Posthog(
                project_api_key=self.api_key,
                host=self.host,
                flush_at=1,
                flush_interval=0.5,
            )
            logger.info("Initialized PostHogAgentObserver with session %s", self.session_id)
        except Exception as e:
            logger.warning("Failed to initialize PostHog client, logging locally: %s", e)
            self.client = None

        # Feature flags helper
        self.flags = PostHogFeatureFlags(client=self.client, default_distinct_id=self.session_id)

    def attach_to_session(self, session: Any):
        """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
        if hasattr(session, "on"):
            session.on("metrics_collected", self.on_metrics_collected)
            logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

    def link_session_replay(
        self,
        livekit_room: str,
        participant_identity: str,
        replay_id: Optional[str] = None,
    ) -> str:
        """
        Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
        Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
        """
        rep_id = replay_id or self.session_id
        replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

        properties = {
            "$session_id": self.session_id,
            "livekit_room": livekit_room,
            "participant_identity": participant_identity,
            "replay_url": replay_url,
            "webrtc_stream_linked": True,
            "timestamp": time.time(),
        }

        logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$session_replay_link",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to emit $session_replay_link event: %s", e)

        return replay_url

    def on_metrics_collected(self, metrics: Any):
        """Dispatches collected metrics into PostHog telemetry."""
        metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

        if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
            self.record_llm_metrics(metrics)
        elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
            self.record_interruption_metrics(metrics)
        elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
            self.record_tts_metrics(metrics)
        elif "STT" in metrics_type:
            self.record_stt_metrics(metrics)
        elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
            self.record_eou_metrics(metrics)
        elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
            self.record_session_usage(metrics)
        else:
            logger.debug("Captured general metric: %s", metrics_type)

    def record_llm_metrics(self, m: Any):
        """
        Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
        """
        input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
        output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
        duration = getattr(m, "duration", getattr(m, "latency", 0.0))
        ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
        model = getattr(m, "model", "gpt-4o-mini")
        cancelled = getattr(m, "cancelled", False)

        tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

        properties = {
            "$session_id": self.session_id,
            "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
            "$ai_model": model,
            "$ai_provider": "openai",
            "$ai_input_tokens": input_tokens,
            "$ai_output_tokens": output_tokens,
            "$ai_latency": round(duration, 4),
            "$ai_time_to_first_token": round(ttft, 4),
            "tokens_per_second": round(tps, 2),
            "cancelled": cancelled,
        }

        logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                    input_tokens, output_tokens, duration)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$ai_generation",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture $ai_generation event: %s", e)

    def record_interruption_metrics(self, m: Any):
        """Records conversational interruption / user frustration event."""
        properties = {
            "$session_id": self.session_id,
            "timestamp": time.time(),
            "speech_duration_before_cut": getattr(m, "duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_interrupted",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_interrupted event: %s", e)

    def record_tts_metrics(self, m: Any):
        """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
        properties = {
            "$session_id": self.session_id,
            "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
            "tts_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "characters_count": getattr(m, "characters_count", 0),
            "tts_model": getattr(m, "label", "tts"),
            "cancelled": getattr(m, "cancelled", False),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                    properties["tts_ttfb"], properties["tts_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_tts_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

    def record_stt_metrics(self, m: Any):
        """Records Speech-to-Text transcription latency metrics."""
        properties = {
            "$session_id": self.session_id,
            "stt_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "stt_model": getattr(m, "label", "stt"),
            "streamed": getattr(m, "streamed", True),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                    properties["stt_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_stt_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

    def record_eou_metrics(self, m: Any):
        """Records End-of-Utterance (endpointing) conversational turn transition delay."""
        properties = {
            "$session_id": self.session_id,
            "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
            "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
            "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                    properties["eou_delay"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_endpointing_delay",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

    def record_session_usage(self, m: Any):
        """Records cumulative session resource and token usage."""
        properties = {
            "$session_id": self.session_id,
            "total_tokens": getattr(m, "total_tokens", 0),
            "input_tokens": getattr(m, "input_tokens", 0),
            "output_tokens": getattr(m, "output_tokens", 0),
            "audio_duration": getattr(m, "audio_duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_session_completed")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_session_completed",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_session_completed event: %s", e)

    def shutdown(self):
        """Flushes remaining events to PostHog."""
        if self.client:
            try:
                self.client.flush()
            except Exception:
                pass

Subscribes to LiveKit Agent metrics and emits standardized PostHog events for AI observability ($ai_generation), conversational frustration (interruptions), voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.

Methods

def attach_to_session(self, session: Any)

Expand source code

def attach_to_session(self, session: Any):
    """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
    if hasattr(session, "on"):
        session.on("metrics_collected", self.on_metrics_collected)
        logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent.

def link_session_replay(self, livekit_room: str, participant_identity: str, replay_id: str | None = None) ‑> str

Expand source code

def link_session_replay(
    self,
    livekit_room: str,
    participant_identity: str,
    replay_id: Optional[str] = None,
) -> str:
    """
    Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
    Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
    """
    rep_id = replay_id or self.session_id
    replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

    properties = {
        "$session_id": self.session_id,
        "livekit_room": livekit_room,
        "participant_identity": participant_identity,
        "replay_url": replay_url,
        "webrtc_stream_linked": True,
        "timestamp": time.time(),
    }

    logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$session_replay_link",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to emit $session_replay_link event: %s", e)

    return replay_url

Cross-links the LiveKit WebRTC voice session with PostHog Session Replay. Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.

def on_metrics_collected(self, metrics: Any)

Expand source code

def on_metrics_collected(self, metrics: Any):
    """Dispatches collected metrics into PostHog telemetry."""
    metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

    if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
        self.record_llm_metrics(metrics)
    elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
        self.record_interruption_metrics(metrics)
    elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
        self.record_tts_metrics(metrics)
    elif "STT" in metrics_type:
        self.record_stt_metrics(metrics)
    elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
        self.record_eou_metrics(metrics)
    elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
        self.record_session_usage(metrics)
    else:
        logger.debug("Captured general metric: %s", metrics_type)

Dispatches collected metrics into PostHog telemetry.

def record_eou_metrics(self, m: Any)

Expand source code

def record_eou_metrics(self, m: Any):
    """Records End-of-Utterance (endpointing) conversational turn transition delay."""
    properties = {
        "$session_id": self.session_id,
        "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
        "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
        "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                properties["eou_delay"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_endpointing_delay",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

Records End-of-Utterance (endpointing) conversational turn transition delay.

def record_interruption_metrics(self, m: Any)

Expand source code

def record_interruption_metrics(self, m: Any):
    """Records conversational interruption / user frustration event."""
    properties = {
        "$session_id": self.session_id,
        "timestamp": time.time(),
        "speech_duration_before_cut": getattr(m, "duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_interrupted",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_interrupted event: %s", e)

Records conversational interruption / user frustration event.

def record_llm_metrics(self, m: Any)

Expand source code

def record_llm_metrics(self, m: Any):
    """
    Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
    """
    input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
    output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
    duration = getattr(m, "duration", getattr(m, "latency", 0.0))
    ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
    model = getattr(m, "model", "gpt-4o-mini")
    cancelled = getattr(m, "cancelled", False)

    tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

    properties = {
        "$session_id": self.session_id,
        "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
        "$ai_model": model,
        "$ai_provider": "openai",
        "$ai_input_tokens": input_tokens,
        "$ai_output_tokens": output_tokens,
        "$ai_latency": round(duration, 4),
        "$ai_time_to_first_token": round(ttft, 4),
        "tokens_per_second": round(tps, 2),
        "cancelled": cancelled,
    }

    logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                input_tokens, output_tokens, duration)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$ai_generation",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture $ai_generation event: %s", e)

Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).

def record_session_usage(self, m: Any)

Expand source code

def record_session_usage(self, m: Any):
    """Records cumulative session resource and token usage."""
    properties = {
        "$session_id": self.session_id,
        "total_tokens": getattr(m, "total_tokens", 0),
        "input_tokens": getattr(m, "input_tokens", 0),
        "output_tokens": getattr(m, "output_tokens", 0),
        "audio_duration": getattr(m, "audio_duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_session_completed")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_session_completed",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_session_completed event: %s", e)

Records cumulative session resource and token usage.

def record_stt_metrics(self, m: Any)

Expand source code

def record_stt_metrics(self, m: Any):
    """Records Speech-to-Text transcription latency metrics."""
    properties = {
        "$session_id": self.session_id,
        "stt_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "stt_model": getattr(m, "label", "stt"),
        "streamed": getattr(m, "streamed", True),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                properties["stt_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_stt_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

Records Speech-to-Text transcription latency metrics.

def record_tts_metrics(self, m: Any)

Expand source code

def record_tts_metrics(self, m: Any):
    """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
    properties = {
        "$session_id": self.session_id,
        "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
        "tts_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "characters_count": getattr(m, "characters_count", 0),
        "tts_model": getattr(m, "label", "tts"),
        "cancelled": getattr(m, "cancelled", False),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                properties["tts_ttfb"], properties["tts_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_tts_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration).

def shutdown(self)

Expand source code

def shutdown(self):
    """Flushes remaining events to PostHog."""
    if self.client:
        try:
            self.client.flush()
        except Exception:
            pass

Flushes remaining events to PostHog.

class PostHogAgentObserver (api_key: str | None = None,
host: str | None = None,
session_id: str | None = None,
project_id: str | None = None)

Expand source code

class PostHogAgentObserver:
    """
    Subscribes to LiveKit Agent metrics and emits standardized PostHog events
    for AI observability ($ai_generation), conversational frustration (interruptions),
    voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: Optional[str] = None,
        session_id: Optional[str] = None,
        project_id: Optional[str] = None,
    ):
        self.api_key = api_key or os.getenv("POSTHOG_API_KEY", "phc_mock_key_prism")
        self.host = host or os.getenv("POSTHOG_HOST", "https://us.i.posthog.com")
        self.session_id = session_id or f"RM_prism_{int(time.time())}"
        self.project_id = project_id or os.getenv("POSTHOG_PROJECT_ID", "default_proj")

        # Initialize PostHog client (flush_at=1 for real-time streaming telemetry)
        try:
            self.client = posthog.Posthog(
                project_api_key=self.api_key,
                host=self.host,
                flush_at=1,
                flush_interval=0.5,
            )
            logger.info("Initialized PostHogAgentObserver with session %s", self.session_id)
        except Exception as e:
            logger.warning("Failed to initialize PostHog client, logging locally: %s", e)
            self.client = None

        # Feature flags helper
        self.flags = PostHogFeatureFlags(client=self.client, default_distinct_id=self.session_id)

    def attach_to_session(self, session: Any):
        """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
        if hasattr(session, "on"):
            session.on("metrics_collected", self.on_metrics_collected)
            logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

    def link_session_replay(
        self,
        livekit_room: str,
        participant_identity: str,
        replay_id: Optional[str] = None,
    ) -> str:
        """
        Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
        Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
        """
        rep_id = replay_id or self.session_id
        replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

        properties = {
            "$session_id": self.session_id,
            "livekit_room": livekit_room,
            "participant_identity": participant_identity,
            "replay_url": replay_url,
            "webrtc_stream_linked": True,
            "timestamp": time.time(),
        }

        logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$session_replay_link",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to emit $session_replay_link event: %s", e)

        return replay_url

    def on_metrics_collected(self, metrics: Any):
        """Dispatches collected metrics into PostHog telemetry."""
        metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

        if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
            self.record_llm_metrics(metrics)
        elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
            self.record_interruption_metrics(metrics)
        elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
            self.record_tts_metrics(metrics)
        elif "STT" in metrics_type:
            self.record_stt_metrics(metrics)
        elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
            self.record_eou_metrics(metrics)
        elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
            self.record_session_usage(metrics)
        else:
            logger.debug("Captured general metric: %s", metrics_type)

    def record_llm_metrics(self, m: Any):
        """
        Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
        """
        input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
        output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
        duration = getattr(m, "duration", getattr(m, "latency", 0.0))
        ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
        model = getattr(m, "model", "gpt-4o-mini")
        cancelled = getattr(m, "cancelled", False)

        tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

        properties = {
            "$session_id": self.session_id,
            "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
            "$ai_model": model,
            "$ai_provider": "openai",
            "$ai_input_tokens": input_tokens,
            "$ai_output_tokens": output_tokens,
            "$ai_latency": round(duration, 4),
            "$ai_time_to_first_token": round(ttft, 4),
            "tokens_per_second": round(tps, 2),
            "cancelled": cancelled,
        }

        logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                    input_tokens, output_tokens, duration)

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="$ai_generation",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture $ai_generation event: %s", e)

    def record_interruption_metrics(self, m: Any):
        """Records conversational interruption / user frustration event."""
        properties = {
            "$session_id": self.session_id,
            "timestamp": time.time(),
            "speech_duration_before_cut": getattr(m, "duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_interrupted",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_interrupted event: %s", e)

    def record_tts_metrics(self, m: Any):
        """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
        properties = {
            "$session_id": self.session_id,
            "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
            "tts_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "characters_count": getattr(m, "characters_count", 0),
            "tts_model": getattr(m, "label", "tts"),
            "cancelled": getattr(m, "cancelled", False),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                    properties["tts_ttfb"], properties["tts_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_tts_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

    def record_stt_metrics(self, m: Any):
        """Records Speech-to-Text transcription latency metrics."""
        properties = {
            "$session_id": self.session_id,
            "stt_duration": round(getattr(m, "duration", 0.0), 4),
            "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
            "stt_model": getattr(m, "label", "stt"),
            "streamed": getattr(m, "streamed", True),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                    properties["stt_duration"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_stt_latency",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

    def record_eou_metrics(self, m: Any):
        """Records End-of-Utterance (endpointing) conversational turn transition delay."""
        properties = {
            "$session_id": self.session_id,
            "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
            "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
            "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
        }

        logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                    properties["eou_delay"])

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_agent_endpointing_delay",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

    def record_session_usage(self, m: Any):
        """Records cumulative session resource and token usage."""
        properties = {
            "$session_id": self.session_id,
            "total_tokens": getattr(m, "total_tokens", 0),
            "input_tokens": getattr(m, "input_tokens", 0),
            "output_tokens": getattr(m, "output_tokens", 0),
            "audio_duration": getattr(m, "audio_duration", 0.0),
        }

        logger.info("[PostHog Telemetry] Emitting voice_session_completed")

        if self.client:
            try:
                self.client.capture(
                    distinct_id=self.session_id,
                    event="voice_session_completed",
                    properties=properties,
                )
            except Exception as e:
                logger.error("Failed to capture voice_session_completed event: %s", e)

    def shutdown(self):
        """Flushes remaining events to PostHog."""
        if self.client:
            try:
                self.client.flush()
            except Exception:
                pass

Subscribes to LiveKit Agent metrics and emits standardized PostHog events for AI observability ($ai_generation), conversational frustration (interruptions), voice latency waterfalls (TTS, STT, EOU), Session Replay cross-linking, and session costs.

Methods

def attach_to_session(self, session: Any)

Expand source code

def attach_to_session(self, session: Any):
    """Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent."""
    if hasattr(session, "on"):
        session.on("metrics_collected", self.on_metrics_collected)
        logger.info("Attached PostHogAgentObserver to session.on('metrics_collected')")

Attaches metrics listener to a LiveKit AgentSession or VoicePipelineAgent.

def link_session_replay(self, livekit_room: str, participant_identity: str, replay_id: str | None = None) ‑> str

Expand source code

def link_session_replay(
    self,
    livekit_room: str,
    participant_identity: str,
    replay_id: Optional[str] = None,
) -> str:
    """
    Cross-links the LiveKit WebRTC voice session with PostHog Session Replay.
    Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.
    """
    rep_id = replay_id or self.session_id
    replay_url = f"{self.host.rstrip('/')}/project/{self.project_id}/replay/{rep_id}"

    properties = {
        "$session_id": self.session_id,
        "livekit_room": livekit_room,
        "participant_identity": participant_identity,
        "replay_url": replay_url,
        "webrtc_stream_linked": True,
        "timestamp": time.time(),
    }

    logger.info("[PostHog SessionReplay] Linked LiveKit room '%s' to replay: %s", livekit_room, replay_url)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$session_replay_link",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to emit $session_replay_link event: %s", e)

    return replay_url

Cross-links the LiveKit WebRTC voice session with PostHog Session Replay. Emits a $session_replay_link event so the voice recording is indexed alongside DOM replays.

def on_metrics_collected(self, metrics: Any)

Expand source code

def on_metrics_collected(self, metrics: Any):
    """Dispatches collected metrics into PostHog telemetry."""
    metrics_type = getattr(metrics, "type", metrics.__class__.__name__)

    if "LLM" in metrics_type or hasattr(metrics, "input_tokens"):
        self.record_llm_metrics(metrics)
    elif "Interruption" in metrics_type or hasattr(metrics, "interrupted"):
        self.record_interruption_metrics(metrics)
    elif "TTS" in metrics_type or hasattr(metrics, "ttfb"):
        self.record_tts_metrics(metrics)
    elif "STT" in metrics_type:
        self.record_stt_metrics(metrics)
    elif "EOU" in metrics_type or hasattr(metrics, "end_of_utterance_delay"):
        self.record_eou_metrics(metrics)
    elif "Usage" in metrics_type or hasattr(metrics, "total_tokens"):
        self.record_session_usage(metrics)
    else:
        logger.debug("Captured general metric: %s", metrics_type)

Dispatches collected metrics into PostHog telemetry.

def record_eou_metrics(self, m: Any)

Expand source code

def record_eou_metrics(self, m: Any):
    """Records End-of-Utterance (endpointing) conversational turn transition delay."""
    properties = {
        "$session_id": self.session_id,
        "eou_delay": round(getattr(m, "end_of_utterance_delay", 0.0), 4),
        "transcription_delay": round(getattr(m, "transcription_delay", 0.0), 4),
        "turn_completed_delay": round(getattr(m, "on_user_turn_completed_delay", 0.0), 4),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_endpointing_delay: eou_delay=%.3fs",
                properties["eou_delay"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_endpointing_delay",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_endpointing_delay event: %s", e)

Records End-of-Utterance (endpointing) conversational turn transition delay.

def record_interruption_metrics(self, m: Any)

Expand source code

def record_interruption_metrics(self, m: Any):
    """Records conversational interruption / user frustration event."""
    properties = {
        "$session_id": self.session_id,
        "timestamp": time.time(),
        "speech_duration_before_cut": getattr(m, "duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_interrupted")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_interrupted",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_interrupted event: %s", e)

Records conversational interruption / user frustration event.

def record_llm_metrics(self, m: Any)

Expand source code

def record_llm_metrics(self, m: Any):
    """
    Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).
    """
    input_tokens = getattr(m, "input_tokens", getattr(m, "prompt_tokens", 0))
    output_tokens = getattr(m, "output_tokens", getattr(m, "completion_tokens", 0))
    duration = getattr(m, "duration", getattr(m, "latency", 0.0))
    ttft = getattr(m, "time_to_first_token", getattr(m, "ttft", 0.0))
    model = getattr(m, "model", "gpt-4o-mini")
    cancelled = getattr(m, "cancelled", False)

    tps = (output_tokens / duration) if (duration > 0 and output_tokens > 0) else 0.0

    properties = {
        "$session_id": self.session_id,
        "$ai_trace_id": getattr(m, "request_id", f"req_{int(time.time()*1000)}"),
        "$ai_model": model,
        "$ai_provider": "openai",
        "$ai_input_tokens": input_tokens,
        "$ai_output_tokens": output_tokens,
        "$ai_latency": round(duration, 4),
        "$ai_time_to_first_token": round(ttft, 4),
        "tokens_per_second": round(tps, 2),
        "cancelled": cancelled,
    }

    logger.info("[PostHog Telemetry] Emitting $ai_generation: tokens=(in:%d, out:%d) latency=%.2fs",
                input_tokens, output_tokens, duration)

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="$ai_generation",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture $ai_generation event: %s", e)

Conforms strictly to Contract C: PostHog AI Observability Schema ($ai_generation).

def record_session_usage(self, m: Any)

Expand source code

def record_session_usage(self, m: Any):
    """Records cumulative session resource and token usage."""
    properties = {
        "$session_id": self.session_id,
        "total_tokens": getattr(m, "total_tokens", 0),
        "input_tokens": getattr(m, "input_tokens", 0),
        "output_tokens": getattr(m, "output_tokens", 0),
        "audio_duration": getattr(m, "audio_duration", 0.0),
    }

    logger.info("[PostHog Telemetry] Emitting voice_session_completed")

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_session_completed",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_session_completed event: %s", e)

Records cumulative session resource and token usage.

def record_stt_metrics(self, m: Any)

Expand source code

def record_stt_metrics(self, m: Any):
    """Records Speech-to-Text transcription latency metrics."""
    properties = {
        "$session_id": self.session_id,
        "stt_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "stt_model": getattr(m, "label", "stt"),
        "streamed": getattr(m, "streamed", True),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_stt_latency: duration=%.3fs",
                properties["stt_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_stt_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_stt_latency event: %s", e)

Records Speech-to-Text transcription latency metrics.

def record_tts_metrics(self, m: Any)

Expand source code

def record_tts_metrics(self, m: Any):
    """Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration)."""
    properties = {
        "$session_id": self.session_id,
        "tts_ttfb": round(getattr(m, "ttfb", 0.0), 4),
        "tts_duration": round(getattr(m, "duration", 0.0), 4),
        "audio_duration": round(getattr(m, "audio_duration", 0.0), 4),
        "characters_count": getattr(m, "characters_count", 0),
        "tts_model": getattr(m, "label", "tts"),
        "cancelled": getattr(m, "cancelled", False),
    }

    logger.info("[PostHog Telemetry] Emitting voice_agent_tts_latency: ttfb=%.3fs duration=%.3fs",
                properties["tts_ttfb"], properties["tts_duration"])

    if self.client:
        try:
            self.client.capture(
                distinct_id=self.session_id,
                event="voice_agent_tts_latency",
                properties=properties,
            )
        except Exception as e:
            logger.error("Failed to capture voice_agent_tts_latency event: %s", e)

Records Text-to-Speech synthesis latency waterfall metrics (TTFB, duration).

def shutdown(self)

Expand source code

def shutdown(self):
    """Flushes remaining events to PostHog."""
    if self.client:
        try:
            self.client.flush()
        except Exception:
            pass

Flushes remaining events to PostHog.

class Tools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)

Expand source code

class PostHogDualChannelTools:
    """
    Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine.
    Splits analytical query responses into:
      1. High-throughput WebRTC data-channel visualization packet.
      2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
    """

    def __init__(
        self,
        gateway_url: str = DEFAULT_GATEWAY_URL,
        room_name: str = "prism-demo",
        livekit_room: Optional[Any] = None,
    ):
        self.gateway_url = gateway_url.rstrip("/")
        self.room_name = room_name
        self.livekit_room = livekit_room
        self.http_client = httpx.Client(timeout=3.0)

        # Wrap tools with LiveKit function_tool
        self.query_funnel_tool = function_tool(self._query_funnel_impl)
        self.query_errors_tool = function_tool(self._query_errors_impl)
        self.query_retention_tool = function_tool(self._query_retention_impl)
        self.query_hogql_tool = function_tool(self._query_hogql_impl)

    def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str:
        """
        Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup).
        Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser,
        and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_funnel",
            "args": {"funnel_id": funnel_id},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload)
            if resp.status_code == 200:
                data = resp.json()
                return data.get(
                    "speech_ready_delta",
                    "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        # Local fallback if Go Gateway is offline
        mock_card = {
            "type": "funnel_viz",
            "title": "Checkout Conversion Dropoff",
            "funnel_id": funnel_id,
            "steps": [
                {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0},
                {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1},
                {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5},
            ],
            "primary_drop_step": "Payment Confirmed",
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen."

    def _query_errors_impl(self, timeframe: str = "24h") -> str:
        """
        Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d).
        Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_errors",
            "args": {"timeframe": timeframe},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "error_waterfall",
            "title": f"API Error Waterfall ({timeframe})",
            "timeframe": timeframe,
            "total_errors": 42,
            "primary_root_cause": "PaymentGatewayTimeout (504)",
            "errors": [
                {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504},
                {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400},
                {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen."

    def _query_retention_impl(self, cohort: str = "weekly") -> str:
        """
        Query PostHog cohort retention trends over an 8-week horizon.
        Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_retention",
            "args": {"cohort": cohort},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "retention_trend",
            "title": "8-Week Cohort Retention Trend",
            "cohort_name": "Weekly Active Users",
            "terminal_retention": 19.5,
            "intervals": [
                {"interval": "Week 1", "retention_rate": 100.0, "users": 8400},
                {"interval": "Week 2", "retention_rate": 44.2, "users": 3712},
                {"interval": "Week 3", "retention_rate": 34.0, "users": 2856},
                {"interval": "Week 4", "retention_rate": 28.1, "users": 2360},
                {"interval": "Week 6", "retention_rate": 23.4, "users": 1965},
                {"interval": "Week 8", "retention_rate": 19.5, "users": 1638},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen."

    def _query_hogql_impl(self, query: str) -> str:
        """
        Execute an analytical HogQL query against PostHog and ClickHouse.
        Renders the analytical results card over WebRTC and returns a concise verbal summary.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_hogql",
            "args": {"query": query},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your HogQL query completed successfully. I've projected the analytical results onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "hogql_viz",
            "title": "HogQL Analytical Query Result",
            "query": query,
            "columns": ["event", "count()"],
            "rows": [
                ["$pageview", 48200],
                ["user_signed_up", 1240],
                ["checkout_step_completed", 890],
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen."

    def _publish_webrtc_card(self, card_dict: dict):
        """Broadcasts data card locally if LiveKit room participant is available."""
        if self.livekit_room and hasattr(self.livekit_room, "local_participant"):
            try:
                card_bytes = json.dumps(card_dict).encode("utf-8")
                import asyncio
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    asyncio.create_task(
                        self.livekit_room.local_participant.publish_data(
                            card_bytes,
                            reliable=True,
                            topic="posthog-viz-card",
                        )
                    )
            except Exception as ex:
                logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex)

    def get_tools(self) -> List[Any]:
        """Returns all LiveKit FunctionTools provided by this suite."""
        return [
            self.query_funnel_tool,
            self.query_errors_tool,
            self.query_retention_tool,
            self.query_hogql_tool,
        ]

    @property
    def query_funnel(self):
        return self.query_funnel_tool

    @property
    def query_errors(self):
        return self.query_errors_tool

    @property
    def query_retention(self):
        return self.query_retention_tool

    @property
    def query_hogql(self):
        return self.query_hogql_tool

Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.

Instance variables

prop query_errors

Expand source code

@property
def query_errors(self):
    return self.query_errors_tool

prop query_funnel

Expand source code

@property
def query_funnel(self):
    return self.query_funnel_tool

prop query_hogql

Expand source code

@property
def query_hogql(self):
    return self.query_hogql_tool

prop query_retention

Expand source code

@property
def query_retention(self):
    return self.query_retention_tool

Methods

def get_tools(self) ‑> List[Any]

Expand source code

def get_tools(self) -> List[Any]:
    """Returns all LiveKit FunctionTools provided by this suite."""
    return [
        self.query_funnel_tool,
        self.query_errors_tool,
        self.query_retention_tool,
        self.query_hogql_tool,
    ]

Returns all LiveKit FunctionTools provided by this suite.

class DualChannelTools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)

Expand source code

class PostHogDualChannelTools:
    """
    Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine.
    Splits analytical query responses into:
      1. High-throughput WebRTC data-channel visualization packet.
      2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
    """

    def __init__(
        self,
        gateway_url: str = DEFAULT_GATEWAY_URL,
        room_name: str = "prism-demo",
        livekit_room: Optional[Any] = None,
    ):
        self.gateway_url = gateway_url.rstrip("/")
        self.room_name = room_name
        self.livekit_room = livekit_room
        self.http_client = httpx.Client(timeout=3.0)

        # Wrap tools with LiveKit function_tool
        self.query_funnel_tool = function_tool(self._query_funnel_impl)
        self.query_errors_tool = function_tool(self._query_errors_impl)
        self.query_retention_tool = function_tool(self._query_retention_impl)
        self.query_hogql_tool = function_tool(self._query_hogql_impl)

    def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str:
        """
        Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup).
        Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser,
        and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_funnel",
            "args": {"funnel_id": funnel_id},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload)
            if resp.status_code == 200:
                data = resp.json()
                return data.get(
                    "speech_ready_delta",
                    "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        # Local fallback if Go Gateway is offline
        mock_card = {
            "type": "funnel_viz",
            "title": "Checkout Conversion Dropoff",
            "funnel_id": funnel_id,
            "steps": [
                {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0},
                {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1},
                {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5},
            ],
            "primary_drop_step": "Payment Confirmed",
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen."

    def _query_errors_impl(self, timeframe: str = "24h") -> str:
        """
        Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d).
        Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_errors",
            "args": {"timeframe": timeframe},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "error_waterfall",
            "title": f"API Error Waterfall ({timeframe})",
            "timeframe": timeframe,
            "total_errors": 42,
            "primary_root_cause": "PaymentGatewayTimeout (504)",
            "errors": [
                {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504},
                {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400},
                {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen."

    def _query_retention_impl(self, cohort: str = "weekly") -> str:
        """
        Query PostHog cohort retention trends over an 8-week horizon.
        Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_retention",
            "args": {"cohort": cohort},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "retention_trend",
            "title": "8-Week Cohort Retention Trend",
            "cohort_name": "Weekly Active Users",
            "terminal_retention": 19.5,
            "intervals": [
                {"interval": "Week 1", "retention_rate": 100.0, "users": 8400},
                {"interval": "Week 2", "retention_rate": 44.2, "users": 3712},
                {"interval": "Week 3", "retention_rate": 34.0, "users": 2856},
                {"interval": "Week 4", "retention_rate": 28.1, "users": 2360},
                {"interval": "Week 6", "retention_rate": 23.4, "users": 1965},
                {"interval": "Week 8", "retention_rate": 19.5, "users": 1638},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen."

    def _query_hogql_impl(self, query: str) -> str:
        """
        Execute an analytical HogQL query against PostHog and ClickHouse.
        Renders the analytical results card over WebRTC and returns a concise verbal summary.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_hogql",
            "args": {"query": query},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your HogQL query completed successfully. I've projected the analytical results onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "hogql_viz",
            "title": "HogQL Analytical Query Result",
            "query": query,
            "columns": ["event", "count()"],
            "rows": [
                ["$pageview", 48200],
                ["user_signed_up", 1240],
                ["checkout_step_completed", 890],
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen."

    def _publish_webrtc_card(self, card_dict: dict):
        """Broadcasts data card locally if LiveKit room participant is available."""
        if self.livekit_room and hasattr(self.livekit_room, "local_participant"):
            try:
                card_bytes = json.dumps(card_dict).encode("utf-8")
                import asyncio
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    asyncio.create_task(
                        self.livekit_room.local_participant.publish_data(
                            card_bytes,
                            reliable=True,
                            topic="posthog-viz-card",
                        )
                    )
            except Exception as ex:
                logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex)

    def get_tools(self) -> List[Any]:
        """Returns all LiveKit FunctionTools provided by this suite."""
        return [
            self.query_funnel_tool,
            self.query_errors_tool,
            self.query_retention_tool,
            self.query_hogql_tool,
        ]

    @property
    def query_funnel(self):
        return self.query_funnel_tool

    @property
    def query_errors(self):
        return self.query_errors_tool

    @property
    def query_retention(self):
        return self.query_retention_tool

    @property
    def query_hogql(self):
        return self.query_hogql_tool

Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.

Instance variables

prop query_errors

Expand source code

@property
def query_errors(self):
    return self.query_errors_tool

prop query_funnel

Expand source code

@property
def query_funnel(self):
    return self.query_funnel_tool

prop query_hogql

Expand source code

@property
def query_hogql(self):
    return self.query_hogql_tool

prop query_retention

Expand source code

@property
def query_retention(self):
    return self.query_retention_tool

Methods

def get_tools(self) ‑> List[Any]

Expand source code

def get_tools(self) -> List[Any]:
    """Returns all LiveKit FunctionTools provided by this suite."""
    return [
        self.query_funnel_tool,
        self.query_errors_tool,
        self.query_retention_tool,
        self.query_hogql_tool,
    ]

Returns all LiveKit FunctionTools provided by this suite.

class PostHogDualChannelTools (gateway_url: str = 'http://localhost:8080',
room_name: str = 'prism-demo',
livekit_room: Any | None = None)

Expand source code

class PostHogDualChannelTools:
    """
    Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine.
    Splits analytical query responses into:
      1. High-throughput WebRTC data-channel visualization packet.
      2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.
    """

    def __init__(
        self,
        gateway_url: str = DEFAULT_GATEWAY_URL,
        room_name: str = "prism-demo",
        livekit_room: Optional[Any] = None,
    ):
        self.gateway_url = gateway_url.rstrip("/")
        self.room_name = room_name
        self.livekit_room = livekit_room
        self.http_client = httpx.Client(timeout=3.0)

        # Wrap tools with LiveKit function_tool
        self.query_funnel_tool = function_tool(self._query_funnel_impl)
        self.query_errors_tool = function_tool(self._query_errors_impl)
        self.query_retention_tool = function_tool(self._query_retention_impl)
        self.query_hogql_tool = function_tool(self._query_hogql_impl)

    def _query_funnel_impl(self, funnel_id: str = "checkout_v2") -> str:
        """
        Query PostHog user conversion funnel analytics by funnel ID (e.g., checkout_v2 or signup).
        Visual chart data is pushed over the WebRTC Data Channel directly to the user's browser,
        and an ultra-compact speech-ready string (< 30 tokens) is returned to the agent context.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_funnel",
            "args": {"funnel_id": funnel_id},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/funnel", json=payload)
            if resp.status_code == 200:
                data = resp.json()
                return data.get(
                    "speech_ready_delta",
                    "The checkout funnel has a 22.5 percent conversion rate. I've projected the conversion funnel onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        # Local fallback if Go Gateway is offline
        mock_card = {
            "type": "funnel_viz",
            "title": "Checkout Conversion Dropoff",
            "funnel_id": funnel_id,
            "steps": [
                {"name": "Cart Viewed", "visitors": 14200, "conversion_rate": 100.0},
                {"name": "Billing Entered", "visitors": 9100, "conversion_rate": 64.1},
                {"name": "Payment Confirmed", "visitors": 3200, "conversion_rate": 22.5},
            ],
            "primary_drop_step": "Payment Confirmed",
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "The checkout funnel has a 22.5 percent conversion rate. The main drop-off is at the payment confirmation step. I've projected the conversion funnel onto your screen."

    def _query_errors_impl(self, timeframe: str = "24h") -> str:
        """
        Query PostHog API error waterfall and exception logs over a timeframe (e.g., 24h, 7d).
        Renders an interactive Error Waterfall card over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_errors",
            "args": {"timeframe": timeframe},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "error_waterfall",
            "title": f"API Error Waterfall ({timeframe})",
            "timeframe": timeframe,
            "total_errors": 42,
            "primary_root_cause": "PaymentGatewayTimeout (504)",
            "errors": [
                {"error_type": "PaymentGatewayTimeout (504)", "count": 22, "percentage": 52.4, "status_code": 504},
                {"error_type": "StripeCardDeclined (400)", "count": 13, "percentage": 30.9, "status_code": 400},
                {"error_type": "DatabaseDeadlock (500)", "count": 7, "percentage": 16.7, "status_code": 500},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "We detected 42 server errors in the last 24 hours. The highest failure rate is Payment Gateway Timeouts at 52 percent. The error waterfall is on your screen."

    def _query_retention_impl(self, cohort: str = "weekly") -> str:
        """
        Query PostHog cohort retention trends over an 8-week horizon.
        Renders a Retention Trend line chart over WebRTC and returns a conversational speech delta.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_retention",
            "args": {"cohort": cohort},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/execute", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "retention_trend",
            "title": "8-Week Cohort Retention Trend",
            "cohort_name": "Weekly Active Users",
            "terminal_retention": 19.5,
            "intervals": [
                {"interval": "Week 1", "retention_rate": 100.0, "users": 8400},
                {"interval": "Week 2", "retention_rate": 44.2, "users": 3712},
                {"interval": "Week 3", "retention_rate": 34.0, "users": 2856},
                {"interval": "Week 4", "retention_rate": 28.1, "users": 2360},
                {"interval": "Week 6", "retention_rate": 23.4, "users": 1965},
                {"interval": "Week 8", "retention_rate": 19.5, "users": 1638},
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your eight-week user retention stands at 19.5 percent, beating the industry average of 15 percent. The retention curve is displayed on your screen."

    def _query_hogql_impl(self, query: str) -> str:
        """
        Execute an analytical HogQL query against PostHog and ClickHouse.
        Renders the analytical results card over WebRTC and returns a concise verbal summary.
        """
        payload = {
            "room_name": self.room_name,
            "session_id": f"session_{int(time.time())}",
            "tool": "query_hogql",
            "args": {"query": query},
        }

        try:
            resp = self.http_client.post(f"{self.gateway_url}/api/tools/hogql", json=payload)
            if resp.status_code == 200:
                return resp.json().get(
                    "speech_ready_delta",
                    "Your HogQL query completed successfully. I've projected the analytical results onto your screen.",
                )
        except Exception as e:
            logger.warning("[Prism Tools] Gateway unreachable (%s). Using local fallback.", e)

        mock_card = {
            "type": "hogql_viz",
            "title": "HogQL Analytical Query Result",
            "query": query,
            "columns": ["event", "count()"],
            "rows": [
                ["$pageview", 48200],
                ["user_signed_up", 1240],
                ["checkout_step_completed", 890],
            ],
            "timestamp": int(time.time()),
        }
        self._publish_webrtc_card(mock_card)
        return "Your HogQL query completed successfully across 48,000 pageviews and 1,240 signups. I've projected the analytical results onto your screen."

    def _publish_webrtc_card(self, card_dict: dict):
        """Broadcasts data card locally if LiveKit room participant is available."""
        if self.livekit_room and hasattr(self.livekit_room, "local_participant"):
            try:
                card_bytes = json.dumps(card_dict).encode("utf-8")
                import asyncio
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    asyncio.create_task(
                        self.livekit_room.local_participant.publish_data(
                            card_bytes,
                            reliable=True,
                            topic="posthog-viz-card",
                        )
                    )
            except Exception as ex:
                logger.error("[Prism Tools] Failed to broadcast local data packet: %s", ex)

    def get_tools(self) -> List[Any]:
        """Returns all LiveKit FunctionTools provided by this suite."""
        return [
            self.query_funnel_tool,
            self.query_errors_tool,
            self.query_retention_tool,
            self.query_hogql_tool,
        ]

    @property
    def query_funnel(self):
        return self.query_funnel_tool

    @property
    def query_errors(self):
        return self.query_errors_tool

    @property
    def query_retention(self):
        return self.query_retention_tool

    @property
    def query_hogql(self):
        return self.query_hogql_tool

Dual-Channel Tool suite connecting LiveKit Voice Agents with PostHog's analytical engine. Splits analytical query responses into: 1. High-throughput WebRTC data-channel visualization packet. 2. Concise, speech-ready delta (< 30 tokens) for conversational LLM context.

Instance variables

prop query_errors

Expand source code

@property
def query_errors(self):
    return self.query_errors_tool

prop query_funnel

Expand source code

@property
def query_funnel(self):
    return self.query_funnel_tool

prop query_hogql

Expand source code

@property
def query_hogql(self):
    return self.query_hogql_tool

prop query_retention

Expand source code

@property
def query_retention(self):
    return self.query_retention_tool

Methods

def get_tools(self) ‑> List[Any]

Expand source code

def get_tools(self) -> List[Any]:
    """Returns all LiveKit FunctionTools provided by this suite."""
    return [
        self.query_funnel_tool,
        self.query_errors_tool,
        self.query_retention_tool,
        self.query_hogql_tool,
    ]

Returns all LiveKit FunctionTools provided by this suite.

class FeatureFlags (api_key: str | None = None,
host: str | None = None,
client: posthog.Posthog | None = None,
default_distinct_id: str = 'agent-system')

Expand source code

class PostHogFeatureFlags:
    """
    Client for evaluating PostHog feature flags, prompt experiments, and dynamic
    tool gates for LiveKit voice agents.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: Optional[str] = None,
        client: Optional[posthog.Posthog] = None,
        default_distinct_id: str = "agent-system",
    ):
        self.api_key = api_key or os.getenv("POSTHOG_API_KEY", "phc_mock_key_prism")
        self.host = host or os.getenv("POSTHOG_HOST", "https://us.i.posthog.com")
        self.default_distinct_id = default_distinct_id

        if client is not None:
            self.client = client
        else:
            try:
                self.client = posthog.Posthog(
                    project_api_key=self.api_key,
                    host=self.host,
                )
            except Exception as e:
                logger.warning("[PostHog FeatureFlags] Failed to init client: %s", e)
                self.client = None

    def is_feature_enabled(
        self,
        key: str,
        distinct_id: Optional[str] = None,
        default: bool = False,
        person_properties: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """
        Evaluates whether a feature flag is enabled for the given distinct_id.
        Useful for gating experimental tools (e.g., beta funnel tools).
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return default

        try:
            res = self.client.feature_enabled(
                key=key,
                distinct_id=target_id,
                person_properties=person_properties,
            )
            return bool(res) if res is not None else default
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error evaluating %s: %s", key, e)
            return default

    def get_feature_flag(
        self,
        key: str,
        distinct_id: Optional[str] = None,
        default: Any = None,
        person_properties: Optional[Dict[str, Any]] = None,
    ) -> Any:
        """
        Retrieves the multivariate variant for a feature flag.
        Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return default

        try:
            val = self.client.get_feature_flag(
                key=key,
                distinct_id=target_id,
                person_properties=person_properties,
            )
            return val if val is not None else default
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error fetching flag %s: %s", key, e)
            return default

    def get_prompt_variant(
        self,
        key: str,
        fallback_prompt: str,
        distinct_id: Optional[str] = None,
    ) -> str:
        """
        Retrieves a dynamic prompt instruction payload from a PostHog experiment flag.
        Enables zero-deploy A/B testing of system prompts.
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return fallback_prompt

        try:
            payload = self.client.get_feature_flag_payload(key=key, distinct_id=target_id)
            if isinstance(payload, str) and payload.strip():
                return payload
            elif isinstance(payload, dict) and "prompt" in payload:
                return str(payload["prompt"])
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error fetching prompt variant for %s: %s", key, e)

        return fallback_prompt

Client for evaluating PostHog feature flags, prompt experiments, and dynamic tool gates for LiveKit voice agents.

Methods

def get_feature_flag(self,
key: str,
distinct_id: str | None = None,
default: Any = None,
person_properties: Dict[str, Any] | None = None) ‑> Any

Expand source code

def get_feature_flag(
    self,
    key: str,
    distinct_id: Optional[str] = None,
    default: Any = None,
    person_properties: Optional[Dict[str, Any]] = None,
) -> Any:
    """
    Retrieves the multivariate variant for a feature flag.
    Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return default

    try:
        val = self.client.get_feature_flag(
            key=key,
            distinct_id=target_id,
            person_properties=person_properties,
        )
        return val if val is not None else default
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error fetching flag %s: %s", key, e)
        return default

Retrieves the multivariate variant for a feature flag. Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').

def get_prompt_variant(self, key: str, fallback_prompt: str, distinct_id: str | None = None) ‑> str

Expand source code

def get_prompt_variant(
    self,
    key: str,
    fallback_prompt: str,
    distinct_id: Optional[str] = None,
) -> str:
    """
    Retrieves a dynamic prompt instruction payload from a PostHog experiment flag.
    Enables zero-deploy A/B testing of system prompts.
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return fallback_prompt

    try:
        payload = self.client.get_feature_flag_payload(key=key, distinct_id=target_id)
        if isinstance(payload, str) and payload.strip():
            return payload
        elif isinstance(payload, dict) and "prompt" in payload:
            return str(payload["prompt"])
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error fetching prompt variant for %s: %s", key, e)

    return fallback_prompt

Retrieves a dynamic prompt instruction payload from a PostHog experiment flag. Enables zero-deploy A/B testing of system prompts.

def is_feature_enabled(self,
key: str,
distinct_id: str | None = None,
default: bool = False,
person_properties: Dict[str, Any] | None = None) ‑> bool

Expand source code

def is_feature_enabled(
    self,
    key: str,
    distinct_id: Optional[str] = None,
    default: bool = False,
    person_properties: Optional[Dict[str, Any]] = None,
) -> bool:
    """
    Evaluates whether a feature flag is enabled for the given distinct_id.
    Useful for gating experimental tools (e.g., beta funnel tools).
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return default

    try:
        res = self.client.feature_enabled(
            key=key,
            distinct_id=target_id,
            person_properties=person_properties,
        )
        return bool(res) if res is not None else default
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error evaluating %s: %s", key, e)
        return default

Evaluates whether a feature flag is enabled for the given distinct_id. Useful for gating experimental tools (e.g., beta funnel tools).

class PostHogFeatureFlags (api_key: str | None = None,
host: str | None = None,
client: posthog.Posthog | None = None,
default_distinct_id: str = 'agent-system')

Expand source code

class PostHogFeatureFlags:
    """
    Client for evaluating PostHog feature flags, prompt experiments, and dynamic
    tool gates for LiveKit voice agents.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: Optional[str] = None,
        client: Optional[posthog.Posthog] = None,
        default_distinct_id: str = "agent-system",
    ):
        self.api_key = api_key or os.getenv("POSTHOG_API_KEY", "phc_mock_key_prism")
        self.host = host or os.getenv("POSTHOG_HOST", "https://us.i.posthog.com")
        self.default_distinct_id = default_distinct_id

        if client is not None:
            self.client = client
        else:
            try:
                self.client = posthog.Posthog(
                    project_api_key=self.api_key,
                    host=self.host,
                )
            except Exception as e:
                logger.warning("[PostHog FeatureFlags] Failed to init client: %s", e)
                self.client = None

    def is_feature_enabled(
        self,
        key: str,
        distinct_id: Optional[str] = None,
        default: bool = False,
        person_properties: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """
        Evaluates whether a feature flag is enabled for the given distinct_id.
        Useful for gating experimental tools (e.g., beta funnel tools).
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return default

        try:
            res = self.client.feature_enabled(
                key=key,
                distinct_id=target_id,
                person_properties=person_properties,
            )
            return bool(res) if res is not None else default
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error evaluating %s: %s", key, e)
            return default

    def get_feature_flag(
        self,
        key: str,
        distinct_id: Optional[str] = None,
        default: Any = None,
        person_properties: Optional[Dict[str, Any]] = None,
    ) -> Any:
        """
        Retrieves the multivariate variant for a feature flag.
        Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return default

        try:
            val = self.client.get_feature_flag(
                key=key,
                distinct_id=target_id,
                person_properties=person_properties,
            )
            return val if val is not None else default
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error fetching flag %s: %s", key, e)
            return default

    def get_prompt_variant(
        self,
        key: str,
        fallback_prompt: str,
        distinct_id: Optional[str] = None,
    ) -> str:
        """
        Retrieves a dynamic prompt instruction payload from a PostHog experiment flag.
        Enables zero-deploy A/B testing of system prompts.
        """
        target_id = distinct_id or self.default_distinct_id
        if not self.client:
            return fallback_prompt

        try:
            payload = self.client.get_feature_flag_payload(key=key, distinct_id=target_id)
            if isinstance(payload, str) and payload.strip():
                return payload
            elif isinstance(payload, dict) and "prompt" in payload:
                return str(payload["prompt"])
        except Exception as e:
            logger.debug("[PostHog FeatureFlags] Error fetching prompt variant for %s: %s", key, e)

        return fallback_prompt

Client for evaluating PostHog feature flags, prompt experiments, and dynamic tool gates for LiveKit voice agents.

Methods

def get_feature_flag(self,
key: str,
distinct_id: str | None = None,
default: Any = None,
person_properties: Dict[str, Any] | None = None) ‑> Any

Expand source code

def get_feature_flag(
    self,
    key: str,
    distinct_id: Optional[str] = None,
    default: Any = None,
    person_properties: Optional[Dict[str, Any]] = None,
) -> Any:
    """
    Retrieves the multivariate variant for a feature flag.
    Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return default

    try:
        val = self.client.get_feature_flag(
            key=key,
            distinct_id=target_id,
            person_properties=person_properties,
        )
        return val if val is not None else default
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error fetching flag %s: %s", key, e)
        return default

Retrieves the multivariate variant for a feature flag. Useful for A/B testing different voice models (e.g. 'gpt-4o-mini' vs 'claude-3-5-sonnet').

def get_prompt_variant(self, key: str, fallback_prompt: str, distinct_id: str | None = None) ‑> str

Expand source code

def get_prompt_variant(
    self,
    key: str,
    fallback_prompt: str,
    distinct_id: Optional[str] = None,
) -> str:
    """
    Retrieves a dynamic prompt instruction payload from a PostHog experiment flag.
    Enables zero-deploy A/B testing of system prompts.
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return fallback_prompt

    try:
        payload = self.client.get_feature_flag_payload(key=key, distinct_id=target_id)
        if isinstance(payload, str) and payload.strip():
            return payload
        elif isinstance(payload, dict) and "prompt" in payload:
            return str(payload["prompt"])
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error fetching prompt variant for %s: %s", key, e)

    return fallback_prompt

Retrieves a dynamic prompt instruction payload from a PostHog experiment flag. Enables zero-deploy A/B testing of system prompts.

def is_feature_enabled(self,
key: str,
distinct_id: str | None = None,
default: bool = False,
person_properties: Dict[str, Any] | None = None) ‑> bool

Expand source code

def is_feature_enabled(
    self,
    key: str,
    distinct_id: Optional[str] = None,
    default: bool = False,
    person_properties: Optional[Dict[str, Any]] = None,
) -> bool:
    """
    Evaluates whether a feature flag is enabled for the given distinct_id.
    Useful for gating experimental tools (e.g., beta funnel tools).
    """
    target_id = distinct_id or self.default_distinct_id
    if not self.client:
        return default

    try:
        res = self.client.feature_enabled(
            key=key,
            distinct_id=target_id,
            person_properties=person_properties,
        )
        return bool(res) if res is not None else default
    except Exception as e:
        logger.debug("[PostHog FeatureFlags] Error evaluating %s: %s", key, e)
        return default

Evaluates whether a feature flag is enabled for the given distinct_id. Useful for gating experimental tools (e.g., beta funnel tools).

class PostHogPlugin

Expand source code

class PostHogPlugin(Plugin):
    def __init__(self) -> None:
        super().__init__(__name__, __version__, __package__, logger)

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.plugin.Plugin
• abc.ABC

class PluginOptions (api_key: str | None = None,
host: str = 'https://us.i.posthog.com',
project_id: str | None = None,
personal_api_key: str | None = None,
flush_at: int = 1,
flush_interval: float = 0.5,
capture_tts_metrics: bool = True,
capture_stt_metrics: bool = True,
capture_eou_metrics: bool = True,
capture_interruption_metrics: bool = True)

Expand source code

@dataclass
class PostHogPluginOptions:
    """Configuration options for the PostHog LiveKit plugin."""
    api_key: Optional[str] = None
    host: str = "https://us.i.posthog.com"
    project_id: Optional[str] = None
    personal_api_key: Optional[str] = None
    flush_at: int = 1
    flush_interval: float = 0.5
    capture_tts_metrics: bool = True
    capture_stt_metrics: bool = True
    capture_eou_metrics: bool = True
    capture_interruption_metrics: bool = True

Configuration options for the PostHog LiveKit plugin.

Instance variables

var api_key : str | None

The type of the None singleton.

var capture_eou_metrics : bool

The type of the None singleton.

var capture_interruption_metrics : bool

The type of the None singleton.

var capture_stt_metrics : bool

The type of the None singleton.

var capture_tts_metrics : bool

The type of the None singleton.

var flush_at : int

The type of the None singleton.

var flush_interval : float

The type of the None singleton.

var host : str

The type of the None singleton.

var personal_api_key : str | None

The type of the None singleton.

var project_id : str | None

The type of the None singleton.

class PostHogPluginOptions (api_key: str | None = None,
host: str = 'https://us.i.posthog.com',
project_id: str | None = None,
personal_api_key: str | None = None,
flush_at: int = 1,
flush_interval: float = 0.5,
capture_tts_metrics: bool = True,
capture_stt_metrics: bool = True,
capture_eou_metrics: bool = True,
capture_interruption_metrics: bool = True)

Expand source code

@dataclass
class PostHogPluginOptions:
    """Configuration options for the PostHog LiveKit plugin."""
    api_key: Optional[str] = None
    host: str = "https://us.i.posthog.com"
    project_id: Optional[str] = None
    personal_api_key: Optional[str] = None
    flush_at: int = 1
    flush_interval: float = 0.5
    capture_tts_metrics: bool = True
    capture_stt_metrics: bool = True
    capture_eou_metrics: bool = True
    capture_interruption_metrics: bool = True

Configuration options for the PostHog LiveKit plugin.

Instance variables

var api_key : str | None

The type of the None singleton.

var capture_eou_metrics : bool

The type of the None singleton.

var capture_interruption_metrics : bool

The type of the None singleton.

var capture_stt_metrics : bool

The type of the None singleton.

var capture_tts_metrics : bool

The type of the None singleton.

var flush_at : int

The type of the None singleton.

var flush_interval : float

The type of the None singleton.

var host : str

The type of the None singleton.

var personal_api_key : str | None

The type of the None singleton.

var project_id : str | None

The type of the None singleton.

class VisualFunnelCard (type: str = 'funnel_viz',
title: str = 'Conversion Funnel',
funnel_id: str = 'default_funnel',
steps: List[FunnelStep] = <factory>,
primary_drop_step: str = '',
timestamp: int = 0)

Expand source code

@dataclass
class VisualFunnelCard:
    """Payload pushed over WebRTC data channel conforming to Contract A."""
    type: str = "funnel_viz"
    title: str = "Conversion Funnel"
    funnel_id: str = "default_funnel"
    steps: List[FunnelStep] = field(default_factory=list)
    primary_drop_step: str = ""
    timestamp: int = 0

Payload pushed over WebRTC data channel conforming to Contract A.

Instance variables

var funnel_id : str

The type of the None singleton.

var primary_drop_step : str

The type of the None singleton.

var steps : List[FunnelStep]

The type of the None singleton.

var timestamp : int

The type of the None singleton.

var title : str

The type of the None singleton.

var type : str

The type of the None singleton.