mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(langfuse_otel): name the OTel v2 request root once, from the request's own call
The v2 Langfuse logger re-derived langfuse.trace.name on the request root for every LLM call in the request. A post_call llm_as_a_judge guardrail's judge call replaced the caller's trace_name on the root with litellm-acompletion, and with another v2 callback (otel, arize_phoenix) listed before langfuse_otel that vendor's root span gained langfuse.trace.name on every request. The derived name now fills only an unnamed root this logger's provider started, and never comes from a call litellm makes for its own features. A caller trace_name still names the root on every call, as on main.
This commit is contained in:
parent
23fd6ad3ad
commit
84f2f28f9e
3 changed files with 203 additions and 5 deletions
|
|
@ -1,19 +1,25 @@
|
|||
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping
|
||||
from typing import TYPE_CHECKING, Final
|
||||
|
||||
from opentelemetry.sdk.trace import ReadableSpan
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.integrations.otel.logger import OpenTelemetryV2
|
||||
from litellm.integrations.otel.mappers.langfuse import (
|
||||
LANGFUSE_OBSERVATION_INPUT,
|
||||
LANGFUSE_OBSERVATION_OUTPUT,
|
||||
LANGFUSE_TRACE_NAME,
|
||||
LangfuseMapper,
|
||||
)
|
||||
from litellm.integrations.otel.model.metadata import internal_call_origin
|
||||
from litellm.integrations.otel.model.request_io import request_input, response_output, stream_output
|
||||
from litellm.integrations.otel.model.trace_controls import caller_trace_controls
|
||||
from litellm.integrations.otel.model.utils import as_str
|
||||
from litellm.integrations.otel.plumbing.context import request_root_span
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span
|
||||
|
||||
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
|
||||
from litellm.types.utils import ModelResponseStream
|
||||
|
||||
|
|
@ -28,11 +34,23 @@ class LangfuseOpenTelemetryV2(OpenTelemetryV2):
|
|||
root.set_attributes(
|
||||
LangfuseMapper.trace_attributes(
|
||||
caller_trace_controls(kwargs),
|
||||
as_str(kwargs.get("call_type")) if getattr(root, "parent", None) is None else None,
|
||||
as_str(kwargs.get("call_type")) if self._derives_trace_name(root, kwargs) else None,
|
||||
)
|
||||
)
|
||||
super().log_pre_api_call(model, messages, kwargs)
|
||||
|
||||
def _derives_trace_name(self, root: "Span", kwargs: Mapping[str, object]) -> bool:
|
||||
"""Whether this call may name the request root after its call type: the call is not one litellm made for
|
||||
its own features, this logger's provider started the root with no parent, and neither the caller nor an
|
||||
earlier call in the request has named it."""
|
||||
return (
|
||||
internal_call_origin(None, kwargs) is None
|
||||
and isinstance(root, ReadableSpan)
|
||||
and root.parent is None
|
||||
and root.resource is self.tracer_provider.resource
|
||||
and LANGFUSE_TRACE_NAME not in (root.attributes or {})
|
||||
)
|
||||
|
||||
|
||||
class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
|
||||
"""Stamps the request's input and output on the root observation while it is still recording.
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import json
|
|||
import threading
|
||||
import uuid
|
||||
from itertools import product
|
||||
from collections.abc import Iterator, Mapping
|
||||
from collections.abc import Iterator, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
|
@ -28,6 +28,8 @@ SURFACE_CALL_TYPES: Final = {
|
|||
"responses_stream": "aresponses",
|
||||
}
|
||||
ANTHROPIC_MODEL: Final = "anthropic/claude-sonnet-4-5-20250929"
|
||||
JUDGE_MODEL: Final = "lit8281-judge-model"
|
||||
JUDGE_GUARDRAIL: Final = "lit8281-judge"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -123,6 +125,32 @@ class LangfuseSink:
|
|||
return eventually(roots, lambda found: len(found) == 1, seconds=70)[0]
|
||||
|
||||
|
||||
class OtlpSink:
|
||||
"""A generic OTLP endpoint: keeps every span it receives."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.lock: Final = threading.Lock()
|
||||
self.spans: Final[list[SpanRecord]] = [] # mutable-ok: spans accumulate as the exporter delivers them
|
||||
|
||||
def respond(self, request: Request) -> Reply:
|
||||
if request.target.endswith("/v1/traces"):
|
||||
records: Final = _span_records(request.body)
|
||||
with self.lock:
|
||||
self.spans.extend(records)
|
||||
return Reply(body=b"")
|
||||
|
||||
def root_of_request(self, marker: str) -> SpanRecord:
|
||||
"""The root span of the one trace whose spans carry the marker."""
|
||||
|
||||
def roots() -> tuple[SpanRecord, ...]:
|
||||
with self.lock:
|
||||
snapshot: Final = tuple(self.spans)
|
||||
traces: Final = {record.trace_id for record in snapshot if marker in str(tuple(record.attributes.values()))}
|
||||
return tuple(record for record in snapshot if record.trace_id in traces and not record.parent_span_id)
|
||||
|
||||
return eventually(roots, lambda found: len(found) == 1, seconds=70)[0]
|
||||
|
||||
|
||||
def _chat_completion(marker: str, text: str = "langfuse-otel scripted reply") -> bytes:
|
||||
return json.dumps(
|
||||
{
|
||||
|
|
@ -152,6 +180,8 @@ def _provider(request: Request) -> Reply:
|
|||
}
|
||||
).encode()
|
||||
)
|
||||
if body.get("response_format") == {"type": "json_object"}:
|
||||
return Reply(body=_chat_completion("lit8281-judge-verdict", json.dumps({"overall_score": 100})))
|
||||
messages: Final = body.get("messages")
|
||||
first: Final = messages[0]["content"] if isinstance(messages, list) and messages else None
|
||||
if isinstance(first, str) and first.startswith("fail-"):
|
||||
|
|
@ -159,10 +189,33 @@ def _provider(request: Request) -> Reply:
|
|||
return surface_reply(request)
|
||||
|
||||
|
||||
def _langfuse_config(directory: Path, name: str, settings: Mapping[str, object]) -> Path:
|
||||
def _langfuse_config(directory: Path, name: str, settings: Mapping[str, object], provider_url: str) -> Path:
|
||||
"""The integration config with `langfuse_otel`, plus an opt-in post_call `llm_as_a_judge` guardrail whose judge
|
||||
model is served by the scripted provider."""
|
||||
base: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
|
||||
judge: Final = {
|
||||
"model_name": JUDGE_MODEL,
|
||||
"litellm_params": {
|
||||
"model": "openai/gpt-4o-mini",
|
||||
"api_base": provider_url + "/v1",
|
||||
"api_key": "synthetic-provider-key",
|
||||
},
|
||||
}
|
||||
config: Final = {
|
||||
**base,
|
||||
"model_list": [*base["model_list"], judge],
|
||||
"guardrails": [
|
||||
{
|
||||
"guardrail_name": JUDGE_GUARDRAIL,
|
||||
"litellm_params": {
|
||||
"guardrail": "llm_as_a_judge",
|
||||
"mode": "post_call",
|
||||
"judge_model": JUDGE_MODEL,
|
||||
"on_failure": "log",
|
||||
"criteria": [{"name": "helpful", "description": "The reply answers the user", "weight": 100}],
|
||||
},
|
||||
}
|
||||
],
|
||||
"litellm_settings": {**base["litellm_settings"], "callbacks": ["langfuse_otel"], **settings},
|
||||
}
|
||||
path: Final = directory / f"{name}.yaml"
|
||||
|
|
@ -198,7 +251,7 @@ def _rig(
|
|||
"LANGFUSE_HOST": destination.url,
|
||||
**environment,
|
||||
},
|
||||
config=_langfuse_config(directory, name, settings),
|
||||
config=_langfuse_config(directory, name, settings, provider.url),
|
||||
workers=2,
|
||||
) as proxy,
|
||||
proxy.scenario() as scenario,
|
||||
|
|
@ -237,8 +290,35 @@ def v2_llm_only(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
|
|||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SharedRig:
|
||||
rig: Rig
|
||||
otel: OtlpSink
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def v2_after_otel(tmp_path_factory: pytest.TempPathFactory) -> Iterator[SharedRig]:
|
||||
"""OTel v2 with the generic `otel` callback listed before `langfuse_otel`, so the request root belongs to the
|
||||
generic exporter."""
|
||||
otel: Final = OtlpSink()
|
||||
with wire_server(otel.respond) as destination:
|
||||
environment: Final = {
|
||||
**V2_ENVIRONMENT,
|
||||
"OTEL_EXPORTER": "otlp_http",
|
||||
"OTEL_ENDPOINT": destination.url + "/v1/traces",
|
||||
}
|
||||
for rig in _rig(
|
||||
tmp_path_factory, "langfuse-v2-after-otel", environment, {"callbacks": ["otel", "langfuse_otel"]}
|
||||
):
|
||||
yield SharedRig(rig, otel)
|
||||
|
||||
|
||||
def _chat(
|
||||
rig: Rig, content: str, metadata: Mapping[str, object] | None = None, headers: Mapping[str, str] | None = None
|
||||
rig: Rig,
|
||||
content: str,
|
||||
metadata: Mapping[str, object] | None = None,
|
||||
headers: Mapping[str, str] | None = None,
|
||||
guardrails: Sequence[str] = (),
|
||||
) -> str:
|
||||
response: Final = rig.proxy.request(
|
||||
"POST",
|
||||
|
|
@ -247,6 +327,7 @@ def _chat(
|
|||
"model": rig.openai_model,
|
||||
"messages": [{"role": "user", "content": content}],
|
||||
**({"metadata": dict(metadata)} if metadata is not None else {}),
|
||||
**({"guardrails": list(guardrails)} if guardrails else {}),
|
||||
},
|
||||
key=rig.key,
|
||||
headers=headers,
|
||||
|
|
@ -478,6 +559,33 @@ def test_otel_v2_inbound_traceparent_leaves_the_trace_name_to_the_caller(v2: Rig
|
|||
assert not set(_trace_fields(spans)), spans
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("metadata", "headers"),
|
||||
[({"trace_name": "caller-trace"}, None), (None, {"langfuse_trace_name": "caller-trace"})],
|
||||
ids=["body", "header"],
|
||||
)
|
||||
def test_otel_v2_post_call_judge_keeps_the_caller_trace_name_on_the_root(
|
||||
v2: Rig, metadata: Mapping[str, object] | None, headers: Mapping[str, str] | None
|
||||
) -> None:
|
||||
marker: Final = "lit8281-v2-judge-" + uuid.uuid4().hex
|
||||
_chat(v2, marker, metadata=metadata, headers=headers, guardrails=[JUDGE_GUARDRAIL])
|
||||
generations: Final = eventually(lambda: v2.sink.matching(marker), lambda found: len(found) == 2, seconds=70)
|
||||
assert len({generation.trace_id for generation in generations}) == 1, generations
|
||||
root: Final = v2.sink.root_of(generations[0])
|
||||
assert root.attributes.get("langfuse.trace.name") == "caller-trace", root.attributes
|
||||
|
||||
|
||||
def test_otel_v2_generic_otel_root_gets_no_derived_langfuse_trace_name(v2_after_otel: SharedRig) -> None:
|
||||
derived: Final = "lit8281-v2-after-otel-" + uuid.uuid4().hex
|
||||
caller: Final = "lit8281-v2-after-otel-caller-" + uuid.uuid4().hex
|
||||
_chat(v2_after_otel.rig, derived)
|
||||
_chat(v2_after_otel.rig, caller, metadata={"trace_name": "caller-trace"})
|
||||
root: Final = v2_after_otel.otel.root_of_request(derived)
|
||||
assert "langfuse.trace.name" not in root.attributes, root.attributes
|
||||
caller_root: Final = v2_after_otel.otel.root_of_request(caller)
|
||||
assert caller_root.attributes.get("langfuse.trace.name") == "caller-trace", caller_root.attributes
|
||||
|
||||
|
||||
def test_otel_v2_llm_only_root_generation_keeps_its_span_name_as_the_trace_name(v2_llm_only: Rig) -> None:
|
||||
marker: Final = "lit8281-v2-llm-only-" + uuid.uuid4().hex
|
||||
_chat(v2_llm_only, marker)
|
||||
|
|
|
|||
|
|
@ -437,6 +437,78 @@ def test_named_request_under_an_inbound_traceparent_keeps_the_caller_trace_name(
|
|||
assert generation_attrs[TRACE_NAME_ATTR] == "from-body"
|
||||
|
||||
|
||||
UNNAMED: Final = {"proxy_server_request": {"headers": {}}}
|
||||
JUDGE_PARAMS: Final = {"metadata": {"internal_call_origin": "llm_as_a_judge_guardrail"}}
|
||||
CALLER_NAMES: Final = (
|
||||
{"metadata": {"trace_name": "from-caller"}, "proxy_server_request": {"headers": {}}},
|
||||
{"proxy_server_request": {"headers": {"langfuse_trace_name": "from-caller"}}},
|
||||
)
|
||||
|
||||
|
||||
def _root_trace_name_after(
|
||||
logger: OpenTelemetryV2,
|
||||
calls: Sequence[tuple[str, Mapping[str, object]]],
|
||||
root_logger: OpenTelemetryV2 | None = None,
|
||||
) -> object:
|
||||
root: Final = _start_root(root_logger or logger)
|
||||
for index, (call_type, litellm_params) in enumerate(calls):
|
||||
logger.log_pre_api_call(
|
||||
model="gpt-5.4-mini",
|
||||
messages=[],
|
||||
kwargs={"litellm_call_id": f"call_{index}", "litellm_params": litellm_params, "call_type": call_type},
|
||||
)
|
||||
root.end()
|
||||
return dict(root.attributes or {}).get(TRACE_NAME_ATTR)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("later", [JUDGE_PARAMS, UNNAMED], ids=["litellm-sub-call", "untagged-call"])
|
||||
@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"])
|
||||
def test_a_later_call_in_the_request_keeps_the_caller_trace_name_on_the_root(caller, later):
|
||||
logger, _ = _logger()
|
||||
|
||||
assert _root_trace_name_after(logger, (("acompletion", caller), ("acompletion", later))) == "from-caller"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"])
|
||||
def test_a_caller_trace_name_replaces_the_name_an_earlier_call_derived(caller):
|
||||
logger, _ = _logger()
|
||||
|
||||
assert _root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("acompletion", caller))) == "from-caller"
|
||||
|
||||
|
||||
def test_a_later_call_keeps_the_trace_name_the_first_call_derived():
|
||||
logger, _ = _logger()
|
||||
|
||||
assert (
|
||||
_root_trace_name_after(logger, (("anthropic_messages", UNNAMED), ("acompletion", UNNAMED)))
|
||||
== "litellm-anthropic_messages"
|
||||
)
|
||||
|
||||
|
||||
def test_a_litellm_sub_call_before_the_request_call_leaves_the_trace_name_to_the_request_call():
|
||||
logger, _ = _logger()
|
||||
|
||||
assert (
|
||||
_root_trace_name_after(logger, (("acompletion", JUDGE_PARAMS), ("anthropic_messages", UNNAMED)))
|
||||
== "litellm-anthropic_messages"
|
||||
)
|
||||
|
||||
|
||||
def test_a_root_another_vendor_started_gets_no_derived_trace_name():
|
||||
logger, _ = _logger()
|
||||
other_vendor, _ = _logger(mappers=("genai",))
|
||||
|
||||
assert _root_trace_name_after(logger, (("acompletion", UNNAMED),), root_logger=other_vendor) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("caller", CALLER_NAMES, ids=["body", "header"])
|
||||
def test_a_root_another_vendor_started_still_gets_the_caller_trace_name(caller):
|
||||
logger, _ = _logger()
|
||||
other_vendor, _ = _logger(mappers=("genai",))
|
||||
|
||||
assert _root_trace_name_after(logger, (("acompletion", caller),), root_logger=other_vendor) == "from-caller"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("capture", ["span_only", "no_content"])
|
||||
def test_body_metadata_user_session_and_tags_land_on_the_root_and_the_generation(capture):
|
||||
logger, exporter = _logger(capture=capture)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue