scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-26 01:12:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 145

refactor(langfuse): emit observations as plain OTel spans, keep the SDK for prompts and auth

Browse source 
The callback now owns an isolated TracerProvider and OTLP exporter and builds generation and child spans with public OpenTelemetry APIs plus the LangfuseOtelSpanAttributes constants. Caller trace ids, generation ids, parent observation ids and historical start and end times are honoured through the OTel id generator, remote SpanContext and explicit span timestamps, so no private Langfuse SDK tracing handle is used any more. The Langfuse client stays only for get_prompt and auth_check

This also resolves the gauntlet findings on the previous draft: fresh traces start from an empty context so caller application spans are never stamped, the Slack trace link is read from the request logging state instead of constructing a logger per alert, a truthy non-mapping trace_metadata is serialized instead of raising, trace_input and trace_output land on the root generation, discarding a cached client is done under the lock, and the prompt cache no longer leaks a task manager because the client cache no longer tears down shared providers

Fixtures under tests/logging_callback_tests lose the SDK-private langfuse.internal.as_root marker; every other exported attribute is unchanged

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-17 21:18:36 +00:00
parent 2dbbea5b6b
commit 2f33397626
30 changed files with 1179 additions and 1961 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -66,26 +66,20 @@ async def _add_langfuse_trace_id_to_alert(
-> trace_id
-> litellm_call_id
"""
from litellm.integrations.langfuse.langfuse import LangFuseLogger
from litellm.integrations.langfuse.langfuse import LangFuseLogger, resolve_langfuse_host
callbacks: Final = litellm.logging_callback_manager._get_all_callbacks()
if not any(callback == "langfuse" or isinstance(callback, LangFuseLogger) for callback in callbacks):
return None
if request_data is not None and request_data.get("litellm_logging_obj", None) is not None:
trace_id: str | None = None
litellm_logging_obj: Final[Logging] = request_data["litellm_logging_obj"]
if request_data is None or request_data.get("litellm_logging_obj", None) is None:
return None
for _ in range(3):
trace_id = litellm_logging_obj._get_trace_id(service_name="langfuse")
if trace_id is not None:
break
await asyncio.sleep(3) # wait 3s before retrying for trace id
if trace_id is None:
return None
langfuse_object: Final = litellm_logging_obj._get_callback_object(service_name="langfuse")
if isinstance(langfuse_object, LangFuseLogger):
return f"{langfuse_object.langfuse_host}/trace/{trace_id}"
litellm_logging_obj: Final[Logging] = request_data["litellm_logging_obj"]
host: Final = resolve_langfuse_host(litellm_logging_obj.standard_callback_dynamic_params.get("langfuse_host"))
for _ in range(3):
if (trace_id := litellm_logging_obj._get_trace_id(service_name="langfuse")) is not None:
return f"{host}/trace/{trace_id}"
await asyncio.sleep(3) # wait 3s before retrying for trace id
return None

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

@ -3,7 +3,7 @@
import os
import re
import traceback
from collections.abc import Callable, Iterable, Mapping, Sequence
from collections.abc import Callable, Iterable, Mapping
from datetime import datetime
from functools import lru_cache
from importlib.metadata import version
@ -45,13 +45,15 @@ from litellm.types.utils import (
)
if TYPE_CHECKING:
from langfuse import Langfuse, LangfuseGeneration
from langfuse import Langfuse
from litellm.integrations.langfuse.langfuse_sdk import LangfuseObservation, LangfuseTracing
from litellm.litellm_core_utils.litellm_logging import DynamicLoggingCache
else:
DynamicLoggingCache = Any
Langfuse = Any
LangfuseGeneration = Any
LangfuseObservation = Any
LangfuseTracing = Any
_DENIED_STEERING_KEYS: Final = frozenset({"headers", "endpoint", "caching_groups", "previous_models"})
@ -190,8 +192,8 @@ 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.
`LangfuseOtelSpanAttributes` 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
@ -205,52 +207,6 @@ def raise_if_unsupported_langfuse_version(installed_version: str) -> None:
)
_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 _propagated_trace_metadata(value: object) -> Mapping[str, str] | None:
"""v2's ``trace(metadata=...)`` took any JSON; v4 propagates one flat string per key."""
entries: Final = _object_mapping(value)
if not entries:
return None
return MappingProxyType({str(key): str(item)[:_PROPAGATED_VALUE_MAX_CHARS] for key, item in entries.items()})
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.
"""
trace_metadata: Final = _propagated_trace_metadata(trace_params.get("metadata"))
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
},
**({"metadata": trace_metadata} if trace_metadata is not None else {}),
}
)
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
@ -263,28 +219,6 @@ def _trace_public_flag(value: object) -> bool | 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,
@ -299,11 +233,15 @@ 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 = (
return public_key, secret_key, resolve_langfuse_host(langfuse_host)
def resolve_langfuse_host(langfuse_host: object = None) -> str:
"""The Langfuse base URL for ``langfuse_host`` with the env fallbacks, always carrying a scheme."""
resolved: Final = str(
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
return resolved if resolved.startswith(("http://", "https://")) else f"http://{resolved}"
def warn_if_upstream_langfuse_configured() -> None:
@ -356,9 +294,6 @@ class LangFuseLogger:
langfuse_host=langfuse_host,
allow_env_credentials=allow_env_credentials,
)
if not (self.langfuse_host.startswith("http://") or self.langfuse_host.startswith("https://")):
# add http:// if unset, assume communicating over private network - e.g. render
self.langfuse_host = "http://" + self.langfuse_host
_env_override: Final = str(langfuse_environment).strip() if langfuse_environment is not None else None
if _env_override:
validate_langfuse_environment_value(_env_override)
@ -388,6 +323,17 @@ class LangFuseLogger:
"environment": self.langfuse_environment,
}
self.Langfuse: Langfuse = self.safe_init_langfuse_client(self.langfuse_client_parameters)
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_tracing
self.tracing: LangfuseTracing = acquire_langfuse_tracing(
public_key=str(self.public_key),
secret_key=str(self.secret_key),
base_url=self.langfuse_host,
environment=self.langfuse_environment,
release=self.langfuse_release,
flush_interval=self.langfuse_flush_interval,
mock_mode=self.is_mock_mode,
)
# set the current langfuse project id in the environ
# this is used by Alerting to link to the correct project
@ -421,11 +367,11 @@ class LangFuseLogger:
raise Exception(
f"Max langfuse clients reached: {litellm.initialized_langfuse_clients} is greater than {MAX_LANGFUSE_INITIALIZED_CLIENTS}"
)
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client
from litellm.integrations.langfuse.langfuse_sdk import build_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(
langfuse_client: Final = build_langfuse_client(
parameters=parameters,
environment=environment_param,
release=release_param,
@ -435,20 +381,10 @@ class LangFuseLogger:
verbose_logger.debug("Created langfuse client number %s", litellm.initialized_langfuse_clients)
return langfuse_client
def _renew_langfuse_client(self) -> Langfuse:
"""Replace a client the cache evicted after handing this logger to the callback.
Bypasses the initialized-client ceiling: eviction already released this logger's slot, and the
replacement is never evicted itself, so its provider is retired with the logger instead.
"""
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client
return acquire_langfuse_client(
parameters=self.langfuse_client_parameters,
environment=self.langfuse_environment,
release=self.langfuse_release,
mock_mode=self.is_mock_mode,
)
def flush(self) -> None:
"""Push every queued observation to Langfuse before the process goes away."""
self.tracing.flush()
self.Langfuse.flush()
@staticmethod
def add_metadata_from_header(litellm_params: dict, metadata: dict) -> dict[str, object]:
@ -544,24 +480,20 @@ class LangFuseLogger:
status_message=status_message,
)
verbose_logger.debug("OUTPUT IN LANGFUSE: %s; original: %s", output, response_obj)
from litellm.integrations.langfuse.langfuse_sdk import lease_langfuse_client
with lease_langfuse_client(self.Langfuse, self._renew_langfuse_client) as leased:
self.Langfuse = leased
trace_id, generation_id = self._log_langfuse_v2(
user_id=user_id,
metadata=metadata,
litellm_params=litellm_params,
output=output,
start_time=start_time,
end_time=end_time,
kwargs=kwargs,
optional_params=optional_params,
input=input,
response_obj=response_obj,
level=level,
litellm_call_id=litellm_call_id,
)
trace_id, generation_id = self._log_langfuse_v2(
user_id=user_id,
metadata=metadata,
litellm_params=litellm_params,
output=output,
start_time=start_time,
end_time=end_time,
kwargs=kwargs,
optional_params=optional_params,
input=input,
response_obj=response_obj,
level=level,
litellm_call_id=litellm_call_id,
)
verbose_logger.debug("Langfuse Layer Logging - final response object: %s", response_obj)
verbose_logger.info("Langfuse Layer Logging - logging success")
@ -939,7 +871,7 @@ class LangFuseLogger:
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 {}),
**(_object_mapping(trace_params.get("metadata")) or _NO_METADATA),
**log_requester_metadata(redact_user_api_key_info(metadata=allowlisted_metadata)), # pyright: ignore[reportArgumentType] # TypedDict in, plain metadata dict out
**enrichments,
**_lookup_ids(litellm_call_id, response_obj),
@ -961,57 +893,66 @@ class LangFuseLogger:
if masked_output is not None and isinstance(masked_output, str) and level == "ERROR":
generation_params["status_message"] = masked_output
generation_params["completion_start_time"] = kwargs.get("completion_start_time", None)
# 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,
observation_attributes,
resolve_observation_id,
resolve_trace_id,
start_generation,
to_unix_nanos,
trace_attributes,
)
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
existing_trace=existing_trace_id is not None,
)
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")),
observation_id=resolve_observation_id(generation_params["id"]),
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,
)
log_provider_specific_information_as_span(
client=self.Langfuse, parent=generation, enrichments=enrichments
)
self._log_guardrail_information_as_span(
client=self.Langfuse, parent=generation, standard_logging_object=standard_logging_object
)
generation.end(end_time=to_unix_nanos(end_time))
continued_trace: Final = existing_trace_id is not None
trace_public: Final = _trace_public_flag(trace_params.get("public"))
trace_input: Final = trace_params.get("input")
trace_output: Final = trace_params.get("output")
trace_level_attributes: Final = trace_attributes(
name=trace_params.get("name"),
user_id=trace_params.get("user_id"),
session_id=trace_params.get("session_id"),
version=trace_params.get("version"),
release=trace_params.get("release"),
tags=trace_params.get("tags"),
metadata=trace_params.get("metadata"),
public=trace_public,
input=trace_input if continued_trace or trace_input != generation_params["input"] else None,
output=trace_output if continued_trace or trace_output != generation_params["output"] else None,
)
generation_attributes: Final = observation_attributes(
observation_type="generation",
input=generation_params["input"],
output=generation_params["output"],
metadata=generation_params["metadata"],
level=level,
status_message=generation_params.get("status_message"),
version=generation_params["version"],
model=model_name,
model_parameters=optional_params,
usage_details=usage_details,
cost_details=generation_params["cost_details"],
completion_start_time=kwargs.get("completion_start_time", None),
prompt=generation_params.get("prompt"),
)
generation: Final = start_generation(
tracing=self.tracing,
trace_id=resolved_trace_id,
parent_observation_id=resolve_observation_id(parent_observation_id), # pyright: ignore[reportArgumentType] # metadata value, str or None at runtime
existing_trace=continued_trace,
observation_id=resolve_observation_id(generation_params["id"]),
name=generation_params["name"], # pyright: ignore[reportArgumentType] # always the str set a few lines up
start_time=start_time,
public=trace_public,
attributes=MappingProxyType({**generation_attributes, **trace_level_attributes}),
)
log_provider_specific_information_as_span(tracing=self.tracing, parent=generation, enrichments=enrichments)
self._log_guardrail_information_as_span(
tracing=self.tracing, parent=generation, standard_logging_object=standard_logging_object
)
generation.end(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 requested generation_id after
# resolve_observation_id, unless the provider was adopted from user code.
# The observation id is the requested generation_id after resolve_observation_id.
return resolved_trace_id, generation.id
except Exception:
verbose_logger.error("Langfuse Layer Error - %s", traceback.format_exc())
@ -1150,8 +1091,8 @@ class LangFuseLogger:
def _log_guardrail_information_as_span(
self,
client: "Langfuse",
parent: "LangfuseGeneration",
tracing: "LangfuseTracing",
parent: "LangfuseObservation",
standard_logging_object: StandardLoggingPayload | None,
):
"""
@ -1173,7 +1114,7 @@ class LangFuseLogger:
)
return
from litellm.integrations.langfuse.langfuse_sdk import start_child_span, to_unix_nanos
from litellm.integrations.langfuse.langfuse_sdk import observation_attributes, start_child_span
for guardrail_entry in guardrail_information:
if not isinstance(guardrail_entry, dict):
@ -1184,23 +1125,26 @@ class LangFuseLogger:
continue
span = start_child_span(
client=client,
tracing=tracing,
parent=parent,
name="guardrail",
start_time=guardrail_entry.get("start_time", None),
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),
},
},
attributes=observation_attributes(
observation_type="span",
input=guardrail_entry.get("guardrail_request", None),
output=guardrail_entry.get("guardrail_response", None),
metadata=MappingProxyType(
{
"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(end_time=to_unix_nanos(guardrail_entry.get("end_time", None)))
span.end(guardrail_entry.get("end_time", None))
def _add_prompt_to_generation_params(
@ -1278,8 +1222,8 @@ def _add_prompt_to_generation_params(
def log_provider_specific_information_as_span(
*,
client: "Langfuse",
parent: "LangfuseGeneration",
tracing: "LangfuseTracing",
parent: "LangfuseObservation",
enrichments: Mapping[str, Any],
):
"""Logs provider-specific information as spans under the generation."""
@ -1295,24 +1239,24 @@ def log_provider_specific_information_as_span(
for elem in vertex_ai_grounding_metadata:
if isinstance(elem, dict):
for key, value in elem.items():
_end_grounding_span(client=client, parent=parent, name=key, value=value)
_end_grounding_span(tracing=tracing, parent=parent, name=key, value=value)
else:
_end_grounding_span(client=client, parent=parent, name="vertex_ai_grounding_metadata", value=elem)
_end_grounding_span(tracing=tracing, parent=parent, name="vertex_ai_grounding_metadata", value=elem)
else:
_end_grounding_span(
client=client, parent=parent, name="vertex_ai_grounding_metadata", value=vertex_ai_grounding_metadata
tracing=tracing, parent=parent, name="vertex_ai_grounding_metadata", value=vertex_ai_grounding_metadata
)
def _end_grounding_span(*, client: "Langfuse", parent: "LangfuseGeneration", name: str, value: object) -> None:
from litellm.integrations.langfuse.langfuse_sdk import start_child_span
def _end_grounding_span(*, tracing: "LangfuseTracing", parent: "LangfuseObservation", name: str, value: object) -> None:
from litellm.integrations.langfuse.langfuse_sdk import observation_attributes, start_child_span
start_child_span(
client=client,
tracing=tracing,
parent=parent,
name=name,
start_time=None,
attributes={"input": value}, # mutable-ok: langfuse serializes this payload
attributes=observation_attributes(observation_type="span", input=value),
).end()

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

@ -72,7 +72,7 @@ def langfuse_client_init(
"""
try:
from langfuse import (
Langfuse, # noqa: F401 # the import is the install probe; construction moved to acquire_langfuse_client
Langfuse, # noqa: F401 # the import is the install probe; construction happens in build_langfuse_client
)
except Exception as e:
raise Exception(
@ -124,9 +124,9 @@ def langfuse_client_init(
parameters["environment"] = LangFuseLogger.resolve_deployment_environment()
from .langfuse_sdk import acquire_langfuse_client
from .langfuse_sdk import build_langfuse_client
client: Final = acquire_langfuse_client(
client: Final = build_langfuse_client(
parameters=parameters,
environment=parameters["environment"],
release=langfuse_release,
@ -152,6 +152,8 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
flush_interval=1,
):
from .langfuse_sdk import acquire_langfuse_tracing
self.langfuse_sdk_version = installed_langfuse_version()
self.Langfuse = langfuse_client_init(
langfuse_public_key=langfuse_public_key,
@ -159,6 +161,20 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
langfuse_host=langfuse_host,
flush_interval=flush_interval,
)
public_key, secret_key, host = resolve_langfuse_credentials(
langfuse_public_key=langfuse_public_key,
langfuse_secret=langfuse_secret,
langfuse_host=langfuse_host,
)
self.tracing = acquire_langfuse_tracing(
public_key=str(public_key),
secret_key=str(secret_key),
base_url=host,
environment=LangFuseLogger.resolve_deployment_environment(),
release=os.getenv("LANGFUSE_RELEASE"),
flush_interval=LangFuseLogger._get_langfuse_flush_interval(flush_interval), # pyright: ignore[reportPrivateUsage] # shared env-fallback helper, not part of the logger's API
mock_mode=should_use_langfuse_mock(),
)
@property
def integration_name(self):

906
litellm/integrations/langfuse/langfuse_sdk.py
View file

File diff suppressed because it is too large Load diff

34
litellm/litellm_core_utils/litellm_logging.py
View file

@ -3683,40 +3683,6 @@ class Logging(LiteLLMLoggingBaseClass):
return trace_id
def _get_callback_object(self, service_name: Literal["langfuse"]) -> Any | None:
"""
Return dynamic callback object.
Meant to solve issue when doing key-based/team-based logging
"""
global langFuseLogger
if service_name == "langfuse":
if langFuseLogger is None or (
(
self.standard_callback_dynamic_params.get("langfuse_public_key") is not None
and self.standard_callback_dynamic_params.get("langfuse_public_key") != langFuseLogger.public_key
)
or (
self.standard_callback_dynamic_params.get("langfuse_public_key") is not None
and self.standard_callback_dynamic_params.get("langfuse_public_key") != langFuseLogger.public_key
)
or (
self.standard_callback_dynamic_params.get("langfuse_host") is not None
and self.standard_callback_dynamic_params.get("langfuse_host") != langFuseLogger.langfuse_host
)
):
return LangFuseLogger(
langfuse_public_key=self.standard_callback_dynamic_params.get("langfuse_public_key"),
langfuse_secret=self.standard_callback_dynamic_params.get("langfuse_secret")
or self.standard_callback_dynamic_params.get("langfuse_secret_key"),
langfuse_host=self.standard_callback_dynamic_params.get("langfuse_host"),
allow_env_credentials=self.standard_callback_dynamic_params.get("langfuse_host") is None,
)
return langFuseLogger
return None
def handle_sync_success_callbacks_for_async_calls(
self,
result: Any,

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

@ -1,10 +1,8 @@
"""
This is a cache for LangfuseLoggers.
Langfuse Python SDK initializes a thread for each client.
This ensures we do
1. Proper cleanup of Langfuse initialized clients.
1. Release the initialized-client slot a LangfuseLogger holds when it expires.
2. Re-use created langfuse clients.
"""
@ -21,34 +19,17 @@ from ...caching import InMemoryCache
class LangfuseInMemoryCache(InMemoryCache):
"""
Ensures we do proper cleanup of Langfuse initialized clients.
Releases the initialized-client slot of a LangfuseLogger when it expires.
Langfuse Python SDK initializes a thread for each client, we need to call Langfuse.shutdown() to properly cleanup.
This ensures we do proper cleanup of Langfuse initialized clients.
Export channels are shared per credential set and outlive the logger, so
nothing else needs tearing down (https://github.com/BerriAI/litellm/issues/11169).
"""
def _remove_key(self, key: str) -> None:
"""
Override _remove_key in InMemoryCache to ensure we do proper cleanup of Langfuse initialized clients.
LangfuseLoggers consume threads when initalized, this shuts them down when they are expired
Relevant Issue: https://github.com/BerriAI/litellm/issues/11169
"""
from litellm.integrations.langfuse.langfuse import LangFuseLogger
if isinstance(self.cache_dict[key], LangFuseLogger):
_created_langfuse_logger: Final[LangFuseLogger] = self.cache_dict[key]
#########################################################
# Clean up Langfuse initialized clients
#########################################################
from litellm.integrations.langfuse.langfuse_sdk import (
shutdown_langfuse_client,
)
litellm.initialized_langfuse_clients -= 1
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.

2
litellm/proxy/proxy_server.py
View file

@ -1033,7 +1033,7 @@ async def proxy_shutdown_event(worker_heartbeat: ProxyWorkerHeartbeat | None = N
from litellm.utils import langFuseLogger
if langFuseLogger is not None:
langFuseLogger.Langfuse.flush()
langFuseLogger.flush()
except Exception:
# [DO NOT BLOCK shutdown events for this]
pass

1
tests/logging_callback_tests/langfuse_expected_request_body/completion.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_bedrock_call.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 6e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_complex_metadata.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_no_choices.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 1.9999999999999998e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_router.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 1.9999999999999998e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_tags.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_tags_stream.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/completion_with_vertex_call.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 7.5e-06
},

1
tests/logging_callback_tests/langfuse_expected_request_body/complex_metadata.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/complex_metadata_2.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/empty_metadata.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/metadata_with_function.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/metadata_with_lock.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/nested_metadata.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/simple_metadata.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/simple_metadata2.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

1
tests/logging_callback_tests/langfuse_expected_request_body/simple_metadata3.json
View file

@ -2,7 +2,6 @@
"name": "litellm-acompletion",
"parent_span_id": null,
"attributes": {
"langfuse.internal.as_root": true,
"langfuse.observation.cost_details": {
"total": 3.5e-05
},

9
tests/logging_callback_tests/test_langfuse_e2e_test.py
View file

@ -27,15 +27,16 @@ import pytest_asyncio
LANGFUSE_EXPORT_POST: Final = "requests.Session.post"
LANGFUSE_EXPORT_PATH: Final = "/api/public/otel/v1/traces"
_LITELLM_OWNED_ATTRIBUTES: Final = frozenset(
_PER_RUN_ATTRIBUTES: Final = frozenset(
{
"langfuse.internal.is_app_root",
"langfuse.observation.completion_start_time",
"langfuse.observation.metadata.applied_guardrails",
"langfuse.observation.metadata.cache_hit",
"langfuse.observation.metadata.hidden_params",
"langfuse.observation.metadata.litellm_call_id",
"langfuse.observation.metadata.litellm_response_cost",
"langfuse.observation.metadata.requester_metadata",
"langfuse.observation.metadata.response_id",
"langfuse.observation.metadata.usage_object",
}
)
@ -87,8 +88,8 @@ def _comparable(span: Mapping[str, object]) -> dict[str, object]:
assert isinstance(attributes, dict)
return {
"name": span["name"],
"parent_span_id": None if attributes.get("langfuse.internal.as_root") else span["parent_span_id"],
"attributes": {key: value for key, value in sorted(attributes.items()) if key not in _LITELLM_OWNED_ATTRIBUTES},
"parent_span_id": span["parent_span_id"],
"attributes": {key: value for key, value in sorted(attributes.items()) if key not in _PER_RUN_ATTRIBUTES},
}

35
tests/test_litellm/integrations/SlackAlerting/test_slack_alerting_utils.py
View file

@ -31,32 +31,35 @@ async def test_langfuse_not_initialized_returns_none_early():
@pytest.mark.asyncio
async def test_langfuse_trace_url_uses_logger_host(monkeypatch):
from litellm.integrations.langfuse.langfuse import LangFuseLogger
async def test_langfuse_trace_url_uses_the_request_host_without_building_a_logger(monkeypatch):
"""Key-scoped callbacks point at their own Langfuse host; the alert link follows it.
The lookup must not construct a LangFuseLogger per alert, or an alert storm
exhausts the initialized-client ceiling and takes the callback down with it.
"""
monkeypatch.setattr(litellm, "success_callback", ["langfuse"])
logger = LangFuseLogger(
langfuse_public_key="pk-slack-test",
langfuse_secret="sk-slack-test",
langfuse_host="http://127.0.0.1:1",
)
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
logging_obj = MagicMock()
logging_obj._get_trace_id.return_value = "abc123"
logging_obj._get_callback_object.return_value = logger
logging_obj.standard_callback_dynamic_params = {"langfuse_host": "http://127.0.0.1:1"}
result = await _add_langfuse_trace_id_to_alert({"litellm_logging_obj": logging_obj})
assert result == "http://127.0.0.1:1/trace/abc123"
assert litellm.initialized_langfuse_clients == 0
@pytest.mark.asyncio
async def test_langfuse_trace_url_skips_non_langfuse_callback(monkeypatch):
async def test_langfuse_trace_url_falls_back_to_the_env_host(monkeypatch):
monkeypatch.setattr(litellm, "success_callback", ["langfuse"])
monkeypatch.setenv("LANGFUSE_HOST", "langfuse.internal:3000")
logging_obj = MagicMock()
logging_obj._get_trace_id.return_value = "abc123"
logging_obj._get_callback_object.return_value = object()
logging_obj.standard_callback_dynamic_params = {}
assert await _add_langfuse_trace_id_to_alert({"litellm_logging_obj": logging_obj}) is None
assert await _add_langfuse_trace_id_to_alert({"litellm_logging_obj": logging_obj}) == (
"http://langfuse.internal:3000/trace/abc123"
)
@pytest.mark.asyncio
@ -75,7 +78,7 @@ async def test_langfuse_trace_url_when_callback_registered_as_logger_instance(mo
monkeypatch.setattr(litellm, "callbacks", [])
logging_obj = MagicMock()
logging_obj._get_trace_id.return_value = "trace-from-instance"
logging_obj._get_callback_object.return_value = logger
logging_obj.standard_callback_dynamic_params = {"langfuse_host": "http://127.0.0.1:1"}
result = await _add_langfuse_trace_id_to_alert({"litellm_logging_obj": logging_obj})
@ -84,16 +87,10 @@ async def test_langfuse_trace_url_when_callback_registered_as_logger_instance(mo
@pytest.mark.asyncio
async def test_langfuse_trace_url_absent_when_trace_id_never_arrives(monkeypatch):
from litellm.integrations.langfuse.langfuse import LangFuseLogger
monkeypatch.setattr(litellm, "success_callback", ["langfuse"])
monkeypatch.setattr("litellm.integrations.SlackAlerting.utils.asyncio.sleep", AsyncMock())
logging_obj = MagicMock()
logging_obj._get_trace_id.return_value = None
logging_obj._get_callback_object.return_value = LangFuseLogger(
langfuse_public_key="pk-slack-none",
langfuse_secret="sk-slack-none",
langfuse_host="http://127.0.0.1:1",
)
logging_obj.standard_callback_dynamic_params = {"langfuse_host": "http://127.0.0.1:1"}
assert await _add_langfuse_trace_id_to_alert({"litellm_logging_obj": logging_obj}) is None

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

@ -83,7 +83,7 @@ class TestLangfusePromptManagement:
),
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
), # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
patch(
"litellm.llms.custom_httpx.http_handler.get_ssl_configuration",
return_value=False,
@ -127,7 +127,7 @@ def test_langfuse_client_init_resolves_deployment_environment(monkeypatch, env_v
monkeypatch.setattr(_RecordingLangfuseForEnv, "last_environment", None)
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
): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
langfuse_client_init.cache_clear()
langfuse_client_init()
langfuse_client_init.cache_clear()
@ -145,7 +145,7 @@ def test_langfuse_client_init_warns_that_upstream_langfuse_is_ignored(monkeypatc
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
), # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
caplog.at_level("WARNING", logger="LiteLLM"),
):
langfuse_client_init.cache_clear()
@ -158,20 +158,13 @@ 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.
this logger is the one the standard proxy path emits observations through;
they travel over litellm's 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,
)
import litellm
received = []
@ -192,35 +185,34 @@ def test_langfuse_client_init_mock_mode_makes_no_network_calls(monkeypatch):
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()
now: Final = datetime.now(timezone.utc)
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",
logger = LangfusePromptManagement()
logged = logger.log_event_on_langfuse(
kwargs={
"litellm_call_id": "call-pm-mock-egress",
"call_type": "completion",
"litellm_params": {"metadata": {"trace_id": "a" * 32}},
"messages": [{"role": "user", "content": "hi"}],
"optional_params": {},
},
response_obj=litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "ok"}}]),
start_time=now,
claim_trace_root=claim_root,
attributes={},
).end(end_time=to_unix_nanos(now))
client.flush()
time.sleep(1)
end_time=now,
)
logger.flush()
finally:
server.shutdown()
langfuse_client_init.cache_clear()
LangfuseResourceManager._instances.pop("pk-pm-mock-egress", None)
assert logged["trace_id"] == "a" * 32
assert received == [], f"LANGFUSE_MOCK still sent spans to the configured host: {received}"
@pytest.mark.asyncio
async def test_async_log_failure_event_records_trace_id_for_alerting(monkeypatch):
from langfuse._client.resource_manager import LangfuseResourceManager
from litellm.integrations.langfuse.langfuse_sdk import resolve_trace_id
from litellm.litellm_core_utils.specialty_caches.service_trace_id_cache import in_memory_trace_id_cache
@ -228,7 +220,6 @@ async def test_async_log_failure_event_records_trace_id_for_alerting(monkeypatch
monkeypatch.setenv("LANGFUSE_HOST", "http://127.0.0.1:1")
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-pm-trace-cache")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-pm-trace-cache")
LangfuseResourceManager._instances.pop("pk-pm-trace-cache", None)
langfuse_client_init.cache_clear()
call_id: Final = "call-trace-cache-1"
now: Final = datetime.now(timezone.utc)
@ -248,7 +239,6 @@ async def test_async_log_failure_event_records_trace_id_for_alerting(monkeypatch
)
finally:
langfuse_client_init.cache_clear()
LangfuseResourceManager._instances.pop("pk-pm-trace-cache", None)
assert in_memory_trace_id_cache.get_cache(litellm_call_id=call_id, service_name="langfuse") == resolve_trace_id(
"alert-trace-1"

1285
tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py
View file

File diff suppressed because it is too large Load diff

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

@ -1,5 +1,7 @@
import datetime
import json
import threading
import time
import types
import unittest
from typing import Final, Optional
@ -10,7 +12,7 @@ import pytest
import litellm
from litellm.integrations.langfuse import langfuse as langfuse_module
from litellm.integrations.langfuse.langfuse import LangFuseLogger
from litellm.integrations.langfuse.langfuse_sdk import _lifecycle_state, resolve_trace_id
from litellm.integrations.langfuse.langfuse_sdk import resolve_trace_id
# Import LangfuseUsageDetails directly from the module where it's defined
@ -122,27 +124,26 @@ class TestLangfuseUsageDetails(unittest.TestCase):
self.env_patcher.stop()
def use_real_langfuse_client(self):
"""Point the logger at a real v4 client whose spans land in memory."""
from langfuse._client.resource_manager import LangfuseResourceManager
from opentelemetry.sdk.trace import TracerProvider
"""Point the logger at an export channel whose spans land in memory."""
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.langfuse.langfuse_sdk import build_langfuse_tracing
self.span_exporter = InMemorySpanExporter()
self.real_provider = TracerProvider()
LangfuseResourceManager._instances.pop("pk-unit-test", None)
self.logger.Langfuse = self.real_langfuse_class(
public_key="pk-unit-test",
secret_key="sk-unit-test",
host="http://127.0.0.1:1",
tracer_provider=self.real_provider,
span_exporter=self.span_exporter,
self.logger.tracing = build_langfuse_tracing(
exporter=self.span_exporter,
environment=None,
release=None,
sample_rate=1.0,
flush_interval_millis=10,
)
return self.logger.Langfuse
self.real_provider = self.logger.tracing.provider
return self.logger.tracing
def exported_generation(self):
self.logger.Langfuse.flush()
self.logger.tracing.flush()
spans = [s for s in self.span_exporter.get_finished_spans()]
assert spans, "no spans were exported"
return spans[-1]
@ -524,7 +525,7 @@ class TestLangfuseUsageDetails(unittest.TestCase):
"""Every attribute this logger exported to langfuse, as one searchable string."""
import json
self.logger.Langfuse.flush()
self.logger.tracing.flush()
return json.dumps(
[dict(span.attributes or {}) for span in self.span_exporter.get_finished_spans()],
default=repr,
@ -554,7 +555,7 @@ class TestLangfuseUsageDetails(unittest.TestCase):
}
def exported_spans_named(self, name):
self.logger.Langfuse.flush()
self.logger.tracing.flush()
return [span for span in self.span_exporter.get_finished_spans() if span.name == name]
def _drive_with_canary(self, extra_metadata=None, hidden_params=None, guardrail_information=None):
@ -645,9 +646,9 @@ class TestLangfuseUsageDetails(unittest.TestCase):
def test_only_the_generation_claims_the_trace_root(self):
"""
Langfuse v4 derives trace name and I/O from every observation marked root, and
the one with the latest start wins. A post-call guardrail starts after the model
call, so it must nest under the generation instead of claiming root itself, or
Langfuse derives trace name and I/O from the root observation, and with several
roots the one with the latest start wins. A post-call guardrail starts after the
model call, so it must nest under the generation instead of being a root itself, or
the trace shows the guardrail's request instead of the model's.
"""
self._drive_with_canary(
@ -667,11 +668,12 @@ class TestLangfuseUsageDetails(unittest.TestCase):
[generation] = [span for span in self.span_exporter.get_finished_spans() if span.name.startswith("litellm-")]
[guardrail] = self.exported_spans_named("guardrail")
[grounding] = self.exported_spans_named("vertex_ai_grounding_metadata")
assert generation.attributes.get("langfuse.internal.as_root") is True
assert generation.parent is None
assert generation.attributes["langfuse.trace.name"] == "canary-trace"
for child in (guardrail, grounding):
assert child.parent.span_id == generation.context.span_id
assert child.context.trace_id == generation.context.trace_id
assert "langfuse.internal.as_root" not in child.attributes
assert "langfuse.trace.name" not in child.attributes
def test_caller_cannot_spoof_an_allowlisted_identity_field(self):
"""
@ -1203,6 +1205,53 @@ async def test_async_log_failure_event_works_without_standard_logging_object():
assert "InternalServerError" in call_kwargs["status_message"]
class _OtlpReceiver:
"""A local HTTP server that records the paths of every POST it gets, standing in for Langfuse."""
def __init__(self) -> None:
from http.server import BaseHTTPRequestHandler, HTTPServer
self.received: list[str] = []
received = self.received
class _Handler(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
self.server = HTTPServer(("127.0.0.1", 0), _Handler)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
@property
def url(self) -> str:
return f"http://127.0.0.1:{self.server.server_port}"
def close(self) -> None:
self.server.shutdown()
def _log_one_completion(logger: LangFuseLogger) -> None:
now = datetime.datetime.now()
logger.log_event_on_langfuse(
kwargs={
"call_type": "completion",
"litellm_params": {"metadata": {}, "proxy_server_request": {"headers": {}}},
"messages": [{"role": "user", "content": "hi"}],
"optional_params": {},
},
response_obj=litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "yo"}}]),
start_time=now,
end_time=now,
)
logger.flush()
def test_mock_mode_makes_no_network_calls(monkeypatch):
"""LANGFUSE_MOCK promises full execution without egress.
@ -1210,55 +1259,21 @@ def test_mock_mode_makes_no_network_calls(monkeypatch):
exporter, so nothing stops a real request to the configured host without an
exporter that drops them.
"""
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from langfuse._client.resource_manager import LangfuseResourceManager
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()
receiver = _OtlpReceiver()
monkeypatch.setenv("LANGFUSE_MOCK", "true")
monkeypatch.setenv("LANGFUSE_HOST", f"http://127.0.0.1:{server.server_port}")
monkeypatch.setenv("LANGFUSE_HOST", receiver.url)
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-mock-egress")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-mock-egress")
LangfuseResourceManager._instances.pop("pk-mock-egress", None)
try:
logger = LangFuseLogger()
assert logger.is_mock_mode is True
now = datetime.datetime.now()
logger.log_event_on_langfuse(
kwargs={
"call_type": "completion",
"litellm_params": {"metadata": {}, "proxy_server_request": {"headers": {}}},
"messages": [{"role": "user", "content": "hi"}],
"optional_params": {},
},
response_obj=litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "yo"}}]),
start_time=now,
end_time=now,
)
logger.Langfuse.flush()
_log_one_completion(logger)
time.sleep(1)
finally:
server.shutdown()
LangfuseResourceManager._instances.pop("pk-mock-egress", None)
receiver.close()
assert received == [], f"mock mode sent real requests: {received}"
assert receiver.received == [], f"mock mode sent real requests: {receiver.received}"
def test_max_langfuse_clients_limit():
@ -1323,7 +1338,7 @@ class _RecordingLangfuse:
def _build_langfuse_logger(monkeypatch) -> LangFuseLogger:
monkeypatch.setenv("LANGFUSE_MOCK", "false")
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
return LangFuseLogger(
langfuse_public_key="pk-lit5228",
langfuse_secret="sk-lit5228",
@ -1335,7 +1350,7 @@ def test_langfuse_environment_is_passed_to_sdk_client(monkeypatch):
monkeypatch.setenv("LANGFUSE_MOCK", "false")
monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
logger = LangFuseLogger(
langfuse_public_key="pk-env",
langfuse_secret="sk-env",
@ -1350,7 +1365,7 @@ def test_langfuse_environment_falls_back_to_deployment_env_var(monkeypatch):
monkeypatch.setenv("LANGFUSE_MOCK", "false")
monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "deployment-wide")
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
logger = LangFuseLogger(
langfuse_public_key="pk-env",
langfuse_secret="sk-env",
@ -1424,76 +1439,47 @@ _LANGFUSE_REDACTED = "redacted-by-litellm"
def _steering_logger():
"""``__new__`` skips the network setup in ``__init__``; spans land in memory."""
from langfuse import Langfuse
from langfuse._client.resource_manager import LangfuseResourceManager
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.langfuse.langfuse import installed_langfuse_version
from litellm.integrations.langfuse.langfuse_sdk import build_langfuse_client, build_langfuse_tracing
exporter = InMemorySpanExporter()
LangfuseResourceManager._instances.pop("pk-steering-test", None)
logger = LangFuseLogger.__new__(LangFuseLogger)
logger.Langfuse = Langfuse(
public_key="pk-steering-test",
secret_key="sk-steering-test",
host="http://127.0.0.1:1",
tracer_provider=TracerProvider(),
span_exporter=exporter,
logger.tracing = build_langfuse_tracing(
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
)
logger.Langfuse = build_langfuse_client(
parameters={
"public_key": "pk-steering-test",
"secret_key": "sk-steering-test",
"base_url": "http://127.0.0.1:1",
"tracing_enabled": False,
},
environment=None,
release=None,
mock_mode=True,
)
logger.langfuse_sdk_version = installed_langfuse_version()
return logger, exporter
def test_log_event_holds_a_client_lease_during_export():
logger, _ = _steering_logger()
state = _lifecycle_state(logger.Langfuse)
def test_log_event_keeps_exporting_after_the_dynamic_cache_evicts_the_logger():
"""Per-key loggers are evicted from ``DynamicLoggingCache`` while a callback may still hold them.
def assert_lease_is_active(**_: object) -> tuple[str, str]:
assert state.active_leases == 1
return "trace-id", "generation-id"
now = datetime.datetime.now()
with patch.object(logger, "_log_langfuse_v2", side_effect=assert_lease_is_active):
returned = logger.log_event_on_langfuse(
kwargs={
"call_type": "completion",
"litellm_params": {"metadata": {}},
"messages": [{"role": "user", "content": "the-input"}],
"optional_params": {},
},
response_obj=litellm.ModelResponse(
choices=[{"message": {"role": "assistant", "content": "the-output"}}]
),
start_time=now,
end_time=now,
)
assert returned == {"trace_id": "trace-id", "generation_id": "generation-id"}
assert state.active_leases == 0
def test_log_event_renews_a_client_the_cache_evicted_before_the_lease():
"""The cache can evict this logger after handing it to the callback and before the lease opens.
v2 lost that callback's events to a shut-down client; the callback must export through a fresh one.
v2 lost that callback's events to a shut-down client; the export channel is shared per
credential set and outlives any one logger, so the events still land.
"""
from litellm.integrations.langfuse.langfuse_sdk import register_langfuse_client, shutdown_langfuse_client
from litellm.litellm_core_utils.specialty_caches.dynamic_logging_cache import LangfuseInMemoryCache
logger, _ = _steering_logger()
evicted = logger.Langfuse
logger.langfuse_client_parameters = {
"public_key": "pk-steering-test",
"secret_key": "sk-steering-test",
"base_url": "http://127.0.0.1:1",
}
logger.langfuse_environment = None
logger.langfuse_release = None
logger.is_mock_mode = True
register_langfuse_client(evicted)
shutdown_langfuse_client(evicted)
logger, exporter = _steering_logger()
cache = LangfuseInMemoryCache()
cache.set_cache("langfuse-evicted", logger)
litellm.initialized_langfuse_clients += 1
before = litellm.initialized_langfuse_clients
cache._remove_key("langfuse-evicted")
now = datetime.datetime.now()
returned = logger.log_event_on_langfuse(
@ -1508,18 +1494,37 @@ def test_log_event_renews_a_client_the_cache_evicted_before_the_lease():
end_time=now,
)
assert returned["trace_id"] is not None
assert logger.Langfuse is not evicted
assert logger.Langfuse._resources is not evicted._resources
assert logger.Langfuse not in _lifecycle_state(logger.Langfuse).retired
shutdown_langfuse_client(logger.Langfuse)
assert litellm.initialized_langfuse_clients == before - 1
assert _span_trace_id(_exported_span(logger, exporter)) == returned["trace_id"]
def _exported_span(logger, exporter):
logger.Langfuse.flush()
logger.flush()
return exporter.get_finished_spans()[-1]
_TRACE_FIELD_KEYS = {
"user.id": "user_id",
"session.id": "session_id",
"langfuse.version": "version",
"langfuse.release": "release",
}
def _trace_params(span):
"""The trace-level fields of the exported span, keyed as v2's ``trace_params`` were."""
prefix = "langfuse.trace."
attributes = span.attributes or {}
return {
**{
key[len(prefix) :]: value
for key, value in attributes.items()
if key.startswith(prefix) and not key.startswith(prefix + "metadata.")
},
**{name: attributes[key] for key, name in _TRACE_FIELD_KEYS.items() if key in attributes},
}
def _span_trace_id(span):
return format(span.context.trace_id, "032x")
@ -1527,49 +1532,35 @@ def _span_trace_id(span):
def _emit(rig, *, metadata=None, headers=None):
"""``log_event_on_langfuse`` is the entry point that folds ``langfuse_*`` headers into metadata.
v4 has no trace object, so the trace-level fields are captured where the
callback hands them to propagation, and the observation fields are read back
off the span langfuse actually exported.
Both the trace-level and the observation fields are read back off the span litellm exported.
"""
from litellm.integrations.langfuse import langfuse as langfuse_module
logger, exporter = rig
exporter.clear()
captured_trace_params = {}
propagate_for_real = langfuse_module._trace_attributes_for_propagation
def capture(trace_params):
captured_trace_params.update(trace_params)
return propagate_for_real(trace_params)
now = datetime.datetime.now()
response_obj = litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "the-output"}}])
with patch.object( # test-quality-ok: v4 has no trace object to read back; the propagation call is the only observable trace-level boundary
langfuse_module, "_trace_attributes_for_propagation", capture
):
logger.log_event_on_langfuse(
kwargs={
"call_type": "completion",
"litellm_params": {
"metadata": dict(metadata or {}),
"proxy_server_request": {"headers": dict(headers or {})},
},
"messages": [{"role": "user", "content": "the-input"}],
"optional_params": {},
logger.log_event_on_langfuse(
kwargs={
"call_type": "completion",
"litellm_params": {
"metadata": dict(metadata or {}),
"proxy_server_request": {"headers": dict(headers or {})},
},
response_obj=response_obj,
start_time=now,
end_time=now,
)
logger.Langfuse.flush()
"messages": [{"role": "user", "content": "the-input"}],
"optional_params": {},
},
response_obj=response_obj,
start_time=now,
end_time=now,
)
prefix = "langfuse.observation."
span = exporter.get_finished_spans()[-1]
span = _exported_span(logger, exporter)
generation_params = {
key[len(prefix) :]: value
for key, value in (span.attributes or {}).items()
if key.startswith(prefix) and not key.startswith(prefix + "metadata.")
}
return captured_trace_params, generation_params, span
return _trace_params(span), generation_params, span
@pytest.mark.parametrize("level", ["DEFAULT", "ERROR"])
@ -1831,7 +1822,7 @@ def test_mask_input_header_false_keeps_the_prompt():
trace_params, generation_params, _ = _emit(rig, headers={"langfuse_mask_input": "false"})
assert trace_params["input"] == {"messages": [{"role": "user", "content": "the-input"}]}
assert "input" not in trace_params
assert json.loads(generation_params["input"]) == {"messages": [{"role": "user", "content": "the-input"}]}
@ -1840,7 +1831,7 @@ def test_mask_input_header_true_redacts_the_prompt():
trace_params, generation_params, _ = _emit(rig, headers={"langfuse_mask_input": "true"})
assert trace_params["input"] == _LANGFUSE_REDACTED
assert "input" not in trace_params
assert generation_params["input"] == _LANGFUSE_REDACTED
@ -1849,8 +1840,8 @@ def test_mask_output_header_false_keeps_the_completion():
trace_params, generation_params, _ = _emit(rig, headers={"langfuse_mask_output": "false"})
assert trace_params["output"] != _LANGFUSE_REDACTED
assert generation_params["output"] != _LANGFUSE_REDACTED
assert "output" not in trace_params
assert "the-output" in generation_params["output"]
def test_mask_output_header_true_redacts_the_completion():
@ -1858,7 +1849,7 @@ def test_mask_output_header_true_redacts_the_completion():
trace_params, generation_params, _ = _emit(rig, headers={"langfuse_mask_output": "true"})
assert trace_params["output"] == _LANGFUSE_REDACTED
assert "output" not in trace_params
assert generation_params["output"] == _LANGFUSE_REDACTED
@ -1874,9 +1865,9 @@ def test_mask_output_header_true_redacts_the_completion():
def test_mask_input_from_the_request_body_is_unchanged(mask_input, expect_redacted):
rig = _steering_logger()
trace_params, _, _ = _emit(rig, metadata={"mask_input": mask_input})
_, generation_params, _ = _emit(rig, metadata={"mask_input": mask_input})
assert (trace_params["input"] == _LANGFUSE_REDACTED) is expect_redacted
assert (generation_params["input"] == _LANGFUSE_REDACTED) is expect_redacted
@pytest.mark.parametrize("flag", [True, "true"])
@ -1895,8 +1886,6 @@ def test_update_trace_keys_header_applies_every_key_when_enabled(flag, monkeypat
)
assert trace_params["release"] == "v1.2.3"
assert trace_params["tail"] == "last"
# v4 models release, so it reaches langfuse; a key it does not model cannot
assert span.attributes["langfuse.release"] == "v1.2.3"
assert not [key for key in span.attributes if key.endswith("tail")]
@ -1921,7 +1910,6 @@ def test_update_trace_keys_is_off_by_default():
assert "user_api_key_auth" not in trace_params
assert "release" not in trace_params
assert "sk-canary" not in json.dumps(trace_params, default=repr)
assert "sk-canary" not in json.dumps(dict(span.attributes or {}), default=repr)
@ -1964,7 +1952,6 @@ def test_existing_trace_id_appends_without_claiming_trace_root():
_, _, span = _emit(rig, metadata={"existing_trace_id": "trace-1", "trace_name": "second-call"})
assert span.parent is not None
assert span.attributes.get("langfuse.internal.as_root") is None
assert "langfuse.trace.name" not in (span.attributes or {})
@ -1973,7 +1960,7 @@ def test_a_fresh_trace_still_claims_root_so_its_generation_names_it():
_, _, span = _emit(rig, metadata={"trace_id": "a" * 32, "trace_name": "first-call"})
assert span.attributes.get("langfuse.internal.as_root") is True
assert span.parent is None
assert span.attributes["langfuse.trace.name"] == "first-call"
@ -2026,11 +2013,22 @@ def test_update_trace_keys_trace_metadata_reaches_the_trace_not_just_the_generat
},
)
assert span.attributes["langfuse.trace.metadata.step"] == "2"
assert span.attributes["langfuse.trace.metadata.note"] == "x" * 200
assert span.attributes["langfuse.trace.metadata.step"] == 2
assert span.attributes["langfuse.trace.metadata.note"] == "x" * 300
assert span.attributes["langfuse.observation.metadata.step"] == 2
def test_non_mapping_trace_metadata_does_not_lose_the_event():
"""A caller who passes ``trace_metadata`` as a string still gets a generation, and the string is not spread."""
rig = _steering_logger()
trace_params, generation_params, span = _emit(rig, metadata={"trace_metadata": "just-a-note"})
assert json.loads(generation_params["output"])["content"] == "the-output"
assert trace_params["name"] == "litellm-completion"
assert not any(key.startswith("langfuse.trace.metadata.") for key in span.attributes or {})
def test_trace_metadata_is_not_propagated_when_absent():
rig = _steering_logger()
@ -2054,7 +2052,7 @@ def test_langfuse_environment_is_coerced_and_validated(monkeypatch):
monkeypatch.setenv("LANGFUSE_MOCK", "false")
monkeypatch.delenv("LANGFUSE_TRACING_ENVIRONMENT", raising=False)
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
logger = LangFuseLogger(
langfuse_public_key="pk-env",
langfuse_secret="sk-env",
@ -2081,7 +2079,7 @@ def test_langfuse_empty_environment_falls_back_and_is_not_dynamic(monkeypatch):
# '' falls back to the deployment env var at init
monkeypatch.setenv("LANGFUSE_MOCK", "false")
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where acquire_langfuse_client resolves it; a real client spawns export threads
with patch("litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuse): # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
logger = LangFuseLogger(
langfuse_public_key="pk-env",
langfuse_secret="sk-env",
@ -2136,9 +2134,8 @@ def test_continued_trace_keeps_the_generation_version():
"""v2 set ``version`` on the generation even when the trace was not being updated."""
rig = _steering_logger()
captured_trace_params, _, span = _emit(rig, metadata={"existing_trace_id": "b" * 32, "version": "gen-7"})
_, _, span = _emit(rig, metadata={"existing_trace_id": "b" * 32, "version": "gen-7"})
assert "version" not in captured_trace_params
assert span.attributes["langfuse.version"] == "gen-7"
@ -2195,41 +2192,38 @@ def test_langfuse_debug_env_string_false_stays_off(monkeypatch):
The v4 client does ``if debug:`` and then mutates root logging via
``logging.basicConfig``, so the unparsed string "false" turns debug ON.
"""
from langfuse._client.resource_manager import LangfuseResourceManager
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-debug-parse-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-debug-parse-test")
monkeypatch.setenv("LANGFUSE_MOCK", "true")
monkeypatch.setenv("LANGFUSE_DEBUG", "false")
monkeypatch.setattr(litellm, "initialized_langfuse_clients", litellm.initialized_langfuse_clients)
logger = LangFuseLogger()
try:
assert logger.langfuse_debug is False
finally:
LangfuseResourceManager._instances.pop("pk-debug-parse-test", None)
assert LangFuseLogger().langfuse_debug is False
def test_explicit_langfuse_host_beats_the_v4_base_url_env(monkeypatch):
"""Per-key/per-team ``langfuse_host`` must win over LANGFUSE_BASE_URL.
v4 resolves ``base_url or $LANGFUSE_BASE_URL or host``, so passing the
resolved host as ``host=`` lets a stray env var silently redirect every
tenant's traces to one server.
v4 resolves ``base_url or $LANGFUSE_BASE_URL or host``, so a stray env var
could silently redirect every tenant's traces to one server. The proof is a
real round trip: the observation lands on the configured host.
"""
from langfuse._client.resource_manager import LangfuseResourceManager
receiver = _OtlpReceiver()
monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-base-url-test")
monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-base-url-test")
monkeypatch.setenv("LANGFUSE_MOCK", "true")
monkeypatch.setenv("LANGFUSE_BASE_URL", "https://elsewhere.example")
monkeypatch.delenv("LANGFUSE_MOCK", raising=False)
monkeypatch.setenv("LANGFUSE_BASE_URL", "http://127.0.0.1:1")
monkeypatch.setenv("LANGFUSE_FLUSH_INTERVAL", "1")
monkeypatch.setattr(litellm, "initialized_langfuse_clients", litellm.initialized_langfuse_clients)
logger = LangFuseLogger(langfuse_host="https://good.example")
try:
assert logger.Langfuse._base_url == "https://good.example"
logger = LangFuseLogger(langfuse_host=receiver.url)
_log_one_completion(logger)
finally:
LangfuseResourceManager._instances.pop("pk-base-url-test", None)
receiver.close()
assert logger.langfuse_host == receiver.url
assert receiver.received == ["/api/public/otel/v1/traces"]
def test_resolve_credentials_falls_back_to_langfuse_base_url(monkeypatch):
@ -2257,30 +2251,25 @@ def test_version_gate_rejects_v5_prereleases():
langfuse_module.raise_if_unsupported_langfuse_version("5.0.0")
def test_int_steering_values_survive_v4_propagation():
"""v4 drops non-string propagated values outright; v2's pydantic coerced them."""
def test_int_steering_values_reach_langfuse_as_strings():
"""Langfuse models user, session and version as strings; v2's pydantic coerced ints for the caller."""
rig = _steering_logger()
_, _, span = _emit(
rig, metadata={"trace_user_id": 12345, "session_id": 67, "trace_version": 3, "tags": ["ok", 99]}
)
_, _, span = _emit(rig, metadata={"trace_user_id": 12345, "session_id": 67, "trace_version": 3})
# the SDK validates AFTER litellm's coercion: a surviving attribute proves the value was a str
assert span.attributes["user.id"] == "12345"
assert span.attributes["session.id"] == "67"
assert span.attributes["langfuse.version"] == "3"
# tags reach propagation as a list; non-str entries must be coerced item-wise
assert langfuse_module._coerce_propagated_value(["ok", 99]) == ["ok", "99"]
def test_long_steering_values_are_capped_not_dropped():
"""The SDK drops any propagated value over 200 characters with only a warning."""
def test_long_steering_values_are_neither_capped_nor_dropped():
"""v2 sent ids of any length; the SDK's 200 character rule belongs to baggage propagation, which litellm no longer uses."""
rig = _steering_logger()
long_user: Final = "u" * 250
_, _, span = _emit(rig, metadata={"trace_user_id": long_user})
assert span.attributes["user.id"] == "u" * 200
assert span.attributes["user.id"] == long_user
def test_returned_generation_id_names_the_exported_observation():
@ -2298,7 +2287,6 @@ def test_returned_generation_id_names_the_exported_observation():
start_time=datetime.datetime.now(),
end_time=datetime.datetime.now(),
)
logger.Langfuse.flush()
span = exporter.get_finished_spans()[-1]
span = _exported_span(logger, exporter)
assert returned["generation_id"] == format(span.context.span_id, "016x")

37
tests/test_litellm/litellm_core_utils/specialty_caches/test_dynamic_logging_cache.py
View file

@ -50,29 +50,22 @@ class TestLangfuseInMemoryCache:
assert litellm.initialized_langfuse_clients == initial_count - 1
@patch("litellm.initialized_langfuse_clients", 3)
def test_langfuse_client_shutdown_called_on_eviction(self):
"""Test that langfuse client shutdown is called to close the thread."""
def test_evicted_logger_keeps_its_client_alive(self):
"""The SDK keeps one resource bundle per public key, shared by every logger on that key.
# Create a mock LangFuseLogger class
class MockLangFuseLogger:
def __init__(self):
self.Langfuse = MagicMock()
self.Langfuse.flush = MagicMock()
self.Langfuse.shutdown = MagicMock()
Shutting it down on eviction would stop prompt fetches and the exit flush for the
loggers still using it, so eviction only releases the initialized-client slot.
"""
from litellm.integrations.langfuse.langfuse import LangFuseLogger
mock_logger = MockLangFuseLogger()
logger = LangFuseLogger.__new__(LangFuseLogger)
logger.Langfuse = MagicMock()
logger.Langfuse.get_prompt.return_value = "prompt-after-eviction"
self.cache.cache_dict["test_key"] = logger
self.cache.ttl_dict["test_key"] = time.time() + 100
# Patch the LangFuseLogger import to return our mock class
with patch(
"litellm.integrations.langfuse.langfuse.LangFuseLogger", MockLangFuseLogger
):
# Add the mock logger to cache
self.cache.cache_dict["test_key"] = mock_logger
self.cache.ttl_dict["test_key"] = time.time() + 100
self.cache._remove_key("test_key")
# Remove the key (this should trigger cleanup)
self.cache._remove_key("test_key")
# Verify flush and shutdown were called
mock_logger.Langfuse.flush.assert_called_once()
mock_logger.Langfuse.shutdown.assert_called_once()
assert litellm.initialized_langfuse_clients == 2
assert logger.Langfuse.get_prompt("greeting") == "prompt-after-eviction"
logger.Langfuse.shutdown.assert_not_called()