diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md index da5f3c297..29f2e8987 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Changed + +- Align streaming timing, reasoning-token usage, provider names, and agent + version attributes with the current upstream GenAI conventions. + ### Fixed - Start AgentScope v2 streaming LLM spans before invoking the underlying model diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py index 982adf987..3e5e8b9cd 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_v2_middleware.py @@ -20,7 +20,13 @@ import json import logging import timeit -from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence +from collections.abc import ( + AsyncGenerator, + Awaitable, + Callable, + Mapping, + Sequence, +) from contextvars import ContextVar from dataclasses import asdict, is_dataclass from typing import Any @@ -179,6 +185,7 @@ async def on_model_call( try: result = await next_handler(**input_kwargs) if inspect.isasyncgen(result): + invocation.stream = True return self._wrap_model_stream( result, invocation, @@ -201,14 +208,18 @@ async def _wrap_model_stream( invocation: LLMInvocation, handler: ExtendedTelemetryHandler, ) -> AsyncGenerator[ChatResponse, None]: - first_token_seen = False last_chunk = None closed = False try: async for chunk in result: - if not first_token_seen: - invocation.monotonic_first_token_s = timeit.default_timer() - first_token_seen = True + now = timeit.default_timer() + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _chat_response_has_token(chunk) + ): + invocation.monotonic_first_token_s = now last_chunk = chunk yield chunk except BaseException as exc: @@ -324,6 +335,7 @@ def _create_llm_invocation( provider=_get_provider_name(model), input_messages=_messages_to_inputs(input_kwargs.get("messages")), tool_definitions=_tool_definitions(input_kwargs.get("tools")), + stream=True if _is_streaming_model(model, input_kwargs) else None, ) parameters = getattr(model, "parameters", None) for source in (parameters, input_kwargs): @@ -461,6 +473,30 @@ def _is_first_token_event(item: Any) -> bool: return event_type in _FIRST_TOKEN_EVENT_TYPES +def _chat_response_has_token(response: Any) -> bool: + content = _safe_get(response, "content") + if isinstance(content, str): + return bool(content) + for block in content or []: + block_type = _safe_get(block, "type") + if block_type == "text" and _safe_get(block, "text"): + return True + if block_type == "thinking" and _safe_get(block, "thinking"): + return True + if block_type == "tool_call": + return True + return False + + +def _safe_get(value: Any, name: str) -> Any: + if isinstance(value, Mapping): + return value.get(name) + try: + return getattr(value, name, None) + except (AttributeError, KeyError): + return None + + def _middleware_arg_position() -> int | None: try: parameters = list(inspect.signature(Agent.__init__).parameters) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_wrapper.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_wrapper.py index 38a9cc85e..3e600df8d 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_wrapper.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/_wrapper.py @@ -60,6 +60,7 @@ import threading import timeit import weakref +from collections.abc import Mapping from dataclasses import dataclass, field from functools import wraps from typing import Any, AsyncGenerator, MutableMapping @@ -80,6 +81,31 @@ logger = logging.getLogger(__name__) + +def _safe_get(value: Any, name: str) -> Any: + if isinstance(value, Mapping): + return value.get(name) + try: + return getattr(value, name, None) + except (AttributeError, KeyError): + return None + + +def _stream_chunk_has_token(chunk: Any) -> bool: + content = _safe_get(chunk, "content") + if isinstance(content, str): + return bool(content) + for block in content or []: + block_type = _safe_get(block, "type") + if block_type == "text" and _safe_get(block, "text"): + return True + if block_type == "thinking" and _safe_get(block, "thinking"): + return True + if block_type == "tool_call": + return True + return bool(_safe_get(chunk, "tool_calls")) + + # Fixed hook name shared by all concurrent invocations on a given agent. # Using a stable name (rather than a per-call uuid) lets AgentScope's own # registry naturally deduplicate, so a single set of callbacks fires for @@ -392,12 +418,16 @@ async def _wrap_streaming_response( """Wrap streaming response to update invocation when done.""" try: last_chunk = None - first_token_received = False async for chunk in generator: - # Record time when first token is received - if not first_token_received: - first_token_received = True - invocation.monotonic_first_token_s = timeit.default_timer() + now = timeit.default_timer() + invocation.stream = True + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _stream_chunk_has_token(chunk) + ): + invocation.monotonic_first_token_s = now last_chunk = chunk yield chunk diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/message_converter.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/message_converter.py index 4ca8072ca..2b009336c 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/message_converter.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/message_converter.py @@ -674,7 +674,7 @@ def get_message_converter( elif provide_name == "ollama": # AgentScopeGenAiProviderName.OLLAMA.value parser = OllamaMessageParser() elif ( - provide_name == "moonshot" + provide_name == "moonshot_ai" ): # AgentScopeGenAiProviderName.MOONSHOT.value # Moonshot uses OpenAI-compatible API format parser = OpenAIMessageParser() diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/utils.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/utils.py index 54ff15d97..99dc3e557 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/utils.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/src/opentelemetry/instrumentation/agentscope/utils.py @@ -69,7 +69,7 @@ class AgentScopeGenAiProviderName(str, Enum): OLLAMA = "ollama" DASHSCOPE = "dashscope" - MOONSHOT = "moonshot" + MOONSHOT = "moonshot_ai" # Provider name mapping based on class names @@ -79,6 +79,7 @@ class AgentScopeGenAiProviderName(str, Enum): "anthropic": GenAIAttributes.GenAiProviderNameValues.ANTHROPIC.value, "dashscope": AgentScopeGenAiProviderName.DASHSCOPE.value, "ollama": AgentScopeGenAiProviderName.OLLAMA.value, + "moonshot": AgentScopeGenAiProviderName.MOONSHOT.value, } # Base URL to provider mapping for OpenAI-compatible APIs @@ -89,6 +90,7 @@ class AgentScopeGenAiProviderName(str, Enum): GenAIAttributes.GenAiProviderNameValues.DEEPSEEK.value, ), ("dashscope.aliyuncs.com", AgentScopeGenAiProviderName.DASHSCOPE.value), + ("api.moonshot.cn", AgentScopeGenAiProviderName.MOONSHOT.value), ] diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_utils.py b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_utils.py index b8ade817f..c8d778297 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_utils.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agentscope/tests/test_utils.py @@ -37,6 +37,7 @@ create_embedding_invocation, create_llm_invocation, entry_baggage_identity_attributes, + get_provider_name, ) from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( GEN_AI_SESSION_ID, @@ -47,6 +48,18 @@ class TestUtils: + def test_moonshot_provider_uses_upstream_value(self): + class MoonshotChatModel: + pass + + assert get_provider_name(MoonshotChatModel()) == "moonshot_ai" + + def test_moonshot_compatible_base_url_uses_upstream_value(self): + class DashScopeChatModel: + base_http_api_url = "https://api.moonshot.cn/v1" + + assert get_provider_name(DashScopeChatModel()) == "moonshot_ai" + def test_convert_msg_with_tool_result(self): """Test conversion of AgentScope Msg with ToolResultBlock (End-to-End)""" # Construct a Msg object simulating a tool execution result diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agno/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-agno/CHANGELOG.md index 540752d9e..d1e6569be 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agno/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agno/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Align streaming timing and response extraction with the current upstream + GenAI conventions. - Split Agno 2.x tool-call loops into ReAct step spans with one LLM span per provider request, and populate streaming agent token usage from child LLM spans when Agno run events do not include metrics. diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-agno/src/opentelemetry/instrumentation/agno/_wrapper.py b/instrumentation-loongsuite/loongsuite-instrumentation-agno/src/opentelemetry/instrumentation/agno/_wrapper.py index f432497b4..bfcf5be61 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-agno/src/opentelemetry/instrumentation/agno/_wrapper.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-agno/src/opentelemetry/instrumentation/agno/_wrapper.py @@ -81,6 +81,37 @@ def _is_stream_close(exc: BaseException) -> bool: return isinstance(exc, (GeneratorExit, StopIteration, StopAsyncIteration)) +def _safe_get(value: Any, name: str) -> Any: + if isinstance(value, Mapping): + return value.get(name) + try: + return getattr(value, name, None) + except (AttributeError, KeyError): + return None + + +def _response_has_token(response: Any) -> bool: + return any( + ( + _safe_get(response, "content"), + _safe_get(response, "reasoning_content"), + _safe_get(response, "redacted_reasoning_content"), + _safe_get(response, "tool_calls"), + ) + ) + + +def _mark_llm_stream_item(invocation: Any, response: Any) -> None: + now = timeit.default_timer() + invocation.stream = True + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if invocation.monotonic_first_token_s is None and _response_has_token( + response + ): + invocation.monotonic_first_token_s = now + + @dataclass class _AgnoRunState: handler: ExtendedTelemetryHandler @@ -698,8 +729,7 @@ def process_response_stream( try: stream = wrapped(*args, **kwargs) for response in stream: - if invocation.monotonic_first_token_s is None: - invocation.monotonic_first_token_s = timeit.default_timer() + _mark_llm_stream_item(invocation, response) responses.append(response) yield response self._finish_llm_call( @@ -744,8 +774,7 @@ async def aprocess_response_stream( if inspect.isawaitable(stream): stream = await stream async for response in stream: - if invocation.monotonic_first_token_s is None: - invocation.monotonic_first_token_s = timeit.default_timer() + _mark_llm_stream_item(invocation, response) responses.append(response) yield response self._finish_llm_call( @@ -886,8 +915,7 @@ def response_stream( try: stream = wrapped(*args, **kwargs) for response in stream: - if invocation.monotonic_first_token_s is None: - invocation.monotonic_first_token_s = timeit.default_timer() + _mark_llm_stream_item(invocation, response) responses.append(response) yield response if responses: @@ -967,8 +995,7 @@ async def aresponse_stream( if inspect.isawaitable(stream): stream = await stream async for response in stream: - if invocation.monotonic_first_token_s is None: - invocation.monotonic_first_token_s = timeit.default_timer() + _mark_llm_stream_item(invocation, response) responses.append(response) yield response if responses: diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/CHANGELOG.md index e5eea384b..ede3d3ddc 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Capture time to first chunk independently from time to first token, and emit + the stream request indicator only when AutoGen enables model streaming. - Avoid enriching non-AutoGen spans that use overlapping GenAI operation names. ## Version 0.7.0 (2026-07-03) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/src/opentelemetry/instrumentation/autogen/patch.py b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/src/opentelemetry/instrumentation/autogen/patch.py index a49f1b021..90a43e277 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/src/opentelemetry/instrumentation/autogen/patch.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/src/opentelemetry/instrumentation/autogen/patch.py @@ -464,15 +464,20 @@ def _direct_model_create_stream_wrapper(wrapped, instance, args, kwargs): # typ async def _generator(): # type: ignore[no-untyped-def] handler = _get_handler() invocation = _make_direct_model_invocation(instance, args, kwargs) + invocation.stream = True handler.start_llm(invocation) _mark_autogen_live_span(invocation) try: async for item in wrapped(*args, **kwargs): + now = timeit.default_timer() + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now if ( isinstance(item, str) + and bool(item) and invocation.monotonic_first_token_s is None ): - invocation.monotonic_first_token_s = timeit.default_timer() + invocation.monotonic_first_token_s = now elif type(item).__name__ == "CreateResult": apply_create_result(invocation, item) yield item @@ -523,6 +528,11 @@ async def _generator(): # type: ignore[no-untyped-def] agent_name=agent_name, output_type=output_type, ) + invocation.stream = ( + True + if _arg_value(args, kwargs, "model_client_stream") is True + else None + ) handler.start_llm(invocation) _mark_autogen_live_span(invocation) try: @@ -531,11 +541,15 @@ async def _generator(): # type: ignore[no-untyped-def] _suppress_direct_model_span, True, ): + now = timeit.default_timer() + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now if ( type(item).__name__ == "ModelClientStreamingChunkEvent" + and bool(getattr(item, "content", None)) and invocation.monotonic_first_token_s is None ): - invocation.monotonic_first_token_s = timeit.default_timer() + invocation.monotonic_first_token_s = now elif type(item).__name__ == "CreateResult": apply_create_result(invocation, item) yield item diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/tests/test_patch.py b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/tests/test_patch.py index 898d69d7c..5cb81f1f5 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-autogen/tests/test_patch.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-autogen/tests/test_patch.py @@ -368,6 +368,37 @@ async def wrapped(*args, **kwargs): ("start_llm", "qwen-plus"), ("stop_llm", "qwen-plus"), ] + assert handler.invocations[0].stream is None + + +@pytest.mark.asyncio +async def test_llm_wrapper_marks_streaming_call(): + handler = Handler() + _set_handler(handler) + + async def wrapped(*args, **kwargs): + yield CreateResult() + + items = [ + item + async for item in patch._call_llm_wrapper( + wrapped, + None, + (), + { + "model_client": ModelClient(), + "model_client_stream": True, + "system_messages": [], + "model_context": ModelContext(), + "workbench": [], + "handoff_tools": [], + "agent_name": "assistant", + }, + ) + ] + + assert len(items) == 1 + assert handler.invocations[0].stream is True @pytest.mark.asyncio diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/CHANGELOG.md index ec8f896f8..c517ec46b 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Use the upstream `moonshot_ai` provider name. - Capture Claude Agent SDK default model names from stream metadata when the client options and environment do not provide an explicit model. - Fail open LLM spans when Claude Agent SDK streams raise exceptions, preserving diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/tests/test_unit.py b/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/tests/test_unit.py index 627ef3de0..b2cbe1eaf 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/tests/test_unit.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-claude-agent-sdk/tests/test_unit.py @@ -111,7 +111,7 @@ def test_infer_provider_from_base_url(): # Test Moonshot (extended provider) result = infer_provider_from_base_url("https://api.moonshot.cn/v1") - assert result == "moonshot" + assert result == "moonshot_ai" # Test Anthropic (defaults to anthropic) os.environ["ANTHROPIC_BASE_URL"] = "https://api.anthropic.com" diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/CHANGELOG.md index 323fc2e1f..6e3eb8d59 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Capture stream request state, reasoning token usage, response IDs, and + independent time-to-first-chunk and time-to-first-token timestamps. + ## Version 0.7.0 (2026-07-03) ### Fixed diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/generation.py b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/generation.py index 3bd927664..b5469701e 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/generation.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/generation.py @@ -32,6 +32,23 @@ logger = logging.getLogger(__name__) +def _generation_chunk_has_token(chunk) -> bool: + output = getattr(chunk, "output", None) + if output is None: + return False + if getattr(output, "text", None) or getattr(output, "content", None): + return True + for choice in getattr(output, "choices", None) or []: + message = getattr(choice, "message", None) + if message is None: + continue + if getattr(message, "content", None): + return True + if getattr(message, "tool_calls", None): + return True + return False + + def wrap_generation_call(wrapped, instance, args, kwargs, handler=None): """Wrapper for Generation.call (sync). @@ -170,14 +187,17 @@ def _wrap_sync_generator( """ last_response = None accumulated_text = "" - first_token_received = False - try: for chunk in generator: - # Record time when first token is received - if not first_token_received: - first_token_received = True - invocation.monotonic_first_token_s = timeit.default_timer() + now = timeit.default_timer() + invocation.stream = True + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _generation_chunk_has_token(chunk) + ): + invocation.monotonic_first_token_s = now last_response = chunk @@ -231,14 +251,17 @@ async def _wrap_async_generator( """ last_response = None accumulated_text = "" - first_token_received = False - try: async for chunk in generator: - # Record time when first token is received - if not first_token_received: - first_token_received = True - invocation.monotonic_first_token_s = timeit.default_timer() + now = timeit.default_timer() + invocation.stream = True + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _generation_chunk_has_token(chunk) + ): + invocation.monotonic_first_token_s = now last_response = chunk diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/multimodal_conversation.py b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/multimodal_conversation.py index a870136e7..769e0ec9b 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/multimodal_conversation.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-dashscope/src/opentelemetry/instrumentation/dashscope/patch/multimodal_conversation.py @@ -32,6 +32,24 @@ logger = logging.getLogger(__name__) +def _multimodal_chunk_has_token(chunk) -> bool: + output = getattr(chunk, "output", None) + for choice in getattr(output, "choices", None) or []: + message = getattr(choice, "message", None) + if message is None: + continue + content = getattr(message, "content", None) + if isinstance(content, str) and content: + return True + if isinstance(content, list): + for item in content: + if isinstance(item, dict) and item.get("text"): + return True + if getattr(message, "tool_calls", None): + return True + return False + + def wrap_multimodal_conversation_call( wrapped, instance, args, kwargs, handler=None ): @@ -112,14 +130,17 @@ def _wrap_multimodal_sync_generator( """ last_response = None accumulated_text = "" - first_token_received = False - try: for chunk in generator: - # Record time when first token is received - if not first_token_received: - first_token_received = True - invocation.monotonic_first_token_s = timeit.default_timer() + now = timeit.default_timer() + invocation.stream = True + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _multimodal_chunk_has_token(chunk) + ): + invocation.monotonic_first_token_s = now last_response = chunk diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/CHANGELOG.md index 53341c6fc..8bc84d223 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Capture stream request state, reasoning token usage, response IDs, and + independent time-to-first-chunk and time-to-first-token timestamps. - Capture `gen_ai.skill.*` attributes on Google ADK SkillToolset `load_skill` and `load_skill_resource` execute-tool spans. diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/src/opentelemetry/instrumentation/google_adk/internal/_plugin.py b/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/src/opentelemetry/instrumentation/google_adk/internal/_plugin.py index a77ece0f9..ebc27d9b4 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/src/opentelemetry/instrumentation/google_adk/internal/_plugin.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-google-adk/src/opentelemetry/instrumentation/google_adk/internal/_plugin.py @@ -463,6 +463,9 @@ async def before_model_callback( invocation = LLMInvocation( request_model=model_name, provider=self._extractors._extract_provider_name(model_name), + stream=( + True if getattr(llm_request, "stream", False) else None + ), ) # Extract input messages @@ -484,6 +487,9 @@ async def before_model_callback( top_p = self._get_real_attr(config, "top_p") if top_p is not None: invocation.top_p = top_p + top_k = self._get_real_attr(config, "top_k") + if isinstance(top_k, int): + invocation.top_k = top_k # Extract conversation_id and user_id session_id = self._session_id_from_callback_context( @@ -541,10 +547,15 @@ async def after_model_callback( ) if self._is_streaming_partial_response(llm_response): - if llm_invocation.monotonic_first_token_s is None: - llm_invocation.monotonic_first_token_s = ( - timeit.default_timer() - ) + now = timeit.default_timer() + llm_invocation.stream = True + if llm_invocation.monotonic_first_chunk_s is None: + llm_invocation.monotonic_first_chunk_s = now + if ( + llm_invocation.monotonic_first_token_s is None + and self._llm_response_has_token(llm_response) + ): + llm_invocation.monotonic_first_token_s = now _logger.debug( "Captured partial LLM response for %s", request_key, @@ -1013,6 +1024,12 @@ def _update_llm_invocation_from_response( if response_model: invocation.response_model_name = response_model + for attr_name in ("response_id", "responseId"): + response_id = self._get_real_attr(llm_response, attr_name) + if response_id: + invocation.response_id = str(response_id) + break + usage = getattr(llm_response, "usage_metadata", None) if self._is_mock_placeholder( usage @@ -1032,6 +1049,15 @@ def _update_llm_invocation_from_response( if output_tokens is not None: invocation.output_tokens = output_tokens + reasoning_tokens = self._get_real_attr( + usage, "thoughts_token_count" + ) + if isinstance(reasoning_tokens, int): + invocation.reasoning_output_tokens = reasoning_tokens + invocation.output_tokens = ( + output_tokens if isinstance(output_tokens, int) else 0 + ) + reasoning_tokens + finish_reason = self._get_real_attr(llm_response, "finish_reason") if finish_reason: if hasattr(finish_reason, "value"): @@ -1122,6 +1148,15 @@ def _extract_text_from_llm_response( return "" + def _llm_response_has_token(self, llm_response: LlmResponse) -> bool: + if self._extract_text_from_llm_response(llm_response): + return True + content = self._get_real_attr(llm_response, "content") + for part in self._get_real_attr(content, "parts") or []: + if self._get_real_attr(part, "function_call"): + return True + return False + def _convert_text_to_output_messages( self, text: str, llm_response: LlmResponse ) -> List[OutputMessage]: diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/CHANGELOG.md index 74194d067..754793d57 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Populate upstream stream and time-to-first-chunk fields for Hermes LLM spans. + ## Version 0.7.0 (2026-07-03) ### Fixed diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/wrappers.py b/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/wrappers.py index 73427e2c3..b948463e5 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/wrappers.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/wrappers.py @@ -307,6 +307,7 @@ def __call__(self, wrapped, instance, args, kwargs): api_kwargs = args[0] if args else kwargs.get("api_kwargs", {}) invocation = create_llm_invocation(instance, api_kwargs) + invocation.stream = True if self._streaming else None provider = invocation.provider or provider_name(instance) model = invocation.request_model or "" self._handler.start_llm(invocation) @@ -318,6 +319,8 @@ def __call__(self, wrapped, instance, args, kwargs): def _wrapped_first_delta(): now = timeit.default_timer() + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now invocation.monotonic_first_token_s = now if current_state["first_token_monotonic_s"] is None: current_state["first_token_monotonic_s"] = now diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/CHANGELOG.md index 2edd2506e..0b1e922f5 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Mark streaming LLM requests with `gen_ai.request.stream=true`. +- Omit opt-in Retrieval query and document attributes unless content capture is + enabled. + ## Version 0.7.0 (2026-07-03) ### Added diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/internal/_tracer.py b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/internal/_tracer.py index aae2dd507..bcd8a195b 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/internal/_tracer.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/internal/_tracer.py @@ -273,6 +273,7 @@ def on_llm_new_token( # type: ignore[override] and rd.invocation is not None ): inv: LLMInvocation = rd.invocation + inv.stream = True if inv.monotonic_first_token_s is None: inv.monotonic_first_token_s = timeit.default_timer() return None diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/tests/test_retriever_spans.py b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/tests/test_retriever_spans.py index 974de566e..228e78c26 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-langchain/tests/test_retriever_spans.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-langchain/tests/test_retriever_spans.py @@ -117,7 +117,7 @@ def test_retrieval_documents_captured(self, instrument, span_exporter): def test_no_content_when_disabled( self, instrument_no_content, span_exporter ): - """When content capture is NO_CONTENT: query omitted; documents record id and score only.""" + """NO_CONTENT omits the opt-in query and document attributes.""" retriever = FakeRetriever() retriever.invoke("secret query") @@ -128,8 +128,6 @@ def test_no_content_when_disabled( assert GEN_AI_RETRIEVAL_QUERY_TEXT not in attrs, ( "Query should NOT be captured when content capture is disabled" ) - # Documents are recorded with id and score only (no content) when NO_CONTENT - docs_val = attrs.get(GEN_AI_RETRIEVAL_DOCUMENTS, "") - assert "secret query" not in docs_val, ( - "Document content should NOT be captured when NO_CONTENT" + assert GEN_AI_RETRIEVAL_DOCUMENTS not in attrs, ( + "Documents are opt-in and should not be captured when disabled" ) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/CHANGELOG.md index c8fe378c3..f2ddd3122 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Capture stream request state, reasoning request level, reasoning token usage, + and independent first-chunk and first-token timestamps. + ## Version 0.7.0 (2026-07-03) There are no changelog entries for this release. diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_stream_wrapper.py b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_stream_wrapper.py index 20f6af70d..7d88578e7 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_stream_wrapper.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_stream_wrapper.py @@ -43,6 +43,13 @@ def __init__(self, invocation: Any = None): self._choice_states: dict[int, dict[str, Any]] = {} def record_chunk(self, chunk: Any) -> None: + if self.invocation is not None: + self.invocation.stream = True + if self.invocation.monotonic_first_chunk_s is None: + self.invocation.monotonic_first_chunk_s = ( + timeit.default_timer() + ) + choices = get_litellm_value(chunk, "choices") or [] if not choices: return diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_utils.py b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_utils.py index cceebb523..0905495e3 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_utils.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/src/opentelemetry/instrumentation/litellm/_utils.py @@ -239,6 +239,15 @@ def _apply_usage_to_invocation( if include_output_tokens and output_tokens is not None: invocation.output_tokens = output_tokens + completion_details = get_litellm_value(usage, "completion_tokens_details") + reasoning_tokens = get_litellm_value( + completion_details, "reasoning_tokens" + ) + if reasoning_tokens is not None and hasattr( + invocation, "reasoning_output_tokens" + ): + invocation.reasoning_output_tokens = reasoning_tokens + prompt_details = get_litellm_value(usage, "prompt_tokens_details") cached_tokens = get_litellm_value(prompt_details, "cached_tokens") if cached_tokens is not None and hasattr( @@ -265,7 +274,7 @@ def parse_provider_from_model(model: str) -> Optional[str]: return None if "/" in model: - return model.split("/")[0] + return model.split("/", 1)[0] # Fallback: try to infer from model name patterns if "gpt" in model.lower(): @@ -522,6 +531,10 @@ def create_llm_invocation_from_litellm(**kwargs): operation_name=GenAiOperationNameValues.CHAT.value, input_messages=input_messages, ) + invocation.stream = True if kwargs.get("stream") is True else None + reasoning_level = kwargs.get("reasoning_effort") + if reasoning_level is not None: + invocation.reasoning_level = str(reasoning_level) if system_instruction: invocation.system_instruction = system_instruction diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/tests/test_genai_util_wrapper.py b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/tests/test_genai_util_wrapper.py index 7704b2406..ba33679a9 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-litellm/tests/test_genai_util_wrapper.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-litellm/tests/test_genai_util_wrapper.py @@ -444,6 +444,14 @@ def fake_completion(*args, **kwargs): assert len(spans) == 1 span = spans[0] assert "gen_ai.response.time_to_first_token" in span.attributes + assert "gen_ai.response.time_to_first_chunk" in span.attributes + assert span.attributes["gen_ai.request.stream"] is True + assert isinstance( + span.attributes["gen_ai.response.time_to_first_chunk"], float + ) + assert isinstance( + span.attributes["gen_ai.response.time_to_first_token"], int + ) assert span.attributes["gen_ai.request.choice.count"] == 2 assert span.attributes["gen_ai.usage.input_tokens"] == 6 assert span.attributes["gen_ai.usage.output_tokens"] == 5 diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/CHANGELOG.md index 7726d0f3f..6c3f2750f 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Map Mem0 methods to upstream Memory operations and emit store, record, count, + query, and records attributes while preserving the LoongSuite `MEMORY` span + classification. +- Stop emitting legacy Mem0 user, agent, run, limit, and input/output message + attributes and the legacy `gen_ai.memory.operation.details` event on standard + Memory telemetry. + ## Version 0.7.0 (2026-07-03) There are no changelog entries for this release. diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/pyproject.toml b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/pyproject.toml index 39128d0f7..2dd2bb234 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/pyproject.toml +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ "opentelemetry-api ~=1.37", "opentelemetry-instrumentation >=0.58b0", "opentelemetry-semantic-conventions >=0.58b0", - "opentelemetry-util-genai >= 0.2b0", + "opentelemetry-util-genai", ] [project.optional-dependencies] diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/src/opentelemetry/instrumentation/mem0/internal/_wrapper.py b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/src/opentelemetry/instrumentation/mem0/internal/_wrapper.py index 8b237ca8d..1e9fbae69 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/src/opentelemetry/instrumentation/mem0/internal/_wrapper.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/src/opentelemetry/instrumentation/mem0/internal/_wrapper.py @@ -27,11 +27,11 @@ ) from opentelemetry.instrumentation.mem0.internal._extractors import ( GraphOperationAttributeExtractor, - MemoryOperationAttributeExtractor, RerankerAttributeExtractor, VectorOperationAttributeExtractor, ) from opentelemetry.instrumentation.mem0.internal._util import ( + extract_result_count, extract_server_info, get_exception_type, safe_str, @@ -57,11 +57,85 @@ ) from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler from opentelemetry.util.genai.extended_memory import MemoryInvocation +from opentelemetry.util.genai.extended_semconv.gen_ai_memory_attributes import ( + GenAiMemoryOperationNameValues, +) from opentelemetry.util.genai.types import Error logger = logging.getLogger(__name__) +_MEM0_OPERATION_NAMES = { + "add": GenAiMemoryOperationNameValues.UPSERT_MEMORY.value, + "update": GenAiMemoryOperationNameValues.UPDATE_MEMORY.value, + "batch_update": GenAiMemoryOperationNameValues.UPDATE_MEMORY.value, + "search": GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + "get": GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + "get_all": GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + "history": GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + "delete": GenAiMemoryOperationNameValues.DELETE_MEMORY.value, + "batch_delete": GenAiMemoryOperationNameValues.DELETE_MEMORY.value, + "delete_all": GenAiMemoryOperationNameValues.DELETE_MEMORY.value, +} + + +def _standard_memory_operation(operation_name: str) -> str: + return _MEM0_OPERATION_NAMES.get(operation_name, operation_name) + + +def _memory_record(value: Any) -> Optional[dict[str, Any]]: + """Normalize one Mem0 value to the upstream MemoryRecord shape.""" + if isinstance(value, str): + return {"content": value} + if not isinstance(value, dict): + return None + + content = None + for key in ("content", "memory", "text"): + if value.get(key) is not None: + content = value[key] + break + if content is None and value.get("role") is not None: + content = value.get("content") + if content is None: + return None + + record: dict[str, Any] = {"content": content} + record_id = value.get("id", value.get("memory_id")) + if record_id is not None: + record["id"] = safe_str(record_id) + if isinstance(value.get("metadata"), dict): + record["metadata"] = value["metadata"] + if value.get("score") is not None: + try: + record["score"] = float(value["score"]) + except (TypeError, ValueError): + pass + return record + + +def _memory_records(value: Any) -> list[dict[str, Any]]: + """Normalize Mem0 request/response containers to MemoryRecord objects.""" + if value is None: + return [] + if isinstance(value, dict): + for key in ("results", "memories", "data", "items"): + if isinstance(value.get(key), list): + value = value[key] + break + else: + value = [value] + elif not isinstance(value, (list, tuple)): + value = [value] + + records: list[dict[str, Any]] = [] + for item in value: + record = _memory_record(item) + if record is not None: + records.append(record) + return records + + def _get_field(payload: dict, field_name: str) -> Any: """ fetch a field if present (distinguish between missing vs present None). @@ -80,15 +154,6 @@ def _coerce_optional_int(value: Any) -> Optional[int]: return None -def _coerce_optional_float(value: Any) -> Optional[float]: - try: - if value is None: - return None - return float(value) - except Exception: - return None - - def _normalize_call_parameters( func: Callable, args: tuple, @@ -176,7 +241,6 @@ def __init__(self, telemetry_handler: ExtendedTelemetryHandler): telemetry_handler: GenAI ExtendedTelemetryHandler from opentelemetry-util-genai. """ self.telemetry_handler = telemetry_handler - self.extractor = MemoryOperationAttributeExtractor() self._memory_before_hook: MemoryBeforeHook = None self._memory_after_hook: MemoryAfterHook = None @@ -235,53 +299,12 @@ async def async_wrapper(instance: Any, *args, **kwargs): def _set_invocation_fields_from_kwargs( invocation: MemoryInvocation, normalized_kwargs: dict ) -> None: - """Populate MemoryInvocation standard fields from normalized kwargs.""" - - def _int_or_none(value: Any) -> Optional[int]: - try: - if value is None: - return None - return int(value) - except Exception: - return None - - def _float_or_none(value: Any) -> Optional[float]: - try: - if value is None: - return None - return float(value) - except Exception: - return None - - if (user_id := normalized_kwargs.get("user_id")) is not None: - invocation.user_id = safe_str(user_id) - if (agent_id := normalized_kwargs.get("agent_id")) is not None: - invocation.agent_id = safe_str(agent_id) - if (run_id := normalized_kwargs.get("run_id")) is not None: - invocation.run_id = safe_str(run_id) - if (app_id := normalized_kwargs.get("app_id")) is not None: - invocation.app_id = safe_str(app_id) - + """Populate stable upstream Memory identifiers from request data.""" + store_id = normalized_kwargs.get("store_id") + if store_id is not None: + invocation.store_id = safe_str(store_id) if (memory_id := normalized_kwargs.get("memory_id")) is not None: - invocation.memory_id = safe_str(memory_id) - if (limit := normalized_kwargs.get("limit")) is not None: - invocation.limit = _int_or_none(limit) - if (page := normalized_kwargs.get("page")) is not None: - invocation.page = _int_or_none(page) - if (page_size := normalized_kwargs.get("page_size")) is not None: - invocation.page_size = _int_or_none(page_size) - if (top_k := normalized_kwargs.get("top_k")) is not None: - invocation.top_k = _int_or_none(top_k) - if (memory_type := normalized_kwargs.get("memory_type")) is not None: - invocation.memory_type = safe_str(memory_type) - if (threshold := normalized_kwargs.get("threshold")) is not None: - invocation.threshold = _float_or_none(threshold) - if "rerank" in normalized_kwargs: - # rerank can be explicitly False - raw_rerank = normalized_kwargs.get("rerank") - invocation.rerank = ( - bool(raw_rerank) if raw_rerank is not None else None - ) + invocation.record_id = safe_str(memory_id) @staticmethod def _apply_custom_extractor_output_to_invocation( @@ -291,66 +314,30 @@ def _apply_custom_extractor_output_to_invocation( Apply custom extractor output directly onto MemoryInvocation. Expected format (no compatibility guarantees): - - Any MemoryInvocation field name can be provided directly, e.g.: - user_id/agent_id/run_id/app_id/memory_id/limit/page/page_size/top_k/ - memory_type/threshold/rerank/input_messages/output_messages/ - server_address/server_port + - Standard MemoryInvocation fields can be provided directly. - Custom attributes can be provided as: - key "attributes": dict[str, Any] - or any other leftover keys will be treated as custom attributes """ - if "user_id" in extracted: - raw = _get_field(extracted, "user_id") - invocation.user_id = safe_str(raw) if raw is not None else None - if "agent_id" in extracted: - raw = _get_field(extracted, "agent_id") - invocation.agent_id = safe_str(raw) if raw is not None else None - if "run_id" in extracted: - raw = _get_field(extracted, "run_id") - invocation.run_id = safe_str(raw) if raw is not None else None - if "app_id" in extracted: - raw = _get_field(extracted, "app_id") - invocation.app_id = safe_str(raw) if raw is not None else None - if "memory_id" in extracted: - raw = _get_field(extracted, "memory_id") - invocation.memory_id = safe_str(raw) if raw is not None else None - - if "limit" in extracted: - invocation.limit = _coerce_optional_int( - _get_field(extracted, "limit") - ) - if "page" in extracted: - invocation.page = _coerce_optional_int( - _get_field(extracted, "page") - ) - if "page_size" in extracted: - invocation.page_size = _coerce_optional_int( - _get_field(extracted, "page_size") - ) - if "top_k" in extracted: - invocation.top_k = _coerce_optional_int( - _get_field(extracted, "top_k") - ) - - if "memory_type" in extracted: - raw = _get_field(extracted, "memory_type") - invocation.memory_type = safe_str(raw) if raw is not None else None - if "threshold" in extracted: - invocation.threshold = _coerce_optional_float( - _get_field(extracted, "threshold") - ) - if "rerank" in extracted: - raw_rerank = _get_field(extracted, "rerank") - invocation.rerank = ( - bool(raw_rerank) if raw_rerank is not None else None - ) - - if "input_messages" in extracted: - invocation.input_messages = _get_field(extracted, "input_messages") - if "output_messages" in extracted: - invocation.output_messages = _get_field( - extracted, "output_messages" + for field_name in ( + "provider", + "store_id", + "record_id", + "query_text", + ): + if field_name in extracted: + raw = _get_field(extracted, field_name) + setattr( + invocation, + field_name, + safe_str(raw) if raw is not None else None, + ) + if "record_count" in extracted: + invocation.record_count = _coerce_optional_int( + _get_field(extracted, "record_count") ) + if "records" in extracted: + invocation.records = _get_field(extracted, "records") if "server_address" in extracted: raw = _get_field(extracted, "server_address") @@ -369,20 +356,12 @@ def _apply_custom_extractor_output_to_invocation( # Any leftover keys -> invocation.attributes (excluding known fields) known = { - "user_id", - "agent_id", - "run_id", - "app_id", - "memory_id", - "limit", - "page", - "page_size", - "top_k", - "memory_type", - "threshold", - "rerank", - "input_messages", - "output_messages", + "provider", + "store_id", + "record_id", + "record_count", + "query_text", + "records", "server_address", "server_port", "attributes", @@ -415,24 +394,39 @@ def _apply_extracted_attrs_to_invocation( ) return - # Built-in extractor path: directly extract for MemoryInvocation fields - input_msg, output_msg = self.extractor.extract_invocation_content( - operation_name, - normalized_kwargs, - result, - is_memory_client=is_memory_client, - ) - if input_msg is not None: - invocation.input_messages = input_msg - if output_msg is not None: - invocation.output_messages = output_msg - - # Extra attributes only (NOT covered by MemoryInvocation fields) - extra_attrs = self.extractor.extract_invocation_attributes( - operation_name, normalized_kwargs, result - ) - if extra_attrs: - invocation.attributes.update(extra_attrs) + if operation_name == "search": + query = normalized_kwargs.get("query") + if query is not None: + invocation.query_text = safe_str(query) + + candidate_records: Any = result + if candidate_records is None and operation_name in ( + "add", + "update", + "batch_update", + ): + candidate_records = normalized_kwargs.get( + "memories", + normalized_kwargs.get( + "messages", + normalized_kwargs.get( + "data", normalized_kwargs.get("text") + ), + ), + ) + records = _memory_records(candidate_records) + if records: + invocation.records = records + + if result is not None: + invocation.record_count = extract_result_count(result) + if invocation.record_count is None: + for key in ("memories", "ids", "items", "data_list"): + if isinstance(normalized_kwargs.get(key), list): + invocation.record_count = len(normalized_kwargs[key]) + break + if invocation.record_count is None and invocation.record_id: + invocation.record_count = 1 except Exception as e: logger.debug(f"Failed to extract invocation attributes: {e}") @@ -449,7 +443,13 @@ def _execute_with_handler( """Top-level Memory operation execution using util GenAI memory handler (sync).""" normalized_kwargs = _normalize_call_parameters(func, args, kwargs) - invocation = MemoryInvocation(operation=operation_name) + invocation = MemoryInvocation( + operation=_standard_memory_operation(operation_name), + span_kind=( + SpanKind.CLIENT if is_memory_client else SpanKind.INTERNAL + ), + provider="mem0" if is_memory_client else None, + ) self._set_invocation_fields_from_kwargs(invocation, normalized_kwargs) # Server info (MemoryClient/AsyncMemoryClient) @@ -541,7 +541,13 @@ async def _execute_with_handler_async( """Top-level Memory operation execution using util GenAI memory handler (async).""" normalized_kwargs = _normalize_call_parameters(func, args, kwargs) - invocation = MemoryInvocation(operation=operation_name) + invocation = MemoryInvocation( + operation=_standard_memory_operation(operation_name), + span_kind=( + SpanKind.CLIENT if is_memory_client else SpanKind.INTERNAL + ), + provider="mem0" if is_memory_client else None, + ) self._set_invocation_fields_from_kwargs(invocation, normalized_kwargs) # Server info (MemoryClient/AsyncMemoryClient) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_hooks.py b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_hooks.py index 68ef43928..9258253fa 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_hooks.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_hooks.py @@ -38,11 +38,16 @@ ) from opentelemetry.instrumentation.mem0.types import set_memory_hooks from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.trace import SpanKind from opentelemetry.util.genai.extended_memory import MemoryInvocation class _DummyTelemetryHandler: + def __init__(self): + self.started = [] + def start_memory(self, invocation: Any) -> None: + self.started.append(invocation) return None def stop_memory(self, invocation: Any) -> None: @@ -52,6 +57,28 @@ def fail_memory(self, invocation: Any, error: Any) -> None: return None +@pytest.mark.parametrize( + ("mem0_operation", "upstream_operation"), + [ + ("add", "upsert_memory"), + ("update", "update_memory"), + ("batch_update", "update_memory"), + ("search", "search_memory"), + ("get", "search_memory"), + ("get_all", "search_memory"), + ("history", "search_memory"), + ("delete", "delete_memory"), + ("batch_delete", "delete_memory"), + ("delete_all", "delete_memory"), + ], +) +def test_memory_operation_mapping(mem0_operation, upstream_operation): + assert ( + wrapper_mod._standard_memory_operation(mem0_operation) + == upstream_operation + ) + + @pytest.mark.asyncio async def test_memory_hooks_sync_async_and_exception_paths(): calls: List[Tuple[str, Dict[str, Any]]] = [] @@ -70,7 +97,8 @@ def after( seen_exc["exc"] = exc calls.append(("after", hook_context)) - w = MemoryOperationWrapper(_DummyTelemetryHandler()) + dummy_handler = _DummyTelemetryHandler() + w = MemoryOperationWrapper(dummy_handler) set_memory_hooks(w, memory_before_hook=before, memory_after_hook=after) # Cover helper: _normalize_call_parameters with positional args mapping @@ -89,20 +117,24 @@ def _sig(self, memory_id, data, *, user_id=None): # noqa: ARG001 MemoryOperationWrapper._apply_custom_extractor_output_to_invocation( inv, { - "user_id": "u1", - "limit": "3", - "threshold": "0.7", - "rerank": False, + "provider": "mem0", + "store_id": "store-1", + "record_id": "memory-1", + "record_count": "3", + "query_text": "preferences", + "records": [{"content": "likes apples"}], "server_address": "example.com", "server_port": "443", "attributes": {"k1": "v1"}, "custom_k": "custom_v", }, ) - assert inv.user_id == "u1" - assert inv.limit == 3 - assert inv.threshold == 0.7 - assert inv.rerank is False + assert inv.provider == "mem0" + assert inv.store_id == "store-1" + assert inv.record_id == "memory-1" + assert inv.record_count == 3 + assert inv.query_text == "preferences" + assert inv.records == [{"content": "likes apples"}] assert inv.server_address == "example.com" assert inv.server_port == 443 assert inv.attributes.get("k1") == "v1" @@ -122,6 +154,8 @@ def _fn_sync(*a, **k): is_memory_client=False, ) assert res_sync == "ok" + assert dummy_handler.started[-1].operation == "upsert_memory" + assert dummy_handler.started[-1].span_kind is SpanKind.INTERNAL # cover extract_server_info exception branch (host property raises) class _BadHost: @@ -138,6 +172,9 @@ def host(self): # pragma: no cover extract_attributes_func=None, is_memory_client=True, ) + assert dummy_handler.started[-1].operation == "search_memory" + assert dummy_handler.started[-1].span_kind is SpanKind.CLIENT + assert dummy_handler.started[-1].provider == "mem0" # cover custom extractor function path (extract_attributes_func provided) inv2 = MemoryInvocation(operation="add") @@ -148,12 +185,12 @@ def host(self): # pragma: no cover operation_name="add", result={"results": []}, extract_attributes_func=lambda kwargs, result: { - "user_id": "u2", + "record_id": "memory-2", "attributes": {"x": 1}, }, is_memory_client=False, ) - assert inv2.user_id == "u2" + assert inv2.record_id == "memory-2" assert inv2.attributes.get("x") == 1 # async success @@ -170,6 +207,7 @@ async def _fn_async(*a, **k): is_memory_client=False, ) assert res_async == "ok2" + assert dummy_handler.started[-1].operation == "search_memory" # hook exceptions swallowed (both before/after) def before_boom(span, operation, instance, args, kwargs, hook_context): diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_client_vcr.py b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_client_vcr.py index 78fbf59b0..5041ab439 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_client_vcr.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_client_vcr.py @@ -40,6 +40,15 @@ E2E_RUN_ID = "test_run_client_full" +def _memory_spans(spans, operation_name): + return [ + span + for span in spans + if span.attributes.get("gen_ai.operation.name") == operation_name + and span.attributes.get("gen_ai.span.kind") == "MEMORY" + ] + + def _new_client(): pytest.importorskip("mem0") MemoryClient = pytest.importorskip("mem0.client.main").MemoryClient # type: ignore @@ -100,31 +109,19 @@ def test_client_add_vcr(span_exporter, instrument_with_content): # Verify add operation span attributes spans = span_exporter.get_finished_spans() - add_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "add" - ] + add_spans = _memory_spans(spans, "upsert_memory") assert add_spans, "Should have add operation span" add_span = add_spans[0] - # Basic attributes - assert ( - add_span.attributes.get("gen_ai.operation.name") == "memory_operation" - ) - assert add_span.attributes.get("gen_ai.memory.operation") == "add" - assert add_span.attributes.get("gen_ai.memory.user_id") == E2E_USER_ID - assert add_span.attributes.get("gen_ai.memory.agent_id") == E2E_AGENT_ID + assert add_span.name == "upsert_memory" + assert isinstance(add_span.attributes["gen_ai.memory.record.count"], int) # Server attributes assert "server.address" in add_span.attributes assert "server.port" in add_span.attributes - # Input messages - if "gen_ai.memory.input.messages" in add_span.attributes: - assert isinstance( - add_span.attributes["gen_ai.memory.input.messages"], str - ) + if "gen_ai.memory.records" in add_span.attributes: + assert isinstance(add_span.attributes["gen_ai.memory.records"], str) @pytest.mark.vcr(cassette_name="test_client_get_all_vcr") @@ -141,26 +138,16 @@ def test_client_get_all_vcr(span_exporter, instrument_with_content): # Verify get_all operation span attributes spans = span_exporter.get_finished_spans() - getall_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "get_all" - ] + getall_spans = _memory_spans(spans, "search_memory") assert getall_spans, "Should have get_all operation span" getall_span = getall_spans[0] - assert ( - getall_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + assert isinstance( + getall_span.attributes["gen_ai.memory.record.count"], int ) - assert getall_span.attributes.get("gen_ai.memory.operation") == "get_all" assert "server.address" in getall_span.attributes assert "server.port" in getall_span.attributes - # top_k parameter - if "gen_ai.memory.top_k" in getall_span.attributes: - assert getall_span.attributes["gen_ai.memory.top_k"] == top_k - @pytest.mark.vcr(cassette_name="test_client_get_vcr") def test_client_get_vcr(span_exporter, instrument_with_content): @@ -174,20 +161,13 @@ def test_client_get_vcr(span_exporter, instrument_with_content): # Verify get operation span attributes spans = span_exporter.get_finished_spans() - get_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "get" - ] + get_spans = _memory_spans(spans, "search_memory") assert get_spans, "Should have get operation span" get_span = get_spans[0] assert ( - get_span.attributes.get("gen_ai.operation.name") == "memory_operation" + get_span.attributes["gen_ai.memory.record.id"] == HARDCODED_MEMORY_ID_1 ) - assert get_span.attributes.get("gen_ai.memory.operation") == "get" - assert "gen_ai.memory.id" in get_span.attributes - assert get_span.attributes["gen_ai.memory.id"] == HARDCODED_MEMORY_ID_1 assert "server.address" in get_span.attributes @@ -205,25 +185,14 @@ def test_client_search_vcr(span_exporter, instrument_with_content): # Verify search operation span attributes spans = span_exporter.get_finished_spans() - search_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "search" - ] + search_spans = _memory_spans(spans, "search_memory") assert search_spans, "Should have search operation span" search_span = search_spans[0] - assert ( - search_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + assert search_span.attributes["gen_ai.memory.query.text"] == query + assert isinstance( + search_span.attributes["gen_ai.memory.record.count"], int ) - assert search_span.attributes.get("gen_ai.memory.operation") == "search" - - # Query-related attributes - if "gen_ai.memory.input.messages" in search_span.attributes: - assert query in str( - search_span.attributes["gen_ai.memory.input.messages"] - ) assert "server.address" in search_span.attributes @@ -241,28 +210,18 @@ def test_client_update_vcr(span_exporter, instrument_with_content): # Verify update operation span attributes spans = span_exporter.get_finished_spans() - update_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "update" - ] + update_spans = _memory_spans(spans, "update_memory") assert update_spans, "Should have update operation span" update_span = update_spans[0] assert ( - update_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + update_span.attributes["gen_ai.memory.record.id"] + == HARDCODED_MEMORY_ID_1 ) - assert update_span.attributes.get("gen_ai.memory.operation") == "update" - assert "gen_ai.memory.id" in update_span.attributes - assert update_span.attributes["gen_ai.memory.id"] == HARDCODED_MEMORY_ID_1 assert "server.address" in update_span.attributes - # Input messages (new content) - if "gen_ai.memory.input.messages" in update_span.attributes: - assert new_text in str( - update_span.attributes["gen_ai.memory.input.messages"] - ) + if "gen_ai.memory.records" in update_span.attributes: + assert new_text in update_span.attributes["gen_ai.memory.records"] @pytest.mark.vcr(cassette_name="test_client_delete_vcr") @@ -290,21 +249,11 @@ def test_client_delete_vcr(span_exporter, instrument_with_content): # Verify delete operation span attributes spans = span_exporter.get_finished_spans() - delete_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "delete" - ] + delete_spans = _memory_spans(spans, "delete_memory") assert delete_spans, "Should have delete operation span" delete_span = delete_spans[0] - assert ( - delete_span.attributes.get("gen_ai.operation.name") - == "memory_operation" - ) - assert delete_span.attributes.get("gen_ai.memory.operation") == "delete" - assert "gen_ai.memory.id" in delete_span.attributes - assert delete_span.attributes["gen_ai.memory.id"] == mem_id + assert delete_span.attributes["gen_ai.memory.record.id"] == mem_id assert "server.address" in delete_span.attributes @@ -320,21 +269,14 @@ def test_client_history_vcr(span_exporter, instrument_with_content): # Verify history operation span attributes spans = span_exporter.get_finished_spans() - history_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "history" - ] + history_spans = _memory_spans(spans, "search_memory") assert history_spans, "Should have history operation span" history_span = history_spans[0] assert ( - history_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + history_span.attributes["gen_ai.memory.record.id"] + == HARDCODED_MEMORY_ID_1 ) - assert history_span.attributes.get("gen_ai.memory.operation") == "history" - assert "gen_ai.memory.id" in history_span.attributes - assert history_span.attributes["gen_ai.memory.id"] == HARDCODED_MEMORY_ID_1 assert "server.address" in history_span.attributes diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_record_vcr.py b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_record_vcr.py index 96f7fdd88..4c1509d03 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_record_vcr.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_record_vcr.py @@ -24,6 +24,15 @@ import pytest +def _memory_spans(spans, operation_name): + return [ + span + for span in spans + if span.attributes.get("gen_ai.operation.name") == operation_name + and span.attributes.get("gen_ai.span.kind") == "MEMORY" + ] + + def _build_demo_config_from_env() -> Dict[str, Any]: # Allow overriding model and dimensions via environment variables llm_model = os.environ.get("OPENAI_LLM_MODEL", "qwen-plus") @@ -191,22 +200,17 @@ def test_record_memory_full_flow_vcr( # Verify OTel spans at least contain each main operation spans: List[Any] = span_exporter.get_finished_spans() - ops_needed: set[str] = { - "add", - "search", - "get", - "get_all", - "update", - "delete", - "delete_all", - "history", + ops_needed = { + "upsert_memory", + "search_memory", + "update_memory", + "delete_memory", + } + ops_seen = { + span.attributes["gen_ai.operation.name"] + for span in spans + if span.attributes.get("gen_ai.span.kind") == "MEMORY" } - ops_seen: set[str] = set() - for s in spans: - op = s.attributes.get("gen_ai.memory.operation") - if isinstance(op, str): - if op in ops_needed: - ops_seen.add(op) missing = ops_needed - ops_seen assert not missing, f"should contain span attributes: {missing}" @@ -256,37 +260,13 @@ def test_record_memory_add_vcr( # Verify add operation span attributes spans = span_exporter.get_finished_spans() - add_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "add" - ] + add_spans = _memory_spans(spans, "upsert_memory") assert add_spans, "should have add operation span" add_span = add_spans[0] - # Required attributes - assert ( - add_span.attributes.get("gen_ai.operation.name") == "memory_operation" - ) - assert add_span.attributes.get("gen_ai.memory.operation") == "add" - assert add_span.attributes.get("gen_ai.memory.user_id") == user_id - assert add_span.attributes.get("gen_ai.memory.agent_id") == agent_id - assert add_span.attributes.get("gen_ai.memory.run_id") == run_id - - # Operation specific attributes - assert "gen_ai.memory.infer" in add_span.attributes - assert add_span.attributes["gen_ai.memory.infer"] is False - assert "gen_ai.memory.metadata" in add_span.attributes # metadata keys - assert "gen_ai.memory.input.messages" in add_span.attributes - - # Result attributes - if "gen_ai.memory.result_count" in add_span.attributes: - assert add_span.attributes["gen_ai.memory.result_count"] > 0 - if "gen_ai.memory.output.messages" in add_span.attributes: - # Output message should be JSON string - assert isinstance( - add_span.attributes["gen_ai.memory.output.messages"], str - ) + assert add_span.name == "upsert_memory" + assert isinstance(add_span.attributes["gen_ai.memory.record.count"], int) + assert isinstance(add_span.attributes["gen_ai.memory.records"], str) @pytest.mark.vcr() @@ -332,35 +312,14 @@ def test_record_memory_get_all_vcr( # Verify get_all operation span attributes spans = span_exporter.get_finished_spans() - getall_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "get_all" - ] + getall_spans = _memory_spans(spans, "search_memory") assert getall_spans, "should have get_all operation span" getall_span = getall_spans[0] - # Required attributes - assert ( - getall_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + assert isinstance( + getall_span.attributes["gen_ai.memory.record.count"], int ) - assert getall_span.attributes.get("gen_ai.memory.operation") == "get_all" - assert getall_span.attributes.get("gen_ai.memory.user_id") == user_id - assert getall_span.attributes.get("gen_ai.memory.agent_id") == agent_id - assert getall_span.attributes.get("gen_ai.memory.run_id") == run_id - - # Operation specific attributes - assert "gen_ai.memory.limit" in getall_span.attributes - assert getall_span.attributes["gen_ai.memory.limit"] == limit - - # Result attributes - assert "gen_ai.memory.result_count" in getall_span.attributes - if "gen_ai.memory.output.messages" in getall_span.attributes: - # Output message should be JSON string - assert isinstance( - getall_span.attributes["gen_ai.memory.output.messages"], str - ) + assert isinstance(getall_span.attributes["gen_ai.memory.records"], str) @pytest.mark.vcr() @@ -413,25 +372,13 @@ def test_record_memory_get_vcr( # Verify get operation span attributes spans = span_exporter.get_finished_spans() - get_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "get" - ] + get_spans = _memory_spans(spans, "search_memory") assert get_spans, "should have get operation span" get_span = get_spans[0] - # Required attributes - assert ( - get_span.attributes.get("gen_ai.operation.name") == "memory_operation" - ) - assert get_span.attributes.get("gen_ai.memory.operation") == "get" - assert "gen_ai.memory.id" in get_span.attributes - assert get_span.attributes["gen_ai.memory.id"] == mem_id - - # Output attributes (if has results) - if "gen_ai.memory.output.messages" in get_span.attributes: - output = get_span.attributes["gen_ai.memory.output.messages"] + assert get_span.attributes["gen_ai.memory.record.id"] == mem_id + if "gen_ai.memory.records" in get_span.attributes: + output = get_span.attributes["gen_ai.memory.records"] assert isinstance(output, str) and len(output) > 0 @@ -478,35 +425,16 @@ def test_record_memory_search_vcr( # Verify search operation span attributes spans = span_exporter.get_finished_spans() - search_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "search" - ] + search_spans = _memory_spans(spans, "search_memory") assert search_spans, "shouldhas search operation span" search_span = search_spans[0] - # Required attributes - assert ( - search_span.attributes.get("gen_ai.operation.name") - == "memory_operation" + assert search_span.attributes["gen_ai.memory.query.text"] == query + assert isinstance( + search_span.attributes["gen_ai.memory.record.count"], int ) - assert search_span.attributes.get("gen_ai.memory.operation") == "search" - assert search_span.attributes.get("gen_ai.memory.user_id") == user_id - - # Operation specific attributes - assert "gen_ai.memory.limit" in search_span.attributes - assert search_span.attributes["gen_ai.memory.limit"] == limit - assert "gen_ai.memory.input.messages" in search_span.attributes - # query should be in input.messages - assert query in str(search_span.attributes["gen_ai.memory.input.messages"]) - - # Result attributes - assert "gen_ai.memory.result_count" in search_span.attributes - if "gen_ai.memory.output.messages" in search_span.attributes: - assert isinstance( - search_span.attributes["gen_ai.memory.output.messages"], str - ) + if "gen_ai.memory.records" in search_span.attributes: + assert isinstance(search_span.attributes["gen_ai.memory.records"], str) @pytest.mark.vcr() @@ -562,28 +490,12 @@ def test_record_memory_update_vcr( # Verify update operation span attributes spans = span_exporter.get_finished_spans() - update_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "update" - ] + update_spans = _memory_spans(spans, "update_memory") assert update_spans, "shouldhas update operation span" update_span = update_spans[0] - # Required attributes - assert ( - update_span.attributes.get("gen_ai.operation.name") - == "memory_operation" - ) - assert update_span.attributes.get("gen_ai.memory.operation") == "update" - assert "gen_ai.memory.id" in update_span.attributes - assert update_span.attributes["gen_ai.memory.id"] == mem_id - - # Input attributes (new content) - assert "gen_ai.memory.input.messages" in update_span.attributes - assert new_content in str( - update_span.attributes["gen_ai.memory.input.messages"] - ) + assert update_span.attributes["gen_ai.memory.record.id"] == mem_id + assert new_content in update_span.attributes["gen_ai.memory.records"] @pytest.mark.vcr() @@ -638,22 +550,11 @@ def test_record_memory_delete_vcr( # Verify delete operation span attributes spans = span_exporter.get_finished_spans() - delete_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "delete" - ] + delete_spans = _memory_spans(spans, "delete_memory") assert delete_spans, "shouldhas delete operation span" delete_span = delete_spans[0] - # Required attributes - assert ( - delete_span.attributes.get("gen_ai.operation.name") - == "memory_operation" - ) - assert delete_span.attributes.get("gen_ai.memory.operation") == "delete" - assert "gen_ai.memory.id" in delete_span.attributes - assert delete_span.attributes["gen_ai.memory.id"] == mem_id + assert delete_span.attributes["gen_ai.memory.record.id"] == mem_id @pytest.mark.vcr() @@ -711,26 +612,12 @@ def test_record_memory_history_vcr( # Verify history operation span attributes spans = span_exporter.get_finished_spans() - history_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "history" - ] + history_spans = _memory_spans(spans, "search_memory") assert history_spans, "should have history operation span" history_span = history_spans[0] - # Required attributes - assert ( - history_span.attributes.get("gen_ai.operation.name") - == "memory_operation" - ) - assert history_span.attributes.get("gen_ai.memory.operation") == "history" - assert "gen_ai.memory.id" in history_span.attributes - assert history_span.attributes["gen_ai.memory.id"] == mem_id - - # Result attributes (history count) - if "gen_ai.memory.result_count" in history_span.attributes: - assert history_span.attributes["gen_ai.memory.result_count"] >= 0 + assert history_span.attributes["gen_ai.memory.record.id"] == mem_id + assert history_span.attributes["gen_ai.memory.record.count"] >= 0 @pytest.mark.vcr() @@ -779,23 +666,7 @@ def test_record_memory_delete_all_vcr( # Verify delete_all operation span attributes spans = span_exporter.get_finished_spans() - deleteall_spans = [ - s - for s in spans - if s.attributes.get("gen_ai.memory.operation") == "delete_all" - ] + deleteall_spans = _memory_spans(spans, "delete_memory") assert deleteall_spans, "shouldhas delete_all operation span" - deleteall_span = deleteall_spans[0] - # Required attributes - assert ( - deleteall_span.attributes.get("gen_ai.operation.name") - == "memory_operation" - ) - assert ( - deleteall_span.attributes.get("gen_ai.memory.operation") - == "delete_all" - ) - assert deleteall_span.attributes.get("gen_ai.memory.user_id") == user_id - assert deleteall_span.attributes.get("gen_ai.memory.agent_id") == agent_id - assert deleteall_span.attributes.get("gen_ai.memory.run_id") == run_id + assert deleteall_spans[0].name == "delete_memory" diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_vcr.py b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_vcr.py index e8d96fe42..883d887fa 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_vcr.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-mem0/tests/test_memory_operations_vcr.py @@ -23,10 +23,6 @@ import pytest -from opentelemetry.semconv._incubating.attributes import ( - gen_ai_attributes as GenAIAttributes, -) - _mem0 = pytest.importorskip("mem0") _factory = pytest.importorskip("mem0.utils.factory") @@ -228,25 +224,16 @@ def test_memory_add_full_flow( assert m.add(test_message, user_id="u_123") is not None spans = span_exporter.get_finished_spans() - # verify top-level operation type via attributes, not relying on span.name + # Verify the upstream Memory operation and LoongSuite classification. add_span = next( s for s in spans - if s.attributes.get("gen_ai.memory.operation") == "add" + if s.attributes.get("gen_ai.operation.name") == "upsert_memory" + and s.attributes.get("gen_ai.span.kind") == "MEMORY" ) - assert GenAIAttributes.GEN_AI_OPERATION_NAME in add_span.attributes - - # Verify input messages collection (messages passed as positional parameter) - assert "gen_ai.memory.input.messages" in add_span.attributes - assert test_message in add_span.attributes["gen_ai.memory.input.messages"] - - # Verify output messages collection (if has results) - if ( - "gen_ai.memory.result.count" in add_span.attributes - and add_span.attributes["gen_ai.memory.result.count"] > 0 - ): - # If has results, should have output messages - assert "gen_ai.memory.output.messages" in add_span.attributes + assert add_span.name == "upsert_memory" + assert isinstance(add_span.attributes["gen_ai.memory.record.count"], int) + assert test_message in add_span.attributes["gen_ai.memory.records"] @pytest.mark.vcr() @@ -272,23 +259,17 @@ def test_memory_search_full_flow( assert m.search(test_query, user_id="u_123", limit=2) is not None spans = span_exporter.get_finished_spans() - # verify top-level operation type via attributes + # Verify the upstream Memory search operation. search_span = next( s for s in spans - if s.attributes.get("gen_ai.memory.operation") == "search" + if s.attributes.get("gen_ai.operation.name") == "search_memory" + and s.attributes.get("gen_ai.span.kind") == "MEMORY" + ) + assert search_span.attributes["gen_ai.memory.query.text"] == test_query + assert isinstance( + search_span.attributes["gen_ai.memory.record.count"], int ) - - # Verify input messages collection (query passed as positional parameter) - assert "gen_ai.memory.input.messages" in search_span.attributes - assert test_query in search_span.attributes["gen_ai.memory.input.messages"] - - # Verify output messages collection (if has results) - # FakeVectorStore search will return results, so should have output messages - if "gen_ai.memory.result.count" in search_span.attributes: - # As long as executed, regardless of whether results are empty, may have output messages - # here not mandatory, because results may be empty - pass @pytest.mark.vcr() diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/CHANGELOG.md index 7c8928699..eac13fa67 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Capture stream request state, response IDs, and independent + time-to-first-chunk and time-to-first-token timestamps. + ## Version 0.7.0 (2026-07-03) ### Fixed diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/patch.py b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/patch.py index c395cd3f5..256e1c4a3 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/patch.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/patch.py @@ -290,11 +290,15 @@ def _wrap_streaming_llm_response( """Wrap a streaming LLM response iterator to capture output on completion.""" try: last_response = None - first_token = True for response in response_iter: - if first_token: - invocation.monotonic_first_token_s = timeit.default_timer() - first_token = False + now = timeit.default_timer() + if invocation.monotonic_first_chunk_s is None: + invocation.monotonic_first_chunk_s = now + if ( + invocation.monotonic_first_token_s is None + and _qwen_response_has_token(response) + ): + invocation.monotonic_first_token_s = now _apply_usage_to_llm_invocation(invocation, response) last_response = response yield response @@ -332,6 +336,34 @@ def _wrap_streaming_llm_response( raise +def _qwen_response_has_token(response: Any) -> bool: + messages = response if isinstance(response, list) else [response] + for message in messages: + content = ( + message.get("content") + if isinstance(message, dict) + else getattr(message, "content", None) + ) + if isinstance(content, str) and content: + return True + if isinstance(content, list): + for part in content: + if isinstance(part, str) and part: + return True + if isinstance(part, dict) and ( + part.get("text") or part.get("function_call") + ): + return True + function_call = ( + message.get("function_call") + if isinstance(message, dict) + else getattr(message, "function_call", None) + ) + if function_call: + return True + return False + + def wrap_agent_call_llm( wrapped, instance, args, kwargs, handler: ExtendedTelemetryHandler ): diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/utils.py b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/utils.py index 3d10eab96..c2fdd24ae 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/utils.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-qwen-agent/src/opentelemetry/instrumentation/qwen_agent/utils.py @@ -564,6 +564,7 @@ def _create_llm_invocation( request_model=request_model, provider=provider_name, input_messages=input_messages, + stream=True if stream else None, ) # Set generation parameters diff --git a/tox-loongsuite.ini b/tox-loongsuite.ini index 6482b0196..b6d8bca4d 100644 --- a/tox-loongsuite.ini +++ b/tox-loongsuite.ini @@ -222,7 +222,7 @@ deps = util-genai: {[testenv]test_deps} util-genai: -r {toxinidir}/util/opentelemetry-util-genai/test-requirements.txt - util-genai: {toxinidir}/util/opentelemetry-util-genai + util-genai: -e {toxinidir}/util/opentelemetry-util-genai detect-loongsuite-changes: pytest detect-loongsuite-changes: tomli diff --git a/util/opentelemetry-util-genai/CHANGELOG.md b/util/opentelemetry-util-genai/CHANGELOG.md index 44716b83c..b39cdd928 100644 --- a/util/opentelemetry-util-genai/CHANGELOG.md +++ b/util/opentelemetry-util-genai/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Align LLM, Retrieval, Agent, and Memory span attributes with the + current upstream OpenTelemetry GenAI semantic conventions while retaining the + existing invocation lifecycle. +- Record `gen_ai.response.time_to_first_chunk` as double seconds independently + from the LoongSuite `gen_ai.response.time_to_first_token` integer nanoseconds. +- Change Retrieval spans to client spans and emit `gen_ai.retrieval.top_k`; + omit the opt-in `gen_ai.retrieval.documents` attribute when content capture + is disabled instead of retaining document ids and scores. + Memory spans now use upstream operation names and Memory attributes instead + of legacy LoongSuite `gen_ai.memory.*` fields and the legacy Memory operation + event. - Avoid import-time warnings when optional audio dependencies for PCM16-to-WAV conversion are not installed. ## Version 0.3b0 (2026-02-20) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_handler.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_handler.py index 3e6cfee7e..4c38658e1 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_handler.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_handler.py @@ -87,7 +87,6 @@ from opentelemetry.util.genai.extended_memory import ( MemoryInvocation, _apply_memory_finish_attributes, - _maybe_emit_memory_event, ) from opentelemetry.util.genai.extended_metrics import ( ExtendedInvocationMetricsRecorder, @@ -576,7 +575,7 @@ def start_retrieval( """Start a retrieval invocation and create a pending span entry.""" span = self._tracer.start_span( name="retrieval", - kind=SpanKind.INTERNAL, + kind=SpanKind.CLIENT, context=context, ) # Record a monotonic start timestamp (seconds) for duration @@ -716,12 +715,9 @@ def start_memory( context: Context | None = None, ) -> MemoryInvocation: """Start a memory operation invocation and create a pending span entry.""" - span_name = f"memory_operation {invocation.operation}" - - # Memory operations are CLIENT operations to remote services span = self._tracer.start_span( - name=span_name, - kind=SpanKind.CLIENT, + name=invocation.operation, + kind=invocation.span_kind, context=context, ) # Record a monotonic start timestamp (seconds) for duration @@ -741,7 +737,6 @@ def stop_memory(self, invocation: MemoryInvocation) -> MemoryInvocation: # pyli return invocation _apply_memory_finish_attributes(invocation.span, invocation) - _maybe_emit_memory_event(self._logger, invocation.span, invocation) self._record_extended_metrics(invocation.span, invocation) _safe_detach(invocation.context_token) @@ -758,7 +753,6 @@ def fail_memory( # pylint: disable=no-self-use span = invocation.span _apply_memory_finish_attributes(span, invocation) _apply_error_attributes(span, error) - _maybe_emit_memory_event(self._logger, span, invocation, error) # pylint: disable=too-many-function-args _safe_detach(invocation.context_token) span.end() return invocation diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/__init__.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/__init__.py index 26fdb82c9..5500baf0c 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/__init__.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/__init__.py @@ -22,11 +22,9 @@ from opentelemetry.util.genai.extended_memory.memory_utils import ( MemoryInvocation, _apply_memory_finish_attributes, - _maybe_emit_memory_event, ) __all__ = [ "MemoryInvocation", "_apply_memory_finish_attributes", - "_maybe_emit_memory_event", ] diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_types.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_types.py index 4ab797afd..00c357249 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_types.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_types.py @@ -17,7 +17,8 @@ from dataclasses import dataclass, field from typing import Any, Dict -from opentelemetry.util.genai.types import ContextToken, Span +from opentelemetry.trace import Span, SpanKind +from opentelemetry.util.genai.types import ContextToken def _new_str_any_dict() -> Dict[str, Any]: @@ -33,11 +34,24 @@ class MemoryInvocation: The span and context_token attributes are set by the TelemetryHandler. """ - operation: str # Memory operation type (add, search, update, etc.) + operation: str + """Upstream operation name, for example ``search_memory``.""" context_token: ContextToken | None = None span: Span | None = None attributes: Dict[str, Any] = field(default_factory=_new_str_any_dict) - # Memory identifiers (conditionally required) + provider: str | None = None + store_id: str | None = None + record_id: str | None = None + record_count: int | None = None + query_text: str | None = None + records: Any = None + span_kind: SpanKind = SpanKind.CLIENT + server_address: str | None = None + server_port: int | None = None + monotonic_start_s: float | None = None + + # Legacy input fields retained for source compatibility. They are not + # emitted as attributes by the standard Memory span implementation. user_id: str | None = None agent_id: str | None = None run_id: str | None = None @@ -54,7 +68,3 @@ class MemoryInvocation: # Memory content (optional, controlled by content capturing mode) input_messages: Any = None # Original memory content output_messages: Any = None # Query results - # Server information - server_address: str | None = None - server_port: int | None = None - monotonic_start_s: float | None = None diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_utils.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_utils.py index 91424dcd7..39a64a203 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_utils.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_memory/memory_utils.py @@ -21,19 +21,13 @@ from typing import Any -from opentelemetry._logs import Logger, LogRecord -from opentelemetry.context import get_current from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) -from opentelemetry.semconv.attributes import ( - error_attributes as ErrorAttributes, -) from opentelemetry.semconv.attributes import ( server_attributes as ServerAttributes, ) from opentelemetry.trace import Span -from opentelemetry.trace.propagation import set_span_in_context from opentelemetry.util.genai.extended_memory.memory_types import ( MemoryInvocation, ) @@ -42,147 +36,69 @@ GenAiSpanKindValues, ) from opentelemetry.util.genai.extended_semconv.gen_ai_memory_attributes import ( - GEN_AI_MEMORY_AGENT_ID, - GEN_AI_MEMORY_APP_ID, - GEN_AI_MEMORY_ID, - GEN_AI_MEMORY_INPUT_MESSAGES, - GEN_AI_MEMORY_LIMIT, - GEN_AI_MEMORY_MEMORY_TYPE, - GEN_AI_MEMORY_OPERATION, - GEN_AI_MEMORY_OUTPUT_MESSAGES, - GEN_AI_MEMORY_PAGE, - GEN_AI_MEMORY_PAGE_SIZE, - GEN_AI_MEMORY_RERANK, - GEN_AI_MEMORY_RUN_ID, - GEN_AI_MEMORY_THRESHOLD, - GEN_AI_MEMORY_TOP_K, - GEN_AI_MEMORY_USER_ID, + GEN_AI_MEMORY_QUERY_TEXT, + GEN_AI_MEMORY_RECORD_COUNT, + GEN_AI_MEMORY_RECORD_ID, + GEN_AI_MEMORY_RECORDS, + GEN_AI_MEMORY_STORE_ID, ) -from opentelemetry.util.genai.types import Error from opentelemetry.util.genai.utils import ( ContentCapturingMode, gen_ai_json_dumps, get_content_capturing_mode, is_experimental_mode, - should_emit_event, ) def _get_memory_common_attributes( invocation: MemoryInvocation, ) -> dict[str, Any]: - """Get common memory attributes (operation_name, operation type, identifiers). - - Returns a dictionary of attributes. - """ - attributes: dict[str, Any] = {} - - # Operation name - attributes[GenAI.GEN_AI_OPERATION_NAME] = "memory_operation" - - # Required: memory operation type - attributes[GEN_AI_MEMORY_OPERATION] = invocation.operation - - # Conditionally required identifiers - if invocation.user_id is not None: - attributes[GEN_AI_MEMORY_USER_ID] = invocation.user_id - if invocation.agent_id is not None: - attributes[GEN_AI_MEMORY_AGENT_ID] = invocation.agent_id - if invocation.run_id is not None: - attributes[GEN_AI_MEMORY_RUN_ID] = invocation.run_id - if invocation.app_id is not None: - attributes[GEN_AI_MEMORY_APP_ID] = invocation.app_id - - return attributes - - -def _get_memory_parameter_attributes( - invocation: MemoryInvocation, -) -> dict[str, Any]: - """Get memory operation parameter attributes. + """Get common attributes defined by the upstream Memory span convention. Returns a dictionary of attributes. """ attributes: dict[str, Any] = {} - # Optional parameters - if invocation.memory_id is not None: - attributes[GEN_AI_MEMORY_ID] = invocation.memory_id - if invocation.limit is not None: - attributes[GEN_AI_MEMORY_LIMIT] = invocation.limit - if invocation.page is not None: - attributes[GEN_AI_MEMORY_PAGE] = invocation.page - if invocation.page_size is not None: - attributes[GEN_AI_MEMORY_PAGE_SIZE] = invocation.page_size - if invocation.top_k is not None: - attributes[GEN_AI_MEMORY_TOP_K] = invocation.top_k - if invocation.memory_type is not None: - attributes[GEN_AI_MEMORY_MEMORY_TYPE] = invocation.memory_type - if invocation.threshold is not None: - attributes[GEN_AI_MEMORY_THRESHOLD] = invocation.threshold - if invocation.rerank is not None: - attributes[GEN_AI_MEMORY_RERANK] = invocation.rerank + attributes[GenAI.GEN_AI_OPERATION_NAME] = invocation.operation + if invocation.provider is not None: + attributes[GenAI.GEN_AI_PROVIDER_NAME] = invocation.provider + if invocation.store_id is not None: + attributes[GEN_AI_MEMORY_STORE_ID] = invocation.store_id + if invocation.record_id is not None: + attributes[GEN_AI_MEMORY_RECORD_ID] = invocation.record_id + if invocation.record_count is not None: + attributes[GEN_AI_MEMORY_RECORD_COUNT] = invocation.record_count return attributes def _get_memory_content_attributes( invocation: MemoryInvocation, - for_span: bool = True, ) -> dict[str, Any]: """ Get memory content attributes (input/output messages). This is a controlled operation that only records content when: - Experimental mode is enabled - - For spans: content capturing mode is SPAN_ONLY or SPAN_AND_EVENT - - For events: content capturing mode is EVENT_ONLY or SPAN_AND_EVENT - - Args: - invocation: The memory invocation - for_span: If True, check for span content capturing mode; if False, check for event mode + - Content capturing mode is SPAN_ONLY or SPAN_AND_EVENT Returns empty dict if not in experimental mode or content capturing is disabled. """ attributes: dict[str, Any] = {} - # Check experimental mode and content capturing mode based on usage - if not is_experimental_mode(): + if not is_experimental_mode() or get_content_capturing_mode() not in ( + ContentCapturingMode.SPAN_ONLY, + ContentCapturingMode.SPAN_AND_EVENT, + ): return attributes - if for_span: - # For spans: only capture if SPAN_ONLY or SPAN_AND_EVENT - if get_content_capturing_mode() not in ( - ContentCapturingMode.SPAN_ONLY, - ContentCapturingMode.SPAN_AND_EVENT, - ): - return attributes - else: - # For events: only capture if EVENT_ONLY or SPAN_AND_EVENT - if get_content_capturing_mode() not in ( - ContentCapturingMode.EVENT_ONLY, - ContentCapturingMode.SPAN_AND_EVENT, - ): - return attributes - - if invocation.input_messages is not None: - if isinstance(invocation.input_messages, str): - attributes[GEN_AI_MEMORY_INPUT_MESSAGES] = ( - invocation.input_messages - ) - else: - attributes[GEN_AI_MEMORY_INPUT_MESSAGES] = gen_ai_json_dumps( - invocation.input_messages - ) - - if invocation.output_messages is not None: - if isinstance(invocation.output_messages, str): - attributes[GEN_AI_MEMORY_OUTPUT_MESSAGES] = ( - invocation.output_messages - ) - else: - attributes[GEN_AI_MEMORY_OUTPUT_MESSAGES] = gen_ai_json_dumps( - invocation.output_messages - ) + if invocation.query_text is not None: + attributes[GEN_AI_MEMORY_QUERY_TEXT] = invocation.query_text + if invocation.records is not None: + attributes[GEN_AI_MEMORY_RECORDS] = ( + invocation.records + if isinstance(invocation.records, str) + else gen_ai_json_dumps(invocation.records) + ) return attributes @@ -191,18 +107,13 @@ def _apply_memory_finish_attributes( span: Span, invocation: MemoryInvocation ) -> None: """Apply attributes for memory operations.""" - # Update span name with actual operation - if invocation.operation: - span.update_name(f"memory_operation {invocation.operation}") - else: - span.update_name("memory_operation") + span.update_name(invocation.operation) span.set_attribute(GEN_AI_SPAN_KIND, GenAiSpanKindValues.MEMORY.value) # Build all attributes attributes: dict[str, Any] = {} attributes.update(_get_memory_common_attributes(invocation)) - attributes.update(_get_memory_parameter_attributes(invocation)) # Recommended attributes if invocation.server_address is not None: @@ -210,11 +121,7 @@ def _apply_memory_finish_attributes( if invocation.server_port is not None: attributes[ServerAttributes.SERVER_PORT] = invocation.server_port - # Content attributes (controlled by content capturing mode) - # For spans, only capture if SPAN_ONLY or SPAN_AND_EVENT - attributes.update( - _get_memory_content_attributes(invocation, for_span=True) - ) + attributes.update(_get_memory_content_attributes(invocation)) # Custom attributes attributes.update(invocation.attributes) @@ -224,46 +131,7 @@ def _apply_memory_finish_attributes( span.set_attributes(attributes) -def _maybe_emit_memory_event( - logger: Logger | None, - span: Span, - invocation: MemoryInvocation, - error: Error | None = None, -) -> None: - """Emit a gen_ai.memory.operation.details event to the logger. - - This function creates a LogRecord event for memory operations following - the semantic convention for gen_ai.memory.operation.details. - """ - if not is_experimental_mode() or not should_emit_event() or logger is None: - return - - # Build event attributes - attributes: dict[str, Any] = {} - attributes.update(_get_memory_common_attributes(invocation)) - attributes.update(_get_memory_parameter_attributes(invocation)) - - # Content attributes for events (controlled by content capturing mode) - attributes.update( - _get_memory_content_attributes(invocation, for_span=False) - ) - - # Add error.type if operation ended in error - if error is not None: - attributes[ErrorAttributes.ERROR_TYPE] = error.type.__qualname__ - - # Create and emit the event with span context - context = set_span_in_context(span, get_current()) - event = LogRecord( - event_name="gen_ai.memory.operation.details", - attributes=attributes, - context=context, - ) - logger.emit(event) - - __all__ = [ "MemoryInvocation", "_apply_memory_finish_attributes", - "_maybe_emit_memory_event", ] diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_extended_attributes.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_extended_attributes.py index b471e8a66..569c6a854 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_extended_attributes.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_extended_attributes.py @@ -26,6 +26,43 @@ The tool definitions used by the model. """ +# Attributes present in the current upstream GenAI semantic conventions but +# not yet shipped by the minimum opentelemetry-semantic-conventions version +# supported by LoongSuite. +GEN_AI_PROMPT_NAME: Final = "gen_ai.prompt.name" +"""The name that uniquely identifies the prompt template.""" + +GEN_AI_PROMPT_VERSION: Final = "gen_ai.prompt.version" +"""The version of the prompt template.""" + +GEN_AI_PROMPT_VARIABLE_PREFIX: Final = "gen_ai.prompt.variable." +"""Prefix for prompt-template variables; append the variable name.""" + +GEN_AI_REQUEST_STREAM: Final = "gen_ai.request.stream" +"""Whether the request uses streaming. Emit only for streaming requests.""" + +GEN_AI_REQUEST_REASONING_LEVEL: Final = "gen_ai.request.reasoning.level" +"""The provider-specific reasoning or thinking effort requested.""" + +GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK: Final = ( + "gen_ai.response.time_to_first_chunk" +) +"""Time from request issuance to the first response chunk, in seconds.""" + +GEN_AI_USAGE_REASONING_OUTPUT_TOKENS: Final = ( + "gen_ai.usage.reasoning.output_tokens" +) +"""Output tokens used for reasoning, included in output-token usage.""" + +GEN_AI_CONVERSATION_COMPACTED: Final = "gen_ai.conversation.compacted" +"""Positive indicator that the effective conversation was compacted.""" + +GEN_AI_AGENT_VERSION: Final = "gen_ai.agent.version" +"""The version of the GenAI agent.""" + +GEN_AI_WORKFLOW_NAME: Final = "gen_ai.workflow.name" +"""The low-cardinality, human-readable name of the GenAI workflow.""" + # Embedding attributes GEN_AI_EMBEDDINGS_DIMENSION_COUNT: Final = "gen_ai.embeddings.dimension.count" """ @@ -46,7 +83,7 @@ # Retrieval attributes GEN_AI_RETRIEVAL_QUERY_TEXT: Final = "gen_ai.retrieval.query.text" """ -The retrieval query text (short phrase). Per LoongSuite semantic convention. +The retrieval query text (short phrase). Per upstream OpenTelemetry GenAI conventions. """ GEN_AI_RETRIEVAL_DOCUMENTS: Final = "gen_ai.retrieval.documents" @@ -54,6 +91,9 @@ The documents retrieved from the search or vector database. """ +GEN_AI_RETRIEVAL_TOP_K: Final = "gen_ai.retrieval.top_k" +"""The maximum number of retrieval results requested.""" + # Rerank attributes GEN_AI_RERANK_DOCUMENTS_COUNT: Final = "gen_ai.rerank.documents.count" """ @@ -233,8 +273,14 @@ class GenAiSpanKindValues(Enum): class GenAiExtendedOperationNameValues(Enum): + INVOKE_WORKFLOW = "invoke_workflow" + """Invoke a coordinated GenAI workflow.""" + + PLAN = "plan" + """Agent planning or task-decomposition phase.""" + RETRIEVAL = "retrieval" - """Retrieval operation (vector store / database lookup). Per LoongSuite semantic convention.""" + """Retrieval operation (vector store / database lookup) from upstream OpenTelemetry GenAI conventions.""" RERANK_DOCUMENTS = "rerank_documents" """Rerank documents operation.""" @@ -253,5 +299,5 @@ class GenAiExtendedProviderNameValues(Enum): OLLAMA = "ollama" """Ollama.""" - MOONSHOT = "moonshot" + MOONSHOT = "moonshot_ai" """Moonshot.""" diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_memory_attributes.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_memory_attributes.py index 302b2ba05..1ee8ea470 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_memory_attributes.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_semconv/gen_ai_memory_attributes.py @@ -20,7 +20,26 @@ from enum import Enum from typing import Final -# Memory operation type +# Current upstream OpenTelemetry GenAI Memory attributes. These are defined +# locally until LoongSuite can raise its semantic-conventions dependency floor. +GEN_AI_MEMORY_STORE_ID: Final = "gen_ai.memory.store.id" +"""Unique identifier of the memory store.""" + +GEN_AI_MEMORY_RECORD_ID: Final = "gen_ai.memory.record.id" +"""Unique identifier of a memory record.""" + +GEN_AI_MEMORY_RECORD_COUNT: Final = "gen_ai.memory.record.count" +"""Number of memory records affected by the operation.""" + +GEN_AI_MEMORY_QUERY_TEXT: Final = "gen_ai.memory.query.text" +"""Opt-in text used to query a memory store.""" + +GEN_AI_MEMORY_RECORDS: Final = "gen_ai.memory.records" +"""Opt-in structured records stored or retrieved by the operation.""" + +# Legacy LoongSuite Memory attributes. Kept as import-compatible constants for +# one transition cycle, but the standard Memory span utility no longer emits +# them. GEN_AI_MEMORY_OPERATION: Final = "gen_ai.memory.operation" """ The type of memory operation being performed. @@ -133,3 +152,15 @@ class GenAiMemoryOperationValues(Enum): DELETE_ALL = "delete_all" """Delete all memory records.""" + + +class GenAiMemoryOperationNameValues(Enum): + """Operation names from the upstream GenAI Memory span convention.""" + + CREATE_MEMORY_STORE = "create_memory_store" + SEARCH_MEMORY = "search_memory" + CREATE_MEMORY = "create_memory" + UPDATE_MEMORY = "update_memory" + UPSERT_MEMORY = "upsert_memory" + DELETE_MEMORY = "delete_memory" + DELETE_MEMORY_STORE = "delete_memory_store" diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_span_utils.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_span_utils.py index 22952132f..25e1d52a5 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_span_utils.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_span_utils.py @@ -37,6 +37,7 @@ from opentelemetry.trace import Span from opentelemetry.trace.propagation import set_span_in_context from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( # pylint: disable=no-name-in-module + GEN_AI_AGENT_VERSION, GEN_AI_EMBEDDINGS_DIMENSION_COUNT, GEN_AI_REACT_FINISH_REASON, GEN_AI_REACT_ROUND, @@ -53,6 +54,7 @@ GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, GEN_AI_RETRIEVAL_DOCUMENTS, GEN_AI_RETRIEVAL_QUERY_TEXT, + GEN_AI_RETRIEVAL_TOP_K, GEN_AI_SESSION_ID, GEN_AI_SKILL_DESCRIPTION, GEN_AI_SKILL_ID, @@ -148,6 +150,8 @@ def _get_invoke_agent_common_attributes( attributes[GenAI.GEN_AI_AGENT_ID] = invocation.agent_id if invocation.agent_name is not None: attributes[GenAI.GEN_AI_AGENT_NAME] = invocation.agent_name + if invocation.agent_version is not None: + attributes[GEN_AI_AGENT_VERSION] = invocation.agent_version if invocation.conversation_id is not None: attributes[GenAI.GEN_AI_CONVERSATION_ID] = invocation.conversation_id if invocation.data_source_id is not None: @@ -318,32 +322,23 @@ def _get_retrieval_documents_attributes( documents: list[RetrievalDocument] | None, ) -> dict[str, Any]: """ - Get retrieval attributes. - Records documents only when experimental mode is enabled. - Serialization follows ToolDefinition pattern: - - When content capturing is SPAN_ONLY or SPAN_AND_EVENT: full (id, score, content, metadata) - - Otherwise (NO_CONTENT): only id and score + Get opt-in retrieval document attributes. + + The upstream convention marks the entire ``gen_ai.retrieval.documents`` + attribute as opt-in, so it must be omitted unless span content capture is + explicitly enabled. When enabled, serialize the complete document model. """ attributes: dict[str, Any] = {} - if not is_experimental_mode(): + if not is_experimental_mode() or get_content_capturing_mode() not in ( + ContentCapturingMode.SPAN_ONLY, + ContentCapturingMode.SPAN_AND_EVENT, + ): return attributes if not documents: return attributes - should_record_full = get_content_capturing_mode() in ( - ContentCapturingMode.SPAN_ONLY, - ContentCapturingMode.SPAN_AND_EVENT, - ) - - doc_dicts: list[dict[str, Any]] = [] - for doc in documents: - if should_record_full: - doc_dicts.append(asdict(doc)) - else: - doc_dicts.append({"id": doc.id, "score": doc.score}) - - if doc_dicts: - attributes[GEN_AI_RETRIEVAL_DOCUMENTS] = gen_ai_json_dumps(doc_dicts) + doc_dicts = [asdict(doc) for doc in documents] + attributes[GEN_AI_RETRIEVAL_DOCUMENTS] = gen_ai_json_dumps(doc_dicts) return attributes @@ -481,6 +476,8 @@ def _apply_create_agent_finish_attributes( attributes[GenAI.GEN_AI_AGENT_ID] = invocation.agent_id if invocation.agent_name is not None: attributes[GenAI.GEN_AI_AGENT_NAME] = invocation.agent_name + if invocation.agent_version is not None: + attributes[GEN_AI_AGENT_VERSION] = invocation.agent_version if invocation.request_model is not None: attributes[GenAI.GEN_AI_REQUEST_MODEL] = invocation.request_model if invocation.server_port is not None: @@ -638,7 +635,7 @@ def _get_retrieval_span_name(invocation: RetrievalInvocation) -> str: def _apply_retrieval_finish_attributes( span: Span, invocation: RetrievalInvocation ) -> None: - """Apply attributes for retrieval operations (per LoongSuite semantic convention).""" + """Apply attributes for upstream OpenTelemetry GenAI retrieval operations.""" span.update_name(_get_retrieval_span_name(invocation)) # Build all attributes @@ -662,7 +659,7 @@ def _apply_retrieval_finish_attributes( # Recommended if invocation.top_k is not None: - attributes[GenAI.GEN_AI_REQUEST_TOP_K] = invocation.top_k + attributes[GEN_AI_RETRIEVAL_TOP_K] = invocation.top_k # Optional: retrieval query (sensitive - controlled by content capturing mode) if invocation.query is not None and ( diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_types.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_types.py index d74131cf9..8b8b0808d 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_types.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/extended_types.py @@ -145,6 +145,7 @@ class CreateAgentInvocation: # Agent-specific attributes agent_id: str | None = None agent_description: str | None = None + agent_version: str | None = None request_model: str | None = None # Server information server_address: str | None = None @@ -180,6 +181,7 @@ class InvokeAgentInvocation: # Agent-specific attributes agent_id: str | None = None agent_description: str | None = None + agent_version: str | None = None conversation_id: str | None = None data_source_id: str | None = None request_model: str | None = None @@ -233,7 +235,7 @@ class RetrievalInvocation: data_source_id: str | None = None provider: str | None = None request_model: str | None = None - top_k: float | None = None + top_k: int | None = None server_address: str | None = None server_port: int | None = None monotonic_start_s: float | None = None diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py index 509b953af..d486f50e7 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py @@ -15,7 +15,7 @@ from __future__ import annotations from dataclasses import asdict -from typing import Any +from typing import Any, Dict from opentelemetry._logs import Logger, LogRecord from opentelemetry.context import get_current @@ -35,10 +35,18 @@ gen_ai_extended_attributes as GenAIExtended, # LoongSuite Extension ) from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( # pylint: disable=no-name-in-module + GEN_AI_CONVERSATION_COMPACTED, + GEN_AI_PROMPT_NAME, + GEN_AI_PROMPT_VARIABLE_PREFIX, + GEN_AI_PROMPT_VERSION, + GEN_AI_REQUEST_REASONING_LEVEL, + GEN_AI_REQUEST_STREAM, + GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK, GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, # LoongSuite Extension GEN_AI_SPAN_KIND, # LoongSuite Extension GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, # LoongSuite Extension GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, # LoongSuite Extension + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, GEN_AI_USAGE_TOTAL_TOKENS, # LoongSuite Extension GenAiSpanKindValues, # LoongSuite Extension ) @@ -75,6 +83,10 @@ def _get_llm_common_attributes( GenAI.GEN_AI_CONVERSATION_ID, invocation.conversation_id, ), # LoongSuite Extension + ( + GEN_AI_CONVERSATION_COMPACTED, + True if invocation.conversation_compacted is True else None, + ), (server_attributes.SERVER_ADDRESS, invocation.server_address), (server_attributes.SERVER_PORT, invocation.server_port), ) @@ -256,6 +268,9 @@ def _maybe_emit_llm_event( attributes: dict[str, Any] = {} attributes.update(_get_llm_common_attributes(invocation)) attributes.update(_get_llm_request_attributes(invocation)) + attributes.update( + _get_prompt_variables_attributes(invocation, for_span=False) + ) attributes.update(_get_llm_response_attributes(invocation)) attributes.update( _get_llm_messages_attributes_for_event( @@ -293,6 +308,9 @@ def _apply_llm_finish_attributes( attributes: dict[str, Any] = {} attributes.update(_get_llm_common_attributes(invocation)) attributes.update(_get_llm_request_attributes(invocation)) + attributes.update( + _get_prompt_variables_attributes(invocation, for_span=True) + ) attributes.update(_get_llm_response_attributes(invocation)) attributes.update( _get_llm_messages_attributes_for_span( @@ -325,6 +343,10 @@ def _get_llm_request_attributes( ) -> dict[str, Any]: """Get GenAI request semantic convention attributes.""" optional_attrs = ( + (GEN_AI_PROMPT_NAME, invocation.prompt_name), + (GEN_AI_PROMPT_VERSION, invocation.prompt_version), + (GEN_AI_REQUEST_STREAM, True if invocation.stream is True else None), + (GEN_AI_REQUEST_REASONING_LEVEL, invocation.reasoning_level), ( GenAI.GEN_AI_OUTPUT_TYPE, invocation.output_type, @@ -349,6 +371,31 @@ def _get_llm_request_attributes( return {key: value for key, value in optional_attrs if value is not None} +def _get_prompt_variables_attributes( + invocation: LLMInvocation, *, for_span: bool +) -> dict[str, str]: + """Return opt-in prompt variables for the requested telemetry signal.""" + if not is_experimental_mode(): + return {} + + allowed_modes = ( + (ContentCapturingMode.SPAN_ONLY, ContentCapturingMode.SPAN_AND_EVENT) + if for_span + else ( + ContentCapturingMode.EVENT_ONLY, + ContentCapturingMode.SPAN_AND_EVENT, + ) + ) + if get_content_capturing_mode() not in allowed_modes: + return {} + + return { + f"{GEN_AI_PROMPT_VARIABLE_PREFIX}{name}": value + for name, value in invocation.prompt_variables.items() + if name and value is not None # pyright: ignore[reportUnnecessaryComparison] + } + + def _get_llm_response_attributes( invocation: LLMInvocation, ) -> dict[str, Any]: @@ -379,6 +426,10 @@ def _get_llm_response_attributes( (GenAI.GEN_AI_RESPONSE_ID, invocation.response_id), (GenAI.GEN_AI_USAGE_INPUT_TOKENS, invocation.input_tokens), (GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, invocation.output_tokens), + ( + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, + invocation.reasoning_output_tokens, + ), ( GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, # LoongSuite Extension invocation.usage_cache_creation_input_tokens, @@ -389,7 +440,9 @@ def _get_llm_response_attributes( ), ) - result = {key: value for key, value in optional_attrs if value is not None} + result: Dict[str, Any] = { + key: value for key, value in optional_attrs if value is not None + } # LoongSuite Extension: Calculate total_tokens as sum of input and output tokens when both are available total_tokens = 0 @@ -400,6 +453,15 @@ def _get_llm_response_attributes( if total_tokens > 0: result[GEN_AI_USAGE_TOTAL_TOKENS] = total_tokens + if ( + invocation.monotonic_first_chunk_s is not None + and invocation.monotonic_start_s is not None + and invocation.monotonic_first_chunk_s >= invocation.monotonic_start_s + ): + result[GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK] = ( + invocation.monotonic_first_chunk_s - invocation.monotonic_start_s + ) + # LoongSuite Extension: Time to first token for streaming responses (in nanoseconds) if ( invocation.monotonic_first_token_s is not None diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py index 032b95f1d..1da6d3a00 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py @@ -214,6 +214,10 @@ def _new_str_any_dict() -> dict[str, Any]: return {} +def _new_str_str_dict() -> dict[str, str]: + return {} + + # LoongSuite Extension @@ -280,12 +284,26 @@ class LLMInvocation(GenAIInvocation): server_port: int | None = None conversation_id: str | None = None # LoongSuite Extension """``gen_ai.conversation.id``.""" + conversation_compacted: bool | None = None + """``gen_ai.conversation.compacted``; only ``True`` is emitted.""" + prompt_name: str | None = None + """``gen_ai.prompt.name``.""" + prompt_version: str | None = None + """``gen_ai.prompt.version``.""" + prompt_variables: dict[str, str] = field(default_factory=_new_str_str_dict) + """Opt-in ``gen_ai.prompt.variable.`` attributes.""" + stream: bool | None = None + """``gen_ai.request.stream``; only streaming requests emit ``True``.""" + reasoning_level: str | None = None + """``gen_ai.request.reasoning.level``.""" output_type: str | None = None # LoongSuite Extension """``gen_ai.output.type`` (e.g. text, json, image).""" choice_count: int | None = None # LoongSuite Extension """``gen_ai.request.choice.count`` (omit on span when ``1``).""" - top_k: float | None = None # LoongSuite Extension + top_k: int | None = None # LoongSuite Extension """``gen_ai.request.top_k``.""" + reasoning_output_tokens: int | None = None + """``gen_ai.usage.reasoning.output_tokens``.""" usage_cache_creation_input_tokens: int | None = ( None # LoongSuite Extension ) @@ -316,6 +334,12 @@ class LLMInvocation(GenAIInvocation): attribute is calculated as (monotonic_first_token_s - monotonic_start_s) * 1e9 and stored in nanoseconds. """ + monotonic_first_chunk_s: float | None = None + """ + Monotonic time when the first response chunk was received. The + ``gen_ai.response.time_to_first_chunk`` attribute is calculated in seconds. + This timestamp is independent from ``monotonic_first_token_s``. + """ @dataclass diff --git a/util/opentelemetry-util-genai/tests/test_extended_handler.py b/util/opentelemetry-util-genai/tests/test_extended_handler.py index e279d9f43..c68b89d7f 100644 --- a/util/opentelemetry-util-genai/tests/test_extended_handler.py +++ b/util/opentelemetry-util-genai/tests/test_extended_handler.py @@ -63,12 +63,14 @@ get_extended_telemetry_handler, ) from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( + GEN_AI_AGENT_VERSION, GEN_AI_EMBEDDINGS_DIMENSION_COUNT, GEN_AI_REACT_FINISH_REASON, GEN_AI_REACT_ROUND, GEN_AI_RERANK_DOCUMENTS_COUNT, GEN_AI_RETRIEVAL_DOCUMENTS, GEN_AI_RETRIEVAL_QUERY_TEXT, + GEN_AI_RETRIEVAL_TOP_K, GEN_AI_SESSION_ID, GEN_AI_SKILL_DESCRIPTION, GEN_AI_SKILL_ID, @@ -198,6 +200,7 @@ def test_create_agent_start_and_stop_creates_span(self): invocation.agent_name = "TestAgent" invocation.agent_id = "agent_123" invocation.agent_description = "A test agent" + invocation.agent_version = "1.2.3" invocation.request_model = "gpt-4" invocation.server_address = "api.openai.com" invocation.server_port = 443 @@ -217,6 +220,7 @@ def test_create_agent_start_and_stop_creates_span(self): GenAI.GEN_AI_AGENT_NAME: "TestAgent", GenAI.GEN_AI_AGENT_ID: "agent_123", GenAI.GEN_AI_AGENT_DESCRIPTION: "A test agent", + GEN_AI_AGENT_VERSION: "1.2.3", GenAI.GEN_AI_REQUEST_MODEL: "gpt-4", ServerAttributes.SERVER_ADDRESS: "api.openai.com", ServerAttributes.SERVER_PORT: 443, @@ -557,6 +561,7 @@ def test_invoke_agent_start_and_stop_creates_span(self): invocation.agent_name = "CustomerAgent" invocation.agent_id = "agent_abc" invocation.agent_description = "Customer service agent" + invocation.agent_version = "2026-07-14" invocation.conversation_id = "conv_123" invocation.request_model = "gpt-4" invocation.temperature = 0.7 @@ -581,6 +586,7 @@ def test_invoke_agent_start_and_stop_creates_span(self): GenAI.GEN_AI_AGENT_NAME: "CustomerAgent", GenAI.GEN_AI_AGENT_ID: "agent_abc", GenAI.GEN_AI_AGENT_DESCRIPTION: "Customer service agent", + GEN_AI_AGENT_VERSION: "2026-07-14", GenAI.GEN_AI_CONVERSATION_ID: "conv_123", GenAI.GEN_AI_REQUEST_MODEL: "gpt-4", GenAI.GEN_AI_REQUEST_TEMPERATURE: 0.7, @@ -1060,7 +1066,7 @@ def test_retrieval_start_and_stop_creates_span(self): span = _get_single_span(self.span_exporter) self.assertEqual(span.name, "retrieval") - self.assertEqual(span.kind, trace.SpanKind.INTERNAL) + self.assertEqual(span.kind, trace.SpanKind.CLIENT) _assert_span_time_order(span) span_attrs = _get_span_attributes(span) @@ -1153,8 +1159,8 @@ def test_retrieval_without_sensitive_data(self): stability_mode="gen_ai_latest_experimental", content_capturing="NO_CONTENT", ) - def test_retrieval_no_content_records_id_score_only(self): - """When content capture is NO_CONTENT, query is omitted; documents record id and score only.""" + def test_retrieval_no_content_omits_opt_in_attributes(self): + """NO_CONTENT omits both opt-in query and document attributes.""" documents = [ RetrievalDocument( id="doc_123", @@ -1174,20 +1180,11 @@ def test_retrieval_no_content_records_id_score_only(self): span_attrs, "Query should NOT be captured when content capture is NO_CONTENT", ) - self.assertIn( + self.assertNotIn( GEN_AI_RETRIEVAL_DOCUMENTS, span_attrs, - "Documents should be recorded with id and score only", + "Documents are opt-in and should be omitted when content capture is disabled", ) - docs_val = span_attrs[GEN_AI_RETRIEVAL_DOCUMENTS] - self.assertIn("doc_123", docs_val) - self.assertIn("0.95", docs_val) - self.assertNotIn( - "sensitive doc content", - docs_val, - "Content should NOT be in documents when NO_CONTENT", - ) - self.assertNotIn("secret", docs_val) @patch_env_vars( stability_mode="gen_ai_latest_experimental", @@ -1224,7 +1221,7 @@ def test_retrieval_span_name_with_data_source_id(self): invocation.query = "test query" invocation.provider = "chroma" invocation.request_model = "embedding-model" - invocation.top_k = 5.0 + invocation.top_k = 5 span = _get_single_span(self.span_exporter) self.assertEqual(span.name, "retrieval H7STPQYOND") @@ -1236,7 +1233,7 @@ def test_retrieval_span_name_with_data_source_id(self): GenAI.GEN_AI_DATA_SOURCE_ID: "H7STPQYOND", GenAI.GEN_AI_PROVIDER_NAME: "chroma", GenAI.GEN_AI_REQUEST_MODEL: "embedding-model", - GenAI.GEN_AI_REQUEST_TOP_K: 5.0, + GEN_AI_RETRIEVAL_TOP_K: 5, GEN_AI_RETRIEVAL_QUERY_TEXT: "test query", }, ) diff --git a/util/opentelemetry-util-genai/tests/test_extended_memory.py b/util/opentelemetry-util-genai/tests/test_extended_memory.py index 48bfd2025..de61b277b 100644 --- a/util/opentelemetry-util-genai/tests/test_extended_memory.py +++ b/util/opentelemetry-util-genai/tests/test_extended_memory.py @@ -12,35 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json import os -import unittest -from typing import Any, Mapping from unittest.mock import patch -from opentelemetry import trace +import pytest + from opentelemetry.instrumentation._semconv import ( OTEL_SEMCONV_STABILITY_OPT_IN, _OpenTelemetrySemanticConventionStability, ) - -# Backward compatibility for InMemoryLogExporter -> InMemoryLogRecordExporter rename -# Changed in opentelemetry-sdk@0.60b0 -try: - from opentelemetry.sdk._logs.export import ( # pylint: disable=no-name-in-module - InMemoryLogRecordExporter, - SimpleLogRecordProcessor, - ) -except ImportError: - # Fallback to old name for compatibility with older SDK versions - from opentelemetry.sdk._logs.export import ( - InMemoryLogExporter as InMemoryLogRecordExporter, - ) - from opentelemetry.sdk._logs.export import ( - SimpleLogRecordProcessor, - ) - -from opentelemetry.sdk._logs import LoggerProvider -from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, @@ -48,564 +30,182 @@ from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) -from opentelemetry.semconv.attributes import ( - error_attributes as ErrorAttributes, -) -from opentelemetry.semconv.attributes import ( - server_attributes as ServerAttributes, -) +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.trace import SpanKind from opentelemetry.trace.status import StatusCode from opentelemetry.util.genai.environment_variables import ( OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT, - OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, ) from opentelemetry.util.genai.extended_handler import ( get_extended_telemetry_handler, ) from opentelemetry.util.genai.extended_memory import MemoryInvocation +from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( + GEN_AI_SPAN_KIND, + GenAiSpanKindValues, +) from opentelemetry.util.genai.extended_semconv.gen_ai_memory_attributes import ( - GEN_AI_MEMORY_AGENT_ID, - GEN_AI_MEMORY_APP_ID, - GEN_AI_MEMORY_ID, - GEN_AI_MEMORY_INPUT_MESSAGES, - GEN_AI_MEMORY_LIMIT, - GEN_AI_MEMORY_MEMORY_TYPE, - GEN_AI_MEMORY_OPERATION, - GEN_AI_MEMORY_OUTPUT_MESSAGES, - GEN_AI_MEMORY_PAGE, - GEN_AI_MEMORY_PAGE_SIZE, - GEN_AI_MEMORY_RERANK, - GEN_AI_MEMORY_RUN_ID, - GEN_AI_MEMORY_THRESHOLD, - GEN_AI_MEMORY_TOP_K, - GEN_AI_MEMORY_USER_ID, + GEN_AI_MEMORY_QUERY_TEXT, + GEN_AI_MEMORY_RECORD_COUNT, + GEN_AI_MEMORY_RECORD_ID, + GEN_AI_MEMORY_RECORDS, + GEN_AI_MEMORY_STORE_ID, + GenAiMemoryOperationNameValues, ) -def patch_env_vars(stability_mode, content_capturing=None, emit_event=None): - def decorator(test_case): - env_vars = { - OTEL_SEMCONV_STABILITY_OPT_IN: stability_mode, - } - if content_capturing is not None: - env_vars[OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT] = ( - content_capturing - ) - if emit_event is not None: - env_vars[OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT] = emit_event +@pytest.fixture(name="memory_telemetry") +def fixture_memory_telemetry(): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + handler = get_extended_telemetry_handler(tracer_provider=provider) + yield handler, exporter + exporter.clear() + if hasattr(get_extended_telemetry_handler, "_default_handler"): + delattr(get_extended_telemetry_handler, "_default_handler") + - @patch.dict(os.environ, env_vars) - def wrapper(*args, **kwargs): - # Reset state. - _OpenTelemetrySemanticConventionStability._initialized = False - _OpenTelemetrySemanticConventionStability._initialize() - return test_case(*args, **kwargs) +def _span(exporter): + spans = exporter.get_finished_spans() + assert len(spans) == 1 + return spans[0] - return wrapper - return decorator +def _enable_content_capture(): + return patch.dict( + os.environ, + { + OTEL_SEMCONV_STABILITY_OPT_IN: "gen_ai_latest_experimental", + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "SPAN_ONLY", + }, + ) -def _get_single_span(span_exporter: InMemorySpanExporter) -> ReadableSpan: - spans = span_exporter.get_finished_spans() - assert len(spans) == 1 - return spans[0] +def _reset_semconv_mode(): + _OpenTelemetrySemanticConventionStability._initialized = False + _OpenTelemetrySemanticConventionStability._initialize() -def _assert_span_time_order(span: ReadableSpan) -> None: - assert span.start_time is not None - assert span.end_time is not None - assert span.end_time >= span.start_time - - -def _get_span_attributes(span: ReadableSpan) -> Mapping[str, Any]: - attrs = span.attributes - assert attrs is not None - return attrs - - -def _assert_span_attributes( - span_attrs: Mapping[str, Any], expected_values: Mapping[str, Any] -) -> None: - for key, value in expected_values.items(): - assert span_attrs.get(key) == value - - -class TestMemoryOperations(unittest.TestCase): # pylint: disable=too-many-public-methods - def setUp(self): - self.span_exporter = InMemorySpanExporter() - tracer_provider = TracerProvider() - tracer_provider.add_span_processor( - SimpleSpanProcessor(self.span_exporter) - ) - - self.log_exporter = InMemoryLogRecordExporter() - logger_provider = LoggerProvider() - logger_provider.add_log_record_processor( - SimpleLogRecordProcessor(self.log_exporter) - ) - - self.telemetry_handler = get_extended_telemetry_handler( - tracer_provider=tracer_provider, - logger_provider=logger_provider, - ) - - def tearDown(self): - # Clear spans, logs and reset the singleton telemetry handler so each test starts clean - self.span_exporter.clear() - self.log_exporter.clear() - if hasattr(get_extended_telemetry_handler, "_default_handler"): - delattr(get_extended_telemetry_handler, "_default_handler") - - # ==================== Memory Operation Tests ==================== - - def test_memory_add_start_and_stop_creates_span(self): - invocation = MemoryInvocation(operation="add") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - invocation.run_id = "run_789" - invocation.input_messages = "Remember that user likes apples" - invocation.server_address = "api.mem0.ai" - invocation.server_port = 443 - invocation.attributes = {"custom": "value"} - - span = _get_single_span(self.span_exporter) - self.assertEqual(span.name, "memory_operation add") - self.assertEqual(span.kind, trace.SpanKind.CLIENT) - _assert_span_time_order(span) - - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GenAI.GEN_AI_OPERATION_NAME: "memory_operation", - GEN_AI_MEMORY_OPERATION: "add", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - GEN_AI_MEMORY_RUN_ID: "run_789", - ServerAttributes.SERVER_ADDRESS: "api.mem0.ai", - ServerAttributes.SERVER_PORT: 443, - "custom": "value", - }, - ) - - def test_memory_search_with_parameters(self): - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - invocation.limit = 10 - invocation.threshold = 0.7 - invocation.rerank = True - invocation.top_k = 5 - - span = _get_single_span(self.span_exporter) - self.assertEqual(span.name, "memory_operation search") - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "search", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - GEN_AI_MEMORY_LIMIT: 10, - GEN_AI_MEMORY_THRESHOLD: 0.7, - GEN_AI_MEMORY_RERANK: True, - GEN_AI_MEMORY_TOP_K: 5, - }, - ) - - def test_memory_update_operation(self): - invocation = MemoryInvocation(operation="update") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.memory_id = "mem_abc123" - invocation.user_id = "user_123" - invocation.input_messages = "Updated memory content" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "update", - GEN_AI_MEMORY_ID: "mem_abc123", - GEN_AI_MEMORY_USER_ID: "user_123", - }, - ) - - def test_memory_get_operation(self): - invocation = MemoryInvocation(operation="get") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.memory_id = "mem_xyz789" - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "get", - GEN_AI_MEMORY_ID: "mem_xyz789", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - }, - ) - - def test_memory_get_all_with_pagination(self): - invocation = MemoryInvocation(operation="get_all") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.page = 1 - invocation.page_size = 100 - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "get_all", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_PAGE: 1, - GEN_AI_MEMORY_PAGE_SIZE: 100, - }, - ) - - def test_memory_history_operation(self): - invocation = MemoryInvocation(operation="history") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - invocation.run_id = "run_789" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "history", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - GEN_AI_MEMORY_RUN_ID: "run_789", - }, - ) - - def test_memory_with_app_id(self): - """Test memory operation with app_id (for managed platforms).""" - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.app_id = "app_001" - invocation.user_id = "user_123" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "search", - GEN_AI_MEMORY_APP_ID: "app_001", - GEN_AI_MEMORY_USER_ID: "user_123", - }, - ) - - def test_memory_with_memory_type(self): - invocation = MemoryInvocation(operation="add") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.memory_type = "procedural_memory" - invocation.user_id = "user_123" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "add", - GEN_AI_MEMORY_MEMORY_TYPE: "procedural_memory", - GEN_AI_MEMORY_USER_ID: "user_123", - }, - ) - - def test_memory_manual_start_and_stop(self): - invocation = MemoryInvocation(operation="search") - invocation.user_id = "user_123" - invocation.limit = 20 - - self.telemetry_handler.start_memory(invocation) - assert invocation.span is not None - self.telemetry_handler.stop_memory(invocation) - - span = _get_single_span(self.span_exporter) - self.assertEqual(span.name, "memory_operation search") - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, +def test_search_memory_uses_upstream_name_and_fields(memory_telemetry): + handler, exporter = memory_telemetry + invocation = MemoryInvocation( + operation=GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + provider="mem0", + store_id="store-1", + record_id="memory-1", + record_count=2, + ) + + with handler.memory(invocation): + pass + + span = _span(exporter) + assert span.name == "search_memory" + assert span.kind is SpanKind.CLIENT + assert span.attributes == { + GenAI.GEN_AI_OPERATION_NAME: "search_memory", + GenAI.GEN_AI_PROVIDER_NAME: "mem0", + GEN_AI_MEMORY_STORE_ID: "store-1", + GEN_AI_MEMORY_RECORD_ID: "memory-1", + GEN_AI_MEMORY_RECORD_COUNT: 2, + GEN_AI_SPAN_KIND: GenAiSpanKindValues.MEMORY.value, + } + + +def test_local_memory_operation_can_be_internal(memory_telemetry): + handler, exporter = memory_telemetry + invocation = MemoryInvocation( + operation=GenAiMemoryOperationNameValues.UPSERT_MEMORY.value, + span_kind=SpanKind.INTERNAL, + ) + + with handler.memory(invocation): + pass + + assert _span(exporter).kind is SpanKind.INTERNAL + + +def test_memory_content_is_opt_in(memory_telemetry): + handler, exporter = memory_telemetry + invocation = MemoryInvocation( + operation=GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + query_text="vegetarian preferences", + records=[ { - GEN_AI_MEMORY_OPERATION: "search", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_LIMIT: 20, - }, - ) - - def test_memory_error_handling(self): - class MemoryOperationError(RuntimeError): + "id": "memory-1", + "content": "User prefers vegetarian meals", + "score": 0.95, + } + ], + ) + + with _enable_content_capture(): + _reset_semconv_mode() + with handler.memory(invocation): pass - with self.assertRaises(MemoryOperationError): - invocation = MemoryInvocation(operation="add") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - raise MemoryOperationError("Memory operation failed") - - span = _get_single_span(self.span_exporter) - self.assertEqual(span.status.status_code, StatusCode.ERROR) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - ErrorAttributes.ERROR_TYPE: MemoryOperationError.__qualname__, - }, - ) + attributes = _span(exporter).attributes + assert attributes[GEN_AI_MEMORY_QUERY_TEXT] == "vegetarian preferences" + assert json.loads(attributes[GEN_AI_MEMORY_RECORDS]) == invocation.records - @patch_env_vars( - stability_mode="gen_ai_latest_experimental", - content_capturing="SPAN_ONLY", - ) - def test_memory_with_content_capturing(self): - """Test that input/output messages are captured when content capturing is enabled.""" - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = "What does the user like?" - invocation.output_messages = "The user likes apples" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - - # Verify messages are captured - self.assertIn(GEN_AI_MEMORY_INPUT_MESSAGES, span_attrs) - self.assertIn(GEN_AI_MEMORY_OUTPUT_MESSAGES, span_attrs) - self.assertEqual( - span_attrs[GEN_AI_MEMORY_INPUT_MESSAGES], - "What does the user like?", - ) - self.assertEqual( - span_attrs[GEN_AI_MEMORY_OUTPUT_MESSAGES], "The user likes apples" - ) - - def test_memory_without_content_capturing(self): - """Test that messages are NOT captured when content capturing is disabled.""" - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = "What does the user like?" - invocation.output_messages = "The user likes apples" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - - # Verify messages are NOT captured - self.assertNotIn(GEN_AI_MEMORY_INPUT_MESSAGES, span_attrs) - self.assertNotIn(GEN_AI_MEMORY_OUTPUT_MESSAGES, span_attrs) - - @patch_env_vars( - stability_mode="gen_ai_latest_experimental", - content_capturing="EVENT_ONLY", - emit_event="true", + +def test_memory_content_is_not_captured_by_default(memory_telemetry): + handler, exporter = memory_telemetry + invocation = MemoryInvocation( + operation=GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + query_text="sensitive query", + records=[{"content": "sensitive memory"}], ) - def test_memory_emits_event(self): - """Test that memory operation emits events when emit_event is enabled.""" - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - invocation.input_messages = "What does the user like?" - invocation.output_messages = "The user likes apples" - - # Check that event was emitted - logs = self.log_exporter.get_finished_logs() - self.assertEqual(len(logs), 1) - log_record = logs[0].log_record - - # Verify event name - self.assertEqual( - log_record.event_name, - "gen_ai.memory.operation.details", - ) - - # Verify event attributes - attrs = log_record.attributes - self.assertIsNotNone(attrs) - self.assertEqual( - attrs[GenAI.GEN_AI_OPERATION_NAME], "memory_operation" - ) - self.assertEqual(attrs[GEN_AI_MEMORY_OPERATION], "search") - self.assertEqual(attrs[GEN_AI_MEMORY_USER_ID], "user_123") - self.assertEqual(attrs[GEN_AI_MEMORY_AGENT_ID], "agent_456") - self.assertIn(GEN_AI_MEMORY_INPUT_MESSAGES, attrs) - self.assertIn(GEN_AI_MEMORY_OUTPUT_MESSAGES, attrs) - - @patch_env_vars( - stability_mode="gen_ai_latest_experimental", - content_capturing="SPAN_AND_EVENT", - emit_event="true", + + with handler.memory(invocation): + pass + + attributes = _span(exporter).attributes + assert GEN_AI_MEMORY_QUERY_TEXT not in attributes + assert GEN_AI_MEMORY_RECORDS not in attributes + + +def test_legacy_memory_fields_are_not_emitted(memory_telemetry): + handler, exporter = memory_telemetry + invocation = MemoryInvocation( + operation=GenAiMemoryOperationNameValues.SEARCH_MEMORY.value, + user_id="user-1", + agent_id="agent-1", + memory_id="old-memory-id", + limit=10, + top_k=5, ) - def test_memory_emits_event_and_span(self): - """Test that memory operation emits both event and span when emit_event is enabled.""" - invocation = MemoryInvocation(operation="add") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = "User likes apples" - - # Check span was created - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - self.assertIn(GEN_AI_MEMORY_INPUT_MESSAGES, span_attrs) - - # Check event was emitted - logs = self.log_exporter.get_finished_logs() - self.assertEqual(len(logs), 1) - log_record = logs[0].log_record - self.assertEqual( - log_record.event_name, - "gen_ai.memory.operation.details", - ) - self.assertIn(GEN_AI_MEMORY_INPUT_MESSAGES, log_record.attributes) - - @patch_env_vars( - stability_mode="gen_ai_latest_experimental", - content_capturing="EVENT_ONLY", - emit_event="true", + + with handler.memory(invocation): + pass + + attributes = _span(exporter).attributes + assert not any( + key.startswith("gen_ai.memory.user_") + or key + in { + "gen_ai.memory.operation", + "gen_ai.memory.agent_id", + "gen_ai.memory.id", + "gen_ai.memory.limit", + "gen_ai.memory.top_k", + } + for key in attributes ) - def test_memory_emits_event_with_error(self): - """Test that memory operation emits event with error when operation fails.""" - class MemoryOperationError(RuntimeError): - pass - with self.assertRaises(MemoryOperationError): - invocation = MemoryInvocation(operation="add") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = "Test memory" - raise MemoryOperationError("Memory operation failed") - - # Check event was emitted - logs = self.log_exporter.get_finished_logs() - self.assertEqual(len(logs), 1) - log_record = logs[0].log_record - attrs = log_record.attributes - - # Verify error attribute is present - self.assertEqual( - attrs[ErrorAttributes.ERROR_TYPE], - MemoryOperationError.__qualname__, - ) - self.assertEqual( - attrs[GenAI.GEN_AI_OPERATION_NAME], "memory_operation" - ) - - def test_memory_does_not_emit_event_when_disabled(self): - """Test that memory operation does not emit event when emit_event is disabled.""" - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = "Test query" - - # Check that no event was emitted - logs = self.log_exporter.get_finished_logs() - self.assertEqual(len(logs), 0) - - def test_memory_batch_update_operation(self): - invocation = MemoryInvocation(operation="batch_update") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "batch_update", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - }, - ) - - def test_memory_delete_operation(self): - invocation = MemoryInvocation(operation="delete") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.memory_id = "mem_to_delete" - invocation.user_id = "user_123" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "delete", - GEN_AI_MEMORY_ID: "mem_to_delete", - GEN_AI_MEMORY_USER_ID: "user_123", - }, - ) - - def test_memory_batch_delete_operation(self): - invocation = MemoryInvocation(operation="batch_delete") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "batch_delete", - GEN_AI_MEMORY_USER_ID: "user_123", - }, - ) - - def test_memory_delete_all_operation(self): - invocation = MemoryInvocation(operation="delete_all") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.agent_id = "agent_456" - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - _assert_span_attributes( - span_attrs, - { - GEN_AI_MEMORY_OPERATION: "delete_all", - GEN_AI_MEMORY_USER_ID: "user_123", - GEN_AI_MEMORY_AGENT_ID: "agent_456", - }, - ) - - @patch_env_vars( - stability_mode="gen_ai_latest_experimental", - content_capturing="SPAN_ONLY", - ) - def test_memory_with_json_input_output(self): - """Test that JSON input/output messages are properly serialized.""" - input_data = {"query": "What does user like?", "context": "shopping"} - output_data = [ - {"memory_id": "mem1", "content": "User likes apples"}, - {"memory_id": "mem2", "content": "User likes oranges"}, - ] - - invocation = MemoryInvocation(operation="search") - with self.telemetry_handler.memory(invocation) as invocation: - invocation.user_id = "user_123" - invocation.input_messages = input_data - invocation.output_messages = output_data - - span = _get_single_span(self.span_exporter) - span_attrs = _get_span_attributes(span) - - # Verify messages are captured as JSON strings - self.assertIn(GEN_AI_MEMORY_INPUT_MESSAGES, span_attrs) - self.assertIn(GEN_AI_MEMORY_OUTPUT_MESSAGES, span_attrs) - # Should be JSON strings - self.assertIsInstance(span_attrs[GEN_AI_MEMORY_INPUT_MESSAGES], str) - self.assertIsInstance(span_attrs[GEN_AI_MEMORY_OUTPUT_MESSAGES], str) +def test_memory_error_sets_error_type(memory_telemetry): + handler, exporter = memory_telemetry + + with pytest.raises(RuntimeError): + with handler.memory( + MemoryInvocation( + operation=GenAiMemoryOperationNameValues.DELETE_MEMORY.value + ) + ): + raise RuntimeError("delete failed") + + span = _span(exporter) + assert span.status.status_code is StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "RuntimeError" diff --git a/util/opentelemetry-util-genai/tests/test_utils.py b/util/opentelemetry-util-genai/tests/test_utils.py index 04798684e..1745943af 100644 --- a/util/opentelemetry-util-genai/tests/test_utils.py +++ b/util/opentelemetry-util-genai/tests/test_utils.py @@ -47,7 +47,15 @@ OTEL_INSTRUMENTATION_GENAI_EMIT_EVENT, ) from opentelemetry.util.genai.extended_semconv.gen_ai_extended_attributes import ( # pylint: disable=no-name-in-module + GEN_AI_CONVERSATION_COMPACTED, + GEN_AI_PROMPT_NAME, + GEN_AI_PROMPT_VERSION, + GEN_AI_REQUEST_REASONING_LEVEL, + GEN_AI_REQUEST_STREAM, + GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK, + GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN, GEN_AI_SPAN_KIND, # LoongSuite Extension + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, GEN_AI_USAGE_TOTAL_TOKENS, # LoongSuite Extension GenAiSpanKindValues, # LoongSuite Extension ) @@ -446,6 +454,63 @@ def test_llm_span_finish_reasons_deduplicated_from_output_messages(self): ("length", "stop"), ) + @patch_env_vars( + stability_mode="gen_ai_latest_experimental", + content_capturing="SPAN_ONLY", + emit_event="", + ) + def test_llm_current_semconv_fields_and_separate_stream_times(self): + invocation = LLMInvocation( + request_model="reasoning-model", + provider="test-provider", + stream=True, + reasoning_level="high", + reasoning_output_tokens=13, + conversation_compacted=True, + prompt_name="support-answer", + prompt_version="v2", + prompt_variables={"language": "Chinese"}, + ) + + self.telemetry_handler.start_llm(invocation) + assert invocation.monotonic_start_s is not None + invocation.monotonic_first_chunk_s = ( + invocation.monotonic_start_s + 0.125 + ) + invocation.monotonic_first_token_s = ( + invocation.monotonic_start_s + 0.25 + ) + self.telemetry_handler.stop_llm(invocation) + + attrs = _get_span_attributes(_get_single_span(self.span_exporter)) + self.assertEqual(attrs[GEN_AI_REQUEST_STREAM], True) + self.assertEqual(attrs[GEN_AI_REQUEST_REASONING_LEVEL], "high") + self.assertEqual(attrs[GEN_AI_USAGE_REASONING_OUTPUT_TOKENS], 13) + self.assertEqual(attrs[GEN_AI_CONVERSATION_COMPACTED], True) + self.assertEqual(attrs[GEN_AI_PROMPT_NAME], "support-answer") + self.assertEqual(attrs[GEN_AI_PROMPT_VERSION], "v2") + self.assertEqual(attrs["gen_ai.prompt.variable.language"], "Chinese") + self.assertAlmostEqual( + attrs[GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK], 0.125 + ) + self.assertEqual( + attrs[GEN_AI_RESPONSE_TIME_TO_FIRST_TOKEN], 250_000_000 + ) + + def test_false_stream_and_compacted_flags_are_omitted(self): + invocation = LLMInvocation( + request_model="nonstream-model", + stream=False, + conversation_compacted=False, + ) + + self.telemetry_handler.start_llm(invocation) + self.telemetry_handler.stop_llm(invocation) + + attrs = _get_span_attributes(_get_single_span(self.span_exporter)) + self.assertNotIn(GEN_AI_REQUEST_STREAM, attrs) + self.assertNotIn(GEN_AI_CONVERSATION_COMPACTED, attrs) + def test_llm_span_uses_expected_schema_url(self): invocation = LLMInvocation( request_model="schema-model",