mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-24 00:52:24 +00:00
fix(otel v2): map completions, images, speech, transcription and moderation output onto the Langfuse generation output (#42394)
* fix(otel v2): map completions, images, speech, transcription and moderation output onto the generation output Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(redaction): redact text completion choices in the standard logging payload Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(e2e): compare decoded generation output text and follow the live moderation verdict Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(otel v2): compare logged byte counts with the received media and move e2e schemas into models.py Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ci(e2e): keep the otel_v2 Langfuse output e2e file out of the stage-mirror gate it cannot run in Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: yucheng <yucheng@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
55e95c0279
commit
13b374873d
16 changed files with 738 additions and 61 deletions
1
.github/e2e-stack/select_tests.py
vendored
1
.github/e2e-stack/select_tests.py
vendored
|
|
@ -9,6 +9,7 @@ UNSUPPORTED: Final = re.compile(
|
|||
r"|^tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e\.py$"
|
||||
r"|^tests/e2e/batches/test_managed_files_enforcement_e2e\.py$"
|
||||
r"|^tests/e2e/guardrails/test_presidio_masking_e2e\.py$"
|
||||
r"|^tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e\.py$"
|
||||
)
|
||||
HARNESS: Final = re.compile(
|
||||
r"^tests/e2e/[A-Za-z0-9_.-]+\.(py|ini)$"
|
||||
|
|
|
|||
|
|
@ -427,7 +427,7 @@ class LLMCallSpanData:
|
|||
# plain ``.get`` — no repeated ``isinstance`` guards.
|
||||
raw_response: Final = payload.get("response")
|
||||
response: Final = cast(Mapping[str, object], raw_response if isinstance(raw_response, dict) else {})
|
||||
choices_out: Final = _dicts(response.get("choices")) or _responses_choices(response) or _ocr_choices(response)
|
||||
choices_out: Final = _output_choices(response)
|
||||
# ``finish_reasons`` is metadata, not content, so derive it from
|
||||
# ``choices_out`` before gating. The raw message/choice bodies are only
|
||||
# retained when content capture is enabled (see ``capture_span_content``);
|
||||
|
|
@ -752,20 +752,99 @@ def _responses_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
|||
return (choice,)
|
||||
|
||||
|
||||
def _ocr_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
markdowns: Final = tuple(
|
||||
text for page in _dicts(response.get("pages")) if (text := as_str(page.get("markdown"))) is not None
|
||||
def _output_choices(response: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
|
||||
"""The response output as chat-shaped choices; images and binary bodies become size summaries, never bytes."""
|
||||
return (
|
||||
_completion_choices(response)
|
||||
or _responses_choices(response)
|
||||
or _ocr_choices(response)
|
||||
or _transcription_choices(response)
|
||||
or _moderation_choices(response)
|
||||
or _image_choices(response)
|
||||
or _binary_choices(response)
|
||||
)
|
||||
if not markdowns:
|
||||
|
||||
|
||||
def _text_choice(content: str, finish_reason: str | None = None) -> _Choice:
|
||||
message: Final[_AssistantMessage] = {"role": "assistant", "content": content, "refusal": None, "tool_calls": None}
|
||||
return {"message": message, "finish_reason": finish_reason}
|
||||
|
||||
|
||||
def _joined_choice(parts: tuple[str, ...]) -> tuple[_Choice, ...]:
|
||||
return (_text_choice("\n\n".join(parts)),) if parts else ()
|
||||
|
||||
|
||||
def _completion_choices(response: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
|
||||
return tuple(
|
||||
_text_choice(text, as_str(choice.get("finish_reason")))
|
||||
if "message" not in choice and isinstance(text := choice.get("text"), str)
|
||||
else choice
|
||||
for choice in _dicts(response.get("choices"))
|
||||
)
|
||||
|
||||
|
||||
def _ocr_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
return _joined_choice(
|
||||
tuple(text for page in _dicts(response.get("pages")) if (text := as_str(page.get("markdown"))) is not None)
|
||||
)
|
||||
|
||||
|
||||
def _transcription_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
text: Final = response.get("text")
|
||||
return (_text_choice(text),) if isinstance(text, str) and text else ()
|
||||
|
||||
|
||||
def _moderation_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
return _joined_choice(
|
||||
tuple(
|
||||
_moderation_verdict(flagged, result.get("categories"))
|
||||
for result in _dicts(response.get("results"))
|
||||
if isinstance(flagged := result.get("flagged"), bool)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _moderation_verdict(flagged: bool, categories: object) -> str:
|
||||
if not flagged:
|
||||
return "not flagged"
|
||||
hits: Final = (
|
||||
tuple(name for name, hit in cast(Mapping[str, object], categories).items() if hit is True)
|
||||
if isinstance(categories, dict)
|
||||
else ()
|
||||
)
|
||||
return f"flagged: {', '.join(hits)}" if hits else "flagged"
|
||||
|
||||
|
||||
def _image_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
return _joined_choice(
|
||||
tuple(summary for item in _dicts(response.get("data")) if (summary := _image_summary(item)) is not None)
|
||||
)
|
||||
|
||||
|
||||
def _image_summary(item: Mapping[str, object]) -> str | None:
|
||||
location: Final = _image_location(item)
|
||||
if location is None:
|
||||
return None
|
||||
revised: Final = as_str(item.get("revised_prompt"))
|
||||
return f"{revised}\n{location}" if revised else location
|
||||
|
||||
|
||||
def _image_location(item: Mapping[str, object]) -> str | None:
|
||||
url: Final = as_str(item.get("url"))
|
||||
if url is not None:
|
||||
return url
|
||||
encoded: Final = item.get("b64_json")
|
||||
if not isinstance(encoded, str):
|
||||
return None
|
||||
return f"b64_json image ({len(encoded) * 3 // 4 - encoded[-2:].count('=')} bytes)"
|
||||
|
||||
|
||||
def _binary_choices(response: Mapping[str, object]) -> tuple[_Choice, ...]:
|
||||
size: Final = as_int(response.get("num_bytes"))
|
||||
if size is None:
|
||||
return ()
|
||||
message: Final[_AssistantMessage] = {
|
||||
"role": "assistant",
|
||||
"content": "\n\n".join(markdowns),
|
||||
"refusal": None,
|
||||
"tool_calls": None,
|
||||
}
|
||||
choice: Final[_Choice] = {"message": message, "finish_reason": None}
|
||||
return (choice,)
|
||||
content_type: Final = as_str(response.get("content_type"))
|
||||
return (_text_choice(f"{content_type} ({size} bytes)" if content_type else f"{size} bytes"),)
|
||||
|
||||
|
||||
def _responses_parts_text(parts: tuple[Mapping[str, object], ...], part_type: str, field: str) -> str | None:
|
||||
|
|
|
|||
|
|
@ -6277,6 +6277,8 @@ def _extract_response_obj_and_hidden_params(
|
|||
hidden_params = getattr(init_response_obj, "_hidden_params", None)
|
||||
elif isinstance(init_response_obj, dict):
|
||||
response_obj = init_response_obj
|
||||
elif isinstance(init_response_obj, HttpxBinaryResponseContent):
|
||||
response_obj = dict(init_response_obj.logging_summary())
|
||||
else:
|
||||
response_obj = {}
|
||||
|
||||
|
|
|
|||
|
|
@ -245,6 +245,8 @@ def _redact_model_response_dict_choices(choices, redacted_str: str):
|
|||
if "audio" in choice["delta"]:
|
||||
choice["delta"]["audio"] = None
|
||||
_redact_tool_calls_dict(choice["delta"])
|
||||
elif choice.get("text") is not None:
|
||||
choice["text"] = redacted_str
|
||||
else:
|
||||
_redact_choice_content(choice)
|
||||
|
||||
|
|
|
|||
|
|
@ -110,6 +110,14 @@ FileTypes = (
|
|||
EmbeddingInput = str | list[str]
|
||||
|
||||
|
||||
class BinaryResponseSummary(TypedDict):
|
||||
"""What logging keeps of a binary response (speech audio, file content): size and media type, never the bytes."""
|
||||
|
||||
object: ReadOnly[Literal["binary"]]
|
||||
content_type: ReadOnly[str | None]
|
||||
num_bytes: ReadOnly[int]
|
||||
|
||||
|
||||
class HttpxBinaryResponseContent(_HttpxBinaryResponseContent):
|
||||
_hidden_params: dict
|
||||
|
||||
|
|
@ -117,6 +125,19 @@ class HttpxBinaryResponseContent(_HttpxBinaryResponseContent):
|
|||
super().__init__(response)
|
||||
self._hidden_params = {} # mutable-ok: mutable-dict contract shared with ModelResponse logging consumers
|
||||
|
||||
def logging_summary(self) -> BinaryResponseSummary:
|
||||
return {
|
||||
"object": "binary",
|
||||
"content_type": self.response.headers.get("content-type"),
|
||||
"num_bytes": self._num_bytes(),
|
||||
}
|
||||
|
||||
def _num_bytes(self) -> int:
|
||||
try:
|
||||
return len(self.response.content)
|
||||
except httpx.ResponseNotRead:
|
||||
return self.response.num_bytes_downloaded
|
||||
|
||||
def set_response_cost(self, response_cost: float | None) -> None:
|
||||
if response_cost is None:
|
||||
self._hidden_params.pop("response_cost", None)
|
||||
|
|
|
|||
|
|
@ -212,6 +212,11 @@ def select_tests(changed: tuple[str, ...]) -> tuple[str, ...]:
|
|||
(("tests/e2e/batches/test_managed_files_enforcement_e2e.py",), ()),
|
||||
(("tests/e2e/guardrails/test_presidio_masking_e2e.py",), ()),
|
||||
(("tests/e2e/llm_translation/realtime/test_realtime_pipecat_audio_e2e.py",), ()),
|
||||
(("tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e.py",), ()),
|
||||
(
|
||||
("tests/e2e/logging/test_team_langfuse_callback_e2e.py",),
|
||||
("tests/e2e/logging/test_team_langfuse_callback_e2e.py",),
|
||||
),
|
||||
(
|
||||
("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
|
||||
("tests/e2e/llm_translation/realtime/test_realtime_e2e.py",),
|
||||
|
|
|
|||
|
|
@ -105,7 +105,7 @@ A couple of logging destinations are configured on the proxy rather than by the
|
|||
|
||||
### The pull request check
|
||||
|
||||
Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite and both JWT suites as canaries, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Keycloak, Jaeger, and TLS cluster-mode Valkey. Realm-only edits also trigger these canaries. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, and `guardrails/test_presidio_masking_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start. The Redis chaos test under `load/` needs a proxy it can pause the Redis of on the same host (`gateway/redis_chaos_ci_config.yml`), which `.github/workflows/test-e2e-redis-chaos.yml` boots, and which the Buildkite `e2e-redis-chaos` step in project-releaser runs co-located with Postgres and Valkey in one pod; it is deselected unless `E2E_REDIS_CHAOS` is set
|
||||
Every same-repository PR that adds, modifies, or renames a `tests/e2e/**/test_*.py` file runs those changed files three times. A change to the harness itself, meaning a root-level `tests/e2e/*.py` file or `pytest.ini`, `tests/e2e/gateway/`, `.github/e2e-stack/`, or the workflow, also runs the `access_control` suite and both JWT suites as canaries, because those files have no test of their own that exercises the stack. `.github/e2e-stack/select_tests.py` applies both rules. The stack config at `tests/e2e/gateway/stage_mirror_ci_config.yml` must declare every model the selected suites use; a missing one shows up as a failed test id in the public log. The suite's own single rerun for network errors and 5xx responses (see `pytest.ini`) applies on every pass, so a transport blip does not fail the check while a race inside a test still does. The stage-mirror stack has a control-plane backend, two gateways behind nginx, Postgres, Keycloak, Jaeger, and TLS cluster-mode Valkey. Realm-only edits also trigger these canaries. The stack exports every gateway address in `LITELLM_PROXY_REPLICA_URLS`, so model registration waits until each gateway lists the new model rather than whichever one the load balancer answered from. Documentation, deleted-file, and application-only changes do not start the stack or request environment approval. The `ui/`, `claude_code/`, and `load/` directories, `batches/test_managed_files_enforcement_e2e.py`, `llm_translation/realtime/test_realtime_pipecat_audio_e2e.py`, and `guardrails/test_presidio_masking_e2e.py` remain outside this check because they use separate tooling or need a differently configured stack: the pipecat audio suite skips itself at import time unless the NLTK `punkt_tab` data is installed, and the presidio suite fails without the analyzer and anonymizer services this stack does not start. `logging/test_otel_v2_langfuse_generation_output_e2e.py` is marked `otel_v2` and deselects itself unless `E2E_OTEL_V2` is set, because it needs a gateway booted with `LITELLM_OTEL_V2=true` and Langfuse credentials, neither of which this stack provides, so run it with `E2E_OTEL_V2=1` against a local OTel v2 proxy. The Redis chaos test under `load/` needs a proxy it can pause the Redis of on the same host (`gateway/redis_chaos_ci_config.yml`), which `.github/workflows/test-e2e-redis-chaos.yml` boots, and which the Buildkite `e2e-redis-chaos` step in project-releaser runs co-located with Postgres and Valkey in one pod; it is deselected unless `E2E_REDIS_CHAOS` is set
|
||||
|
||||
Every selected file must execute at least one passing test in each pass, and any test failure, collection error, or entirely skipped or deselected file fails the check. A file whose tests are all marked skip therefore cannot pass this check, so unskip at least one of them, or add the file to `UNSUPPORTED` in `select_tests.py` with the reason, before changing one. A failed pass stops the run. The public log prints pytest's one-line summary for each pass, including the rerun count, and names each failed or errored test as `classname::name`, so a retried network error or a failing test is visible without the raw output. The final `e2e-changed-tests` job succeeds only when no supported test files changed or the approved run completed all three passes. Fork PRs with selected tests fail this gate until a maintainer brings the reviewed change onto a same-repository branch
|
||||
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ from e2e_config import (
|
|||
FIXTURE_MODE_RAW,
|
||||
MANAGED_FILES_OPT_IN_ENV,
|
||||
MCP_OAUTH_LIVE_OPT_IN_ENV,
|
||||
OTEL_V2_OPT_IN_ENV,
|
||||
PROMPT_CACHING_OPT_IN_ENV,
|
||||
PROVIDER_EDGE_HOST_OPT_IN_ENV,
|
||||
PROXY_BASE_URL,
|
||||
|
|
@ -61,6 +62,7 @@ OPT_IN_MARKERS: Final = MappingProxyType(
|
|||
"cli_determinism": CLI_DETERMINISM_OPT_IN_ENV,
|
||||
"mcp_oauth_live": MCP_OAUTH_LIVE_OPT_IN_ENV,
|
||||
"provider_edge_host": PROVIDER_EDGE_HOST_OPT_IN_ENV,
|
||||
"otel_v2": OTEL_V2_OPT_IN_ENV,
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -150,6 +152,10 @@ def pytest_configure(config: pytest.Config) -> None:
|
|||
"provider_edge_host: routes provider traffic through the pytest host's edge in every fixture mode, so the "
|
||||
"gateway must reach the pytest host; deselected unless E2E_PROVIDER_EDGE_HOST_REACHABLE is set",
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers",
|
||||
"otel_v2: needs a proxy running with LITELLM_OTEL_V2=true; deselected unless E2E_OTEL_V2 is set",
|
||||
)
|
||||
|
||||
|
||||
def pytest_sessionstart(session: pytest.Session) -> None:
|
||||
|
|
|
|||
|
|
@ -147,6 +147,7 @@ REDIS_CHAOS_OPT_IN_ENV = "E2E_REDIS_CHAOS"
|
|||
CLI_DETERMINISM_OPT_IN_ENV = "E2E_CLI_DETERMINISM"
|
||||
MCP_OAUTH_LIVE_OPT_IN_ENV: Final = "E2E_MCP_OAUTH_LIVE"
|
||||
PROVIDER_EDGE_HOST_OPT_IN_ENV: Final = "E2E_PROVIDER_EDGE_HOST_REACHABLE"
|
||||
OTEL_V2_OPT_IN_ENV: Final = "E2E_OTEL_V2"
|
||||
ANOMALY_SESSIONS = int(os.environ.get("E2E_ANOMALY_SESSIONS", "6"))
|
||||
ANOMALY_TURNS_PER_SESSION = int(os.environ.get("E2E_ANOMALY_TURNS_PER_SESSION", "6"))
|
||||
ANOMALY_TURN_ATTEMPTS = int(os.environ.get("E2E_ANOMALY_TURN_ATTEMPTS", "3"))
|
||||
|
|
|
|||
|
|
@ -146,6 +146,29 @@ class LangfuseObservationList(BaseModel):
|
|||
data: list[LangfuseObservation] = []
|
||||
|
||||
|
||||
class LangfuseOtelMetadata(BaseModel):
|
||||
"""Langfuse stores every OTel span attribute under metadata.attributes."""
|
||||
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
attributes: dict[str, str] = {}
|
||||
|
||||
|
||||
def otel_attributes(obs: LangfuseObservation) -> dict[str, str]:
|
||||
try:
|
||||
return LangfuseOtelMetadata.model_validate(obs.metadata).attributes
|
||||
except ValidationError:
|
||||
return {}
|
||||
|
||||
|
||||
def is_otel_v2_generation(obs: LangfuseObservation, *, key_alias: str) -> bool:
|
||||
attributes = otel_attributes(obs)
|
||||
return (
|
||||
attributes.get("langfuse.observation.type") == "generation"
|
||||
and attributes.get("litellm.metadata.user_api_key_alias") == key_alias
|
||||
)
|
||||
|
||||
|
||||
class LangfuseListParams(BaseModel):
|
||||
model_config = ConfigDict(populate_by_name=True)
|
||||
|
||||
|
|
@ -630,6 +653,18 @@ class LoggingClient:
|
|||
time.sleep(POLL_INTERVAL)
|
||||
return last
|
||||
|
||||
def poll_langfuse_generation(
|
||||
self, creds: LangfuseCreds, *, key_alias: str, from_start_time: str
|
||||
) -> LangfuseObservation | None:
|
||||
"""The OTel v2 generation the proxy exported for one key alias since from_start_time."""
|
||||
deadline = time.monotonic() + POLL_TIMEOUT
|
||||
while time.monotonic() < deadline:
|
||||
for obs in self.list_langfuse_observations(creds, from_start_time=from_start_time):
|
||||
if is_otel_v2_generation(obs, key_alias=key_alias):
|
||||
return obs
|
||||
time.sleep(POLL_INTERVAL)
|
||||
return None
|
||||
|
||||
def poll_langfuse_trace_observations(
|
||||
self,
|
||||
creds: LangfuseCreds,
|
||||
|
|
|
|||
210
tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e.py
Normal file
210
tests/e2e/logging/test_otel_v2_langfuse_generation_output_e2e.py
Normal file
|
|
@ -0,0 +1,210 @@
|
|||
"""Live e2e: the OTel v2 Langfuse generation carries output for every non-chat endpoint (LIT-8309).
|
||||
|
||||
With LITELLM_OTEL_V2=true the proxy exports one generation per request to the
|
||||
team's Langfuse destination. Chat, Responses, embeddings and OCR already fill
|
||||
its output; this file pins the remaining five families. Each test registers a
|
||||
real OpenAI deployment, drives the endpoint through the shared transport, then
|
||||
reads the generation back from Langfuse and asserts its output reflects what
|
||||
the caller received: the completion text, the transcript, the moderation
|
||||
verdict, and for images and speech a bounded summary that never carries the
|
||||
raw base64 or audio bytes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from e2e_config import unique_marker
|
||||
from e2e_http import unwrap
|
||||
from lifecycle import ResourceManager
|
||||
from logging_client import LangfuseCreds, LangfuseObservation, LoggingClient, load_langfuse_creds
|
||||
from models import (
|
||||
CompletionBody,
|
||||
CompletionResponse,
|
||||
ImageGenerationBody,
|
||||
ImageGenerationResponse,
|
||||
LiteLLMParamsBody,
|
||||
ModerationBody,
|
||||
ModerationResponse,
|
||||
SpeechBody,
|
||||
TranscriptionForm,
|
||||
TranscriptionResponse,
|
||||
)
|
||||
from pydantic import BaseModel, TypeAdapter, ValidationError
|
||||
|
||||
pytestmark = [pytest.mark.e2e, pytest.mark.otel_v2]
|
||||
|
||||
WEATHER_WAV: Final = (
|
||||
Path(__file__).resolve().parent.parent / "llm_translation" / "realtime" / "fixtures" / "weather_question_24k.wav"
|
||||
)
|
||||
BOUNDED_OUTPUT_CHARS: Final = 1024
|
||||
|
||||
|
||||
class _OutputMessage(BaseModel):
|
||||
"""One assistant message of the Langfuse generation output; only the text is read."""
|
||||
|
||||
content: str = ""
|
||||
|
||||
|
||||
_OUTPUT_MESSAGES: Final = TypeAdapter(list[_OutputMessage])
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def langfuse_creds() -> LangfuseCreds:
|
||||
return load_langfuse_creds()
|
||||
|
||||
|
||||
def _langfuse_key(
|
||||
client: LoggingClient, creds: LangfuseCreds, resources: ResourceManager, params: LiteLLMParamsBody
|
||||
) -> tuple[str, str, str]:
|
||||
"""A model registered for this run plus a key on a team whose Langfuse callback is `creds`."""
|
||||
model: Final = f"e2e-otel-out-{unique_marker()}"
|
||||
model_id: Final = client.proxy.create_model(model, params)
|
||||
resources.defer(lambda: client.proxy.delete_model(model_id))
|
||||
team_id: Final = client.create_team(f"otel-out-team-{unique_marker()}", models=[model])
|
||||
resources.defer(lambda: client.delete_team(team_id))
|
||||
client.add_team_langfuse_callback(team_id, creds)
|
||||
alias: Final = f"otel-out-key-{unique_marker()}"
|
||||
key: Final = client.key_with_alias(alias, models=[model], team_id=team_id)
|
||||
resources.defer(lambda: client.delete_key(key))
|
||||
return model, key, alias
|
||||
|
||||
|
||||
def _generation(client: LoggingClient, creds: LangfuseCreds, *, alias: str, started: datetime) -> LangfuseObservation:
|
||||
since: Final = (started - timedelta(seconds=5)).isoformat()
|
||||
observation: Final = client.poll_langfuse_generation(creds, key_alias=alias, from_start_time=since)
|
||||
assert observation is not None, f"no OTel v2 generation reached Langfuse for key alias {alias!r}"
|
||||
return observation
|
||||
|
||||
|
||||
def _output_text(observation: LangfuseObservation) -> str:
|
||||
assert observation.output not in (None, "", [], {}), f"generation output is empty: {observation!r}"
|
||||
try:
|
||||
messages: Final = _OUTPUT_MESSAGES.validate_python(observation.output)
|
||||
except ValidationError:
|
||||
pytest.fail(f"generation output is not a list of assistant messages: {observation!r}")
|
||||
assert messages, f"generation output is empty: {observation!r}"
|
||||
return "\n".join(message.content for message in messages)
|
||||
|
||||
|
||||
def _openai(model: str) -> LiteLLMParamsBody:
|
||||
return LiteLLMParamsBody(model=model, api_key="os.environ/OPENAI_API_KEY")
|
||||
|
||||
|
||||
class TestOtelV2LangfuseGenerationOutput:
|
||||
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["completions"])
|
||||
def test_completions_output_is_the_completion_text(
|
||||
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key, alias = _langfuse_key(client, langfuse_creds, resources, _openai("openai/gpt-3.5-turbo-instruct"))
|
||||
started: Final = datetime.now(timezone.utc)
|
||||
response: Final = unwrap(
|
||||
client.proxy.transport.post(
|
||||
"/v1/completions",
|
||||
headers=client.proxy.transport.bearer(key),
|
||||
json=CompletionBody(model=model, prompt=f"Repeat exactly: {unique_marker()}", n=2),
|
||||
response_type=CompletionResponse,
|
||||
)
|
||||
)
|
||||
texts: Final = tuple(choice.text.strip() for choice in response.choices)
|
||||
assert len(texts) == 2 and all(texts), f"/v1/completions returned no text: {response!r}"
|
||||
|
||||
output: Final = _output_text(_generation(client, langfuse_creds, alias=alias, started=started))
|
||||
assert all(text in output for text in texts), (
|
||||
f"generation output lacks the completion texts {texts!r}: {output!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["images_generations"])
|
||||
def test_images_output_is_a_bounded_summary_without_base64(
|
||||
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key, alias = _langfuse_key(client, langfuse_creds, resources, _openai("openai/gpt-image-1-mini"))
|
||||
started: Final = datetime.now(timezone.utc)
|
||||
response: Final = unwrap(
|
||||
client.proxy.transport.post(
|
||||
"/v1/images/generations",
|
||||
headers=client.proxy.transport.bearer(key),
|
||||
json=ImageGenerationBody(model=model, prompt=f"a plain red square {unique_marker()}"),
|
||||
response_type=ImageGenerationResponse,
|
||||
timeout=180.0,
|
||||
)
|
||||
)
|
||||
assert response.data, f"/v1/images/generations returned no data: {response!r}"
|
||||
encoded: Final = response.data[0].b64_json or ""
|
||||
assert encoded, f"expected a b64_json image from gpt-image-1-mini: {response.data[0].url!r}"
|
||||
image_bytes: Final = len(base64.b64decode(encoded))
|
||||
|
||||
output: Final = _output_text(_generation(client, langfuse_creds, alias=alias, started=started))
|
||||
assert len(output) <= BOUNDED_OUTPUT_CHARS, f"image generation output is not bounded ({len(output)} chars)"
|
||||
assert encoded[:64] not in output, "image generation output leaks the raw base64 payload"
|
||||
assert output == f"b64_json image ({image_bytes} bytes)", (
|
||||
f"image generation output does not report the {image_bytes} decoded bytes: {output!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["audio_speech"])
|
||||
def test_speech_output_is_a_bounded_summary_without_audio_bytes(
|
||||
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key, alias = _langfuse_key(client, langfuse_creds, resources, _openai("openai/gpt-4o-mini-tts"))
|
||||
started: Final = datetime.now(timezone.utc)
|
||||
audio: Final = client.proxy.transport.stream_binary(
|
||||
"/v1/audio/speech",
|
||||
headers=client.proxy.transport.bearer(key),
|
||||
json=SpeechBody(model=model, input=f"hello {unique_marker()}"),
|
||||
)
|
||||
assert audio.ok and audio.total_bytes > 0, f"/v1/audio/speech returned no audio: {audio!r}"
|
||||
|
||||
output: Final = _output_text(_generation(client, langfuse_creds, alias=alias, started=started))
|
||||
assert len(output) <= BOUNDED_OUTPUT_CHARS, f"speech output is not bounded ({len(output)} chars)"
|
||||
assert output.endswith(f" ({audio.total_bytes} bytes)"), (
|
||||
f"speech output does not report the {audio.total_bytes} audio bytes the caller received: {output!r}"
|
||||
)
|
||||
|
||||
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["audio_transcriptions"])
|
||||
def test_transcription_output_is_the_transcript(
|
||||
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key, alias = _langfuse_key(client, langfuse_creds, resources, _openai("openai/gpt-4o-mini-transcribe"))
|
||||
started: Final = datetime.now(timezone.utc)
|
||||
response: Final = unwrap(
|
||||
client.proxy.transport.upload(
|
||||
"/v1/audio/transcriptions",
|
||||
headers=client.proxy.transport.bearer(key),
|
||||
form=TranscriptionForm(model=model),
|
||||
filename=WEATHER_WAV.name,
|
||||
content=WEATHER_WAV.read_bytes(),
|
||||
file_content_type="audio/wav",
|
||||
response_type=TranscriptionResponse,
|
||||
)
|
||||
)
|
||||
transcript: Final = response.text.strip()
|
||||
assert transcript, f"/v1/audio/transcriptions returned no text: {response!r}"
|
||||
|
||||
output: Final = _output_text(_generation(client, langfuse_creds, alias=alias, started=started))
|
||||
assert transcript in output, f"generation output lacks the transcript {transcript!r}: {output!r}"
|
||||
|
||||
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["moderations"])
|
||||
def test_moderations_output_is_the_verdict(
|
||||
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
|
||||
) -> None:
|
||||
model, key, alias = _langfuse_key(client, langfuse_creds, resources, _openai("openai/omni-moderation-latest"))
|
||||
started: Final = datetime.now(timezone.utc)
|
||||
response: Final = unwrap(
|
||||
client.proxy.transport.post(
|
||||
"/v1/moderations",
|
||||
headers=client.proxy.transport.bearer(key),
|
||||
json=ModerationBody(model=model, input=f"I will find you and hurt you badly {unique_marker()}"),
|
||||
response_type=ModerationResponse,
|
||||
)
|
||||
)
|
||||
assert response.results, f"/v1/moderations returned no results: {response!r}"
|
||||
verdict: Final = "flagged: " if response.results[0].flagged else "not flagged"
|
||||
|
||||
output: Final = _output_text(_generation(client, langfuse_creds, alias=alias, started=started))
|
||||
assert output.startswith(verdict), (
|
||||
f"generation output does not carry the moderation verdict {verdict!r}: {output!r}"
|
||||
)
|
||||
|
|
@ -760,6 +760,77 @@ class OcrResponse(BaseModel):
|
|||
pages: list[OcrPage] = []
|
||||
|
||||
|
||||
# ---------- completions ----------
|
||||
|
||||
|
||||
class CompletionBody(BaseModel):
|
||||
model: str
|
||||
prompt: str
|
||||
max_tokens: int = 8
|
||||
n: int = 1
|
||||
|
||||
|
||||
class CompletionChoice(BaseModel):
|
||||
text: str = ""
|
||||
|
||||
|
||||
class CompletionResponse(BaseModel):
|
||||
choices: list[CompletionChoice] = []
|
||||
|
||||
|
||||
# ---------- images ----------
|
||||
|
||||
|
||||
class ImageGenerationBody(BaseModel):
|
||||
model: str
|
||||
prompt: str
|
||||
n: int = 1
|
||||
size: str = "1024x1024"
|
||||
quality: str = "low"
|
||||
|
||||
|
||||
class ImageDatum(BaseModel):
|
||||
url: str | None = None
|
||||
b64_json: str | None = None
|
||||
|
||||
|
||||
class ImageGenerationResponse(BaseModel):
|
||||
data: list[ImageDatum] = []
|
||||
|
||||
|
||||
# ---------- audio ----------
|
||||
|
||||
|
||||
class SpeechBody(BaseModel):
|
||||
model: str
|
||||
input: str
|
||||
voice: str = "alloy"
|
||||
|
||||
|
||||
class TranscriptionForm(BaseModel):
|
||||
model: str
|
||||
|
||||
|
||||
class TranscriptionResponse(BaseModel):
|
||||
text: str = ""
|
||||
|
||||
|
||||
# ---------- moderations ----------
|
||||
|
||||
|
||||
class ModerationBody(BaseModel):
|
||||
model: str
|
||||
input: str
|
||||
|
||||
|
||||
class ModerationResult(BaseModel):
|
||||
flagged: bool
|
||||
|
||||
|
||||
class ModerationResponse(BaseModel):
|
||||
results: list[ModerationResult] = []
|
||||
|
||||
|
||||
# ---------- spend logs ----------
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -14,3 +14,4 @@ markers =
|
|||
redis_chaos: load test that pauses the proxy's Redis outright mid-run; needs a proxy booted from gateway/redis_chaos_ci_config.yml on the same host, and is deselected unless E2E_REDIS_CHAOS is set
|
||||
mcp_oauth_live: real Linear OAuth consent via a captured browser session; deselected unless E2E_MCP_OAUTH_LIVE is set
|
||||
provider_edge_host: routes provider traffic through the pytest host's edge in every fixture mode, so the gateway must reach the pytest host; deselected unless E2E_PROVIDER_EDGE_HOST_REACHABLE is set
|
||||
otel_v2: needs a proxy running with LITELLM_OTEL_V2=true; deselected unless E2E_OTEL_V2 is set
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ and the typed StandardLoggingPayload adapter. These need no OTel SDK."""
|
|||
import json
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -151,9 +152,7 @@ def test_llm_call_span_name():
|
|||
|
||||
def _all_constants(cls):
|
||||
return {
|
||||
getattr(cls, name)
|
||||
for name in vars(cls)
|
||||
if not name.startswith("__") and isinstance(getattr(cls, name), str)
|
||||
getattr(cls, name) for name in vars(cls) if not name.startswith("__") and isinstance(getattr(cls, name), str)
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -465,9 +464,7 @@ def test_mcp_tool_call_content_gated_off_by_default():
|
|||
off = MCPToolCallSpanData.from_standard_logging_payload(_mcp_payload())
|
||||
assert off.arguments_json is None and off.result_json is None
|
||||
|
||||
on = MCPToolCallSpanData.from_standard_logging_payload(
|
||||
_mcp_payload(), capture_content=True
|
||||
)
|
||||
on = MCPToolCallSpanData.from_standard_logging_payload(_mcp_payload(), capture_content=True)
|
||||
assert on.arguments_json is not None and '"Paris"' in on.arguments_json
|
||||
assert on.result_json is not None and "21" in on.result_json
|
||||
|
||||
|
|
@ -689,9 +686,7 @@ def test_content_capture_gated_off_by_default():
|
|||
payload = _sample_payload(
|
||||
messages=[{"role": "user", "content": "secret prompt"}],
|
||||
)
|
||||
payload["response"]["choices"] = [
|
||||
{"finish_reason": "stop", "message": {"role": "assistant", "content": "secret"}}
|
||||
]
|
||||
payload["response"]["choices"] = [{"finish_reason": "stop", "message": {"role": "assistant", "content": "secret"}}]
|
||||
data = LLMCallSpanData.from_standard_logging_payload(payload)
|
||||
assert data.messages_in == ()
|
||||
assert data.choices_out == ()
|
||||
|
|
@ -911,6 +906,221 @@ def test_ocr_pages_without_markdown_stay_empty():
|
|||
assert data.choices_out == ()
|
||||
|
||||
|
||||
def _assistant_choice(content: str, finish_reason: str | None = None) -> dict[str, object]:
|
||||
return {
|
||||
"message": {"role": "assistant", "content": content, "refusal": None, "tool_calls": None},
|
||||
"finish_reason": finish_reason,
|
||||
}
|
||||
|
||||
|
||||
def _route_payload(call_type: str, model: str, response: Mapping[str, object]) -> dict[str, object]:
|
||||
return _sample_payload(call_type=call_type, model=model, messages=None, response=response)
|
||||
|
||||
|
||||
def test_text_completion_choices_become_assistant_messages_in_choice_order() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload(
|
||||
"atext_completion",
|
||||
"gpt-3.5-turbo-instruct",
|
||||
{
|
||||
"id": "cmpl-1",
|
||||
"object": "text_completion",
|
||||
"choices": [
|
||||
{"index": 0, "text": " first", "finish_reason": "length", "logprobs": None},
|
||||
{"index": 1, "text": " second", "finish_reason": "stop", "logprobs": None},
|
||||
],
|
||||
},
|
||||
),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == (_assistant_choice(" first", "length"), _assistant_choice(" second", "stop"))
|
||||
assert data.finish_reasons == ("length", "stop")
|
||||
assert data.response_id == "cmpl-1"
|
||||
|
||||
|
||||
def test_text_completion_choices_follow_the_content_capture_gate_but_finish_reasons_do_not() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload(
|
||||
"atext_completion", "gpt-3.5-turbo-instruct", {"choices": [{"text": "x", "finish_reason": "stop"}]}
|
||||
)
|
||||
)
|
||||
|
||||
assert data.choices_out == ()
|
||||
assert data.finish_reasons == ("stop",)
|
||||
|
||||
|
||||
def test_chat_choices_with_a_message_are_passed_through_untouched_even_beside_a_stray_text_key() -> None:
|
||||
choice: Final = {
|
||||
"index": 0,
|
||||
"finish_reason": "stop",
|
||||
"text": "no",
|
||||
"message": {"role": "assistant", "content": "chat"},
|
||||
}
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_sample_payload(response={"choices": [choice]}), capture_content=True
|
||||
)
|
||||
|
||||
assert data.choices_out == (choice,)
|
||||
|
||||
|
||||
def test_transcription_text_becomes_one_assistant_choice() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("atranscription", "gpt-4o-mini-transcribe", {"text": "What is the weather like?", "task": "x"}),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == (_assistant_choice("What is the weather like?"),)
|
||||
assert data.finish_reasons == ()
|
||||
|
||||
|
||||
def test_empty_transcription_text_stays_empty() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("atranscription", "gpt-4o-mini-transcribe", {"text": ""}), capture_content=True
|
||||
)
|
||||
|
||||
assert data.choices_out == ()
|
||||
|
||||
|
||||
def test_moderation_results_become_one_verdict_per_input_naming_the_hit_categories() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload(
|
||||
"amoderation",
|
||||
"omni-moderation-latest",
|
||||
{
|
||||
"id": "modr-1",
|
||||
"results": [
|
||||
{
|
||||
"flagged": True,
|
||||
"categories": {"harassment": False, "violence": True, "self-harm": True},
|
||||
"category_scores": {"harassment": 0.01, "violence": 0.98, "self-harm": 0.7},
|
||||
},
|
||||
{"flagged": False, "categories": {"violence": False}},
|
||||
{"flagged": True},
|
||||
],
|
||||
},
|
||||
),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == (_assistant_choice("flagged: violence, self-harm\n\nnot flagged\n\nflagged"),)
|
||||
assert data.response_id == "modr-1"
|
||||
|
||||
|
||||
def test_moderation_output_follows_the_content_capture_gate() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("amoderation", "omni-moderation-latest", {"results": [{"flagged": True}]})
|
||||
)
|
||||
|
||||
assert data.choices_out == ()
|
||||
|
||||
|
||||
def test_moderation_results_without_a_verdict_produce_no_output() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("amoderation", "omni-moderation-latest", {"results": [{"categories": {"violence": True}}]}),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == ()
|
||||
|
||||
|
||||
def test_image_data_becomes_a_size_summary_and_never_carries_the_base64_payload() -> None:
|
||||
encoded: Final = "QUJDRA=="
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload(
|
||||
"aimage_generation",
|
||||
"gpt-image-1-mini",
|
||||
{
|
||||
"created": 1,
|
||||
"data": [
|
||||
{"b64_json": encoded, "revised_prompt": "a red bicycle"},
|
||||
{"url": "https://images.example/cat.png"},
|
||||
{"b64_json": "QUJDREVGR0g="},
|
||||
],
|
||||
},
|
||||
),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == (
|
||||
_assistant_choice(
|
||||
"a red bicycle\nb64_json image (4 bytes)\n\nhttps://images.example/cat.png\n\nb64_json image (8 bytes)"
|
||||
),
|
||||
)
|
||||
assert encoded not in json.dumps(data.choices_out)
|
||||
|
||||
|
||||
def test_image_data_without_a_url_or_payload_stays_empty_and_embeddings_are_not_images() -> None:
|
||||
images: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("aimage_generation", "gpt-image-1-mini", {"data": [{"revised_prompt": "x"}]}),
|
||||
capture_content=True,
|
||||
)
|
||||
embeddings: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_embedding_payload([[0.1, 0.2]]), capture_content=True
|
||||
)
|
||||
|
||||
assert images.choices_out == ()
|
||||
assert embeddings.choices_out == ()
|
||||
|
||||
|
||||
def test_speech_summary_becomes_a_media_type_and_byte_count_choice() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload(
|
||||
"aspeech", "gpt-4o-mini-tts", {"object": "binary", "content_type": "audio/mpeg", "num_bytes": 48210}
|
||||
),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == (_assistant_choice("audio/mpeg (48210 bytes)"),)
|
||||
|
||||
|
||||
def test_speech_summary_without_a_media_type_is_the_byte_count_and_follows_the_capture_gate() -> None:
|
||||
response: Final = {"object": "binary", "content_type": None, "num_bytes": 7}
|
||||
shown: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("aspeech", "gpt-4o-mini-tts", response), capture_content=True
|
||||
)
|
||||
gated: Final = LLMCallSpanData.from_standard_logging_payload(_route_payload("aspeech", "gpt-4o-mini-tts", response))
|
||||
|
||||
assert shown.choices_out == (_assistant_choice("7 bytes"),)
|
||||
assert gated.choices_out == ()
|
||||
|
||||
|
||||
def test_speech_response_without_a_byte_count_produces_no_output() -> None:
|
||||
data: Final = LLMCallSpanData.from_standard_logging_payload(
|
||||
_route_payload("aspeech", "gpt-4o-mini-tts", {"object": "binary", "content_type": "audio/mpeg"}),
|
||||
capture_content=True,
|
||||
)
|
||||
|
||||
assert data.choices_out == ()
|
||||
|
||||
|
||||
def test_speech_binary_response_is_logged_as_its_summary_not_dropped() -> None:
|
||||
import httpx
|
||||
|
||||
from litellm.litellm_core_utils.litellm_logging import _extract_response_obj_and_hidden_params
|
||||
from litellm.types.llms.openai import HttpxBinaryResponseContent
|
||||
|
||||
raw: Final = httpx.Response(200, headers={"content-type": "audio/mpeg"}, content=b"\x00" * 1234)
|
||||
response_obj, hidden_params = _extract_response_obj_and_hidden_params(HttpxBinaryResponseContent(raw), None)
|
||||
|
||||
assert response_obj == {"object": "binary", "content_type": "audio/mpeg", "num_bytes": 1234}
|
||||
assert hidden_params is None
|
||||
|
||||
|
||||
def test_speech_binary_response_still_streaming_reports_the_bytes_downloaded_so_far() -> None:
|
||||
import httpx
|
||||
|
||||
from litellm.types.llms.openai import HttpxBinaryResponseContent
|
||||
|
||||
unread: Final = httpx.Response(200, stream=httpx.ByteStream(b"\x00" * 10))
|
||||
|
||||
assert HttpxBinaryResponseContent(unread).logging_summary() == {
|
||||
"object": "binary",
|
||||
"content_type": None,
|
||||
"num_bytes": 0,
|
||||
}
|
||||
|
||||
|
||||
def test_request_identity_prefers_canonical_team_keys():
|
||||
from litellm.integrations.otel.model.payloads import RequestIdentity
|
||||
|
||||
|
|
@ -931,9 +1141,7 @@ def test_request_identity_prefers_canonical_team_keys():
|
|||
def test_request_identity_falls_back_to_legacy_team_keys():
|
||||
from litellm.integrations.otel.model.payloads import RequestIdentity
|
||||
|
||||
payload = _sample_payload(
|
||||
metadata={"team_id": "legacy-team", "team_alias": "legacy"}
|
||||
)
|
||||
payload = _sample_payload(metadata={"team_id": "legacy-team", "team_alias": "legacy"})
|
||||
ident = RequestIdentity.from_payload(payload)
|
||||
assert ident.team_id == "legacy-team"
|
||||
assert ident.team_alias == "legacy"
|
||||
|
|
@ -952,7 +1160,10 @@ def test_request_identity_falls_back_to_legacy_team_keys():
|
|||
},
|
||||
"from-header",
|
||||
),
|
||||
({"proxy_server_request": {"headers": {"langfuse_trace_name": ""}}, "metadata": {"trace_name": "body"}}, "body"),
|
||||
(
|
||||
{"proxy_server_request": {"headers": {"langfuse_trace_name": ""}}, "metadata": {"trace_name": "body"}},
|
||||
"body",
|
||||
),
|
||||
({"proxy_server_request": {"headers": {}}, "metadata": {"user_api_key_team_id": "t1"}}, None),
|
||||
({}, None),
|
||||
],
|
||||
|
|
@ -1002,7 +1213,15 @@ def test_caller_trace_name_prefers_the_langfuse_header_over_body_metadata(reques
|
|||
),
|
||||
({}, TraceControls()),
|
||||
],
|
||||
ids=["body", "headers-beat-body", "anthropic-body", "non-string-tags-dropped", "scalar-coercion", "mutation-controls-ignored", "empty"],
|
||||
ids=[
|
||||
"body",
|
||||
"headers-beat-body",
|
||||
"anthropic-body",
|
||||
"non-string-tags-dropped",
|
||||
"scalar-coercion",
|
||||
"mutation-controls-ignored",
|
||||
"empty",
|
||||
],
|
||||
)
|
||||
def test_caller_trace_controls_carry_user_session_and_tags(request_data, expected):
|
||||
assert caller_trace_controls({"litellm_params": request_data}) == expected
|
||||
|
|
@ -1168,9 +1387,7 @@ def test_content_capture_opt_in_retains_bodies():
|
|||
payload = _sample_payload(
|
||||
messages=[{"role": "user", "content": "secret prompt"}],
|
||||
)
|
||||
payload["response"]["choices"] = [
|
||||
{"finish_reason": "stop", "message": {"role": "assistant", "content": "hi"}}
|
||||
]
|
||||
payload["response"]["choices"] = [{"finish_reason": "stop", "message": {"role": "assistant", "content": "hi"}}]
|
||||
data = LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True)
|
||||
assert data.messages_in and data.messages_in[0]["content"] == "secret prompt"
|
||||
assert data.choices_out and data.choices_out[0]["message"]["content"] == "hi"
|
||||
|
|
@ -1187,41 +1404,17 @@ def test_capture_span_content_resolves_modes():
|
|||
|
||||
# default (no_content) → off
|
||||
assert OpenTelemetryV2Config().capture_span_content is False
|
||||
assert OpenTelemetryV2Config(capture_message_content=CaptureMessageContent.SPAN_ONLY).capture_span_content is True
|
||||
assert (
|
||||
OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.SPAN_ONLY
|
||||
).capture_span_content
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.SPAN_AND_EVENT
|
||||
).capture_span_content
|
||||
is True
|
||||
OpenTelemetryV2Config(capture_message_content=CaptureMessageContent.SPAN_AND_EVENT).capture_span_content is True
|
||||
)
|
||||
# event-only does not authorize span-attribute content
|
||||
assert (
|
||||
OpenTelemetryV2Config(
|
||||
capture_message_content=CaptureMessageContent.EVENT_ONLY
|
||||
).capture_span_content
|
||||
is False
|
||||
)
|
||||
assert OpenTelemetryV2Config(capture_message_content=CaptureMessageContent.EVENT_ONLY).capture_span_content is False
|
||||
# V1 accepted UPPER_SNAKE_CASE; the env value is case-insensitive so an
|
||||
# operator carrying ``SPAN_AND_EVENT`` forward still enables capture.
|
||||
assert (
|
||||
OpenTelemetryV2Config(
|
||||
capture_message_content="SPAN_AND_EVENT"
|
||||
).capture_span_content
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
OpenTelemetryV2Config(capture_message_content="SPAN_ONLY").capture_span_content
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
OpenTelemetryV2Config(capture_message_content="NO_CONTENT").capture_span_content
|
||||
is False
|
||||
)
|
||||
assert OpenTelemetryV2Config(capture_message_content="SPAN_AND_EVENT").capture_span_content is True
|
||||
assert OpenTelemetryV2Config(capture_message_content="SPAN_ONLY").capture_span_content is True
|
||||
assert OpenTelemetryV2Config(capture_message_content="NO_CONTENT").capture_span_content is False
|
||||
|
||||
|
||||
def test_capture_message_content_normalizer_only_touches_strings():
|
||||
|
|
|
|||
|
|
@ -6,6 +6,8 @@ backends, so one trace lights up every configured destination.
|
|||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -249,6 +251,34 @@ def test_langfuse_mapper_renders_an_ocr_call_with_the_page_markdown_as_output():
|
|||
assert attrs["langfuse.observation.type"] == "generation"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("call_type", "response", "expected_content"),
|
||||
[
|
||||
("atext_completion", {"choices": [{"text": "Paris.", "finish_reason": "stop"}]}, "Paris."),
|
||||
("atranscription", {"text": "What is the weather like?"}, "What is the weather like?"),
|
||||
("amoderation", {"results": [{"flagged": True, "categories": {"violence": True}}]}, "flagged: violence"),
|
||||
("aimage_generation", {"data": [{"b64_json": "QUJDRA=="}]}, "b64_json image (4 bytes)"),
|
||||
("aspeech", {"object": "binary", "content_type": "audio/mpeg", "num_bytes": 9}, "audio/mpeg (9 bytes)"),
|
||||
],
|
||||
)
|
||||
def test_langfuse_mapper_renders_every_non_chat_route_output_as_an_assistant_message(
|
||||
call_type: str, response: Mapping[str, object], expected_content: str
|
||||
) -> None:
|
||||
payload: Final[dict[str, object]] = {
|
||||
"call_type": call_type,
|
||||
"custom_llm_provider": "openai",
|
||||
"model": "m",
|
||||
"messages": None,
|
||||
"response": response,
|
||||
}
|
||||
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True))
|
||||
|
||||
assert json.loads(attrs["langfuse.observation.output"]) == [
|
||||
{"role": "assistant", "content": expected_content, "refusal": None, "tool_calls": None}
|
||||
]
|
||||
assert attrs["langfuse.observation.type"] == "generation"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Weave
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
|
|
|||
|
|
@ -304,6 +304,26 @@ class TestPerformRedaction:
|
|||
assert delta["thinking_blocks"] is None
|
||||
assert delta["audio"] is None
|
||||
|
||||
def test_redacts_text_completion_choices_in_standard_logging_object(self):
|
||||
details = {
|
||||
"standard_logging_object": {
|
||||
"response": {
|
||||
"object": "text_completion",
|
||||
"choices": [
|
||||
{"text": " Paris.", "finish_reason": "stop", "index": 0},
|
||||
{"text": "\n\nBlue", "finish_reason": "length", "index": 1},
|
||||
],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
perform_redaction(details, None)
|
||||
|
||||
assert details["standard_logging_object"]["response"]["choices"] == [
|
||||
{"text": "redacted-by-litellm", "finish_reason": "stop", "index": 0},
|
||||
{"text": "redacted-by-litellm", "finish_reason": "length", "index": 1},
|
||||
]
|
||||
|
||||
def test_redacts_object_choices_inside_model_response_dict(self):
|
||||
result = {
|
||||
"choices": [
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue