fix(instrumentation-py): repair LiveKit, LlamaIndex, and Marqo OTel 2.x spans

.release-intents/20260817-python-livekit-llama-index-marqo-otel2.json

@@ -0,0 +1,8 @@
+{
+  "summary": "Repair OTel 2.x span translation and validation for LiveKit, LlamaIndex, and Marqo instrumentations",
+  "packages": {
+    "respan-instrumentation-livekit": "patch",
+    "respan-instrumentation-llama-index": "patch",
+    "respan-instrumentation-marqo": "patch"
+  }
+}

python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_constants.py

@@ -29,4 +29,4 @@
 EVENT_GEN_AI_CHOICE = "gen_ai.choice"
 
 LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR = "lk.respan.function_tools"
-
+LIVEKIT_RESPAN_PROVIDER_NAME_ATTR = "lk.respan.provider_name"

python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_instrumentation.py

@@ -6,19 +6,22 @@
 import importlib
 import logging
 import time
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from opentelemetry import trace
+from opentelemetry.semconv_ai import SpanAttributes
+from respan_tracing.core.tracer import RespanTracer
 
 from respan_instrumentation_livekit._constants import (
     LIVEKIT_INSTRUMENTATION_NAME,
+    LIVEKIT_RESPAN_PROVIDER_NAME_ATTR,
     LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR,
 )
 from respan_instrumentation_livekit._otel_emitter import emit_livekit_tool_span
 from respan_instrumentation_livekit._processor import LiveKitSpanProcessor
 from respan_instrumentation_livekit._serialization import get_value, safe_json
 from respan_instrumentation_livekit._translator import normalize_livekit_tools
-from respan_tracing.core.tracer import RespanTracer
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +33,34 @@
 _PATCHED_LLM_STREAM_CLASS: Any = None
 _ACTIVE_INSTANCES = 0
 
+_MODEL_PROVIDER_PREFIXES = (
+    ("claude", "anthropic"),
+    ("gemini", "google"),
+    ("gpt-", "openai"),
+    ("o1", "openai"),
+    ("o3", "openai"),
+    ("o4", "openai"),
+)
+
+
+def _provider_name_from_llm(llm: Any) -> str | None:
+    module_parts = type(llm).__module__.lower().split(".")
+    try:
+        plugin_index = module_parts.index("plugins")
+        plugin_name = module_parts[plugin_index + 1]
+    except (ValueError, IndexError):
+        plugin_name = ""
+    if plugin_name:
+        return "openai" if plugin_name in {"azure", "azure_openai"} else plugin_name
+
+    model = str(get_value(llm, "model") or "").lower()
+    for prefix, provider_name in _MODEL_PROVIDER_PREFIXES:
+        if model.startswith(prefix):
+            return provider_name
+
+    provider = get_value(llm, "provider")
+    return str(provider).lower() if provider else None
+
 
 def _active_span_processors() -> tuple[Any, tuple[Any, ...] | None]:
     tracer_provider = trace.get_tracer_provider()
@@ -74,7 +105,7 @@ def _patch_execute_function_call(utils_module: Any, llm_module: Any) -> None:
     if _ORIGINAL_UTILS_EXECUTE_FUNCTION_CALL is not None:
         return
 
-    original = getattr(utils_module, "execute_function_call")
+    original = utils_module.execute_function_call
     exported_original = getattr(llm_module, "execute_function_call", None)
 
     @functools.wraps(original)
@@ -93,9 +124,8 @@ async def wrapped_execute_function_call(*args: Any, **kwargs: Any) -> Any:
             or get_value(tool_call, "arguments")
             or {}
         )
-        call_id = (
-            get_value(get_value(result, "fnc_call"), "call_id")
-            or get_value(tool_call, "call_id")
+        call_id = get_value(get_value(result, "fnc_call"), "call_id") or get_value(
+            tool_call, "call_id"
         )
         raw_exception = get_value(result, "raw_exception")
         fnc_call_out = get_value(result, "fnc_call_out")
@@ -117,9 +147,9 @@ async def wrapped_execute_function_call(*args: Any, **kwargs: Any) -> Any:
     _ORIGINAL_EXPORTED_EXECUTE_FUNCTION_CALL = exported_original
     _PATCHED_UTILS_MODULE = utils_module
     _PATCHED_LLM_MODULE = llm_module
-    setattr(utils_module, "execute_function_call", wrapped_execute_function_call)
+    utils_module.execute_function_call = wrapped_execute_function_call
     if exported_original is not None:
-        setattr(llm_module, "execute_function_call", wrapped_execute_function_call)
+        llm_module.execute_function_call = wrapped_execute_function_call
 
 
 def _restore_execute_function_call() -> None:
@@ -129,16 +159,12 @@ def _restore_execute_function_call() -> None:
     global _PATCHED_LLM_MODULE
 
     if _PATCHED_UTILS_MODULE is not None and _ORIGINAL_UTILS_EXECUTE_FUNCTION_CALL:
-        setattr(
-            _PATCHED_UTILS_MODULE,
-            "execute_function_call",
-            _ORIGINAL_UTILS_EXECUTE_FUNCTION_CALL,
+        _PATCHED_UTILS_MODULE.execute_function_call = (
+            _ORIGINAL_UTILS_EXECUTE_FUNCTION_CALL
         )
     if _PATCHED_LLM_MODULE is not None and _ORIGINAL_EXPORTED_EXECUTE_FUNCTION_CALL:
-        setattr(
-            _PATCHED_LLM_MODULE,
-            "execute_function_call",
-            _ORIGINAL_EXPORTED_EXECUTE_FUNCTION_CALL,
+        _PATCHED_LLM_MODULE.execute_function_call = (
+            _ORIGINAL_EXPORTED_EXECUTE_FUNCTION_CALL
         )
 
     _ORIGINAL_UTILS_EXECUTE_FUNCTION_CALL = None
@@ -154,11 +180,21 @@ def _patch_llm_stream_main_task(llm_stream_class: Any) -> None:
     if _ORIGINAL_LLM_STREAM_MAIN_TASK is not None:
         return
 
-    original = getattr(llm_stream_class, "_main_task")
+    original = llm_stream_class._main_task
 
     @functools.wraps(original)
     async def wrapped_main_task(self: Any, *args: Any, **kwargs: Any) -> Any:
         current_span = trace.get_current_span()
+        try:
+            current_span.set_attribute(SpanAttributes.LLM_IS_STREAMING, True)
+            provider_name = _provider_name_from_llm(getattr(self, "_llm", None))
+            if provider_name:
+                current_span.set_attribute(
+                    LIVEKIT_RESPAN_PROVIDER_NAME_ATTR,
+                    provider_name,
+                )
+        except Exception:
+            logger.debug("Failed to classify LiveKit LLM stream", exc_info=True)
         tool_definitions = normalize_livekit_tools(getattr(self, "_tools", None))
         if tool_definitions:
             try:
@@ -172,15 +208,15 @@ async def wrapped_main_task(self: Any, *args: Any, **kwargs: Any) -> Any:
 
     _ORIGINAL_LLM_STREAM_MAIN_TASK = original
     _PATCHED_LLM_STREAM_CLASS = llm_stream_class
-    setattr(llm_stream_class, "_main_task", wrapped_main_task)
+    llm_stream_class._main_task = wrapped_main_task
 
 
 def _restore_llm_stream_main_task() -> None:
     global _ORIGINAL_LLM_STREAM_MAIN_TASK
     global _PATCHED_LLM_STREAM_CLASS
 
     if _PATCHED_LLM_STREAM_CLASS is not None and _ORIGINAL_LLM_STREAM_MAIN_TASK:
-        setattr(_PATCHED_LLM_STREAM_CLASS, "_main_task", _ORIGINAL_LLM_STREAM_MAIN_TASK)
+        _PATCHED_LLM_STREAM_CLASS._main_task = _ORIGINAL_LLM_STREAM_MAIN_TASK
     _ORIGINAL_LLM_STREAM_MAIN_TASK = None
     _PATCHED_LLM_STREAM_CLASS = None
 
@@ -209,7 +245,9 @@ def activate(self) -> None:
             return
 
         if not self._is_respan_tracing_enabled():
-            logger.info("LiveKit instrumentation skipped because Respan tracing is disabled")
+            logger.info(
+                "LiveKit instrumentation skipped because Respan tracing is disabled"
+            )
             return
 
         try:
@@ -229,7 +267,7 @@ def activate(self) -> None:
 
         _register_processor(self._processor)
         _patch_execute_function_call(livekit_llm_utils, livekit_llm)
-        _patch_llm_stream_main_task(getattr(livekit_llm, "LLMStream"))
+        _patch_llm_stream_main_task(livekit_llm.LLMStream)
 
         _ACTIVE_INSTANCES += 1
         self._is_instrumented = True

python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_processor.py

@@ -7,9 +7,18 @@
 
 from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor
 from opentelemetry.semconv_ai import SpanAttributes
+from respan_sdk.constants.span_attributes import (
+    RESPAN_SPAN_CUSTOM_ID,
+    RESPAN_TRACE_GROUP_ID,
+)
+from respan_sdk.utils.data_processing.id_processing import (
+    format_span_id,
+    format_trace_id,
+)
 
 from respan_instrumentation_livekit._constants import (
     ATTR_LLM_METRICS,
+    LIVEKIT_RESPAN_PROVIDER_NAME_ATTR,
     LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR,
     LIVEKIT_SCOPE_NAME,
 )
@@ -20,14 +29,6 @@
     build_livekit_llm_attrs,
     is_livekit_llm_span,
 )
-from respan_sdk.constants.span_attributes import (
-    RESPAN_SPAN_CUSTOM_ID,
-    RESPAN_TRACE_GROUP_ID,
-)
-from respan_sdk.utils.data_processing.id_processing import (
-    format_span_id,
-    format_trace_id,
-)
 
 
 def _mutable_attrs(span: ReadableSpan) -> dict[str, Any] | None:
@@ -54,7 +55,7 @@ def _tool_call_ids(attrs: dict[str, Any]) -> list[str]:
         return []
     try:
         parsed = json.loads(value) if isinstance(value, str) else value
-    except Exception:
+    except json.JSONDecodeError:
         return []
     if not isinstance(parsed, list):
         return []
@@ -127,6 +128,8 @@ def on_end(self, span: ReadableSpan) -> None:
             events=tuple(getattr(span, "events", ()) or ()),
         )
         attrs.update(translated)
+        attrs.pop(LIVEKIT_RESPAN_PROVIDER_NAME_ATTR, None)
+        attrs.pop(LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR, None)
         _rename_span_to_workflow(span=span, attrs=attrs)
         _register_tool_parent_contexts(span=span, attrs=attrs)
         span._attributes = attrs

python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_translator.py

@@ -10,6 +10,17 @@
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.semconv_ai import LLMRequestTypeValues, SpanAttributes
+from respan_sdk.constants.llm_logging import (
+    LOG_TYPE_CHAT,
+    LOG_TYPE_TOOL,
+    LogMethodChoices,
+)
+from respan_sdk.constants.span_attributes import (
+    RESPAN_LOG_METHOD,
+    RESPAN_LOG_TYPE,
+    RESPAN_METADATA,
+    RESPAN_TRACE_GROUP_ID,
+)
 
 from respan_instrumentation_livekit._constants import (
     ASSISTANT_ROLE,
@@ -25,6 +36,7 @@
     ID_KEY,
     LIVEKIT_CHAT_SPAN_NAME,
     LIVEKIT_LLM_REQUEST_SPAN_NAME,
+    LIVEKIT_RESPAN_PROVIDER_NAME_ATTR,
     LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR,
     NAME_KEY,
     ROLE_KEY,
@@ -37,13 +49,6 @@
     parse_jsonish,
     safe_json,
 )
-from respan_sdk.constants.llm_logging import LOG_TYPE_CHAT, LOG_TYPE_TOOL, LogMethodChoices
-from respan_sdk.constants.span_attributes import (
-    RESPAN_LOG_METHOD,
-    RESPAN_LOG_TYPE,
-    RESPAN_METADATA,
-    RESPAN_TRACE_GROUP_ID,
-)
 
 _PROMPT_EVENT_ROLES = {
     EVENT_GEN_AI_SYSTEM_MESSAGE: "system",
@@ -75,9 +80,7 @@ def _tool_calls_from_event(value: Any) -> list[dict[str, Any]]:
         return []
     if isinstance(value, (str, bytes, bytearray)):
         value = [value.decode() if isinstance(value, bytes) else value]
-    elif isinstance(value, Mapping):
-        value = [value]
-    elif not isinstance(value, Sequence):
+    elif isinstance(value, Mapping) or not isinstance(value, Sequence):
         value = [value]
 
     tool_calls: list[dict[str, Any]] = []
@@ -115,7 +118,7 @@ def _metrics(attrs: Mapping[str, Any]) -> Mapping[str, Any]:
         return {}
     try:
         loaded = json.loads(value)
-    except Exception:
+    except json.JSONDecodeError:
         return {}
     return loaded if isinstance(loaded, Mapping) else {}
 
@@ -131,8 +134,10 @@ def _int_value(value: Any) -> int | None:
 
 
 def _provider_name(attrs: Mapping[str, Any]) -> str:
-    provider = attrs.get(GenAIAttributes.GEN_AI_PROVIDER_NAME) or attrs.get(
-        SpanAttributes.LLM_SYSTEM
+    provider = (
+        attrs.get(LIVEKIT_RESPAN_PROVIDER_NAME_ATTR)
+        or attrs.get(GenAIAttributes.GEN_AI_PROVIDER_NAME)
+        or attrs.get(SpanAttributes.LLM_SYSTEM)
     )
     if not provider:
         return "livekit"
@@ -210,7 +215,9 @@ def _apply_completion_event(
 def _apply_usage(translated: dict[str, Any], attrs: Mapping[str, Any]) -> None:
     metrics = _metrics(attrs)
     input_tokens = _int_value(
-        attrs.get(GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS, metrics.get("prompt_tokens"))
+        attrs.get(
+            GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS, metrics.get("prompt_tokens")
+        )
     )
     output_tokens = _int_value(
         attrs.get(
@@ -241,7 +248,9 @@ def _apply_usage(translated: dict[str, Any], attrs: Mapping[str, Any]) -> None:
         translated[cache_read_attr] = cached_tokens
 
 
-def _apply_livekit_metadata(translated: dict[str, Any], attrs: Mapping[str, Any]) -> None:
+def _apply_livekit_metadata(
+    translated: dict[str, Any], attrs: Mapping[str, Any]
+) -> None:
     metrics = attrs.get(ATTR_LLM_METRICS)
     if isinstance(metrics, str) and metrics:
         translated[f"{RESPAN_METADATA}.livekit_llm_metrics"] = metrics
@@ -255,7 +264,9 @@ def _apply_livekit_metadata(translated: dict[str, Any], attrs: Mapping[str, Any]
         )
 
 
-def _apply_tool_definitions(translated: dict[str, Any], attrs: Mapping[str, Any]) -> None:
+def _apply_tool_definitions(
+    translated: dict[str, Any], attrs: Mapping[str, Any]
+) -> None:
     value = attrs.get(LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR)
     parsed = parse_jsonish(value)
     if parsed:
@@ -279,10 +290,12 @@ def build_livekit_llm_attrs(
     events: Sequence[Any],
 ) -> dict[str, Any]:
     """Build canonical Respan chat attributes from a LiveKit LLM span."""
+    provider_name = _provider_name(attrs)
     translated: dict[str, Any] = {
         RESPAN_LOG_METHOD: LogMethodChoices.TRACING_INTEGRATION.value,
         RESPAN_LOG_TYPE: LOG_TYPE_CHAT,
-        SpanAttributes.LLM_SYSTEM: _provider_name(attrs),
+        GenAIAttributes.GEN_AI_PROVIDER_NAME: provider_name,
+        SpanAttributes.LLM_SYSTEM: provider_name,
         SpanAttributes.LLM_REQUEST_TYPE: LLMRequestTypeValues.CHAT.value,
         SpanAttributes.TRACELOOP_ENTITY_NAME: LIVEKIT_CHAT_SPAN_NAME,
         SpanAttributes.TRACELOOP_ENTITY_PATH: LIVEKIT_CHAT_SPAN_NAME,
@@ -294,6 +307,9 @@ def build_livekit_llm_attrs(
     if model:
         translated[SpanAttributes.LLM_REQUEST_MODEL] = str(model)
 
+    if attrs.get(SpanAttributes.LLM_IS_STREAMING):
+        translated[SpanAttributes.LLM_IS_STREAMING] = True
+
     workflow_name = attrs.get(RESPAN_TRACE_GROUP_ID)
     if workflow_name:
         translated[SpanAttributes.TRACELOOP_WORKFLOW_NAME] = str(workflow_name)
@@ -325,15 +341,17 @@ def normalize_livekit_tools(tools: Any) -> list[dict[str, Any]]:
         parsed = ToolContext(tools).parse_function_tools("openai")
         if isinstance(parsed, list):
             return [tool for tool in parsed if isinstance(tool, dict)]
-    except Exception:
-        pass
+    except Exception:  # noqa: BLE001 - third-party tool objects may fail arbitrarily
+        parsed = None
 
     normalized: list[dict[str, Any]] = []
     for tool in tools:
         info = get_value(tool, "info")
         raw_schema = get_value(info, "raw_schema")
         if isinstance(raw_schema, Mapping) and raw_schema.get(NAME_KEY):
-            normalized.append({TYPE_KEY: FUNCTION_TOOL_TYPE, FUNCTION_KEY: dict(raw_schema)})
+            normalized.append(
+                {TYPE_KEY: FUNCTION_TOOL_TYPE, FUNCTION_KEY: dict(raw_schema)}
+            )
             continue
 
         name = get_value(info, NAME_KEY) or get_value(tool, ID_KEY)