mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
fix(langfuse): harden client lifecycle and v4 value coercion per review
- hold the SDK registry lock across evict+construct; reuse the provider for a cached key (atexit pin leak); tear down only litellm-built providers; unpublish the registry entry before the blocking shutdown - rebuild litellm's TLS material (CA bundle / client cert) onto the OTLP exporter, which no longer flows through the injected httpx client - coerce and cap propagated trace attributes (v4 drops non-str and >200 chars) - claim trace root on guardrail/grounding sibling spans like the SDK does - return the exported observation id as generation_id - reject langfuse v5 prereleases in the version gate - pin langfuse==4.15.1 in Dockerfile.build_from_pip; drop dead version sniffs; warn once on ignored UPSTREAM_LANGFUSE_* vars
This commit is contained in:
Yucheng Zhu
parent
38b6b72fde
commit
64ed88b9b6
7 changed files with 350 additions and 122 deletions
|
|
@ -32,7 +32,7 @@ RUN uv venv --python python && \
|
|||
"anthropic[vertex]==0.84.0" \
|
||||
"grpcio==1.78.0" \
|
||||
"prometheus-client==0.20.0" \
|
||||
"langfuse>=4.7,<5.0" \
|
||||
"langfuse==4.15.1" \
|
||||
"opentelemetry-api==1.33.1" \
|
||||
"opentelemetry-sdk==1.33.1" \
|
||||
"opentelemetry-exporter-otlp==1.33.1" \
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
# On success, logs events to Langfuse
|
||||
import os
|
||||
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
|
||||
|
|
@ -141,7 +141,9 @@ def raise_if_unsupported_langfuse_version(installed_version: str) -> None:
|
|||
`propagate_attributes` raises inside the per-request handler and the broad
|
||||
except there turns it into silent total data loss.
|
||||
"""
|
||||
if Version(MINIMUM_LANGFUSE_VERSION) <= Version(installed_version) < Version(UNSUPPORTED_LANGFUSE_VERSION):
|
||||
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 "
|
||||
|
|
@ -159,17 +161,37 @@ _GENERATION_ONLY_KEYS: Final = frozenset(
|
|||
)
|
||||
|
||||
|
||||
_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."""
|
||||
"""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: trace_params[key]
|
||||
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:
|
||||
|
|
@ -290,8 +312,8 @@ class LangFuseLogger:
|
|||
"debug": self.langfuse_debug,
|
||||
"flush_interval": self.langfuse_flush_interval, # flush interval in seconds
|
||||
"httpx_client": self.langfuse_client,
|
||||
"environment": self.langfuse_environment,
|
||||
}
|
||||
parameters["environment"] = self.langfuse_environment
|
||||
self.Langfuse: Langfuse = self.safe_init_langfuse_client(parameters)
|
||||
|
||||
# set the current langfuse project id in the environ
|
||||
|
|
@ -307,6 +329,10 @@ class LangFuseLogger:
|
|||
verbose_logger.debug("Langfuse project id unavailable, alerting links will omit it")
|
||||
|
||||
if os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY") is not 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")
|
||||
|
|
@ -321,33 +347,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}"
|
||||
)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import (
|
||||
DiscardingSpanExporter,
|
||||
build_isolated_tracer_provider,
|
||||
evict_stale_langfuse_resources,
|
||||
register_langfuse_client,
|
||||
)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client
|
||||
|
||||
evict_stale_langfuse_resources(
|
||||
public_key=parameters.get("public_key"),
|
||||
secret_key=parameters.get("secret_key"),
|
||||
base_url=parameters.get("base_url"),
|
||||
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,
|
||||
)
|
||||
langfuse_client: Final = Langfuse(
|
||||
**parameters,
|
||||
tracer_provider=build_isolated_tracer_provider(
|
||||
environment=parameters.get("environment"),
|
||||
release=parameters.get("release"),
|
||||
),
|
||||
span_exporter=DiscardingSpanExporter() if self.is_mock_mode else None,
|
||||
)
|
||||
register_langfuse_client(langfuse_client)
|
||||
litellm.initialized_langfuse_clients += 1
|
||||
verbose_logger.debug("Created langfuse client number %s", litellm.initialized_langfuse_clients)
|
||||
return langfuse_client
|
||||
|
|
@ -446,8 +459,6 @@ class LangFuseLogger:
|
|||
status_message=status_message,
|
||||
)
|
||||
verbose_logger.debug("OUTPUT IN LANGFUSE: %s; original: %s", output, response_obj)
|
||||
trace_id = None
|
||||
generation_id = None
|
||||
trace_id, generation_id = self._log_langfuse_v2(
|
||||
user_id=user_id,
|
||||
metadata=metadata,
|
||||
|
|
@ -579,11 +590,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
|
||||
|
|
@ -739,17 +746,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:
|
||||
|
|
@ -837,20 +843,19 @@ class LangFuseLogger:
|
|||
**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
|
||||
|
||||
|
|
@ -876,14 +881,18 @@ class LangFuseLogger:
|
|||
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
|
||||
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,
|
||||
)
|
||||
start_generation(
|
||||
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
|
||||
|
|
@ -892,10 +901,13 @@ class LangFuseLogger:
|
|||
release=trace_params.get("release"),
|
||||
public=_trace_public_flag(trace_params.get("public")),
|
||||
attributes=_generation_attributes(generation_params, propagated=propagated_trace_attributes),
|
||||
).end(end_time=to_unix_nanos(end_time))
|
||||
)
|
||||
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
|
||||
return resolved_trace_id, generation_id
|
||||
# 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
|
||||
|
|
@ -969,14 +981,6 @@ class LangFuseLogger:
|
|||
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")
|
||||
|
||||
@staticmethod
|
||||
def _apply_masking_function(data: object, masking_function: Callable[[object], object]) -> object:
|
||||
"""
|
||||
|
|
@ -1044,6 +1048,7 @@ class LangFuseLogger:
|
|||
client: "Langfuse",
|
||||
context: "Context",
|
||||
standard_logging_object: StandardLoggingPayload | None,
|
||||
claim_trace_root: bool,
|
||||
):
|
||||
"""
|
||||
Log guardrail information as a span
|
||||
|
|
@ -1079,6 +1084,7 @@ class LangFuseLogger:
|
|||
context=context,
|
||||
name="guardrail",
|
||||
start_time=guardrail_entry.get("start_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),
|
||||
|
|
@ -1172,6 +1178,7 @@ def log_provider_specific_information_as_span(
|
|||
client: "Langfuse",
|
||||
context: "Context",
|
||||
enrichments: Mapping[str, Any],
|
||||
claim_trace_root: bool,
|
||||
):
|
||||
"""
|
||||
Logs provider-specific information as spans.
|
||||
|
|
@ -1195,19 +1202,30 @@ 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, context=context, name=key, value=value)
|
||||
_end_grounding_span(
|
||||
client=client, context=context, name=key, value=value, claim_trace_root=claim_trace_root
|
||||
)
|
||||
else:
|
||||
_end_grounding_span(client=client, context=context, name="vertex_ai_grounding_metadata", value=elem)
|
||||
_end_grounding_span(
|
||||
client=client,
|
||||
context=context,
|
||||
name="vertex_ai_grounding_metadata",
|
||||
value=elem,
|
||||
claim_trace_root=claim_trace_root,
|
||||
)
|
||||
else:
|
||||
_end_grounding_span(
|
||||
client=client,
|
||||
context=context,
|
||||
name="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) -> None:
|
||||
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(
|
||||
|
|
@ -1215,6 +1233,7 @@ def _end_grounding_span(*, client: "Langfuse", context: "Context", name: str, va
|
|||
context=context,
|
||||
name=name,
|
||||
start_time=None,
|
||||
claim_trace_root=claim_trace_root,
|
||||
attributes={"input": value}, # mutable-ok: langfuse serializes this payload
|
||||
).end()
|
||||
|
||||
|
|
|
|||
|
|
@ -68,7 +68,9 @@ def langfuse_client_init(
|
|||
Exception: If langfuse package is not installed
|
||||
"""
|
||||
try:
|
||||
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"
|
||||
|
|
@ -118,20 +120,14 @@ def langfuse_client_init(
|
|||
|
||||
parameters["environment"] = LangFuseLogger.resolve_deployment_environment()
|
||||
|
||||
from .langfuse_sdk import (
|
||||
DiscardingSpanExporter,
|
||||
build_isolated_tracer_provider,
|
||||
evict_stale_langfuse_resources,
|
||||
register_langfuse_client,
|
||||
)
|
||||
from .langfuse_sdk import acquire_langfuse_client
|
||||
|
||||
evict_stale_langfuse_resources(public_key=public_key, secret_key=secret_key, base_url=langfuse_host)
|
||||
client: Final = Langfuse(
|
||||
**parameters, # pyright: ignore[reportArgumentType] # kwargs-ok: dict mirrors the typed ctor, values resolved above
|
||||
tracer_provider=build_isolated_tracer_provider(environment=parameters["environment"], release=langfuse_release),
|
||||
span_exporter=DiscardingSpanExporter() if is_mock_mode else None,
|
||||
client: Final = acquire_langfuse_client(
|
||||
parameters=parameters,
|
||||
environment=parameters["environment"],
|
||||
release=langfuse_release,
|
||||
mock_mode=is_mock_mode,
|
||||
)
|
||||
register_langfuse_client(client)
|
||||
|
||||
return client
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
from base64 import b64encode
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime
|
||||
from hashlib import sha256
|
||||
|
|
@ -22,6 +24,7 @@ __all__ = (
|
|||
"PUBLIC_ATTRIBUTE",
|
||||
"RELEASE_ATTRIBUTE",
|
||||
"DiscardingSpanExporter",
|
||||
"acquire_langfuse_client",
|
||||
"build_isolated_tracer_provider",
|
||||
"evict_stale_langfuse_resources",
|
||||
"open_trace_context",
|
||||
|
|
@ -135,17 +138,28 @@ def start_child_span(
|
|||
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."""
|
||||
"""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"
|
||||
_RELEASE_ATTRIBUTE: Final = "langfuse.release"
|
||||
|
||||
# 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:
|
||||
|
|
@ -162,11 +176,13 @@ def build_isolated_tracer_provider(*, environment: str | None, release: str | No
|
|||
attributes: Final = MappingProxyType(
|
||||
{
|
||||
key: value
|
||||
for key, value in ((_ENVIRONMENT_ATTRIBUTE, environment), (_RELEASE_ATTRIBUTE, release))
|
||||
for key, value in ((_ENVIRONMENT_ATTRIBUTE, environment), (RELEASE_ATTRIBUTE, release))
|
||||
if value is not None
|
||||
}
|
||||
)
|
||||
return TracerProvider(resource=Resource.create(dict(attributes)))
|
||||
provider: Final = TracerProvider(resource=Resource.create(dict(attributes)))
|
||||
_litellm_built_providers.add(provider)
|
||||
return provider
|
||||
|
||||
|
||||
class DiscardingSpanExporter(SpanExporter):
|
||||
|
|
@ -187,23 +203,122 @@ class DiscardingSpanExporter(SpanExporter):
|
|||
return True
|
||||
|
||||
|
||||
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.
|
||||
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 cached bundle, if any.
|
||||
|
||||
langfuse keys its client registry on the public key alone, so a rotated
|
||||
secret or a moved host silently keeps exporting with the original values.
|
||||
Only the one stale entry is removed; the SDK's own reset would shut down
|
||||
every other tenant in the process.
|
||||
every other tenant in the process. A stale bundle no live litellm client
|
||||
still holds also has its export thread stopped — but only when litellm
|
||||
built the provider, because a bundle adopted from user code may share the
|
||||
process-global provider.
|
||||
"""
|
||||
if not public_key:
|
||||
return
|
||||
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
|
||||
with _LIVE_CLIENTS_LOCK:
|
||||
holders: Final = _live_clients.get(cached)
|
||||
abandoned: Final = holders is None or len(holders) == 0
|
||||
provider: Final = getattr(cached, "tracer_provider", None)
|
||||
if abandoned and provider is not None and provider in _litellm_built_providers:
|
||||
provider.shutdown()
|
||||
return None
|
||||
|
||||
|
||||
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
|
||||
cached: Final = LangfuseResourceManager._instances.get(public_key) # pyright: ignore[reportPrivateUsage] # registry has no public accessor
|
||||
if cached is None:
|
||||
return
|
||||
if getattr(cached, "secret_key", None) == secret_key and getattr(cached, "base_url", None) == base_url:
|
||||
return
|
||||
LangfuseResourceManager._instances.pop(public_key, None) # pyright: ignore[reportPrivateUsage] # registry has no public accessor
|
||||
_evict_if_stale_locked(public_key=public_key, secret_key=secret_key, base_url=base_url)
|
||||
|
||||
|
||||
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)
|
||||
return client
|
||||
|
||||
|
||||
_LIVE_CLIENTS_LOCK: Final = threading.Lock()
|
||||
|
|
@ -256,21 +371,24 @@ def shutdown_langfuse_client(client: Langfuse) -> None:
|
|||
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
|
||||
if not _release_langfuse_resources(resources, client):
|
||||
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 not isinstance(provider, otel_trace.ProxyTracerProvider):
|
||||
if provider is not None and provider in _litellm_built_providers:
|
||||
provider.shutdown()
|
||||
public_key: Final = getattr(resources, "public_key", None)
|
||||
if public_key is None:
|
||||
return
|
||||
with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor
|
||||
if 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
|
||||
|
|
|
|||
|
|
@ -91,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",
|
||||
|
|
@ -128,15 +127,12 @@ 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()
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ from litellm.integrations.langfuse.langfuse import (
|
|||
)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import (
|
||||
AS_ROOT_ATTRIBUTE,
|
||||
_litellm_built_providers,
|
||||
PUBLIC_ATTRIBUTE,
|
||||
RELEASE_ATTRIBUTE,
|
||||
build_isolated_tracer_provider,
|
||||
|
|
@ -100,9 +101,9 @@ def test_guardrail_span_with_float_timestamps_does_not_break_the_generation(clie
|
|||
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, attributes={}).end(
|
||||
end_time=to_unix_nanos(guardrail_start + 2)
|
||||
)
|
||||
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))
|
||||
|
|
@ -141,9 +142,9 @@ 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, attributes={}).end(
|
||||
end_time=to_unix_nanos(guardrail_start + timedelta(seconds=2))
|
||||
)
|
||||
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))
|
||||
|
|
@ -152,8 +153,10 @@ def test_child_span_keeps_its_own_window_and_stays_a_sibling(client):
|
|||
guardrail = _only_span(exporter, "guardrail")
|
||||
generation = _only_span(exporter, "gen")
|
||||
assert (guardrail.end_time - guardrail.start_time) == 2 * 1_000_000_000
|
||||
assert guardrail.parent.span_id == generation.parent.span_id
|
||||
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):
|
||||
|
|
@ -423,6 +426,7 @@ def _shared_resources_pair():
|
|||
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",
|
||||
|
|
@ -502,3 +506,43 @@ def test_shutdown_of_a_stale_client_does_not_deregister_the_live_one():
|
|||
|
||||
assert stale_resources is not live._resources
|
||||
assert LangfuseResourceManager._instances.get(PUBLIC_KEY) is live._resources
|
||||
|
||||
|
||||
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."""
|
||||
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_client
|
||||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -309,7 +309,6 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
side_effect=lambda generation_params, **kwargs: generation_params,
|
||||
create=True,
|
||||
) as mock_add_prompt_params,
|
||||
patch.object(self.logger, "_supports_prompt", return_value=True),
|
||||
):
|
||||
# Create a mock response object with usage information containing None values
|
||||
response_obj = MagicMock()
|
||||
|
|
@ -980,7 +979,7 @@ def test_failure_handler_langfuse_kwargs_excludes_original_response():
|
|||
|
||||
try:
|
||||
# Mock LangFuseHandler to return our capturing mock logger
|
||||
with patch("litellm.litellm_core_utils.litellm_logging.LangFuseHandler") as mock_handler_class:
|
||||
with patch("litellm.litellm_core_utils.litellm_logging.LangFuseHandler") as mock_handler_class: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients
|
||||
mock_handler_class.get_langfuse_logger_for_request.return_value = mock_langfuse_logger
|
||||
|
||||
# Call the actual failure_handler
|
||||
|
|
@ -1047,7 +1046,7 @@ async def test_async_log_failure_event_logs_to_langfuse():
|
|||
"generation_id": "mock-gen",
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler:
|
||||
with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients
|
||||
mock_handler.get_langfuse_logger_for_request.return_value = mock_logger
|
||||
|
||||
kwargs = {
|
||||
|
|
@ -1112,7 +1111,7 @@ async def test_async_log_failure_event_works_without_standard_logging_object():
|
|||
"generation_id": "mock-gen",
|
||||
}
|
||||
|
||||
with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler:
|
||||
with patch("litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler") as mock_handler: # test-quality-ok: route the request to the capturing logger; the real handler builds live clients
|
||||
mock_handler.get_langfuse_logger_for_request.return_value = mock_logger
|
||||
|
||||
kwargs = {
|
||||
|
|
@ -1263,7 +1262,7 @@ class _RecordingLangfuse:
|
|||
def _build_langfuse_logger(monkeypatch) -> LangFuseLogger:
|
||||
monkeypatch.setenv("LANGFUSE_MOCK", "false")
|
||||
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
|
||||
with patch("langfuse.Langfuse", _RecordingLangfuse):
|
||||
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
|
||||
return LangFuseLogger(
|
||||
langfuse_public_key="pk-lit5228",
|
||||
langfuse_secret="sk-lit5228",
|
||||
|
|
@ -1275,7 +1274,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("langfuse.Langfuse", _RecordingLangfuse):
|
||||
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
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
|
|
@ -1290,7 +1289,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("langfuse.Langfuse", _RecordingLangfuse):
|
||||
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
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
|
|
@ -1577,7 +1576,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("langfuse.Langfuse", _RecordingLangfuse):
|
||||
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
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
|
|
@ -1604,7 +1603,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("langfuse.Langfuse", _RecordingLangfuse):
|
||||
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
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
|
|
@ -1769,3 +1768,59 @@ def test_resolve_credentials_falls_back_to_langfuse_base_url(monkeypatch):
|
|||
|
||||
_, _, host = langfuse_module.resolve_langfuse_credentials(langfuse_host="https://explicit.example")
|
||||
assert host == "https://explicit.example"
|
||||
|
||||
|
||||
def test_version_gate_rejects_v5_prereleases():
|
||||
""""5.0.0rc1" sorts below "5", so a plain version comparison would admit it."""
|
||||
langfuse_module.raise_if_unsupported_langfuse_version("4.7")
|
||||
with pytest.raises(ImportError):
|
||||
langfuse_module.raise_if_unsupported_langfuse_version("5.0.0rc1")
|
||||
with pytest.raises(ImportError):
|
||||
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."""
|
||||
rig = _steering_logger()
|
||||
|
||||
_, _, span = _emit(
|
||||
rig, metadata={"trace_user_id": 12345, "session_id": 67, "trace_version": 3, "tags": ["ok", 99]}
|
||||
)
|
||||
|
||||
# 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."""
|
||||
rig = _steering_logger()
|
||||
long_user: Final = "u" * 250
|
||||
|
||||
_, _, span = _emit(rig, metadata={"trace_user_id": long_user})
|
||||
|
||||
assert span.attributes["user.id"] == "u" * 200
|
||||
|
||||
|
||||
def test_returned_generation_id_names_the_exported_observation():
|
||||
"""v4 derives observation ids from the OTel span, so a pre-computed id would name nothing."""
|
||||
logger, exporter = _steering_logger()
|
||||
|
||||
returned = logger.log_event_on_langfuse(
|
||||
kwargs={
|
||||
"call_type": "completion",
|
||||
"litellm_params": {"metadata": {"trace_id": "d" * 32}},
|
||||
"messages": [{"role": "user", "content": "the-input"}],
|
||||
"optional_params": {},
|
||||
},
|
||||
response_obj=litellm.ModelResponse(choices=[{"message": {"role": "assistant", "content": "the-output"}}]),
|
||||
start_time=datetime.datetime.now(),
|
||||
end_time=datetime.datetime.now(),
|
||||
)
|
||||
logger.Langfuse.flush()
|
||||
|
||||
span = exporter.get_finished_spans()[-1]
|
||||
assert returned["generation_id"] == format(span.context.span_id, "016x")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue