mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-26 01:12:21 +00:00
fix(langfuse): own the tracer config and drop the SDK client for prompts and auth
The callback's TracerProvider now sets its sampler, span limits and id generator explicitly so unrelated OTEL_* variables no longer change what Langfuse receives, and trace metadata is written once on the trace instead of folded into the generation, which kept input and output under the attribute cap. Spans are emitted under the langfuse-sdk scope so Langfuse renders them natively, the batch processor queues 100k spans and honors LANGFUSE_FLUSH_AT, and the proxy shutdown flush runs off the event loop with a 10s deadline and logs a miss. Prompts, auth_check and the project id now go through LangfuseAPI directly with a litellm-owned TTL cache, so no Langfuse() client is built and a host application's client on the same public key is left alone. Dead attributes, the unreachable exporter branch and the export list are cleaned up, and the client-budget eviction behavior is documented. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng
parent
6093893052
commit
bbef26a90b
15 changed files with 571 additions and 456 deletions
|
|
@ -573,6 +573,7 @@ FIREWORKS_AI_80_B: Final = int(os.getenv("FIREWORKS_AI_80_B", 80))
|
|||
#### Logging callback constants ####
|
||||
REDACTED_BY_LITELM_STRING: Final = "REDACTED_BY_LITELM"
|
||||
MAX_LANGFUSE_INITIALIZED_CLIENTS: Final = int(os.getenv("MAX_LANGFUSE_INITIALIZED_CLIENTS", 50))
|
||||
LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS: Final = 10_000
|
||||
# Backpressure + lifetime bounds for the /v1/messages streaming relay (see
|
||||
# BaseAnthropicMessagesStreamingIterator.async_sse_wrapper). The relay queue is
|
||||
# bounded so a slow client throttles the upstream pump instead of letting it
|
||||
|
|
|
|||
|
|
@ -45,13 +45,11 @@ from litellm.types.utils import (
|
|||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from langfuse import Langfuse
|
||||
|
||||
from litellm.integrations.langfuse.langfuse_sdk import LangfuseObservation, LangfuseTracing
|
||||
from litellm.integrations.langfuse.langfuse_sdk import LangfuseApiClient, LangfuseObservation, LangfuseTracing
|
||||
from litellm.litellm_core_utils.litellm_logging import DynamicLoggingCache
|
||||
else:
|
||||
DynamicLoggingCache = Any
|
||||
Langfuse = Any
|
||||
LangfuseApiClient = Any
|
||||
LangfuseObservation = Any
|
||||
LangfuseTracing = Any
|
||||
|
||||
|
|
@ -280,7 +278,7 @@ class LangFuseLogger:
|
|||
allow_env_credentials: bool = True,
|
||||
):
|
||||
try:
|
||||
from langfuse import Langfuse
|
||||
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_tracing
|
||||
except Exception as e:
|
||||
raise Exception(
|
||||
f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\n{traceback.format_exc()}\033[0m"
|
||||
|
|
@ -312,19 +310,7 @@ class LangFuseLogger:
|
|||
self.langfuse_client = self._http_handler.client
|
||||
self.is_mock_mode = False
|
||||
|
||||
self.langfuse_client_parameters: Final[dict[str, object]] = {
|
||||
"public_key": self.public_key,
|
||||
"secret_key": self.secret_key,
|
||||
"base_url": self.langfuse_host,
|
||||
"release": self.langfuse_release,
|
||||
"debug": self.langfuse_debug,
|
||||
"flush_interval": self.langfuse_flush_interval, # flush interval in seconds
|
||||
"httpx_client": self.langfuse_client,
|
||||
"environment": self.langfuse_environment,
|
||||
}
|
||||
self.Langfuse: Langfuse = self.safe_init_langfuse_client(self.langfuse_client_parameters)
|
||||
from litellm.integrations.langfuse.langfuse_sdk import acquire_langfuse_tracing
|
||||
|
||||
self.api_client: LangfuseApiClient = self.safe_init_langfuse_client()
|
||||
self.tracing: LangfuseTracing = acquire_langfuse_tracing(
|
||||
public_key=str(self.public_key),
|
||||
secret_key=str(self.secret_key),
|
||||
|
|
@ -342,26 +328,19 @@ class LangFuseLogger:
|
|||
verbose_logger.debug("Langfuse Mock: Using mock project ID")
|
||||
else:
|
||||
try:
|
||||
project_id: Final = self.Langfuse.api.projects.get().data[0].id
|
||||
os.environ["LANGFUSE_PROJECT_ID"] = project_id
|
||||
project_id: Final = self.api_client.project_id()
|
||||
if project_id is not None:
|
||||
os.environ["LANGFUSE_PROJECT_ID"] = project_id
|
||||
except Exception:
|
||||
verbose_logger.debug("Langfuse project id unavailable, alerting links will omit it")
|
||||
|
||||
warn_if_upstream_langfuse_configured()
|
||||
if os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY") is not None:
|
||||
self.upstream_langfuse_secret_key = os.getenv("UPSTREAM_LANGFUSE_SECRET_KEY")
|
||||
self.upstream_langfuse_public_key = os.getenv("UPSTREAM_LANGFUSE_PUBLIC_KEY")
|
||||
self.upstream_langfuse_host = os.getenv("UPSTREAM_LANGFUSE_HOST")
|
||||
self.upstream_langfuse_release = os.getenv("UPSTREAM_LANGFUSE_RELEASE")
|
||||
self.upstream_langfuse_debug = os.getenv("UPSTREAM_LANGFUSE_DEBUG")
|
||||
|
||||
def safe_init_langfuse_client(self, parameters: dict) -> Langfuse:
|
||||
"""
|
||||
Safely init a langfuse client if the number of initialized clients is less than the max
|
||||
def safe_init_langfuse_client(self) -> LangfuseApiClient:
|
||||
"""Build the REST client while the process is under its logger budget.
|
||||
|
||||
Note:
|
||||
- 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.
|
||||
The budget dates from the SDK client, which started a consumer thread per instance and once
|
||||
pinned a CPU at 100% when many were built; it still bounds the number of per-key loggers.
|
||||
"""
|
||||
if litellm.initialized_langfuse_clients >= MAX_LANGFUSE_INITIALIZED_CLIENTS:
|
||||
raise Exception(
|
||||
|
|
@ -369,22 +348,19 @@ class LangFuseLogger:
|
|||
)
|
||||
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 = build_langfuse_client(
|
||||
parameters=parameters,
|
||||
environment=environment_param,
|
||||
release=release_param,
|
||||
mock_mode=self.is_mock_mode,
|
||||
api_client: Final = build_langfuse_client(
|
||||
public_key=self.public_key,
|
||||
secret_key=self.secret_key,
|
||||
base_url=self.langfuse_host,
|
||||
httpx_client=self.langfuse_client,
|
||||
)
|
||||
litellm.initialized_langfuse_clients += 1
|
||||
verbose_logger.debug("Created langfuse client number %s", litellm.initialized_langfuse_clients)
|
||||
return langfuse_client
|
||||
return api_client
|
||||
|
||||
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]:
|
||||
|
|
@ -859,36 +835,25 @@ class LangFuseLogger:
|
|||
generation_params = {
|
||||
"name": generation_name,
|
||||
"id": clean_metadata.pop("generation_id", generation_id),
|
||||
"start_time": start_time,
|
||||
"end_time": end_time,
|
||||
"model": model_name,
|
||||
"model_parameters": optional_params,
|
||||
"input": masked_input if not mask_input else "redacted-by-litellm",
|
||||
"output": masked_output if not mask_output else "redacted-by-litellm",
|
||||
"usage": usage,
|
||||
"usage_details": usage_details,
|
||||
"cost_details": {"total": cost} # mutable-ok: langfuse serializes this payload
|
||||
if usage is not None and isinstance(cost, (int, float))
|
||||
else None,
|
||||
"metadata": { # mutable-ok: langfuse serializes this payload, a proxy is not json-encodable
|
||||
**(_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),
|
||||
},
|
||||
"level": level,
|
||||
"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
|
||||
|
||||
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,
|
||||
langfuse_client=self.api_client,
|
||||
)
|
||||
if masked_output is not None and isinstance(masked_output, str) and level == "ERROR":
|
||||
generation_params["status_message"] = masked_output
|
||||
|
|
@ -1151,9 +1116,8 @@ def _add_prompt_to_generation_params(
|
|||
generation_params: dict,
|
||||
clean_metadata: dict,
|
||||
prompt_management_metadata: StandardLoggingPromptManagementMetadata | None,
|
||||
langfuse_client: object,
|
||||
langfuse_client: "LangfuseApiClient",
|
||||
) -> dict:
|
||||
from langfuse import Langfuse
|
||||
from langfuse.model import (
|
||||
ChatPromptClient,
|
||||
Prompt_Chat,
|
||||
|
|
@ -1161,8 +1125,6 @@ def _add_prompt_to_generation_params(
|
|||
TextPromptClient,
|
||||
)
|
||||
|
||||
langfuse_client = cast(Langfuse, langfuse_client)
|
||||
|
||||
user_prompt: Final = clean_metadata.pop("prompt", None)
|
||||
if user_prompt is None and prompt_management_metadata is None:
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@ from ..prompt_management_base import PromptManagementBase
|
|||
from .langfuse import (
|
||||
LangFuseLogger,
|
||||
installed_langfuse_version,
|
||||
parse_langfuse_debug,
|
||||
raise_if_unsupported_langfuse_version,
|
||||
resolve_langfuse_credentials,
|
||||
warn_if_upstream_langfuse_configured,
|
||||
|
|
@ -31,12 +30,13 @@ from .langfuse_handler import LangFuseHandler
|
|||
from .langfuse_mock_client import create_mock_langfuse_client, should_use_langfuse_mock
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from langfuse import Langfuse
|
||||
from langfuse.model import ChatPromptClient, TextPromptClient
|
||||
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
|
||||
LangfuseClass: TypeAlias = Langfuse
|
||||
from .langfuse_sdk import LangfuseApiClient
|
||||
|
||||
LangfuseClass: TypeAlias = LangfuseApiClient
|
||||
|
||||
PROMPT_CLIENT = TextPromptClient | ChatPromptClient
|
||||
else:
|
||||
|
|
@ -56,24 +56,22 @@ def langfuse_client_init(
|
|||
allow_env_credentials: bool = True,
|
||||
) -> LangfuseClass:
|
||||
"""
|
||||
Initialize Langfuse client with caching to prevent multiple initializations.
|
||||
Initialize the Langfuse REST client with caching to prevent multiple initializations.
|
||||
|
||||
Args:
|
||||
langfuse_public_key (str, optional): Public key for Langfuse. Defaults to None.
|
||||
langfuse_secret (str, optional): Secret key for Langfuse. Defaults to None.
|
||||
langfuse_host (str, optional): Host URL for Langfuse. Defaults to None.
|
||||
flush_interval (int, optional): Flush interval in seconds. Defaults to 1.
|
||||
flush_interval (int, optional): Kept in the signature so cached callers keep their cache key.
|
||||
|
||||
Returns:
|
||||
Langfuse: Initialized Langfuse client instance
|
||||
LangfuseApiClient: prompt, auth and project lookups for one credential set
|
||||
|
||||
Raises:
|
||||
Exception: If langfuse package is not installed
|
||||
"""
|
||||
try:
|
||||
from langfuse import (
|
||||
Langfuse, # noqa: F401 # the import is the install probe; construction happens in build_langfuse_client
|
||||
)
|
||||
from .langfuse_sdk import build_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"
|
||||
|
|
@ -91,18 +89,6 @@ def langfuse_client_init(
|
|||
# add http:// if unset, assume communicating over private network - e.g. render
|
||||
langfuse_host = "http://" + langfuse_host
|
||||
|
||||
langfuse_release: Final = os.getenv("LANGFUSE_RELEASE")
|
||||
langfuse_debug: Final = parse_langfuse_debug(os.getenv("LANGFUSE_DEBUG"))
|
||||
|
||||
parameters: Final = {
|
||||
"public_key": public_key,
|
||||
"secret_key": secret_key,
|
||||
"base_url": langfuse_host,
|
||||
"release": langfuse_release,
|
||||
"debug": langfuse_debug,
|
||||
"flush_interval": LangFuseLogger._get_langfuse_flush_interval(flush_interval), # pyright: ignore[reportPrivateUsage] # shared env-fallback helper, not part of the logger's API
|
||||
}
|
||||
|
||||
raise_if_unsupported_langfuse_version(installed_langfuse_version())
|
||||
warn_if_upstream_langfuse_configured()
|
||||
|
||||
|
|
@ -112,29 +98,21 @@ def langfuse_client_init(
|
|||
|
||||
from ...llms.custom_httpx.http_handler import get_ssl_configuration
|
||||
|
||||
is_mock_mode: Final = should_use_langfuse_mock()
|
||||
parameters["httpx_client"] = (
|
||||
httpx_client: Final = (
|
||||
create_mock_langfuse_client()
|
||||
if is_mock_mode
|
||||
if should_use_langfuse_mock()
|
||||
else httpx.Client(
|
||||
verify=get_ssl_configuration(),
|
||||
cert=os.getenv("SSL_CERTIFICATE", litellm.ssl_certificate),
|
||||
)
|
||||
)
|
||||
|
||||
parameters["environment"] = LangFuseLogger.resolve_deployment_environment()
|
||||
|
||||
from .langfuse_sdk import build_langfuse_client
|
||||
|
||||
client: Final = build_langfuse_client(
|
||||
parameters=parameters,
|
||||
environment=parameters["environment"],
|
||||
release=langfuse_release,
|
||||
mock_mode=is_mock_mode,
|
||||
return build_langfuse_client(
|
||||
public_key=public_key,
|
||||
secret_key=secret_key,
|
||||
base_url=langfuse_host,
|
||||
httpx_client=httpx_client,
|
||||
)
|
||||
|
||||
return client
|
||||
|
||||
|
||||
def _remember_trace_id(litellm_call_id: object, logged: LangfuseLoggedEvent) -> None:
|
||||
trace_id: Final = logged["trace_id"]
|
||||
|
|
@ -157,7 +135,7 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
|
|||
|
||||
from .langfuse_sdk import acquire_langfuse_tracing
|
||||
|
||||
self.Langfuse = langfuse_client_init(
|
||||
self.api_client = langfuse_client_init(
|
||||
langfuse_public_key=langfuse_public_key,
|
||||
langfuse_secret=langfuse_secret,
|
||||
langfuse_host=langfuse_host,
|
||||
|
|
|
|||
|
|
@ -17,15 +17,15 @@ from typing import Final, Literal
|
|||
|
||||
import httpx
|
||||
import opentelemetry.trace as otel_trace
|
||||
from langfuse import Langfuse, LangfuseOtelSpanAttributes
|
||||
from langfuse.api import LangfuseAPI
|
||||
from langfuse.model import BasePromptClient
|
||||
from langfuse import LangfuseOtelSpanAttributes
|
||||
from langfuse.api import LangfuseAPI, Prompt, Prompt_Chat
|
||||
from langfuse.model import BasePromptClient, ChatPromptClient, PromptClient, TextPromptClient
|
||||
from opentelemetry.context import Context
|
||||
from opentelemetry.sdk.resources import Resource
|
||||
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
|
||||
from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult
|
||||
from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
|
||||
from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult
|
||||
from opentelemetry.sdk.trace.sampling import ALWAYS_ON, Decision, Sampler, SamplingResult
|
||||
from opentelemetry.trace import Link, NonRecordingSpan, Span, SpanContext, SpanKind, TraceFlags, Tracer, TraceState
|
||||
from opentelemetry.util.types import Attributes, AttributeValue
|
||||
from requests import PreparedRequest, RequestException, Response, Session
|
||||
|
|
@ -36,6 +36,7 @@ from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
|||
|
||||
__all__ = (
|
||||
"DiscardingSpanExporter",
|
||||
"LangfuseApiClient",
|
||||
"LangfuseObservation",
|
||||
"LangfuseTracing",
|
||||
"RetryingSpanExporter",
|
||||
|
|
@ -43,7 +44,9 @@ __all__ = (
|
|||
"acquire_langfuse_tracing",
|
||||
"build_langfuse_client",
|
||||
"build_langfuse_tracing",
|
||||
"configured_flush_at",
|
||||
"configured_sample_rate",
|
||||
"flush_langfuse_tracing",
|
||||
"observation_attributes",
|
||||
"resolve_observation_id",
|
||||
"resolve_trace_id",
|
||||
|
|
@ -55,7 +58,19 @@ __all__ = (
|
|||
|
||||
_TRACE_ID_PATTERN: Final = re.compile(r"^(?=.*[1-9a-f])[0-9a-f]{32}$")
|
||||
_OBSERVATION_ID_PATTERN: Final = re.compile(r"^(?=.*[1-9a-f])[0-9a-f]{16}$")
|
||||
_TRACER_NAME: Final = "litellm.langfuse"
|
||||
_TRACER_NAME: Final = "langfuse-sdk"
|
||||
_MAX_QUEUE_SIZE: Final = 100_000
|
||||
_DEFAULT_FLUSH_AT: Final = 512
|
||||
_SPAN_LIMITS: Final = SpanLimits(
|
||||
max_attributes=SpanLimits.UNSET,
|
||||
max_events=128,
|
||||
max_links=128,
|
||||
max_span_attributes=SpanLimits.UNSET,
|
||||
max_event_attributes=128,
|
||||
max_link_attributes=128,
|
||||
max_attribute_length=SpanLimits.UNSET,
|
||||
max_span_attribute_length=SpanLimits.UNSET,
|
||||
)
|
||||
|
||||
|
||||
def to_unix_nanos(value: datetime | float | None) -> int | None:
|
||||
|
|
@ -76,7 +91,9 @@ def resolve_trace_id(trace_id: object | None) -> str:
|
|||
normalized: Final = serialized.lower().replace("-", "")
|
||||
if _TRACE_ID_PATTERN.fullmatch(normalized):
|
||||
return normalized
|
||||
return Langfuse.create_trace_id(seed=serialized) if serialized else Langfuse.create_trace_id()
|
||||
if not serialized:
|
||||
return format(RandomIdGenerator().generate_trace_id(), "032x")
|
||||
return sha256(serialized.encode("utf-8")).digest()[:16].hex()
|
||||
|
||||
|
||||
def resolve_observation_id(observation_id: object | None) -> str | None:
|
||||
|
|
@ -395,12 +412,6 @@ def _parse_sample_rate(raw: str) -> float | None:
|
|||
return rate if 0.0 <= rate <= 1.0 else None
|
||||
|
||||
|
||||
def _usable_sample_rate() -> float:
|
||||
raw: Final = os.environ.get("LANGFUSE_SAMPLE_RATE")
|
||||
parsed: Final = _parse_sample_rate(raw) if raw is not None else 1.0
|
||||
return 1.0 if parsed is None else parsed
|
||||
|
||||
|
||||
def configured_sample_rate() -> float:
|
||||
"""``LANGFUSE_SAMPLE_RATE`` as a fraction, exporting everything when it is unset or unusable."""
|
||||
raw: Final = os.environ.get("LANGFUSE_SAMPLE_RATE")
|
||||
|
|
@ -415,10 +426,27 @@ def configured_sample_rate() -> float:
|
|||
return parsed
|
||||
|
||||
|
||||
def configured_flush_at() -> int:
|
||||
"""``LANGFUSE_FLUSH_AT`` as the export batch size, the SDK's own knob, with its default when unset or unusable."""
|
||||
raw: Final = os.environ.get("LANGFUSE_FLUSH_AT")
|
||||
if raw is None:
|
||||
return _DEFAULT_FLUSH_AT
|
||||
parsed: Final = int(raw) if raw.strip().isdigit() else None
|
||||
if parsed is None or not 0 < parsed <= _MAX_QUEUE_SIZE:
|
||||
verbose_logger.warning(
|
||||
"LANGFUSE_FLUSH_AT=%r is not a whole number between 1 and %d; exporting batches of %d",
|
||||
raw,
|
||||
_MAX_QUEUE_SIZE,
|
||||
_DEFAULT_FLUSH_AT,
|
||||
)
|
||||
return _DEFAULT_FLUSH_AT
|
||||
return parsed
|
||||
|
||||
|
||||
class DiscardingSpanExporter(SpanExporter):
|
||||
"""Accept and drop every span, for mock mode.
|
||||
|
||||
The mock intercepts the httpx client the SDK uses for its API, but observations
|
||||
The mock intercepts the httpx client behind the REST API, but observations
|
||||
travel over OTLP, so without this the "no network calls" contract silently sends
|
||||
real traces to the configured host.
|
||||
"""
|
||||
|
|
@ -447,16 +475,17 @@ class RetryingSpanExporter(SpanExporter):
|
|||
delays: Sequence[float] = (1.0, 2.0, 4.0)
|
||||
|
||||
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
|
||||
for delay in chain(self.delays, (None,)):
|
||||
for delay in self.delays:
|
||||
try:
|
||||
return self.exporter.export(spans)
|
||||
except RequestException as error:
|
||||
if delay is None:
|
||||
verbose_logger.error("Langfuse export failed after %d retries: %s", len(self.delays), error)
|
||||
return SpanExportResult.FAILURE
|
||||
verbose_logger.warning("Langfuse export raised %s, retrying in %ss", error, delay)
|
||||
sleep(delay)
|
||||
return SpanExportResult.FAILURE
|
||||
try:
|
||||
return self.exporter.export(spans)
|
||||
except RequestException as error:
|
||||
verbose_logger.error("Langfuse export failed after %d retries: %s", len(self.delays), error)
|
||||
return SpanExportResult.FAILURE
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.exporter.shutdown()
|
||||
|
|
@ -552,6 +581,7 @@ class _TracingKey:
|
|||
environment: str | None
|
||||
release: str | None
|
||||
sample_rate: float
|
||||
flush_at: int
|
||||
flush_interval_millis: int
|
||||
mock_mode: bool
|
||||
|
||||
|
|
@ -584,6 +614,7 @@ def acquire_langfuse_tracing(
|
|||
environment=environment,
|
||||
release=release,
|
||||
sample_rate=configured_sample_rate(),
|
||||
flush_at=configured_flush_at(),
|
||||
flush_interval_millis=int(flush_interval * 1000),
|
||||
mock_mode=mock_mode,
|
||||
)
|
||||
|
|
@ -598,6 +629,7 @@ def acquire_langfuse_tracing(
|
|||
environment=environment,
|
||||
release=release,
|
||||
sample_rate=key.sample_rate,
|
||||
flush_at=key.flush_at,
|
||||
flush_interval_millis=key.flush_interval_millis,
|
||||
)
|
||||
_TRACING[key] = created
|
||||
|
|
@ -641,63 +673,116 @@ def build_langfuse_tracing(
|
|||
release: str | None,
|
||||
sample_rate: float,
|
||||
flush_interval_millis: int,
|
||||
flush_at: int = _DEFAULT_FLUSH_AT,
|
||||
) -> LangfuseTracing:
|
||||
"""Wire the provider from litellm's own settings so a host's ``OTEL_*`` variables do not steer it.
|
||||
|
||||
An unset sampler or span limit falls back to ``OTEL_TRACES_SAMPLER`` and
|
||||
``OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT`` style variables, which are meant for the
|
||||
host application's own tracing. ``OTEL_SDK_DISABLED`` still applies, as it does to the SDK.
|
||||
|
||||
The tracer carries the SDK's scope name because Langfuse keys on it: spans from any other
|
||||
scope are treated as foreign OTel traffic and get their raw attributes echoed into metadata.
|
||||
"""
|
||||
if os.environ.get("OTEL_SDK_DISABLED", "").strip().lower() == "true":
|
||||
verbose_logger.warning("OTEL_SDK_DISABLED=true also disables the langfuse callback's export channel")
|
||||
provider: Final = TracerProvider(
|
||||
resource=_resource(environment=environment, release=release),
|
||||
sampler=TraceIdHashSampler(sample_rate) if sample_rate < 1 else None,
|
||||
sampler=ALWAYS_ON if sample_rate >= 1 else TraceIdHashSampler(sample_rate),
|
||||
id_generator=_RequestedIdGenerator(),
|
||||
span_limits=_SPAN_LIMITS,
|
||||
)
|
||||
provider.add_span_processor(
|
||||
BatchSpanProcessor(
|
||||
exporter,
|
||||
max_queue_size=_MAX_QUEUE_SIZE,
|
||||
max_export_batch_size=flush_at,
|
||||
schedule_delay_millis=flush_interval_millis,
|
||||
)
|
||||
)
|
||||
provider.add_span_processor(BatchSpanProcessor(exporter, schedule_delay_millis=flush_interval_millis))
|
||||
return LangfuseTracing(provider=provider, tracer=provider.get_tracer(_TRACER_NAME))
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _CachedPrompt:
|
||||
prompt: PromptClient
|
||||
fetched_at: float
|
||||
|
||||
|
||||
def _prompt_client(prompt: Prompt) -> PromptClient:
|
||||
return ChatPromptClient(prompt) if isinstance(prompt, Prompt_Chat) else TextPromptClient(prompt)
|
||||
|
||||
|
||||
class LangfuseApiClient:
|
||||
"""litellm's handle on one Langfuse project over its REST API: prompts, ``auth_check`` and the project id.
|
||||
|
||||
The SDK's ``Langfuse`` client is deliberately not constructed. It keeps one tracing bundle per
|
||||
public key and hands it to every ``Langfuse()`` a host application builds for the same key, so
|
||||
litellm's exporter, host and masking would leak into that application. Observations travel
|
||||
over ``LangfuseTracing``; nothing here exports spans.
|
||||
|
||||
Prompts are cached for ``LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS`` (60 by default) as the SDK
|
||||
does, refreshed on the request that finds them stale; a refresh that fails keeps serving the
|
||||
stale prompt rather than failing the request, again like the SDK.
|
||||
"""
|
||||
|
||||
def __init__(self, api: LangfuseAPI, *, prompt_cache_ttl_seconds: float) -> None:
|
||||
self.api: Final = api
|
||||
self.prompt_cache_ttl_seconds: Final = prompt_cache_ttl_seconds
|
||||
self._prompts: Final[dict[str, _CachedPrompt]] = {} # mutable-ok: per-client prompt cache, guarded by _lock
|
||||
self._lock: Final = threading.Lock()
|
||||
|
||||
def auth_check(self) -> bool:
|
||||
try:
|
||||
self.api.projects.get()
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
def project_id(self) -> str | None:
|
||||
projects: Final = self.api.projects.get().data
|
||||
return projects[0].id if projects else None
|
||||
|
||||
def get_prompt(self, name: str, *, label: str | None = None, version: int | None = None) -> PromptClient:
|
||||
key: Final = f"{name}:version:{version}" if version is not None else f"{name}:label:{label}"
|
||||
with self._lock:
|
||||
cached: Final = self._prompts.get(key)
|
||||
if cached is not None and monotonic() - cached.fetched_at < self.prompt_cache_ttl_seconds:
|
||||
return cached.prompt
|
||||
try:
|
||||
fetched: Final = _prompt_client(self.api.prompts.get(name, version=version, label=label))
|
||||
except Exception as error:
|
||||
if cached is None:
|
||||
raise
|
||||
verbose_logger.warning("Langfuse prompt %r refresh failed, serving the cached version: %s", name, error)
|
||||
return cached.prompt
|
||||
with self._lock:
|
||||
self._prompts[key] = _CachedPrompt(prompt=fetched, fetched_at=monotonic())
|
||||
return fetched
|
||||
|
||||
|
||||
def build_langfuse_client(
|
||||
*,
|
||||
parameters: Mapping[str, object],
|
||||
environment: str | None,
|
||||
release: str | None,
|
||||
mock_mode: bool,
|
||||
) -> Langfuse:
|
||||
"""The SDK client litellm keeps for prompt management and ``auth_check``.
|
||||
public_key: str | None,
|
||||
secret_key: str | None,
|
||||
base_url: str,
|
||||
httpx_client: httpx.Client | None,
|
||||
) -> LangfuseApiClient:
|
||||
"""The REST client for prompt management, ``auth_check`` and the Slack project link.
|
||||
|
||||
Observations never go through it, but the SDK still builds a tracer for it and, given no
|
||||
provider, claims the process-global one, which disables litellm's other OTel exporters.
|
||||
It gets a provider of its own instead. The SDK caches one resource bundle per public key,
|
||||
so a user application constructing ``Langfuse`` for the same key afterwards shares this
|
||||
bundle; the exporter it carries is litellm's so that application's spans still reach Langfuse.
|
||||
|
||||
That same cache keeps the first secret and host it saw for a public key, so the REST client
|
||||
behind ``get_prompt`` and ``auth_check`` is rebuilt from the credentials actually supplied.
|
||||
Without both keys the SDK disables the client, which has no REST client to rebuild.
|
||||
|
||||
The SDK reads ``LANGFUSE_SAMPLE_RATE`` itself and raises on anything it cannot parse, so it
|
||||
gets the rate litellm already validated; the sampler that matters is on litellm's provider.
|
||||
Missing keys are passed through as absent credentials: the server answers 401, which
|
||||
``auth_check`` reports as ``False`` rather than raising at construction.
|
||||
"""
|
||||
public_key: Final = parameters.get("public_key")
|
||||
secret_key: Final = parameters.get("secret_key")
|
||||
base_url: Final = str(parameters.get("base_url"))
|
||||
httpx_client: Final = parameters.get("httpx_client")
|
||||
credentialed: Final = isinstance(public_key, str) and isinstance(secret_key, str)
|
||||
client: Final = Langfuse(
|
||||
**parameters, # pyright: ignore[reportArgumentType] # kwargs-ok: dict mirrors the typed ctor, values resolved by the callers
|
||||
sample_rate=_usable_sample_rate(),
|
||||
tracer_provider=TracerProvider(
|
||||
resource=_resource(environment=environment, release=release), shutdown_on_exit=False
|
||||
return LangfuseApiClient(
|
||||
LangfuseAPI(
|
||||
base_url=base_url,
|
||||
username=public_key,
|
||||
password=secret_key,
|
||||
x_langfuse_sdk_name="python",
|
||||
x_langfuse_sdk_version=version("langfuse"),
|
||||
x_langfuse_public_key=public_key,
|
||||
httpx_client=httpx_client,
|
||||
timeout=int(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
),
|
||||
span_exporter=_build_span_exporter(public_key=str(public_key), secret_key=str(secret_key), base_url=base_url)
|
||||
if credentialed and not mock_mode
|
||||
else DiscardingSpanExporter(),
|
||||
prompt_cache_ttl_seconds=float(os.getenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", "60")),
|
||||
)
|
||||
if not isinstance(public_key, str) or not isinstance(secret_key, str):
|
||||
return client
|
||||
client.api = LangfuseAPI(
|
||||
base_url=base_url,
|
||||
username=public_key,
|
||||
password=secret_key,
|
||||
x_langfuse_sdk_name="python",
|
||||
x_langfuse_sdk_version=version("langfuse"),
|
||||
x_langfuse_public_key=public_key,
|
||||
httpx_client=httpx_client if isinstance(httpx_client, httpx.Client) else None,
|
||||
timeout=int(os.getenv("LANGFUSE_TIMEOUT", "5")),
|
||||
)
|
||||
return client
|
||||
|
|
|
|||
|
|
@ -19,10 +19,12 @@ from ...caching import InMemoryCache
|
|||
|
||||
class LangfuseInMemoryCache(InMemoryCache):
|
||||
"""
|
||||
Releases the initialized-client slot of a LangfuseLogger when it expires.
|
||||
Decrements ``litellm.initialized_langfuse_clients`` when a LangFuseLogger entry expires.
|
||||
|
||||
Export channels are shared per credential set and outlive the logger, so
|
||||
nothing else needs tearing down (https://github.com/BerriAI/litellm/issues/11169).
|
||||
The counter is a soft budget: loggers built concurrently for one credential set before the
|
||||
first lands in the cache each take a slot, and only the cached one gives it back on expiry.
|
||||
Export channels are shared per credential set and outlive the logger, so nothing else is
|
||||
torn down here (https://github.com/BerriAI/litellm/issues/11169).
|
||||
"""
|
||||
|
||||
def _remove_key(self, key: str) -> None:
|
||||
|
|
|
|||
|
|
@ -394,7 +394,7 @@ async def health_services_endpoint(
|
|||
from litellm.integrations.langfuse.langfuse import LangFuseLogger
|
||||
|
||||
langfuse_logger: Final = LangFuseLogger()
|
||||
if langfuse_logger.Langfuse.auth_check() is False:
|
||||
if langfuse_logger.api_client.auth_check() is False:
|
||||
raise ValueError(
|
||||
"langfuse auth_check failed - verify LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY are set correctly"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -67,6 +67,7 @@ from litellm.constants import (
|
|||
DEFAULT_SHARED_HEALTH_CHECK_LOCK_TTL,
|
||||
DEFAULT_SHARED_HEALTH_CHECK_TTL,
|
||||
DEFAULT_SLACK_ALERTING_THRESHOLD,
|
||||
LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS,
|
||||
LITELLM_EMBEDDING_PROVIDERS_SUPPORTING_INPUT_ARRAY_OF_TOKENS,
|
||||
LITELLM_SETTINGS_SAFE_DB_OVERRIDES,
|
||||
LITELLM_UI_ALLOW_HEADERS,
|
||||
|
|
@ -1040,10 +1041,16 @@ async def proxy_shutdown_event(worker_heartbeat: ProxyWorkerHeartbeat | None = N
|
|||
try:
|
||||
from litellm.integrations.langfuse.langfuse_sdk import flush_langfuse_tracing
|
||||
|
||||
flush_langfuse_tracing()
|
||||
except Exception:
|
||||
# [DO NOT BLOCK shutdown events for this]
|
||||
pass
|
||||
flushed: Final = await asyncio.to_thread(flush_langfuse_tracing, LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS)
|
||||
if flushed:
|
||||
verbose_proxy_logger.info("Langfuse export channels flushed")
|
||||
else:
|
||||
verbose_proxy_logger.warning(
|
||||
"Langfuse export did not finish within %dms; remaining spans are left to the background exporter",
|
||||
LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 # shutdown must continue even if the flush fails
|
||||
verbose_proxy_logger.exception("Error flushing Langfuse export channels on shutdown: %s", e)
|
||||
|
||||
## RESET CUSTOM VARIABLES ##
|
||||
cleanup_router_config_variables()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,3 @@
|
|||
import sys
|
||||
from types import ModuleType, SimpleNamespace
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.langfuse.langfuse import resolve_langfuse_credentials
|
||||
from litellm.integrations.langfuse.langfuse_handler import LangFuseHandler
|
||||
|
|
@ -51,46 +48,29 @@ def test_resolve_langfuse_credentials_keeps_env_for_global_config(monkeypatch):
|
|||
assert host == "https://admin-configured.example"
|
||||
|
||||
|
||||
def test_upstream_langfuse_env_is_read_without_building_a_client(monkeypatch):
|
||||
"""The UPSTREAM_LANGFUSE_* values are recorded, and nothing consumes them.
|
||||
|
||||
A client was built here and never referenced. On v4 that means a second
|
||||
exporter, its own threads, and an entry in the SDK's per-key registry, so it
|
||||
is no longer constructed.
|
||||
"""
|
||||
def test_upstream_langfuse_env_only_warns_and_opens_no_second_channel(monkeypatch, caplog):
|
||||
"""UPSTREAM_LANGFUSE_* configured a second v2 ingestion client. v4 has one export channel per
|
||||
credential set, so the values are ignored with a startup warning and never build anything."""
|
||||
from litellm.integrations.langfuse import langfuse_sdk
|
||||
from litellm.integrations.langfuse.langfuse import LangFuseLogger
|
||||
|
||||
class FakeLangfuse:
|
||||
instances = []
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.kwargs = kwargs
|
||||
FakeLangfuse.instances.append(self)
|
||||
|
||||
fake_langfuse_module = ModuleType("langfuse")
|
||||
fake_langfuse_module.Langfuse = FakeLangfuse
|
||||
fake_langfuse_module.version = SimpleNamespace(__version__="2.6.0")
|
||||
|
||||
monkeypatch.setitem(sys.modules, "langfuse", fake_langfuse_module)
|
||||
monkeypatch.setattr(litellm, "initialized_langfuse_clients", 0)
|
||||
monkeypatch.setattr(langfuse_sdk, "_TRACING", {})
|
||||
monkeypatch.setenv("LANGFUSE_MOCK", "true")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_SECRET_KEY", "upstream-secret")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_PUBLIC_KEY", "upstream-public")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_HOST", "https://upstream.example")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_RELEASE", "release")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_DEBUG", "true")
|
||||
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="public",
|
||||
langfuse_secret="secret",
|
||||
langfuse_host="https://langfuse.example",
|
||||
)
|
||||
with caplog.at_level("WARNING", logger="LiteLLM"):
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="public",
|
||||
langfuse_secret="secret",
|
||||
langfuse_host="https://langfuse.example",
|
||||
)
|
||||
|
||||
assert logger.upstream_langfuse_debug == "true"
|
||||
assert logger.upstream_langfuse_public_key == "upstream-public"
|
||||
assert logger.upstream_langfuse_host == "https://upstream.example"
|
||||
assert logger.upstream_langfuse_release == "release"
|
||||
assert not hasattr(logger, "upstream_langfuse")
|
||||
assert any("UPSTREAM_LANGFUSE_* is no longer supported" in record.getMessage() for record in caplog.records)
|
||||
assert list(langfuse_sdk._TRACING.values()) == [logger.tracing]
|
||||
assert all(key.public_key == "public" for key in langfuse_sdk._TRACING)
|
||||
|
||||
|
||||
def test_langfuse_handler_accepts_secret_key_alias(monkeypatch):
|
||||
|
|
|
|||
|
|
@ -354,7 +354,7 @@ def test_langfuse_e2e_sync(monkeypatch):
|
|||
)
|
||||
for logger in litellm.logging_callback_manager._get_all_callbacks():
|
||||
if isinstance(logger, LangFuseLogger):
|
||||
logger.Langfuse.flush()
|
||||
logger.flush()
|
||||
deadline = time.time() + 10
|
||||
while not received_paths and time.time() < deadline:
|
||||
time.sleep(0.1)
|
||||
|
|
|
|||
|
|
@ -71,20 +71,15 @@ class TestLangfusePromptManagement:
|
|||
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
|
||||
|
||||
shared_client = _get_httpx_client().client
|
||||
|
||||
mock_langfuse_class = MagicMock()
|
||||
built = MagicMock()
|
||||
with (
|
||||
patch(
|
||||
"litellm.integrations.langfuse.langfuse_prompt_management.resolve_langfuse_credentials",
|
||||
return_value=("pk-1234", "sk-1234", "https://localhost"),
|
||||
),
|
||||
patch(
|
||||
"litellm.integrations.langfuse.langfuse_prompt_management.LangFuseLogger._get_langfuse_flush_interval",
|
||||
return_value=1,
|
||||
),
|
||||
patch(
|
||||
"litellm.integrations.langfuse.langfuse_sdk.Langfuse", mock_langfuse_class
|
||||
), # test-quality-ok: the ctor must be intercepted where build_langfuse_client resolves it; a real client spawns export threads
|
||||
"litellm.integrations.langfuse.langfuse_sdk.build_langfuse_client", built
|
||||
), # test-quality-ok: the REST client is built where langfuse_client_init resolves it; the transport it gets is the behavior under test
|
||||
patch(
|
||||
"litellm.llms.custom_httpx.http_handler.get_ssl_configuration",
|
||||
return_value=False,
|
||||
|
|
@ -96,10 +91,8 @@ class TestLangfusePromptManagement:
|
|||
langfuse_host="https://localhost",
|
||||
)
|
||||
|
||||
mock_langfuse_class.assert_called_once()
|
||||
call_kwargs = mock_langfuse_class.call_args[1]
|
||||
assert "httpx_client" in call_kwargs
|
||||
passed_client = call_kwargs["httpx_client"]
|
||||
built.assert_called_once()
|
||||
passed_client = built.call_args.kwargs["httpx_client"]
|
||||
assert isinstance(passed_client, httpx.Client)
|
||||
assert passed_client is not shared_client
|
||||
mock_get_ssl.assert_called_once()
|
||||
|
|
@ -107,32 +100,22 @@ class TestLangfusePromptManagement:
|
|||
langfuse_client_init.cache_clear()
|
||||
|
||||
|
||||
class _RecordingLangfuseForEnv:
|
||||
last_environment: str | None = None
|
||||
|
||||
def __init__(
|
||||
self, *, environment: str | None = None, **parameters: object
|
||||
) -> None: # kwargs-ok: records only environment out of whatever langfuse_client_init forwards
|
||||
type(self).last_environment = environment
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("env_value", "expected"),
|
||||
(("Production", "default"), ("production ", "production"), ("prod", "prod")),
|
||||
)
|
||||
def test_langfuse_client_init_resolves_deployment_environment(monkeypatch, env_value, expected):
|
||||
def test_prompt_management_logger_exports_the_resolved_deployment_environment(monkeypatch, env_value, expected):
|
||||
from langfuse import LangfuseOtelSpanAttributes
|
||||
|
||||
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_HOST", "http://127.0.0.1:1")
|
||||
monkeypatch.setenv("LANGFUSE_MOCK", "true")
|
||||
monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", env_value)
|
||||
monkeypatch.setattr(_RecordingLangfuseForEnv, "last_environment", None)
|
||||
with patch(
|
||||
"litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuseForEnv
|
||||
): # 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()
|
||||
assert _RecordingLangfuseForEnv.last_environment == expected
|
||||
logger = LangfusePromptManagement()
|
||||
langfuse_client_init.cache_clear()
|
||||
assert logger.tracing.provider.resource.attributes[LangfuseOtelSpanAttributes.ENVIRONMENT] == expected
|
||||
|
||||
|
||||
def test_langfuse_client_init_warns_that_upstream_langfuse_is_ignored(monkeypatch, caplog):
|
||||
|
|
@ -143,12 +126,7 @@ def test_langfuse_client_init_warns_that_upstream_langfuse_is_ignored(monkeypatc
|
|||
monkeypatch.setenv("LANGFUSE_HOST", "https://test.langfuse.com")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_SECRET_KEY", "sk-upstream")
|
||||
monkeypatch.setenv("UPSTREAM_LANGFUSE_HOST", "https://upstream.example")
|
||||
with (
|
||||
patch(
|
||||
"litellm.integrations.langfuse.langfuse_sdk.Langfuse", _RecordingLangfuseForEnv
|
||||
), # 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"),
|
||||
):
|
||||
with caplog.at_level("WARNING", logger="LiteLLM"):
|
||||
langfuse_client_init.cache_clear()
|
||||
langfuse_client_init()
|
||||
langfuse_client_init.cache_clear()
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ import httpx
|
|||
import opentelemetry.trace as otel_trace
|
||||
import pytest
|
||||
from langfuse import LangfuseOtelSpanAttributes as A
|
||||
from langfuse.api import UnauthorizedError
|
||||
from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
|
|
@ -36,6 +35,7 @@ from litellm.integrations.langfuse.langfuse_sdk import (
|
|||
acquire_langfuse_tracing,
|
||||
build_langfuse_client,
|
||||
build_langfuse_tracing,
|
||||
configured_flush_at,
|
||||
configured_sample_rate,
|
||||
flush_langfuse_tracing,
|
||||
observation_attributes,
|
||||
|
|
@ -457,6 +457,165 @@ def test_configured_sample_rate_reads_the_env_var(monkeypatch: pytest.MonkeyPatc
|
|||
assert configured_sample_rate() == 0.25
|
||||
|
||||
|
||||
def _exported_generations(exporter: InMemorySpanExporter, tracing: LangfuseTracing, count: int) -> tuple:
|
||||
for _ in range(count):
|
||||
_generation(tracing, trace_id=resolve_trace_id(uuid.uuid4())).end(CALL_END)
|
||||
tracing.flush()
|
||||
return exporter.get_finished_spans()
|
||||
|
||||
|
||||
def test_full_sample_rate_exports_every_trace_even_when_the_host_turned_otel_sampling_off(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
"""A provider built without a sampler reads ``OTEL_TRACES_SAMPLER``, which belongs to the host's tracing."""
|
||||
monkeypatch.setenv("OTEL_TRACES_SAMPLER", "always_off")
|
||||
exporter = InMemorySpanExporter()
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
|
||||
)
|
||||
assert len(_exported_generations(exporter, tracing, 5)) == 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("variable", "value"),
|
||||
[
|
||||
("OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT", "4"),
|
||||
("OTEL_ATTRIBUTE_COUNT_LIMIT", "4"),
|
||||
("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "8"),
|
||||
("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "8"),
|
||||
],
|
||||
)
|
||||
def test_host_otel_span_limits_do_not_truncate_langfuse_observations(
|
||||
monkeypatch: pytest.MonkeyPatch, variable: str, value: str
|
||||
):
|
||||
monkeypatch.setenv(variable, value)
|
||||
exporter = InMemorySpanExporter()
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
|
||||
)
|
||||
attributes = {f"langfuse.observation.metadata.k{i}": "v" * 32 for i in range(40)}
|
||||
_generation(tracing, attributes=attributes).end(CALL_END)
|
||||
tracing.flush()
|
||||
|
||||
span = _only_span(exporter, "gen")
|
||||
assert span.dropped_attributes == 0
|
||||
assert all(span.attributes[key] == "v" * 32 for key in attributes)
|
||||
|
||||
|
||||
def test_many_metadata_keys_never_evict_the_generation_input_and_output():
|
||||
"""OTel's default 128-attribute cap drops the earliest attributes, and v2 never capped metadata."""
|
||||
exporter = InMemorySpanExporter()
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
|
||||
)
|
||||
attributes = {
|
||||
A.OBSERVATION_INPUT: "the-prompt",
|
||||
A.OBSERVATION_OUTPUT: "the-completion",
|
||||
**{f"langfuse.observation.metadata.k{i}": str(i) for i in range(300)},
|
||||
}
|
||||
_generation(tracing, attributes=attributes).end(CALL_END)
|
||||
tracing.flush()
|
||||
|
||||
span = _only_span(exporter, "gen")
|
||||
assert span.dropped_attributes == 0
|
||||
assert span.attributes[A.OBSERVATION_INPUT] == "the-prompt"
|
||||
assert span.attributes[A.OBSERVATION_OUTPUT] == "the-completion"
|
||||
assert span.attributes["langfuse.observation.metadata.k299"] == "299"
|
||||
|
||||
|
||||
def test_otel_sdk_disabled_still_wins_but_is_called_out(monkeypatch: pytest.MonkeyPatch, caplog):
|
||||
monkeypatch.setenv("OTEL_SDK_DISABLED", "true")
|
||||
exporter = InMemorySpanExporter()
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
|
||||
)
|
||||
assert "OTEL_SDK_DISABLED" in caplog.text
|
||||
assert _exported_generations(exporter, tracing, 3) == ()
|
||||
|
||||
|
||||
def test_spans_carry_the_langfuse_sdk_scope_name(channel):
|
||||
"""Langfuse keys on the SDK's instrumentation scope (langfuse 4.15.2, ``langfuse/_client/constants.py``,
|
||||
read 2026-09-17); any other scope is foreign OTel traffic whose raw attributes get echoed into metadata."""
|
||||
tracing, exporter = channel
|
||||
_generation(tracing).end(CALL_END)
|
||||
tracing.flush()
|
||||
assert _only_span(exporter, "gen").instrumentation_scope.name == "langfuse-sdk"
|
||||
|
||||
|
||||
class _GatedExporter(SpanExporter):
|
||||
"""Hold the export thread until released, so spans pile up in the processor queue."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.gate = threading.Event()
|
||||
self.batches: list[int] = []
|
||||
|
||||
def export(self, spans) -> SpanExportResult:
|
||||
self.gate.wait(timeout=30)
|
||||
self.batches.append(len(spans))
|
||||
return SpanExportResult.SUCCESS
|
||||
|
||||
def shutdown(self) -> None:
|
||||
return None
|
||||
|
||||
def force_flush(self, timeout_millis: int = 30_000) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def test_export_queue_holds_a_v2_sized_burst_while_the_destination_stalls():
|
||||
"""v2 queued 100k events; OTel's default 2048 dropped most of a burst during a destination stall."""
|
||||
exporter = _GatedExporter()
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
|
||||
)
|
||||
for _ in range(6000):
|
||||
_generation(tracing, trace_id=resolve_trace_id(uuid.uuid4())).end(CALL_END)
|
||||
exporter.gate.set()
|
||||
assert tracing.flush(timeout_millis=30_000) is True
|
||||
assert sum(exporter.batches) == 6000
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw", "expected"),
|
||||
[(None, 512), ("64", 64), ("0", 512), ("-5", 512), ("abc", 512), ("100001", 512), ("100000", 100_000)],
|
||||
ids=["unset", "valid", "zero", "negative", "text", "over-queue", "at-queue"],
|
||||
)
|
||||
def test_langfuse_flush_at_is_parsed_like_the_sdk_did(monkeypatch: pytest.MonkeyPatch, raw, expected, caplog):
|
||||
if raw is None:
|
||||
monkeypatch.delenv("LANGFUSE_FLUSH_AT", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("LANGFUSE_FLUSH_AT", raw)
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
|
||||
assert configured_flush_at() == expected
|
||||
assert ("LANGFUSE_FLUSH_AT" in caplog.text) is (raw is not None and str(expected) != raw)
|
||||
|
||||
|
||||
def test_langfuse_flush_at_sizes_the_export_batches(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setenv("LANGFUSE_FLUSH_AT", "64")
|
||||
exporter = _GatedExporter()
|
||||
exporter.gate.set()
|
||||
tracing = build_langfuse_tracing(
|
||||
exporter=exporter,
|
||||
environment=None,
|
||||
release=None,
|
||||
sample_rate=1.0,
|
||||
flush_interval_millis=60_000,
|
||||
flush_at=configured_flush_at(),
|
||||
)
|
||||
for _ in range(200):
|
||||
_generation(tracing, trace_id=resolve_trace_id(uuid.uuid4())).end(CALL_END)
|
||||
tracing.flush()
|
||||
assert sum(exporter.batches) == 200
|
||||
assert max(exporter.batches) == 64
|
||||
|
||||
|
||||
def test_acquired_channel_reads_langfuse_flush_at(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setenv("LANGFUSE_FLUSH_AT", "7")
|
||||
small = _acquire(public_key="pk-flush-at-test")
|
||||
monkeypatch.setenv("LANGFUSE_FLUSH_AT", "9")
|
||||
assert _acquire(public_key="pk-flush-at-test") is not small
|
||||
|
||||
|
||||
def test_channel_does_not_take_over_the_process_tracer_provider():
|
||||
provider_before = otel_trace.get_tracer_provider()
|
||||
|
||||
|
|
@ -589,82 +748,97 @@ def test_a_changed_sample_rate_rebuilds_the_channel(monkeypatch: pytest.MonkeyPa
|
|||
assert "TraceIdHashSampler" not in full.provider.sampler.get_description()
|
||||
|
||||
|
||||
def test_sdk_client_rest_api_follows_the_supplied_credentials_not_the_registry():
|
||||
"""The SDK keeps one resource bundle per public key, so a rotated secret or another host
|
||||
would otherwise keep authenticating prompt fetches with whatever it saw first."""
|
||||
requests = []
|
||||
|
||||
def _recording_transport(requests: list[httpx.Request], status: int = 401) -> httpx.Client:
|
||||
def record(request: httpx.Request) -> httpx.Response:
|
||||
requests.append(request)
|
||||
return httpx.Response(401, json={"message": "unauthorized"})
|
||||
return httpx.Response(status, json=_PROJECTS_BODY if status == 200 else {"message": "unauthorized"})
|
||||
|
||||
parameters = {
|
||||
"public_key": "pk-rest-test",
|
||||
"secret_key": "sk-first",
|
||||
"base_url": "http://127.0.0.1:1",
|
||||
"httpx_client": httpx.Client(transport=httpx.MockTransport(record)),
|
||||
}
|
||||
build_langfuse_client(parameters=parameters, environment=None, release=None, mock_mode=True)
|
||||
return httpx.Client(transport=httpx.MockTransport(record))
|
||||
|
||||
|
||||
_PROJECTS_BODY: Final = {
|
||||
"data": [{"id": "proj-under-test", "name": "p", "metadata": {}, "organization": {"id": "o", "name": "o"}}]
|
||||
}
|
||||
|
||||
|
||||
def test_rest_client_authenticates_with_the_credentials_it_was_built_with():
|
||||
"""Two loggers for one public key but different secrets or hosts each talk to their own project."""
|
||||
requests: list[httpx.Request] = []
|
||||
build_langfuse_client(
|
||||
public_key="pk-rest-test",
|
||||
secret_key="sk-first",
|
||||
base_url="http://127.0.0.1:1",
|
||||
httpx_client=_recording_transport(requests),
|
||||
)
|
||||
rotated = build_langfuse_client(
|
||||
parameters={**parameters, "secret_key": "sk-second", "base_url": "http://127.0.0.1:2"},
|
||||
environment=None,
|
||||
release=None,
|
||||
mock_mode=True,
|
||||
public_key="pk-rest-test",
|
||||
secret_key="sk-second",
|
||||
base_url="http://127.0.0.1:2",
|
||||
httpx_client=_recording_transport(requests),
|
||||
)
|
||||
|
||||
with pytest.raises(UnauthorizedError):
|
||||
rotated.auth_check()
|
||||
assert rotated.auth_check() is False
|
||||
assert requests[-1].url.host == "127.0.0.1" and requests[-1].url.port == 2
|
||||
assert requests[-1].headers["authorization"] == "Basic " + b64encode(b"pk-rest-test:sk-second").decode()
|
||||
|
||||
|
||||
def test_sdk_client_without_keys_is_built_disabled_and_fails_auth_check(monkeypatch):
|
||||
def test_rest_client_without_keys_fails_auth_check_instead_of_raising(monkeypatch):
|
||||
"""``/health/services?service=langfuse`` with no credentials must report a failed check, not crash."""
|
||||
for name in ("LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
client = build_langfuse_client(
|
||||
parameters={"public_key": None, "secret_key": None, "base_url": "http://127.0.0.1:1"},
|
||||
environment=None,
|
||||
release=None,
|
||||
mock_mode=True,
|
||||
)
|
||||
client = build_langfuse_client(public_key=None, secret_key=None, base_url="http://127.0.0.1:1", httpx_client=None)
|
||||
assert client.auth_check() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", ["1.5", "-0.5", "abc"])
|
||||
def test_sdk_client_is_built_despite_an_unusable_sample_rate(monkeypatch: pytest.MonkeyPatch, raw: str):
|
||||
"""The SDK parses ``LANGFUSE_SAMPLE_RATE`` itself and would raise, which took the whole callback down."""
|
||||
monkeypatch.setenv("LANGFUSE_SAMPLE_RATE", raw)
|
||||
requests = []
|
||||
|
||||
def record(request: httpx.Request) -> httpx.Response:
|
||||
requests.append(request)
|
||||
return httpx.Response(401, json={"message": "unauthorized"})
|
||||
|
||||
def test_rest_client_reports_the_project_id_and_a_passing_auth_check():
|
||||
requests: list[httpx.Request] = []
|
||||
client = build_langfuse_client(
|
||||
parameters={
|
||||
"public_key": "pk-sr-test-" + raw,
|
||||
"secret_key": "sk",
|
||||
"base_url": "http://127.0.0.1:1",
|
||||
"httpx_client": httpx.Client(transport=httpx.MockTransport(record)),
|
||||
},
|
||||
environment=None,
|
||||
release=None,
|
||||
mock_mode=True,
|
||||
public_key="pk-project-test",
|
||||
secret_key="sk",
|
||||
base_url="http://127.0.0.1:1",
|
||||
httpx_client=_recording_transport(requests, status=200),
|
||||
)
|
||||
|
||||
with pytest.raises(UnauthorizedError):
|
||||
client.auth_check()
|
||||
assert requests[-1].headers["authorization"] == "Basic " + b64encode(f"pk-sr-test-{raw}:sk".encode()).decode()
|
||||
assert client.project_id() == "proj-under-test"
|
||||
assert client.auth_check() is True
|
||||
|
||||
|
||||
def test_sdk_client_does_not_take_over_the_process_tracer_provider():
|
||||
def test_rest_client_leaves_a_host_applications_langfuse_client_alone():
|
||||
"""The SDK hands every ``Langfuse()`` built for one public key the same resource bundle, so a
|
||||
litellm-built SDK client used to make a host application's client fetch through litellm's
|
||||
host, secret and httpx client. litellm now speaks REST directly and registers nothing."""
|
||||
from langfuse import Langfuse
|
||||
|
||||
requests: list[httpx.Request] = []
|
||||
litellm_client = build_langfuse_client(
|
||||
public_key="pk-shared-with-host",
|
||||
secret_key="sk-litellm",
|
||||
base_url="http://litellm.example",
|
||||
httpx_client=_recording_transport(requests, status=200),
|
||||
)
|
||||
assert litellm_client.project_id() == "proj-under-test"
|
||||
|
||||
host_requests: list[httpx.Request] = []
|
||||
host = Langfuse(
|
||||
public_key="pk-shared-with-host",
|
||||
secret_key="sk-host",
|
||||
base_url="http://host.example",
|
||||
httpx_client=_recording_transport(host_requests, status=200),
|
||||
tracing_enabled=False,
|
||||
)
|
||||
try:
|
||||
assert host.auth_check() is True
|
||||
finally:
|
||||
host.shutdown()
|
||||
|
||||
assert [request.url.host for request in requests] == ["litellm.example"]
|
||||
assert host_requests[-1].url.host == "host.example"
|
||||
assert host_requests[-1].headers["authorization"] == "Basic " + b64encode(b"pk-shared-with-host:sk-host").decode()
|
||||
|
||||
|
||||
def test_rest_client_does_not_take_over_the_process_tracer_provider():
|
||||
provider_before = otel_trace.get_tracer_provider()
|
||||
build_langfuse_client(
|
||||
parameters={"public_key": "pk-sdk-global-test", "secret_key": "sk", "base_url": "http://127.0.0.1:1"},
|
||||
environment=None,
|
||||
release=None,
|
||||
mock_mode=True,
|
||||
public_key="pk-sdk-global-test", secret_key="sk", base_url="http://127.0.0.1:1", httpx_client=None
|
||||
)
|
||||
assert otel_trace.get_tracer_provider() is provider_before
|
||||
|
||||
|
|
|
|||
|
|
@ -35,29 +35,6 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
)
|
||||
self.env_patcher.start()
|
||||
|
||||
# Create mock objects
|
||||
self.mock_langfuse_client = MagicMock()
|
||||
# Mock the client attribute to prevent errors during logger initialization
|
||||
self.mock_langfuse_client.client = MagicMock()
|
||||
self.mock_langfuse_trace = MagicMock()
|
||||
self.mock_langfuse_generation = MagicMock()
|
||||
self.mock_langfuse_generation.trace_id = "test-trace-id"
|
||||
|
||||
# Mock span method for trace (used by log_provider_specific_information_as_span and _log_guardrail_information_as_span)
|
||||
self.mock_langfuse_span = MagicMock()
|
||||
self.mock_langfuse_span.end = MagicMock()
|
||||
self.mock_langfuse_trace.span.return_value = self.mock_langfuse_span
|
||||
|
||||
# Setup the trace and generation chain
|
||||
self.mock_langfuse_trace.generation.return_value = self.mock_langfuse_generation
|
||||
self.last_trace_kwargs = {}
|
||||
|
||||
def _trace_side_effect(*args, **kwargs):
|
||||
self.last_trace_kwargs = kwargs
|
||||
return self.mock_langfuse_trace
|
||||
|
||||
self.mock_langfuse_client.trace.side_effect = _trace_side_effect
|
||||
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
|
||||
|
|
@ -68,18 +45,10 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
self.real_provider = TracerProvider()
|
||||
self.real_provider.add_span_processor(SimpleSpanProcessor(self.span_exporter))
|
||||
|
||||
# the real SDK is installed; inject the client instead of replacing the module,
|
||||
# so the v4 imports under test resolve normally
|
||||
import langfuse as _langfuse_module
|
||||
|
||||
self.real_langfuse_class = _langfuse_module.Langfuse
|
||||
# no patching: the host above is unreachable, so a real client is cheap to build
|
||||
# and each test swaps in the client it wants
|
||||
# the host above is unreachable, so the REST client is cheap to build
|
||||
# and each test swaps in the export channel it wants
|
||||
self.logger = LangFuseLogger()
|
||||
|
||||
# Explicitly set the Langfuse client to our mock
|
||||
self.logger.Langfuse = self.mock_langfuse_client
|
||||
|
||||
# Add the log_event_on_langfuse method to the instance
|
||||
def log_event_on_langfuse(
|
||||
self,
|
||||
|
|
@ -113,9 +82,6 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
def tearDown(self):
|
||||
# Clean up logger instance to prevent state leakage
|
||||
if hasattr(self, "logger"):
|
||||
# Reset logger's Langfuse client to break any references
|
||||
self.logger.Langfuse = None
|
||||
# Delete logger instance to ensure complete cleanup
|
||||
del self.logger
|
||||
|
||||
# Restore global Langfuse client counter to prevent cross-test pollution
|
||||
|
|
@ -281,18 +247,6 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
Test that _log_langfuse_v2 correctly handles None values in the usage object
|
||||
by converting them to 0, preventing validation errors.
|
||||
"""
|
||||
# Reset the mock to ensure clean state; clear side_effect so return_value takes effect
|
||||
self.mock_langfuse_client.reset_mock(side_effect=True)
|
||||
self.mock_langfuse_trace.reset_mock(side_effect=True)
|
||||
self.mock_langfuse_generation.reset_mock(side_effect=True)
|
||||
|
||||
# Re-setup the trace and generation chain with clean state
|
||||
self.mock_langfuse_generation.trace_id = "test-trace-id"
|
||||
mock_span = MagicMock()
|
||||
mock_span.end = MagicMock()
|
||||
self.mock_langfuse_trace.span.return_value = mock_span
|
||||
self.mock_langfuse_trace.generation.return_value = self.mock_langfuse_generation
|
||||
|
||||
self.use_real_langfuse_client()
|
||||
|
||||
with (
|
||||
|
|
@ -611,9 +565,13 @@ class TestLangfuseUsageDetails(unittest.TestCase):
|
|||
debug_langfuse dumps request metadata into the trace as a second emit site.
|
||||
It must be sourced from the allowlisted payload too.
|
||||
"""
|
||||
dumped = self._drive_with_canary(extra_metadata={"debug_langfuse": True})["metadata_passed_to_litellm"]
|
||||
import json
|
||||
|
||||
self._drive_with_canary(extra_metadata={"debug_langfuse": True})
|
||||
dumped = json.loads(self.exported_generation().attributes["langfuse.trace.metadata.metadata_passed_to_litellm"])
|
||||
|
||||
assert "user_api_key_auth" not in dumped
|
||||
assert dumped["first_custom"] == "keep-first"
|
||||
assert self.CANARY not in self._emitted_payload_text()
|
||||
|
||||
def test_raw_request_metadata_reaches_the_emitted_blob_through_no_key(self):
|
||||
|
|
@ -1327,52 +1285,40 @@ def test_max_langfuse_clients_limit():
|
|||
litellm.initialized_langfuse_clients = original_initialized_langfuse_clients
|
||||
|
||||
|
||||
class _RecordingLangfuse:
|
||||
last_parameters: Optional[dict] = None
|
||||
|
||||
def __init__(self, environment=None, **parameters):
|
||||
type(self).last_parameters = {"environment": environment, **parameters}
|
||||
self.client = MagicMock()
|
||||
_UNREACHABLE_HOST: Final = "http://127.0.0.1:1"
|
||||
|
||||
|
||||
def _build_langfuse_logger(monkeypatch) -> LangFuseLogger:
|
||||
def _build_langfuse_logger(monkeypatch, **overrides) -> 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 build_langfuse_client resolves it; a real client spawns export threads
|
||||
return LangFuseLogger(
|
||||
langfuse_public_key="pk-lit5228",
|
||||
langfuse_secret="sk-lit5228",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
)
|
||||
return LangFuseLogger(
|
||||
**{
|
||||
"langfuse_public_key": "pk-lit5228",
|
||||
"langfuse_secret": "sk-lit5228",
|
||||
"langfuse_host": _UNREACHABLE_HOST,
|
||||
**overrides,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_langfuse_environment_is_passed_to_sdk_client(monkeypatch):
|
||||
monkeypatch.setenv("LANGFUSE_MOCK", "false")
|
||||
def _exported_environment(logger: LangFuseLogger):
|
||||
from langfuse import LangfuseOtelSpanAttributes
|
||||
|
||||
return logger.tracing.provider.resource.attributes.get(LangfuseOtelSpanAttributes.ENVIRONMENT)
|
||||
|
||||
|
||||
def test_langfuse_environment_lands_on_every_exported_span(monkeypatch):
|
||||
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 build_langfuse_client resolves it; a real client spawns export threads
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
langfuse_environment="staging",
|
||||
)
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-env", langfuse_environment="staging")
|
||||
assert logger.langfuse_environment == "staging"
|
||||
assert _RecordingLangfuse.last_parameters["environment"] == "staging"
|
||||
assert _exported_environment(logger) == "staging"
|
||||
|
||||
|
||||
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 build_langfuse_client resolves it; a real client spawns export threads
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
)
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-env")
|
||||
assert logger.langfuse_environment == "deployment-wide"
|
||||
assert _RecordingLangfuse.last_parameters["environment"] == "deployment-wide"
|
||||
assert _exported_environment(logger) == "deployment-wide"
|
||||
|
||||
|
||||
def test_dynamic_langfuse_environment_triggers_dynamic_logger():
|
||||
|
|
@ -1389,7 +1335,7 @@ def test_dynamic_langfuse_environment_triggers_dynamic_logger():
|
|||
assert config["langfuse_environment"] == "team-a-env"
|
||||
|
||||
|
||||
def test_langfuse_sdk_client_survives_httpx_cache_eviction(monkeypatch):
|
||||
def test_langfuse_rest_client_survives_httpx_cache_eviction(monkeypatch):
|
||||
import gc
|
||||
import weakref
|
||||
|
||||
|
|
@ -1399,21 +1345,20 @@ def test_langfuse_sdk_client_survives_httpx_cache_eviction(monkeypatch):
|
|||
|
||||
monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache())
|
||||
logger = _build_langfuse_logger(monkeypatch)
|
||||
sdk_client = _RecordingLangfuse.last_parameters["httpx_client"]
|
||||
|
||||
cached_handler = _get_httpx_client()
|
||||
handler_ref = weakref.ref(cached_handler)
|
||||
|
||||
assert sdk_client is logger.langfuse_client
|
||||
assert sdk_client is cached_handler.client
|
||||
assert logger.langfuse_client is cached_handler.client
|
||||
|
||||
litellm.in_memory_llm_clients_cache = LLMClientCache()
|
||||
del cached_handler
|
||||
gc.collect()
|
||||
|
||||
assert litellm.in_memory_llm_clients_cache.get_cache("httpx_client") is None
|
||||
assert handler_ref() is not None, "logger must keep the handler that owns the client it handed the SDK"
|
||||
assert not sdk_client.is_closed
|
||||
assert handler_ref() is not None, "logger must keep the handler that owns the client behind its REST API"
|
||||
assert not logger.langfuse_client.is_closed
|
||||
assert logger.api_client.auth_check() is False
|
||||
|
||||
|
||||
def test_langfuse_logger_reuses_the_shared_cached_client(monkeypatch):
|
||||
|
|
@ -1451,16 +1396,8 @@ def _steering_logger():
|
|||
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.api_client = build_langfuse_client(
|
||||
public_key="pk-steering-test", secret_key="sk-steering-test", base_url=_UNREACHABLE_HOST, httpx_client=None
|
||||
)
|
||||
logger.langfuse_sdk_version = installed_langfuse_version()
|
||||
return logger, exporter
|
||||
|
|
@ -1998,8 +1935,7 @@ def test_update_trace_keys_from_the_request_body_list_applies_when_enabled(monke
|
|||
assert span.attributes["langfuse.release"] == "v1.2.3"
|
||||
|
||||
|
||||
def test_update_trace_keys_trace_metadata_reaches_the_trace_not_just_the_generation(monkeypatch):
|
||||
"""v2 updated the trace object's metadata; v4 has to propagate it as a trace attribute."""
|
||||
def test_update_trace_keys_trace_metadata_reaches_the_trace_and_stays_off_the_generation(monkeypatch):
|
||||
rig = _steering_logger()
|
||||
|
||||
monkeypatch.setattr(litellm, "langfuse_enable_update_trace_keys", True)
|
||||
|
|
@ -2015,7 +1951,7 @@ 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" * 300
|
||||
assert span.attributes["langfuse.observation.metadata.step"] == 2
|
||||
assert "langfuse.observation.metadata.step" not in span.attributes
|
||||
|
||||
|
||||
def test_non_mapping_trace_metadata_does_not_lose_the_event():
|
||||
|
|
@ -2049,25 +1985,12 @@ def test_update_trace_keys_matches_whole_keys_not_substrings():
|
|||
|
||||
|
||||
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 build_langfuse_client resolves it; a real client spawns export threads
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
langfuse_environment=123, # non-string: must coerce, not crash
|
||||
)
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-env", langfuse_environment=123)
|
||||
assert logger.langfuse_environment == "123"
|
||||
|
||||
with pytest.raises(ValueError, match="langfuse_environment"):
|
||||
LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
langfuse_environment="Production",
|
||||
)
|
||||
_build_langfuse_logger(monkeypatch, langfuse_public_key="pk-env", langfuse_environment="Production")
|
||||
|
||||
|
||||
def test_langfuse_empty_environment_falls_back_and_is_not_dynamic(monkeypatch):
|
||||
|
|
@ -2077,15 +2000,7 @@ def test_langfuse_empty_environment_falls_back_and_is_not_dynamic(monkeypatch):
|
|||
monkeypatch.setenv("LANGFUSE_TRACING_ENVIRONMENT", "production")
|
||||
|
||||
# '' 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 build_langfuse_client resolves it; a real client spawns export threads
|
||||
logger = LangFuseLogger(
|
||||
langfuse_public_key="pk-env",
|
||||
langfuse_secret="sk-env",
|
||||
langfuse_host="https://test.langfuse.com",
|
||||
langfuse_environment="",
|
||||
)
|
||||
logger = _build_langfuse_logger(monkeypatch, langfuse_public_key="pk-env", langfuse_environment="")
|
||||
assert logger.langfuse_environment == "production"
|
||||
|
||||
# env-only params that add nothing do not select a dynamic logger
|
||||
|
|
|
|||
|
|
@ -24,10 +24,7 @@ class TestLangfuseInMemoryCache:
|
|||
|
||||
# Create a mock LangFuseLogger class
|
||||
class MockLangFuseLogger:
|
||||
def __init__(self):
|
||||
self.Langfuse = MagicMock()
|
||||
self.Langfuse.flush = MagicMock()
|
||||
self.Langfuse.shutdown = MagicMock()
|
||||
pass
|
||||
|
||||
mock_logger = MockLangFuseLogger()
|
||||
|
||||
|
|
@ -50,22 +47,30 @@ class TestLangfuseInMemoryCache:
|
|||
assert litellm.initialized_langfuse_clients == initial_count - 1
|
||||
|
||||
@patch("litellm.initialized_langfuse_clients", 3)
|
||||
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.
|
||||
|
||||
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.
|
||||
"""
|
||||
def test_evicted_logger_keeps_its_export_channel_and_api_client_alive(self):
|
||||
"""Export channels are shared per credential set and outlive the logger, so eviction only
|
||||
releases the initialized-client slot: prompts still resolve and the channel still flushes."""
|
||||
from litellm.integrations.langfuse.langfuse import LangFuseLogger
|
||||
from litellm.integrations.langfuse.langfuse_sdk import DiscardingSpanExporter, build_langfuse_tracing
|
||||
|
||||
logger = LangFuseLogger.__new__(LangFuseLogger)
|
||||
logger.Langfuse = MagicMock()
|
||||
logger.Langfuse.get_prompt.return_value = "prompt-after-eviction"
|
||||
logger.api_client = MagicMock()
|
||||
logger.api_client.get_prompt.return_value = "prompt-after-eviction"
|
||||
logger.tracing = build_langfuse_tracing(
|
||||
exporter=DiscardingSpanExporter(),
|
||||
environment=None,
|
||||
release=None,
|
||||
sample_rate=1.0,
|
||||
flush_at=512,
|
||||
flush_interval_millis=1000,
|
||||
)
|
||||
self.cache.cache_dict["test_key"] = logger
|
||||
self.cache.ttl_dict["test_key"] = time.time() + 100
|
||||
|
||||
self.cache._remove_key("test_key")
|
||||
|
||||
assert litellm.initialized_langfuse_clients == 2
|
||||
assert logger.Langfuse.get_prompt("greeting") == "prompt-after-eviction"
|
||||
logger.Langfuse.shutdown.assert_not_called()
|
||||
assert logger.api_client.get_prompt("greeting") == "prompt-after-eviction"
|
||||
with logger.tracing.tracer.start_as_current_span("still-open"):
|
||||
pass
|
||||
assert logger.tracing.flush(1000) is True
|
||||
|
|
|
|||
|
|
@ -81,35 +81,29 @@ class TestCallbackManagementEndpoints:
|
|||
# Setup test client
|
||||
client = TestClient(app)
|
||||
|
||||
# Initialize Langfuse logger and add to callbacks
|
||||
with patch("litellm.integrations.langfuse.langfuse.Langfuse") as mock_langfuse:
|
||||
# Mock the Langfuse client initialization
|
||||
mock_langfuse_client = MagicMock()
|
||||
mock_langfuse.return_value = mock_langfuse_client
|
||||
# Add string representation to callback lists (this is how the system typically works)
|
||||
litellm.success_callback.append("langfuse")
|
||||
litellm._async_success_callback.append("langfuse")
|
||||
|
||||
# Add string representation to callback lists (this is how the system typically works)
|
||||
litellm.success_callback.append("langfuse")
|
||||
litellm._async_success_callback.append("langfuse")
|
||||
# Make request to list callbacks endpoint
|
||||
response = client.get(
|
||||
"/callbacks/list", headers={"Authorization": "Bearer sk-1234"}
|
||||
)
|
||||
|
||||
# Make request to list callbacks endpoint
|
||||
response = client.get(
|
||||
"/callbacks/list", headers={"Authorization": "Bearer sk-1234"}
|
||||
)
|
||||
# Verify response
|
||||
assert response.status_code == 200
|
||||
|
||||
# Verify response
|
||||
assert response.status_code == 200
|
||||
response_data = response.json()
|
||||
|
||||
response_data = response.json()
|
||||
# Verify langfuse appears in success callbacks
|
||||
assert "langfuse" in response_data["success"]
|
||||
assert response_data["failure"] == []
|
||||
assert response_data["success_and_failure"] == []
|
||||
|
||||
# Verify langfuse appears in success callbacks
|
||||
assert "langfuse" in response_data["success"]
|
||||
assert response_data["failure"] == []
|
||||
assert response_data["success_and_failure"] == []
|
||||
|
||||
# Verify the response structure is correct
|
||||
assert isinstance(response_data["success"], list)
|
||||
assert isinstance(response_data["failure"], list)
|
||||
assert isinstance(response_data["success_and_failure"], list)
|
||||
# Verify the response structure is correct
|
||||
assert isinstance(response_data["success"], list)
|
||||
assert isinstance(response_data["failure"], list)
|
||||
assert isinstance(response_data["success_and_failure"], list)
|
||||
|
||||
def test_alist_callbacks_with_datadog_logger(self):
|
||||
"""Test /callbacks/list endpoint with DataDog logger configuration"""
|
||||
|
|
|
|||
|
|
@ -152,6 +152,40 @@ async def test_proxy_shutdown_flushes_every_langfuse_export_channel(monkeypatch)
|
|||
assert flushed.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_proxy_shutdown_flushes_langfuse_off_the_event_loop_and_logs_a_timeout(monkeypatch, caplog):
|
||||
"""The flush blocks on OTLP exports for up to its deadline, so it must run on a worker thread
|
||||
with the shutdown deadline, and a channel that misses it is reported instead of ignored."""
|
||||
import threading
|
||||
|
||||
from litellm.constants import LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS
|
||||
from litellm.integrations.langfuse import langfuse_sdk
|
||||
|
||||
ran_on = MagicMock()
|
||||
|
||||
def flushed(timeout_millis: int) -> bool:
|
||||
ran_on(threading.current_thread(), timeout_millis)
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(langfuse_sdk, "flush_langfuse_tracing", flushed)
|
||||
monkeypatch.setattr(ps, "prisma_client", None, raising=False)
|
||||
monkeypatch.setattr(ps, "jwt_handler", MagicMock(close=AsyncMock()), raising=False)
|
||||
monkeypatch.setattr(ps, "db_writer_client", None, raising=False)
|
||||
|
||||
import litellm
|
||||
|
||||
monkeypatch.setattr(litellm, "cache", None, raising=False)
|
||||
monkeypatch.setattr(litellm, "success_callback", [], raising=False)
|
||||
|
||||
with caplog.at_level("WARNING", logger="LiteLLM Proxy"):
|
||||
await proxy_shutdown_event()
|
||||
|
||||
(flush_thread, timeout_millis), _ = ran_on.call_args
|
||||
assert flush_thread is not threading.main_thread()
|
||||
assert timeout_millis == LANGFUSE_SHUTDOWN_FLUSH_TIMEOUT_MILLIS
|
||||
assert any("Langfuse export did not finish" in record.getMessage() for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_proxy_shutdown_drains_gateway_requests_before_disconnecting(monkeypatch):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue