scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 48

feat(langfuse): migrate the sdk callback to langfuse v4

Browse source 
Replace the v2 trace()/generation()/span() calls with SDK v4 observations exported over OpenTelemetry, with one isolated tracer provider per Langfuse credential set, a discarding exporter for mock mode, and v4 trace and observation id normalization. Keeps the session-header trace provenance logic from main so each call under a session alias still gets its own trace

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-14 20:42:36 +00:00
parent 351a54e849
commit 0d93439762
15 changed files with 2503 additions and 485 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docker/build_from_pip/Dockerfile.build_from_pip
View file

@ -32,10 +32,10 @@ RUN uv venv --python python && \
"anthropic[vertex]==0.84.0" \
"grpcio==1.78.0" \
"prometheus-client==0.20.0" \
"langfuse==2.59.7" \
"opentelemetry-api==1.28.0" \
"opentelemetry-sdk==1.28.0" \
"opentelemetry-exporter-otlp==1.28.0" \
"langfuse==4.15.1" \
"opentelemetry-api==1.33.1" \
"opentelemetry-sdk==1.33.1" \
"opentelemetry-exporter-otlp==1.33.1" \
"ddtrace==4.11.0" \
"sentry-sdk==2.21.0" \
"mangum==0.17.0" \

5
litellm/integrations/SlackAlerting/utils.py
View file

@ -84,7 +84,8 @@ async def _add_langfuse_trace_id_to_alert(
#########################################################
langfuse_object: Final = litellm_logging_obj._get_callback_object(service_name="langfuse")
if langfuse_object is not None:
base_url: Final = langfuse_object.Langfuse.base_url
return f"{base_url}/trace/{trace_id}"
base_url: Final = getattr(langfuse_object, "langfuse_host", None)
if base_url is not None:
return f"{base_url}/trace/{trace_id}"
return None

463
litellm/integrations/langfuse/langfuse.py
View file

@ -1,12 +1,12 @@
#### What this does ####
# On success, logs events to Langfuse
import inspect
import os
import re
import traceback
from collections.abc import Callable, Iterable, Mapping
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import datetime
from functools import lru_cache
from importlib.metadata import version
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast
@ -45,12 +45,13 @@ from litellm.types.utils import (
)
if TYPE_CHECKING:
from langfuse.client import Langfuse, StatefulTraceClient
from langfuse import Langfuse
from opentelemetry.context import Context
from litellm.litellm_core_utils.litellm_logging import DynamicLoggingCache
else:
Context = Any
DynamicLoggingCache = Any
StatefulTraceClient = Any
Langfuse = Any
@ -158,6 +159,107 @@ def _as_steering_key_sequence(value: object) -> tuple[str, ...]:
return ()
MINIMUM_LANGFUSE_VERSION: Final = "4.7"
UNSUPPORTED_LANGFUSE_VERSION: Final = "5"
def installed_langfuse_version() -> str:
"""Only ``importlib.metadata`` reads correctly on every major.
``langfuse.version`` was removed in v4, ``langfuse.__version__`` does not
exist in v3, and in v2 it reports a different value from the distribution
that is actually installed.
"""
return version("langfuse")
def raise_if_unsupported_langfuse_version(installed_version: str) -> None:
"""Fail at logger construction rather than dropping every event at request time.
v4 moved the callback onto OpenTelemetry, so on an older SDK the import of
`propagate_attributes` raises inside the per-request handler and the broad
except there turns it into silent total data loss.
"""
installed: Final = Version(installed_version)
# compare majors, not versions: "5.0.0rc1" sorts below "5" but is just as unsupported
if Version(MINIMUM_LANGFUSE_VERSION) <= installed and installed.major < Version(UNSUPPORTED_LANGFUSE_VERSION).major:
return
raise ImportError(
f"\033[91mlitellm requires langfuse>={MINIMUM_LANGFUSE_VERSION},<{UNSUPPORTED_LANGFUSE_VERSION} for the "
f"'langfuse' callback, but {installed_version} is installed. Run "
f"'pip install \"langfuse>={MINIMUM_LANGFUSE_VERSION},<{UNSUPPORTED_LANGFUSE_VERSION}\"' to upgrade, or use "
f"the 'langfuse_otel' callback, which does not depend on the langfuse SDK\033[0m"
)
_PROPAGATED_TRACE_KEYS: Final = MappingProxyType(
{"name": "trace_name", "user_id": "user_id", "session_id": "session_id", "version": "version", "tags": "tags"}
)
_GENERATION_ONLY_KEYS: Final = frozenset(
{"id", "start_time", "end_time", "parent_observation_id", "usage", "name", "version"}
)
_PROPAGATED_VALUE_MAX_CHARS: Final = 200
def _coerce_propagated_value(value: object) -> str | Sequence[str]:
"""v4 silently drops non-string or >200-char propagated values; v2's pydantic coerced them."""
if isinstance(value, (list, tuple)):
return [str(item)[:_PROPAGATED_VALUE_MAX_CHARS] for item in value]
return str(value)[:_PROPAGATED_VALUE_MAX_CHARS]
def _trace_attributes_for_propagation(trace_params: Mapping[str, object]) -> Mapping[str, object]:
"""Trace-level fields in v4 are propagated onto the observations, not set on a trace object.
Values are coerced and capped up front: the SDK drops offenders with only a
warning, and a dropped ``version`` would vanish from the generation too,
because ``_generation_attributes`` already stripped it as propagated.
"""
return MappingProxyType(
{
propagated: _coerce_propagated_value(trace_params[key])
for key, propagated in _PROPAGATED_TRACE_KEYS.items()
if trace_params.get(key) is not None
}
)
def _optional_str(value: object) -> str | None:
"""v4 sets attribute values raw; a non-string version would be dropped by the server."""
return str(value) if value is not None else None
def _trace_public_flag(value: object) -> bool | None:
"""``trace_public`` reaches here as a bool from metadata or a string from a ``langfuse_*`` header."""
if value is None:
return None
return _as_steering_flag(value)
def _generation_attributes(
generation_params: Mapping[str, object], *, propagated: Mapping[str, object]
) -> Mapping[str, object]:
"""Drop what the v4 wrapper cannot take: ids it generates, and timings set on the span itself.
``usage`` is the v2 shape that v4 replaced with ``usage_details``, which the
caller already builds alongside it.
v4 has one ``version`` for a trace and its observations, so the trace's
propagated value covers the generation; a continued trace propagates none
and the generation keeps its own, as it did in v2.
"""
keep_version: Final = "version" not in propagated and generation_params.get("version") is not None
return MappingProxyType(
{
key: value
for key, value in generation_params.items()
if key not in _GENERATION_ONLY_KEYS or (key == "version" and keep_version)
}
)
def resolve_langfuse_credentials(
langfuse_public_key=None,
langfuse_secret=None,
@ -172,11 +274,18 @@ def resolve_langfuse_credentials(
secret_key = langfuse_secret or langfuse_secret_key or os.getenv("LANGFUSE_SECRET_KEY")
public_key = langfuse_public_key or os.getenv("LANGFUSE_PUBLIC_KEY")
resolved_host: Final = langfuse_host or os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
resolved_host: Final = (
langfuse_host or os.getenv("LANGFUSE_HOST") or os.getenv("LANGFUSE_BASE_URL") or "https://cloud.langfuse.com"
)
return public_key, secret_key, resolved_host
def parse_langfuse_debug(raw_value: str | None) -> bool:
"""Parse the LANGFUSE_DEBUG value into the boolean flag the langfuse client expects."""
return raw_value is not None and raw_value.strip().lower() in ("true", "1")
@lru_cache(maxsize=8)
def _warn_invalid_deployment_environment(raw_value: str, error: str) -> None:
verbose_logger.warning(
@ -199,12 +308,14 @@ class LangFuseLogger:
allow_env_credentials: bool = True,
):
try:
import langfuse
from langfuse import Langfuse
except Exception as e:
raise Exception(
f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\n{traceback.format_exc()}\033[0m"
)
self.langfuse_sdk_version: str = installed_langfuse_version()
raise_if_unsupported_langfuse_version(self.langfuse_sdk_version)
self.public_key, self.secret_key, self.langfuse_host = resolve_langfuse_credentials(
langfuse_public_key=langfuse_public_key,
langfuse_secret=langfuse_secret,
@ -221,7 +332,7 @@ class LangFuseLogger:
else:
self.langfuse_environment = self.resolve_deployment_environment()
self.langfuse_release = os.getenv("LANGFUSE_RELEASE")
self.langfuse_debug = os.getenv("LANGFUSE_DEBUG")
self.langfuse_debug = parse_langfuse_debug(os.getenv("LANGFUSE_DEBUG"))
self.langfuse_flush_interval = LangFuseLogger._get_langfuse_flush_interval(flush_interval)
if should_use_langfuse_mock():
@ -235,18 +346,13 @@ class LangFuseLogger:
parameters: Final = {
"public_key": self.public_key,
"secret_key": self.secret_key,
"host": self.langfuse_host,
"base_url": self.langfuse_host,
"release": self.langfuse_release,
"debug": self.langfuse_debug,
"flush_interval": self.langfuse_flush_interval, # flush interval in seconds
"httpx_client": self.langfuse_client,
"environment": self.langfuse_environment,
}
self.langfuse_sdk_version: str = langfuse.version.__version__
if "environment" in inspect.signature(Langfuse.__init__).parameters:
parameters["environment"] = self.langfuse_environment
if Version(self.langfuse_sdk_version) >= Version("2.6.0"):
parameters["sdk_integration"] = "litellm"
self.Langfuse: Langfuse = self.safe_init_langfuse_client(parameters)
# set the current langfuse project id in the environ
@ -256,30 +362,21 @@ class LangFuseLogger:
verbose_logger.debug("Langfuse Mock: Using mock project ID")
else:
try:
project_id = self.Langfuse.client.projects.get().data[0].id
project_id: Final = self.Langfuse.api.projects.get().data[0].id
os.environ["LANGFUSE_PROJECT_ID"] = project_id
except Exception:
project_id = None
verbose_logger.debug("Langfuse project id unavailable, alerting links will omit it")
if os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY") is not None:
upstream_langfuse_debug_env: Final = os.getenv("UPSTREAM_LANGFUSE_DEBUG")
upstream_langfuse_debug: Final = (
str_to_bool(upstream_langfuse_debug_env) if upstream_langfuse_debug_env is not None else None
verbose_logger.warning(
"UPSTREAM_LANGFUSE_* is no longer supported: the langfuse callback moved to SDK v4, "
"which has no second ingestion client. The values are ignored."
)
self.upstream_langfuse_secret_key = os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY")
self.upstream_langfuse_public_key = os.getenv("UPSTREAM_LANGFUSE_PUBLIC_KEY")
self.upstream_langfuse_host = os.getenv("UPSTREAM_LANGFUSE_HOST")
self.upstream_langfuse_release = os.getenv("UPSTREAM_LANGFUSE_RELEASE")
self.upstream_langfuse_debug = upstream_langfuse_debug_env
self.upstream_langfuse = Langfuse(
public_key=self.upstream_langfuse_public_key,
secret_key=self.upstream_langfuse_secret_key,
host=self.upstream_langfuse_host,
release=self.upstream_langfuse_release,
debug=(upstream_langfuse_debug if upstream_langfuse_debug is not None else False),
)
else:
self.upstream_langfuse = None
self.upstream_langfuse_debug = os.getenv("UPSTREAM_LANGFUSE_DEBUG")
def safe_init_langfuse_client(self, parameters: dict) -> Langfuse:
"""
@ -289,13 +386,20 @@ class LangFuseLogger:
- Langfuse initializes 1 thread everytime a client is initialized.
- We've had an incident in the past where we reached 100% cpu utilization because Langfuse was initialized several times.
"""
from langfuse import Langfuse
if litellm.initialized_langfuse_clients >= MAX_LANGFUSE_INITIALIZED_CLIENTS:
raise Exception(
f"Max langfuse clients reached: {litellm.initialized_langfuse_clients} is greater than {MAX_LANGFUSE_INITIALIZED_CLIENTS}"
)
langfuse_client: Final = Langfuse(**parameters)
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client
environment_param: Final = cast(str | None, parameters.get("environment")) # cast-ok: untyped dict
release_param: Final = cast(str | None, parameters.get("release")) # cast-ok: untyped dict
langfuse_client: Final = acquire_langfuse_client(
parameters=parameters,
environment=environment_param,
release=release_param,
mock_mode=self.is_mock_mode,
)
litellm.initialized_langfuse_clients += 1
verbose_logger.debug("Created langfuse client number %s", litellm.initialized_langfuse_clients)
return langfuse_client
@ -394,9 +498,9 @@ class LangFuseLogger:
status_message=status_message,
)
verbose_logger.debug("OUTPUT IN LANGFUSE: %s; original: %s", output, response_obj)
trace_id = None
generation_id = None
if self._is_langfuse_v2():
from litellm.integrations.langfuse.langfuse_sdk import lease_langfuse_client
with lease_langfuse_client(self.Langfuse):
trace_id, generation_id = self._log_langfuse_v2(
user_id=user_id,
metadata=metadata,
@ -411,18 +515,6 @@ class LangFuseLogger:
level=level,
litellm_call_id=litellm_call_id,
)
elif response_obj is not None:
self._log_langfuse_v1(
user_id=user_id,
metadata=metadata,
output=output,
start_time=start_time,
end_time=end_time,
kwargs=kwargs,
optional_params=optional_params,
input=input,
response_obj=response_obj,
)
verbose_logger.debug("Langfuse Layer Logging - final response object: %s", response_obj)
verbose_logger.info("Langfuse Layer Logging - logging success")
@ -518,58 +610,6 @@ class LangFuseLogger:
This approach does not impact latency and runs in the background
"""
def _is_langfuse_v2(self):
import langfuse
return Version(langfuse.version.__version__) >= Version("2.0.0")
def _log_langfuse_v1(
self,
user_id,
metadata,
output,
start_time,
end_time,
kwargs,
optional_params,
input,
response_obj,
):
from langfuse.model import CreateGeneration, CreateTrace
verbose_logger.warning(
"Please upgrade langfuse to v2.0.0 or higher: https://github.com/langfuse/langfuse-python/releases/tag/v2.0.1"
)
trace: Final = self.Langfuse.trace(
CreateTrace(
name=metadata.get("generation_name", "litellm-completion"),
input=input,
output=output,
userId=user_id,
)
)
custom_llm_provider: Final = cast(str | None, kwargs.get("custom_llm_provider"))
model_name: Final = reconstruct_model_name(kwargs.get("model", ""), custom_llm_provider, metadata)
trace.generation(
CreateGeneration(
name=metadata.get("generation_name", "litellm-completion"),
startTime=start_time,
endTime=end_time,
model=model_name,
modelParameters=optional_params,
prompt=input,
completion=output,
usage={
"prompt_tokens": response_obj.usage.prompt_tokens,
"completion_tokens": response_obj.usage.completion_tokens,
},
metadata=metadata,
)
)
def _log_langfuse_v2(
self,
user_id: str | None,
@ -592,11 +632,7 @@ class LangFuseLogger:
StandardLoggingPayload | None,
kwargs.get("standard_logging_object", None),
)
tags = (
self._get_langfuse_tags(standard_logging_object=standard_logging_object)
if self._supports_tags()
else []
)
tags = self._get_langfuse_tags(standard_logging_object=standard_logging_object)
allowlisted_metadata: Final[StandardLoggingMetadata | Mapping[str, object]] = (
standard_logging_object["metadata"] if standard_logging_object is not None else _NO_METADATA
@ -648,17 +684,17 @@ class LangFuseLogger:
# This allows continuing an existing trace while still returning the correct trace_id
if existing_trace_id is not None:
trace_id = existing_trace_id
resolved_trace_id: Final = (
call_trace_id: Final = (
litellm_call_id or trace_id
if existing_trace_id is None
and _is_session_header_trace(trace_id, session_id, litellm_params.get("proxy_server_request"))
else trace_id
)
if resolved_trace_id != trace_id:
if call_trace_id != trace_id:
verbose_logger.debug(
"Langfuse: trace_id %s came from a session header; using call id %s so each call gets its own trace",
trace_id,
resolved_trace_id,
call_trace_id,
)
requested_trace_keys: Final = _as_steering_key_sequence(clean_metadata.pop("update_trace_keys", ()))
update_trace_keys: Final = (
@ -714,7 +750,7 @@ class LangFuseLogger:
trace_params["output"] = masked_output if not mask_output else "redacted-by-litellm"
else: # don't overwrite an existing trace
trace_params = {
"id": resolved_trace_id,
"id": call_trace_id,
"name": trace_name,
"session_id": session_id,
"input": masked_input if not mask_input else "redacted-by-litellm",
@ -764,17 +800,16 @@ class LangFuseLogger:
("api_base", api_base, bool(api_base)),
("vertex_location", vertex_location, bool(vertex_location)),
("aws_region_name", aws_region_name, bool(aws_region_name)),
("cache_hit", kwargs.get("cache_hit") or False, self._supports_tags() and "cache_hit" in kwargs),
("cache_hit", kwargs.get("cache_hit") or False, "cache_hit" in kwargs),
)
enrichments: Final[Mapping[str, object]] = {
key: value for key, value, include in candidate_enrichments if include
}
if self._supports_tags():
if "cache_hit" in kwargs and kwargs["cache_hit"] is None:
kwargs["cache_hit"] = False # rebind-ok: pre-existing normalization other integrations rely on
if existing_trace_id is None:
trace_params.update({"tags": tags})
if "cache_hit" in kwargs and kwargs["cache_hit"] is None:
kwargs["cache_hit"] = False # rebind-ok: pre-existing normalization other integrations rely on
if existing_trace_id is None:
trace_params.update({"tags": tags})
proxy_server_request: Final = litellm_params.get("proxy_server_request", None)
if proxy_server_request:
@ -788,17 +823,6 @@ class LangFuseLogger:
if key.lower() not in _REDACTED_PROXY_HEADERS:
clean_headers[key] = value
trace: Final[StatefulTraceClient] = self.Langfuse.trace(**trace_params)
# Log provider specific information as a span
log_provider_specific_information_as_span(trace, enrichments)
# Log guardrail information as a span
self._log_guardrail_information_as_span(
trace=trace,
standard_logging_object=standard_logging_object,
)
generation_id = None
usage = None
usage_details = None
@ -820,7 +844,7 @@ class LangFuseLogger:
usage = {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_cost": cost if self._supports_costs() else None,
"total_cost": cost,
}
# According to langfuse documentation: "the input value must be reduced by the number of cache_read_input_tokens"
input_tokens: Final = prompt_tokens - cache_read_input_tokens
@ -864,45 +888,87 @@ class LangFuseLogger:
"output": masked_output if not mask_output else "redacted-by-litellm",
"usage": usage,
"usage_details": usage_details,
"metadata": {
**log_requester_metadata(redact_user_api_key_info(metadata=allowlisted_metadata)),
"cost_details": {"total": cost} # mutable-ok: langfuse serializes this payload
if usage is not None and isinstance(cost, (int, float))
else None,
"metadata": { # mutable-ok: langfuse serializes this payload, a proxy is not json-encodable
**(trace_params.get("metadata") or {}),
**log_requester_metadata(redact_user_api_key_info(metadata=allowlisted_metadata)), # pyright: ignore[reportArgumentType] # TypedDict in, plain metadata dict out
**enrichments,
},
"level": level,
"version": clean_metadata.pop("version", None),
"version": _optional_str(clean_metadata.pop("version", None)),
}
parent_observation_id: Final = metadata.get("parent_observation_id", None)
if parent_observation_id is not None:
generation_params["parent_observation_id"] = parent_observation_id
if self._supports_prompt():
generation_params = _add_prompt_to_generation_params(
generation_params=generation_params,
clean_metadata=clean_metadata,
prompt_management_metadata=prompt_management_metadata,
langfuse_client=self.Langfuse,
)
generation_params = _add_prompt_to_generation_params(
generation_params=generation_params,
clean_metadata=clean_metadata,
prompt_management_metadata=prompt_management_metadata,
langfuse_client=self.Langfuse,
)
if masked_output is not None and isinstance(masked_output, str) and level == "ERROR":
generation_params["status_message"] = masked_output
if self._supports_completion_start_time():
generation_params["completion_start_time"] = kwargs.get("completion_start_time", None)
generation_params["completion_start_time"] = kwargs.get("completion_start_time", None)
generation_client: Final = trace.generation(**generation_params)
# langfuse ships in the proxy-runtime extra, so this module must import cleanly without it
from litellm.integrations.langfuse.langfuse_sdk import (
open_trace_context,
propagate_attributes,
resolve_observation_id,
resolve_trace_id,
start_generation,
to_unix_nanos,
)
# Return the trace_id we set (which should be litellm_call_id when no explicit trace_id provided)
# We explicitly set trace_id in trace_params["id"], so langfuse should use it
# Verify langfuse accepted our trace_id; if it differs, log a warning but still return our intended value
# to match expected test behavior
if hasattr(generation_client, "trace_id") and generation_client.trace_id:
if generation_client.trace_id != resolved_trace_id:
verbose_logger.warning(
"Langfuse trace_id mismatch: set %s, but langfuse returned %s. Using our intended trace_id for consistency.",
resolved_trace_id,
generation_client.trace_id,
resolved_trace_id: Final = resolve_trace_id(call_trace_id) # pyright: ignore[reportArgumentType] # metadata value, str or None at runtime
propagated_trace_attributes: Final = _trace_attributes_for_propagation(trace_params)
with propagate_attributes(**propagated_trace_attributes): # pyright: ignore[reportArgumentType] # kwargs-ok: keys fixed by _PROPAGATED_TRACE_KEYS, values are the SDK's own trace fields
trace_context, claim_trace_root = open_trace_context(
client=self.Langfuse,
trace_id=resolved_trace_id,
parent_observation_id=resolve_observation_id(parent_observation_id), # pyright: ignore[reportArgumentType] # metadata value, str or None at runtime
)
log_provider_specific_information_as_span(
client=self.Langfuse,
context=trace_context,
enrichments=enrichments,
claim_trace_root=claim_trace_root,
)
self._log_guardrail_information_as_span(
client=self.Langfuse,
context=trace_context,
standard_logging_object=standard_logging_object,
claim_trace_root=claim_trace_root,
)
generation: Final = start_generation(
client=self.Langfuse,
context=trace_context,
name=generation_params["name"], # pyright: ignore[reportArgumentType] # always the str set a few lines up
start_time=start_time,
claim_trace_root=claim_trace_root,
release=trace_params.get("release"),
public=_trace_public_flag(trace_params.get("public")),
attributes=_generation_attributes(generation_params, propagated=propagated_trace_attributes),
)
if existing_trace_id is not None and ("input" in update_trace_keys or "output" in update_trace_keys):
# with a real parent the generation is not the trace root, so trace-level
# I/O has to be stamped explicitly; v2 updated the trace object directly
generation.set_trace_io( # pyright: ignore[reportDeprecated] # the SDK keeps it exactly for this legacy trace-level contract
input=trace_params.get("input") if "input" in update_trace_keys else None,
output=trace_params.get("output") if "output" in update_trace_keys else None,
)
return resolved_trace_id, generation_id
generation.end(end_time=to_unix_nanos(end_time))
# log_event_on_langfuse tuple-unpacks this and re-wraps it in the dict callers cache.
# The wrapper's id is the exported observation id; the pre-computed generation_id would
# name nothing in langfuse, because v4 derives observation ids from the OTel span.
return resolved_trace_id, generation.id
except Exception:
verbose_logger.error("Langfuse Layer Error - %s", traceback.format_exc())
return None, None
@ -971,27 +1037,11 @@ class LangFuseLogger:
_cache_key = _hidden_params.get("cache_key", None)
if _cache_key is None and litellm.cache is not None:
# fallback to using "preset_cache_key"
_preset_cache_key: Final = litellm.cache._get_preset_cache_key_from_kwargs(**kwargs)
_preset_cache_key: Final = litellm.cache._get_preset_cache_key_from_kwargs(**kwargs) # pyright: ignore[reportPrivateUsage] # kwargs-ok: no public preset-cache-key accessor
_cache_key = _preset_cache_key
tags.append(f"cache_key:{_cache_key}")
return tags
def _supports_tags(self):
"""Check if current langfuse version supports tags"""
return Version(self.langfuse_sdk_version) >= Version("2.6.3")
def _supports_prompt(self):
"""Check if current langfuse version supports prompt"""
return Version(self.langfuse_sdk_version) >= Version("2.7.3")
def _supports_costs(self):
"""Check if current langfuse version supports costs"""
return Version(self.langfuse_sdk_version) >= Version("2.7.3")
def _supports_completion_start_time(self):
"""Check if current langfuse version supports completion start time"""
return Version(self.langfuse_sdk_version) >= Version("2.7.3")
@staticmethod
def _apply_masking_function(data: object, masking_function: Callable[[object], object]) -> object:
"""
@ -1056,8 +1106,10 @@ class LangFuseLogger:
def _log_guardrail_information_as_span(
self,
trace: StatefulTraceClient,
client: "Langfuse",
context: "Context",
standard_logging_object: StandardLoggingPayload | None,
claim_trace_root: bool,
):
"""
Log guardrail information as a span
@ -1078,6 +1130,8 @@ class LangFuseLogger:
)
return
from litellm.integrations.langfuse.langfuse_sdk import start_child_span, to_unix_nanos
for guardrail_entry in guardrail_information:
if not isinstance(guardrail_entry, dict):
verbose_logger.debug(
@ -1086,21 +1140,25 @@ class LangFuseLogger:
)
continue
span = trace.span(
span = start_child_span(
client=client,
context=context,
name="guardrail",
input=guardrail_entry.get("guardrail_request", None),
output=guardrail_entry.get("guardrail_response", None),
metadata={
"guardrail_name": guardrail_entry.get("guardrail_name", None),
"guardrail_mode": guardrail_entry.get("guardrail_mode", None),
"guardrail_masked_entity_count": guardrail_entry.get("masked_entity_count", None),
},
start_time=guardrail_entry.get("start_time", None),
end_time=guardrail_entry.get("end_time", None),
claim_trace_root=claim_trace_root,
attributes={ # mutable-ok: langfuse serializes this payload, a proxy is not json-encodable
"input": guardrail_entry.get("guardrail_request", None),
"output": guardrail_entry.get("guardrail_response", None),
"metadata": {
"guardrail_name": guardrail_entry.get("guardrail_name", None),
"guardrail_mode": guardrail_entry.get("guardrail_mode", None),
"guardrail_masked_entity_count": guardrail_entry.get("masked_entity_count", None),
},
},
)
verbose_logger.debug("Logged guardrail information as span: %s", span)
span.end()
span.end(end_time=to_unix_nanos(guardrail_entry.get("end_time", None)))
def _add_prompt_to_generation_params(
@ -1142,7 +1200,7 @@ def _add_prompt_to_generation_params(
if "labels" in prompt_text_params and "tags" in prompt_text_params:
_data["labels"] = user_prompt.get("labels", []) or []
_data["tags"] = user_prompt.get("tags", []) or []
_prompt_obj = Prompt_Text(**_data)
_prompt_obj = Prompt_Text(**_data) # pyright: ignore[reportArgumentType] # kwargs-ok: shape mirrors the pydantic model, values from the user's prompt dict
generation_params["prompt"] = TextPromptClient(prompt=_prompt_obj)
elif isinstance(user_prompt["prompt"], list):
@ -1157,7 +1215,7 @@ def _add_prompt_to_generation_params(
_data["labels"] = user_prompt.get("labels", []) or []
_data["tags"] = user_prompt.get("tags", []) or []
_prompt_obj = Prompt_Chat(**_data)
_prompt_obj = Prompt_Chat(**_data) # pyright: ignore[reportArgumentType] # kwargs-ok: shape mirrors the pydantic model, values from the user's prompt dict
generation_params["prompt"] = ChatPromptClient(prompt=_prompt_obj)
else:
@ -1177,21 +1235,24 @@ def _add_prompt_to_generation_params(
def log_provider_specific_information_as_span(
trace,
clean_metadata: Mapping[str, Any],
*,
client: "Langfuse",
context: "Context",
enrichments: Mapping[str, Any],
claim_trace_root: bool,
):
"""
Logs provider-specific information as spans.
Parameters:
trace: The tracing object used to log spans.
clean_metadata: A dictionary containing metadata to be logged.
enrichments: The litellm-computed fields on the emitted payload.
Returns:
None
"""
_hidden_params: Final[Mapping[str, object] | None] = clean_metadata.get("hidden_params", None)
_hidden_params: Final[Mapping[str, object] | None] = enrichments.get("hidden_params", None)
if _hidden_params is None:
return
@ -1202,22 +1263,42 @@ def log_provider_specific_information_as_span(
for elem in vertex_ai_grounding_metadata:
if isinstance(elem, dict):
for key, value in elem.items():
trace.span(
name=key,
input=value,
_end_grounding_span(
client=client, context=context, name=key, value=value, claim_trace_root=claim_trace_root
)
else:
trace.span(
_end_grounding_span(
client=client,
context=context,
name="vertex_ai_grounding_metadata",
input=elem,
value=elem,
claim_trace_root=claim_trace_root,
)
else:
trace.span(
_end_grounding_span(
client=client,
context=context,
name="vertex_ai_grounding_metadata",
input=vertex_ai_grounding_metadata,
value=vertex_ai_grounding_metadata,
claim_trace_root=claim_trace_root,
)
def _end_grounding_span(
*, client: "Langfuse", context: "Context", name: str, value: object, claim_trace_root: bool
) -> None:
from litellm.integrations.langfuse.langfuse_sdk import start_child_span
start_child_span(
client=client,
context=context,
name=name,
start_time=None,
claim_trace_root=claim_trace_root,
attributes={"input": value}, # mutable-ok: langfuse serializes this payload
).end()
def log_requester_metadata(clean_metadata: Mapping[str, Any]):
returned_metadata: Final = {}
requester_metadata: Final = clean_metadata.get("requester_metadata") or {}

57
litellm/integrations/langfuse/langfuse_prompt_management.py
View file

@ -2,13 +2,10 @@
Call Hook for LiteLLM Proxy which allows Langfuse prompt management.
"""
import inspect
import os
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast
from packaging.version import Version
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.prompt_management_base import PromptManagementClient
from litellm.litellm_core_utils.asyncify import run_async_function
@ -20,12 +17,19 @@ from ...litellm_core_utils.specialty_caches.dynamic_logging_cache import (
DynamicLoggingCache,
)
from ..prompt_management_base import PromptManagementBase
from .langfuse import LangFuseLogger, resolve_langfuse_credentials
from .langfuse import (
LangFuseLogger,
installed_langfuse_version,
parse_langfuse_debug,
raise_if_unsupported_langfuse_version,
resolve_langfuse_credentials,
)
from .langfuse_handler import LangFuseHandler
from .langfuse_mock_client import create_mock_langfuse_client, should_use_langfuse_mock
if TYPE_CHECKING:
from langfuse import Langfuse
from langfuse.client import ChatPromptClient, TextPromptClient
from langfuse.model import ChatPromptClient, TextPromptClient
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -64,8 +68,9 @@ def langfuse_client_init(
Exception: If langfuse package is not installed
"""
try:
import langfuse
from langfuse import Langfuse
from langfuse import (
Langfuse, # noqa: F401 # the import is the install probe; construction moved to acquire_langfuse_client
)
except Exception as e:
raise Exception(
f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\n\033[0m"
@ -84,36 +89,45 @@ def langfuse_client_init(
langfuse_host = "http://" + langfuse_host
langfuse_release: Final = os.getenv("LANGFUSE_RELEASE")
langfuse_debug: Final = os.getenv("LANGFUSE_DEBUG")
langfuse_debug: Final = parse_langfuse_debug(os.getenv("LANGFUSE_DEBUG"))
parameters: Final = {
"public_key": public_key,
"secret_key": secret_key,
"host": langfuse_host,
"base_url": langfuse_host,
"release": langfuse_release,
"debug": langfuse_debug,
"flush_interval": LangFuseLogger._get_langfuse_flush_interval(flush_interval), # flush interval in seconds
"flush_interval": LangFuseLogger._get_langfuse_flush_interval(flush_interval), # pyright: ignore[reportPrivateUsage] # shared env-fallback helper, not part of the logger's API
}
if Version(langfuse.version.__version__) >= Version("2.6.0"):
parameters["sdk_integration"] = "litellm"
raise_if_unsupported_langfuse_version(installed_langfuse_version())
if Version(langfuse.version.__version__) >= Version("2.7.3"):
import httpx
import httpx
import litellm
import litellm
from ...llms.custom_httpx.http_handler import get_ssl_configuration
from ...llms.custom_httpx.http_handler import get_ssl_configuration
parameters["httpx_client"] = httpx.Client(
is_mock_mode: Final = should_use_langfuse_mock()
parameters["httpx_client"] = (
create_mock_langfuse_client()
if is_mock_mode
else httpx.Client(
verify=get_ssl_configuration(),
cert=os.getenv("SSL_CERTIFICATE", litellm.ssl_certificate),
)
)
if "environment" in inspect.signature(Langfuse.__init__).parameters:
parameters["environment"] = LangFuseLogger.resolve_deployment_environment()
parameters["environment"] = LangFuseLogger.resolve_deployment_environment()
client: Final = Langfuse(**parameters)
from .langfuse_sdk import acquire_langfuse_client
client: Final = acquire_langfuse_client(
parameters=parameters,
environment=parameters["environment"],
release=langfuse_release,
mock_mode=is_mock_mode,
)
return client
@ -126,9 +140,8 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
langfuse_host=None,
flush_interval=1,
):
import langfuse
self.langfuse_sdk_version = langfuse.version.__version__
self.langfuse_sdk_version = installed_langfuse_version()
self.Langfuse = langfuse_client_init(
langfuse_public_key=langfuse_public_key,
langfuse_secret=langfuse_secret,

572
litellm/integrations/langfuse/langfuse_sdk.py Normal file
View file

@ -0,0 +1,572 @@
from __future__ import annotations
import os
import re
import threading
from base64 import b64encode
from collections.abc import Generator, Mapping
from contextlib import contextmanager
from datetime import datetime
from hashlib import sha256
from types import MappingProxyType
from typing import Final
from weakref import WeakKeyDictionary, WeakSet
import opentelemetry.trace as otel_trace
from langfuse import Langfuse, LangfuseGeneration, LangfuseSpan, propagate_attributes
from langfuse._client.resource_manager import LangfuseResourceManager
from opentelemetry.context import Context
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
__all__ = (
"AS_ROOT_ATTRIBUTE",
"PUBLIC_ATTRIBUTE",
"RELEASE_ATTRIBUTE",
"DiscardingSpanExporter",
"acquire_langfuse_client",
"build_isolated_tracer_provider",
"evict_stale_langfuse_resources",
"lease_langfuse_client",
"open_trace_context",
"propagate_attributes",
"register_langfuse_client",
"resolve_observation_id",
"resolve_trace_id",
"shutdown_langfuse_client",
"start_child_span",
"start_generation",
"to_unix_nanos",
)
AS_ROOT_ATTRIBUTE: Final = "langfuse.internal.as_root"
PUBLIC_ATTRIBUTE: Final = "langfuse.trace.public"
RELEASE_ATTRIBUTE: Final = "langfuse.release"
_TRACE_ID_PATTERN: Final = re.compile(r"^[0-9a-f]{32}$")
_OBSERVATION_ID_PATTERN: Final = re.compile(r"^[0-9a-f]{16}$")
def to_unix_nanos(value: datetime | float | None) -> int | None:
"""Langfuse v4 takes OTel timestamps, which are integer nanoseconds since the epoch.
Guardrail entries carry unix seconds as floats rather than datetimes, so both
shapes have to convert; the v2 SDK accepted either through a pydantic model.
"""
if value is None:
return None
seconds: Final = value.timestamp() if isinstance(value, datetime) else float(value)
return int(seconds * 1_000_000_000)
def resolve_trace_id(trace_id: object | None) -> str:
"""Map a caller's trace id onto the 32 lowercase hex characters v4 requires."""
serialized: Final = "" if trace_id is None else str(trace_id)
normalized: Final = serialized.lower().replace("-", "")
if _TRACE_ID_PATTERN.fullmatch(normalized):
return normalized
return Langfuse.create_trace_id(seed=serialized) if serialized else Langfuse.create_trace_id()
def resolve_observation_id(observation_id: object | None) -> str | None:
"""Map a caller's parent observation id onto v4's 16 lowercase hex characters."""
serialized: Final = "" if observation_id is None else str(observation_id)
normalized: Final = serialized.lower().replace("-", "")
if _OBSERVATION_ID_PATTERN.fullmatch(normalized):
return normalized
if not serialized:
return None
return sha256(serialized.encode("utf-8")).digest()[:8].hex()
def open_trace_context(
*,
client: Langfuse,
trace_id: str,
parent_observation_id: str | None,
) -> tuple[Context, bool]:
"""Build the OTel context that places new observations inside ``trace_id``.
Returns the context plus whether the caller must claim trace root. Langfuse
fabricates a random parent span id when no real parent is supplied, so the
observation is a child of something that will never be exported; the public
SDK path compensates by marking the span as root and this path must do the
same.
"""
remote_parent: Final = client._create_remote_parent_span( # pyright: ignore[reportPrivateUsage] # no public equivalent in v4
trace_id=trace_id, parent_span_id=parent_observation_id
)
return otel_trace.set_span_in_context(remote_parent), parent_observation_id is None
def start_generation(
*,
client: Langfuse,
context: Context,
name: str,
start_time: datetime | float | None,
claim_trace_root: bool,
release: str | None = None,
public: bool | None = None,
attributes: Mapping[str, object],
) -> LangfuseGeneration:
"""Create a generation whose start time is when the model call began.
No public v4 API accepts a historical start time, so this drives the SDK's
own OTel tracer, which does. Langfuse documents this route for backdated
ingestion.
``public`` is the v2 ``trace(public=...)`` flag; v4 reads it off the root
observation's ``langfuse.trace.public`` attribute instead.
"""
otel_span: Final = client._otel_tracer.start_span( # pyright: ignore[reportPrivateUsage] # only route to a historical start time
name=name, context=context, start_time=to_unix_nanos(start_time)
)
if claim_trace_root:
otel_span.set_attribute(AS_ROOT_ATTRIBUTE, True)
if public is not None:
otel_span.set_attribute(PUBLIC_ATTRIBUTE, public)
generation: Final = LangfuseGeneration(otel_span=otel_span, langfuse_client=client, **attributes) # pyright: ignore[reportArgumentType] # kwargs-ok: callback-built params, v2 accepted the same shapes
if release is not None:
# after the wrapper, which stamps the client-wide release and would otherwise
# overwrite the release this request asked for
otel_span.set_attribute(RELEASE_ATTRIBUTE, release)
return generation
def start_child_span(
*,
client: Langfuse,
context: Context,
name: str,
start_time: datetime | float | None,
claim_trace_root: bool,
attributes: Mapping[str, object],
) -> LangfuseSpan:
"""Create a sibling observation inside the same trace, keeping its own window.
When the shared parent is the fabricated remote span, every observation must
claim trace root itself — the SDK's own remote-parent paths stamp each span —
or it exports with a parent id that is never exported.
"""
otel_span: Final = client._otel_tracer.start_span( # pyright: ignore[reportPrivateUsage] # only route to a historical start time
name=name, context=context, start_time=to_unix_nanos(start_time)
)
if claim_trace_root:
otel_span.set_attribute(AS_ROOT_ATTRIBUTE, True)
return LangfuseSpan(otel_span=otel_span, langfuse_client=client, **attributes) # pyright: ignore[reportArgumentType] # kwargs-ok: callback-built params, v2 accepted the same shapes
_ENVIRONMENT_ATTRIBUTE: Final = "langfuse.environment"
# providers litellm itself constructed; a bundle adopted from user code may hold the
# process-global provider, which litellm must never shut down.
_litellm_built_providers: Final[WeakSet] = WeakSet()
def build_isolated_tracer_provider(*, environment: str | None, release: str | None) -> TracerProvider:
"""Give the langfuse client a provider of its own instead of the process-wide one.
v4 is built on OpenTelemetry and otherwise either claims the global tracer
provider, which silently disables litellm's own exporters, or attaches its
processor to litellm's, which sends litellm spans to every langfuse project
and langfuse spans to every other litellm destination.
The resource is rebuilt here because langfuse only applies ``environment``
and ``release`` when it constructs the provider itself, and the sampler is
rebuilt for the same reason: ``LANGFUSE_SAMPLE_RATE`` is otherwise silently
ignored and every trace exports.
"""
raw_sample_rate: Final = os.environ.get("LANGFUSE_SAMPLE_RATE")
sample_rate: Final = float(raw_sample_rate) if raw_sample_rate is not None else 1.0
if not 0.0 <= sample_rate <= 1.0:
raise ValueError(f"Sample rate must be between 0.0 and 1.0, got {sample_rate}")
attributes: Final = MappingProxyType(
{
key: value
for key, value in ((_ENVIRONMENT_ATTRIBUTE, environment), (RELEASE_ATTRIBUTE, release))
if value is not None
}
)
provider: Final = TracerProvider(
resource=Resource.create(dict(attributes)),
sampler=TraceIdRatioBased(sample_rate) if sample_rate < 1 else None,
)
with _LIVE_CLIENTS_LOCK:
_litellm_built_providers.add(provider)
return provider
class DiscardingSpanExporter(SpanExporter):
"""Accept and drop every span, for mock mode.
The mock intercepts the httpx client langfuse used to take, but v4 ships
observations through its own OTLP exporter, so without this the "no network
calls" contract silently sends real traces to the configured host.
"""
def export(self, spans: object) -> SpanExportResult:
return SpanExportResult.SUCCESS
def shutdown(self) -> None:
return None
def force_flush(self, timeout_millis: int = 30_000) -> bool:
return True
_LIVE_CLIENTS_LOCK: Final = threading.Lock()
# litellm clients still using each SDK resource bundle; the bundle is torn down with the last one.
# Both sides are weak so a throwaway client (a health probe, an alerting lookup) that is simply
# garbage-collected stops holding the bundle open rather than inflating a counter forever.
_live_clients: Final[WeakKeyDictionary[LangfuseResourceManager, WeakSet]] = WeakKeyDictionary()
class _LangfuseLifecycleState:
"""How many callbacks are leasing one SDK resource bundle, and what eviction has queued behind them.
``lock`` is never held across a teardown, which takes the SDK's own registry lock.
"""
def __init__(self) -> None:
self.lock = threading.Lock()
self.active_leases = 0
self.teardown_in_progress = False
self.teardown_owner: int | None = None
self.pending_clients: set[Langfuse] = set() # mutable-ok: eviction and callback threads queue into it
def open_lease(self) -> None:
with self.lock:
self.active_leases += 1
def claim_for_teardown(self, client: Langfuse) -> bool:
"""Whether this thread owns ``client``'s teardown; a lease or another teardown in flight queues it instead."""
with self.lock:
if self.active_leases > 0 or self.teardown_in_progress:
self.pending_clients.add(client)
return False
self.teardown_in_progress = True
self.teardown_owner = threading.get_ident()
return True
def release_lease(self) -> tuple[Langfuse, ...]:
"""Drop this lease and take ownership of the teardowns it was holding up, if it was the last one."""
with self.lock:
self.active_leases -= 1
if self.active_leases > 0 or self.teardown_in_progress or not self.pending_clients:
return ()
claimed: Final = tuple(self.pending_clients)
self.pending_clients.clear()
self.teardown_in_progress = True
self.teardown_owner = threading.get_ident()
return claimed
def next_teardown_batch(self) -> tuple[Langfuse, ...]:
"""Whatever eviction queued while the last batch was draining, handing the ownership flag back when empty."""
with self.lock:
if self.active_leases == 0 and self.pending_clients:
claimed: Final = tuple(self.pending_clients)
self.pending_clients.clear()
return claimed
self.teardown_in_progress = False
self.teardown_owner = None
return ()
def requeue(self, clients: tuple[Langfuse, ...]) -> None:
with self.lock:
self.pending_clients.update(clients)
def end_teardown(self) -> None:
with self.lock:
if self.teardown_owner == threading.get_ident():
self.teardown_in_progress = False
self.teardown_owner = None
_LIFECYCLE_STATES_LOCK: Final = threading.Lock()
_LIFECYCLE_STATES: Final[WeakKeyDictionary[object, _LangfuseLifecycleState]] = WeakKeyDictionary()
def _lifecycle_state(client: Langfuse) -> _LangfuseLifecycleState:
"""One state per resource bundle, since teardown closes the provider every client on that bundle exports through."""
resources: Final = getattr(client, "_resources", None)
key: Final = client if resources is None else resources
with _LIFECYCLE_STATES_LOCK:
existing: Final = _LIFECYCLE_STATES.get(key)
if existing is not None:
return existing
created: Final = _LangfuseLifecycleState()
_LIFECYCLE_STATES[key] = created
return created
@contextmanager
def lease_langfuse_client(client: Langfuse) -> Generator[None]:
"""Hold off cache eviction's teardown of ``client`` while the export inside is in flight.
Eviction reaches a client the cache handed a callback moments earlier, so closing the SDK client
and its tracer provider there drops the spans that callback is still writing. The lease protects
exactly the window it wraps: an eviction arriving inside it is deferred to the last lease exit.
Taking a lease never blocks; a teardown already running keeps running, because the spans of a
lease taken that late were lost before the lease began, and stalling every other callback in the
process would not bring them back. A client the registry hands out during the deferral registers
as a holder, and the reference count keeps its bundle alive from there.
"""
state: Final = _lifecycle_state(client)
state.open_lease()
try:
yield
finally:
_run_teardowns(state, state.release_lease(), propagate_base_exception=False)
def _run_teardowns(
state: _LangfuseLifecycleState,
clients: tuple[Langfuse, ...],
*,
propagate_base_exception: bool = True,
) -> None:
"""Tear down ``clients``, then whatever eviction queued meanwhile, and hand the flag back.
A failing ordinary teardown is logged and skipped rather than raised: the thread here is usually a
request callback that merely held the last lease, and its request must not fail on eviction's behalf.
Interrupts requeue the unfinished batch and normally propagate, while a callback exception already
in flight takes precedence over an eviction interrupt.
"""
batch = clients # rebind-ok: drains each batch queued while the previous one was being torn down
try:
from litellm._logging import verbose_logger
while batch:
for index, client in enumerate(batch):
try:
_teardown_langfuse_client(client)
except Exception:
verbose_logger.exception("Langfuse client teardown failed during cache eviction")
except BaseException:
state.requeue(batch[index:])
if propagate_base_exception:
raise
return
batch = state.next_teardown_batch()
finally:
state.end_teardown()
def _evict_if_stale_locked(
*, public_key: object, secret_key: object, base_url: object
) -> LangfuseResourceManager | None:
"""Assumes ``LangfuseResourceManager._lock`` is held; returns the still-valid bundle, evicting a stale one."""
if not public_key:
return None
cached: Final = LangfuseResourceManager._instances.get(public_key) # pyright: ignore[reportPrivateUsage] # registry has no public accessor
if cached is None:
return None
if getattr(cached, "secret_key", None) == secret_key and getattr(cached, "base_url", None) == base_url:
return cached
LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor
return None
def _retire_orphaned_providers() -> None:
"""Shut down every provider litellm built whose bundle nothing uses any more.
A rotated-out bundle whose last client is simply garbage collected, which is how the
prompt-management LRU drops clients, never reaches ``shutdown_langfuse_client``, and the
provider's own atexit hook would keep its export thread alive for the rest of the process.
Holders are snapshotted last: a client is registered in the same registry-locked block
that builds its provider, so once the registry snapshot's lock has been acquired, the
client of any provider from the first snapshot is visible to the final one even when a
concurrent rotation already evicted its bundle again. Runs outside both locks because
provider shutdown flushes and joins the export thread.
"""
with _LIVE_CLIENTS_LOCK:
candidates: Final = tuple(_litellm_built_providers)
with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
registered: Final = tuple(
getattr(resources, "tracer_provider", None)
for resources in LangfuseResourceManager._instances.values() # pyright: ignore[reportPrivateUsage] # registry has no public accessor
)
with _LIVE_CLIENTS_LOCK:
held: Final = tuple(
getattr(resources, "tracer_provider", None)
for resources, holders in _live_clients.items()
if len(holders) > 0
)
orphaned: Final = tuple(provider for provider in candidates if provider not in registered and provider not in held)
for provider in orphaned:
_litellm_built_providers.discard(provider)
provider.shutdown()
def evict_stale_langfuse_resources(*, public_key: str | None, secret_key: str | None, base_url: str | None) -> None:
"""Drop a cached client whose credentials no longer match the ones being requested."""
with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
_evict_if_stale_locked(public_key=public_key, secret_key=secret_key, base_url=base_url)
_retire_orphaned_providers()
def _build_verified_span_exporter(*, public_key: object, secret_key: object, base_url: object) -> SpanExporter | None:
"""Rebuild litellm's TLS material onto the export channel.
v2 ingested through the injected httpx client, which carried litellm's CA
bundle and client certificate; v4 ships every observation through its own
OTLP exporter, so a private-CA deployment would fail TLS on every export in
a background thread while ``auth_check`` (still on the httpx client) stays
green. Only built when custom TLS material is configured; endpoint and
headers mirror ``langfuse._client.span_processor``.
"""
import litellm
ca_bundle: Final = litellm.ssl_verify if isinstance(litellm.ssl_verify, str) else None
configured_certificate: Final = os.getenv("SSL_CERTIFICATE") or litellm.ssl_certificate
client_certificate: Final = configured_certificate if isinstance(configured_certificate, str) else None
if ca_bundle is None and client_certificate is None:
return None
import langfuse as langfuse_package
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
langfuse_version: Final = getattr(langfuse_package, "__version__", "unknown")
export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH")
endpoint: Final = f"{base_url}/{export_path}" if export_path else f"{base_url}/api/public/otel/v1/traces"
encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii")
return OTLPSpanExporter(
endpoint=endpoint,
headers={ # mutable-ok: the exporter copies these into its session headers
"Authorization": "Basic " + encoded_auth,
"x-langfuse-sdk-name": "python",
"x-langfuse-sdk-version": langfuse_version,
"x-langfuse-public-key": str(public_key),
},
certificate_file=ca_bundle,
client_certificate_file=client_certificate,
)
def acquire_langfuse_client(
*,
parameters: Mapping[str, object],
environment: str | None,
release: str | None,
mock_mode: bool,
) -> Langfuse:
"""Evict-check, construct, and register a client as one atomic step.
The SDK registry lock is held across the whole sequence: released between
eviction and construction, two concurrent inits for the same public key
with different secrets can bind one tenant's logger to the other tenant's
exporter. The isolated provider is only built when the registry does not
already hold the key — a discarded ``TracerProvider`` stays pinned forever
by its atexit hook, so building one per health probe or alerting lookup
would leak a provider each time.
"""
public_key: Final = parameters.get("public_key")
span_exporter: Final = (
DiscardingSpanExporter()
if mock_mode
else _build_verified_span_exporter(
public_key=public_key,
secret_key=parameters.get("secret_key"),
base_url=parameters.get("base_url"),
)
)
with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
cached: Final = _evict_if_stale_locked(
public_key=public_key,
secret_key=parameters.get("secret_key"),
base_url=parameters.get("base_url"),
)
client: Final = Langfuse(
**parameters, # pyright: ignore[reportArgumentType] # kwargs-ok: dict mirrors the typed ctor, values resolved by the callers
tracer_provider=None
if cached is not None
else build_isolated_tracer_provider(environment=environment, release=release),
span_exporter=span_exporter,
)
register_langfuse_client(client)
_retire_orphaned_providers()
return client
def register_langfuse_client(client: Langfuse) -> None:
"""Track the client against the SDK resources it ended up with.
langfuse keys its resources on the public key alone, so a second client for
the same key (a per-key ``langfuse_environment`` override, a team whose
callback_vars repeat the global credentials) is handed the first client's
tracer provider and export thread rather than its own. Only the last live
client may shut those down; see ``shutdown_langfuse_client``.
"""
resources: Final = getattr(client, "_resources", None)
if resources is None:
return
with _LIVE_CLIENTS_LOCK:
holders = _live_clients.get(resources)
if holders is None:
holders = WeakSet()
_live_clients[resources] = holders
holders.add(client)
def _release_langfuse_resources(resources: LangfuseResourceManager, client: Langfuse) -> bool:
"""Drop the client's claim; True when no other live client still uses ``resources``."""
with _LIVE_CLIENTS_LOCK:
holders: Final = _live_clients.get(resources)
if holders is None:
return True
holders.discard(client)
if len(holders) > 0:
return False
_live_clients.pop(resources, None)
return True
def shutdown_langfuse_client(client: Langfuse) -> None:
"""Release everything the client owns, which the SDK's own shutdown does not.
``Langfuse.shutdown`` joins the score and media consumers but leaves the
tracer provider's export thread running and leaves the client in the
registry, so a later request for the same key gets a dead client back.
A callback holding a lease on the client's bundle postpones all of this to
the moment that lease ends, so eviction cannot close the provider out from
under an export the lease is wrapping. See ``lease_langfuse_client``.
"""
state: Final = _lifecycle_state(client)
if not state.claim_for_teardown(client):
return
_run_teardowns(state, (client,))
def _teardown_langfuse_client(client: Langfuse) -> None:
"""The blocking teardown behind ``shutdown_langfuse_client``.
A client that shares its resources with another live client only flushes:
shutting the shared provider down here would silence the other client for
the rest of its life, as it did before the reference count existed.
The registry entry is removed before the blocking shutdown so a concurrent
construct builds a fresh bundle instead of adopting a dying one, and the
provider is only shut down when litellm built it: a bundle adopted from
user code may share the process-global provider.
"""
resources: Final = getattr(client, "_resources", None)
client.flush()
if resources is None:
client.shutdown()
return
public_key: Final = getattr(resources, "public_key", None)
with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
if not _release_langfuse_resources(resources, client):
return
if public_key is not None and LangfuseResourceManager._instances.get(public_key) is resources: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor
client.shutdown()
provider: Final = getattr(resources, "tracer_provider", None)
if provider is not None and provider in _litellm_built_providers:
_litellm_built_providers.discard(provider)
provider.shutdown()
_retire_orphaned_providers()

7
litellm/litellm_core_utils/specialty_caches/dynamic_logging_cache.py
View file

@ -43,9 +43,12 @@ class LangfuseInMemoryCache(InMemoryCache):
#########################################################
# Clean up Langfuse initialized clients
#########################################################
from litellm.integrations.langfuse.langfuse_sdk import (
shutdown_langfuse_client,
)
litellm.initialized_langfuse_clients -= 1
_created_langfuse_logger.Langfuse.flush()
_created_langfuse_logger.Langfuse.shutdown()
shutdown_langfuse_client(_created_langfuse_logger.Langfuse)
# Loggers with a periodic flush task (e.g. NewRelicMetricsLogger) expose
# stop() so eviction actually ends the task instead of leaking it.

5
litellm/proxy/health_endpoints/_health_endpoints.py
View file

@ -394,7 +394,10 @@ async def health_services_endpoint(
from litellm.integrations.langfuse.langfuse import LangFuseLogger
langfuse_logger: Final = LangFuseLogger()
langfuse_logger.Langfuse.auth_check()
if langfuse_logger.Langfuse.auth_check() is False:
raise ValueError(
"langfuse auth_check failed - verify LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY are set correctly"
)
_ = litellm.completion(
model="openai/litellm-mock-response-model",
messages=[{"role": "user", "content": "Hey, how's it going?"}],

30
pyproject.toml
View file

@ -155,11 +155,11 @@ proxy-runtime = [
"anthropic[vertex]>=0.84.0,<1.0",
"grpcio==1.78.0",
"prometheus-client>=0.20.0,<1.0",
"langfuse>=2.59.7,<3.0",
"opentelemetry-api==1.28.0",
"opentelemetry-sdk==1.28.0",
"opentelemetry-exporter-otlp==1.28.0",
"opentelemetry-instrumentation-fastapi==0.49b0",
"langfuse>=4.7,<5.0",
"opentelemetry-api==1.33.1",
"opentelemetry-sdk==1.33.1",
"opentelemetry-exporter-otlp==1.33.1",
"opentelemetry-instrumentation-fastapi==0.54b1",
"ddtrace>=4.8.2,<5.0",
"sentry-sdk>=2.21.0,<3.0",
"mangum>=0.17.0,<1.0",
@ -204,11 +204,11 @@ dev = [
"types-PyYAML==6.0.12.20250915",
"botocore-stubs==1.43.14",
"types-boto3[bedrock,bedrock-agent,bedrock-runtime,kms,s3,sagemaker-runtime,sts]==1.43.30",
"opentelemetry-api==1.28.0",
"opentelemetry-sdk==1.28.0",
"opentelemetry-exporter-otlp==1.28.0",
"opentelemetry-instrumentation-fastapi==0.49b0",
"langfuse==2.59.7",
"opentelemetry-api==1.33.1",
"opentelemetry-sdk==1.33.1",
"opentelemetry-exporter-otlp==1.33.1",
"opentelemetry-instrumentation-fastapi==0.54b1",
"langfuse>=4.7,<5.0",
"fastapi-offline==1.7.6",
"fakeredis==2.34.1",
"pytest-rerunfailures==15.1",
@ -230,10 +230,10 @@ proxy-dev = [
"prisma==0.11.0",
"hypercorn==0.17.3",
"prometheus-client==0.20.0",
"opentelemetry-api==1.28.0",
"opentelemetry-sdk==1.28.0",
"opentelemetry-exporter-otlp==1.28.0",
"opentelemetry-instrumentation-fastapi==0.49b0",
"opentelemetry-api==1.33.1",
"opentelemetry-sdk==1.33.1",
"opentelemetry-exporter-otlp==1.33.1",
"opentelemetry-instrumentation-fastapi==0.54b1",
"azure-identity==1.25.2",
"a2a-sdk==1.1.0",
]
@ -253,7 +253,7 @@ ci = [
"lunary==1.4.36; python_version == '3.10'",
"lunary==1.4.37; python_version >= '3.11'",
"logfire==4.6.0",
"traceloop-sdk==0.33.12",
"traceloop-sdk==0.34.0",
"detect-secrets==1.5.0",
"PyGithub==2.8.1",
"aiodynamo==24.7",

25
tests/litellm_utils_tests/test_utils.py
View file

@ -914,6 +914,7 @@ def test_logging_trace_id(langfuse_trace_id, langfuse_existing_trace_id):
"""
- Unit test for `_get_trace_id` function in Logging obj
"""
from litellm.integrations.langfuse.langfuse_sdk import resolve_trace_id
from litellm.litellm_core_utils.litellm_logging import Logging
litellm.success_callback = ["langfuse"]
@ -946,24 +947,18 @@ def test_logging_trace_id(langfuse_trace_id, langfuse_existing_trace_id):
time.sleep(3)
assert litellm_logging_obj._get_trace_id(service_name="langfuse") is not None
## if existing_trace_id exists
# langfuse addresses a trace by a 32-hex id, so the id litellm reports back is the
# resolved form of whichever source won; that is what the alerting deep link needs
if langfuse_existing_trace_id is not None:
assert (
litellm_logging_obj._get_trace_id(service_name="langfuse")
== langfuse_existing_trace_id
)
## if trace_id exists
expected_source = langfuse_existing_trace_id
elif langfuse_trace_id is not None:
assert (
litellm_logging_obj._get_trace_id(service_name="langfuse")
== langfuse_trace_id
)
## if no trace_id or existing_trace_id is provided, use litellm_trace_id
expected_source = langfuse_trace_id
else:
assert (
litellm_logging_obj._get_trace_id(service_name="langfuse")
== litellm_logging_obj.litellm_trace_id
)
expected_source = litellm_logging_obj.litellm_trace_id
assert litellm_logging_obj._get_trace_id(service_name="langfuse") == resolve_trace_id(
expected_source
)
def test_convert_model_response_object():

13
tests/logging_callback_tests/test_langfuse_dynamic_credentials.py
View file

@ -51,7 +51,13 @@ def test_resolve_langfuse_credentials_keeps_env_for_global_config(monkeypatch):
assert host == "https://admin-configured.example"
def test_upstream_langfuse_debug_env_is_passed(monkeypatch):
def test_upstream_langfuse_env_is_read_without_building_a_client(monkeypatch):
"""The UPSTREAM_LANGFUSE_* values are recorded, and nothing consumes them.
A client was built here and never referenced. On v4 that means a second
exporter, its own threads, and an entry in the SDK's per-key registry, so it
is no longer constructed.
"""
from litellm.integrations.langfuse.langfuse import LangFuseLogger
class FakeLangfuse:
@ -81,7 +87,10 @@ def test_upstream_langfuse_debug_env_is_passed(monkeypatch):
)
assert logger.upstream_langfuse_debug == "true"
assert FakeLangfuse.instances[-1].kwargs["debug"] is True
assert logger.upstream_langfuse_public_key == "upstream-public"
assert logger.upstream_langfuse_host == "https://upstream.example"
assert logger.upstream_langfuse_release == "release"
assert not hasattr(logger, "upstream_langfuse")
def test_langfuse_handler_accepts_secret_key_alias(monkeypatch):

57
tests/logging_callback_tests/test_langfuse_unit_tests.py
View file

@ -306,35 +306,58 @@ def test_get_langfuse_flush_interval():
def test_langfuse_e2e_sync(monkeypatch):
from litellm import completion
import litellm
import respx
import httpx
"""A sync completion must reach langfuse over the wire, not just build a span.
v4 exports OTLP over ``requests`` rather than the v2 ingestion endpoint over
httpx, so this stands up a real receiver and asserts langfuse posted to it.
"""
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
litellm.disable_aiohttp_transport = (
True # since this uses respx, we need to set use_aiohttp_transport to False
)
import litellm
from litellm import completion
from litellm.integrations.langfuse.langfuse import LangFuseLogger
litellm._turn_on_debug()
received_paths = []
class _Receiver(BaseHTTPRequestHandler):
def do_POST(self):
received_paths.append(self.path)
self.rfile.read(int(self.headers.get("Content-Length") or 0))
self.send_response(200)
self.send_header("Content-Length", "0")
self.end_headers()
def log_message(self, *args):
pass
server = HTTPServer(("127.0.0.1", 0), _Receiver)
threading.Thread(target=server.serve_forever, daemon=True).start()
monkeypatch.setenv("LANGFUSE_HOST", f"http://127.0.0.1:{server.server_port}")
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-e2e-sync")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-e2e-sync")
monkeypatch.setattr(litellm, "success_callback", ["langfuse"])
monkeypatch.setattr(litellm, "_langfuse_logger_cache", {}, raising=False)
with respx.mock:
# Mock Langfuse
# Mock any Langfuse endpoint
langfuse_mock = respx.post(
"https://*.cloud.langfuse.com/api/public/ingestion"
).mock(return_value=httpx.Response(200))
try:
completion(
model="openai/my-fake-endpoint",
messages=[{"role": "user", "content": "hello from litellm"}],
stream=False,
mock_response="Hello from litellm 2",
)
for logger in litellm.logging_callback_manager._get_all_callbacks():
if isinstance(logger, LangFuseLogger):
logger.Langfuse.flush()
deadline = time.time() + 10
while not received_paths and time.time() < deadline:
time.sleep(0.1)
finally:
server.shutdown()
time.sleep(3)
assert langfuse_mock.called
assert received_paths, "langfuse exported nothing"
assert all(path.endswith("/api/public/otel/v1/traces") for path in received_paths)
def test_get_chat_content_for_langfuse():

75
tests/test_litellm/integrations/langfuse/test_langfuse_prompt_management.py
View file

@ -1,9 +1,13 @@
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Final
from unittest.mock import MagicMock, patch
import pytest
# langfuse_client_init imports this lazily; cache it before any test mocks
# sys.modules["langfuse"], or a single-file run dies on the real import
import litellm.integrations.langfuse.langfuse_sdk # noqa: F401
from litellm.integrations.langfuse.langfuse_prompt_management import (
LangfusePromptManagement,
langfuse_client_init,
@ -87,13 +91,12 @@ class TestLangfusePromptManagement:
"litellm.integrations.langfuse.langfuse_prompt_management.LangFuseLogger._get_langfuse_flush_interval",
return_value=1,
),
patch.dict("sys.modules", {"langfuse": self._mock_langfuse}),
patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", mock_langfuse_class), # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
patch(
"litellm.llms.custom_httpx.http_handler.get_ssl_configuration",
return_value=False,
) as mock_get_ssl,
):
self._mock_langfuse.Langfuse = mock_langfuse_class
langfuse_client_init(
langfuse_public_key="pk-1234",
@ -124,16 +127,76 @@ class _RecordingLangfuseForEnv:
(("Production", "default"), ("production ", "production"), ("prod", "prod")),
)
def test_langfuse_client_init_resolves_deployment_environment(monkeypatch, env_value, expected):
mock_langfuse_module: Final = MagicMock()
mock_langfuse_module.version.__version__ = "2.60.0"
mock_langfuse_module.Langfuse = _RecordingLangfuseForEnv
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-test")
monkeypatch.setenv("LANGFUSE_HOST", "https://test.langfuse.com")
monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", env_value)
monkeypatch.setattr(_RecordingLangfuseForEnv, "last_environment", None)
with patch.dict("sys.modules", MappingProxyType({"langfuse": mock_langfuse_module})):
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuseForEnv): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
langfuse_client_init.cache_clear()
langfuse_client_init()
langfuse_client_init.cache_clear()
assert _RecordingLangfuseForEnv.last_environment == expected
def test_langfuse_client_init_mock_mode_makes_no_network_calls(monkeypatch):
"""LANGFUSE_MOCK promises full execution without egress.
The registry maps the "langfuse" callback to LangfusePromptManagement, so
this client is the one the standard proxy path emits observations through;
v4 ships them over its own OTLP exporter, which the httpx mock cannot see.
"""
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from langfuse._client.resource_manager import LangfuseResourceManager
from litellm.integrations.langfuse.langfuse_sdk import (
open_trace_context,
start_generation,
to_unix_nanos,
)
received = []
class _Receiver(BaseHTTPRequestHandler):
def do_POST(self):
received.append(self.path)
self.rfile.read(int(self.headers.get("Content-Length") or 0))
self.send_response(200)
self.send_header("Content-Length", "0")
self.end_headers()
def log_message(self, *args):
pass
server = HTTPServer(("127.0.0.1", 0), _Receiver)
threading.Thread(target=server.serve_forever, daemon=True).start()
monkeypatch.setenv("LANGFUSE_MOCK", "true")
monkeypatch.setenv("LANGFUSE_HOST", f"http://127.0.0.1:{server.server_port}")
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-pm-mock-egress")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-pm-mock-egress")
LangfuseResourceManager._instances.pop("pk-pm-mock-egress", None)
langfuse_client_init.cache_clear()
try:
client = langfuse_client_init()
context, claim_root = open_trace_context(client=client, trace_id="a" * 32, parent_observation_id=None)
now = datetime.now(timezone.utc)
start_generation(
client=client,
context=context,
name="pm-mock-gen",
start_time=now,
claim_trace_root=claim_root,
attributes={},
).end(end_time=to_unix_nanos(now))
client.flush()
time.sleep(1)
finally:
server.shutdown()
langfuse_client_init.cache_clear()
LangfuseResourceManager._instances.pop("pk-pm-mock-egress", None)
assert received == [], f"LANGFUSE_MOCK still sent spans to the configured host: {received}"

893
tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py Normal file
View file

@ -0,0 +1,893 @@
"""Covers the v4 observation plumbing: historical timestamps and id normalisation.
The timestamp assertions are the regression guard for the migration: v4 has no
public API for an observation start time, so a callback running after the model
call would otherwise record its own duration instead of the call's.
"""
import json
import threading
from datetime import datetime, timedelta, timezone
import opentelemetry.trace as otel_trace
import pytest
from langfuse import Langfuse
from langfuse._client.resource_manager import LangfuseResourceManager
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from litellm.integrations.langfuse.langfuse import (
MINIMUM_LANGFUSE_VERSION,
installed_langfuse_version,
raise_if_unsupported_langfuse_version,
)
from litellm.integrations.langfuse.langfuse_sdk import (
AS_ROOT_ATTRIBUTE,
PUBLIC_ATTRIBUTE,
RELEASE_ATTRIBUTE,
_lifecycle_state,
_litellm_built_providers,
_teardown_langfuse_client,
build_isolated_tracer_provider,
evict_stale_langfuse_resources,
lease_langfuse_client,
open_trace_context,
register_langfuse_client,
resolve_observation_id,
resolve_trace_id,
shutdown_langfuse_client,
start_child_span,
start_generation,
to_unix_nanos,
)
CALL_START = datetime(2024, 3, 1, 12, 0, 0, tzinfo=timezone.utc)
FIRST_TOKEN = CALL_START + timedelta(seconds=5)
CALL_END = CALL_START + timedelta(seconds=20)
@pytest.fixture(name="client")
def _client():
LangfuseResourceManager._instances.pop("pk-obs-test", None)
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
client = Langfuse(
public_key="pk-obs-test",
secret_key="sk-obs-test",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
yield client, exporter
LangfuseResourceManager._instances.pop("pk-obs-test", None)
def _only_span(exporter, name):
return next(s for s in exporter.get_finished_spans() if s.name == name)
def test_generation_records_the_model_call_window_not_the_callback(client):
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="a" * 32, parent_observation_id=None)
start_generation(
client=lf,
context=context,
name="gen",
start_time=CALL_START,
claim_trace_root=claim_root,
attributes={"completion_start_time": FIRST_TOKEN},
).end(end_time=to_unix_nanos(CALL_END))
lf.flush()
span = _only_span(exporter, "gen")
assert span.start_time == to_unix_nanos(CALL_START)
assert span.end_time == to_unix_nanos(CALL_END)
assert (span.end_time - span.start_time) == 20 * 1_000_000_000
assert json.loads(span.attributes["langfuse.observation.completion_start_time"]) == FIRST_TOKEN.isoformat().replace(
"+00:00", "Z"
)
@pytest.mark.parametrize(
"supplied",
[1709294400.5, datetime(2024, 3, 1, 12, 0, 0, 500000, tzinfo=timezone.utc)],
ids=["unix-seconds-float", "datetime"],
)
def test_timestamps_accept_both_shapes_guardrails_and_callbacks_use(supplied):
"""Guardrail entries carry unix seconds as floats, the callback carries datetimes."""
assert to_unix_nanos(supplied) == 1709294400500000000
def test_guardrail_span_with_float_timestamps_does_not_break_the_generation(client):
"""A guardrail entry must not take the whole event down with it."""
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="9" * 32, parent_observation_id=None)
guardrail_start = 1709294400.0
start_child_span(
client=lf,
context=context,
name="guardrail",
start_time=guardrail_start,
claim_trace_root=claim_root,
attributes={},
).end(end_time=to_unix_nanos(guardrail_start + 2))
start_generation(
client=lf, context=context, name="gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={}
).end(end_time=to_unix_nanos(CALL_END))
lf.flush()
guardrail = _only_span(exporter, "guardrail")
assert (guardrail.end_time - guardrail.start_time) == 2 * 1_000_000_000
assert _only_span(exporter, "gen") is not None
def test_generation_claims_trace_root_only_without_a_real_parent(client):
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="b" * 32, parent_observation_id=None)
assert claim_root is True
start_generation(
client=lf, context=context, name="root-gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={}
).end()
parented_context, parented_claim = open_trace_context(client=lf, trace_id="b" * 32, parent_observation_id="c" * 16)
assert parented_claim is False
start_generation(
client=lf,
context=parented_context,
name="child-gen",
start_time=CALL_START,
claim_trace_root=parented_claim,
attributes={},
).end()
lf.flush()
assert _only_span(exporter, "root-gen").attributes.get(AS_ROOT_ATTRIBUTE) is True
assert _only_span(exporter, "child-gen").attributes.get(AS_ROOT_ATTRIBUTE) is None
def test_child_span_keeps_its_own_window_and_stays_a_sibling(client):
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="d" * 32, parent_observation_id=None)
guardrail_start = CALL_START + timedelta(seconds=1)
start_child_span(
client=lf,
context=context,
name="guardrail",
start_time=guardrail_start,
claim_trace_root=claim_root,
attributes={},
).end(end_time=to_unix_nanos(guardrail_start + timedelta(seconds=2)))
start_generation(
client=lf, context=context, name="gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={}
).end(end_time=to_unix_nanos(CALL_END))
lf.flush()
guardrail = _only_span(exporter, "guardrail")
generation = _only_span(exporter, "gen")
assert (guardrail.end_time - guardrail.start_time) == 2 * 1_000_000_000
assert guardrail.context.trace_id == generation.context.trace_id
# the shared remote parent is fabricated and never exported, so both must claim trace root
assert guardrail.attributes.get(AS_ROOT_ATTRIBUTE) is True
assert generation.attributes.get(AS_ROOT_ATTRIBUTE) is True
def test_release_is_carried_on_the_root_observation(client):
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="e" * 32, parent_observation_id=None)
start_generation(
client=lf,
context=context,
name="gen",
start_time=CALL_START,
claim_trace_root=claim_root,
release="v1.2.3",
attributes={},
).end()
lf.flush()
assert _only_span(exporter, "gen").attributes[RELEASE_ATTRIBUTE] == "v1.2.3"
@pytest.mark.parametrize("public", [True, False], ids=["public", "private"])
def test_trace_public_flag_lands_on_the_root_observation(client, public):
"""v2 took ``public`` on ``trace()``; v4 reads ``langfuse.trace.public`` off the root observation."""
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="a" * 32, parent_observation_id=None)
start_generation(
client=lf,
context=context,
name="gen",
start_time=CALL_START,
claim_trace_root=claim_root,
public=public,
attributes={},
).end()
lf.flush()
assert _only_span(exporter, "gen").attributes[PUBLIC_ATTRIBUTE] is public
def test_trace_public_flag_is_absent_when_not_requested(client):
lf, exporter = client
context, claim_root = open_trace_context(client=lf, trace_id="b" * 32, parent_observation_id=None)
start_generation(
client=lf, context=context, name="gen", start_time=CALL_START, claim_trace_root=claim_root, attributes={}
).end()
lf.flush()
assert PUBLIC_ATTRIBUTE not in _only_span(exporter, "gen").attributes
def test_request_release_beats_the_client_wide_release(monkeypatch):
"""A client configured with its own release must not overwrite trace_release."""
monkeypatch.setenv("LANGFUSE_RELEASE", "client-wide-release")
LangfuseResourceManager._instances.pop("pk-release-test", None)
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
lf = Langfuse(
public_key="pk-release-test",
secret_key="sk-release-test",
host="http://127.0.0.1:1",
release="client-wide-release",
tracer_provider=provider,
span_exporter=exporter,
)
context, claim_root = open_trace_context(client=lf, trace_id="f" * 32, parent_observation_id=None)
start_generation(
client=lf,
context=context,
name="gen",
start_time=CALL_START,
claim_trace_root=claim_root,
release="per-request-release",
attributes={},
).end()
lf.flush()
LangfuseResourceManager._instances.pop("pk-release-test", None)
assert _only_span(exporter, "gen").attributes[RELEASE_ATTRIBUTE] == "per-request-release"
@pytest.mark.parametrize(
"supplied, expected",
[
("0123456789abcdef0123456789abcdef", "0123456789abcdef0123456789abcdef"),
("0123456789ABCDEF0123456789ABCDEF", "0123456789abcdef0123456789abcdef"),
("3fe0c940-b69a-de3b-a77c-06102505349a", "3fe0c940b69ade3ba77c06102505349a"),
],
ids=["already-hex", "uppercase-hex", "uuid-with-dashes"],
)
def test_trace_id_passes_through_when_it_is_already_usable(supplied, expected):
assert resolve_trace_id(supplied) == expected
def test_arbitrary_trace_id_is_hashed_deterministically():
first = resolve_trace_id("order-4471")
assert first == resolve_trace_id("order-4471")
assert len(first) == 32 and first == first.lower()
assert first != resolve_trace_id("order-4472")
@pytest.mark.parametrize("supplied", [12345, 12.5, True, False], ids=["int", "float", "true", "false"])
def test_non_string_trace_id_is_normalized(supplied):
resolved = resolve_trace_id(supplied)
assert len(resolved) == 32
assert resolved == resolve_trace_id(supplied)
@pytest.mark.parametrize("supplied", [12345, 12.5, True, False], ids=["int", "float", "true", "false"])
def test_non_string_observation_id_is_normalized(supplied):
resolved = resolve_observation_id(supplied)
assert len(resolved) == 16
assert resolved == resolve_observation_id(supplied)
def test_hyphen_only_trace_ids_are_deterministic():
assert resolve_trace_id("---") == resolve_trace_id("---")
def test_trace_id_with_trailing_newline_is_hashed():
supplied = "a" * 32 + "\n"
resolved = resolve_trace_id(supplied)
assert resolved != supplied
assert len(resolved) == 32
def test_missing_trace_id_still_yields_a_valid_trace_id():
generated = resolve_trace_id(None)
assert len(generated) == 32
assert int(generated, 16) >= 0
@pytest.mark.parametrize(
"supplied, expected",
[
("0123456789abcdef", "0123456789abcdef"),
(None, None),
("", None),
],
ids=["already-hex", "none", "empty"],
)
def test_observation_id_normalisation(supplied, expected):
assert resolve_observation_id(supplied) == expected
def test_arbitrary_observation_id_is_hashed_to_a_span_id():
resolved = resolve_observation_id("my-parent-observation")
assert len(resolved) == 16
assert resolved == resolve_observation_id("my-parent-observation")
PUBLIC_KEY = "pk-lifecycle-test"
@pytest.fixture(autouse=True)
def _clean_registry():
LangfuseResourceManager._instances.pop(PUBLIC_KEY, None)
yield
LangfuseResourceManager._instances.pop(PUBLIC_KEY, None)
def _lifecycle_client(secret_key="sk-original", host="http://127.0.0.1:1"):
return Langfuse(
public_key=PUBLIC_KEY,
secret_key=secret_key,
host=host,
tracer_provider=build_isolated_tracer_provider(environment=None, release=None),
)
@pytest.mark.parametrize("unsupported", ["2.59.7", "3.15.0", "5.0.0"], ids=["v2", "v3", "v5"])
def test_unsupported_sdk_fails_loudly_rather_than_dropping_every_event(unsupported):
with pytest.raises(ImportError) as raised:
raise_if_unsupported_langfuse_version(unsupported)
assert unsupported in str(raised.value)
assert MINIMUM_LANGFUSE_VERSION in str(raised.value)
def test_supported_sdk_is_accepted():
assert raise_if_unsupported_langfuse_version(installed_langfuse_version()) is None
def test_isolated_provider_carries_environment_and_release():
provider = build_isolated_tracer_provider(environment="staging", release="v9")
attributes = provider.resource.attributes
assert attributes["langfuse.environment"] == "staging"
assert attributes["langfuse.release"] == "v9"
def test_langfuse_sample_rate_drops_spans_on_the_isolated_provider(monkeypatch):
"""The SDK only installs its sampler on providers it builds itself; v2 sampled via the same env var."""
monkeypatch.setenv("LANGFUSE_SAMPLE_RATE", "0")
dropped_exporter = InMemorySpanExporter()
dropping_provider = build_isolated_tracer_provider(environment=None, release=None)
dropping_provider.add_span_processor(SimpleSpanProcessor(dropped_exporter))
dropping_provider.get_tracer("test").start_span("dropped").end()
assert not dropped_exporter.get_finished_spans()
monkeypatch.delenv("LANGFUSE_SAMPLE_RATE")
kept_exporter = InMemorySpanExporter()
keeping_provider = build_isolated_tracer_provider(environment=None, release=None)
keeping_provider.add_span_processor(SimpleSpanProcessor(kept_exporter))
keeping_provider.get_tracer("test").start_span("kept").end()
assert [span.name for span in kept_exporter.get_finished_spans()] == ["kept"]
def test_invalid_sample_rate_fails_at_construction_like_the_sdk(monkeypatch):
monkeypatch.setenv("LANGFUSE_SAMPLE_RATE", "1.5")
with pytest.raises(ValueError, match=r"between 0\.0 and 1\.0"):
build_isolated_tracer_provider(environment=None, release=None)
def test_environment_override_lands_per_span_despite_shared_resources():
"""The SDK registry is keyed on public key alone, so a second client for the
same key adopts the first client's provider; the observation wrapper stamps
each span with its own client's environment, which the server prefers over
the resource-level value."""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
first = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
environment="prod",
tracer_provider=provider,
span_exporter=exporter,
)
second = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
environment="staging",
)
assert second._resources is first._resources
for client, environment in ((first, "prod"), (second, "staging")):
context, claim_trace_root = open_trace_context(client=client, trace_id="a" * 32, parent_observation_id=None)
start_generation(
client=client,
context=context,
name=f"generation-{environment}",
start_time=CALL_START,
claim_trace_root=claim_trace_root,
attributes={},
).end()
first.flush()
spans = {span.name: span for span in exporter.get_finished_spans()}
assert spans["generation-prod"].attributes["langfuse.environment"] == "prod"
assert spans["generation-staging"].attributes["langfuse.environment"] == "staging"
def test_client_does_not_take_over_the_process_tracer_provider():
# the global provider can only be set once per process, so assert it is left
# alone rather than assuming this test is the one that installed it
provider_before = otel_trace.get_tracer_provider()
client = _lifecycle_client()
assert otel_trace.get_tracer_provider() is provider_before
assert client._resources.tracer_provider is not provider_before
active = getattr(provider_before, "_active_span_processor", None)
if active is not None:
assert not any("Langfuse" in type(processor).__name__ for processor in active._span_processors)
def test_rotated_credentials_replace_the_cached_client():
original = _lifecycle_client(secret_key="sk-original", host="http://127.0.0.1:1")
original_resources = original._resources
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-rotated", base_url="http://127.0.0.1:2")
rotated = _lifecycle_client(secret_key="sk-rotated", host="http://127.0.0.1:2")
assert rotated._resources is not original_resources
assert rotated._resources.secret_key == "sk-rotated"
assert rotated._resources.base_url == "http://127.0.0.1:2"
def test_unchanged_credentials_keep_the_cached_client():
original = _lifecycle_client()
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-original", base_url="http://127.0.0.1:1")
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is original._resources
def test_eviction_flushes_queued_observations_before_tearing_down():
"""An observation already ended when the cache evicts must still reach langfuse."""
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.langfuse.langfuse_sdk import (
open_trace_context,
start_generation,
)
exporter = InMemorySpanExporter()
provider = TracerProvider()
# a long delay keeps the span queued, so only the shutdown can flush it
provider.add_span_processor(BatchSpanProcessor(exporter, schedule_delay_millis=600000))
client = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
context, claim_root = open_trace_context(client=client, trace_id="a" * 32, parent_observation_id=None)
start_generation(
client=client, context=context, name="in-flight", start_time=None, claim_trace_root=claim_root, attributes={}
).end()
assert exporter.get_finished_spans() == ()
shutdown_langfuse_client(client)
assert any(span.name == "in-flight" for span in exporter.get_finished_spans())
def test_shutdown_deregisters_so_a_later_client_is_not_a_corpse():
client = _lifecycle_client()
resources = client._resources
shutdown_langfuse_client(client)
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is not resources
def _shared_resources_pair():
"""The SDK hands a second client on the same public key the first client's resources."""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
_litellm_built_providers.add(provider)
first = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
second = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=build_isolated_tracer_provider(environment="per-key-override", release=None),
)
assert second._resources is first._resources
register_langfuse_client(first)
register_langfuse_client(second)
return first, second, exporter
def _exports(client, exporter, name):
client.start_observation(name=name).end()
client.flush()
return any(span.name == name for span in exporter.get_finished_spans())
def test_garbage_collected_throwaway_clients_do_not_hold_shared_resources_open():
"""A health probe or alerting lookup builds a client it never shuts down.
Once such a client is garbage collected it must stop counting, or the last
managed client's shutdown would skip the teardown forever.
"""
import gc
first, second, exporter = _shared_resources_pair()
throwaway = Langfuse(public_key=PUBLIC_KEY, secret_key="sk-original", host="http://127.0.0.1:1")
register_langfuse_client(throwaway)
shutdown_langfuse_client(second)
del throwaway
gc.collect()
shutdown_langfuse_client(first)
assert not _exports(first, exporter, "after-managed-teardown")
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is not first._resources
def test_evicting_a_client_that_shares_resources_keeps_the_other_exporting():
"""A per-key ``langfuse_environment`` override is a second client on the global key.
When the cache evicts it, the global logger must keep exporting.
"""
first, second, exporter = _shared_resources_pair()
shutdown_langfuse_client(second)
assert _exports(first, exporter, "after-sibling-eviction")
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is first._resources
def test_eviction_defers_teardown_until_active_callback_finishes():
"""A cached client must keep exporting while its callback lease is active."""
exporter = InMemorySpanExporter()
provider = build_isolated_tracer_provider(environment=None, release=None)
provider.add_span_processor(SimpleSpanProcessor(exporter))
client = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
register_langfuse_client(client)
def evict() -> None:
shutdown_langfuse_client(client)
with lease_langfuse_client(client):
evictor = threading.Thread(target=evict)
evictor.start()
evictor.join(timeout=5)
assert not evictor.is_alive()
assert not exporter._stopped
context, claim_root = open_trace_context(client=client, trace_id="a" * 32, parent_observation_id=None)
start_generation(
client=client,
context=context,
name="active-callback",
start_time=None,
claim_trace_root=claim_root,
attributes={},
).end()
client.flush()
assert any(span.name == "active-callback" for span in exporter.get_finished_spans())
evictor.join(timeout=5)
assert not evictor.is_alive()
assert exporter._stopped
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is not client._resources
def test_teardown_failure_does_not_strand_queued_clients(monkeypatch):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
first = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
clients = (
first,
Langfuse(public_key=PUBLIC_KEY, secret_key="sk-original", host="http://127.0.0.1:1"),
Langfuse(public_key=PUBLIC_KEY, secret_key="sk-original", host="http://127.0.0.1:1"),
)
assert len({client._resources for client in clients}) == 1
for client in clients:
register_langfuse_client(client)
state = _lifecycle_state(clients[0])
original_teardown = _teardown_langfuse_client
calls = []
def teardown(client):
calls.append(client)
original_teardown(client)
if len(calls) == 1:
raise RuntimeError("teardown failed")
monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk._teardown_langfuse_client", teardown)
with lease_langfuse_client(clients[0]):
for client in clients:
shutdown_langfuse_client(client)
assert len(calls) == 3
assert not state.pending_clients
assert not state.teardown_in_progress
def test_queued_eviction_waits_for_the_last_of_two_overlapping_leases():
exporter = InMemorySpanExporter()
provider = build_isolated_tracer_provider(environment=None, release=None)
provider.add_span_processor(SimpleSpanProcessor(exporter))
client = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
register_langfuse_client(client)
with lease_langfuse_client(client):
with lease_langfuse_client(client):
shutdown_langfuse_client(client)
assert not exporter._stopped
assert exporter._stopped
def test_a_client_adopted_during_deferred_teardown_keeps_exporting():
"""The registry hands the same bundle back out while its teardown is queued behind a lease;
the holder count must degrade that teardown to a flush."""
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
_litellm_built_providers.add(provider)
evicted = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
register_langfuse_client(evicted)
with lease_langfuse_client(evicted):
shutdown_langfuse_client(evicted)
adopter = Langfuse(public_key=PUBLIC_KEY, secret_key="sk-original", host="http://127.0.0.1:1")
assert adopter._resources is evicted._resources
register_langfuse_client(adopter)
assert _exports(adopter, exporter, "after-deferred-teardown")
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is adopter._resources
def test_leases_on_one_client_do_not_serialise_callbacks():
"""Every langfuse callback in the process shares one client, so leases must overlap."""
client = _lifecycle_client()
both_inside = threading.Barrier(2, timeout=5)
def hold_lease() -> None:
with lease_langfuse_client(client):
both_inside.wait()
holders = tuple(threading.Thread(target=hold_lease) for _ in range(2))
for holder in holders:
holder.start()
for holder in holders:
holder.join(timeout=5)
assert not any(holder.is_alive() for holder in holders)
assert not both_inside.broken
def test_last_client_on_shared_resources_tears_them_down():
first, second, exporter = _shared_resources_pair()
shutdown_langfuse_client(second)
shutdown_langfuse_client(first)
assert not _exports(first, exporter, "after-last-eviction")
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is not first._resources
def test_shutdown_of_a_stale_client_does_not_deregister_the_live_one():
stale = _lifecycle_client(secret_key="sk-original", host="http://127.0.0.1:1")
stale_resources = stale._resources
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-rotated", base_url="http://127.0.0.1:2")
live = _lifecycle_client(secret_key="sk-rotated", host="http://127.0.0.1:2")
shutdown_langfuse_client(stale)
assert stale_resources is not live._resources
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is live._resources
def _rotation_provider():
"""A litellm-built provider on the lifecycle public key, exporting in memory."""
exporter = InMemorySpanExporter()
provider = build_isolated_tracer_provider(environment=None, release=None)
provider.add_span_processor(SimpleSpanProcessor(exporter))
client = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
register_langfuse_client(client)
assert _exports(client, exporter, "before-rotation")
return client, exporter
def test_rotation_retires_the_provider_no_client_is_left_on():
"""Prompt management builds throwaway clients from request credentials.
Alternating the secret for one public key evicts a bundle nobody holds any
more, and its export thread has to go with it or every rotation leaks one.
"""
import gc
client, exporter = _rotation_provider()
del client
gc.collect()
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-rotated", base_url="http://127.0.0.1:2")
assert exporter._stopped
def test_rotation_keeps_a_still_live_client_exporting():
"""The evicted bundle is only retired when nothing is on it; a live logger must survive."""
client, exporter = _rotation_provider()
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-rotated", base_url="http://127.0.0.1:2")
assert _exports(client, exporter, "after-rotation")
def test_a_client_dropped_without_shutdown_gets_its_provider_retired():
"""The prompt-management LRU drops rotated-out clients without shutting them down.
Nothing ever calls ``shutdown_langfuse_client`` on such a client, so the next
lifecycle call has to reap the bundle instead of leaking its export thread.
"""
import gc
client, exporter = _rotation_provider()
evict_stale_langfuse_resources(public_key=PUBLIC_KEY, secret_key="sk-rotated", base_url="http://127.0.0.1:2")
assert _exports(client, exporter, "still-held")
del client
gc.collect()
evict_stale_langfuse_resources(public_key="pk-unrelated", secret_key="sk", base_url="http://127.0.0.1:3")
assert exporter._stopped
def test_the_registrys_current_bundle_is_not_reaped_when_its_clients_die():
"""The registry hands its bundle to the next client on the same key, so a bundle
that is still current keeps its provider even after every client is collected."""
import gc
client, exporter = _rotation_provider()
del client
gc.collect()
evict_stale_langfuse_resources(public_key="pk-unrelated", secret_key="sk", base_url="http://127.0.0.1:3")
successor = Langfuse(public_key=PUBLIC_KEY, secret_key="sk-original", host="http://127.0.0.1:1")
assert _exports(successor, exporter, "after-collection")
def test_a_sweep_overlapping_registration_and_rotation_keeps_the_live_provider():
"""A sweep can snapshot providers before a client registers, then wait on the registry
lock while that client registers and a rotation evicts its fresh bundle. Holders are
re-read after the registry snapshot, so the stale first look must not win."""
import threading
from litellm.integrations.langfuse.langfuse_sdk import _retire_orphaned_providers
exporter = InMemorySpanExporter()
provider = build_isolated_tracer_provider(environment=None, release=None)
provider.add_span_processor(SimpleSpanProcessor(exporter))
client = Langfuse(
public_key=PUBLIC_KEY,
secret_key="sk-original",
host="http://127.0.0.1:1",
tracer_provider=provider,
span_exporter=exporter,
)
registry_lock = LangfuseResourceManager._lock
registry_lock.acquire()
try:
sweeper = threading.Thread(target=_retire_orphaned_providers)
sweeper.start()
sweeper.join(timeout=0.5) # parks on the registry lock once its provider snapshot is taken
register_langfuse_client(client)
LangfuseResourceManager._instances.pop(PUBLIC_KEY, None) # the rotation that evicts the fresh bundle
finally:
registry_lock.release()
sweeper.join(timeout=5)
assert not sweeper.is_alive()
assert _exports(client, exporter, "after-racing-sweep")
def test_ssl_exporter_is_only_built_with_custom_tls_material(monkeypatch, tmp_path):
"""v4 exports over its own OTLP channel, so litellm's CA bundle must be rebuilt onto it."""
import litellm
from litellm.integrations.langfuse.langfuse_sdk import _build_verified_span_exporter
monkeypatch.delenv("SSL_CERTIFICATE", raising=False)
monkeypatch.setattr(litellm, "ssl_verify", True)
monkeypatch.setattr(litellm, "ssl_certificate", None)
assert (
_build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") is None
)
ca_path = tmp_path / "private-ca.pem"
ca_path.write_text("dummy")
monkeypatch.setattr(litellm, "ssl_verify", str(ca_path))
exporter = _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
assert exporter is not None
assert exporter._endpoint == "https://lf.internal.example/api/public/otel/v1/traces"
assert exporter._certificate_file == str(ca_path)
assert exporter._headers["x-langfuse-public-key"] == "pk"
def test_second_client_on_the_same_key_does_not_build_another_provider():
"""A discarded TracerProvider is pinned forever by its atexit hook."""
import gc
from litellm.integrations.langfuse.langfuse_sdk import (
_retire_orphaned_providers,
acquire_langfuse_client,
)
# reap earlier tests' orphans first, so the count below only moves if a provider is built
gc.collect()
_retire_orphaned_providers()
pk = "pk-provider-reuse-test"
LangfuseResourceManager._instances.pop(pk, None)
parameters = {"public_key": pk, "secret_key": "sk-reuse", "base_url": "http://127.0.0.1:1"}
try:
first = acquire_langfuse_client(parameters=parameters, environment=None, release=None, mock_mode=True)
providers_after_first = len(_litellm_built_providers)
second = acquire_langfuse_client(parameters=parameters, environment=None, release=None, mock_mode=True)
assert second._resources is first._resources
assert len(_litellm_built_providers) == providers_after_first
finally:
LangfuseResourceManager._instances.pop(pk, None)

762
tests/test_litellm/integrations/test_langfuse.py
View file

File diff suppressed because it is too large Load diff

16
tests/test_litellm/proxy/health_endpoints/test_health_endpoints.py
View file

@ -4179,3 +4179,19 @@ async def test_health_services_endpoint_pointfive_blocks_non_admin(monkeypatch,
assert str(raised.value.code) == "403"
logger_class.assert_not_called()
@pytest.mark.asyncio
async def test_health_services_endpoint_langfuse_missing_keys_errors(monkeypatch):
"""A disabled v4 client returns False from auth_check instead of raising.
v2 raised out of ``auth_check`` on missing keys, so the endpoint errored;
the endpoint must not report success when the return value says the check
failed.
"""
for key in ("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", "LANGFUSE_HOST", "LANGFUSE_BASE_URL", "LANGFUSE_MOCK"):
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(litellm, "initialized_langfuse_clients", litellm.initialized_langfuse_clients)
with pytest.raises(ProxyException, match="auth_check failed"):
await health_services_endpoint(service="langfuse")