fix(langfuse): retry a cold prompt miss once and drop upstream headers from prompt errors

A cold prompt fetch makes one immediate second attempt after a 5xx or a
transport failure, as the v2 client did, still with the generated client's
sleeping retries and Retry-After handling off so the event loop never stalls.
A failed fetch raises LangfusePromptError carrying only the status and body,
so the proxy no longer forwards Langfuse's response headers to its client

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-23 19:28:30 +00:00
parent 20043aec3f
commit 59a2e81b3a
2 changed files with 102 additions and 10 deletions

View file

@ -9,7 +9,7 @@ from collections.abc import Iterable, Mapping, Sequence
from contextvars import ContextVar
from dataclasses import dataclass, replace
from datetime import datetime
from functools import reduce
from functools import partial, reduce
from hashlib import sha256
from importlib.metadata import version
from itertools import chain
@ -47,6 +47,7 @@ __all__ = (
"DiscardingSpanExporter",
"LangfuseApiClient",
"LangfuseObservation",
"LangfusePromptError",
"LangfuseSpanExporter",
"LangfuseTracing",
"TraceIdHashSampler",
@ -1075,6 +1076,20 @@ def _api_error_reason(error: ApiError) -> str:
return f"status_code: {detail.status_code}, body: {detail.body}"
class LangfusePromptError(Exception):
"""An ``ApiError`` without its ``headers``, which the proxy would otherwise forward to its own client."""
def __init__(self, error: ApiError) -> None:
detail: Final = _ApiErrorDetail.model_validate(error)
super().__init__(f"status_code: {detail.status_code}, body: {detail.body}")
self.status_code: Final = detail.status_code
self.body: Final = detail.body
def _is_server_error(error: ApiError) -> bool:
return error.status_code is not None and error.status_code >= 500
class LangfuseApiClient:
"""litellm's handle on one Langfuse project over its REST API: prompts, ``auth_check`` and the project id.
@ -1129,14 +1144,31 @@ class LangfuseApiClient:
return cached.prompt
def _fetch(self, key: _PromptKey) -> PromptClient:
name, version, label = key
fetched: Final = _prompt_client(
self.api.prompts.get(quote(name, safe=""), version=version, label=label, request_options=_NO_REST_RETRIES)
)
fetched: Final = _prompt_client(self._request_prompt(key))
with self._lock:
self._prompts[key] = _CachedPrompt(prompt=fetched, fetched_at=monotonic())
return fetched
def _request_prompt(self, key: _PromptKey) -> Prompt:
"""Retried once, at once, after a 5xx or a transport failure: a cold miss runs on the caller's event
loop, so the generated client's sleeping retries stay off."""
name, version, label = key
request: Final = partial(
self.api.prompts.get, quote(name, safe=""), version=version, label=label, request_options=_NO_REST_RETRIES
)
try:
return request()
except ApiError as error:
if not _is_server_error(error):
raise LangfusePromptError(error) from None
verbose_logger.debug("Langfuse prompt %r fetch failed (%s), retrying once", name, _api_error_reason(error))
except httpx.TransportError as error:
verbose_logger.debug("Langfuse prompt %r fetch failed (%s), retrying once", name, error)
try:
return request()
except ApiError as error:
raise LangfusePromptError(error) from None
def _refresh_in_background(self, key: _PromptKey) -> None:
with self._lock:
if key in self._refreshing:

View file

@ -34,6 +34,7 @@ from litellm.integrations.langfuse.langfuse import (
from litellm.integrations.langfuse.langfuse_sdk import (
DiscardingSpanExporter,
LangfuseApiClient,
LangfusePromptError,
LangfuseSpanExporter,
LangfuseTracing,
_build_span_exporter,
@ -1065,10 +1066,15 @@ def test_auth_check_and_project_id_make_one_round_trip_when_langfuse_is_down(sta
assert monotonic() - started < 0.5
@pytest.mark.parametrize("status", [500, 503, 429], ids=["http-500", "http-503", "http-429"])
def test_cold_prompt_miss_makes_one_round_trip_when_langfuse_is_down(status: int):
@pytest.mark.parametrize(
("status", "round_trips"),
[(500, 2), (503, 2), (429, 1), (404, 1)],
ids=["http-500", "http-503", "http-429", "http-404"],
)
def test_cold_prompt_miss_never_sleeps_when_langfuse_is_down(status: int, round_trips: int):
"""A cold ``get_prompt`` fetches inline on the event loop; with the generated client's default retries a
429 carrying ``Retry-After: 30`` used to hold the loop for a minute."""
429 carrying ``Retry-After: 30`` used to hold the loop for a minute. A 5xx gets the v2 client's one
quick retry, a 429 or 4xx none."""
requests: list[httpx.Request] = []
def fail(request: httpx.Request) -> httpx.Response:
@ -1083,10 +1089,64 @@ def test_cold_prompt_miss_makes_one_round_trip_when_langfuse_is_down(status: int
)
started = monotonic()
with pytest.raises(ApiError):
with pytest.raises(LangfusePromptError) as caught:
client.get_prompt("greeting")
assert len(requests) == 1
assert len(requests) == round_trips
assert monotonic() - started < 0.5
assert caught.value.status_code == status
@pytest.mark.parametrize("first_failure", [503, "connect-error"], ids=["http-503", "connect-error"])
def test_one_transient_failure_on_a_cold_prompt_miss_does_not_fail_the_call(first_failure: int | str):
"""The v2 client retried a cold fetch once; a single Langfuse blip must not fail the LLM call."""
requests: list[httpx.Request] = []
def flaky(request: httpx.Request) -> httpx.Response:
requests.append(request)
if len(requests) > 1:
return httpx.Response(200, request=request, json=_TEXT_PROMPT_BODY)
if isinstance(first_failure, int):
return httpx.Response(first_failure, request=request, json={"message": "down"})
raise httpx.ConnectError("refused", request=request)
client = build_langfuse_client(
public_key="pk",
secret_key="sk",
base_url="http://127.0.0.1:1",
httpx_client=httpx.Client(transport=httpx.MockTransport(flaky)),
)
started = monotonic()
assert client.get_prompt("greeting").compile() == "hello"
assert len(requests) == 2
assert monotonic() - started < 0.5
assert client.get_prompt("greeting").compile() == "hello", "the retried prompt is cached like any other"
assert len(requests) == 2
def test_prompt_fetch_error_carries_status_and_body_but_no_upstream_headers():
"""The proxy forwards an exception's ``headers`` to its client and prints ``str(e)``; the generated
``ApiError`` carries Langfuse's response headers in both."""
upstream_headers = {"server": "langfuse-edge", "set-cookie": "session=abc; HttpOnly", "x-upstream-internal": "1"}
def not_found(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, request=request, headers=upstream_headers, json={"message": "Prompt not found"})
client = build_langfuse_client(
public_key="pk",
secret_key="sk",
base_url="http://127.0.0.1:1",
httpx_client=httpx.Client(transport=httpx.MockTransport(not_found)),
)
with pytest.raises(Exception, match="Prompt not found") as caught:
client.get_prompt("missing")
error = caught.value
assert getattr(error, "headers", None) is None
assert getattr(error, "status_code", None) == 404
assert not any(header in str(error) for header in upstream_headers)
assert error.__cause__ is None and error.__suppress_context__, "the header-bearing ApiError must not ride along"
_TEXT_PROMPT_BODY: Final[dict[str, object]] = {