Module livekit.plugins.posthog.feature_flags

PostHog Feature Flags & Dynamic Prompt Experimentation for LiveKit Agents.

Classes

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 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).