mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-26 01:12:21 +00:00
fix(langfuse): coerce generation names, keep v2 release, timeout and retry defaults, refresh stale prompts off the loop
A non-string metadata generation_name reached the OTLP encoder and took the whole batch down; it is now exported as its text and the exporter drops only the span the encoder rejects. LANGFUSE_RELEASE falls back to the deploy platform's commit variable again, the export deadline is back to the v2 default of 20 s and LANGFUSE_MAX_RETRIES sizes the retry ladder. An expired prompt is served at once while one background thread refreshes it, a re-acquired export channel cancels the pending retire timer, and flush reports delivery rather than a drained queue Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng
parent
548c486372
commit
148b8f19d2
6 changed files with 362 additions and 50 deletions
|
|
@ -284,6 +284,7 @@ class LangFuseLogger:
|
|||
f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\033[0m"
|
||||
) from e
|
||||
raise_if_unsupported_langfuse_version(self.langfuse_sdk_version)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import configured_release
|
||||
|
||||
self.public_key, self.secret_key, self.langfuse_host = resolve_langfuse_credentials(
|
||||
langfuse_public_key=langfuse_public_key,
|
||||
|
|
@ -297,7 +298,7 @@ class LangFuseLogger:
|
|||
self.langfuse_environment: str | None = _env_override
|
||||
else:
|
||||
self.langfuse_environment = self.resolve_deployment_environment()
|
||||
self.langfuse_release = os.getenv("LANGFUSE_RELEASE")
|
||||
self.langfuse_release = configured_release()
|
||||
self.langfuse_debug = parse_langfuse_debug(os.getenv("LANGFUSE_DEBUG"))
|
||||
self.langfuse_flush_interval = LangFuseLogger._get_langfuse_flush_interval(flush_interval)
|
||||
|
||||
|
|
@ -826,15 +827,15 @@ class LangFuseLogger:
|
|||
cache_read_input_tokens=cache_read_input_tokens,
|
||||
)
|
||||
|
||||
generation_name = clean_metadata.pop("generation_name", None)
|
||||
if generation_name is None:
|
||||
# if `generation_name` is None, use sensible default values
|
||||
# If using litellm proxy user `key_alias` if not None
|
||||
# If `key_alias` is None, just log `litellm-{call_type}` as the generation name
|
||||
_user_api_key_alias: Final = cast(str | None, clean_metadata.get("user_api_key_alias", None))
|
||||
generation_name = f"litellm-{cast(str, kwargs.get('call_type', 'completion'))}"
|
||||
if _user_api_key_alias is not None:
|
||||
generation_name = f"litellm:{_user_api_key_alias}"
|
||||
requested_generation_name: Final = clean_metadata.pop("generation_name", None)
|
||||
_user_api_key_alias: Final = cast(str | None, clean_metadata.get("user_api_key_alias", None))
|
||||
generation_name: Final = (
|
||||
str(requested_generation_name)
|
||||
if requested_generation_name is not None
|
||||
else f"litellm:{_user_api_key_alias}"
|
||||
if _user_api_key_alias is not None
|
||||
else f"litellm-{cast(str, kwargs.get('call_type', 'completion'))}"
|
||||
)
|
||||
|
||||
if response_obj is not None:
|
||||
system_fingerprint = getattr(response_obj, "system_fingerprint", None)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@
|
|||
Call Hook for LiteLLM Proxy which allows Langfuse prompt management.
|
||||
"""
|
||||
|
||||
import os
|
||||
from functools import lru_cache
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast
|
||||
|
||||
|
|
@ -121,7 +120,7 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
|
|||
self.langfuse_sdk_version = installed_langfuse_version()
|
||||
raise_if_unsupported_langfuse_version(self.langfuse_sdk_version)
|
||||
|
||||
from .langfuse_sdk import acquire_langfuse_tracing
|
||||
from .langfuse_sdk import acquire_langfuse_tracing, configured_release
|
||||
|
||||
self.api_client = langfuse_client_init(
|
||||
langfuse_public_key=langfuse_public_key,
|
||||
|
|
@ -139,7 +138,7 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
|
|||
secret_key=str(self.secret_key),
|
||||
base_url=self.langfuse_host,
|
||||
environment=LangFuseLogger.resolve_deployment_environment(),
|
||||
release=os.getenv("LANGFUSE_RELEASE"),
|
||||
release=configured_release(),
|
||||
flush_interval=LangFuseLogger._get_langfuse_flush_interval(flush_interval), # pyright: ignore[reportPrivateUsage] # shared env-fallback helper, not part of the logger's API
|
||||
mock_mode=should_use_langfuse_mock(),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -46,7 +46,10 @@ __all__ = (
|
|||
"build_langfuse_client",
|
||||
"build_langfuse_tracing",
|
||||
"configured_flush_at",
|
||||
"configured_max_retries",
|
||||
"configured_release",
|
||||
"configured_sample_rate",
|
||||
"configured_timeout",
|
||||
"flush_langfuse_tracing",
|
||||
"observation_attributes",
|
||||
"release_langfuse_tracing",
|
||||
|
|
@ -64,6 +67,20 @@ _TRACER_NAME: Final = "langfuse-sdk"
|
|||
_MAX_QUEUE_SIZE: Final = 100_000
|
||||
_DEFAULT_FLUSH_AT: Final = 512
|
||||
_CHANNEL_RETIRE_GRACE_SECONDS: Final = 60.0
|
||||
_DEFAULT_TIMEOUT_SECONDS: Final = 20.0
|
||||
_DEFAULT_MAX_RETRIES: Final = 3
|
||||
_COMMON_RELEASE_ENVS: Final = (
|
||||
"RENDER_GIT_COMMIT",
|
||||
"CI_COMMIT_SHA",
|
||||
"CIRCLE_SHA1",
|
||||
"SOURCE_VERSION",
|
||||
"TRAVIS_COMMIT",
|
||||
"GIT_COMMIT",
|
||||
"GITHUB_SHA",
|
||||
"BITBUCKET_COMMIT",
|
||||
"BUILD_SOURCEVERSION",
|
||||
"DRONE_COMMIT_SHA",
|
||||
)
|
||||
_SPAN_LIMITS: Final = SpanLimits(
|
||||
max_attributes=SpanLimits.UNSET,
|
||||
max_events=128,
|
||||
|
|
@ -429,6 +446,34 @@ def configured_sample_rate() -> float:
|
|||
return parsed
|
||||
|
||||
|
||||
def configured_timeout() -> float:
|
||||
"""``LANGFUSE_TIMEOUT`` in seconds for every export and REST call, the v2 SDK's 20 s when unset.
|
||||
|
||||
A value that is not a number raises, as the v2 client did at construction, so a typo is not silently ignored.
|
||||
"""
|
||||
return float(os.environ.get("LANGFUSE_TIMEOUT", _DEFAULT_TIMEOUT_SECONDS))
|
||||
|
||||
|
||||
def configured_max_retries() -> int:
|
||||
"""``LANGFUSE_MAX_RETRIES`` as the number of re-sends after a failed export, the v2 SDK's knob and default."""
|
||||
raw: Final = os.environ.get("LANGFUSE_MAX_RETRIES")
|
||||
if raw is None:
|
||||
return _DEFAULT_MAX_RETRIES
|
||||
if not raw.strip().isdigit():
|
||||
verbose_logger.warning(
|
||||
"LANGFUSE_MAX_RETRIES=%r is not a whole number; retrying %d times", raw, _DEFAULT_MAX_RETRIES
|
||||
)
|
||||
return _DEFAULT_MAX_RETRIES
|
||||
return int(raw)
|
||||
|
||||
|
||||
def configured_release() -> str | None:
|
||||
"""``LANGFUSE_RELEASE``, else the commit variable of the CI or deploy platform, as both SDK generations resolve it."""
|
||||
return os.environ.get("LANGFUSE_RELEASE") or next(
|
||||
(os.environ[name] for name in _COMMON_RELEASE_ENVS if name in os.environ), None
|
||||
)
|
||||
|
||||
|
||||
def configured_flush_at() -> int:
|
||||
"""``LANGFUSE_FLUSH_AT`` as the export batch size, the SDK's own knob, with its default when unset or unusable."""
|
||||
raw: Final = os.environ.get("LANGFUSE_FLUSH_AT")
|
||||
|
|
@ -485,7 +530,12 @@ class LangfuseSpanExporter(SpanExporter):
|
|||
delays: Sequence[float] = (1.0, 2.0, 4.0)
|
||||
|
||||
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
|
||||
body: Final = encode_spans(spans).SerializeToString()
|
||||
body: Final = _encode(spans)
|
||||
if body is None:
|
||||
return SpanExportResult.FAILURE
|
||||
return self._send(body)
|
||||
|
||||
def _send(self, body: bytes) -> SpanExportResult:
|
||||
for delay in self.delays:
|
||||
outcome: _ExportOutcome = self._post(body)
|
||||
if outcome != "retry":
|
||||
|
|
@ -518,8 +568,26 @@ class LangfuseSpanExporter(SpanExporter):
|
|||
return True
|
||||
|
||||
|
||||
def _encode(spans: Sequence[ReadableSpan]) -> bytes | None:
|
||||
"""The OTLP body, or ``None`` when nothing survived: a span the encoder rejects is dropped, not the whole batch."""
|
||||
try:
|
||||
return encode_spans(spans).SerializeToString()
|
||||
except Exception: # noqa: BLE001 # protobuf raises TypeError or ValueError depending on the field
|
||||
kept: Final = tuple(span for span in spans if _encodes(span))
|
||||
verbose_logger.error("Langfuse export dropped %d span(s) the OTLP encoder rejected", len(spans) - len(kept))
|
||||
return encode_spans(kept).SerializeToString() if kept else None
|
||||
|
||||
|
||||
def _encodes(span: ReadableSpan) -> bool:
|
||||
try:
|
||||
encode_spans((span,))
|
||||
except Exception: # noqa: BLE001 # same encoder failure modes as above
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> LangfuseSpanExporter:
|
||||
"""Endpoint, headers and timeout mirror the SDK's span processor so the server treats the spans as v4 SDK traffic."""
|
||||
"""Endpoint, headers, timeout and retries mirror the v2 SDK's so the server treats the spans as SDK traffic."""
|
||||
export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH") or "/api/public/otel/v1/traces"
|
||||
encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii")
|
||||
return LangfuseSpanExporter(
|
||||
|
|
@ -534,7 +602,8 @@ def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) ->
|
|||
"x-langfuse-public-key": public_key,
|
||||
}
|
||||
),
|
||||
timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
timeout=configured_timeout(),
|
||||
delays=tuple(2.0**attempt for attempt in range(configured_max_retries())),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -549,6 +618,26 @@ def _resource(*, environment: str | None, release: str | None) -> Resource:
|
|||
)
|
||||
|
||||
|
||||
class _ExportLedger(SpanExporter):
|
||||
"""Counts the batches the exporter gave up on, so a flush can report delivery rather than a drained queue."""
|
||||
|
||||
def __init__(self, exporter: SpanExporter) -> None:
|
||||
self.exporter: Final = exporter
|
||||
self.failed_batches = 0
|
||||
|
||||
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
|
||||
result: Final = self.exporter.export(spans)
|
||||
if result is not SpanExportResult.SUCCESS:
|
||||
self.failed_batches += 1
|
||||
return result
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.exporter.shutdown()
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30_000) -> bool:
|
||||
return self.exporter.force_flush(timeout_millis)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LangfuseTracing:
|
||||
"""litellm's own export channel to one Langfuse project: a provider, its tracer and the exporter behind them.
|
||||
|
|
@ -559,9 +648,12 @@ class LangfuseTracing:
|
|||
|
||||
provider: TracerProvider
|
||||
tracer: Tracer
|
||||
ledger: _ExportLedger
|
||||
|
||||
def flush(self, timeout_millis: int = 30_000) -> bool:
|
||||
return self.provider.force_flush(timeout_millis)
|
||||
"""``True`` only when the queue drained in time and every batch it held was accepted by the destination."""
|
||||
failed_before: Final = self.ledger.failed_batches
|
||||
return self.provider.force_flush(timeout_millis) and self.ledger.failed_batches == failed_before
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.provider.shutdown()
|
||||
|
|
@ -584,6 +676,7 @@ class _TracingKey:
|
|||
class _Lease:
|
||||
tracing: LangfuseTracing
|
||||
holders: int
|
||||
retire: threading.Timer | None = None
|
||||
|
||||
|
||||
_TRACING_LOCK: Final = threading.Lock()
|
||||
|
|
@ -619,7 +712,9 @@ def acquire_langfuse_tracing(
|
|||
with _TRACING_LOCK:
|
||||
cached: Final = _TRACING.get(key)
|
||||
if cached is not None:
|
||||
_TRACING[key] = replace(cached, holders=cached.holders + 1)
|
||||
if cached.retire is not None:
|
||||
cached.retire.cancel()
|
||||
_TRACING[key] = replace(cached, holders=cached.holders + 1, retire=None)
|
||||
return cached.tracing
|
||||
created: Final = build_langfuse_tracing(
|
||||
exporter=DiscardingSpanExporter()
|
||||
|
|
@ -648,27 +743,30 @@ def release_langfuse_tracing(tracing: LangfuseTracing, *, grace_seconds: float =
|
|||
key, lease = held
|
||||
if lease.holders <= 0:
|
||||
return
|
||||
_TRACING[key] = replace(lease, holders=lease.holders - 1)
|
||||
if lease.holders > 1:
|
||||
_TRACING[key] = replace(lease, holders=lease.holders - 1)
|
||||
return
|
||||
if grace_seconds <= 0:
|
||||
_retire_if_unheld(key, tracing)
|
||||
return
|
||||
retire: Final = threading.Timer(grace_seconds, _retire_if_unheld, args=(key, tracing))
|
||||
retire.name = "langfuse-retire"
|
||||
retire.daemon = True
|
||||
retire.start()
|
||||
|
||||
|
||||
def _retire_if_unheld(key: _TracingKey, tracing: LangfuseTracing) -> None:
|
||||
with _TRACING_LOCK:
|
||||
lease: Final = _TRACING.get(key)
|
||||
if lease is None or lease.tracing is not tracing or lease.holders > 0:
|
||||
if grace_seconds > 0:
|
||||
retire: Final = threading.Timer(grace_seconds, lambda: _retire_unless_reacquired(key, retire))
|
||||
retire.name = "langfuse-retire"
|
||||
retire.daemon = True
|
||||
_TRACING[key] = _Lease(tracing=tracing, holders=0, retire=retire)
|
||||
retire.start()
|
||||
return
|
||||
del _TRACING[key]
|
||||
tracing.shutdown()
|
||||
|
||||
|
||||
def _retire_unless_reacquired(key: _TracingKey, timer: threading.Timer) -> None:
|
||||
"""Only the timer the lease still points at may retire it; a re-acquire cancels and clears the pending one."""
|
||||
with _TRACING_LOCK:
|
||||
lease: Final = _TRACING.get(key)
|
||||
if lease is None or lease.retire is not timer:
|
||||
return
|
||||
del _TRACING[key]
|
||||
lease.tracing.shutdown()
|
||||
|
||||
|
||||
class _FlushWorker(threading.Thread):
|
||||
"""Daemon, so a channel still blocked at the deadline cannot hold up interpreter exit."""
|
||||
|
||||
|
|
@ -725,15 +823,16 @@ def build_langfuse_tracing(
|
|||
id_generator=_RequestedIdGenerator(),
|
||||
span_limits=_SPAN_LIMITS,
|
||||
)
|
||||
ledger: Final = _ExportLedger(exporter)
|
||||
provider.add_span_processor(
|
||||
BatchSpanProcessor(
|
||||
exporter,
|
||||
ledger,
|
||||
max_queue_size=_MAX_QUEUE_SIZE,
|
||||
max_export_batch_size=flush_at,
|
||||
schedule_delay_millis=flush_interval_millis,
|
||||
)
|
||||
)
|
||||
return LangfuseTracing(provider=provider, tracer=provider.get_tracer(_TRACER_NAME))
|
||||
return LangfuseTracing(provider=provider, tracer=provider.get_tracer(_TRACER_NAME), ledger=ledger)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -758,8 +857,9 @@ class LangfuseApiClient:
|
|||
over ``LangfuseTracing``; nothing here exports spans.
|
||||
|
||||
Prompts are cached for ``LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS`` (60 by default) as the SDK
|
||||
does, refreshed on the request that finds them stale; a refresh that fails keeps serving the
|
||||
stale prompt rather than failing the request, again like the SDK.
|
||||
does. A stale prompt is served at once and refreshed on a background thread, so the request
|
||||
that finds it stale, and the event loop it runs on, never wait for the REST round trip; a
|
||||
refresh that fails keeps serving the stale prompt rather than failing the request, again like the SDK.
|
||||
"""
|
||||
|
||||
def __init__(self, api: LangfuseAPI, *, prompt_cache_ttl_seconds: float) -> None:
|
||||
|
|
@ -767,6 +867,8 @@ class LangfuseApiClient:
|
|||
self.prompt_cache_ttl_seconds: Final = prompt_cache_ttl_seconds
|
||||
# mutable-ok: per-client prompt cache, guarded by _lock
|
||||
self._prompts: Final[dict[_PromptKey, _CachedPrompt]] = {}
|
||||
# mutable-ok: keys with a refresh in flight, guarded by _lock
|
||||
self._refreshing: Final[set[_PromptKey]] = set()
|
||||
self._lock: Final = threading.Lock()
|
||||
|
||||
def auth_check(self) -> bool:
|
||||
|
|
@ -784,19 +886,35 @@ class LangfuseApiClient:
|
|||
key: Final[_PromptKey] = (name, version, label)
|
||||
with self._lock:
|
||||
cached: Final = self._prompts.get(key)
|
||||
if cached is not None and monotonic() - cached.fetched_at < self.prompt_cache_ttl_seconds:
|
||||
return cached.prompt
|
||||
try:
|
||||
fetched: Final = _prompt_client(self.api.prompts.get(name, version=version, label=label))
|
||||
except Exception as error:
|
||||
if cached is None:
|
||||
raise
|
||||
verbose_logger.warning("Langfuse prompt %r refresh failed, serving the cached version: %s", name, error)
|
||||
return cached.prompt
|
||||
if cached is None:
|
||||
return self._fetch(key)
|
||||
if monotonic() - cached.fetched_at >= self.prompt_cache_ttl_seconds:
|
||||
self._refresh_in_background(key)
|
||||
return cached.prompt
|
||||
|
||||
def _fetch(self, key: _PromptKey) -> PromptClient:
|
||||
name, version, label = key
|
||||
fetched: Final = _prompt_client(self.api.prompts.get(name, version=version, label=label))
|
||||
with self._lock:
|
||||
self._prompts[key] = _CachedPrompt(prompt=fetched, fetched_at=monotonic())
|
||||
return fetched
|
||||
|
||||
def _refresh_in_background(self, key: _PromptKey) -> None:
|
||||
with self._lock:
|
||||
if key in self._refreshing:
|
||||
return
|
||||
self._refreshing.add(key)
|
||||
threading.Thread(target=self._refresh, args=(key,), name="langfuse-prompt-refresh", daemon=True).start()
|
||||
|
||||
def _refresh(self, key: _PromptKey) -> None:
|
||||
try:
|
||||
self._fetch(key)
|
||||
except Exception as error: # noqa: BLE001 # a failed refresh keeps the stale prompt in service
|
||||
verbose_logger.warning("Langfuse prompt %r refresh failed, serving the cached version: %s", key[0], error)
|
||||
finally:
|
||||
with self._lock:
|
||||
self._refreshing.discard(key)
|
||||
|
||||
|
||||
def build_langfuse_client(
|
||||
*,
|
||||
|
|
@ -819,7 +937,7 @@ def build_langfuse_client(
|
|||
x_langfuse_sdk_version=version("langfuse"),
|
||||
x_langfuse_public_key=public_key,
|
||||
httpx_client=httpx_client,
|
||||
timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
timeout=configured_timeout(),
|
||||
),
|
||||
prompt_cache_ttl_seconds=float(os.getenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", "60")),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1058,7 +1058,8 @@ async def proxy_shutdown_event(worker_heartbeat: ProxyWorkerHeartbeat | None = N
|
|||
verbose_proxy_logger.info("Langfuse export channels flushed")
|
||||
else:
|
||||
verbose_proxy_logger.warning(
|
||||
"Langfuse export did not finish within %dms; remaining spans are left to the background exporter",
|
||||
"Langfuse shutdown flush incomplete: a channel did not finish within %dms or a batch was rejected "
|
||||
"(see the export errors above); remaining spans are left to the background exporter",
|
||||
LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 # shutdown must continue even if the flush fails
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import threading
|
|||
import uuid
|
||||
from base64 import b64encode
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from time import monotonic, sleep
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -56,6 +57,12 @@ TRACE_A = "a" * 32
|
|||
PARENT_C = "c" * 16
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _own_channel_registry(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Channels leaked by other test modules would otherwise take part in every process-wide flush here."""
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk._TRACING", {})
|
||||
|
||||
|
||||
@pytest.fixture(name="channel")
|
||||
def _channel() -> tuple[LangfuseTracing, InMemorySpanExporter]:
|
||||
exporter = InMemorySpanExporter()
|
||||
|
|
@ -725,6 +732,47 @@ def test_channel_reacquired_within_the_grace_is_kept(monkeypatch: pytest.MonkeyP
|
|||
assert len(exporter.get_finished_spans()) == 1
|
||||
|
||||
|
||||
def test_retire_timer_of_an_earlier_release_cannot_kill_a_reacquired_channel(monkeypatch: pytest.MonkeyPatch):
|
||||
"""release, re-acquire, release: the first timer used to fire into a channel that a later holder still
|
||||
counted on for its own grace period, shutting the batch thread down while spans were still queued."""
|
||||
tracing, exporter = _acquire_recorded(monkeypatch, "pk-lease-race-test")
|
||||
|
||||
release_langfuse_tracing(tracing, grace_seconds=0.2)
|
||||
assert _acquire(public_key="pk-lease-race-test", mock_mode=False, flush_interval=600.0) is tracing
|
||||
release_langfuse_tracing(tracing, grace_seconds=600.0)
|
||||
|
||||
threading.Event().wait(0.5)
|
||||
assert exporter.shutdowns == 0
|
||||
assert _acquire(public_key="pk-lease-race-test", mock_mode=False, flush_interval=600.0) is tracing
|
||||
tracing.tracer.start_span("generation").end()
|
||||
assert tracing.flush() is True
|
||||
assert len(exporter.get_finished_spans()) == 1
|
||||
|
||||
|
||||
class _RejectsEverything(SpanExporter):
|
||||
def export(self, spans) -> SpanExportResult:
|
||||
return SpanExportResult.FAILURE
|
||||
|
||||
def shutdown(self) -> None:
|
||||
return None
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30_000) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def test_flush_is_false_when_the_destination_rejected_a_batch(monkeypatch: pytest.MonkeyPatch):
|
||||
"""The shutdown hook logs "channels flushed" off this value; a drained queue whose batches all
|
||||
failed at the destination is a loss, not a flush."""
|
||||
monkeypatch.setattr(
|
||||
"litellm.integrations.langfuse.langfuse_sdk._build_span_exporter", lambda **_: _RejectsEverything()
|
||||
)
|
||||
tracing = _acquire(public_key="pk-flush-truth-test", mock_mode=False, flush_interval=600.0)
|
||||
tracing.tracer.start_span("generation").end()
|
||||
|
||||
assert tracing.flush() is False
|
||||
assert flush_langfuse_tracing() is True, "an empty queue after the loss has nothing left to fail"
|
||||
|
||||
|
||||
def test_release_of_a_channel_the_registry_never_handed_out_is_a_no_op():
|
||||
exporter = InMemorySpanExporter()
|
||||
tracing = build_langfuse_tracing(
|
||||
|
|
@ -1012,19 +1060,56 @@ def test_exporter_does_not_retry_a_rejected_batch(monkeypatch, status):
|
|||
assert slept == []
|
||||
|
||||
|
||||
def _finished_span_named(name: object):
|
||||
provider = TracerProvider()
|
||||
span = provider.get_tracer("t").start_span("placeholder")
|
||||
span._name = name # pyright: ignore[reportAttributeAccessIssue, reportPrivateUsage] # the SDK only stores str
|
||||
span.end()
|
||||
return span
|
||||
|
||||
|
||||
def test_exporter_drops_a_span_the_encoder_rejects_and_still_posts_the_rest(monkeypatch, caplog):
|
||||
"""One span the OTLP encoder cannot serialize used to raise out of ``export`` and lose every span in the batch."""
|
||||
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
|
||||
|
||||
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", lambda _: None)
|
||||
exporter, seen = _exporter_over([200])
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger="LiteLLM"):
|
||||
result = exporter.export((_finished_span(), _finished_span_named(12345), _finished_span()))
|
||||
|
||||
assert result is SpanExportResult.SUCCESS
|
||||
(request,) = seen
|
||||
decoded = ExportTraceServiceRequest()
|
||||
decoded.ParseFromString(request.content)
|
||||
assert [span.name for span in decoded.resource_spans[0].scope_spans[0].spans] == ["generation", "generation"]
|
||||
assert "dropped 1 span(s)" in caplog.text
|
||||
|
||||
|
||||
def test_exporter_reports_failure_when_no_span_of_the_batch_can_be_encoded(monkeypatch):
|
||||
exporter, seen = _exporter_over([200])
|
||||
|
||||
assert exporter.export((_finished_span_named(12345),)) is SpanExportResult.FAILURE
|
||||
assert seen == []
|
||||
|
||||
|
||||
def test_built_exporter_uses_the_shared_litellm_handler_and_langfuse_headers(monkeypatch):
|
||||
"""No private requests session or TLS adapter: the channel is the same handler the rest of litellm uses."""
|
||||
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
|
||||
|
||||
monkeypatch.delenv("LANGFUSE_TIMEOUT", raising=False)
|
||||
monkeypatch.delenv("LANGFUSE_MAX_RETRIES", raising=False)
|
||||
default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
|
||||
assert default.handler is _get_httpx_client()
|
||||
assert default.timeout == 5
|
||||
assert default.timeout == 20
|
||||
assert len(default.delays) == 3
|
||||
|
||||
monkeypatch.setenv("LANGFUSE_TIMEOUT", "20")
|
||||
monkeypatch.setenv("LANGFUSE_TIMEOUT", "7.5")
|
||||
monkeypatch.setenv("LANGFUSE_MAX_RETRIES", "1")
|
||||
exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
|
||||
assert exporter.endpoint == "https://lf.internal.example/api/public/otel/v1/traces"
|
||||
assert exporter.timeout == 20
|
||||
assert exporter.timeout == 7.5
|
||||
assert exporter.delays == (1.0,)
|
||||
assert exporter.headers["Authorization"] == "Basic " + b64encode(b"pk:sk").decode()
|
||||
assert exporter.headers["x-langfuse-public-key"] == "pk"
|
||||
assert exporter.headers["x-langfuse-sdk-version"] == installed_langfuse_version()
|
||||
|
|
@ -1091,6 +1176,78 @@ class _RecordingPromptsApi:
|
|||
)
|
||||
|
||||
|
||||
class _BlockingPromptsApi(_RecordingPromptsApi):
|
||||
"""Every fetch after the first blocks until the test releases it, and may be told to fail."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.release = threading.Event()
|
||||
self.fail_refresh = False
|
||||
|
||||
def get(self, name: str, *, version: int | None, label: str | None):
|
||||
is_refresh = bool(self.requests)
|
||||
prompt = super().get(name, version=version, label=label)
|
||||
if is_refresh:
|
||||
assert self.release.wait(5), "refresh was never released"
|
||||
if self.fail_refresh:
|
||||
raise RuntimeError("langfuse is down")
|
||||
return prompt
|
||||
|
||||
|
||||
def _wait_until(predicate, timeout: float = 5.0) -> None:
|
||||
for _ in range(int(timeout / 0.01)):
|
||||
if predicate():
|
||||
return
|
||||
sleep(0.01)
|
||||
raise AssertionError("condition not met in time")
|
||||
|
||||
|
||||
def test_stale_prompt_is_served_at_once_while_the_refresh_runs_elsewhere():
|
||||
"""``get_prompt`` runs on the proxy's event loop; a stale entry used to refetch inline and block every
|
||||
request on the REST round trip. The stale prompt is returned immediately and refreshed off-thread."""
|
||||
api = _BlockingPromptsApi()
|
||||
client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0) # pyright: ignore[reportArgumentType] # duck-typed prompts API
|
||||
|
||||
first = client.get_prompt("greeting")
|
||||
started = monotonic()
|
||||
stale = client.get_prompt("greeting")
|
||||
|
||||
assert stale is first, "the stale prompt must come back without waiting on the refresh"
|
||||
assert monotonic() - started < 1.0, "the stale read waited on the blocked refresh"
|
||||
_wait_until(lambda: len(api.requests) == 2)
|
||||
api.release.set()
|
||||
_wait_until(lambda: client.get_prompt("greeting") is not first)
|
||||
|
||||
|
||||
def test_a_failed_background_refresh_keeps_the_stale_prompt_in_service(caplog):
|
||||
api = _BlockingPromptsApi()
|
||||
api.fail_refresh = True
|
||||
client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0) # pyright: ignore[reportArgumentType] # duck-typed prompts API
|
||||
|
||||
first = client.get_prompt("greeting")
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
|
||||
started = monotonic()
|
||||
assert client.get_prompt("greeting") is first
|
||||
assert monotonic() - started < 1.0, "the stale read waited on the blocked refresh"
|
||||
api.release.set()
|
||||
_wait_until(lambda: "refresh failed" in caplog.text)
|
||||
assert client.get_prompt("greeting") is first
|
||||
|
||||
|
||||
def test_only_one_refresh_runs_for_a_stale_prompt_under_concurrent_reads():
|
||||
api = _BlockingPromptsApi()
|
||||
client = LangfuseApiClient(api, prompt_cache_ttl_seconds=0.3) # pyright: ignore[reportArgumentType] # duck-typed prompts API
|
||||
|
||||
first = client.get_prompt("greeting")
|
||||
sleep(0.3)
|
||||
for _ in range(20):
|
||||
assert client.get_prompt("greeting") is first
|
||||
_wait_until(lambda: len(api.requests) == 2)
|
||||
api.release.set()
|
||||
_wait_until(lambda: client.get_prompt("greeting") is not first)
|
||||
assert len(api.requests) == 2
|
||||
|
||||
|
||||
def test_prompt_cache_keeps_a_missing_label_apart_from_the_label_named_none():
|
||||
"""A prompt labelled ``"None"`` and the unlabelled default are different prompts in Langfuse
|
||||
and must not answer each other's requests from the cache."""
|
||||
|
|
|
|||
|
|
@ -1339,6 +1339,42 @@ def test_langfuse_environment_falls_back_to_deployment_env_var(monkeypatch):
|
|||
assert _exported_environment(logger) == "deployment-wide"
|
||||
|
||||
|
||||
def _exported_release(logger: LangFuseLogger):
|
||||
from langfuse import LangfuseOtelSpanAttributes
|
||||
|
||||
return logger.tracing.provider.resource.attributes.get(LangfuseOtelSpanAttributes.RELEASE)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("platform_var", ["GITHUB_SHA", "CI_COMMIT_SHA", "RENDER_GIT_COMMIT", "SOURCE_VERSION"])
|
||||
def test_release_falls_back_to_the_deploy_platforms_commit_variable(monkeypatch, platform_var):
|
||||
"""Deployments that never set ``LANGFUSE_RELEASE`` still got a release on every trace from the v2 SDK, which
|
||||
read the CI or hosting platform's commit variable; dropping that silently blanked their release filter."""
|
||||
from litellm.integrations.langfuse.langfuse_sdk import _COMMON_RELEASE_ENVS
|
||||
|
||||
for name in ("LANGFUSE_RELEASE", *_COMMON_RELEASE_ENVS):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
monkeypatch.setenv(platform_var, "deadbeef")
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key=f"pk-release-{platform_var}")
|
||||
assert logger.langfuse_release == "deadbeef"
|
||||
assert _exported_release(logger) == "deadbeef"
|
||||
|
||||
|
||||
def test_explicit_langfuse_release_wins_over_the_platform_commit(monkeypatch):
|
||||
monkeypatch.setenv("LANGFUSE_RELEASE", "v9")
|
||||
monkeypatch.setenv("GITHUB_SHA", "deadbeef")
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-release-explicit")
|
||||
assert _exported_release(logger) == "v9"
|
||||
|
||||
|
||||
def test_non_string_generation_name_is_exported_as_its_text(monkeypatch):
|
||||
"""v2 coerced ``generation_name`` through pydantic; a raw int would now fail OTLP encoding and lose the batch."""
|
||||
rig = _steering_logger()
|
||||
|
||||
_, _, span = _emit(rig, metadata={"generation_name": 12345})
|
||||
|
||||
assert span.name == "12345"
|
||||
|
||||
|
||||
def test_dynamic_langfuse_environment_triggers_dynamic_logger():
|
||||
from litellm.integrations.langfuse.langfuse_handler import LangFuseHandler
|
||||
from litellm.types.utils import StandardCallbackDynamicParams
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue