From 967ecbb3e9411f11b7634c299542bb09d2b0ced7 Mon Sep 17 00:00:00 2001 From: Juraj Majerik Date: Thu, 16 Jul 2026 14:33:35 +0200 Subject: [PATCH 1/5] feat(prompts): fetch prompts by label --- posthog/ai/prompts.py | 90 ++++++++++++++++++++++++--------- posthog/test/ai/test_prompts.py | 70 +++++++++++++++++++++++++ 2 files changed, 135 insertions(+), 25 deletions(-) diff --git a/posthog/ai/prompts.py b/posthog/ai/prompts.py index 4df46b5b..3aeea0c8 100644 --- a/posthog/ai/prompts.py +++ b/posthog/ai/prompts.py @@ -21,7 +21,7 @@ DEFAULT_CACHE_TTL_SECONDS = 300 # 5 minutes PromptVariables = Dict[str, Union[str, int, float, bool]] -PromptCacheKey = tuple[str, Optional[int]] +PromptCacheKey = tuple[str, Optional[int], Optional[str]] PromptSource = Literal["api", "cache", "stale_cache", "code_fallback"] @@ -34,32 +34,49 @@ class PromptResult: prompt: str name: Optional[str] = None version: Optional[int] = None + label: Optional[str] = None class CachedPrompt: """Cached prompt with metadata.""" - def __init__(self, prompt: str, fetched_at: float, name: str, version: int): + def __init__( + self, + prompt: str, + fetched_at: float, + name: str, + version: int, + label: Optional[str] = None, + ): self.prompt = prompt self.fetched_at = fetched_at self.name = name self.version = version + self.label = label -def _cache_key(name: str, version: Optional[int]) -> PromptCacheKey: - """Build a cache key for latest or versioned prompt fetches.""" - return (name, version) +def _cache_key( + name: str, version: Optional[int], label: Optional[str] = None +) -> PromptCacheKey: + """Build a cache key for latest, versioned, or labeled prompt fetches.""" + return (name, version, label) def _prompt_reference( - name: str, version: Optional[int], *, capitalize: bool = False + name: str, + version: Optional[int], + label: Optional[str] = None, + *, + capitalize: bool = False, ) -> str: """Format a prompt reference for logs and errors.""" prefix = "Prompt" if capitalize else "prompt" - label = f'{prefix} "{name}"' + reference = f'{prefix} "{name}"' if version is not None: - return f"{label} version {version}" - return label + return f"{reference} version {version}" + if label is not None: + return f'{reference} label "{label}"' + return reference def _is_prompt_api_response(data: Any) -> bool: @@ -103,6 +120,9 @@ class Prompts: # Fetch a specific published version prompt_v1 = prompts.get('support-system-prompt', version=1) + # Fetch the version a label currently points to + prod_prompt = prompts.get('support-system-prompt', label='production') + # Compile with variables system_prompt = prompts.compile(template, { 'company': 'Acme Corp', @@ -161,6 +181,7 @@ def get( cache_ttl_seconds: Optional[int] = ..., fallback: Optional[str] = ..., version: Optional[int] = ..., + label: Optional[str] = ..., ) -> PromptResult: ... @overload @@ -172,6 +193,7 @@ def get( cache_ttl_seconds: Optional[int] = ..., fallback: Optional[str] = ..., version: Optional[int] = ..., + label: Optional[str] = ..., ) -> str: ... @overload @@ -182,6 +204,7 @@ def get( cache_ttl_seconds: Optional[int] = ..., fallback: Optional[str] = ..., version: Optional[int] = ..., + label: Optional[str] = ..., ) -> str: ... def get( @@ -192,6 +215,7 @@ def get( cache_ttl_seconds: Optional[int] = None, fallback: Optional[str] = None, version: Optional[int] = None, + label: Optional[str] = None, ) -> Union[str, PromptResult]: """ Fetch a prompt by name from the PostHog API. @@ -207,15 +231,22 @@ def get( Omitting this parameter is deprecated. 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 + version: Specific prompt version to fetch. Mutually exclusive with label. + If neither is given, fetches the latest version + label: Fetch the version this label currently points to, e.g. + 'production'. Mutually exclusive with version Returns: str if with_metadata is False/omitted, PromptResult if True Raises: + ValueError: If both version and label are provided Exception: If the prompt cannot be fetched and no fallback is available """ + if version is not None and label is not None: + raise ValueError( + "[PostHog Prompts] Pass either version or label, not both." + ) if with_metadata is None and not self._has_warned_deprecation: self._has_warned_deprecation = True warnings.warn( @@ -232,13 +263,13 @@ def get( try: result = self._get_internal( - name, cache_ttl_seconds=cache_ttl_seconds, version=version + name, cache_ttl_seconds=cache_ttl_seconds, version=version, label=label ) if with_metadata is True: return result return result.prompt except Exception as error: - prompt_reference = _prompt_reference(name, version) + prompt_reference = _prompt_reference(name, version, label) if fallback is not None: log.warning( "[PostHog Prompts] Failed to fetch %s, using fallback: %s", @@ -256,6 +287,7 @@ def _get_internal( *, cache_ttl_seconds: Optional[int] = None, version: Optional[int] = None, + label: Optional[str] = None, ) -> PromptResult: """ Internal method that handles cache + fetch logic, returning full metadata. @@ -267,7 +299,7 @@ def _get_internal( if cache_ttl_seconds is not None else self._default_cache_ttl_seconds ) - cache_key = _cache_key(name, version) + cache_key = _cache_key(name, version, label) # Check cache first cached = self._cache.get(cache_key) @@ -282,11 +314,12 @@ def _get_internal( prompt=cached.prompt, name=cached.name, version=cached.version, + label=cached.label, ) # Try to fetch from API try: - data = self._fetch_prompt_from_api(name, version) + data = self._fetch_prompt_from_api(name, version, label) # Update cache self._cache[cache_key] = CachedPrompt( @@ -294,6 +327,7 @@ def _get_internal( fetched_at=time.time(), name=data["name"], version=data["version"], + label=data.get("label"), ) return PromptResult( @@ -301,12 +335,13 @@ def _get_internal( prompt=data["prompt"], name=data["name"], version=data["version"], + label=data.get("label"), ) except Exception as error: self._maybe_capture_error(error, name=name, version=version) - prompt_reference = _prompt_reference(name, version) + prompt_reference = _prompt_reference(name, version, label) # Return stale cache (with warning) if cached is not None: log.warning( @@ -319,6 +354,7 @@ def _get_internal( prompt=cached.prompt, name=cached.name, version=cached.version, + label=cached.label, ) raise @@ -395,19 +431,21 @@ def _maybe_capture_error( log.debug("[PostHog Prompts] Failed to capture exception to error tracking") def _fetch_prompt_from_api( - self, name: str, version: Optional[int] = None + self, name: str, version: Optional[int] = None, label: Optional[str] = None ) -> Dict[str, Any]: """ Fetch prompt from PostHog API. Endpoint: {host}/api/environments/@current/llm_prompts/name/{encoded_name}/ - ?token={encoded_project_api_key}[&version={version}] + ?token={encoded_project_api_key}[&version={version}][&label={label}] 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 + version: Specific prompt version to fetch + label: Fetch the version this label points to. If neither version nor + label is given, fetches the latest Returns: The validated API response dict containing prompt, name, and version @@ -430,10 +468,12 @@ def _fetch_prompt_from_api( query_params: Dict[str, Union[str, int]] = {"token": self._project_api_key} if version is not None: query_params["version"] = version + if label is not None: + query_params["label"] = label 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(name, version, capitalize=True) + prompt_reference = _prompt_reference(name, version, label) + prompt_title = _prompt_reference(name, version, label, capitalize=True) headers = { "Authorization": f"Bearer {self._personal_api_key}", @@ -444,7 +484,7 @@ def _fetch_prompt_from_api( if not response.ok: if response.status_code == 404: - raise Exception(f"[PostHog Prompts] {prompt_label} not found") + raise Exception(f"[PostHog Prompts] {prompt_title} not found") if response.status_code == 403: raise Exception( @@ -453,19 +493,19 @@ def _fetch_prompt_from_api( ) raise Exception( - f"[PostHog Prompts] Failed to fetch {prompt_label}: HTTP {response.status_code}" + f"[PostHog Prompts] Failed to fetch {prompt_title}: HTTP {response.status_code}" ) try: data = response.json() except Exception: raise Exception( - f"[PostHog Prompts] Invalid response format for {prompt_label}" + f"[PostHog Prompts] Invalid response format for {prompt_title}" ) if not _is_prompt_api_response(data): raise Exception( - f"[PostHog Prompts] Invalid response format for {prompt_label}" + f"[PostHog Prompts] Invalid response format for {prompt_title}" ) return data diff --git a/posthog/test/ai/test_prompts.py b/posthog/test/ai/test_prompts.py index 47105fbd..6f4df94c 100644 --- a/posthog/test/ai/test_prompts.py +++ b/posthog/test/ai/test_prompts.py @@ -158,6 +158,76 @@ def test_cache_latest_and_versioned_prompts_separately(self, mock_get_session): ) self.assertEqual(mock_get.call_count, 2) + @patch("posthog.ai.prompts._get_session") + def test_fetch_by_label_passes_param_and_returns_label_metadata( + self, mock_get_session + ): + """Should send the label query param and surface label metadata.""" + mock_get = mock_get_session.return_value.get + labeled_prompt_response = { + **self.mock_prompt_response, + "prompt": "Production prompt", + "version": 3, + "label": "production", + } + mock_get.return_value = MockResponse(json_data=labeled_prompt_response) + + posthog = self.create_mock_posthog() + prompts = Prompts(posthog) + + result = prompts.get("test-prompt", label="production", with_metadata=True) + + called_url = mock_get.call_args[0][0] + self.assertIn("label=production", called_url) + self.assertNotIn("version=", called_url) + self.assertEqual(result.prompt, "Production prompt") + self.assertEqual(result.version, 3) + self.assertEqual(result.label, "production") + + def test_version_and_label_together_raises(self): + """Should reject version and label passed together.""" + prompts = Prompts(self.create_mock_posthog()) + + with self.assertRaises(ValueError): + prompts.get("test-prompt", version=1, label="production") + + @patch("posthog.ai.prompts._get_session") + def test_cache_labeled_and_latest_prompts_separately(self, mock_get_session): + """Should cache labeled fetches separately from latest fetches.""" + mock_get = mock_get_session.return_value.get + latest_prompt_response = { + **self.mock_prompt_response, + "prompt": "Latest prompt", + "version": 4, + } + labeled_prompt_response = { + **self.mock_prompt_response, + "prompt": "Production prompt", + "version": 2, + "label": "production", + } + mock_get.side_effect = [ + MockResponse(json_data=latest_prompt_response), + MockResponse(json_data=labeled_prompt_response), + ] + + posthog = self.create_mock_posthog() + prompts = Prompts(posthog) + + # A labeled fetch after a latest fetch must not be served from the + # latest cache entry — that would silently return the wrong version. + self.assertEqual(prompts.get("test-prompt"), latest_prompt_response["prompt"]) + self.assertEqual( + prompts.get("test-prompt", label="production"), + labeled_prompt_response["prompt"], + ) + self.assertEqual(prompts.get("test-prompt"), latest_prompt_response["prompt"]) + self.assertEqual( + prompts.get("test-prompt", label="production"), + labeled_prompt_response["prompt"], + ) + self.assertEqual(mock_get.call_count, 2) + @patch("posthog.ai.prompts._get_session") @patch("posthog.ai.prompts.time.time") def test_refetch_when_cache_is_stale(self, mock_time, mock_get_session): From 1fb1c2a4302ee4ad3ad011d3fca61888bc1bb523 Mon Sep 17 00:00:00 2001 From: Juraj Majerik Date: Thu, 16 Jul 2026 14:54:26 +0200 Subject: [PATCH 2/5] feat(prompts): include label in error capture and docstrings --- posthog/ai/prompts.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/posthog/ai/prompts.py b/posthog/ai/prompts.py index 3aeea0c8..f838bbb1 100644 --- a/posthog/ai/prompts.py +++ b/posthog/ai/prompts.py @@ -28,7 +28,11 @@ @dataclass(frozen=True) class PromptResult: - """Result of a prompt fetch with metadata about its source.""" + """Result of a prompt fetch with metadata about its source. + + ``label`` is the label the prompt resolved through, populated from the API + response when fetching with the ``label`` option; ``None`` otherwise. + """ source: PromptSource prompt: str @@ -339,7 +343,7 @@ def _get_internal( ) except Exception as error: - self._maybe_capture_error(error, name=name, version=version) + self._maybe_capture_error(error, name=name, version=version, label=label) prompt_reference = _prompt_reference(name, version, label) # Return stale cache (with warning) @@ -410,7 +414,12 @@ def clear_cache( self._cache.pop(key, None) def _maybe_capture_error( - self, error: Exception, *, name: str, version: Optional[int] + self, + error: Exception, + *, + name: str, + version: Optional[int], + label: Optional[str] = None, ) -> None: """Report a prompt fetch error to PostHog error tracking if enabled.""" if not self._capture_errors or self._client is None: @@ -424,6 +433,7 @@ def _maybe_capture_error( "$lib_feature": "ai.prompts", "prompt_name": name, "prompt_version": version, + "prompt_label": label, "posthog_host": self._host, }, ) @@ -448,7 +458,8 @@ def _fetch_prompt_from_api( label is given, fetches the latest Returns: - The validated API response dict containing prompt, name, and version + The validated API response dict containing prompt, name, version, + and label (when fetched by label) Raises: Exception: If the prompt cannot be fetched From ab7e6b135f39fa85f0abecdb050d0a9139a5aa86 Mon Sep 17 00:00:00 2001 From: Juraj Majerik Date: Thu, 16 Jul 2026 15:15:23 +0200 Subject: [PATCH 3/5] chore: update public api snapshot --- references/public_api_snapshot.txt | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index f7468453..fec1c0a0 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -404,11 +404,13 @@ attribute posthog.ai.otel.spans.AI_SPAN_PREFIXES = ('gen_ai.', 'llm.', 'ai.', 't attribute posthog.ai.otel.spans.DEFAULT_HOST = 'https://us.i.posthog.com' attribute posthog.ai.prompts.APP_ENDPOINT = 'https://us.posthog.com' attribute posthog.ai.prompts.CachedPrompt.fetched_at = fetched_at +attribute posthog.ai.prompts.CachedPrompt.label = label attribute posthog.ai.prompts.CachedPrompt.name = name attribute posthog.ai.prompts.CachedPrompt.prompt = prompt attribute posthog.ai.prompts.CachedPrompt.version = version attribute posthog.ai.prompts.DEFAULT_CACHE_TTL_SECONDS = 300 -attribute posthog.ai.prompts.PromptCacheKey = tuple[str, Optional[int]] +attribute posthog.ai.prompts.PromptCacheKey = tuple[str, Optional[int], Optional[str]] +attribute posthog.ai.prompts.PromptResult.label: Optional[str] = None attribute posthog.ai.prompts.PromptResult.name: Optional[str] = None attribute posthog.ai.prompts.PromptResult.prompt: str attribute posthog.ai.prompts.PromptResult.source: PromptSource @@ -844,8 +846,8 @@ class posthog.ai.openai.openai_providers.AzureOpenAI(posthog_client: Optional[Po class posthog.ai.openai_agents.processor.PostHogTracingProcessor(client: Optional[Client] = None, distinct_id: Optional[Union[str, Callable[[Trace], Optional[str]]]] = None, privacy_mode: bool = False, groups: Optional[Dict[str, Any]] = None, properties: Optional[Dict[str, Any]] = None) class posthog.ai.otel.exporter.PostHogTraceExporter(api_key: str, host: str = DEFAULT_HOST) class posthog.ai.otel.processor.PostHogSpanProcessor(api_key: str, host: str = DEFAULT_HOST) -class posthog.ai.prompts.CachedPrompt(prompt: str, fetched_at: float, name: str, version: int) -class posthog.ai.prompts.PromptResult(source: PromptSource, prompt: str, name: Optional[str] = None, version: Optional[int] = None) +class posthog.ai.prompts.CachedPrompt(prompt: str, fetched_at: float, name: str, version: int, label: Optional[str] = None) +class posthog.ai.prompts.PromptResult(source: PromptSource, prompt: str, name: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None) class posthog.ai.prompts.Prompts(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, capture_errors: bool = False) class posthog.ai.stream.AsyncStreamWrapper(generator: AsyncGenerator[T, None], stream: Optional[Any] = None) class posthog.ai.types.FormattedFunctionCall @@ -1178,7 +1180,7 @@ method posthog.ai.otel.processor.PostHogSpanProcessor.on_start(span: Span, paren method posthog.ai.otel.processor.PostHogSpanProcessor.shutdown() -> None method posthog.ai.prompts.Prompts.clear_cache(name: Optional[str] = None, *, version: Optional[int] = None) -> None method posthog.ai.prompts.Prompts.compile(prompt: str, variables: PromptVariables) -> str -method posthog.ai.prompts.Prompts.get(name: str, *, with_metadata: Optional[bool] = None, cache_ttl_seconds: Optional[int] = None, fallback: Optional[str] = None, version: Optional[int] = None) -> Union[str, PromptResult] +method posthog.ai.prompts.Prompts.get(name: str, *, with_metadata: Optional[bool] = None, cache_ttl_seconds: Optional[int] = None, fallback: Optional[str] = None, version: Optional[int] = None, label: Optional[str] = None) -> Union[str, PromptResult] method posthog.bucketed_rate_limiter.BucketedRateLimiter.consume_rate_limit(key: Hashable) -> bool method posthog.bucketed_rate_limiter.BucketedRateLimiter.stop() -> None method posthog.client.Client.alias(previous_id: str, distinct_id: Optional[str], timestamp: Optional[Union[datetime, str]] = None, uuid: Optional[str] = None, disable_geoip: Optional[bool] = None) -> Optional[str] From df107893c4d82fae1312a97895c2ebf59f30562c Mon Sep 17 00:00:00 2001 From: Juraj Majerik Date: Fri, 17 Jul 2026 11:29:19 +0200 Subject: [PATCH 4/5] chore: add changeset for prompt label fetching --- .sampo/changesets/prompts-fetch-by-label.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .sampo/changesets/prompts-fetch-by-label.md diff --git a/.sampo/changesets/prompts-fetch-by-label.md b/.sampo/changesets/prompts-fetch-by-label.md new file mode 100644 index 00000000..072c588c --- /dev/null +++ b/.sampo/changesets/prompts-fetch-by-label.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: minor +--- + +Add a `label` option to `Prompts.get()` to fetch the prompt version a label (e.g. `production`) currently points to. Labeled fetches are cached separately, and `PromptResult` carries the resolved `label`. Requires a PostHog version with prompt labels; older servers ignore the parameter and return the latest version. From e8c0be0e960102cffa7d23b34ca9ca1e9f14da9a Mon Sep 17 00:00:00 2001 From: Juraj Majerik Date: Fri, 17 Jul 2026 11:43:43 +0200 Subject: [PATCH 5/5] feat(prompts): warn when server does not resolve requested label --- posthog/ai/prompts.py | 12 ++++++++++++ posthog/test/ai/test_prompts.py | 17 +++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/posthog/ai/prompts.py b/posthog/ai/prompts.py index f838bbb1..dd35a2e3 100644 --- a/posthog/ai/prompts.py +++ b/posthog/ai/prompts.py @@ -325,6 +325,18 @@ def _get_internal( try: data = self._fetch_prompt_from_api(name, version, label) + # An older PostHog server ignores the label param and returns the latest + # version with no label field — surface that instead of failing silently. + if label is not None and data.get("label") != label: + log.warning( + "[PostHog Prompts] Requested label %r for prompt %r but the server " + "resolved %r. It may not support prompt labels yet and returned the " + "latest version instead.", + label, + name, + data.get("label"), + ) + # Update cache self._cache[cache_key] = CachedPrompt( prompt=data["prompt"], diff --git a/posthog/test/ai/test_prompts.py b/posthog/test/ai/test_prompts.py index 6f4df94c..753405b5 100644 --- a/posthog/test/ai/test_prompts.py +++ b/posthog/test/ai/test_prompts.py @@ -184,6 +184,23 @@ def test_fetch_by_label_passes_param_and_returns_label_metadata( self.assertEqual(result.version, 3) self.assertEqual(result.label, "production") + @patch("posthog.ai.prompts._get_session") + @patch("posthog.ai.prompts.log") + def test_warn_when_server_does_not_resolve_requested_label( + self, mock_log, mock_get_session + ): + """Should warn when a labeled fetch gets a response without that label (old server).""" + mock_get = mock_get_session.return_value.get + mock_get.return_value = MockResponse(json_data=self.mock_prompt_response) + + prompts = Prompts(self.create_mock_posthog()) + result = prompts.get("test-prompt", label="production", with_metadata=True) + + self.assertEqual(result.prompt, self.mock_prompt_response["prompt"]) + self.assertIsNone(result.label) + mock_log.warning.assert_called_once() + self.assertIn("may not support prompt labels", mock_log.warning.call_args[0][0]) + def test_version_and_label_together_raises(self): """Should reject version and label passed together.""" prompts = Prompts(self.create_mock_posthog())