diff --git a/tests/integration/observability/_openinference_support.py b/tests/integration/observability/_openinference_support.py new file mode 100644 index 00000000000..b2847c81d9e --- /dev/null +++ b/tests/integration/observability/_openinference_support.py @@ -0,0 +1,1141 @@ +from __future__ import annotations + +import base64 +import json +import uuid +from collections.abc import Callable, Iterator, Mapping, Sequence +from contextlib import AbstractContextManager, ExitStack, contextmanager, nullcontext +from dataclasses import dataclass +from itertools import chain +from pathlib import Path +from typing import Final + +import httpx +import yaml +from integration._support.client import Gateway, eventually, object_value, string_value +from integration._support.process import OwnedProxy, owned_proxy_process +from integration._support.wire import Reply, Request, Wire, wire_server +from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +from opentelemetry.proto.common.v1.common_pb2 import AnyValue +from pydantic import JsonValue, TypeAdapter + +JSON_OBJECT: Final[TypeAdapter[dict[str, JsonValue]]] = TypeAdapter(dict[str, JsonValue]) +JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) +JSON_MESSAGES: Final[TypeAdapter[list[dict[str, JsonValue]]]] = TypeAdapter(list[dict[str, JsonValue]]) + +CHAT_TOOLS: Final = [ + { + "type": "function", + "function": { + "name": "lookup_weather", + "description": "Get weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + }, + } +] + +RESPONSES_TOOLS: Final = [ + { + "type": "function", + "name": "lookup_weather", + "description": "Get weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + } +] + +ANTHROPIC_TOOLS: Final = [ + { + "name": "lookup_weather", + "description": "Get weather", + "input_schema": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + } +] + +_TOOL_PREFIX: Final = "llm.output_messages.{message}.message.tool_calls.{tool}.tool_call." + + +@dataclass(frozen=True, slots=True) +class Rig: + proxy: Gateway + owned: OwnedProxy + model: str + provider: Wire + destination: Wire + + +def _json_object(body: bytes) -> dict[str, JsonValue]: + return JSON_OBJECT.validate_json(body) + + +def _json_object_value(value: JsonValue) -> dict[str, JsonValue]: + return JSON_OBJECT.validate_python(value) + + +def _json_messages(value: str) -> list[dict[str, JsonValue]]: + return JSON_MESSAGES.validate_json(value) + + +def _assert_chat_request( + request: Request, + *, + messages: JsonValue, + tools: JsonValue = CHAT_TOOLS, + tool_choice: JsonValue | None = None, + include_tools: bool = True, + stream: bool | None = None, + stream_options: JsonValue | None = None, + n: int | None = None, +) -> dict[str, JsonValue]: + request_tools: Final = ( + { + "tools": tools, + "tool_choice": ( + tool_choice if tool_choice is not None else {"type": "function", "function": {"name": "lookup_weather"}} + ), + } + if include_tools + else {} + ) + expected: Final[dict[str, JsonValue]] = { + "model": "gpt-4o-mini", + "messages": messages, + **request_tools, + **({"stream": stream} if stream is not None else {}), + **({"stream_options": stream_options} if stream_options is not None else {}), + **({"n": n} if n is not None else {}), + } + observed: Final = _json_object(request.body) + assert observed == expected, (observed, expected) + return observed + + +def _chat_request_marker(request: Request) -> str: + body: Final = _json_object(request.body) + messages: Final = JSON_MESSAGES.validate_python(body["messages"]) + marker: Final = messages[0].get("content") + assert isinstance(marker, str), body + return marker + + +def _assert_responses_request( + request: Request, + *, + marker: str, + input_value: str = "weather in Paris?", + stream: bool = False, +) -> None: + expected: Final[dict[str, JsonValue]] = { + "model": "gpt-4o-mini", + "input": input_value, + "tools": RESPONSES_TOOLS, + "tool_choice": {"type": "function", "name": "lookup_weather"}, + "metadata": {"trace_marker": marker}, + **({"stream": True} if stream else {}), + } + observed: Final = _json_object(request.body) + assert observed == expected, request + + +def _assert_messages_request( + request: Request, + *, + marker: str, + prompt: str = "weather in Paris?", + stream: bool = False, +) -> None: + expected: Final[dict[str, JsonValue]] = { + "model": "claude-opus-5-5", + "max_tokens": 64, + "messages": [{"role": "user", "content": prompt}], + "tools": ANTHROPIC_TOOLS, + "tool_choice": {"type": "auto"}, + "metadata": {}, + "stream": stream, + } + observed: Final = _json_object(request.body) + assert observed == expected, request + + +def _stream_values(reply: Reply) -> tuple[dict[str, JsonValue], ...]: + return tuple( + _json_object(chunk.split(b"data: ", 1)[1].splitlines()[0]) + for chunk in reply.chunks or () + if b"data: " in chunk and b"[DONE]" not in chunk + ) + + +def _sse_json_values(body: bytes) -> tuple[dict[str, JsonValue], ...]: + return tuple( + _json_object(line.removeprefix(b"data: ").strip()) + for line in body.splitlines() + if line.startswith(b"data: ") and line != b"data: [DONE]" + ) + + +def _chat_caller_response(reply: Reply, model: str) -> dict[str, JsonValue]: + body: Final = _json_object(reply.body) + choices: Final = JSON_MESSAGES.validate_python(body["choices"]) + return { + **body, + "model": model, + "choices": [ + { + **choice, + "message": _chat_caller_message(_json_object_value(choice["message"])), + "provider_specific_fields": {}, + } + for choice in choices + ], + } + + +def _chat_caller_message(message: dict[str, JsonValue]) -> dict[str, JsonValue]: + raw_tool_calls: Final = message.get("tool_calls") + tool_calls: Final = JSON_MESSAGES.validate_python(raw_tool_calls) if isinstance(raw_tool_calls, list) else () + return { + **{key: value for key, value in message.items() if key != "tool_calls"}, + **({"tool_calls": [_chat_caller_tool_call(call) for call in tool_calls]} if tool_calls else {}), + "provider_specific_fields": {"refusal": None}, + } + + +def _chat_caller_tool_call(call: dict[str, JsonValue]) -> dict[str, JsonValue]: + function: Final = _json_object_value(call["function"]) + arguments: Final = function.get("arguments") + return { + **call, + "function": { + **function, + **( + {"arguments": json.dumps(arguments)} + if "arguments" in function and not isinstance(arguments, str) + else {} + ), + }, + } + + +def _chat_output_tool_call(call: dict[str, JsonValue]) -> dict[str, JsonValue]: + function: Final = _json_object_value(call["function"]) + return { + **call, + "id": call["id"] if isinstance(call.get("id"), str) else None, + "function": { + **function, + **({"name": None} if "name" not in function else {}), + }, + } + + +def _chat_output_value(reply: Reply) -> str: + body: Final = _json_object(reply.body) + choices: Final = JSON_MESSAGES.validate_python(body["choices"]) + assert len(choices) == 1, body + message: Final = _chat_caller_message(_json_object_value(choices[0]["message"])) + tool_calls: Final = JSON_MESSAGES.validate_python(message["tool_calls"]) if "tool_calls" in message else () + return json.dumps( + [ + { + **{ + key: value + for key, value in message.items() + if key not in {"provider_specific_fields", "tool_calls"} + }, + **({"tool_calls": [_chat_output_tool_call(call) for call in tool_calls]} if tool_calls else {}), + } + ] + ) + + +def _chat_caller_stream(reply: Reply, model: str) -> tuple[dict[str, JsonValue], ...]: + return tuple(chain.from_iterable(_chat_caller_stream_events(event, model) for event in _stream_values(reply))) + + +def _chat_cache_hit_caller_stream( + marker: str, model: str, calls: Sequence[dict[str, JsonValue]] +) -> tuple[dict[str, JsonValue], ...]: + return ( + { + "id": marker, + "object": "chat.completion.chunk", + "created": 1, + "model": model, + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant", + "tool_calls": [{"index": index, **call} for index, call in enumerate(calls)], + }, + } + ], + }, + { + "id": marker, + "object": "chat.completion.chunk", + "created": 1, + "model": model, + "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], + }, + ) + + +def _chat_caller_stream_event(event: dict[str, JsonValue], model: str) -> dict[str, JsonValue]: + choices: Final = JSON_MESSAGES.validate_python(event["choices"]) + return { + "id": event["id"], + "object": event["object"], + "created": 1, + "model": model, + "choices": [_chat_caller_stream_choice(choice) for choice in choices], + } + + +def _chat_caller_stream_events(event: dict[str, JsonValue], model: str) -> tuple[dict[str, JsonValue], ...]: + return ( + _chat_caller_stream_event(event, model), + *((_chat_caller_stream_usage_event(event, model),) if "usage" in event else ()), + ) + + +def _chat_caller_stream_usage_event(event: dict[str, JsonValue], model: str) -> dict[str, JsonValue]: + return { + "id": event["id"], + "object": event["object"], + "created": 1, + "model": model, + "choices": [{"index": 0, "delta": {}}], + "usage": {**_json_object_value(event["usage"]), "cost": 4.05e-6}, + } + + +def _chat_caller_stream_choice(choice: dict[str, JsonValue]) -> dict[str, JsonValue]: + delta: Final = _json_object_value(choice["delta"]) + tool_calls: Final = JSON_MESSAGES.validate_python(delta["tool_calls"]) if "tool_calls" in delta else None + finish_reason: Final = choice.get("finish_reason") + return { + **{ + key: value + for key, value in choice.items() + if key not in {"delta", "finish_reason", "index"} and value is not None + }, + "index": choice["index"], + "delta": { + **{key: value for key, value in delta.items() if key != "tool_calls" and value is not None}, + **( + {"tool_calls": [_chat_caller_stream_tool_call(call) for call in tool_calls]} + if tool_calls is not None + else {} + ), + }, + **({"finish_reason": finish_reason} if finish_reason is not None else {}), + } + + +def _chat_caller_stream_tool_call(call: dict[str, JsonValue]) -> dict[str, JsonValue]: + function: Final = _json_object_value(call["function"]) + return { + **{key: value for key, value in call.items() if key not in {"function", "type"}}, + "type": "function", + "function": {**function, **({"arguments": ""} if "arguments" not in function else {})}, + } + + +def _normalize_chat_caller_stream(events: Sequence[JsonValue]) -> tuple[dict[str, JsonValue], ...]: + return tuple({**_json_object_value(event), "created": 1} for event in events) + + +def _responses_caller_body(body: dict[str, JsonValue], model: str) -> dict[str, JsonValue]: + usage: Final = _json_object_value(body["usage"]) + output: Final = JSON_MESSAGES.validate_python(body["output"]) + return { + **body, + "id": "", + "model": model, + "error": None, + "incomplete_details": None, + "instructions": None, + "metadata": None, + "parallel_tool_calls": None, + "temperature": None, + "tool_choice": None, + "tools": None, + "top_p": None, + "max_output_tokens": None, + "previous_response_id": None, + "reasoning": None, + "text": None, + "truncation": None, + "user": None, + "store": None, + "output": [{**item, **({"namespace": None} if item.get("type") == "function_call" else {})} for item in output], + "usage": { + "input_tokens_details": None, + "output_tokens_details": None, + "cost": None, + **usage, + }, + } + + +def _responses_stream_caller_body(body: dict[str, JsonValue]) -> dict[str, JsonValue]: + response_id: Final = body.get("id") + assert isinstance(response_id, str), body + usage: Final = _json_object_value(body["usage"]) if body.get("status") == "completed" else {} + return { + **body, + "id": "", + **({"usage": {**usage, "cost": 4.05e-06}} if usage else {}), + } + + +def _responses_caller_response(reply: Reply, model: str) -> dict[str, JsonValue]: + return _responses_caller_body(_json_object(reply.body), model) + + +def _normalize_responses_caller_body(body: dict[str, JsonValue]) -> dict[str, JsonValue]: + response_id: Final = body.get("id") + assert isinstance(response_id, str) and response_id.startswith("resp_"), body + return {**body, "id": ""} + + +def _responses_caller_stream(reply: Reply, model: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + { + **event, + "model": model, + **( + {"response": _responses_stream_caller_body(_json_object_value(event["response"]))} + if "response" in event + else {} + ), + } + for event in _stream_values(reply) + ) + + +def _normalize_responses_caller_stream( + events: Sequence[dict[str, JsonValue]], +) -> tuple[dict[str, JsonValue], ...]: + return tuple( + { + **event, + **( + {"response": _normalize_responses_caller_body(_json_object_value(event["response"]))} + if "response" in event + else {} + ), + } + for event in events + ) + + +def _messages_caller_response(reply: Reply, model: str) -> dict[str, JsonValue]: + return {**_json_object(reply.body), "model": model} + + +def _messages_caller_raw_stream(reply: Reply, model: str) -> tuple[dict[str, JsonValue], ...]: + return tuple( + { + **event, + **( + {"message": {**_json_object_value(event["message"]), "model": model}} + if event.get("type") == "message_start" + else {} + ), + } + for event in _stream_values(reply) + ) + + +def _messages_caller_stream_response(reply: Reply, model: str) -> dict[str, JsonValue]: + body: Final = _messages_caller_response(reply, model) + content: Final = JSON_MESSAGES.validate_python(body["content"]) + return { + **body, + "content": [{**block, **({"caller": None} if block.get("type") == "tool_use" else {})} for block in content], + } + + +def _messages_caller_stream( + reply: Reply, model: str, *, final_message: dict[str, JsonValue] +) -> tuple[dict[str, JsonValue], ...]: + events: Final = tuple( + chain.from_iterable( + _messages_caller_stream_event(event, model, final_message) for event in _stream_values(reply) + ) + ) + return (*events, {"type": "message_stop", "message": final_message}) + + +def _messages_caller_stream_event( + event: dict[str, JsonValue], model: str, final_message: dict[str, JsonValue] +) -> tuple[dict[str, JsonValue], ...]: + if event.get("type") == "message_stop": + return () + if event.get("type") == "content_block_stop": + index: Final = event["index"] + assert isinstance(index, int), event + content: Final = JSON_MESSAGES.validate_python(final_message["content"]) + return ({**event, "content_block": content[index]},) + caller_event: Final = { + **event, + **({"message": {**_json_object_value(event["message"]), "model": model}} if "message" in event else {}), + } + if event.get("type") != "content_block_delta": + return (caller_event,) + delta: Final = _json_object_value(event["delta"]) + if delta.get("type") != "input_json_delta": + return (caller_event,) + partial_json: Final = delta["partial_json"] + assert isinstance(partial_json, str), event + return ( + caller_event, + { + "type": "input_json", + "partial_json": partial_json, + "snapshot": JSON_OBJECT.validate_json(partial_json), + }, + ) + + +def _span_attributes(request: Request) -> Iterator[dict[str, str]]: + if request.headers.get("content-type") != "application/x-protobuf": + return + batch: Final = ExportTraceServiceRequest.FromString(request.body) + for resource_spans in batch.resource_spans: + for scope_spans in resource_spans.scope_spans: + for span in scope_spans.spans: + yield {attribute.key: _attribute_text(attribute.value) for attribute in span.attributes} + + +def _attribute_text(value: object) -> str: + assert isinstance(value, AnyValue) + match value.WhichOneof("value"): + case "string_value": + return value.string_value + case "int_value": + return str(value.int_value) + case "double_value": + return str(value.double_value) + case "bool_value": + return str(value.bool_value) + case _: + return "" + + +def _spans(requests: tuple[Request, ...]) -> Iterator[dict[str, str]]: + for request in requests: + yield from _span_attributes(request) + + +def _canonical_response_id(value: str) -> str: + try: + return base64.b64decode(value.removeprefix("resp_").encode()).decode() + except (ValueError, UnicodeDecodeError): + return value + + +def _matching_llm_spans(requests: tuple[Request, ...], response_id: str) -> Iterator[dict[str, str]]: + wanted: Final = _canonical_response_id(response_id) + for attributes in _spans(requests): + if attributes.get("openinference.span.kind") != "LLM": + continue + observed: Final = attributes.get("gen_ai.response.id", "") + if _canonical_response_id(observed) == wanted or response_id in attributes.values(): + yield attributes + + +def _matching_marker_spans(requests: tuple[Request, ...], marker: str) -> Iterator[dict[str, str]]: + for attributes in _spans(requests): + if attributes.get("openinference.span.kind") == "LLM" and marker in attributes.values(): + yield attributes + + +def _matching_span(destination: Wire, response_id: str) -> dict[str, str]: + return eventually( + lambda: tuple(_matching_llm_spans(destination.drain(), response_id)), + lambda spans: len(spans) == 1, + seconds=30, + )[0] + + +def _matching_marker_span(destination: Wire, marker: str) -> dict[str, str]: + return eventually( + lambda: tuple(_matching_marker_spans(destination.drain(), marker)), + lambda spans: len(spans) == 1, + seconds=30, + )[0] + + +def _matching_output_value_span(destination: Wire, marker: str) -> dict[str, str]: + return eventually( + lambda: tuple( + attributes + for attributes in _spans(destination.drain()) + if attributes.get("openinference.span.kind") == "LLM" and marker in attributes.get("output.value", "") + ), + lambda spans: len(spans) == 1, + seconds=30, + )[0] + + +def _matching_genai_marker_span(destination: Wire, marker: str) -> dict[str, str]: + return eventually( + lambda: tuple( + attributes + for attributes in _spans(destination.drain()) + if attributes.get("gen_ai.operation.name") == "chat" and marker in attributes.values() + ), + lambda spans: len(spans) == 1, + seconds=30, + )[0] + + +def _matching_any_marker_span(destination: Wire, marker: str) -> dict[str, str]: + def matches(requests: tuple[Request, ...]) -> tuple[dict[str, str], ...]: + return tuple(attributes for attributes in _spans(requests) if marker in attributes.values()) + + return eventually( + lambda: matches(destination.drain()), + lambda spans: len(spans) == 1, + seconds=30, + )[0] + + +def _collect_marker_spans( + destination: Wire, markers: tuple[str, ...], *, timeout_seconds: float = 30 +) -> tuple[dict[str, str], ...]: + expected: Final = frozenset(markers) + + def matches(requests: tuple[Request, ...]) -> tuple[dict[str, str], ...]: + return tuple( + attributes + for attributes in _spans(requests) + if attributes.get("openinference.span.kind") == "LLM" + and any(marker in attributes.values() for marker in expected) + ) + + def complete(spans: tuple[dict[str, str], ...]) -> bool: + return all(any(marker in attributes.values() for attributes in spans) for marker in expected) + + def collect(previous: tuple[dict[str, str], ...]) -> tuple[dict[str, str], ...]: + current: Final = eventually(lambda: matches(destination.drain()), bool, seconds=timeout_seconds) + combined: Final = (*previous, *current) + return combined if complete(combined) else collect(combined) + + return collect(()) + + +def _write_config( + directory: Path, + *, + callbacks: Sequence[str] = ("arize",), + callback_settings: Mapping[str, JsonValue] | None = None, + litellm_settings: Mapping[str, JsonValue] | None = None, + general_settings: Mapping[str, JsonValue] | None = None, +) -> Path: + loaded: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + config: Final = { + **loaded, + "litellm_settings": { + **loaded["litellm_settings"], + "callbacks": list(callbacks), + **(litellm_settings or {}), + }, + **({"callback_settings": dict(callback_settings)} if callback_settings is not None else {}), + "general_settings": { + **loaded["general_settings"], + **(general_settings or {}), + }, + } + path: Final = directory / f"openinference-{uuid.uuid4().hex}.yaml" + path.write_text(yaml.safe_dump(config)) + return path + + +def _environment(destination: Wire, extra: Mapping[str, str] | None = None) -> dict[str, str]: + return { + "LITELLM_OTEL_V2": "1", + "OTEL_BSP_SCHEDULE_DELAY": "100", + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only", + "LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.trace_marker", + "ARIZE_HTTP_ENDPOINT": destination.url + "/v1/traces", + "ARIZE_SPACE_ID": "integration-space", + "ARIZE_API_KEY": "integration-arize-key", + **(extra or {}), + } + + +def _collector(_request: Request) -> Reply: + return Reply(body=b"", content_type="application/x-protobuf") + + +def _owned_sink_handler(sink: Callable[[Request], Reply]) -> Callable[[Request], Reply]: + def handle(request: Request) -> Reply: + if ( + request.method != "POST" + or not request.body + or request.headers.get("content-type", "").split(";", maxsplit=1)[0].strip().lower() + != "application/x-protobuf" + ): + return Reply(body=b"", content_type="application/x-protobuf") + try: + return sink(request) + except (AssertionError, IndexError, KeyError, TypeError, ValueError) as error: + raise AssertionError(f"{request.method} {request.target}: {error!r}") from error + + return handle + + +def _provider_handler(upstream: Callable[[Request], Reply]) -> Callable[[Request], Reply]: + def handle(request: Request) -> Reply: + if ( + request.method != "POST" + or not request.body + or request.headers.get("content-type", "").split(";", maxsplit=1)[0].strip().lower() != "application/json" + ): + return Reply(body=b"{}", content_type="application/json") + try: + return upstream(request) + except (AssertionError, IndexError, KeyError, TypeError, ValueError) as error: + raise AssertionError(f"{request.method} {request.target}: {error!r}") from error + + return handle + + +@contextmanager +def _rig( + gateway: Gateway, + directory: Path, + upstream: Callable[[Request], Reply], + *, + callbacks: Sequence[str] = ("arize",), + callback_settings: Mapping[str, JsonValue] | None = None, + litellm_settings: Mapping[str, JsonValue] | None = None, + general_settings: Mapping[str, JsonValue] | None = None, + environment: Mapping[str, str] | None = None, + remove_environment: tuple[str, ...] = (), + disabled_environment: tuple[str, ...] = (), + model_name: str = "openai/gpt-4o-mini", + api_base_suffix: str = "/v1", + destination_handler: Callable[[Request], Reply] | None = None, + destination_wire: Wire | None = None, + fresh_client_connections: bool = False, +) -> Iterator[Rig]: + destination_context: Final[AbstractContextManager[Wire]] = ( + nullcontext(destination_wire) + if destination_wire is not None + else wire_server(_owned_sink_handler(destination_handler or _collector)) + ) + with wire_server(_provider_handler(upstream)) as provider, destination_context as destination: + resolved_settings: Final = { + key: ( + {**value, "endpoint": destination.url + "/v1/traces"} + if key == "otel" and isinstance(value, dict) and value.get("endpoint") == "unused" + else value + ) + for key, value in (callback_settings or {}).items() + } + config: Final = _write_config( + directory, + callbacks=callbacks, + callback_settings=resolved_settings, + litellm_settings=litellm_settings, + general_settings=general_settings, + ) + preset_environment: Final = { + **({"PHOENIX_COLLECTOR_ENDPOINT": destination.url + "/v1/traces"} if "arize_phoenix" in callbacks else {}), + **( + { + "WANDB_HOST": destination.url, + "WANDB_API_KEY": "integration-weave-key", + "WANDB_PROJECT_ID": "integration/project", + } + if "weave_otel" in callbacks + else {} + ), + **( + { + "LANGFUSE_OTEL_HOST": destination.url, + "LANGFUSE_PUBLIC_KEY": "integration-public", + "LANGFUSE_SECRET_KEY": "integration-secret", + } + if "langfuse_otel" in callbacks + else {} + ), + **( + { + "LEVOAI_API_KEY": "integration-levo-key", + "LEVOAI_ORG_ID": "integration-org", + "LEVOAI_WORKSPACE_ID": "integration-workspace", + "LEVOAI_COLLECTOR_URL": destination.url + "/v1/traces", + } + if "levo" in callbacks + else {} + ), + **( + { + "SIGNOZ_INGESTION_ENDPOINT": destination.url + "/v1/traces", + "SIGNOZ_INGESTION_KEY": "integration-signoz-key", + } + if "signoz" in callbacks + else {} + ), + **( + {"OTEL_EXPORTER_OTLP_ENDPOINT": destination.url} + if any(callback_name in callbacks for callback_name in ("langtrace", "newrelic", "agentops")) + else {} + ), + } + overrides: Final = { + key: value + for key, value in _environment(destination, {**preset_environment, **(environment or {})}).items() + if key not in disabled_environment + } + with ( + owned_proxy_process( + gateway, + directory, + overrides, + config=config, + remove_environment=remove_environment, + workers=2, + ) as owned, + ExitStack() as resources, + ): + proxy_client: Final = ( + resources.enter_context( + httpx.Client( + base_url=str(owned.gateway.client.base_url), + timeout=15, + trust_env=False, + limits=httpx.Limits(max_keepalive_connections=0), + ) + ) + if fresh_client_connections + else owned.gateway.client + ) + proxy: Final = Gateway( + client=proxy_client, + key=owned.gateway.key, + upstream_url=owned.gateway.upstream_url, + ) + scenario: Final = resources.enter_context(proxy.scenario()) + model: Final = scenario.model( + model=model_name, + api_base=provider.url + api_base_suffix, + ) + yield Rig(proxy, owned, model, provider, destination) + + +def _chat_tool_call(identity: str, city: str = "Paris") -> dict[str, JsonValue]: + return { + "id": "call_" + identity, + "type": "function", + "function": { + "name": "lookup_weather", + "arguments": json.dumps({"city": city}), + }, + } + + +def _chat_response(identity: str, calls: Sequence[dict[str, JsonValue]] | None = None) -> Reply: + tool_calls: Final = list(calls if calls is not None else (_chat_tool_call(identity),)) + return Reply( + body=json.dumps( + { + "id": identity, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "finish_reason": "tool_calls", + "message": {"role": "assistant", "content": None, "tool_calls": tool_calls}, + } + ], + "usage": {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15}, + } + ).encode() + ) + + +def _chat_plain_response(identity: str, content: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": identity, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "message": {"role": "assistant", "content": content}, + } + ], + "usage": {"prompt_tokens": 7, "completion_tokens": 2, "total_tokens": 9}, + } + ).encode() + ) + + +def _chat_stream_response(identity: str, calls: Sequence[dict[str, JsonValue]], *, include_usage: bool = True) -> Reply: + def first_chunk(index: int, call: dict[str, JsonValue]) -> dict[str, JsonValue]: + fields: Final = object_value(JSON_VALUE.validate_python(call)) + return { + "index": index, + "id": string_value(fields["id"]), + "type": "function", + "function": {"name": "lookup_weather"}, + } + + def arguments_chunk(index: int, call: dict[str, JsonValue]) -> dict[str, JsonValue]: + fields: Final = object_value(JSON_VALUE.validate_python(call)) + function: Final = object_value(fields["function"]) + return { + "id": identity, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "index": index, + "function": {"arguments": string_value(function["arguments"])}, + } + ] + }, + "finish_reason": None, + } + ], + } + + chunks: Final = [ + { + "id": identity, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "delta": { + "role": "assistant", + "tool_calls": [first_chunk(index, call) for index, call in enumerate(calls)], + }, + "finish_reason": None, + } + ], + }, + *(arguments_chunk(index, call) for index, call in enumerate(calls)), + { + "id": identity, + "object": "chat.completion.chunk", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], + **({"usage": {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15}} if include_usage else {}), + }, + ] + return Reply( + content_type="text/event-stream", + chunks=tuple(b"data: " + json.dumps(chunk).encode() + b"\n\n" for chunk in chunks) + (b"data: [DONE]\n\n",), + ) + + +def _responses_response(identity: str, calls: Sequence[dict[str, JsonValue]] | None = None) -> Reply: + def response_item(call: dict[str, JsonValue]) -> dict[str, JsonValue]: + fields: Final = object_value(JSON_VALUE.validate_python(call)) + function: Final = object_value(fields["function"]) + call_id: Final = string_value(fields["id"]) + return { + "type": "function_call", + "id": "fc_" + call_id, + "call_id": call_id, + "name": string_value(function["name"]), + "arguments": string_value(function["arguments"]), + "status": "completed", + } + + response: Final = { + "id": identity, + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gpt-4o-mini", + "output": [response_item(call) for call in calls if calls is not None] + if calls is not None + else [response_item(_chat_tool_call(identity))], + "usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15}, + } + return Reply(body=json.dumps(response).encode()) + + +def _responses_stream_response(identity: str, calls: Sequence[dict[str, JsonValue]]) -> Reply: + response: Final = _json_object(_responses_response(identity, calls).body) + output: Final = response["output"] + items: Final = tuple(object_value(item) for item in output) if isinstance(output, list) else () + events: Final = ( + { + "type": "response.created", + "sequence_number": 0, + "response": {**response, "status": "in_progress", "output": []}, + }, + *( + { + "type": "response.output_item.added", + "sequence_number": index + 1, + "output_index": index, + "item": item, + } + for index, item in enumerate(items) + ), + *( + { + "type": "response.function_call_arguments.delta", + "sequence_number": index + len(items) + 1, + "item_id": str(item["id"]), + "output_index": index, + "delta": str(item["arguments"]), + } + for index, item in enumerate(items) + ), + { + "type": "response.completed", + "sequence_number": len(items) * 2 + 1, + "response": response, + }, + ) + return Reply( + content_type="text/event-stream", + chunks=tuple(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events), + ) + + +def _anthropic_response(identity: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "content": [ + { + "type": "tool_use", + "id": "call_" + identity, + "name": "lookup_weather", + "input": {"city": "Paris"}, + } + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 4}, + } + ).encode() + ) + + +def _anthropic_stream_response(identity: str) -> Reply: + message: Final = { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "content": [ + { + "type": "tool_use", + "id": "call_" + identity, + "name": "lookup_weather", + "input": {"city": "Paris"}, + } + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 4}, + } + events: Final = ( + {"type": "message_start", "message": {**message, "content": [], "stop_reason": None}}, + { + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "tool_use", + "id": "call_" + identity, + "name": "lookup_weather", + "input": {}, + }, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": '{"city": "Paris"}'}, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": 4}, + }, + {"type": "message_stop"}, + ) + return Reply( + content_type="text/event-stream", + chunks=tuple(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events), + ) + + +def _tool_call_attributes( + attributes: Mapping[str, str], + *, + message_index: int = 0, + tool_index: int = 0, +) -> tuple[str, str, dict[str, JsonValue]]: + prefix: Final = _TOOL_PREFIX.format(message=message_index, tool=tool_index) + return ( + attributes[prefix + "id"], + attributes[prefix + "function.name"], + JSON_OBJECT.validate_json(attributes[prefix + "function.arguments"]), + ) + + +def _assert_tool_span( + attributes: Mapping[str, str], + *, + marker: str, + output: JsonValue, + calls: Sequence[tuple[str, str, Mapping[str, JsonValue]]], + metadata: Mapping[str, JsonValue] | None = None, + baggage: Mapping[str, str] | None = None, +) -> None: + for index, (call_id, name, arguments) in enumerate(calls): + observed: Final = _tool_call_attributes(attributes, tool_index=index) + assert observed == (call_id, name, dict(arguments)), f"tool call {index} for {marker}: {observed!r}" + observed_output: Final = JSON_MESSAGES.validate_json(attributes["output.value"]) + assert observed_output == output, (marker, observed_output) + if metadata is None: + assert "metadata" not in attributes, attributes + else: + assert json.loads(attributes["metadata"]) == dict(metadata), attributes + for key, value in (baggage or {}).items(): + assert attributes.get("litellm.metadata." + key) == value, attributes + + +def _response_tool_calls(identity: str, cities: Sequence[str] = ("Paris",)) -> list[dict[str, JsonValue]]: + return [_chat_tool_call(identity + "-" + city, city) for city in cities] diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py b/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py new file mode 100644 index 00000000000..88c19215f22 --- /dev/null +++ b/tests/integration/observability/test_arize_otel_v2_openinference_chaos.py @@ -0,0 +1,377 @@ +from __future__ import annotations + +import threading +import uuid +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack +from pathlib import Path +from typing import Final +from urllib.parse import urlsplit + +import httpx +import psutil +import pytest +from _openinference_support import ( + CHAT_TOOLS, + RESPONSES_TOOLS, + _anthropic_response, + _anthropic_stream_response, + _assert_chat_request, + _assert_messages_request, + _assert_responses_request, + _chat_caller_response, + _chat_caller_stream, + _chat_request_marker, + _chat_response, + _chat_stream_response, + _chat_tool_call, + _collect_marker_spans, + _json_object, + _messages_caller_raw_stream, + _messages_caller_response, + _normalize_chat_caller_stream, + _normalize_responses_caller_body, + _normalize_responses_caller_stream, + _owned_sink_handler, + _responses_caller_response, + _responses_caller_stream, + _responses_response, + _responses_stream_response, + _rig, + _spans, + _sse_json_values, +) +from integration._support.client import Gateway, eventually +from integration._support.wire import Reply, Request, wire_server + + +def _call( + proxy: Gateway, + model: str, + marker: str, + *, + surface: str = "chat", + stream: bool = False, + prompt: str | None = None, +) -> httpx.Response: + match surface: + case "chat": + return proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt or "weather in Paris?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + **({"stream": True} if stream else {}), + **({"stream_options": {"include_usage": True}} if stream and surface == "chat" else {}), + "cache": {"no-cache": True}, + }, + ) + case "responses": + return proxy.request( + "POST", + "/v1/responses", + { + "model": model, + "input": prompt or "weather in Paris?", + "tools": RESPONSES_TOOLS, + "tool_choice": {"type": "function", "name": "lookup_weather"}, + "metadata": {"trace_marker": marker}, + **({"stream": True} if stream else {}), + "cache": {"no-cache": True}, + }, + ) + case "messages": + return proxy.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "messages": [{"role": "user", "content": prompt or "weather in Paris?"}], + "tools": [ + { + "name": "lookup_weather", + "description": "Get weather", + "input_schema": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + } + ], + "tool_choice": {"type": "auto"}, + "metadata": {"trace_marker": marker}, + **({"stream": True} if stream else {}), + "cache": {"no-cache": True}, + }, + ) + case _: + raise AssertionError(f"Unknown endpoint: {surface}") + + +def _assert_response(response: httpx.Response, marker: str, surface: str, stream: bool, model: str) -> None: + assert response.status_code == 200, response.text + if stream: + observed_stream: Final = _sse_json_values(response.content) + if surface == "chat": + chat_reply: Final = _chat_stream_response(marker, (_chat_tool_call(marker),)) + assert _normalize_chat_caller_stream(observed_stream) == _chat_caller_stream(chat_reply, model), ( + response.text + ) + return + if surface == "responses": + responses_reply: Final = _responses_stream_response(marker, (_chat_tool_call(marker),)) + assert _normalize_responses_caller_stream(observed_stream) == _responses_caller_stream( + responses_reply, model + ), response.text + return + if surface == "messages": + messages_reply: Final = _anthropic_stream_response(marker) + assert observed_stream == _messages_caller_raw_stream(messages_reply, model), response.text + return + raise AssertionError(f"Unknown endpoint: {surface}") + + observed: Final = _json_object(response.content) + if surface == "chat": + chat_reply: Final = _chat_response(marker) + assert observed == _chat_caller_response(chat_reply, model), response.text + return + if surface == "responses": + responses_reply: Final = _responses_response(marker) + assert _normalize_responses_caller_body(observed) == _responses_caller_response(responses_reply, model), ( + response.text + ) + return + if surface == "messages": + messages_reply: Final = _anthropic_response(marker) + assert observed == _messages_caller_response(messages_reply, model), response.text + return + raise AssertionError(f"Unknown endpoint: {surface}") + + +def _call_without_worker_error( + proxy: Gateway, model: str, marker: str, *, prompt: str | None = None +) -> httpx.Response | None: + try: + return _call(proxy, model, marker, prompt=prompt) + except httpx.HTTPError: + return None + + +def test_arize_otel_v2_f1_sink_outage_and_recovery(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = tuple("f1-" + uuid.uuid4().hex for _ in range(30)) + surfaces: Final = ("chat", "responses", "messages") + calls: Final = tuple( + (marker, surfaces[index % len(surfaces)], index % 2 == 0) for index, marker in enumerate(markers) + ) + + def upstream(request: Request) -> Reply: + body: Final = _json_object(request.body) + if request.target.endswith("/messages"): + messages: Final = body.get("messages") + assert isinstance(messages, list) and isinstance(messages[0], dict), body + marker: Final = messages[0].get("content") + assert isinstance(marker, str), body + _assert_messages_request( + request, + marker=marker, + prompt=marker, + stream=True if body.get("stream") is True else False, + ) + return _anthropic_stream_response(marker) if body.get("stream") is True else _anthropic_response(marker) + if request.target.endswith("/responses"): + marker: Final = body.get("input") + assert isinstance(marker, str), body + _assert_responses_request( + request, + marker=marker, + input_value=marker, + stream=body.get("stream") is True, + ) + return ( + _responses_stream_response(marker, (_chat_tool_call(marker),)) + if body.get("stream") is True + else _responses_response(marker) + ) + marker: Final = _chat_request_marker(request) + _assert_chat_request( + request, + messages=[{"role": "user", "content": marker}], + stream=True if body.get("stream") is True else None, + stream_options={"include_usage": True} if body.get("stream") is True else None, + ) + return ( + _chat_stream_response(marker, (_chat_tool_call(marker),)) + if body.get("stream") is True + else _chat_response(marker) + ) + + def sink(_request: Request) -> Reply: + return Reply(body=b"", content_type="application/x-protobuf") + + with ExitStack() as servers: + initial_stack: Final = servers.enter_context(ExitStack()) + stopped_destination: Final = initial_stack.enter_context(wire_server(_owned_sink_handler(sink))) + sink_port: Final = urlsplit(stopped_destination.url).port + assert sink_port is not None, stopped_destination.url + initial_stack.close() + with _rig( + gateway, + tmp_path, + upstream, + environment={"OTEL_BSP_SCHEDULE_DELAY": "20000"}, + destination_wire=stopped_destination, + ) as rig: + with httpx.Client(trust_env=False) as client, pytest.raises(httpx.ConnectError): + client.get(stopped_destination.url + "/health", timeout=2) + with rig.proxy.scenario() as scenario: + messages_model: Final = scenario.model( + model="anthropic/claude-opus-5-5", + api_base=rig.provider.url, + ) + with ThreadPoolExecutor(max_workers=len(calls)) as executor: + futures: Final = tuple( + executor.submit( + _call, + rig.proxy, + messages_model if surface == "messages" else rig.model, + marker, + surface=surface, + stream=stream, + prompt=marker, + ) + for marker, surface, stream in calls + ) + responses: Final = tuple( + (marker, surface, stream, future.result(timeout=60)) + for (marker, surface, stream), future in zip(calls, futures, strict=True) + ) + for marker, surface, stream, response in responses: + model: Final = messages_model if surface == "messages" else rig.model + _assert_response(response, marker, surface, stream, model) + with httpx.Client(trust_env=False) as client, pytest.raises(httpx.ConnectError): + client.get(stopped_destination.url + "/health", timeout=2) + recovered_stack: Final = servers.enter_context(ExitStack()) + recovered_destination: Final = recovered_stack.enter_context( + wire_server(_owned_sink_handler(sink), port=sink_port) + ) + spans: Final = _collect_marker_spans(recovered_destination, markers, timeout_seconds=90) + assert len(spans) == len(markers), spans + recorded: Final = tuple(span["litellm.metadata.trace_marker"] for span in spans) + assert len(recorded) == len(markers), recorded + assert frozenset(recorded) == frozenset(markers), recorded + + +def test_arize_otel_v2_f2_slow_sink_does_not_deadlock(gateway: Gateway, tmp_path: Path) -> None: + release: Final = threading.Event() + blocked: Final = threading.Event() + completed: Final = threading.Event() + marker: Final = "f2-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + def sink(request: Request) -> Reply: + if any(attributes.get("litellm.metadata.trace_marker") == marker for attributes in _spans((request,))): + blocked.set() + assert release.wait(timeout=5), "slow sink was not released" + completed.set() + return Reply(body=b"", content_type="application/x-protobuf") + + with _rig(gateway, tmp_path, upstream, destination_handler=sink) as rig: + timer: Final = threading.Timer(2, release.set) + try: + response: Final = _call(rig.proxy, rig.model, marker) + _assert_response(response, marker, "chat", False, rig.model) + assert eventually(lambda: blocked.is_set(), bool, seconds=10) + timer.start() + assert eventually(lambda: completed.is_set(), bool, seconds=10) + timer.join(timeout=5) + assert not timer.is_alive(), "Slow sink timer did not finish" + finally: + release.set() + timer.cancel() + if timer.ident is not None: + timer.join(timeout=5) + requests: Final = rig.destination.drain() + spans: Final = tuple( + attributes + for attributes in _spans(requests) + if attributes.get("openinference.span.kind") == "LLM" + and attributes.get("litellm.metadata.trace_marker") == marker + ) + assert len(spans) == 1, spans + + +def test_arize_otel_v2_f3_one_proxy_worker_can_die(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = tuple("f3-" + uuid.uuid4().hex for _ in range(8)) + release: Final = threading.Event() + + def upstream(request: Request) -> Reply: + request_marker: Final = _chat_request_marker(request) + _assert_chat_request( + request, + messages=[{"role": "user", "content": request_marker}], + ) + assert release.wait(timeout=20), "F3 upstream barrier was not released" + return _chat_response(request_marker) + + def sink(_request: Request) -> Reply: + return Reply(body=b"", content_type="application/x-protobuf") + + with _rig( + gateway, + tmp_path, + upstream, + destination_handler=sink, + fresh_client_connections=True, + ) as rig: + children: Final = psutil.Process(rig.owned.process.pid).children(recursive=True) + workers: Final = tuple( + child for child in children if child.is_running() and "resource_tracker" not in " ".join(child.cmdline()) + ) + assert len(workers) >= 2, tuple((worker.pid, worker.name()) for worker in workers) + with ThreadPoolExecutor(max_workers=len(markers)) as executor: + try: + futures: Final = tuple( + executor.submit( + _call_without_worker_error, + rig.proxy, + rig.model, + marker, + prompt=marker, + ) + for marker in markers + ) + observed: Final = eventually( + lambda: rig.provider.received.qsize(), + lambda count: count >= 2, + seconds=10, + ) + assert observed >= 2, observed + workers[0].kill() + assert eventually(lambda: not workers[0].is_running(), bool, seconds=10), workers[0] + release.set() + in_flight: Final = tuple( + (marker, future.result(timeout=60)) for marker, future in zip(markers, futures, strict=True) + ) + finally: + release.set() + survivor: Final = "f3-survivor-" + uuid.uuid4().hex + survivor_response: Final = _call(rig.proxy, rig.model, survivor, prompt=survivor) + _assert_response(survivor_response, survivor, "chat", False, rig.model) + served_responses: Final = tuple( + (marker, response) for marker, response in in_flight if response is not None and response.status_code == 200 + ) + for marker, response in served_responses: + _assert_response(response, marker, "chat", False, rig.model) + served: Final = tuple(marker for marker, _response in served_responses) + (survivor,) + collected: Final = _collect_marker_spans(rig.destination, served) + assert len(collected) == len(served), collected + exported: Final = tuple(span["litellm.metadata.trace_marker"] for span in collected) + assert len(exported) == len(served), exported + assert frozenset(exported) == frozenset(served), exported diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_family.py b/tests/integration/observability/test_arize_otel_v2_openinference_family.py new file mode 100644 index 00000000000..5d51db0b5b0 --- /dev/null +++ b/tests/integration/observability/test_arize_otel_v2_openinference_family.py @@ -0,0 +1,336 @@ +from __future__ import annotations + +import uuid +from collections.abc import Mapping +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import httpx +import pytest +from _openinference_support import ( + CHAT_TOOLS, + _assert_chat_request, + _chat_caller_response, + _chat_response, + _json_messages, + _json_object, + _matching_genai_marker_span, + _matching_marker_span, + _rig, +) +from integration._support.client import Gateway +from integration._support.wire import Reply, Request +from pydantic import JsonValue + +_GENAI_B3_KEYS_WITHOUT_BAGGAGE: Final = frozenset( + { + "gen_ai.input.messages", + "gen_ai.operation.name", + "gen_ai.output.messages", + "gen_ai.provider.name", + "gen_ai.request.model", + "gen_ai.response.finish_reasons", + "gen_ai.response.id", + "gen_ai.response.model", + "gen_ai.system", + "gen_ai.tool.0.description", + "gen_ai.tool.0.name", + "gen_ai.tool.0.parameters", + "gen_ai.usage.completion_tokens", + "gen_ai.usage.input_tokens", + "gen_ai.usage.output_tokens", + "gen_ai.usage.prompt_tokens", + "gen_ai.usage.total_tokens", + "litellm.api_key.hash", + "litellm.call_id", + "litellm.call_type", + "litellm.cost.discount_amount", + "litellm.cost.discount_percent", + "litellm.cost.input", + "litellm.cost.margin_fixed_amount", + "litellm.cost.margin_percent", + "litellm.cost.margin_total_amount", + "litellm.cost.original", + "litellm.cost.output", + "litellm.cost.tool_usage", + "litellm.cost.total", + "litellm.provider.model", + "litellm.request.route", + "litellm.request.tools.declared", + "llm.request.functions.0.description", + "llm.request.functions.0.name", + "llm.request.functions.0.parameters", + "server.address", + "server.port", + } +) +_GENAI_B3_KEYS_WITH_BAGGAGE: Final = _GENAI_B3_KEYS_WITHOUT_BAGGAGE | frozenset({"litellm.metadata.trace_marker"}) +_LANGFUSE_B3_KEYS: Final = frozenset( + { + "gen_ai.input.messages", + "gen_ai.operation.name", + "gen_ai.output.messages", + "gen_ai.provider.name", + "gen_ai.request.model", + "gen_ai.response.finish_reasons", + "gen_ai.response.id", + "gen_ai.response.model", + "gen_ai.system", + "gen_ai.tool.0.description", + "gen_ai.tool.0.name", + "gen_ai.tool.0.parameters", + "gen_ai.usage.completion_tokens", + "gen_ai.usage.input_tokens", + "gen_ai.usage.output_tokens", + "gen_ai.usage.prompt_tokens", + "gen_ai.usage.total_tokens", + "langfuse.observation.cost_details", + "langfuse.observation.id", + "langfuse.observation.input", + "langfuse.observation.metadata.provider", + "langfuse.observation.model.name", + "langfuse.observation.output", + "langfuse.observation.type", + "langfuse.observation.usage_details", + "litellm.api_key.hash", + "litellm.call_id", + "litellm.call_type", + "litellm.cost.discount_amount", + "litellm.cost.discount_percent", + "litellm.cost.input", + "litellm.cost.margin_fixed_amount", + "litellm.cost.margin_percent", + "litellm.cost.margin_total_amount", + "litellm.cost.original", + "litellm.cost.output", + "litellm.cost.tool_usage", + "litellm.cost.total", + "litellm.provider.model", + "litellm.request.route", + "litellm.request.tools.declared", + "llm.request.functions.0.description", + "llm.request.functions.0.name", + "llm.request.functions.0.parameters", + "server.address", + "server.port", + } +) +_B3_ATTRIBUTE_KEYS: Final[Mapping[str, frozenset[str]]] = MappingProxyType( + { + "langfuse_otel": _LANGFUSE_B3_KEYS, + "langtrace": _GENAI_B3_KEYS_WITHOUT_BAGGAGE, + "signoz": _GENAI_B3_KEYS_WITH_BAGGAGE, + "newrelic": _GENAI_B3_KEYS_WITH_BAGGAGE, + "levo": _GENAI_B3_KEYS_WITH_BAGGAGE, + "agentops": _GENAI_B3_KEYS_WITH_BAGGAGE, + "otel": _GENAI_B3_KEYS_WITH_BAGGAGE, + } +) +_B3_BAGGAGE_CALLBACKS: Final = frozenset({"signoz", "newrelic", "levo", "agentops", "otel"}) +_B4_ATTRIBUTE_KEYS: Final = frozenset( + { + "input.value", + "litellm.trace_id", + "llm.cost.total", + "llm.input_messages.0.message.content", + "llm.input_messages.0.message.role", + "llm.invocation_parameters", + "llm.is_streaming", + "llm.model_name", + "llm.output_messages.0.message.content", + "llm.output_messages.0.message.role", + "llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments", + "llm.output_messages.0.message.tool_calls.0.tool_call.function.name", + "llm.output_messages.0.message.tool_calls.0.tool_call.id", + "llm.provider", + "llm.request.type", + "llm.response.cost", + "llm.response.id", + "llm.response.model", + "llm.token_count.completion", + "llm.token_count.prompt", + "llm.token_count.total", + "llm.tools.0.description", + "llm.tools.0.name", + "llm.tools.0.parameters", + "metadata", + "openinference.span.kind", + "output.value", + "user.id", + } +) + + +def _request(proxy: Gateway, model: str, marker: str) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) + + +def _assert_openinference(attributes: dict[str, str], marker: str) -> None: + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == f"call_{marker}", attributes + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.name"] == "lookup_weather", ( + attributes + ) + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] == '{"city": "Paris"}' + ), attributes + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + assert _json_messages(attributes["output.value"]) == [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": f"call_{marker}", + "type": "function", + "function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'}, + } + ], + } + ], attributes + + +def test_arize_otel_v2_b1_phoenix_openinference(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "b1-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + with _rig( + gateway, + tmp_path, + upstream, + callbacks=("arize_phoenix",), + callback_settings={"otel": {"exporter": "http/protobuf", "endpoint": "unused"}}, + environment={"PHOENIX_PROJECT_NAME": "integration"}, + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + _assert_openinference(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_b2_weave_openinference(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "b2-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + with _rig( + gateway, + tmp_path, + upstream, + callbacks=("weave_otel",), + callback_settings={"otel": {"exporter": "http/protobuf", "endpoint": "unused"}}, + environment={"WANDB_BASE_URL": "http://127.0.0.1"}, + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + _assert_openinference(_matching_marker_span(rig.destination, marker), marker) + + +@pytest.mark.parametrize( + "callback", + ("langfuse_otel", "langtrace", "signoz", "newrelic", "levo", "agentops", "otel"), +) +def test_arize_otel_v2_b3_non_openinference_callback_family(callback: str, gateway: Gateway, tmp_path: Path) -> None: + marker: Final = f"b3-{callback}-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + body: Final = _json_object(request.body) + assert body == { + "messages": [{"role": "user", "content": "weather in Paris?"}], + "model": "gpt-4o-mini", + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "tools": CHAT_TOOLS, + }, body + return _chat_response(marker) + + callback_settings: Final[dict[str, JsonValue]] = { + "otel": {"exporter": "http/protobuf", "endpoint": "unused", "mapper_names": ["genai"]} + } + with _rig( + gateway, + tmp_path, + upstream, + callbacks=(callback,), + callback_settings=callback_settings, + environment={ + "LANGFUSE_HOST": "http://127.0.0.1", + **( + { + "HTTPS_PROXY": "http://127.0.0.1:0", + "https_proxy": "http://127.0.0.1:0", + "NO_PROXY": "", + "no_proxy": "", + } + if callback == "agentops" + else {} + ), + }, + remove_environment=( + ("NEW_RELIC_LICENSE_KEY",) + if callback == "newrelic" + else ("AGENTOPS_API_KEY",) + if callback == "agentops" + else () + ), + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + attributes: Final = _matching_genai_marker_span(rig.destination, marker) + assert frozenset(attributes) == _B3_ATTRIBUTE_KEYS[callback], attributes + assert "metadata" not in attributes, attributes + assert not any(".tool_calls." in key for key in attributes), attributes + baggage: Final = tuple( + sorted((key, value) for key, value in attributes.items() if key.startswith("litellm.metadata.")) + ) + expected_baggage: Final = ( + (("litellm.metadata.trace_marker", marker),) if callback in _B3_BAGGAGE_CALLBACKS else () + ) + assert baggage == expected_baggage, attributes + + +def test_arize_otel_v2_b4_legacy_otel_is_unchanged(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "b4-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + body: Final = _json_object(request.body) + assert body == { + "messages": [{"role": "user", "content": "weather in Paris?"}], + "model": "gpt-4o-mini", + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "tools": CHAT_TOOLS, + }, body + return _chat_response(marker) + + with _rig( + gateway, + tmp_path, + upstream, + callbacks=("arize",), + callback_settings={"otel": {"exporter": "http/protobuf", "endpoint": "unused"}}, + remove_environment=("LITELLM_OTEL_V2",), + disabled_environment=("LITELLM_OTEL_V2",), + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + assert frozenset(attributes) == _B4_ATTRIBUTE_KEYS, attributes diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_modes.py b/tests/integration/observability/test_arize_otel_v2_openinference_modes.py new file mode 100644 index 00000000000..b0fef46163f --- /dev/null +++ b/tests/integration/observability/test_arize_otel_v2_openinference_modes.py @@ -0,0 +1,438 @@ +from __future__ import annotations + +import uuid +from collections.abc import Callable +from pathlib import Path +from typing import Final + +import httpx +import pytest +from _openinference_support import ( + CHAT_TOOLS, + _assert_chat_request, + _chat_caller_response, + _chat_plain_response, + _chat_request_marker, + _chat_response, + _json_messages, + _json_object, + _matching_marker_span, + _matching_output_value_span, + _matching_span, + _response_tool_calls, + _rig, + _spans, +) +from integration._support.client import Gateway +from integration._support.wire import Reply, Request + +_DEFAULT_METADATA: Final = { + "requester_ip_address": "127.0.0.1", + "user_api_key_user_id": "default_user_id", +} +_DEFAULT_METADATA_BAGGAGE: Final = frozenset({("litellm.metadata.user_api_key_user_id", "default_user_id")}) + + +def _request( + proxy: Gateway, + model: str, + marker: str, + *, + prompt: str = "weather in Paris?", + headers: dict[str, str] | None = None, + key: str | None = None, +) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + key=key, + headers=headers, + ) + + +def _assert_success_body(response: httpx.Response, marker: str, model: str) -> None: + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), model), response.text + + +def _upstream(marker: str) -> Callable[[Request], Reply]: + def reply(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + return reply + + +def _assert_output_tool_call(attributes: dict[str, str], marker: str) -> None: + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == f"call_{marker}", attributes + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.name"] == "lookup_weather", ( + attributes + ) + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] == '{"city": "Paris"}' + ), attributes + assert _json_messages(attributes["output.value"]) == [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": f"call_{marker}", + "type": "function", + "function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'}, + } + ], + } + ], attributes + + +def _assert_default_allowlist_attributes(attributes: dict[str, str], marker: str) -> None: + observed_baggage: Final = frozenset( + (key, value) for key, value in attributes.items() if key.startswith("litellm.metadata.") + ) + assert observed_baggage == _DEFAULT_METADATA_BAGGAGE, attributes + metadata: Final = _json_object(attributes["metadata"].encode()) + assert metadata == _DEFAULT_METADATA, attributes + assert "trace_marker" not in metadata, attributes + assert "litellm.metadata.trace_marker" not in attributes, attributes + _assert_output_tool_call(attributes, marker) + + +def test_arize_otel_v2_c1_absent_allowlist(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c1-" + uuid.uuid4().hex + with _rig( + gateway, + tmp_path, + _upstream(marker), + remove_environment=("LITELLM_OTEL_BAGGAGE_METADATA_KEYS",), + disabled_environment=("LITELLM_OTEL_BAGGAGE_METADATA_KEYS",), + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_span(rig.destination, marker) + _assert_default_allowlist_attributes(attributes, marker) + + +def test_arize_otel_v2_c2_empty_allowlist(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c2-" + uuid.uuid4().hex + with _rig(gateway, tmp_path, _upstream(marker), environment={"LITELLM_OTEL_BAGGAGE_METADATA_KEYS": ""}) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_span(rig.destination, marker) + assert "metadata" not in attributes, attributes + assert not any(key.startswith("litellm.metadata.") for key in attributes), attributes + _assert_output_tool_call(attributes, marker) + + +def test_arize_otel_v2_c3_absent_allowlisted_key(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c3-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"other": "value"}, + "cache": {"no-cache": True}, + }, + ) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_span(rig.destination, marker) + assert "metadata" not in attributes, attributes + assert "litellm.metadata.trace_marker" not in attributes, attributes + _assert_output_tool_call(attributes, marker) + + +def test_arize_otel_v2_c4_promotes_marker_and_alias(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c4-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker) + + with ( + _rig( + gateway, + tmp_path, + upstream, + environment={"LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.trace_marker,user_api_key_alias"}, + ) as rig, + rig.proxy.scenario() as scenario, + ): + key: Final = scenario.key(key_alias="alias-c4") + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + key=key, + ) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_marker_span(rig.destination, marker) + assert _json_object(attributes["metadata"].encode()) == { + "trace_marker": marker, + "user_api_key_alias": "alias-c4", + }, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + assert attributes["litellm.metadata.user_api_key_alias"] == "alias-c4", attributes + + +def test_arize_otel_v2_c5_yaml_allowlist_does_not_reach_preset_so_default_applies( + gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "c5-" + uuid.uuid4().hex + with _rig( + gateway, + tmp_path, + _upstream(marker), + callback_settings={"otel": {"baggage_metadata_keys": ["requester_metadata.trace_marker"]}}, + remove_environment=("LITELLM_OTEL_BAGGAGE_METADATA_KEYS",), + disabled_environment=("LITELLM_OTEL_BAGGAGE_METADATA_KEYS",), + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_marker_span(rig.destination, marker) + _assert_default_allowlist_attributes(attributes, marker) + + +def test_arize_otel_v2_c6_content_capture_disabled(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c6-" + uuid.uuid4().hex + with _rig( + gateway, + tmp_path, + _upstream(marker), + environment={ + "LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.trace_marker", + }, + remove_environment=("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",), + disabled_environment=("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",), + ) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + _assert_success_body(response, marker, rig.model) + attributes: Final = _matching_marker_span(rig.destination, marker) + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + assert "gen_ai.input.messages" not in attributes, attributes + assert "gen_ai.output.messages" not in attributes, attributes + assert not any( + key.startswith("llm.input_messages.") or key.startswith("llm.output_messages.") for key in attributes + ), attributes + assert "input.value" not in attributes, attributes + assert "output.value" not in attributes, attributes + assert not any(".tool_calls." in key for key in attributes), attributes + + +def test_arize_otel_v2_c7_key_and_team_logging_callbacks(gateway: Gateway, tmp_path: Path) -> None: + key_marker: Final = "c7-key-" + uuid.uuid4().hex + team_marker: Final = "c7-team-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + marker: Final = _chat_request_marker(request) + assert marker in (key_marker, team_marker), request + _assert_chat_request(request, messages=[{"role": "user", "content": marker}]) + return _chat_response(marker) + + logging_metadata: Final = {"logging": [{"callback_name": "arize", "callback_type": "success"}]} + with _rig(gateway, tmp_path, upstream) as rig, rig.proxy.scenario() as scenario: + key: Final = scenario.key(key_alias="key-c7", metadata=logging_metadata) + team: Final = scenario.team(metadata=logging_metadata) + team_key: Final = scenario.key(team_id=team) + key_response: Final = _request(rig.proxy, rig.model, key_marker, prompt=key_marker, key=key) + _assert_success_body(key_response, key_marker, rig.model) + key_attributes: Final = _matching_marker_span(rig.destination, key_marker) + team_response: Final = _request(rig.proxy, rig.model, team_marker, prompt=team_marker, key=team_key) + _assert_success_body(team_response, team_marker, rig.model) + team_attributes: Final = _matching_marker_span(rig.destination, team_marker) + assert _json_object(key_attributes["metadata"].encode()) == {"trace_marker": key_marker}, key_attributes + assert _json_object(team_attributes["metadata"].encode()) == {"trace_marker": team_marker}, team_attributes + assert key_attributes["litellm.metadata.trace_marker"] == key_marker, key_attributes + assert team_attributes["litellm.metadata.trace_marker"] == team_marker, team_attributes + _assert_output_tool_call(key_attributes, key_marker) + _assert_output_tool_call(team_attributes, team_marker) + + +def test_arize_otel_v2_c8_request_callback_disable(gateway: Gateway, tmp_path: Path) -> None: + control_marker: Final = "c8-control-" + uuid.uuid4().hex + disabled_marker: Final = "c8-disabled-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + marker: Final = _chat_request_marker(request) + assert marker in (control_marker, disabled_marker), marker + return _chat_response(marker) + + with _rig( + gateway, + tmp_path, + upstream, + litellm_settings={"allow_dynamic_callback_disabling": True}, + ) as rig: + control_response: Final = _request(rig.proxy, rig.model, control_marker, prompt=control_marker) + _assert_success_body(control_response, control_marker, rig.model) + control_attributes: Final = _matching_marker_span(rig.destination, control_marker) + _assert_output_tool_call(control_attributes, control_marker) + + disabled_response: Final = _request( + rig.proxy, + rig.model, + disabled_marker, + prompt=disabled_marker, + headers={"x-litellm-disable-callbacks": "arize"}, + ) + _assert_success_body(disabled_response, disabled_marker, rig.model) + disabled_spans: Final = tuple( + attributes + for attributes in _spans(rig.destination.drain()) + if attributes.get("openinference.span.kind") == "LLM" + ) + assert disabled_spans == (), disabled_spans + + +@pytest.mark.parametrize("failure_status", (401, 500)) +def test_arize_otel_v2_c9_upstream_failures_are_recorded(failure_status: int, gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c9-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + assert request.method == "POST", request.method + assert request.body, f"{request.method} {request.target}" + body: Final = _json_object(request.body) + assert body == { + "messages": [{"role": "user", "content": "weather in Paris?"}], + "model": "gpt-4o-mini", + }, body + return Reply( + status=failure_status, + body=b'{"error":{"message":"upstream failure"}}', + content_type="application/json", + ) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "metadata": {"trace_marker": marker, "failure_status": failure_status}, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == failure_status, response.text + error_body: Final = _json_object(response.content) + error_name: Final = {401: "AuthenticationError", 500: "InternalServerError"}[failure_status] + error_type: Final = {401: "authentication_error", 500: "internal_server_error"}[failure_status] + provider_message: Final = f"litellm.{error_name}: {error_name}: OpenAIException - upstream failure" + caller_message: Final = ( + f"{provider_message}\n\nLiteLLM: model group '{rig.model}' failed with the error above. " + "No fallback was attempted." + ) + assert error_body == { + "error": { + "message": caller_message, + "type": error_type, + "param": None, + "code": str(failure_status), + } + }, error_body + attributes: Final = _matching_marker_span(rig.destination, marker) + metadata: Final = attributes.get("metadata") + assert metadata is not None, attributes + assert _json_object(metadata.encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + assert attributes["error.message"] == provider_message, attributes + assert attributes["error.type"] == error_name, attributes + assert not any(".tool_calls." in key for key in attributes), attributes + + +def test_arize_otel_v2_c10_attribute_limit_keeps_tool_prefix(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "c10-" + uuid.uuid4().hex + calls: Final = _response_tool_calls(marker, ("Paris", "Berlin", "Rome", "Tokyo", "Oslo", "Lima", "Accra", "Delhi")) + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker, calls) + + with _rig(gateway, tmp_path, upstream, environment={"OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT": "57"}) as rig: + response: Final = _request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker, calls), rig.model), ( + response.text + ) + attributes: Final = _matching_output_value_span(rig.destination, marker) + indexes: Final = tuple( + int(key.split(".tool_calls.")[1].split(".")[0]) + for key in attributes + if ".tool_calls." in key and key.endswith(".tool_call.id") + ) + assert indexes == (0,), attributes + assert attributes["llm.output_messages.0.message.role"] == "assistant", attributes + fields: Final = ("id", "function.name", "function.arguments") + expected_tool_call_keys: Final = frozenset().union( + *( + frozenset(f"llm.output_messages.0.message.tool_calls.{index}.tool_call.{field}" for field in fields) + for index in indexes + ) + ) + observed_tool_call_keys: Final = frozenset(key for key in attributes if ".tool_calls." in key) + assert observed_tool_call_keys == expected_tool_call_keys, attributes + assert _json_messages(attributes["output.value"])[0]["tool_calls"] == calls, attributes + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + + history: Final = [{"role": "user", "content": f"history-{index}"} for index in range(40)] + history_marker: Final = marker + "-history" + + def history_upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=history, include_tools=False) + return _chat_plain_response(history_marker, "history retained") + + with _rig(gateway, tmp_path, history_upstream) as history_rig: + history_response: Final = history_rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": history_rig.model, + "messages": history, + "metadata": {"trace_marker": history_marker}, + "cache": {"no-cache": True}, + }, + ) + assert history_response.status_code == 200, history_response.text + assert _json_object(history_response.content) == _chat_caller_response( + _chat_plain_response(history_marker, "history retained"), history_rig.model + ), history_response.text + history_attributes: Final = _matching_marker_span(history_rig.destination, history_marker) + assert _json_object(history_attributes["metadata"].encode()) == {"trace_marker": history_marker}, ( + history_attributes + ) + assert history_attributes["litellm.metadata.trace_marker"] == history_marker, history_attributes + assert tuple(history_attributes[f"llm.input_messages.{index}.message.role"] for index in range(40)) == tuple( + str(message["role"]) for message in history + ) + assert tuple(history_attributes[f"llm.input_messages.{index}.message.content"] for index in range(40)) == tuple( + str(message["content"]) for message in history + ) + assert _json_messages(history_attributes["output.value"]) == [ + {"role": "assistant", "content": "history retained"} + ], history_attributes diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_sad_edge.py b/tests/integration/observability/test_arize_otel_v2_openinference_sad_edge.py new file mode 100644 index 00000000000..beaba0857a1 --- /dev/null +++ b/tests/integration/observability/test_arize_otel_v2_openinference_sad_edge.py @@ -0,0 +1,512 @@ +from __future__ import annotations + +import json +import uuid +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import Final + +import httpx +import pytest +from _openinference_support import ( + CHAT_TOOLS, + _assert_chat_request, + _chat_caller_response, + _chat_output_value, + _chat_request_marker, + _collect_marker_spans, + _json_messages, + _json_object, + _json_object_value, + _matching_marker_span, + _rig, + _span_attributes, + _spans, +) +from integration._support.client import Gateway +from integration._support.wire import Reply, Request +from pydantic import JsonValue + + +def _call( + proxy: Gateway, + model: str, + marker: str, + metadata: JsonValue | None = None, + key: str | None = None, + *, + prompt: str = "weather in Paris?", +) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": metadata if metadata is not None else {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + key=key, + ) + + +def _success(marker: str) -> Reply: + return Reply( + body=json.dumps( + { + "id": marker, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_" + marker, + "type": "function", + "function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'}, + } + ], + }, + } + ], + "usage": {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}, + } + ).encode() + ) + + +def _chat_message(body: bytes) -> dict[str, JsonValue]: + response: Final = _json_object(body) + choices: Final = response["choices"] + assert isinstance(choices, list) and len(choices) == 1 and isinstance(choices[0], dict), response + message: Final = choices[0]["message"] + assert isinstance(message, dict), response + return message + + +@pytest.mark.parametrize( + ("value", "expected_trace"), + ( + (7, "7"), + (["one", 2], None), + ("", None), + ("x" * 5000, "x" * 5000), + ({"enabled": True}, None), + ), +) +def test_arize_otel_v2_d1_metadata_value_shapes( + value: JsonValue, expected_trace: str | None, gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = "d1-" + uuid.uuid4().hex + metadata: Final = {"trace_marker": value} + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _success(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = _call(rig.proxy, rig.model, marker, metadata) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_success(marker), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + if expected_trace is None: + assert "metadata" not in attributes, attributes + assert "litellm.metadata.trace_marker" not in attributes, attributes + else: + assert json.loads(attributes["metadata"]) == {"trace_marker": expected_trace}, attributes + assert attributes["litellm.metadata.trace_marker"] == expected_trace, attributes + + +def test_arize_otel_v2_d2_duplicate_json_metadata_keys(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "d2-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _success(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + request_body: Final = ( + '{"model":"' + + rig.model + + '","messages":' + + json.dumps([{"role": "user", "content": "weather in Paris?"}]) + + ',"tools":' + + json.dumps(CHAT_TOOLS) + + "," + + '"tool_choice":{"type":"function","function":{"name":"lookup_weather"}},' + + '"metadata":{"trace_marker":"' + + marker + + '","trace_marker":"' + + marker + + '"},"cache":{"no-cache":true}}' + ) + response: Final = rig.proxy.client.post( + "/v1/chat/completions", + content=request_body, + headers={ + "authorization": f"Bearer {rig.proxy.key}", + "content-type": "application/json", + }, + ) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_success(marker), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + + +def test_arize_otel_v2_d4_unauthenticated_request_has_no_span(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "d4-" + uuid.uuid4().hex + with _rig(gateway, tmp_path, lambda _request: _success(marker)) as rig: + with httpx.Client(base_url=str(rig.proxy.client.base_url), trust_env=False) as client: + response: Final = client.post( + "/v1/chat/completions", + json={"model": rig.model, "messages": [{"role": "user", "content": marker}]}, + ) + assert response.status_code == 401, response.text + body: Final = _json_object(response.content) + assert body == { + "error": { + "message": "Authentication Error, No api key passed in.", + "type": "auth_error", + "param": "None", + "code": "401", + } + }, body + assert rig.provider.received.qsize() == 0 + spans: Final = tuple( + attributes + for attributes in _spans(rig.destination.drain()) + if attributes.get("openinference.span.kind") == "LLM" + ) + assert spans == (), "unauthenticated request exported an LLM span" + + +def test_arize_otel_v2_d5_unknown_model_leaves_proxy_ready(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "d5-" + uuid.uuid4().hex + unknown_model: Final = "unknown-model-" + uuid.uuid4().hex + with _rig(gateway, tmp_path, lambda _request: _success(marker)) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + {"model": unknown_model, "messages": [{"role": "user", "content": marker}]}, + ) + assert response.status_code == 400, response.text + body: Final = _json_object(response.content) + error_message: Final = ( + f"/chat/completions: Invalid model name passed in model={unknown_model}. " + "Call `/v1/models` to view available models for your key." + ) + assert body == { + "error": { + "message": error_message, + "type": "invalid_request_error", + "param": None, + "code": "400", + "provider_specific_fields": {"error": error_message}, + } + }, body + assert rig.provider.received.qsize() == 0 + readiness: Final = rig.proxy.client.get("/health/readiness") + assert readiness.status_code == 200, readiness.text + assert _json_object(readiness.content) == {"status": "healthy", "db": "connected"}, readiness.text + spans: Final = tuple( + attributes + for attributes in _spans(rig.destination.drain()) + if attributes.get("openinference.span.kind") == "LLM" + ) + assert spans == (), "unknown model exported an LLM span" + + +def test_arize_otel_v2_d6_sink_rejections_do_not_change_caller_response(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = tuple(f"d6-{status}-" + uuid.uuid4().hex for status in (403, 404)) + unrelated_marker: Final = "d6-unrelated-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + response_marker: Final = _chat_request_marker(request) + _assert_chat_request(request, messages=[{"role": "user", "content": response_marker}]) + return _success(response_marker) + + def sink(request: Request) -> Reply: + exported_markers: Final = tuple( + attributes.get("litellm.metadata.trace_marker") + for attributes in _span_attributes(request) + if "litellm.metadata.trace_marker" in attributes + ) + if markers[0] in exported_markers: + return Reply(status=403, body=b"rejected") + if markers[1] in exported_markers: + return Reply(status=404, body=b"rejected") + return Reply(body=b"", content_type="application/x-protobuf") + + with ( + _rig( + gateway, + tmp_path, + upstream, + destination_handler=sink, + ) as rig, + rig.proxy.scenario() as scenario, + ): + unrelated_key: Final = scenario.key(key_alias="unrelated-d6") + first: Final = _call(rig.proxy, rig.model, markers[0], prompt=markers[0]) + assert first.status_code == 200, first.text + assert _json_object(first.content) == _chat_caller_response(_success(markers[0]), rig.model), first.text + assert _matching_marker_span(rig.destination, markers[0])["litellm.metadata.trace_marker"] == markers[0] + second: Final = _call(rig.proxy, rig.model, markers[1], prompt=markers[1]) + assert second.status_code == 200, second.text + assert _json_object(second.content) == _chat_caller_response(_success(markers[1]), rig.model), second.text + assert _matching_marker_span(rig.destination, markers[1])["litellm.metadata.trace_marker"] == markers[1] + unrelated: Final = _call( + rig.proxy, + rig.model, + unrelated_marker, + key=unrelated_key, + prompt=unrelated_marker, + ) + assert unrelated.status_code == 200, unrelated.text + assert _json_object(unrelated.content) == _chat_caller_response(_success(unrelated_marker), rig.model), ( + unrelated.text + ) + assert ( + _matching_marker_span(rig.destination, unrelated_marker)["litellm.metadata.trace_marker"] + == unrelated_marker + ) + + +def test_arize_otel_v2_d7_missing_space_id_is_stable(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "d7-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _success(marker) + + with _rig( + gateway, + tmp_path, + upstream, + remove_environment=("ARIZE_SPACE_ID",), + disabled_environment=("ARIZE_SPACE_ID",), + ) as rig: + response: Final = _call(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_success(marker), rig.model), response.text + assert _matching_marker_span(rig.destination, marker)["litellm.metadata.trace_marker"] == marker + + +def test_arize_otel_v2_e1_uncached_request_exports_one_llm_span(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = tuple("e1-" + uuid.uuid4().hex for _ in range(3)) + + def upstream(request: Request) -> Reply: + marker: Final = _chat_request_marker(request) + assert marker in markers, request + _assert_chat_request(request, messages=[{"role": "user", "content": marker}]) + return _success(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + responses: Final = tuple(_call(rig.proxy, rig.model, marker, prompt=marker) for marker in markers) + assert all(response.status_code == 200 for response in responses), tuple( + response.text for response in responses + ) + assert tuple(_json_object(response.content) for response in responses) == tuple( + _chat_caller_response(_success(marker), rig.model) for marker in markers + ), responses + spans: Final = _collect_marker_spans(rig.destination, markers) + assert len(spans) == len(markers), spans + spans_by_id: Final = {span["gen_ai.response.id"]: span for span in spans} + assert frozenset(spans_by_id) == frozenset(markers), spans + assert all( + spans_by_id[marker]["gen_ai.response.id"] in response.text + for marker, response in zip(markers, responses, strict=True) + ), spans + + +def test_arize_otel_v2_e2_concurrent_unique_markers(gateway: Gateway, tmp_path: Path) -> None: + markers: Final = tuple("e2-" + uuid.uuid4().hex for _ in range(20)) + + def upstream(request: Request) -> Reply: + marker: Final = _chat_request_marker(request) + assert marker.startswith("e2-"), request + _assert_chat_request(request, messages=[{"role": "user", "content": marker}]) + return _success(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + with ThreadPoolExecutor(max_workers=20) as executor: + responses: Final = tuple( + executor.map( + lambda marker: _call(rig.proxy, rig.model, marker, prompt=marker), + markers, + ) + ) + assert all(response.status_code == 200 for response in responses), tuple( + response.text for response in responses + ) + assert tuple(_json_object(response.content) for response in responses) == tuple( + _chat_caller_response(_success(marker), rig.model) for marker in markers + ), responses + spans: Final = _collect_marker_spans(rig.destination, markers) + assert len(spans) == len(markers), spans + assert tuple( + _json_object(next(span for span in spans if marker in span.values())["metadata"].encode()) + for marker in markers + ) == tuple({"trace_marker": marker} for marker in markers), spans + assert ( + tuple( + next(span for span in spans if marker in span.values())["litellm.metadata.trace_marker"] + for marker in markers + ) + == markers + ), spans + + +@pytest.mark.parametrize( + "shape", + ("object-arguments", "missing-name", "non-dict-call", "null-tool-calls", "integer-id"), +) +def test_arize_otel_v2_d3_malformed_tool_calls_are_normalized(shape: str, gateway: Gateway, tmp_path: Path) -> None: + marker: Final = f"d3-{shape}-" + uuid.uuid4().hex + + def malformed_reply() -> Reply: + call: Final = { + "id": 17 if shape == "integer-id" else f"call_{marker}", + "type": "function", + "function": { + **({} if shape == "missing-name" else {"name": "lookup_weather"}), + "arguments": {"city": "Paris"} if shape == "object-arguments" else '{"city": "Paris"}', + }, + } + tool_calls: Final[JsonValue] = ( + None if shape == "null-tool-calls" else ["not-a-call"] if shape == "non-dict-call" else [call] + ) + message: Final = { + "role": "assistant", + "content": None, + "tool_calls": tool_calls, + } + return Reply( + body=json.dumps( + { + "id": marker, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "finish_reason": "tool_calls", "message": message}], + "usage": {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}, + } + ).encode() + ) + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return malformed_reply() + + with _rig(gateway, tmp_path, upstream) as rig: + proxy_log: Final = rig.owned.log + response: Final = _call(rig.proxy, rig.model, marker) + if shape == "non-dict-call": + assert response.status_code == 400, response.text + error_body: Final = _json_object(response.content) + assert frozenset(error_body) == frozenset({"error"}), error_body + error: Final = _json_object_value(error_body["error"]) + assert frozenset(error) == frozenset({"type", "code", "param", "message"}), error + assert error["type"] == "invalid_request_error", error + assert error["code"] == "400", error + assert error["param"] is None, error + message: Final = error["message"] + assert isinstance(message, str), error + assert "AttributeError: 'str' object has no attribute 'get'" in message, error + spans: Final = tuple(_spans(rig.destination.drain())) + assert all(not any(".tool_calls." in key for key in attributes) for attributes in spans), spans + readiness: Final = rig.proxy.client.get("/health/readiness") + assert readiness.status_code == 200, readiness.text + assert _json_object(readiness.content) == {"status": "healthy", "db": "connected"}, readiness.text + else: + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(malformed_reply(), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + indexed_prefix: Final = "llm.output_messages.0.message.tool_calls.0.tool_call." + expected_fields: Final = ( + ("id", "function.name", "function.arguments"), + ("id", "function.arguments"), + (), + ("function.name", "function.arguments"), + )[("object-arguments", "missing-name", "null-tool-calls", "integer-id").index(shape)] + expected_keys: Final = frozenset(indexed_prefix + field for field in expected_fields) + observed_keys: Final = frozenset(key for key in attributes if ".tool_calls." in key) + assert observed_keys == expected_keys, attributes + if shape == "object-arguments": + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == f"call_{marker}", ( + attributes + ) + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.name"] == "lookup_weather" + ), attributes + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] + == '{"city": "Paris"}' + ), attributes + elif shape == "missing-name": + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == f"call_{marker}", ( + attributes + ) + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] + == '{"city": "Paris"}' + ), attributes + elif shape == "integer-id": + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.name"] == "lookup_weather" + ), attributes + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] + == '{"city": "Paris"}' + ), attributes + assert _json_messages(attributes["output.value"]) == _json_messages( + _chat_output_value(malformed_reply()) + ), attributes + assert "Exception while exporting Span batch" not in proxy_log.read_text(), proxy_log.read_text() + + +@pytest.mark.parametrize("shape", ("empty", "null", "missing")) +def test_arize_otel_v2_e3_empty_or_missing_tool_calls_never_indexed( + shape: str, gateway: Gateway, tmp_path: Path +) -> None: + marker: Final = f"e3-{shape}-" + uuid.uuid4().hex + message: Final = ( + {"role": "assistant", "content": None, "tool_calls": []} + if shape == "empty" + else {"role": "assistant", "content": None, "tool_calls": None} + if shape == "null" + else {"role": "assistant", "content": None} + ) + expected_response: Final[dict[str, JsonValue]] = { + "id": marker, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "finish_reason": "stop", "message": message}], + "usage": {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}, + } + + expected_reply: Final = Reply(body=json.dumps(expected_response).encode()) + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return expected_reply + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = _call(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + response_body: Final = _json_object(response.content) + assert response_body == _chat_caller_response(expected_reply, rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + assert not any(".tool_calls." in key for key in attributes), attributes + assert "tool_calls" not in _json_messages(attributes["output.value"])[0], attributes diff --git a/tests/integration/observability/test_arize_otel_v2_openinference_spans.py b/tests/integration/observability/test_arize_otel_v2_openinference_spans.py index 129a377bb76..f78c48d0f10 100644 --- a/tests/integration/observability/test_arize_otel_v2_openinference_spans.py +++ b/tests/integration/observability/test_arize_otel_v2_openinference_spans.py @@ -1,299 +1,901 @@ +from __future__ import annotations + +import asyncio import json import uuid -from collections.abc import Callable, Iterator -from contextlib import contextmanager +from itertools import chain from pathlib import Path from typing import Final -import yaml -from integration._support.client import Gateway, eventually -from integration._support.process import owned_proxy -from integration._support.wire import Reply, Request, Wire, wire_server -from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest +import anthropic +import httpx +import openai +import pytest +from _openinference_support import ( + CHAT_TOOLS, + RESPONSES_TOOLS, + Rig, + _anthropic_stream_response, + _assert_chat_request, + _assert_messages_request, + _assert_responses_request, + _assert_tool_span, + _chat_cache_hit_caller_stream, + _chat_caller_response, + _chat_caller_stream, + _chat_plain_response, + _chat_response, + _chat_stream_response, + _chat_tool_call, + _json_messages, + _json_object, + _matching_marker_span, + _messages_caller_response, + _messages_caller_stream, + _messages_caller_stream_response, + _normalize_chat_caller_stream, + _normalize_responses_caller_body, + _normalize_responses_caller_stream, + _response_tool_calls, + _responses_caller_response, + _responses_caller_stream, + _responses_response, + _responses_stream_response, + _rig, +) +from integration._support.client import Gateway +from integration._support.wire import Reply, Request +from pydantic import JsonValue -def _span_attributes(request: Request) -> Iterator[dict[str, str]]: - if request.headers.get("content-type") != "application/x-protobuf": - return - batch: Final = ExportTraceServiceRequest.FromString(request.body) - for resource_spans in batch.resource_spans: - for scope_spans in resource_spans.scope_spans: - for span in scope_spans.spans: - yield {attribute.key: attribute.value.string_value for attribute in span.attributes} +def _chat_upstream(request: Request) -> None: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) -def _matching_llm_spans( - requests: tuple[Request, ...], - marker: str, - marker_key: str, -) -> Iterator[dict[str, str]]: - for request in requests: - for attributes in _span_attributes(request): - if attributes.get("openinference.span.kind") == "LLM" and attributes.get(marker_key) == marker: - yield attributes +def _responses_upstream(request: Request, marker: str, *, stream: bool = False) -> None: + _assert_responses_request(request, marker=marker, stream=stream) -def _arize_config(tmp_path: Path) -> Path: - config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) - arize_config: Final = { - **config, - "litellm_settings": { - **config["litellm_settings"], - "callbacks": ["arize"], - }, - } - config_path: Final = tmp_path / "arize-otel-v2.yaml" - config_path.write_text(yaml.safe_dump(arize_config)) - return config_path +def _messages_upstream(request: Request, marker: str, *, stream: bool = False) -> None: + _assert_messages_request(request, marker=marker, stream=stream) -def _arize_environment(destination: Wire) -> dict[str, str]: - return { - "LITELLM_OTEL_V2": "1", - "ARIZE_HTTP_ENDPOINT": destination.url + "/v1/traces", - "ARIZE_SPACE_ID": "integration-space", - "ARIZE_API_KEY": "integration-arize-key", - "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only", - "LITELLM_OTEL_BAGGAGE_METADATA_KEYS": "requester_metadata.trace_marker", - } - - -def _collector(_request: Request) -> Reply: - return Reply(body=b"", content_type="application/x-protobuf") - - -@contextmanager -def _arize_proxy( - gateway: Gateway, - tmp_path: Path, - upstream: Callable[[Request], Reply], -) -> Iterator[tuple[Gateway, Wire, Wire]]: - with wire_server(upstream) as provider, wire_server(_collector) as destination: - config_path: Final = _arize_config(tmp_path) - environment: Final = _arize_environment(destination) - with owned_proxy(gateway, tmp_path, environment, config=config_path) as candidate: - yield candidate, provider, destination - - -def _matching_span( - destination: Wire, - marker: str, - marker_key: str = "gen_ai.response.id", -) -> dict[str, str]: - return eventually( - lambda: tuple(_matching_llm_spans(destination.drain(), marker, marker_key)), - lambda spans: len(spans) == 1, - seconds=30, - )[0] - - -def _assert_tool_call_span(span_attributes: dict[str, str], marker: str) -> None: - tool_call: Final = "llm.output_messages.0.message.tool_calls.0.tool_call." - keys: Final = ( - tool_call + "id", - tool_call + "function.name", - tool_call + "function.arguments", - "metadata", - "litellm.metadata.trace_marker", - ) - observed: Final = { - key: json.loads(span_attributes[key]) - if key in (tool_call + "function.arguments", "metadata") and key in span_attributes - else span_attributes.get(key) - for key in keys - } - expected: Final = { - tool_call + "id": "call_" + marker, - tool_call + "function.name": "lookup_weather", - tool_call + "function.arguments": {"city": "Paris"}, - "metadata": {"trace_marker": marker}, - "litellm.metadata.trace_marker": marker, - } - expected_output: Final = [ - { - "role": "assistant", - "content": None, - "tool_calls": [ - { - "id": "call_" + marker, - "type": "function", - "function": { - "name": "lookup_weather", - "arguments": '{"city": "Paris"}', - }, - } - ], - } - ] - actual: Final = {"attributes": observed, "output.value": json.loads(span_attributes["output.value"])} - expected_values: Final = {"attributes": expected, "output.value": expected_output} - assert actual == expected_values, f"Arize OTel v2 span values for {marker}: {actual!r}" - - -def test_arize_otel_v2_llm_span_carries_openinference_tool_calls_and_metadata(gateway: Gateway, tmp_path: Path) -> None: - marker: Final = "arize-otel-" + uuid.uuid4().hex - - def upstream(request: Request) -> Reply: - assert request.target.endswith("/chat/completions"), request.target - body: Final = json.loads(request.body) - expected_tools: Final = [ +def _messages_response(identity: str) -> Reply: + return Reply( + body=json.dumps( { - "type": "function", - "function": { + "id": identity, + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "content": [ + { + "type": "tool_use", + "id": "call_" + identity, + "name": "lookup_weather", + "input": {"city": "Paris"}, + } + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 11, "output_tokens": 4}, + } + ).encode() + ) + + +def _assert_tool_span_for_marker(attributes: dict[str, str], marker: str, *, content: str | None = None) -> None: + _assert_tool_span( + attributes, + marker=marker, + output=[ + { + "role": "assistant", + "content": content, + "tool_calls": [ + { + "id": "call_" + marker, + "type": "function", + "function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'}, + } + ], + } + ], + calls=[("call_" + marker, "lookup_weather", {"city": "Paris"})], + metadata={"trace_marker": marker}, + baggage={"trace_marker": marker}, + ) + + +def _chat_request( + proxy: Gateway, model: str, marker: str, *, stream: bool = False, no_cache: bool = True +) -> httpx.Response: + return proxy.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + **({"stream": True} if stream else {}), + **({"cache": {"no-cache": True}} if no_cache else {}), + }, + ) + + +def _responses_request(proxy: Gateway, model: str, marker: str) -> httpx.Response: + return proxy.request( + "POST", + "/v1/responses", + { + "model": model, + "input": "weather in Paris?", + "tools": RESPONSES_TOOLS, + "tool_choice": {"type": "function", "name": "lookup_weather"}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) + + +def _messages_request(proxy: Gateway, model: str, marker: str) -> httpx.Response: + return proxy.request( + "POST", + "/v1/messages", + { + "model": model, + "max_tokens": 64, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "tools": [ + { "name": "lookup_weather", "description": "Get weather", - "parameters": { + "input_schema": { "type": "object", "properties": {"city": {"type": "string"}}, }, - }, - } - ] - assert body.get("messages") == [{"role": "user", "content": "weather in Paris?"}], ( - f"Unexpected chat request messages: {body!r}" - ) - assert body.get("tools") == expected_tools, f"Unexpected chat request tools: {body!r}" - assert body.get("tool_choice") == { - "type": "function", - "function": {"name": "lookup_weather"}, - }, f"Unexpected chat request tool_choice: {body!r}" - return Reply( - body=json.dumps( - { - "id": marker, - "object": "chat.completion", - "created": 1, - "model": "gpt-4o-mini", - "choices": [ - { - "index": 0, - "finish_reason": "tool_calls", - "message": { - "role": "assistant", - "content": None, - "tool_calls": [ - { - "id": "call_" + marker, - "type": "function", - "function": { - "name": "lookup_weather", - "arguments": '{"city": "Paris"}', - }, - } - ], - }, - } - ], - "usage": {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15}, } - ).encode() - ) + ], + "tool_choice": {"type": "auto"}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) - with _arize_proxy(gateway, tmp_path, upstream) as (candidate, provider, destination): - with candidate.scenario() as scenario: - model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=provider.url + "/v1") - response: Final = candidate.request( - "POST", - "/v1/chat/completions", - { - "model": model, - "messages": [{"role": "user", "content": "weather in Paris?"}], - "tools": [ - { - "type": "function", - "function": { - "name": "lookup_weather", - "description": "Get weather", - "parameters": { - "type": "object", - "properties": {"city": {"type": "string"}}, - }, - }, - } - ], - "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, - "metadata": {"trace_marker": marker}, - "cache": {"no-cache": True}, - }, + +def _openai_client(proxy: Gateway) -> openai.OpenAI: + return openai.OpenAI(base_url=str(proxy.client.base_url) + "/v1", api_key=proxy.key, max_retries=0) + + +def _async_openai_client(proxy: Gateway) -> openai.AsyncOpenAI: + return openai.AsyncOpenAI(base_url=str(proxy.client.base_url) + "/v1", api_key=proxy.key, max_retries=0) + + +def test_arize_otel_v2_a1_chat_sync_sdk(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a1-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _chat_upstream(request) + body: Final = _json_object(request.body) + assert body == { + "messages": [{"role": "user", "content": "weather in Paris?"}], + "model": "gpt-4o-mini", + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "tools": CHAT_TOOLS, + }, body + return _chat_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + client: Final = _openai_client(rig.proxy) + response: Final = client.chat.completions.create( + model=rig.model, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, + ) + assert response.id == marker, response + assert response.model_dump(mode="json", exclude_unset=True) == _chat_caller_response( + _chat_response(marker), rig.model + ), response + calls: Final = response.choices[0].message.tool_calls + assert calls is not None and len(calls) == 1, response + assert (calls[0].id, calls[0].function.name, calls[0].function.arguments) == ( + f"call_{marker}", + "lookup_weather", + '{"city": "Paris"}', + ), response + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_a2_chat_async_sdk(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a2-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _chat_upstream(request) + return _chat_response(marker) + + async def call() -> None: + with _rig(gateway, tmp_path, upstream) as rig: + client: Final = _async_openai_client(rig.proxy) + response: Final = await client.chat.completions.create( + model=rig.model, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, ) - assert response.status_code == 200, response.text - _assert_tool_call_span(_matching_span(destination, marker), marker) + assert response.id == marker, response + assert response.model_dump(mode="json", exclude_unset=True) == _chat_caller_response( + _chat_response(marker), rig.model + ), response + calls: Final = response.choices[0].message.tool_calls + assert calls is not None and len(calls) == 1, response + assert (calls[0].id, calls[0].function.name, calls[0].function.arguments) == ( + f"call_{marker}", + "lookup_weather", + '{"city": "Paris"}', + ), response + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + asyncio.run(call()) + + +def test_arize_otel_v2_a5_responses_sync_sdk(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a5-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _responses_upstream(request, marker) + return _responses_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + client: Final = _openai_client(rig.proxy) + response: Final = client.responses.create( + model=rig.model, + input="weather in Paris?", + tools=RESPONSES_TOOLS, + tool_choice={"type": "function", "name": "lookup_weather"}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, + ) + assert response.id.startswith("resp_"), response + assert _normalize_responses_caller_body(response.model_dump(mode="json", exclude_unset=True)) == ( + _responses_caller_response(_responses_response(marker), rig.model) + ), response + assert (response.status, response.model) == ("completed", rig.model), response + assert ( + response.output[0].type, + response.output[0].call_id, + response.output[0].name, + response.output[0].arguments, + ) == ("function_call", f"call_{marker}", "lookup_weather", '{"city": "Paris"}'), response + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_a6_responses_async_streaming_sdk(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a6-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _responses_upstream(request, marker, stream=True) + return _responses_stream_response(marker, (_chat_tool_call(marker),)) + + async def call() -> None: + with _rig(gateway, tmp_path, upstream) as rig: + client: Final = _async_openai_client(rig.proxy) + stream: Final = await client.responses.create( + model=rig.model, + input="weather in Paris?", + tools=RESPONSES_TOOLS, + tool_choice={"type": "function", "name": "lookup_weather"}, + stream=True, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, + ) + events: Final = tuple([event async for event in stream]) + assert events[-1].type == "response.completed", events + assert events[-1].response.id.startswith("resp_"), events[-1] + assert _normalize_responses_caller_stream( + tuple(event.model_dump(mode="json", exclude_unset=True) for event in events) + ) == _responses_caller_stream(_responses_stream_response(marker, (_chat_tool_call(marker),)), rig.model), ( + events + ) + call: Final = events[-1].response.output[0] + assert (call.type, call.call_id, call.name, call.arguments) == ( + "function_call", + f"call_{marker}", + "lookup_weather", + '{"city": "Paris"}', + ), events[-1] + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + asyncio.run(call()) + + +def test_arize_otel_v2_a7_messages_sync_sdk(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a7-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _messages_upstream(request, marker) + return _messages_response(marker) + + with _rig(gateway, tmp_path, upstream, model_name="anthropic/claude-opus-5-5", api_base_suffix="") as rig: + client: Final = anthropic.Anthropic( + base_url=str(rig.proxy.client.base_url), api_key=rig.proxy.key, max_retries=0 + ) + response: Final = client.messages.create( + model=rig.model, + max_tokens=64, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=[ + { + "name": "lookup_weather", + "description": "Get weather", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}, + } + ], + tool_choice={"type": "auto"}, + metadata={"trace_marker": marker}, + ) + assert response.id == marker, response + assert response.model_dump(mode="json", exclude_unset=True) == _messages_caller_response( + _messages_response(marker), rig.model + ), response + call: Final = response.content[0] + assert (call.type, call.id, call.name, call.input) == ( + "tool_use", + f"call_{marker}", + "lookup_weather", + {"city": "Paris"}, + ), response + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_a3_chat_streaming(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a3-" + uuid.uuid4().hex + stream_reply: Final = _chat_stream_response(marker, (_chat_tool_call(marker),), include_usage=False) + + def upstream(request: Request) -> Reply: + _assert_chat_request( + request, + messages=[{"role": "user", "content": "weather in Paris?"}], + stream=True, + stream_options={"include_usage": False}, + ) + return stream_reply + + with _rig(gateway, tmp_path, upstream, general_settings={"always_include_stream_usage": False}) as rig: + client: Final = _openai_client(rig.proxy) + stream: Final = client.chat.completions.create( + model=rig.model, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + stream=True, + stream_options={"include_usage": False}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, + ) + chunks: Final = tuple(stream) + assert chunks[0].id == marker and chunks[-1].id == marker, chunks + assert _normalize_chat_caller_stream( + tuple(chunk.model_dump(mode="json", exclude_unset=True) for chunk in chunks) + ) == _chat_caller_stream(stream_reply, rig.model), chunks + assert chunks[-1].choices[0].finish_reason == "tool_calls", chunks + tool_call_deltas: Final = tuple( + chain.from_iterable(chunk.choices[0].delta.tool_calls or () for chunk in chunks) + ) + assert len(tool_call_deltas) == 2, chunks + assert ( + tool_call_deltas[0].id, + tool_call_deltas[0].function.name, + tool_call_deltas[1].function.arguments, + ) == (f"call_{marker}", "lookup_weather", '{"city": "Paris"}'), chunks + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_a4_chat_async_streaming(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a4-" + uuid.uuid4().hex + stream_reply: Final = _chat_stream_response(marker, (_chat_tool_call(marker),), include_usage=False) + + def upstream(request: Request) -> Reply: + _assert_chat_request( + request, + messages=[{"role": "user", "content": "weather in Paris?"}], + stream=True, + stream_options={"include_usage": False}, + ) + return stream_reply + + async def call() -> None: + with _rig(gateway, tmp_path, upstream, general_settings={"always_include_stream_usage": False}) as rig: + client: Final = _async_openai_client(rig.proxy) + stream: Final = await client.chat.completions.create( + model=rig.model, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + stream=True, + stream_options={"include_usage": False}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": True}}, + ) + chunks: Final = tuple([chunk async for chunk in stream]) + assert chunks[0].id == marker and chunks[-1].id == marker, chunks + assert _normalize_chat_caller_stream( + tuple(chunk.model_dump(mode="json", exclude_unset=True) for chunk in chunks) + ) == _chat_caller_stream(stream_reply, rig.model), chunks + assert chunks[-1].choices[0].finish_reason == "tool_calls", chunks + tool_call_deltas: Final = tuple( + chain.from_iterable(chunk.choices[0].delta.tool_calls or () for chunk in chunks) + ) + assert len(tool_call_deltas) == 2, chunks + assert ( + tool_call_deltas[0].id, + tool_call_deltas[0].function.name, + tool_call_deltas[1].function.arguments, + ) == (f"call_{marker}", "lookup_weather", '{"city": "Paris"}'), chunks + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + asyncio.run(call()) + + +def test_arize_otel_v2_a8_messages_streaming(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a8-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _messages_upstream(request, marker, stream=True) + return _anthropic_stream_response(marker) + + async def call() -> None: + with _rig(gateway, tmp_path, upstream, model_name="anthropic/claude-opus-5-5", api_base_suffix="") as rig: + client: Final = anthropic.AsyncAnthropic( + base_url=str(rig.proxy.client.base_url), api_key=rig.proxy.key, max_retries=0 + ) + async with client.messages.stream( + model=rig.model, + max_tokens=64, + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=[ + { + "name": "lookup_weather", + "description": "Get weather", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}, + } + ], + tool_choice={"type": "auto"}, + metadata={"trace_marker": marker}, + ) as stream: + events: Final = tuple([event async for event in stream]) + response: Final = await stream.get_final_message() + assert events[-1].type == "message_stop", events + assert response.id == marker, response + assert response.model_dump(mode="json", exclude_unset=True) == _messages_caller_stream_response( + _messages_response(marker), rig.model + ), response + assert tuple(event.model_dump(mode="json", exclude_unset=True) for event in events) == ( + _messages_caller_stream( + _anthropic_stream_response(marker), + rig.model, + final_message=_messages_caller_stream_response(_messages_response(marker), rig.model), + ) + ), events + call: Final = response.content[0] + assert (call.type, call.id, call.name, call.input) == ( + "tool_use", + f"call_{marker}", + "lookup_weather", + {"city": "Paris"}, + ), response + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker, content="") + + asyncio.run(call()) + + +def test_arize_otel_v2_llm_span_carries_openinference_tool_calls_and_metadata(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a9-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _chat_upstream(request) + return _chat_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = _chat_request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + _assert_tool_span_for_marker(attributes, marker) def test_arize_otel_v2_responses_span_carries_openinference_tool_calls_and_metadata( gateway: Gateway, tmp_path: Path ) -> None: - marker: Final = "arize-otel-responses-" + uuid.uuid4().hex + marker: Final = "a10-" + uuid.uuid4().hex def upstream(request: Request) -> Reply: - assert request.target.endswith("/responses"), request.target - body: Final = json.loads(request.body) - expected_tools: Final = [ - { - "type": "function", - "name": "lookup_weather", - "description": "Get weather", - "parameters": { - "type": "object", - "properties": {"city": {"type": "string"}}, - }, - } - ] - assert body.get("input") == "weather in Paris?", f"Unexpected Responses request input: {body!r}" - assert body.get("tools") == expected_tools, f"Unexpected Responses request tools: {body!r}" - assert body.get("tool_choice") == { - "type": "function", - "name": "lookup_weather", - }, f"Unexpected Responses request tool_choice: {body!r}" - return Reply( - body=json.dumps( - { - "id": marker, - "object": "response", - "created_at": 1, - "status": "completed", - "model": "gpt-4o-mini", - "output": [ - { - "type": "function_call", - "id": "fc_" + marker, - "call_id": "call_" + marker, - "name": "lookup_weather", - "arguments": '{"city": "Paris"}', - "status": "completed", - } - ], - "usage": {"input_tokens": 11, "output_tokens": 4, "total_tokens": 15}, - } - ).encode() + _responses_upstream(request, marker) + return _responses_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = _responses_request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + response_body: Final = _json_object(response.content) + assert _normalize_responses_caller_body(response_body) == _responses_caller_response( + _responses_response(marker), rig.model + ), response.text + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + + +def test_arize_otel_v2_a11_parallel_output_tool_calls(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a11-" + uuid.uuid4().hex + calls: Final = _response_tool_calls(marker, ("Paris", "Berlin")) + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=[{"role": "user", "content": "weather in Paris?"}]) + return _chat_response(marker, calls) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = _chat_request(rig.proxy, rig.model, marker) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker, calls), rig.model), ( + response.text + ) + attributes: Final = _matching_marker_span(rig.destination, marker) + expected_calls: Final = ( + ("call_" + marker + "-Paris", "lookup_weather", {"city": "Paris"}), + ("call_" + marker + "-Berlin", "lookup_weather", {"city": "Berlin"}), + ) + _assert_tool_span( + attributes, + marker=marker, + output=[{"role": "assistant", "content": None, "tool_calls": calls}], + calls=expected_calls, + metadata={"trace_marker": marker}, + baggage={"trace_marker": marker}, ) - with _arize_proxy(gateway, tmp_path, upstream) as (candidate, provider, destination): - with candidate.scenario() as scenario: - model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=provider.url + "/v1") - response: Final = candidate.request( - "POST", - "/v1/responses", + +def test_arize_otel_v2_a12_plain_text_has_metadata_without_tool_calls(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a12-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + _assert_chat_request( + request, + messages=[{"role": "user", "content": "weather in Paris?"}], + include_tools=False, + ) + return _chat_plain_response(marker, "The weather is clear") + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": "weather in Paris?"}], + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response( + _chat_plain_response(marker, "The weather is clear"), rig.model + ), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + assert not any(".tool_calls." in key for key in attributes), attributes + assert _json_messages(attributes["output.value"]) == [ + {"role": "assistant", "content": "The weather is clear"} + ], attributes + + +def test_arize_otel_v2_a13_multiturn_input_tool_calls(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a13-" + uuid.uuid4().hex + messages: Final = [ + {"role": "user", "content": "weather in Paris?"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ { - "model": model, - "input": "weather in Paris?", - "tools": [ - { - "type": "function", - "name": "lookup_weather", - "description": "Get weather", - "parameters": { - "type": "object", - "properties": {"city": {"type": "string"}}, - }, - } - ], - "tool_choice": {"type": "function", "name": "lookup_weather"}, - "metadata": {"trace_marker": marker}, - "cache": {"no-cache": True}, - }, + "id": "call-prior", + "type": "function", + "function": {"name": "lookup_weather", "arguments": '{"city": "Paris"}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "call-prior", "content": '{"temperature": 20}'}, + ] + upstream_messages: Final = [ + {key: value for key, value in message.items() if value is not None} for message in messages + ] + + def upstream(request: Request) -> Reply: + _assert_chat_request(request, messages=upstream_messages) + return _chat_response(marker) + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": messages, + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(_chat_response(marker), rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + _assert_tool_span_for_marker(attributes, marker) + assert not any(key.startswith("llm.input_messages.") and ".tool_calls." in key for key in attributes), ( + attributes + ) + assert tuple(attributes[f"llm.input_messages.{index}.message.role"] for index in range(3)) == ( + "user", + "assistant", + "tool", + ), attributes + assert tuple(attributes[f"llm.input_messages.{index}.message.content"] for index in (0, 2)) == ( + "weather in Paris?", + '{"temperature": 20}', + ), attributes + expected_input_value: Final = [ + messages[0], + messages[1], + {"role": "tool", "content": '{"temperature": 20}'}, + ] + assert _json_messages(attributes["input.value"]) == expected_input_value, attributes + + +def test_arize_otel_v2_a14_two_choices_each_with_tool_calls(gateway: Gateway, tmp_path: Path) -> None: + marker: Final = "a14-" + uuid.uuid4().hex + first_call: Final = _response_tool_calls(marker, ("Paris",))[0] + second_call: Final = _response_tool_calls(marker, ("Berlin",))[0] + + def choice(index: int, call: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + "index": index, + "finish_reason": "tool_calls", + "message": {"role": "assistant", "content": None, "tool_calls": [call]}, + } + + expected_response: Final = Reply( + body=json.dumps( + { + "id": marker, + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [choice(0, first_call), choice(1, second_call)], + "usage": {"prompt_tokens": 11, "completion_tokens": 4, "total_tokens": 15}, + } + ).encode() + ) + + def upstream(request: Request) -> Reply: + _assert_chat_request( + request, + messages=[{"role": "user", "content": "weather in Paris and Berlin?"}], + n=2, + ) + return expected_response + + with _rig(gateway, tmp_path, upstream) as rig: + response: Final = rig.proxy.request( + "POST", + "/v1/chat/completions", + { + "model": rig.model, + "messages": [{"role": "user", "content": "weather in Paris and Berlin?"}], + "tools": CHAT_TOOLS, + "tool_choice": {"type": "function", "function": {"name": "lookup_weather"}}, + "n": 2, + "metadata": {"trace_marker": marker}, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == 200, response.text + assert _json_object(response.content) == _chat_caller_response(expected_response, rig.model), response.text + attributes: Final = _matching_marker_span(rig.destination, marker) + assert _json_messages(attributes["output.value"]) == [ + {"role": "assistant", "content": None, "tool_calls": [first_call]}, + {"role": "assistant", "content": None, "tool_calls": [second_call]}, + ], attributes + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.id"] == str(first_call["id"]), ( + attributes + ) + assert attributes["llm.output_messages.1.message.tool_calls.0.tool_call.id"] == str(second_call["id"]), ( + attributes + ) + assert attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.name"] == "lookup_weather", ( + attributes + ) + assert ( + attributes["llm.output_messages.0.message.tool_calls.0.tool_call.function.arguments"] == '{"city": "Paris"}' + ), attributes + assert attributes["llm.output_messages.1.message.tool_calls.0.tool_call.function.name"] == "lookup_weather", ( + attributes + ) + assert ( + attributes["llm.output_messages.1.message.tool_calls.0.tool_call.function.arguments"] + == '{"city": "Berlin"}' + ), attributes + assert _json_object(attributes["metadata"].encode()) == {"trace_marker": marker}, attributes + assert attributes["litellm.metadata.trace_marker"] == marker, attributes + + +def _cache_call( + rig: Rig, surface: str, marker: str, *, cache_hit: bool = False +) -> tuple[tuple[str, str, str], str, httpx.Headers]: + match surface: + case "chat": + client: Final = _openai_client(rig.proxy) + raw: Final = client.chat.completions.with_raw_response.create( + model=rig.model, + messages=[{"role": "user", "content": marker}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": False}}, ) - assert response.status_code == 200, response.text - _assert_tool_call_span( - _matching_span(destination, marker, marker_key="litellm.metadata.trace_marker"), marker + response: Final = raw.parse() + assert response.model_dump(mode="json", exclude_unset=True) == _chat_caller_response( + _chat_response(marker), rig.model + ), response + assert response.choices[0].message.tool_calls is not None, response + call: Final = response.choices[0].message.tool_calls[0] + return (call.id, call.function.name, call.function.arguments), response.id, raw.headers + case "chat-stream": + client: Final = _openai_client(rig.proxy) + with client.chat.completions.with_streaming_response.create( + model=rig.model, + messages=[{"role": "user", "content": marker}], + tools=CHAT_TOOLS, + tool_choice={"type": "function", "function": {"name": "lookup_weather"}}, + stream=True, + stream_options={"include_usage": False}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": False}}, + ) as raw: + chunks: Final = tuple(raw.parse()) + expected_chunks: Final = ( + _chat_cache_hit_caller_stream(marker, rig.model, (_chat_tool_call(marker),)) + if cache_hit + else _chat_caller_stream( + _chat_stream_response(marker, (_chat_tool_call(marker),), include_usage=False), rig.model + ) + ) + assert ( + _normalize_chat_caller_stream( + tuple(chunk.model_dump(mode="json", exclude_unset=True) for chunk in chunks) + ) + == expected_chunks + ), chunks + calls: Final = tuple(chain.from_iterable(chunk.choices[0].delta.tool_calls or () for chunk in chunks)) + if cache_hit: + assert len(calls) == 1, chunks + call: Final = calls[0] + assert ( + call.id is not None and call.function.name is not None and call.function.arguments is not None + ), chunks + return ( + (call.id, call.function.name, call.function.arguments), + chunks[0].id, + raw.headers, + ) + assert len(calls) == 2, chunks + call: Final = calls[0] + arguments_call: Final = calls[1] + assert ( + call.id is not None + and call.function.name is not None + and arguments_call.function.arguments is not None + ), chunks + return ( + (call.id, call.function.name, arguments_call.function.arguments), + chunks[0].id, + raw.headers, + ) + case "responses": + client: Final = _openai_client(rig.proxy) + raw: Final = client.responses.with_raw_response.create( + model=rig.model, + input=marker, + tools=RESPONSES_TOOLS, + tool_choice={"type": "function", "name": "lookup_weather"}, + extra_body={"metadata": {"trace_marker": marker}, "cache": {"no-cache": False}}, ) + response: Final = raw.parse() + assert _normalize_responses_caller_body(response.model_dump(mode="json", exclude_unset=True)) == ( + _responses_caller_response(_responses_response(marker), rig.model) + ), response + call: Final = response.output[0] + assert call.type == "function_call", response + return (call.call_id, call.name, call.arguments), response.id, raw.headers + case "messages": + client: Final = anthropic.Anthropic( + base_url=str(rig.proxy.client.base_url), api_key=rig.proxy.key, max_retries=0 + ) + raw: Final = client.messages.with_raw_response.create( + model=rig.model, + max_tokens=64, + messages=[{"role": "user", "content": marker}], + tools=[ + { + "name": "lookup_weather", + "description": "Get weather", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}, + } + ], + tool_choice={"type": "auto"}, + extra_body={"cache": {"no-cache": False}, "metadata": {"trace_marker": marker}}, + ) + response: Final = raw.parse() + assert response.model_dump(mode="json", exclude_unset=True) == _messages_caller_response( + _messages_response(marker), rig.model + ), response + call: Final = response.content[0] + assert call.type == "tool_use", response + return (call.id, call.name, json.dumps(call.input)), response.id, raw.headers + case _: + raise AssertionError(f"Unknown cache surface: {surface}") + + +@pytest.mark.parametrize("surface", ("chat", "chat-stream", "responses", "messages")) +def test_arize_otel_v2_a_cache(surface: str, gateway: Gateway, tmp_path: Path) -> None: + marker: Final = f"a-cache-{surface}-" + uuid.uuid4().hex + + def upstream(request: Request) -> Reply: + body: Final = _json_object(request.body) + if surface in ("chat", "chat-stream"): + _assert_chat_request( + request, + messages=[{"role": "user", "content": marker}], + stream=True if surface == "chat-stream" else None, + stream_options={"include_usage": False} if surface == "chat-stream" else None, + ) + return ( + _chat_stream_response(marker, (_chat_tool_call(marker),), include_usage=False) + if surface == "chat-stream" + else _chat_response(marker) + ) + if surface == "responses": + assert body.get("metadata") == {"trace_marker": marker}, body + _assert_responses_request(request, marker=marker, input_value=marker) + return _responses_response(marker) + _assert_messages_request(request, marker=marker, prompt=marker) + return _messages_response(marker) + + with _rig( + gateway, + tmp_path, + upstream, + general_settings={"always_include_stream_usage": False} if surface == "chat-stream" else None, + model_name="anthropic/claude-opus-5-5" if surface == "messages" else "gpt-4o-mini", + api_base_suffix="" if surface == "messages" else "/v1", + ) as rig: + expected_arguments: Final = '{"city": "Paris"}' + first: Final = _cache_call(rig, surface, marker) + assert first[0] == (f"call_{marker}", "lookup_weather", expected_arguments), first + assert first[1].startswith("resp_") if surface == "responses" else first[1] == marker, first + assert not first[2].get("x-litellm-cache-key"), first[2] + rig.destination.drain() + second: Final = _cache_call(rig, surface, marker, cache_hit=True) + assert second[0] == first[0], second + assert second[1].startswith("resp_") if surface == "responses" else second[1] == first[1], second + forwarded: Final = tuple( + request for request in rig.provider.drain() if request.method == "POST" and marker.encode() in request.body + ) + assert len(forwarded) == 1, forwarded + if surface == "messages": + assert not second[2].get("x-litellm-cache-key"), second[2] + else: + assert second[2].get("x-litellm-cache-key"), second[2] + _assert_tool_span_for_marker(_matching_marker_span(rig.destination, marker), marker) + forwarded_body: Final = _json_object(forwarded[0].body) + if surface in ("chat", "chat-stream"): + assert "metadata" not in forwarded_body, forwarded[0] + elif surface == "responses": + assert forwarded_body["metadata"] == {"trace_marker": marker}, forwarded[0] + else: + assert forwarded_body["metadata"] == {}, forwarded[0]