From 66e302193b7552dbb9695bd34e1fe6d8e2803010 Mon Sep 17 00:00:00 2001 From: Nightingalelyy Date: Mon, 17 Aug 2026 19:33:10 +0800 Subject: [PATCH 1/2] fix(instrumentation): repair LiveKit, LlamaIndex, and Marqo OTel 2.x spans --- ...ython-livekit-llama-index-marqo-otel2.json | 8 + .../_constants.py | 2 +- .../_instrumentation.py | 80 +++++-- .../_processor.py | 21 +- .../_translator.py | 58 +++-- .../tests/test_instrumentation.py | 202 +++++++++++++++++- .../README.md | 9 +- .../_constants.py | 1 - .../_handlers.py | 117 ++++++---- .../tests/test_instrumentation.py | 186 +++++++++++++--- .../_instrumentation.py | 1 + .../_native_instrumentation.py | 57 ++++- .../tests/test_instrumentation.py | 176 ++++++++++++++- 13 files changed, 774 insertions(+), 144 deletions(-) create mode 100644 .release-intents/20260817-python-livekit-llama-index-marqo-otel2.json diff --git a/.release-intents/20260817-python-livekit-llama-index-marqo-otel2.json b/.release-intents/20260817-python-livekit-llama-index-marqo-otel2.json new file mode 100644 index 00000000..5de8716f --- /dev/null +++ b/.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" + } +} diff --git a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_constants.py b/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_constants.py index 509cb9ee..725a6517 100644 --- a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_constants.py +++ b/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" diff --git a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_instrumentation.py index fbce1dc4..2cb5b2fd 100644 --- a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_instrumentation.py +++ b/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,7 +208,7 @@ 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: @@ -180,7 +216,7 @@ def _restore_llm_stream_main_task() -> None: 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 diff --git a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_processor.py b/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_processor.py index b3e0aecd..5add3b54 100644 --- a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_processor.py +++ b/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 diff --git a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_translator.py b/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_translator.py index bbb82149..3f6408b7 100644 --- a/python-sdks/instrumentations/respan-instrumentation-livekit/src/respan_instrumentation_livekit/_translator.py +++ b/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) diff --git a/python-sdks/instrumentations/respan-instrumentation-livekit/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-livekit/tests/test_instrumentation.py index 345294fd..d878efd5 100644 --- a/python-sdks/instrumentations/respan-instrumentation-livekit/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-livekit/tests/test_instrumentation.py @@ -5,26 +5,34 @@ from types import ModuleType, SimpleNamespace from typing import Any +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAIAttributes, ) from opentelemetry.semconv_ai import SpanAttributes - -from respan_instrumentation_livekit import _instrumentation -from respan_instrumentation_livekit import _otel_emitter +from respan_instrumentation_livekit import _instrumentation, _otel_emitter from respan_instrumentation_livekit._constants import ( ATTR_LLM_METRICS, EVENT_GEN_AI_CHOICE, EVENT_GEN_AI_USER_MESSAGE, + LIVEKIT_RESPAN_PROVIDER_NAME_ATTR, LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR, ) from respan_instrumentation_livekit._processor import LiveKitSpanProcessor -from respan_instrumentation_livekit._translator import build_tool_span_attrs +from respan_instrumentation_livekit._translator import ( + build_livekit_llm_attrs, + build_tool_span_attrs, +) from respan_sdk.constants.span_attributes import ( RESPAN_LOG_METHOD, RESPAN_LOG_TYPE, RESPAN_TRACE_GROUP_ID, ) +from respan_tracing.exporters.respan import _convert_attributes _OFF_CONTRACT_ALIASES = { "completion_tokens", @@ -90,6 +98,7 @@ def test_processor_translates_livekit_llm_span_to_respan_contract(monkeypatch): GenAIAttributes.GEN_AI_REQUEST_MODEL: "gpt-4o-mini", GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS: 10, GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS: 4, + SpanAttributes.LLM_IS_STREAMING: True, ATTR_LLM_METRICS: json.dumps( { "prompt_tokens": 10, @@ -128,6 +137,7 @@ def test_processor_translates_livekit_llm_span_to_respan_contract(monkeypatch): assert attrs[SpanAttributes.LLM_SYSTEM] == "openai" assert attrs[SpanAttributes.LLM_REQUEST_TYPE] == "chat" assert attrs[SpanAttributes.LLM_REQUEST_MODEL] == "gpt-4o-mini" + assert attrs[SpanAttributes.LLM_IS_STREAMING] is True assert span.name == "livekit_03_tool_calling" assert attrs[SpanAttributes.TRACELOOP_WORKFLOW_NAME] == "livekit_03_tool_calling" assert attrs[f"{SpanAttributes.LLM_PROMPTS}.0.role"] == "user" @@ -140,6 +150,7 @@ def test_processor_translates_livekit_llm_span_to_respan_contract(monkeypatch): tool_call ] assert json.loads(attrs[SpanAttributes.LLM_REQUEST_FUNCTIONS]) == [tool_schema] + assert LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR not in attrs assert attrs[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 10 assert attrs[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 4 assert attrs[SpanAttributes.LLM_USAGE_PROMPT_TOKENS] == 10 @@ -156,6 +167,77 @@ def test_processor_translates_livekit_llm_span_to_respan_contract(monkeypatch): assert _OFF_CONTRACT_ALIASES.isdisjoint(attrs) +def test_stream_and_tool_contract_survives_otlp_attribute_serialization(): + tool_schema = { + "type": "function", + "function": { + "name": "lookup_room", + "description": "Lookup a room.", + "parameters": { + "type": "object", + "properties": {"room": {"type": "string"}}, + "required": ["room"], + }, + }, + } + tool_call = { + "id": "call_123", + "type": "function", + "function": {"name": "lookup_room", "arguments": '{"room":"blue"}'}, + } + attrs = build_livekit_llm_attrs( + span_name="llm_request", + attrs={ + GenAIAttributes.GEN_AI_OPERATION_NAME: "chat", + SpanAttributes.LLM_IS_STREAMING: True, + LIVEKIT_RESPAN_TOOL_DEFINITIONS_ATTR: json.dumps([tool_schema]), + }, + events=[ + _Event( + EVENT_GEN_AI_CHOICE, + {"content": "I will check.", "tool_calls": [json.dumps(tool_call)]}, + ) + ], + ) + + exported = {item["key"]: item["value"] for item in _convert_attributes(attrs)} + + assert exported[SpanAttributes.LLM_IS_STREAMING] == {"boolValue": True} + assert json.loads( + exported[SpanAttributes.LLM_REQUEST_FUNCTIONS]["stringValue"] + ) == [tool_schema] + assert json.loads( + exported[f"{SpanAttributes.LLM_COMPLETIONS}.0.tool_calls"]["stringValue"] + ) == [tool_call] + assert _OFF_CONTRACT_ALIASES.isdisjoint(exported) + + +def test_gateway_plugin_provider_overrides_base_url_host(): + gateway_llm = type( + "LLM", + (), + { + "__module__": "livekit.plugins.openai.llm", + "model": "gpt-4o-mini", + "provider": "api.respan.ai", + }, + )() + + assert _instrumentation._provider_name_from_llm(gateway_llm) == "openai" + + attrs = build_livekit_llm_attrs( + span_name="llm_request", + attrs={ + GenAIAttributes.GEN_AI_OPERATION_NAME: "chat", + GenAIAttributes.GEN_AI_PROVIDER_NAME: "api.respan.ai", + LIVEKIT_RESPAN_PROVIDER_NAME_ATTR: "openai", + }, + events=[], + ) + assert attrs[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "openai" + assert attrs[SpanAttributes.LLM_SYSTEM] == "openai" + + def test_build_tool_span_attrs_uses_tool_contract_without_aliases(): attrs = build_tool_span_attrs( tool_name="lookup_room", @@ -192,7 +274,9 @@ def get_span_context(self): span_id=int("0" * 15 + "2", 16), ) - monkeypatch.setattr(_otel_emitter.trace, "get_current_span", lambda: _FakeCurrentSpan()) + monkeypatch.setattr( + _otel_emitter.trace, "get_current_span", lambda: _FakeCurrentSpan() + ) monkeypatch.setattr(_otel_emitter, "build_readable_span", fake_build_readable_span) monkeypatch.setattr(_otel_emitter, "inject_span", lambda span: True) monkeypatch.setattr( @@ -297,8 +381,21 @@ async def _main_task(self): lambda name: module_map[name], ) monkeypatch.setattr(_instrumentation, "_register_processor", lambda processor: None) - monkeypatch.setattr(_instrumentation, "_unregister_processor", lambda processor: None) + monkeypatch.setattr( + _instrumentation, "_unregister_processor", lambda processor: None + ) emitted = [] + stream_attrs = {} + + class FakeCurrentSpan: + def set_attribute(self, key, value): + stream_attrs[key] = value + + monkeypatch.setattr( + _instrumentation.trace, + "get_current_span", + lambda: FakeCurrentSpan(), + ) monkeypatch.setattr( _instrumentation, "emit_livekit_tool_span", @@ -313,6 +410,11 @@ async def _main_task(self): assert fake_llm.execute_function_call is fake_utils.execute_function_call assert FakeLLMStream._main_task is not original_main_task + stream = FakeLLMStream() + stream._tools = [] + asyncio.run(stream._main_task()) + assert stream_attrs[SpanAttributes.LLM_IS_STREAMING] is True + result = asyncio.run( fake_utils.execute_function_call( SimpleNamespace(name="lookup_room", arguments="{}", call_id="call_1"), @@ -328,3 +430,91 @@ async def _main_task(self): assert fake_llm.execute_function_call is original_execute_function_call assert FakeLLMStream._main_task is original_main_task assert _instrumentation._ORIGINAL_LLM_STREAM_MAIN_TASK is None + + +def test_real_livekit_stream_exports_translated_stream_and_provider(): + from livekit.agents import llm, telemetry + from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS + + class GatewayLLM(llm.LLM): + __module__ = "livekit.plugins.openai.llm" + + def __init__(self) -> None: + super().__init__() + self._model = "gpt-4o-mini" + + @property + def model(self) -> str: + return self._model + + @property + def provider(self) -> str: + return "api.respan.ai" + + def chat(self, *, chat_ctx, tools=None, conn_options=None, **_kwargs): + return GatewayStream( + self, + chat_ctx=chat_ctx, + tools=tools or [], + conn_options=conn_options or DEFAULT_API_CONNECT_OPTIONS, + ) + + class GatewayStream(llm.LLMStream): + async def _run(self) -> None: + self._event_ch.send_nowait( + llm.ChatChunk( + id="gateway-stream", + delta=llm.ChoiceDelta( + role="assistant", + content="streamed gateway response", + ), + ) + ) + await asyncio.sleep(0) + self._event_ch.send_nowait( + llm.ChatChunk( + id="gateway-stream", + usage=llm.CompletionUsage( + prompt_tokens=7, + completion_tokens=3, + total_tokens=10, + ), + ) + ) + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(LiveKitSpanProcessor()) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + previous_provider = telemetry.tracer._tracer_provider + telemetry.set_tracer_provider(provider) + _instrumentation._patch_llm_stream_main_task(llm.LLMStream) + try: + + async def collect_gateway_response(): + chat_ctx = llm.ChatContext.empty() + chat_ctx.add_message(role="user", content="stream a response") + return await GatewayLLM().chat(chat_ctx=chat_ctx).collect() + + response = asyncio.run(collect_gateway_response()) + assert response.text == "streamed gateway response" + + provider.force_flush() + chat_spans = [ + span + for span in exporter.get_finished_spans() + if span.attributes.get(RESPAN_LOG_TYPE) == "chat" + ] + assert len(chat_spans) == 1 + attrs = chat_spans[0].attributes + assert attrs[SpanAttributes.LLM_IS_STREAMING] is True + assert attrs[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "openai" + assert attrs[SpanAttributes.LLM_SYSTEM] == "openai" + assert json.loads(attrs[SpanAttributes.TRACELOOP_ENTITY_OUTPUT]) == { + "content": "streamed gateway response" + } + assert LIVEKIT_RESPAN_PROVIDER_NAME_ATTR not in attrs + finally: + _instrumentation._restore_llm_stream_main_task() + telemetry.tracer.set_provider(previous_provider) + provider.shutdown() diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/README.md b/python-sdks/instrumentations/respan-instrumentation-llama-index/README.md index 3852a4fb..a931b97e 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/README.md +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/README.md @@ -60,10 +60,11 @@ index = SummaryIndex.from_documents( ) query_engine = index.as_query_engine() -response = query_engine.query("What does Respan capture?") -print(response) - -respan.flush() +try: + response = query_engine.query("What does Respan capture?") + print(response) +finally: + respan.shutdown() ``` ### 4. View Dashboard diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_constants.py b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_constants.py index 9b7c721c..b68bc9d2 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_constants.py +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_constants.py @@ -12,7 +12,6 @@ LLAMA_INDEX_CHAT_SPAN_NAME = "llama_index.chat" LLAMA_INDEX_COMPLETION_SPAN_NAME = "llama_index.completion" LLAMA_INDEX_EMBEDDING_SPAN_NAME = "llama_index.embedding" -LLAMA_INDEX_TOOL_SPAN_PREFIX = "llama_index.tool." LLAMA_INDEX_DEFAULT_TOOL_NAME = "llama_index_tool" LLAMA_INDEX_RUN_ID_TAG = "llamaindex.run_id" diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py index d4fa16c0..fb85b833 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py @@ -7,12 +7,12 @@ import re from typing import Any +from llama_index.core.tools.types import BaseTool from llama_index_instrumentation.dispatcher import active_instrument_tags from llama_index_instrumentation.event_handlers import BaseEventHandler from llama_index_instrumentation.span import BaseSpan from llama_index_instrumentation.span_handlers import BaseSpanHandler -from opentelemetry import context -from opentelemetry import trace +from opentelemetry import context, trace from opentelemetry.semconv_ai import LLMRequestTypeValues, SpanAttributes from opentelemetry.trace import Status, StatusCode from pydantic import ConfigDict, PrivateAttr @@ -42,13 +42,12 @@ LLAMA_INDEX_COMPLETION_SPAN_NAME, LLAMA_INDEX_DEFAULT_TOOL_NAME, LLAMA_INDEX_EMBEDDING_SPAN_NAME, - LLAMA_INDEX_USAGE_INPUT_TOKENS, - LLAMA_INDEX_USAGE_OUTPUT_TOKENS, LLAMA_INDEX_RUN_ID_TAG, LLAMA_INDEX_START_EVENT_TAG, LLAMA_INDEX_STEP_INPUT_EVENT_TAG, LLAMA_INDEX_STEP_INPUT_SUMMARY_TAG, - LLAMA_INDEX_TOOL_SPAN_PREFIX, + LLAMA_INDEX_USAGE_INPUT_TOKENS, + LLAMA_INDEX_USAGE_OUTPUT_TOKENS, MESSAGE_ROLE_ASSISTANT, MESSAGE_ROLE_USER, STATUS_CODE_ATTR, @@ -115,20 +114,31 @@ def new_span( **kwargs: Any, ) -> RespanLlamaIndexSpan: tags = tags or active_instrument_tags.get() - entity_name = _span_entity_name(span_id=id_) + source_entity_name = _span_entity_name(span_id=id_) + is_tool_execution = _is_executable_tool_span( + entity_name=source_entity_name, + instance=instance, + ) + entity_name = ( + _tool_name(tool=instance) if is_tool_execution else source_entity_name + ) log_type = _span_log_type( - entity_name=entity_name, + entity_name=source_entity_name, instance=instance, parent_span_id=parent_span_id, ) attributes = _base_attributes( entity_name=entity_name, log_type=log_type, - entity_path=entity_name, + entity_path=entity_name if parent_span_id is not None else "", ) if self.capture_content: attributes[SpanAttributes.TRACELOOP_ENTITY_INPUT] = safe_json( - _span_input_payload(bound_args=bound_args, tags=tags) + _span_input_payload( + bound_args=bound_args, + tags=tags, + tool_name=entity_name if is_tool_execution else None, + ) ) otel_span, context_token, span_context = _start_otel_span( @@ -174,7 +184,7 @@ def _create_synthetic_parent_context(self, *, parent_span_id: str) -> Any: instance=None, parent_span_id=None, ), - entity_path=entity_name, + entity_path="", ) parent_context = context.get_current() otel_span = _start_detached_otel_span( @@ -203,7 +213,7 @@ def prepare_to_exit_span( SpanAttributes.TRACELOOP_ENTITY_OUTPUT, safe_json( _span_output_payload( - entity_name=active_span.entity_name, + log_type=active_span.log_type, result=result, ) ), @@ -276,8 +286,6 @@ def handle(self, event: Any, **kwargs: Any) -> Any: self._handle_embedding_start(event=event) elif event_name == "EmbeddingEndEvent": self._handle_embedding_end(event=event) - elif event_name == "AgentToolCallEvent": - self._handle_tool_call(event=event) elif event_name == "ExceptionEvent": self._handle_exception(event=event) @@ -421,27 +429,6 @@ def _handle_embedding_end(self, *, event: Any) -> None: attributes=attributes, ) - def _handle_tool_call(self, *, event: Any) -> None: - tool = getattr(event, "tool", None) - tool_name = _tool_name(tool=tool) - attributes = _base_attributes( - entity_name=tool_name, - log_type=LOG_TYPE_TOOL, - entity_path=tool_name, - ) - if self.capture_content: - attributes[SpanAttributes.TRACELOOP_ENTITY_INPUT] = safe_json( - { - "name": tool_name, - "arguments": getattr(event, "arguments", ""), - } - ) - active_event_span = _start_event_span( - span_name=f"{LLAMA_INDEX_TOOL_SPAN_PREFIX}{tool_name}", - attributes=attributes, - ) - _finish_event_span(active_event_span=active_event_span, attributes={}) - def _handle_exception(self, *, event: Any) -> None: span_id = getattr(event, "span_id", None) exception = getattr(event, "exception", None) @@ -640,7 +627,9 @@ def _content_attribute(*, value: Any) -> str: return safe_json(jsonable_value) -def _span_output_payload(*, entity_name: str, result: Any) -> Any: +def _span_output_payload(*, log_type: str, result: Any) -> Any: + if log_type == LOG_TYPE_TOOL: + return _tool_output_payload(result=result) return result @@ -648,6 +637,7 @@ def _span_input_payload( *, bound_args: inspect.BoundArguments, tags: dict[str, Any] | None, + tool_name: str | None = None, ) -> Any: """Prefer public Workflows event summaries over internal runtime state. @@ -658,6 +648,12 @@ def _span_input_payload( exported span. """ + if tool_name is not None: + return { + "name": tool_name, + "arguments": _tool_arguments(bound_args=bound_args), + } + tags = tags or {} if LLAMA_INDEX_START_EVENT_TAG in tags: payload: dict[str, Any] = { @@ -704,8 +700,10 @@ def _span_log_type( ) -> str: normalized_name = entity_name.lower() instance_name = type(instance).__name__.lower() if instance is not None else "" - if "tool" in normalized_name or "tool" in instance_name: + if _is_executable_tool_span(entity_name=entity_name, instance=instance): return LOG_TYPE_TOOL + if _is_tool_orchestration_span(entity_name=entity_name): + return LOG_TYPE_TASK if "agent" in normalized_name or "agent" in instance_name: return LOG_TYPE_AGENT if parent_span_id is None: @@ -713,7 +711,34 @@ def _span_log_type( return LOG_TYPE_TASK +def _is_executable_tool_span(*, entity_name: str, instance: Any | None) -> bool: + """Return whether a native span wraps the tool implementation itself. + + Agent workflow methods such as ``call_tool`` and + ``aggregate_tool_results`` coordinate a call but do not execute the + application function. LlamaIndex's ``BaseTool.call`` / ``acall`` boundary + is the one span that owns both the invocation arguments and result. + """ + + operation_name = entity_name.rsplit(".", maxsplit=1)[-1] + return isinstance(instance, BaseTool) and operation_name in {"call", "acall"} + + +def _is_tool_orchestration_span(*, entity_name: str) -> bool: + operation_name = entity_name.rsplit(".", maxsplit=1)[-1] + return operation_name in {"_call_tool", "call_tool", "aggregate_tool_results"} + + def _tool_name(*, tool: Any) -> str: + metadata = getattr(tool, "metadata", None) + if metadata is not None: + for attr_name in ("name", "tool_name"): + value = getattr(metadata, attr_name, None) + if value: + return str(value) + get_name = getattr(metadata, "get_name", None) + if callable(get_name): + return str(get_name()) for attr_name in ("name", "tool_name"): value = getattr(tool, attr_name, None) if value: @@ -722,3 +747,23 @@ def _tool_name(*, tool: Any) -> str: if callable(get_name): return str(get_name()) return LLAMA_INDEX_DEFAULT_TOOL_NAME + + +def _tool_arguments(*, bound_args: inspect.BoundArguments) -> Any: + positional = to_jsonable(getattr(bound_args, "args", ())) + keyword = to_jsonable(getattr(bound_args, "kwargs", {})) + if not positional: + return keyword + if not keyword: + return positional + return {"args": positional, "kwargs": keyword} + + +def _tool_output_payload(*, result: Any) -> Any: + raw_output = getattr(result, "raw_output", None) + if raw_output is not None: + return raw_output + content = getattr(result, "content", None) + if content is not None: + return content + return result diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py index 0e639109..7b145e09 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py @@ -2,14 +2,23 @@ import json import logging from types import SimpleNamespace +from typing import Any import pytest from llama_index.core import instrumentation -from workflows import Workflow, step -from workflows.events import Event, StartEvent, StopEvent +from llama_index.core.agent.workflow import ReActAgent +from llama_index.core.llms import CompletionResponse, LLMMetadata, MockLLM +from llama_index.core.tools import FunctionTool from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.semconv_ai import SpanAttributes from opentelemetry.trace import StatusCode +from pydantic import PrivateAttr +from respan_instrumentation_llama_index import LlamaIndexInstrumentor, _instrumentation +from respan_instrumentation_llama_index._handlers import ( + RespanLlamaIndexEventHandler, + RespanLlamaIndexSpanHandler, +) +from respan_instrumentation_llama_index._serialization import extract_usage from respan_sdk.constants.llm_logging import ( LOG_TYPE_AGENT, LOG_TYPE_CHAT, @@ -26,14 +35,8 @@ from respan_tracing import RespanTelemetry from respan_tracing.core.tracer import RespanTracer from respan_tracing.testing import InMemorySpanExporter - -from respan_instrumentation_llama_index import LlamaIndexInstrumentor -from respan_instrumentation_llama_index import _instrumentation -from respan_instrumentation_llama_index._handlers import ( - RespanLlamaIndexEventHandler, - RespanLlamaIndexSpanHandler, -) -from respan_instrumentation_llama_index._serialization import extract_usage +from workflows import Workflow, step +from workflows.events import Event, StartEvent, StopEvent @pytest.fixture(autouse=True) @@ -164,6 +167,10 @@ def test_span_handler_emits_workflow_and_task_spans(span_exporter): == LOG_TYPE_WORKFLOW ) assert attrs_by_name["BaseRetriever.retrieve"][RESPAN_LOG_TYPE] == LOG_TYPE_TASK + assert ( + attrs_by_name["BaseRetriever.retrieve"][SpanAttributes.TRACELOOP_ENTITY_NAME] + == "BaseRetriever.retrieve" + ) assert ( attrs_by_name["RetrieverQueryEngine.query"][ SpanAttributes.TRACELOOP_ENTITY_OUTPUT @@ -579,8 +586,12 @@ def test_chat_events_can_disable_content_capture(span_exporter): assert SpanAttributes.TRACELOOP_ENTITY_INPUT not in chat_span.attributes assert SpanAttributes.TRACELOOP_ENTITY_OUTPUT not in chat_span.attributes assert not any( - key.startswith(f"{SpanAttributes.LLM_PROMPTS}.") - or key.startswith(f"{SpanAttributes.LLM_COMPLETIONS}.") + key.startswith( + ( + f"{SpanAttributes.LLM_PROMPTS}.", + f"{SpanAttributes.LLM_COMPLETIONS}.", + ) + ) for key in chat_span.attributes ) assert "hidden" not in json.dumps(dict(chat_span.attributes)) @@ -615,8 +626,12 @@ def test_completion_events_can_disable_content_capture(span_exporter): assert SpanAttributes.TRACELOOP_ENTITY_INPUT not in completion_span.attributes assert SpanAttributes.TRACELOOP_ENTITY_OUTPUT not in completion_span.attributes assert not any( - key.startswith(f"{SpanAttributes.LLM_PROMPTS}.") - or key.startswith(f"{SpanAttributes.LLM_COMPLETIONS}.") + key.startswith( + ( + f"{SpanAttributes.LLM_PROMPTS}.", + f"{SpanAttributes.LLM_COMPLETIONS}.", + ) + ) for key in completion_span.attributes ) assert completion_span.attributes[SpanAttributes.LLM_USAGE_TOTAL_TOKENS] == 3 @@ -719,30 +734,111 @@ def test_embedding_events_capture_full_vectors(span_exporter): assert "llm.embeddings.0" not in attributes -def test_tool_event_emits_tool_span(span_exporter): - handler = RespanLlamaIndexEventHandler() +def test_registered_react_runtime_emits_one_logical_tool_execution(span_exporter): + class ScriptedReActLLM(MockLLM): + _responses: list[str] = PrivateAttr() + _response_index: int = PrivateAttr(default=0) - handler.handle( - SimpleNamespace( - class_name=lambda: "AgentToolCallEvent", - tool=SimpleNamespace(name="lookup_order"), - arguments='{"order_id": "ord_123"}', + def __init__(self) -> None: + super().__init__(is_chat_model=True) + self._responses = [ + ( + "Thought: I need multiplication.\n" + "Action: multiply_numbers\n" + 'Action Input: {"a": 7, "b": 6}' + ), + ("Thought: I can answer without using any more tools.\nAnswer: 42"), + ] + + @property + def metadata(self) -> LLMMetadata: + return LLMMetadata( + is_chat_model=True, + model_name="scripted-react", + ) + + def complete( + self, + prompt: str, + formatted: bool = False, + **kwargs: Any, + ) -> CompletionResponse: + del prompt, formatted, kwargs + response = self._responses[ + min(self._response_index, len(self._responses) - 1) + ] + self._response_index += 1 + return CompletionResponse(text=response) + + def multiply_numbers(a: int, b: int) -> int: + return a * b + + async def run_agent(agent: ReActAgent) -> Any: + return await agent.run(user_msg="What is 7 multiplied by 6?") + + instrumentor = LlamaIndexInstrumentor() + instrumentor.activate() + try: + agent = ReActAgent( + tools=[ + FunctionTool.from_defaults( + fn=multiply_numbers, + name="multiply_numbers", + description="Multiply two integers.", + ) + ], + llm=ScriptedReActLLM(), + system_prompt="Use tools for arithmetic.", + streaming=False, ) - ) + result = asyncio.run(run_agent(agent)) + finally: + instrumentor.deactivate() - tool_span = next( + assert str(result) == "42" + spans = span_exporter.get_finished_spans() + tool_spans = [ + span for span in spans if span.attributes.get(RESPAN_LOG_TYPE) == LOG_TYPE_TOOL + ] + assert len(tool_spans) == 1 + + root_span = next(span for span in spans if span.name == "ReActAgent.run") + call_span = next( + span for span in spans if span.name == "BaseWorkflowAgent.call_tool" + ) + aggregate_span = next( span - for span in span_exporter.get_finished_spans() - if span.name == "llama_index.tool.lookup_order" + for span in spans + if span.name == "BaseWorkflowAgent.aggregate_tool_results" + ) + tool_span = tool_spans[0] + + assert root_span.parent is None + assert call_span.parent.span_id == root_span.context.span_id + assert aggregate_span.parent.span_id == root_span.context.span_id + assert tool_span.parent.span_id == call_span.context.span_id + assert tool_span.context.trace_id == root_span.context.trace_id + assert call_span.attributes[RESPAN_LOG_TYPE] == LOG_TYPE_TASK + assert aggregate_span.attributes[RESPAN_LOG_TYPE] == LOG_TYPE_TASK + assert tool_span.name == "multiply_numbers" + assert json.loads(tool_span.attributes[SpanAttributes.TRACELOOP_ENTITY_INPUT]) == { + "name": "multiply_numbers", + "arguments": {"a": 7, "b": 6}, + } + assert ( + json.loads(tool_span.attributes[SpanAttributes.TRACELOOP_ENTITY_OUTPUT]) == 42 ) + for alias in ( + "traceloop.span.kind", + "respan.span.tool_calls", + "tool_calls", + "has_tool_calls", + ): + assert alias not in tool_span.attributes - assert tool_span.attributes[RESPAN_LOG_TYPE] == LOG_TYPE_TOOL - assert "ord_123" in tool_span.attributes[SpanAttributes.TRACELOOP_ENTITY_INPUT] - - -def test_tool_event_respects_content_capture_setting(span_exporter): - handler = RespanLlamaIndexEventHandler(capture_content=False) +def test_agent_tool_call_event_does_not_claim_an_execution(span_exporter): + handler = RespanLlamaIndexEventHandler() handler.handle( SimpleNamespace( class_name=lambda: "AgentToolCallEvent", @@ -751,14 +847,34 @@ def test_tool_event_respects_content_capture_setting(span_exporter): ) ) - tool_span = next( - span - for span in span_exporter.get_finished_spans() - if span.name == "llama_index.tool.lookup_order" + assert not span_exporter.get_finished_spans() + + +def test_native_tool_span_respects_content_capture_setting(span_exporter): + handler = RespanLlamaIndexSpanHandler(capture_content=False) + tool = FunctionTool.from_defaults(fn=lambda value: value, name="echo") + span_id = "FunctionTool.call-55555555-5555-5555-5555-555555555555" + bound_args = SimpleNamespace(args=(), kwargs={"value": "secret"}) + + handler.span_enter( + id_=span_id, + bound_args=bound_args, + instance=tool, + parent_id=None, + ) + handler.span_exit( + id_=span_id, + bound_args=bound_args, + instance=tool, + result=SimpleNamespace(raw_output="secret"), ) + tool_span = span_exporter.get_finished_spans()[0] + assert tool_span.name == "echo" assert tool_span.attributes[RESPAN_LOG_TYPE] == LOG_TYPE_TOOL assert SpanAttributes.TRACELOOP_ENTITY_INPUT not in tool_span.attributes + assert SpanAttributes.TRACELOOP_ENTITY_OUTPUT not in tool_span.attributes + assert "secret" not in json.dumps(dict(tool_span.attributes)) def test_exception_event_marks_open_event_span_error(span_exporter): diff --git a/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_instrumentation.py index 253da987..dde5d334 100644 --- a/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_instrumentation.py @@ -5,6 +5,7 @@ from opentelemetry.semconv_ai import SpanAttributes from respan_sdk.constants.span_attributes import RESPAN_LOG_TYPE + from ._native_instrumentation import ( NativeClientInstrumentor, PatchSpec, diff --git a/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_native_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_native_instrumentation.py index 972ad7ef..d562c8c8 100644 --- a/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_native_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-marqo/src/respan_instrumentation_marqo/_native_instrumentation.py @@ -9,7 +9,8 @@ from collections.abc import Mapping, Sequence from contextvars import ContextVar from dataclasses import dataclass, field -from typing import Any +from typing import Any, ClassVar +from urllib.parse import urlsplit, urlunsplit from opentelemetry import trace from opentelemetry.instrumentation.utils import unwrap @@ -41,7 +42,10 @@ class PatchSpec: def _jsonable(value: Any, *, depth: int = 0) -> Any: if depth > 5: - return repr(value) + return { + "type": f"{type(value).__module__}.{type(value).__qualname__}", + "truncated": True, + } if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, bytes): @@ -70,8 +74,16 @@ def _jsonable(value: Any, *, depth: int = 0) -> Any: try: return _jsonable(method(), depth=depth + 1) except Exception: - pass - return repr(value) + logger.debug( + "Failed to serialize %s with %s", + type(value).__qualname__, + method_name, + exc_info=True, + ) + identity = _instance_identity(value) + if identity: + return identity + return {"type": f"{type(value).__module__}.{type(value).__qualname__}"} def _json_dumps(value: Any) -> str: @@ -89,7 +101,7 @@ def _call_input( try: bound = inspect.signature(wrapped).bind_partial(*args, **kwargs) return {key: value for key, value in bound.arguments.items() if key != "self"} - except Exception: + except (TypeError, ValueError): return {"args": list(args), "kwargs": kwargs} @@ -109,14 +121,43 @@ def _instance_identity(instance: Any) -> dict[str, str]: ): value = getattr(instance, key, None) if isinstance(value, (str, int)): - identity[key.lstrip("_")] = str(value) + identity_key = key.lstrip("_") + identity[identity_key] = ( + _sanitize_endpoint(str(value)) if identity_key == "uri" else str(value) + ) config = getattr(instance, "_config", None) host = getattr(config, "host", None) if isinstance(host, str): - identity["host"] = host + identity["host"] = _sanitize_endpoint(host) return identity +def _sanitize_endpoint(value: str) -> str: + """Remove endpoint credentials while retaining stable routing identity.""" + + candidate = value.strip() + if not candidate: + return candidate + + has_scheme = "://" in candidate + try: + parsed = urlsplit(candidate if has_scheme else f"//{candidate}") + hostname = parsed.hostname + if hostname is None: + if has_scheme: + return urlunsplit((parsed.scheme, "", parsed.path, "", "")) + return "" + port = parsed.port + except ValueError: + return "" + + safe_hostname = f"[{hostname}]" if ":" in hostname else hostname + netloc = f"{safe_hostname}:{port}" if port is not None else safe_hostname + if has_scheme: + return urlunsplit((parsed.scheme, netloc, parsed.path, "", "")) + return urlunsplit(("", netloc, parsed.path, "", "")).removeprefix("//") + + class NativeClientInstrumentor: """Base lifecycle and span mapping for vendor client adapters.""" @@ -125,7 +166,7 @@ class NativeClientInstrumentor: patches: tuple[PatchSpec, ...] = () _patches_applied = False _activation_count = 0 - _patched_targets: list[tuple[type, str]] = [] + _patched_targets: ClassVar[list[tuple[type, str]]] = [] _active_call: ContextVar[bool] = ContextVar( "respan_native_client_active", default=False, diff --git a/python-sdks/instrumentations/respan-instrumentation-marqo/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-marqo/tests/test_instrumentation.py index 45f563c9..149e918a 100644 --- a/python-sdks/instrumentations/respan-instrumentation-marqo/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-marqo/tests/test_instrumentation.py @@ -1,11 +1,19 @@ +import json import sys +import threading +from collections import Counter from contextlib import contextmanager -from types import ModuleType +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import ModuleType, SimpleNamespace import pytest -from opentelemetry.trace import StatusCode +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, +) from opentelemetry.semconv_ai import SpanAttributes - +from opentelemetry.trace import StatusCode from respan_instrumentation_marqo import MarqoInstrumentor from respan_instrumentation_marqo import ( _native_instrumentation as native_instrumentation, @@ -13,6 +21,63 @@ from respan_sdk.constants.span_attributes import RESPAN_LOG_TYPE +class _MarqoLoopbackHandler(BaseHTTPRequestHandler): + def log_message(self, _format, *_args): + return + + def _send_json(self, status, payload): + content = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(content))) + self.end_headers() + self.wfile.write(content) + + def do_GET(self): + if self.path == "/": + self._send_json(200, {"version": "3.18.2"}) + return + if self.path == "/indexes/docs/health": + self._send_json( + 503, + { + "message": "marqo unavailable", + "code": "service_unavailable", + "type": "service_unavailable", + "link": "", + }, + ) + return + self._send_json(404, {"message": "not found"}) + + def do_POST(self): + length = int(self.headers.get("Content-Length", "0")) + self.rfile.read(length) + if self.path == "/indexes/docs/search": + self._send_json( + 200, + { + "hits": [{"_id": "doc-1", "_score": 0.9}], + "processingTimeMs": 1, + }, + ) + return + self._send_json(404, {"message": "not found"}) + + +@contextmanager +def _marqo_loopback_url(): + server = ThreadingHTTPServer(("127.0.0.1", 0), _MarqoLoopbackHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}" + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + class _Span: def __init__(self, name): self.name = name @@ -110,3 +175,108 @@ def health(self): traced_count = len(tracer.spans) assert index.search("deactivated")["hits"] assert len(tracer.spans) == traced_count + + +def test_identity_serialization_removes_endpoint_credentials(): + identity = SimpleNamespace( + uri=( + "https://uri-user:uri-password@marqo.example:8443/indexes/docs" + "?api_key=uri-secret#private" + ), + _config=SimpleNamespace( + host=( + "host-user:host-password@api.marqo.example:9443" + "?token=host-secret#private" + ) + ), + ) + + output = native_instrumentation._json_dumps(identity) + + assert json.loads(output) == { + "host": "api.marqo.example:9443", + "uri": "https://marqo.example:8443/indexes/docs", + } + for secret in ( + "host-password", + "host-secret", + "host-user", + "uri-password", + "uri-secret", + "uri-user", + ): + assert secret not in output + + +def test_real_client_exports_stable_success_and_error_spans(monkeypatch): + import marqo + from marqo.errors import MarqoWebError + + exporter = InMemorySpanExporter() + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr( + native_instrumentation.trace, + "get_tracer", + lambda _: tracer_provider.get_tracer("respan-marqo-test"), + ) + + MarqoInstrumentor._patches_applied = False + instrumentor = MarqoInstrumentor() + instrumentor.activate() + try: + with _marqo_loopback_url() as url: + client = marqo.Client(url=url) + index = client.index("docs") + assert index.search("observability")["hits"] == [ + {"_id": "doc-1", "_score": 0.9} + ] + with pytest.raises(MarqoWebError, match="marqo unavailable"): + index.health() + + finished_spans = exporter.get_finished_spans() + expected_names = Counter( + { + "marqo.client.index": 1, + "marqo.index.health": 1, + "marqo.index.search": 1, + } + ) + assert len(finished_spans) == sum(expected_names.values()) + assert Counter(span.name for span in finished_spans) == expected_names + spans = {span.name: span for span in finished_spans} + + index_output = spans["marqo.client.index"].attributes[ + SpanAttributes.TRACELOOP_ENTITY_OUTPUT + ] + assert json.loads(index_output) == {"index_name": "docs"} + assert "0x" not in index_output + + search_span = spans["marqo.index.search"] + assert search_span.status.status_code == StatusCode.OK + assert json.loads( + search_span.attributes[SpanAttributes.TRACELOOP_ENTITY_OUTPUT] + )["hits"] == [{"_id": "doc-1", "_score": 0.9}] + + failed_span = spans["marqo.index.health"] + assert failed_span.status.status_code == StatusCode.ERROR + failed_output = json.loads( + failed_span.attributes[SpanAttributes.TRACELOOP_ENTITY_OUTPUT] + ) + assert failed_output["error"] == "MarqoWebError" + assert "marqo unavailable" in failed_output["message"] + + forbidden_aliases = { + "has_tool_calls", + "model", + "span_tools", + "tool_calls", + "tools", + } + for span in spans.values(): + assert span.attributes[RESPAN_LOG_TYPE] == "task" + assert SpanAttributes.TRACELOOP_ENTITY_INPUT in span.attributes + assert forbidden_aliases.isdisjoint(span.attributes) + finally: + instrumentor.deactivate() + tracer_provider.shutdown() From 2b64d6c4db71bb57e72d199b02b2a24e74395aa1 Mon Sep 17 00:00:00 2001 From: Nightingalelyy Date: Tue, 18 Aug 2026 13:39:13 +0800 Subject: [PATCH 2/2] fix(instrumentation-py): align LlamaIndex LLM span types --- .../src/respan_instrumentation_llama_index/_handlers.py | 6 +++--- .../tests/test_instrumentation.py | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py index fb85b833..d18bb15f 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/src/respan_instrumentation_llama_index/_handlers.py @@ -20,9 +20,9 @@ from respan_sdk.constants.llm_logging import ( LOG_TYPE_AGENT, LOG_TYPE_CHAT, - LOG_TYPE_COMPLETION, LOG_TYPE_EMBEDDING, LOG_TYPE_TASK, + LOG_TYPE_TEXT, LOG_TYPE_TOOL, LOG_TYPE_WORKFLOW, LogMethodChoices, @@ -349,8 +349,8 @@ def _handle_completion_start(self, *, event: Any) -> None: prompt = getattr(event, "prompt", "") attributes = _llm_base_attributes( entity_name=LLAMA_INDEX_COMPLETION_SPAN_NAME, - log_type=LOG_TYPE_COMPLETION, - request_type=LLMRequestTypeValues.COMPLETION.value, + log_type=LOG_TYPE_TEXT, + request_type=LLMRequestTypeValues.CHAT.value, model_dict=model_dict, ) if self.capture_content: diff --git a/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py b/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py index 7b145e09..899adc49 100644 --- a/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py +++ b/python-sdks/instrumentations/respan-instrumentation-llama-index/tests/test_instrumentation.py @@ -22,9 +22,9 @@ from respan_sdk.constants.llm_logging import ( LOG_TYPE_AGENT, LOG_TYPE_CHAT, - LOG_TYPE_COMPLETION, LOG_TYPE_EMBEDDING, LOG_TYPE_TASK, + LOG_TYPE_TEXT, LOG_TYPE_TOOL, LOG_TYPE_WORKFLOW, ) @@ -674,8 +674,8 @@ def test_completion_events_emit_text_span(span_exporter): ) attributes = text_span.attributes - assert attributes[RESPAN_LOG_TYPE] == LOG_TYPE_COMPLETION - assert attributes[SpanAttributes.LLM_REQUEST_TYPE] == "completion" + assert attributes[RESPAN_LOG_TYPE] == LOG_TYPE_TEXT + assert attributes[SpanAttributes.LLM_REQUEST_TYPE] == "chat" assert ( attributes[f"{SpanAttributes.LLM_PROMPTS}.0.content"] == "Complete this sentence"