insights-python/posthog/ai/prompts.py

"""
Prompt management for PostHog AI SDK.

Fetch and compile LLM prompts from PostHog with caching and fallback support.
"""

import logging
import re
import time
import urllib.parse
from typing import Any, Dict, Optional, Union

from posthog.request import USER_AGENT, _get_session
from posthog.utils import remove_trailing_slash

log = logging.getLogger("posthog")

APP_ENDPOINT = "https://us.posthog.com"
DEFAULT_CACHE_TTL_SECONDS = 300  # 5 minutes

PromptVariables = Dict[str, Union[str, int, float, bool]]
PromptCacheKey = tuple[str, Optional[int]]


class CachedPrompt:
    """Cached prompt with metadata."""

    def __init__(self, prompt: str, fetched_at: float):
        self.prompt = prompt
        self.fetched_at = fetched_at


def _cache_key(name: str, version: Optional[int]) -> PromptCacheKey:
    """Build a cache key for latest or versioned prompt fetches."""
    return (name, version)


def _prompt_reference(name: str, version: Optional[int]) -> str:
    """Format a prompt reference for logs and errors."""
    label = f'prompt "{name}"'
    if version is not None:
        return f"{label} version {version}"
    return label


def _is_prompt_api_response(data: Any) -> bool:
    """Check if the response is a valid prompt API response."""
    return (
        isinstance(data, dict)
        and "prompt" in data
        and isinstance(data.get("prompt"), str)
    )


class Prompts:
    """
    Fetch and compile LLM prompts from PostHog.

    Can be initialized with a PostHog client or with direct options.

    Examples:
        ```python
        from posthog import Posthog
        from posthog.ai.prompts import Prompts

        # With PostHog client
        posthog = Posthog('phc_xxx', host='https://us.posthog.com', personal_api_key='phx_xxx')
        prompts = Prompts(posthog)

        # Or with direct options (no PostHog client needed)
        prompts = Prompts(
            personal_api_key='phx_xxx',
            project_api_key='phc_xxx',
            host='https://us.posthog.com',
        )

        # Fetch with caching and fallback
        template = prompts.get('support-system-prompt', fallback='You are a helpful assistant.')

        # Fetch a specific published version
        prompt_v1 = prompts.get('support-system-prompt', version=1)

        # Compile with variables
        system_prompt = prompts.compile(template, {
            'company': 'Acme Corp',
            'tier': 'premium',
        })
        ```
    """

    def __init__(
        self,
        posthog: Optional[Any] = None,
        *,
        personal_api_key: Optional[str] = None,
        project_api_key: Optional[str] = None,
        host: Optional[str] = None,
        default_cache_ttl_seconds: Optional[int] = None,
    ):
        """
        Initialize Prompts.

        Args:
            posthog: PostHog client instance (optional if personal_api_key provided)
            personal_api_key: Direct personal API key (optional if posthog provided)
            project_api_key: Direct project API key (optional if posthog provided)
            host: PostHog host (defaults to app endpoint)
            default_cache_ttl_seconds: Default cache TTL (defaults to 300)
        """
        self._default_cache_ttl_seconds = (
            default_cache_ttl_seconds or DEFAULT_CACHE_TTL_SECONDS
        )
        self._cache: Dict[PromptCacheKey, CachedPrompt] = {}

        if posthog is not None:
            self._personal_api_key = getattr(posthog, "personal_api_key", None) or ""
            self._project_api_key = getattr(posthog, "api_key", None) or ""
            self._host = remove_trailing_slash(
                getattr(posthog, "raw_host", None) or APP_ENDPOINT
            )
        else:
            self._personal_api_key = personal_api_key or ""
            self._project_api_key = project_api_key or ""
            self._host = remove_trailing_slash(host or APP_ENDPOINT)

    def get(
        self,
        name: str,
        *,
        cache_ttl_seconds: Optional[int] = None,
        fallback: Optional[str] = None,
        version: Optional[int] = None,
    ) -> str:
        """
        Fetch a prompt by name from the PostHog API.

        Caching behavior:
        1. If cache is fresh, return cached value
        2. If fetch fails and cache exists (stale), return stale cache with warning
        3. If fetch fails and fallback provided, return fallback with warning
        4. If fetch fails with no cache/fallback, raise exception

        Args:
            name: The name of the prompt to fetch
            cache_ttl_seconds: Cache TTL in seconds (defaults to instance default)
            fallback: Fallback prompt to use if fetch fails and no cache available
            version: Specific prompt version to fetch. If None, fetches the latest
                version

        Returns:
            The prompt string

        Raises:
            Exception: If the prompt cannot be fetched and no fallback is available
        """
        ttl = (
            cache_ttl_seconds
            if cache_ttl_seconds is not None
            else self._default_cache_ttl_seconds
        )
        cache_key = _cache_key(name, version)

        # Check cache first
        cached = self._cache.get(cache_key)
        now = time.time()

        if cached is not None:
            is_fresh = (now - cached.fetched_at) < ttl

            if is_fresh:
                return cached.prompt

        # Try to fetch from API
        try:
            prompt = self._fetch_prompt_from_api(name, version)
            fetched_at = time.time()

            # Update cache
            self._cache[cache_key] = CachedPrompt(prompt=prompt, fetched_at=fetched_at)

            return prompt

        except Exception as error:
            prompt_reference = _prompt_reference(name, version)
            # Fallback order:
            # 1. Return stale cache (with warning)
            if cached is not None:
                log.warning(
                    "[PostHog Prompts] Failed to fetch %s, using stale cache: %s",
                    prompt_reference,
                    error,
                )
                return cached.prompt

            # 2. Return fallback (with warning)
            if fallback is not None:
                log.warning(
                    "[PostHog Prompts] Failed to fetch %s, using fallback: %s",
                    prompt_reference,
                    error,
                )
                return fallback

            # 3. Raise error
            raise

    def compile(self, prompt: str, variables: PromptVariables) -> str:
        """
        Replace {{variableName}} placeholders with values.

        Unmatched variables are left unchanged.
        Supports variable names with hyphens and dots (e.g., user-id, company.name).

        Args:
            prompt: The prompt template string
            variables: Object containing variable values

        Returns:
            The compiled prompt string
        """

        def replace_variable(match: re.Match) -> str:
            variable_name = match.group(1)

            if variable_name in variables:
                return str(variables[variable_name])

            return match.group(0)

        return re.sub(r"\{\{([\w.-]+)\}\}", replace_variable, prompt)

    def clear_cache(
        self, name: Optional[str] = None, *, version: Optional[int] = None
    ) -> None:
        """
        Clear cached prompts.

        Args:
            name: Specific prompt name to clear. If None, clears all cached prompts.
            version: Specific prompt version to clear. Requires name.
        """
        if version is not None and name is None:
            raise ValueError("'version' requires 'name' to be provided")

        if name is None:
            self._cache.clear()
            return

        if version is not None:
            self._cache.pop(_cache_key(name, version), None)
            return

        keys_to_clear = [key for key in self._cache if key[0] == name]
        for key in keys_to_clear:
            self._cache.pop(key, None)

    def _fetch_prompt_from_api(self, name: str, version: Optional[int] = None) -> str:
        """
        Fetch prompt from PostHog API.

        Endpoint:
            {host}/api/environments/@current/llm_prompts/name/{encoded_name}/
            ?token={encoded_project_api_key}[&version={version}]
        Auth: Bearer {personal_api_key}

        Args:
            name: The name of the prompt to fetch
            version: Specific prompt version to fetch. If None, fetches the latest

        Returns:
            The prompt string

        Raises:
            Exception: If the prompt cannot be fetched
        """
        if not self._personal_api_key:
            raise Exception(
                "[PostHog Prompts] personal_api_key is required to fetch prompts. "
                "Please provide it when initializing the Prompts instance."
            )
        if not self._project_api_key:
            raise Exception(
                "[PostHog Prompts] project_api_key is required to fetch prompts. "
                "Please provide it when initializing the Prompts instance."
            )

        encoded_name = urllib.parse.quote(name, safe="")
        query_params: Dict[str, Union[str, int]] = {"token": self._project_api_key}
        if version is not None:
            query_params["version"] = version
        encoded_query = urllib.parse.urlencode(query_params)
        url = f"{self._host}/api/environments/@current/llm_prompts/name/{encoded_name}/?{encoded_query}"
        prompt_reference = _prompt_reference(name, version)
        prompt_label = prompt_reference[:1].upper() + prompt_reference[1:]

        headers = {
            "Authorization": f"Bearer {self._personal_api_key}",
            "User-Agent": USER_AGENT,
        }

        response = _get_session().get(url, headers=headers, timeout=10)

        if not response.ok:
            if response.status_code == 404:
                raise Exception(f"[PostHog Prompts] {prompt_label} not found")

            if response.status_code == 403:
                raise Exception(
                    f"[PostHog Prompts] Access denied for {prompt_reference}. "
                    "Check that your personal_api_key has the correct permissions and the LLM prompts feature is enabled."
                )

            raise Exception(
                f"[PostHog Prompts] Failed to fetch {prompt_label}: HTTP {response.status_code}"
            )

        try:
            data = response.json()
        except Exception:
            raise Exception(
                f"[PostHog Prompts] Invalid response format for {prompt_label}"
            )

        if not _is_prompt_api_response(data):
            raise Exception(
                f"[PostHog Prompts] Invalid response format for {prompt_label}"
            )

        return data["prompt"]