From b8a5ba3e747111a12748fb59186a6e328aba06ee Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Fri, 11 Sep 2026 19:49:09 +0000 Subject: [PATCH 1/3] Add instrumentation for LLM invocations --- .../README.rst | 6 +- .../instrumentation/genai/dspy/patch.py | 178 +++- .../instrumentation/genai/dspy/utils.py | 668 +++++++++++++- .../tests/conformance/lm.py | 92 ++ .../tests/test_conformance.py | 2 + .../tests/test_lm.py | 860 ++++++++++++++++++ 6 files changed, 1788 insertions(+), 18 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py create mode 100644 instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/README.rst b/instrumentation/opentelemetry-instrumentation-genai-dspy/README.rst index 7dfc90d62..df90673b1 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/README.rst @@ -8,9 +8,9 @@ OpenTelemetry DSPy Instrumentation This package provides OpenTelemetry instrumentation for the `DSPy framework `_, emitting Generative AI -semantic conventions for DSPy Tool executions, ReAct agent loops -(supporting both ``dspy.ReAct`` and ``dspy.ReActV2``), and ``dspy.Retrieve`` -retrieval operations. +semantic conventions for DSPy language models (``dspy.LM``), Tool executions, +ReAct agent loops (supporting both ``dspy.ReAct`` and ``dspy.ReActV2``), and +``dspy.Retrieve`` retrieval operations. Installation ------------ diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py index 87bdc5f3b..d3af8b8bc 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py @@ -7,7 +7,7 @@ import inspect import sys -from collections.abc import Awaitable, Callable, Sequence +from collections.abc import Awaitable, Callable, Mapping, Sequence from copy import copy, deepcopy from importlib import import_module from typing import TYPE_CHECKING, Any, cast @@ -21,14 +21,23 @@ from opentelemetry.instrumentation.genai.dspy.utils import ( SENTINEL_TOOL_NAMES, + _safe_float, + _safe_int, + _safe_stop_sequences, + apply_usage_to_invocation, extract_input_content, + extract_lm_input_messages, + extract_lm_output_messages, extract_output_content, prepare_tool_definitions, + resolve_provider, + resolve_request_model, ) from opentelemetry.instrumentation.utils import unwrap from opentelemetry.util.genai.handler import TelemetryHandler from opentelemetry.util.genai.invocation import ( AgentInvocation, + InferenceInvocation, RetrievalInvocation, ToolInvocation, ) @@ -40,6 +49,8 @@ if TYPE_CHECKING: from dspy.adapters.types.tool import Tool + from dspy.clients.lm import LM + from dspy.core.types import LMResponse from dspy.primitives.module import Module from dspy.primitives.prediction import Prediction from dspy.retrievers.retrieve import Retrieve @@ -217,6 +228,20 @@ def patch_dspy(handler: TelemetryHandler) -> None: _react_aforward(handler, "dspy.ReActV2"), ) + if hasattr(dspy, "LM"): + lm_module = dspy.LM.__module__ + lm_name = dspy.LM.__name__ + _wrap_function( + lm_module, + f"{lm_name}.__call__", + _lm_call(handler), + ) + _wrap_function( + lm_module, + f"{lm_name}.acall", + _lm_acall(handler), + ) + def unpatch_dspy() -> None: """Remove patches from DSPy classes.""" @@ -238,6 +263,157 @@ def unpatch_dspy() -> None: unwrap(react_v2_cls, "forward") unwrap(react_v2_cls, "aforward") + if hasattr(dspy, "LM"): + unwrap(dspy.LM, "__call__") + unwrap(dspy.LM, "acall") + + +def _start_lm_invocation( + handler: TelemetryHandler, + instance: LM, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> InferenceInvocation: + provider = resolve_provider(instance) + request_model = resolve_request_model(instance) + + invocation = handler.inference( + provider=provider, + request_model=request_model, + ) + + merged_kwargs = {**getattr(instance, "kwargs", {}), **kwargs} + + invocation.temperature = _safe_float(merged_kwargs.get("temperature")) + invocation.max_tokens = _safe_int(merged_kwargs.get("max_tokens")) + invocation.top_p = _safe_float( + merged_kwargs.get("top_p") + if merged_kwargs.get("top_p") is not None + else merged_kwargs.get("p") + ) + invocation.frequency_penalty = _safe_float( + merged_kwargs.get("frequency_penalty") + ) + invocation.presence_penalty = _safe_float( + merged_kwargs.get("presence_penalty") + ) + invocation.seed = _safe_int(merged_kwargs.get("seed")) + + invocation.stop_sequences = _safe_stop_sequences(merged_kwargs.get("stop")) + + choice_count = _safe_int(merged_kwargs.get("n")) + if choice_count is not None and choice_count != 1: + invocation.request_choice_count = choice_count + + if handler.should_capture_content(): + invocation.input_messages = extract_lm_input_messages(args, kwargs) + + return invocation + + +def _get_field(obj: Any, key: str) -> Any: + if isinstance(obj, Mapping): + return cast(Mapping[str, Any], obj).get(key) + return getattr(obj, key, None) + + +def _set_lm_invocation_response( + handler: TelemetryHandler, + invocation: InferenceInvocation, + instance: LM, + result: LMResponse | list[dict[str, Any] | str], +) -> None: + if not isinstance(result, list): + if result.model: + invocation.response_model_name = str(result.model) + if result.response_id: + invocation.response_id = str(result.response_id) + + usage_dict = result.usage_as_dict() + if usage_dict: + apply_usage_to_invocation(invocation, usage_dict) + + finish_reasons = [ + out.finish_reason for out in result.outputs if out.finish_reason + ] + if finish_reasons: + invocation.finish_reasons = finish_reasons + + if handler.should_capture_content(): + invocation.output_messages = extract_lm_output_messages(result) + return + + # DSPy 3.x LM calls return a legacy list by default unless experimental=True + # or an LMRequest is used. DSPy appends each call's metadata to instance.history, + # so history[-1] corresponds to the invocation that just finished. + finish_reason: str | None = None + history: Sequence[Mapping[str, Any]] | None = getattr( + instance, "history", None + ) + if isinstance(history, Sequence) and history: + last_entry = history[-1] + resp_model = last_entry.get("response_model") or last_entry.get( + "model" + ) + if resp_model: + invocation.response_model_name = str(resp_model) + + usage = last_entry.get("usage") + if isinstance(usage, Mapping): + apply_usage_to_invocation( + invocation, cast(Mapping[str, Any], usage) + ) + + resp_obj: Any = last_entry.get("response") + if resp_obj is not None: + resp_id = _get_field(resp_obj, "id") + if resp_id: + invocation.response_id = str(resp_id) + + choices = _get_field(resp_obj, "choices") + if isinstance(choices, Sequence) and choices: + fr = _get_field(choices[0], "finish_reason") + if fr: + finish_reason = str(fr) + invocation.finish_reasons = [finish_reason] + + if handler.should_capture_content(): + invocation.output_messages = extract_lm_output_messages( + result, finish_reason=finish_reason + ) + + +def _lm_call(handler: TelemetryHandler) -> Callable[..., Any]: + def traced_method( + wrapped: Callable[..., Any], + instance: LM, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + invocation = _start_lm_invocation(handler, instance, args, kwargs) + with invocation: + result = wrapped(*args, **kwargs) + _set_lm_invocation_response(handler, invocation, instance, result) + return result + + return traced_method + + +def _lm_acall(handler: TelemetryHandler) -> Callable[..., Any]: + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: LM, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + invocation = _start_lm_invocation(handler, instance, args, kwargs) + with invocation: + result = await wrapped(*args, **kwargs) + _set_lm_invocation_response(handler, invocation, instance, result) + return result + + return traced_method + def _extract_tool_arguments( instance: Tool, diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py index 8224b3d3e..702ae0fe5 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py @@ -7,26 +7,664 @@ import json from collections.abc import Callable, Iterable, Mapping, Sequence -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, TypeGuard +from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import ( + GenAiProviderNameValues, +) from opentelemetry.util.genai.types import ( + BlobPart, FunctionToolDefinition, + GenericPart, + InputMessage, + MessagePart, + Modality, + OutputMessage, + ReasoningPart, + TextPart, + ToolCallRequestPart, + ToolCallResponsePart, ToolDefinition, + UriPart, ) +from opentelemetry.util.genai.utils import decode_base64, image_from_url if TYPE_CHECKING: from dspy.adapters.types.tool import Tool + from dspy.clients.base_lm import BaseLM + from dspy.core.types import LMBasePart, LMMessage, LMResponse from dspy.primitives.prediction import Prediction + from opentelemetry.util.genai.invocation import InferenceInvocation + SENTINEL_TOOL_NAMES: frozenset[str] = frozenset({"finish", "submit"}) +_KNOWN_PROVIDERS: dict[str, str] = { + "openai": GenAiProviderNameValues.OPENAI.value, + "anthropic": GenAiProviderNameValues.ANTHROPIC.value, + "cohere": GenAiProviderNameValues.COHERE.value, + "bedrock": GenAiProviderNameValues.AWS_BEDROCK.value, + "vertex_ai": GenAiProviderNameValues.GCP_VERTEX_AI.value, + "vertexai": GenAiProviderNameValues.GCP_VERTEX_AI.value, + "gemini": GenAiProviderNameValues.GCP_GEMINI.value, + "google": GenAiProviderNameValues.GCP_GEMINI.value, + "groq": GenAiProviderNameValues.GROQ.value, + "mistral": GenAiProviderNameValues.MISTRAL_AI.value, + "deepseek": GenAiProviderNameValues.DEEPSEEK.value, + "azure": GenAiProviderNameValues.AZURE_AI_OPENAI.value, + "watsonx": GenAiProviderNameValues.IBM_WATSONX_AI.value, + "perplexity": GenAiProviderNameValues.PERPLEXITY.value, + "xai": GenAiProviderNameValues.X_AI.value, +} + + +def _is_mapping(val: object) -> TypeGuard[Mapping[str, object]]: + return isinstance(val, Mapping) + + +def _is_sequence(val: object) -> TypeGuard[Sequence[object]]: + return isinstance(val, Sequence) and not isinstance(val, (str, bytes)) + + +def safe_int(val: object) -> int | None: + """Safely convert a value to int or return None.""" + if isinstance(val, (int, float, str, bytes)): + try: + return int(val) + except (ValueError, TypeError): + return None + return None + + +def safe_float(val: object) -> float | None: + """Safely convert a value to float or return None.""" + if isinstance(val, (int, float, str, bytes)): + try: + return float(val) + except (ValueError, TypeError): + return None + return None + + +def safe_stop_sequences(val: object) -> list[str] | None: + """Extract stop sequences as a list of strings.""" + if isinstance(val, str): + return [val] + if _is_sequence(val): + return [str(s) for s in val] + return None + + +_safe_int = safe_int +_safe_float = safe_float +_safe_stop_sequences = safe_stop_sequences + + +def parse_provider_and_model( + model_str: str | None, +) -> tuple[str | None, str | None]: + """Parse provider and model name from LiteLLM-style model string.""" + if not isinstance(model_str, str) or not model_str: + return None, None + model_str = model_str.strip().rstrip("/") + if "/" in model_str: + provider, model_name = model_str.split("/", 1) + provider = provider.strip().lower() + model_name = model_name.strip() + if "-" in provider: + provider = provider.split("-")[-1] + return provider, model_name + return None, model_str.strip() if model_str else None + + +def resolve_provider(instance: BaseLM) -> str: + """Resolve gen_ai.provider.name from a DSPy LM instance.""" + provider_obj = getattr(instance, "provider", None) + if provider_obj is not None: + cls_name = provider_obj.__class__.__name__.lower() + if "provider" in cls_name and cls_name != "provider": + p_name = cls_name.removesuffix("provider") + return _KNOWN_PROVIDERS.get(p_name) or p_name + + model = getattr(instance, "model", None) + if model is not None: + p_name, _ = parse_provider_and_model(str(model)) + if p_name: + return _KNOWN_PROVIDERS.get(p_name) or p_name + model_str = str(model).lower() + if model_str.startswith("gemini"): + return GenAiProviderNameValues.GCP_GEMINI.value + if model_str.startswith("claude"): + return GenAiProviderNameValues.ANTHROPIC.value + if model_str.startswith(("gpt-", "o1", "o3", "o4", "chatgpt")): + return GenAiProviderNameValues.OPENAI.value + + if instance.__class__.__name__ == "DummyLM" or model == "dummy": + return "dummy" + + return "unknown" + + +def resolve_request_model(instance: BaseLM) -> str | None: + """Resolve model name from a DSPy LM instance.""" + if (model_name := getattr(instance, "model_name", None)) is not None: + return str(model_name) + model = getattr(instance, "model", None) + if model is not None: + _, m_name = parse_provider_and_model(str(model)) + return m_name or str(model) + return None + + +def _extract_tool_call_part( + tc: object, + capture_content: bool = True, +) -> ToolCallRequestPart | None: + """Extract a ToolCallRequestPart from a tool call dictionary or object.""" + if _is_mapping(tc): + call_id = tc.get("id") + func = tc.get("function") + if _is_mapping(func): + name = str(func.get("name", "")) + args_raw = func.get("arguments") + else: + name = str(tc.get("name", "")) + args_raw = tc.get("args") or tc.get("arguments") + + args = None + if capture_content: + args = args_raw + if isinstance(args_raw, str): + try: + args = json.loads(args_raw) + except (json.JSONDecodeError, TypeError): + args = args_raw + return ToolCallRequestPart( + id=str(call_id) if call_id else None, + name=name, + arguments=args, + ) + if hasattr(tc, "name"): + call_id = getattr(tc, "id", None) + name = str(getattr(tc, "name", "")) + args = None + if capture_content: + args = getattr(tc, "args", None) or getattr(tc, "arguments", None) + return ToolCallRequestPart( + id=str(call_id) if call_id else None, + name=name, + arguments=args, + ) + return None + + +def _derive_modality( + p_type: str | None, media_type: str | None +) -> Modality | str: + if p_type in ("image", "image_url"): + return "image" + if p_type in ("audio", "input_audio"): + return "audio" + if p_type == "video": + return "video" + if p_type == "document": + return "document" + if media_type and "/" in media_type: + prefix = media_type.split("/")[0] + if prefix in ("image", "video", "audio", "document"): + return prefix + return "document" + + +def _to_bytes(data: object) -> bytes | None: + if isinstance(data, bytes): + return data + if isinstance(data, str): + return decode_base64(data) + return None + + +def _extract_part( + p: LMBasePart | Mapping[str, object] | str | object, + capture_content: bool = True, +) -> MessagePart | None: + """Extract a MessagePart from a part object or dictionary.""" + if isinstance(p, str): + return TextPart(content=p) + + if _is_mapping(p): + p_type = p.get("type") + p_type_str = str(p_type) if p_type is not None else None + + if p_type_str == "text" or ("text" in p and not p_type): + content = p.get("text") or p.get("content") + if isinstance(content, str): + return TextPart(content=content) + if p_type_str in ("thinking", "reasoning") or ( + "reasoning" in p and not p_type + ): + content = p.get("thinking") or p.get("reasoning") or p.get("text") + if isinstance(content, str): + return ReasoningPart(content=content) + if p_type_str == "refusal": + return GenericPart(type="refusal") + if p_type_str == "tool_call" or "function" in p: + return _extract_tool_call_part(p, capture_content=capture_content) + if ( + p_type_str == "tool_result" + or "tool_call_id" in p + or p.get("role") == "tool" + ): + call_id = p.get("tool_call_id") or p.get("call_id") or p.get("id") + content = p.get("content") or p.get("response") + return ToolCallResponsePart( + id=str(call_id) if call_id else None, + response=content, + ) + + # URL specified -> UriPart + url = p.get("url") or p.get("path") + if not url and p_type_str == "image_url": + img_url = p.get("image_url") + if _is_mapping(img_url): + url = img_url.get("url") + elif isinstance(img_url, str): + url = img_url + + media_type = p.get("media_type") or p.get("mime_type") + media_type_str = str(media_type) if media_type else None + + if url: + modality = _derive_modality(p_type_str, media_type_str) + if isinstance(url, str) and url.startswith("data:"): + part = image_from_url(url, modality=modality) + if part: + return part + return UriPart( + mime_type=media_type_str, + modality=modality, + uri=str(url), + ) + + # Inline data + data = p.get("data") + if data is None and p_type_str in ("audio", "input_audio"): + data = p.get("input_audio") + + if p_type_str in ("audio", "video") and data is not None: + # Inline audio and video payloads are omitted from telemetry spans + # because they can be massive. + return GenericPart(type=p_type_str) + + if data is not None: + content_bytes = _to_bytes(data) + if content_bytes is not None: + modality = _derive_modality(p_type_str, media_type_str) + return BlobPart( + mime_type=media_type_str, + modality=modality, + content=content_bytes, + ) + + # Fallback to GenericPart for any unhandled dictionary part + return GenericPart(type=p_type_str or "custom") + + # Object handling (DSPy LMBasePart subclasses, etc.) + p_type = getattr(p, "type", None) + p_type_str = str(p_type) if p_type is not None else None + + if p_type_str == "text": + text = getattr(p, "text", None) + if isinstance(text, str): + return TextPart(content=text) + elif p_type_str == "thinking": + text = getattr(p, "text", None) + if isinstance(text, str): + return ReasoningPart(content=text) + elif p_type_str == "refusal": + return GenericPart(type="refusal") + elif p_type_str == "tool_call" or ( + hasattr(p, "args") and hasattr(p, "name") + ): + return _extract_tool_call_part(p, capture_content=capture_content) + elif p_type_str == "tool_result" or hasattr(p, "call_id"): + call_id = getattr(p, "call_id", None) + content = getattr(p, "content", None) + return ToolCallResponsePart( + id=str(call_id) if call_id else None, + response=content, + ) + + # URL specified -> UriPart + url = getattr(p, "url", None) or getattr(p, "path", None) + media_type = getattr(p, "media_type", None) + media_type_str = str(media_type) if media_type else None -def extract_input_content(input_args: dict[str, Any]) -> str: + if url: + modality = _derive_modality(p_type_str, media_type_str) + if isinstance(url, str) and url.startswith("data:"): + part = image_from_url(url, modality=modality) + if part: + return part + return UriPart( + mime_type=media_type_str, + modality=modality, + uri=str(url), + ) + + # Inline data + data = getattr(p, "data", None) + if p_type_str in ("audio", "video") and data is not None: + # Inline audio and video payloads are omitted from telemetry spans + # because their large size causes excessive memory overhead and span bloat. + return GenericPart(type=p_type_str) + + if data is not None: + content_bytes = _to_bytes(data) + if content_bytes is not None: + modality = _derive_modality(p_type_str, media_type_str) + return BlobPart( + mime_type=media_type_str, + modality=modality, + content=content_bytes, + ) + + if p_type_str is None: + text = getattr(p, "text", None) + if isinstance(text, str): + return TextPart(content=text) + + # Fallback to GenericPart for any other part that we haven't covered + fallback_type = p_type_str or type(p).__name__ + return GenericPart(type=fallback_type) + + +def _extract_single_message( + msg: LMMessage | Mapping[str, object] | object, + capture_content: bool = True, +) -> InputMessage | None: + if _is_mapping(msg): + role = str(msg.get("role", "user")) + name = msg.get("name") + parts: list[MessagePart] = [] + + if role == "tool" or "tool_call_id" in msg: + call_id = msg.get("tool_call_id") or msg.get("id") + content = msg.get("content") + parts.append( + ToolCallResponsePart( + id=str(call_id) if call_id else None, + response=content, + ) + ) + else: + tool_calls = msg.get("tool_calls") + if _is_sequence(tool_calls): + for tc in tool_calls: + tcp = _extract_tool_call_part( + tc, capture_content=capture_content + ) + if tcp: + parts.append(tcp) + + reasoning = msg.get("reasoning_content") or msg.get("reasoning") + if isinstance(reasoning, str) and reasoning: + parts.append(ReasoningPart(content=reasoning)) + + raw_parts = msg.get("parts") + if _is_sequence(raw_parts): + for p in raw_parts: + extracted_p = _extract_part( + p, capture_content=capture_content + ) + if extracted_p: + parts.append(extracted_p) + + content = msg.get("content") + if _is_sequence(content): + for p in content: + extracted_p = _extract_part( + p, capture_content=capture_content + ) + if extracted_p: + parts.append(extracted_p) + elif isinstance(content, str): + parts.append(TextPart(content=content)) + elif content is not None and not parts: + parts.append(TextPart(content=str(content))) + + if parts: + return InputMessage( + role=role, + parts=parts, + name=str(name) if name is not None else None, + ) + return None + + role = getattr(msg, "role", None) + if role is None: + return None + name = getattr(msg, "name", None) + parts: list[MessagePart] = [] + raw_msg_parts = getattr(msg, "parts", None) + if _is_sequence(raw_msg_parts): + for p in raw_msg_parts: + extracted = _extract_part(p, capture_content=capture_content) + if extracted: + parts.append(extracted) + + msg_text = getattr(msg, "text", None) + if not parts and isinstance(msg_text, str) and msg_text: + parts.append(TextPart(content=msg_text)) + + if parts: + return InputMessage( + role=str(role), + parts=parts, + name=str(name) if name is not None else None, + ) + + return None + + +def extract_lm_input_messages( + args: tuple[object, ...], + kwargs: Mapping[str, object], + capture_content: bool = True, +) -> list[InputMessage]: + """Extract InputMessage list from LM call arguments.""" + request = kwargs.get("request") + if request is None and args and hasattr(args[0], "messages"): + request = args[0] + + if request is not None and hasattr(request, "messages"): + raw_req_msgs = getattr(request, "messages", None) + if _is_sequence(raw_req_msgs): + msgs: list[InputMessage] = [] + for m in raw_req_msgs: + extracted = _extract_single_message( + m, capture_content=capture_content + ) + if extracted: + msgs.append(extracted) + if msgs: + return msgs + + messages = kwargs.get("messages") + if _is_sequence(messages): + msgs: list[InputMessage] = [] + for m in messages: + extracted = _extract_single_message( + m, capture_content=capture_content + ) + if extracted: + msgs.append(extracted) + if msgs: + return msgs + + prompt = kwargs.get("prompt") + if prompt is None and args and isinstance(args[0], str): + prompt = args[0] + + if prompt is not None: + return [ + InputMessage(role="user", parts=[TextPart(content=str(prompt))]) + ] + + if args: + msgs: list[InputMessage] = [] + for a in args: + extracted = _extract_single_message( + a, capture_content=capture_content + ) + if extracted: + msgs.append(extracted) + if msgs: + return msgs + + return [] + + +def extract_lm_output_messages( + result: LMResponse | Sequence[Mapping[str, object] | str] | str, + finish_reason: str | None = None, + capture_content: bool = True, +) -> list[OutputMessage]: + """Extract OutputMessage list from LM call result.""" + if isinstance(result, str): + return [ + OutputMessage( + role="assistant", + parts=[TextPart(content=result)], + finish_reason=finish_reason, + ) + ] + + if not isinstance(result, Sequence): + msgs: list[OutputMessage] = [] + for out in result.outputs: + parts: list[MessagePart] = [] + if out.parts: + for p in out.parts: + extracted = _extract_part( + p, capture_content=capture_content + ) + if extracted: + parts.append(extracted) + if not parts: + if out.text: + parts.append(TextPart(content=out.text)) + if ( + isinstance(out.reasoning_content, str) + and out.reasoning_content + ): + parts.append(ReasoningPart(content=out.reasoning_content)) + for tc in out.tool_calls: + tcp = _extract_tool_call_part( + tc, capture_content=capture_content + ) + if tcp: + parts.append(tcp) + fr = out.finish_reason or finish_reason + if parts: + msgs.append( + OutputMessage( + role="assistant", + parts=parts, + finish_reason=str(fr) if fr else None, + ) + ) + return msgs + + msgs: list[OutputMessage] = [] + for item in result: + if isinstance(item, str): + msgs.append( + OutputMessage( + role="assistant", + parts=[TextPart(content=item)], + finish_reason=finish_reason, + ) + ) + else: + parts: list[MessagePart] = [] + raw_parts = item.get("parts") + if _is_sequence(raw_parts): + for p in raw_parts: + extracted_p = _extract_part( + p, capture_content=capture_content + ) + if extracted_p: + parts.append(extracted_p) + + content = item.get("content") or item.get("text") + if _is_sequence(content): + for p in content: + extracted_p = _extract_part( + p, capture_content=capture_content + ) + if extracted_p: + parts.append(extracted_p) + elif isinstance(content, str): + parts.append(TextPart(content=content)) + + reasoning = item.get("reasoning_content") or item.get("reasoning") + if isinstance(reasoning, str) and reasoning: + parts.append(ReasoningPart(content=reasoning)) + + tool_calls = item.get("tool_calls") + if _is_sequence(tool_calls): + for tc in tool_calls: + tcp = _extract_tool_call_part( + tc, capture_content=capture_content + ) + if tcp: + parts.append(tcp) + + if not parts: + parts.append(TextPart(content=str(item))) + + fr = item.get("finish_reason") or finish_reason + msgs.append( + OutputMessage( + role="assistant", + parts=parts, + finish_reason=str(fr) if fr else None, + ) + ) + return msgs + + +def apply_usage_to_invocation( + invocation: InferenceInvocation, + usage: Mapping[str, object], +) -> None: + """Apply token usage dictionary to an InferenceInvocation.""" + in_tokens = usage.get("prompt_tokens") + if in_tokens is None: + in_tokens = usage.get("input_tokens") + invocation.input_tokens = _safe_int(in_tokens) + + out_tokens = usage.get("completion_tokens") + if out_tokens is None: + out_tokens = usage.get("output_tokens") + invocation.output_tokens = _safe_int(out_tokens) + + invocation.thinking_tokens = _safe_int(usage.get("reasoning_tokens")) + invocation.cache_read_input_tokens = _safe_int( + usage.get("cache_read_tokens") + ) + invocation.cache_write_input_tokens = _safe_int( + usage.get("cache_write_tokens") + ) + + +def extract_input_content(input_args: Mapping[str, object]) -> str: """Extract input content string from agent invocation arguments.""" if not input_args: return "" if len(input_args) == 1: - val: Any = next(iter(input_args.values())) + val = next(iter(input_args.values())) if isinstance(val, str): return val if isinstance(val, (int, float, bool)): @@ -43,28 +681,30 @@ def extract_input_content(input_args: dict[str, Any]) -> str: def extract_output_content( result: Prediction | None, - signature: Any = None, + signature: object = None, ) -> str: """Extract output content string from a DSPy prediction result.""" if result is None: return "" - output_dict: dict[str, Any] = {} + output_dict: dict[str, object] = {} if hasattr(result, "items") and callable(getattr(result, "items")): try: - items_func: Callable[[], Any] = getattr(result, "items") - for k, v in cast(Iterable[tuple[Any, Any]], items_func()): + items_func: Callable[[], Iterable[tuple[object, object]]] = ( + getattr(result, "items") + ) + for k, v in items_func(): output_dict[str(k)] = v except Exception: pass if output_dict: - filtered_dict: dict[str, Any] = {} - output_fields: Any = ( + filtered_dict: dict[str, object] = {} + output_fields_raw = ( getattr(signature, "output_fields", None) if signature else None ) - if isinstance(output_fields, (dict, list, set, tuple)): - for key in cast(Iterable[Any], output_fields): + if _is_sequence(output_fields_raw): + for key in output_fields_raw: key_str = str(key) if key_str in output_dict: filtered_dict[key_str] = output_dict[key_str] @@ -79,7 +719,7 @@ def extract_output_content( if filtered_dict: if len(filtered_dict) == 1: - val: Any = next(iter(filtered_dict.values())) + val = next(iter(filtered_dict.values())) if isinstance(val, str): return val if isinstance(val, (int, float, bool)): @@ -97,8 +737,8 @@ def extract_output_content( def prepare_tool_definitions( - tools: Sequence[Tool | Callable[..., Any]] - | Mapping[str, Tool | Callable[..., Any]] + tools: Sequence[Tool | Callable[..., object]] + | Mapping[str, Tool | Callable[..., object]] | None, ) -> list[ToolDefinition] | None: """Prepare FunctionToolDefinition instances from a tools collection.""" diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py new file mode 100644 index 000000000..a7d6b3077 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py @@ -0,0 +1,92 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: language model (inference) execution for DSPy.""" + +from __future__ import annotations + +from typing import Any +from unittest import mock + +import dspy + +from opentelemetry.instrumentation.genai.dspy import DSPyInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test_util_genai.conformance import ( + ExpectedViolation, + Scenario, +) +from opentelemetry.test_util_genai.instrumentor import instrument + + +class FakeLM(dspy.LM): + """Test helper inheriting directly from dspy.LM.""" + + def __init__( + self, + responses: list[str] | None = None, + model: str = "openai/gpt-4o", + model_type: str = "chat", + **kwargs: Any, + ) -> None: + super().__init__( + model=model, api_key="fake-api-key", model_type=model_type, **kwargs + ) + self._responses = list(responses or ["Paris"]) + self._idx = 0 + + def forward(self, *args: Any, **kwargs: Any) -> Any: + resp_text = self._responses[self._idx % len(self._responses)] + self._idx += 1 + mock_resp = mock.MagicMock() + mock_choice = mock.MagicMock() + mock_choice.message.content = str(resp_text) + mock_choice.message.reasoning_content = None + mock_choice.message.tool_calls = None + mock_choice.finish_reason = "stop" + mock_resp.choices = [mock_choice] + mock_resp.model = "gpt-4o-2024-05-13" + mock_resp.id = "chatcmpl-123" + mock_resp.usage = { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + } + return mock_resp + + async def aforward(self, *args: Any, **kwargs: Any) -> Any: + return self.forward(*args, **kwargs) + + +class LMScenario(Scenario): + expected_spans = {"chat": 1} + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + ) + expected_violations = ( + ExpectedViolation( + advice_id="genai_expected_attribute_missing", + message_substring="server.address", + ), + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["Paris"]) + lm("What is the capital of France?") diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_conformance.py index cf1c76c3e..1e9b6d69e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_conformance.py @@ -20,6 +20,7 @@ run_conformance, ) +from .conformance.lm import LMScenario from .conformance.react import ReActScenario from .conformance.react_v2 import ReActV2Scenario from .conformance.retrieve import RetrieveScenario @@ -29,6 +30,7 @@ @pytest.mark.parametrize( "scenario", [ + pytest.param(LMScenario()), pytest.param(ReActScenario()), pytest.param(ReActV2Scenario()), pytest.param(RetrieveScenario()), diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py new file mode 100644 index 000000000..f2e25116a --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py @@ -0,0 +1,860 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for DSPy LM (inference) instrumentation.""" + +from __future__ import annotations + +import copy +import json +from typing import Any +from unittest import mock + +import dspy +import pytest +from dspy.utils import DummyLM + +from opentelemetry.instrumentation.genai.dspy import DSPyInstrumentor +from opentelemetry.instrumentation.genai.dspy.utils import ( + parse_provider_and_model, + resolve_provider, + resolve_request_model, +) +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) +from opentelemetry.semconv.attributes import error_attributes +from opentelemetry.test_util_genai.instrumentor import instrument +from opentelemetry.trace import StatusCode + + +class FakeLM(dspy.LM): + """Test helper inheriting directly from dspy.LM.""" + + def __init__( + self, + responses: list[str] | None = None, + model: str = "openai/gpt-4o", + model_type: str = "chat", + **kwargs: Any, + ) -> None: + super().__init__( + model=model, api_key="fake-api-key", model_type=model_type, **kwargs + ) + self._responses = list(responses or ["Paris"]) + self._idx = 0 + + def forward(self, *args: Any, **kwargs: Any) -> Any: + resp_text = self._responses[self._idx % len(self._responses)] + self._idx += 1 + mock_resp = mock.MagicMock() + mock_choice = mock.MagicMock() + mock_choice.message.content = str(resp_text) + mock_choice.finish_reason = "stop" + mock_resp.choices = [mock_choice] + mock_resp.model = "gpt-4o-2024-05-13" + mock_resp.id = "chatcmpl-123" + mock_resp.usage = { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + } + return mock_resp + + async def aforward(self, *args: Any, **kwargs: Any) -> Any: + return self.forward(*args, **kwargs) + + +def test_lm_call_sync_prompt( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["Paris"]) + res = lm("What is the capital of France?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gpt-4o" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + assert span.attributes[GenAI.GEN_AI_PROVIDER_NAME] == "openai" + assert span.attributes[GenAI.GEN_AI_REQUEST_MODEL] == "gpt-4o" + assert span.attributes[GenAI.GEN_AI_RESPONSE_MODEL] == "gpt-4o-2024-05-13" + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ("stop",) + assert span.attributes[GenAI.GEN_AI_USAGE_INPUT_TOKENS] == 10 + assert span.attributes[GenAI.GEN_AI_USAGE_OUTPUT_TOKENS] == 5 + + input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) + assert len(input_messages) == 1 + assert input_messages[0]["role"] == "user" + assert ( + input_messages[0]["parts"][0]["content"] + == "What is the capital of France?" + ) + + output_messages = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert len(output_messages) == 1 + assert output_messages[0]["role"] == "assistant" + assert output_messages[0]["parts"][0]["content"] == "Paris" + assert output_messages[0]["finish_reason"] == "stop" + + +def test_lm_call_sync_messages( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["Berlin"]) + res = lm(messages=[{"role": "user", "content": "Capital of Germany?"}]) + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) + assert len(input_messages) == 1 + assert input_messages[0]["role"] == "user" + assert ( + input_messages[0]["parts"][0]["content"] == "Capital of Germany?" + ) + + +def test_lm_call_request_parameters( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["42"]) + res = lm( + "Compute meaning", + temperature=0.7, + max_tokens=100, + top_p=0.9, + frequency_penalty=0.5, + presence_penalty=0.2, + stop=["STOP"], + seed=42, + n=2, + ) + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.attributes[GenAI.GEN_AI_REQUEST_TEMPERATURE] == 0.7 + assert span.attributes[GenAI.GEN_AI_REQUEST_MAX_TOKENS] == 100 + assert span.attributes[GenAI.GEN_AI_REQUEST_TOP_P] == 0.9 + assert span.attributes[GenAI.GEN_AI_REQUEST_FREQUENCY_PENALTY] == 0.5 + assert span.attributes[GenAI.GEN_AI_REQUEST_PRESENCE_PENALTY] == 0.2 + assert span.attributes[GenAI.GEN_AI_REQUEST_STOP_SEQUENCES] == ("STOP",) + assert span.attributes[GenAI.GEN_AI_REQUEST_SEED] == 42 + assert span.attributes[GenAI.GEN_AI_REQUEST_CHOICE_COUNT] == 2 + + +def test_lm_call_typed_response( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["Rome"]) + with dspy.context(experimental=True): + res = lm("Capital of Italy?") + + assert hasattr(res, "outputs") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.attributes[GenAI.GEN_AI_RESPONSE_MODEL] == "gpt-4o-2024-05-13" + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ("stop",) + assert span.attributes[GenAI.GEN_AI_USAGE_INPUT_TOKENS] == 10 + assert span.attributes[GenAI.GEN_AI_USAGE_OUTPUT_TOKENS] == 5 + + output_messages = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert len(output_messages) == 1 + assert output_messages[0]["role"] == "assistant" + assert output_messages[0]["finish_reason"] == "stop" + + +def test_lm_call_text_model_type( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Madrid"], model_type="text") + res = lm("Capital of Spain?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gpt-4o" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + + +def test_lm_call_gemini_provider( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Madrid"], model="gemini/gemini-1.5-pro") + res = lm("Capital of Spain?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gemini-1.5-pro" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + assert span.attributes[GenAI.GEN_AI_PROVIDER_NAME] == "gcp.gemini" + + +def test_lm_call_vertex_provider( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Madrid"], model="vertex_ai/gemini-1.5-pro") + res = lm("Capital of Spain?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gemini-1.5-pro" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + assert span.attributes[GenAI.GEN_AI_PROVIDER_NAME] == "gcp.vertex_ai" + + +def test_lm_call_model_name_only_provider( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Madrid"], model="gemini-1.5-flash") + res = lm("Capital of Spain?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gemini-1.5-flash" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + assert span.attributes[GenAI.GEN_AI_PROVIDER_NAME] == "gcp.gemini" + + +def test_lm_call_error( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM() + + with mock.patch.object( + lm, "forward", side_effect=RuntimeError("Model unreachable") + ): + with pytest.raises(RuntimeError, match="Model unreachable"): + lm("Hello") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "RuntimeError" + + +@pytest.mark.anyio +async def test_lm_acall_async( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["Tokyo"]) + res = await lm.acall("Capital of Japan?") + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.name == "chat gpt-4o" + assert span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "chat" + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ("stop",) + + input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) + assert input_messages[0]["parts"][0]["content"] == "Capital of Japan?" + + +@pytest.mark.anyio +async def test_lm_acall_async_messages( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = FakeLM(responses=["London"]) + res = await lm.acall( + messages=[{"role": "user", "content": "Capital of UK?"}] + ) + + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) + assert input_messages[0]["parts"][0]["content"] == "Capital of UK?" + + +@pytest.mark.anyio +async def test_lm_acall_async_typed_response( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Ottawa"]) + with dspy.context(experimental=True): + res = await lm.acall("Capital of Canada?") + + assert hasattr(res, "outputs") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ("stop",) + + +@pytest.mark.anyio +async def test_lm_acall_async_error( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM() + + with mock.patch.object( + lm, "aforward", side_effect=ConnectionError("Network timeout") + ): + with pytest.raises(ConnectionError, match="Network timeout"): + await lm.acall("Hello") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.status.status_code == StatusCode.ERROR + assert span.attributes[error_attributes.ERROR_TYPE] == "ConnectionError" + + +def test_lm_content_capture_disabled( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="false", + ): + lm = FakeLM(responses=["Canberra"]) + lm("Capital of Australia?") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert GenAI.GEN_AI_INPUT_MESSAGES not in span.attributes + assert GenAI.GEN_AI_OUTPUT_MESSAGES not in span.attributes + + +def test_dummy_lm_not_instrumented( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + dummy = DummyLM([{"answer": "Paris"}]) + res = dummy("What is the capital of France?") + assert res == ["[[ ## answer ## ]]\nParis"] + + spans = span_exporter.get_finished_spans() + assert len(spans) == 0 + + +def test_provider_and_model_resolution() -> None: + assert parse_provider_and_model("openai/gpt-4o") == ("openai", "gpt-4o") + assert parse_provider_and_model("anthropic/claude-3-5-sonnet") == ( + "anthropic", + "claude-3-5-sonnet", + ) + assert parse_provider_and_model("bedrock/anthropic.claude-3-sonnet") == ( + "bedrock", + "anthropic.claude-3-sonnet", + ) + assert parse_provider_and_model("text-completion-openai/davinci") == ( + "openai", + "davinci", + ) + assert parse_provider_and_model(None) == (None, None) + assert parse_provider_and_model("gpt-4o") == (None, "gpt-4o") + + class MockLM: + def __init__( + self, + model: str | None = None, + model_name: str | None = None, + provider: object | None = None, + ): + self.model = model + self.model_name = model_name + self.provider = provider + + class OpenAIProvider: + pass + + assert resolve_provider(MockLM("openai/gpt-4o")) == "openai" + assert resolve_request_model(MockLM("openai/gpt-4o")) == "gpt-4o" + + assert resolve_provider(MockLM("anthropic/claude-3")) == "anthropic" + assert resolve_request_model(MockLM("anthropic/claude-3")) == "claude-3" + + assert resolve_provider(MockLM("bedrock/anthropic.claude")) == "aws.bedrock" + assert ( + resolve_request_model(MockLM("bedrock/anthropic.claude")) + == "anthropic.claude" + ) + + assert resolve_provider(MockLM("vertex_ai/gemini-pro")) == "gcp.vertex_ai" + assert resolve_provider(MockLM("gemini/gemini-pro")) == "gcp.gemini" + + assert ( + resolve_provider(MockLM("custom-model", provider=OpenAIProvider())) + == "openai" + ) + assert ( + resolve_request_model(MockLM("custom-model", model_name="my-custom")) + == "my-custom" + ) + + assert resolve_provider(DummyLM([])) == "dummy" + assert resolve_request_model(DummyLM([])) == "dummy" + + +def test_copy_and_deepcopy_lm( + instrument_dspy: DSPyInstrumentor, + span_exporter, +) -> None: + lm = FakeLM(responses=["Copy test"]) + lm_copy = copy.copy(lm) + lm_deepcopy = copy.deepcopy(lm) + + assert lm_copy("test 1") is not None + assert lm_deepcopy("test 2") is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + for span in spans: + assert span.name == "chat gpt-4o" + + +def test_extract_message_rich_parts() -> None: + from dspy.core.types import ( + LMMessage, + LMResponse, + LMTextPart, + LMThinkingPart, + LMToolCallPart, + LMToolResultPart, + ) + from opentelemetry.instrumentation.genai.dspy.utils import ( + _extract_single_message, + extract_lm_output_messages, + ) + from opentelemetry.util.genai.types import ( + ReasoningPart, + TextPart, + ToolCallRequestPart, + ToolCallResponsePart, + ) + + # 1. Tool result message dict + tool_msg = _extract_single_message( + {"role": "tool", "tool_call_id": "call_123", "content": "42"} + ) + assert tool_msg is not None + assert tool_msg.role == "tool" + assert len(tool_msg.parts) == 1 + assert isinstance(tool_msg.parts[0], ToolCallResponsePart) + assert tool_msg.parts[0].id == "call_123" + assert tool_msg.parts[0].response == "42" + + # 2. Assistant message dict with tool calls and reasoning + asst_msg = _extract_single_message( + { + "role": "assistant", + "content": "Let me check.", + "reasoning_content": "Thinking about the question...", + "tool_calls": [ + { + "id": "call_abc", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"query": "weather"}', + }, + } + ], + } + ) + assert asst_msg is not None + assert asst_msg.role == "assistant" + assert len(asst_msg.parts) == 3 + assert isinstance(asst_msg.parts[0], ToolCallRequestPart) + assert asst_msg.parts[0].name == "lookup" + assert asst_msg.parts[0].arguments == {"query": "weather"} + assert isinstance(asst_msg.parts[1], ReasoningPart) + assert asst_msg.parts[1].content == "Thinking about the question..." + assert isinstance(asst_msg.parts[2], TextPart) + assert asst_msg.parts[2].content == "Let me check." + + # 3. LMMessage object with thinking and tool call + lm_msg = LMMessage( + role="assistant", + parts=[ + LMThinkingPart(text="Analyzing request"), + LMToolCallPart(id="call_99", name="fetch", args={"id": 1}), + LMTextPart(text="Done"), + ], + ) + extracted_lm_msg = _extract_single_message(lm_msg) + assert extracted_lm_msg is not None + assert len(extracted_lm_msg.parts) == 3 + assert isinstance(extracted_lm_msg.parts[0], ReasoningPart) + assert extracted_lm_msg.parts[0].content == "Analyzing request" + assert isinstance(extracted_lm_msg.parts[1], ToolCallRequestPart) + assert extracted_lm_msg.parts[1].id == "call_99" + assert extracted_lm_msg.parts[1].name == "fetch" + assert isinstance(extracted_lm_msg.parts[2], TextPart) + assert extracted_lm_msg.parts[2].content == "Done" + + # 4. extract_lm_output_messages with LMResponse + lm_resp = LMResponse.from_text("Result text") + lm_resp.outputs[0].parts.insert(0, LMThinkingPart(text="Output thinking")) + lm_resp.outputs[0].parts.append( + LMToolCallPart(id="call_out", name="calc", args={"a": 2}) + ) + output_msgs = extract_lm_output_messages(lm_resp, finish_reason="tool_calls") + assert len(output_msgs) == 1 + assert len(output_msgs[0].parts) == 3 + assert isinstance(output_msgs[0].parts[0], ReasoningPart) + assert output_msgs[0].parts[0].content == "Output thinking" + assert isinstance(output_msgs[0].parts[1], TextPart) + assert output_msgs[0].parts[1].content == "Result text" + assert isinstance(output_msgs[0].parts[2], ToolCallRequestPart) + assert output_msgs[0].parts[2].name == "calc" + assert output_msgs[0].finish_reason == "tool_calls" + + # 5. extract_lm_output_messages with legacy dict item + legacy_msgs = extract_lm_output_messages( + [ + { + "text": "Answer", + "reasoning_content": "Deep thought", + "tool_calls": [ + { + "id": "tc_1", + "function": {"name": "search", "arguments": "{}"}, + } + ], + "finish_reason": "stop", + } + ] + ) + assert len(legacy_msgs) == 1 + assert len(legacy_msgs[0].parts) == 3 + assert isinstance(legacy_msgs[0].parts[0], TextPart) + assert isinstance(legacy_msgs[0].parts[1], ReasoningPart) + assert isinstance(legacy_msgs[0].parts[2], ToolCallRequestPart) + + # 6. capture_content=False omits tool call arguments + no_content_msg = _extract_single_message( + { + "role": "assistant", + "tool_calls": [ + { + "id": "call_abc", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"query": "weather"}', + }, + } + ], + }, + capture_content=False, + ) + assert no_content_msg is not None + assert isinstance(no_content_msg.parts[0], ToolCallRequestPart) + assert no_content_msg.parts[0].arguments is None + + no_content_out = extract_lm_output_messages( + lm_resp, finish_reason="tool_calls", capture_content=False + ) + assert len(no_content_out) == 1 + assert isinstance(no_content_out[0].parts[2], ToolCallRequestPart) + assert no_content_out[0].parts[2].arguments is None + + +def test_extract_multimodal_and_generic_parts() -> None: + from dspy.core.types import ( + LMAudioPart, + LMBinaryPart, + LMCitationPart, + LMDocumentPart, + LMImagePart, + LMMessage, + LMRefusalPart, + LMResponse, + LMSourcePart, + LMVideoPart, + ) + from opentelemetry.instrumentation.genai.dspy.utils import ( + _extract_single_message, + extract_lm_output_messages, + ) + from opentelemetry.util.genai.types import ( + BlobPart, + GenericPart, + UriPart, + ) + + # 1. UriPart for URL specified (image, audio, video, document, dict) + lm_msg = LMMessage( + role="user", + parts=[ + LMImagePart(url="https://example.com/img.png", media_type="image/png"), + LMAudioPart(url="https://example.com/audio.mp3", media_type="audio/mp3"), + LMVideoPart(url="https://example.com/video.mp4", media_type="video/mp4"), + LMDocumentPart(url="https://example.com/doc.pdf", media_type="application/pdf"), + ], + ) + msg = _extract_single_message(lm_msg) + assert msg is not None + assert len(msg.parts) == 4 + assert msg.parts[0] == UriPart(mime_type="image/png", modality="image", uri="https://example.com/img.png") + assert msg.parts[1] == UriPart(mime_type="audio/mp3", modality="audio", uri="https://example.com/audio.mp3") + assert msg.parts[2] == UriPart(mime_type="video/mp4", modality="video", uri="https://example.com/video.mp4") + assert msg.parts[3] == UriPart(mime_type="application/pdf", modality="document", uri="https://example.com/doc.pdf") + + # Dict with image_url + dict_msg = _extract_single_message( + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": "https://example.com/pic.jpg"}} + ], + } + ) + assert dict_msg is not None + assert dict_msg.parts[0] == UriPart( + mime_type=None, modality="image", uri="https://example.com/pic.jpg" + ) + + # Data URL (data:;base64,...) decoded into BlobPart + data_url_dict_msg = _extract_single_message( + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": {"url": "data:image/png;base64,aGVsbG8="}, + } + ], + } + ) + assert data_url_dict_msg is not None + assert data_url_dict_msg.parts[0] == BlobPart( + mime_type="image/png", modality="image", content=b"hello" + ) + + data_url_part_msg = _extract_single_message( + LMMessage( + role="user", + parts=[LMImagePart(url="data:image/png;base64,aGVsbG8=")], + ) + ) + assert data_url_part_msg is not None + assert data_url_part_msg.parts[0] == BlobPart( + mime_type="image/png", modality="image", content=b"hello" + ) + + # 2. BlobPart for inline data (images, documents, binary, source) + blob_msg = LMMessage( + role="user", + parts=[ + LMImagePart(data="aGVsbG8=", media_type="image/png"), + LMDocumentPart(data="aGVsbG8=", media_type="application/pdf"), + LMBinaryPart(data="aGVsbG8=", media_type="application/octet-stream"), + ], + ) + extracted_blob = _extract_single_message(blob_msg) + assert extracted_blob is not None + assert len(extracted_blob.parts) == 3 + assert extracted_blob.parts[0] == BlobPart(mime_type="image/png", modality="image", content=b"hello") + assert extracted_blob.parts[1] == BlobPart(mime_type="application/pdf", modality="document", content=b"hello") + assert extracted_blob.parts[2] == BlobPart(mime_type="application/octet-stream", modality="document", content=b"hello") + + # LMSourcePart extracted directly and from dict + from opentelemetry.instrumentation.genai.dspy.utils import _extract_part + + assert _extract_part( + LMSourcePart(type="source", data="aGVsbG8=", media_type="text/html") + ) == BlobPart(mime_type="text/html", modality="document", content=b"hello") + source_dict_msg = _extract_single_message( + { + "role": "user", + "parts": [ + {"type": "source", "data": "aGVsbG8=", "media_type": "text/html"} + ], + } + ) + assert source_dict_msg is not None + assert source_dict_msg.parts[0] == BlobPart(mime_type="text/html", modality="document", content=b"hello") + + # 3. Inline audio / video omitted as GenericPart + av_msg = LMMessage( + role="user", + parts=[ + LMAudioPart(data="aGVsbG8=", media_type="audio/wav"), + LMVideoPart(data="aGVsbG8=", media_type="video/mp4"), + ], + ) + extracted_av = _extract_single_message(av_msg) + assert extracted_av is not None + assert len(extracted_av.parts) == 2 + assert extracted_av.parts[0] == GenericPart(type="audio") + assert extracted_av.parts[1] == GenericPart(type="video") + + # 4. GenericPart fallback for other parts (refusal, citation, custom dict, invalid b64) + refusal_msg = LMMessage( + role="assistant", + parts=[ + LMRefusalPart(text="I cannot fulfill this request."), + ], + ) + extracted_refusal = _extract_single_message(refusal_msg) + assert extracted_refusal is not None + assert extracted_refusal.parts[0] == GenericPart(type="refusal") + + refusal_dict_msg = _extract_single_message( + { + "role": "assistant", + "parts": [{"type": "refusal", "text": "I refuse."}], + } + ) + assert refusal_dict_msg is not None + assert refusal_dict_msg.parts[0] == GenericPart(type="refusal") + + invalid_b64_msg = _extract_single_message( + { + "role": "user", + "parts": [{"type": "image", "data": "invalid-base64-!@#$"}], + } + ) + assert invalid_b64_msg is not None + assert invalid_b64_msg.parts[0] == GenericPart(type="image") + + other_msg = LMMessage( + role="user", + parts=[ + LMCitationPart(text="cite", title="title"), + ], + ) + extracted_other = _extract_single_message(other_msg) + assert extracted_other is not None + assert len(extracted_other.parts) == 1 + assert extracted_other.parts[0] == GenericPart(type="citation") + + custom_dict_msg = _extract_single_message( + { + "role": "user", + "content": [ + {"type": "custom_extension", "val": 123}, + ], + } + ) + assert custom_dict_msg is not None + assert len(custom_dict_msg.parts) == 1 + assert custom_dict_msg.parts[0] == GenericPart(type="custom_extension") + + # 5. Output messages with multimodal and generic parts in LMResponse + lm_resp = LMResponse.from_text("Text output") + lm_resp.outputs[0].parts.append( + LMImagePart(url="https://example.com/out.png", media_type="image/png") + ) + lm_resp.outputs[0].parts.append( + LMCitationPart(text="source", title="title") + ) + out_msgs = extract_lm_output_messages(lm_resp) + assert len(out_msgs) == 1 + assert len(out_msgs[0].parts) == 3 + assert out_msgs[0].parts[1] == UriPart(mime_type="image/png", modality="image", uri="https://example.com/out.png") + assert out_msgs[0].parts[2] == GenericPart(type="citation") + + From 9d15fd8a0c631f6f9d52d663d63b9f022d329a5c Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Fri, 11 Sep 2026 19:58:05 +0000 Subject: [PATCH 2/3] Add changelog entry --- .../.changelog/683.added | 1 + .../tests/conformance/lm.py | 5 +- .../tests/test_lm.py | 101 +++++++++++++----- 3 files changed, 81 insertions(+), 26 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/683.added diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/683.added b/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/683.added new file mode 100644 index 000000000..bdd379b3a --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/683.added @@ -0,0 +1 @@ +Add instrumentation for ``dspy.LM``. diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py index a7d6b3077..2c6a784ea 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py @@ -32,7 +32,10 @@ def __init__( **kwargs: Any, ) -> None: super().__init__( - model=model, api_key="fake-api-key", model_type=model_type, **kwargs + model=model, + api_key="fake-api-key", + model_type=model_type, + **kwargs, ) self._responses = list(responses or ["Paris"]) self._idx = 0 diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py index f2e25116a..756defdad 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py @@ -42,7 +42,10 @@ def __init__( **kwargs: Any, ) -> None: super().__init__( - model=model, api_key="fake-api-key", model_type=model_type, **kwargs + model=model, + api_key="fake-api-key", + model_type=model_type, + **kwargs, ) self._responses = list(responses or ["Paris"]) self._idx = 0 @@ -139,9 +142,7 @@ def test_lm_call_sync_messages( input_messages = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) assert len(input_messages) == 1 assert input_messages[0]["role"] == "user" - assert ( - input_messages[0]["parts"][0]["content"] == "Capital of Germany?" - ) + assert input_messages[0]["parts"][0]["content"] == "Capital of Germany?" def test_lm_call_request_parameters( @@ -474,7 +475,9 @@ class OpenAIProvider: assert resolve_provider(MockLM("anthropic/claude-3")) == "anthropic" assert resolve_request_model(MockLM("anthropic/claude-3")) == "claude-3" - assert resolve_provider(MockLM("bedrock/anthropic.claude")) == "aws.bedrock" + assert ( + resolve_provider(MockLM("bedrock/anthropic.claude")) == "aws.bedrock" + ) assert ( resolve_request_model(MockLM("bedrock/anthropic.claude")) == "anthropic.claude" @@ -520,8 +523,8 @@ def test_extract_message_rich_parts() -> None: LMTextPart, LMThinkingPart, LMToolCallPart, - LMToolResultPart, ) + from opentelemetry.instrumentation.genai.dspy.utils import ( _extract_single_message, extract_lm_output_messages, @@ -599,7 +602,9 @@ def test_extract_message_rich_parts() -> None: lm_resp.outputs[0].parts.append( LMToolCallPart(id="call_out", name="calc", args={"a": 2}) ) - output_msgs = extract_lm_output_messages(lm_resp, finish_reason="tool_calls") + output_msgs = extract_lm_output_messages( + lm_resp, finish_reason="tool_calls" + ) assert len(output_msgs) == 1 assert len(output_msgs[0].parts) == 3 assert isinstance(output_msgs[0].parts[0], ReasoningPart) @@ -674,6 +679,7 @@ def test_extract_multimodal_and_generic_parts() -> None: LMSourcePart, LMVideoPart, ) + from opentelemetry.instrumentation.genai.dspy.utils import ( _extract_single_message, extract_lm_output_messages, @@ -688,26 +694,53 @@ def test_extract_multimodal_and_generic_parts() -> None: lm_msg = LMMessage( role="user", parts=[ - LMImagePart(url="https://example.com/img.png", media_type="image/png"), - LMAudioPart(url="https://example.com/audio.mp3", media_type="audio/mp3"), - LMVideoPart(url="https://example.com/video.mp4", media_type="video/mp4"), - LMDocumentPart(url="https://example.com/doc.pdf", media_type="application/pdf"), + LMImagePart( + url="https://example.com/img.png", media_type="image/png" + ), + LMAudioPart( + url="https://example.com/audio.mp3", media_type="audio/mp3" + ), + LMVideoPart( + url="https://example.com/video.mp4", media_type="video/mp4" + ), + LMDocumentPart( + url="https://example.com/doc.pdf", media_type="application/pdf" + ), ], ) msg = _extract_single_message(lm_msg) assert msg is not None assert len(msg.parts) == 4 - assert msg.parts[0] == UriPart(mime_type="image/png", modality="image", uri="https://example.com/img.png") - assert msg.parts[1] == UriPart(mime_type="audio/mp3", modality="audio", uri="https://example.com/audio.mp3") - assert msg.parts[2] == UriPart(mime_type="video/mp4", modality="video", uri="https://example.com/video.mp4") - assert msg.parts[3] == UriPart(mime_type="application/pdf", modality="document", uri="https://example.com/doc.pdf") + assert msg.parts[0] == UriPart( + mime_type="image/png", + modality="image", + uri="https://example.com/img.png", + ) + assert msg.parts[1] == UriPart( + mime_type="audio/mp3", + modality="audio", + uri="https://example.com/audio.mp3", + ) + assert msg.parts[2] == UriPart( + mime_type="video/mp4", + modality="video", + uri="https://example.com/video.mp4", + ) + assert msg.parts[3] == UriPart( + mime_type="application/pdf", + modality="document", + uri="https://example.com/doc.pdf", + ) # Dict with image_url dict_msg = _extract_single_message( { "role": "user", "content": [ - {"type": "image_url", "image_url": {"url": "https://example.com/pic.jpg"}} + { + "type": "image_url", + "image_url": {"url": "https://example.com/pic.jpg"}, + } ], } ) @@ -750,15 +783,25 @@ def test_extract_multimodal_and_generic_parts() -> None: parts=[ LMImagePart(data="aGVsbG8=", media_type="image/png"), LMDocumentPart(data="aGVsbG8=", media_type="application/pdf"), - LMBinaryPart(data="aGVsbG8=", media_type="application/octet-stream"), + LMBinaryPart( + data="aGVsbG8=", media_type="application/octet-stream" + ), ], ) extracted_blob = _extract_single_message(blob_msg) assert extracted_blob is not None assert len(extracted_blob.parts) == 3 - assert extracted_blob.parts[0] == BlobPart(mime_type="image/png", modality="image", content=b"hello") - assert extracted_blob.parts[1] == BlobPart(mime_type="application/pdf", modality="document", content=b"hello") - assert extracted_blob.parts[2] == BlobPart(mime_type="application/octet-stream", modality="document", content=b"hello") + assert extracted_blob.parts[0] == BlobPart( + mime_type="image/png", modality="image", content=b"hello" + ) + assert extracted_blob.parts[1] == BlobPart( + mime_type="application/pdf", modality="document", content=b"hello" + ) + assert extracted_blob.parts[2] == BlobPart( + mime_type="application/octet-stream", + modality="document", + content=b"hello", + ) # LMSourcePart extracted directly and from dict from opentelemetry.instrumentation.genai.dspy.utils import _extract_part @@ -770,12 +813,18 @@ def test_extract_multimodal_and_generic_parts() -> None: { "role": "user", "parts": [ - {"type": "source", "data": "aGVsbG8=", "media_type": "text/html"} + { + "type": "source", + "data": "aGVsbG8=", + "media_type": "text/html", + } ], } ) assert source_dict_msg is not None - assert source_dict_msg.parts[0] == BlobPart(mime_type="text/html", modality="document", content=b"hello") + assert source_dict_msg.parts[0] == BlobPart( + mime_type="text/html", modality="document", content=b"hello" + ) # 3. Inline audio / video omitted as GenericPart av_msg = LMMessage( @@ -854,7 +903,9 @@ def test_extract_multimodal_and_generic_parts() -> None: out_msgs = extract_lm_output_messages(lm_resp) assert len(out_msgs) == 1 assert len(out_msgs[0].parts) == 3 - assert out_msgs[0].parts[1] == UriPart(mime_type="image/png", modality="image", uri="https://example.com/out.png") + assert out_msgs[0].parts[1] == UriPart( + mime_type="image/png", + modality="image", + uri="https://example.com/out.png", + ) assert out_msgs[0].parts[2] == GenericPart(type="citation") - - From 478bccb1bf41193fdfc54c2be7216e61110c3a0d Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Mon, 14 Sep 2026 17:59:26 +0000 Subject: [PATCH 3/3] Address copilot --- .../instrumentation/genai/dspy/patch.py | 135 +++++++++++---- .../instrumentation/genai/dspy/utils.py | 28 +++- .../tests/test_lm.py | 155 ++++++++++++++++++ 3 files changed, 281 insertions(+), 37 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py index 06ee08e05..4a9ecef7f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/patch.py @@ -8,6 +8,7 @@ import inspect import sys from collections.abc import Awaitable, Callable, Mapping, Sequence +from contextvars import ContextVar from copy import copy, deepcopy from importlib import import_module from typing import TYPE_CHECKING, Any, cast @@ -61,6 +62,10 @@ _REACT_V2_MODULE = "dspy.predict.react_v2" _REACT_V2_CLASS = "ReActV2" +_current_lm_history_entry: ContextVar[Any] = ContextVar( + "_current_lm_history_entry", default=None +) + if TYPE_CHECKING: _BoundFunctionWrapper = BoundFunctionWrapper[Any, Any] @@ -241,6 +246,12 @@ def patch_dspy(handler: TelemetryHandler) -> None: f"{lm_name}.acall", _lm_acall(handler), ) + if hasattr(dspy.LM, "update_history"): + _wrap_function( + lm_module, + f"{lm_name}.update_history", + _lm_update_history(), + ) def unpatch_dspy() -> None: @@ -266,6 +277,8 @@ def unpatch_dspy() -> None: if hasattr(dspy, "LM"): unwrap(dspy.LM, "__call__") unwrap(dspy.LM, "acall") + if hasattr(dspy.LM, "update_history"): + unwrap(dspy.LM, "update_history") def _start_lm_invocation( @@ -317,11 +330,29 @@ def _get_field(obj: Any, key: str) -> Any: return getattr(obj, key, None) +def _lm_update_history() -> Callable[..., Any]: + """Capture the history entry in context to isolate concurrent calls from shared instance.history.""" + + def traced_method( + wrapped: Callable[..., Any], + instance: LM, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + entry = args[0] if args else kwargs.get("entry") + if entry is not None: + _current_lm_history_entry.set(entry) + return wrapped(*args, **kwargs) + + return traced_method + + def _set_lm_invocation_response( handler: TelemetryHandler, invocation: InferenceInvocation, instance: LM, result: LMResponse | list[dict[str, Any] | str], + history_entry: Any = None, ) -> None: if not isinstance(result, list): if result.model: @@ -344,27 +375,31 @@ def _set_lm_invocation_response( return # DSPy 3.x LM calls return a legacy list by default unless experimental=True - # or an LMRequest is used. DSPy appends each call's metadata to instance.history, - # so history[-1] corresponds to the invocation that just finished. - finish_reason: str | None = None - history: Sequence[Mapping[str, Any]] | None = getattr( - instance, "history", None - ) - if isinstance(history, Sequence) and history: - last_entry = history[-1] - resp_model = last_entry.get("response_model") or last_entry.get( - "model" + # or an LMRequest is used. When available, use history_entry captured from + # LM.update_history for per-call isolation; fall back to instance.history[-1]. + entry = history_entry + if entry is None: + history: Sequence[Mapping[str, Any]] | None = getattr( + instance, "history", None + ) + if isinstance(history, Sequence) and history: + entry = history[-1] + + choice_finish_reasons: list[str | None] | None = None + if entry is not None: + resp_model = _get_field(entry, "response_model") or _get_field( + entry, "model" ) if resp_model: invocation.response_model_name = str(resp_model) - usage = last_entry.get("usage") + usage = _get_field(entry, "usage") if isinstance(usage, Mapping): apply_usage_to_invocation( invocation, cast(Mapping[str, Any], usage) ) - resp_obj: Any = last_entry.get("response") + resp_obj: Any = _get_field(entry, "response") if resp_obj is not None: resp_id = _get_field(resp_obj, "id") if resp_id: @@ -372,14 +407,35 @@ def _set_lm_invocation_response( choices = _get_field(resp_obj, "choices") if isinstance(choices, Sequence) and choices: - fr = _get_field(choices[0], "finish_reason") - if fr: - finish_reason = str(fr) - invocation.finish_reasons = [finish_reason] + choice_finish_reasons = [] + for choice in cast(Sequence[object], choices): + fr = _get_field(choice, "finish_reason") + choice_finish_reasons.append( + str(fr) if fr is not None else None + ) + invocation_finish_reasons = [ + fr for fr in choice_finish_reasons if fr is not None + ] + if invocation_finish_reasons: + invocation.finish_reasons = invocation_finish_reasons + else: + outputs = _get_field(resp_obj, "outputs") + if isinstance(outputs, Sequence) and outputs: + choice_finish_reasons = [] + for out in cast(Sequence[object], outputs): + fr = _get_field(out, "finish_reason") + choice_finish_reasons.append( + str(fr) if fr is not None else None + ) + invocation_finish_reasons = [ + fr for fr in choice_finish_reasons if fr is not None + ] + if invocation_finish_reasons: + invocation.finish_reasons = invocation_finish_reasons if handler.should_capture_content(): invocation.output_messages = extract_lm_output_messages( - result, finish_reason=finish_reason + result, finish_reasons=choice_finish_reasons ) @@ -391,10 +447,22 @@ def traced_method( kwargs: dict[str, Any], ) -> Any: invocation = _start_lm_invocation(handler, instance, args, kwargs) - with invocation: - result = wrapped(*args, **kwargs) - _set_lm_invocation_response(handler, invocation, instance, result) - return result + # Isolate per-call history metadata from concurrent calls on the shared LM instance. + token = _current_lm_history_entry.set(None) + try: + with invocation: + result = wrapped(*args, **kwargs) + history_entry = _current_lm_history_entry.get() + _set_lm_invocation_response( + handler, + invocation, + instance, + result, + history_entry=history_entry, + ) + return result + finally: + _current_lm_history_entry.reset(token) return traced_method @@ -407,10 +475,22 @@ async def traced_method( kwargs: dict[str, Any], ) -> Any: invocation = _start_lm_invocation(handler, instance, args, kwargs) - with invocation: - result = await wrapped(*args, **kwargs) - _set_lm_invocation_response(handler, invocation, instance, result) - return result + # Isolate per-call history metadata from concurrent calls on the shared LM instance. + token = _current_lm_history_entry.set(None) + try: + with invocation: + result = await wrapped(*args, **kwargs) + history_entry = _current_lm_history_entry.get() + _set_lm_invocation_response( + handler, + invocation, + instance, + result, + history_entry=history_entry, + ) + return result + finally: + _current_lm_history_entry.reset(token) return traced_method @@ -606,10 +686,7 @@ def _extract_retrieval_k( if k is None and hasattr(instance, "k"): k = getattr(instance, "k", None) if k is not None: - try: - return int(k) - except (ValueError, TypeError): - return None + return _safe_int(k) return None diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py index 702ae0fe5..dd53762b1 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/src/opentelemetry/instrumentation/genai/dspy/utils.py @@ -71,7 +71,7 @@ def safe_int(val: object) -> int | None: if isinstance(val, (int, float, str, bytes)): try: return int(val) - except (ValueError, TypeError): + except (ValueError, TypeError, OverflowError): return None return None @@ -81,7 +81,7 @@ def safe_float(val: object) -> float | None: if isinstance(val, (int, float, str, bytes)): try: return float(val) - except (ValueError, TypeError): + except (ValueError, TypeError, OverflowError): return None return None @@ -528,21 +528,23 @@ def extract_lm_input_messages( def extract_lm_output_messages( result: LMResponse | Sequence[Mapping[str, object] | str] | str, finish_reason: str | None = None, + finish_reasons: Sequence[str | None] | None = None, capture_content: bool = True, ) -> list[OutputMessage]: """Extract OutputMessage list from LM call result.""" if isinstance(result, str): + fr = finish_reasons[0] if finish_reasons else finish_reason return [ OutputMessage( role="assistant", parts=[TextPart(content=result)], - finish_reason=finish_reason, + finish_reason=str(fr) if fr else None, ) ] if not isinstance(result, Sequence): msgs: list[OutputMessage] = [] - for out in result.outputs: + for idx, out in enumerate(result.outputs): parts: list[MessagePart] = [] if out.parts: for p in out.parts: @@ -565,7 +567,12 @@ def extract_lm_output_messages( ) if tcp: parts.append(tcp) - fr = out.finish_reason or finish_reason + fallback_fr = ( + finish_reasons[idx] + if finish_reasons and idx < len(finish_reasons) + else finish_reason + ) + fr = out.finish_reason or fallback_fr if parts: msgs.append( OutputMessage( @@ -577,13 +584,18 @@ def extract_lm_output_messages( return msgs msgs: list[OutputMessage] = [] - for item in result: + for idx, item in enumerate(result): + fallback_fr = ( + finish_reasons[idx] + if finish_reasons and idx < len(finish_reasons) + else finish_reason + ) if isinstance(item, str): msgs.append( OutputMessage( role="assistant", parts=[TextPart(content=item)], - finish_reason=finish_reason, + finish_reason=str(fallback_fr) if fallback_fr else None, ) ) else: @@ -624,7 +636,7 @@ def extract_lm_output_messages( if not parts: parts.append(TextPart(content=str(item))) - fr = item.get("finish_reason") or finish_reason + fr = item.get("finish_reason") or fallback_fr msgs.append( OutputMessage( role="assistant", diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py index 756defdad..67841209f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py @@ -909,3 +909,158 @@ def test_extract_multimodal_and_generic_parts() -> None: uri="https://example.com/out.png", ) assert out_msgs[0].parts[2] == GenericPart(type="citation") + + +def test_safe_numeric_overflow() -> None: + from opentelemetry.instrumentation.genai.dspy.utils import ( + _safe_float, + _safe_int, + ) + + assert _safe_int(float("inf")) is None + assert _safe_int(float("-inf")) is None + assert _safe_int(float("nan")) is None + assert _safe_float(10**1000) is None + + +def test_lm_multiple_choices_different_finish_reasons( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + class MultiChoiceLM(dspy.LM): + def __init__(self) -> None: + super().__init__( + model="openai/gpt-4o", + api_key="fake-api-key", + model_type="chat", + ) + + def forward(self, *args: Any, **kwargs: Any) -> Any: + resp = mock.MagicMock() + c1 = mock.MagicMock() + c1.message.content = "First choice" + c1.message.reasoning_content = None + c1.message.tool_calls = None + c1.finish_reason = "stop" + + c2 = mock.MagicMock() + c2.message.content = "Second choice" + c2.message.reasoning_content = None + c2.message.tool_calls = None + c2.finish_reason = "length" + + resp.choices = [c1, c2] + resp.model = "gpt-4o-2024-05-13" + resp.id = "chatcmpl-multi" + resp.usage = { + "prompt_tokens": 10, + "completion_tokens": 20, + "total_tokens": 30, + } + return resp + + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = MultiChoiceLM() + res = lm("Tell me two things", n=2) + + assert len(res) == 2 + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + + assert span.attributes[GenAI.GEN_AI_RESPONSE_FINISH_REASONS] == ( + "stop", + "length", + ) + output_messages = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert len(output_messages) == 2 + assert output_messages[0]["parts"][0]["content"] == "First choice" + assert output_messages[0]["finish_reason"] == "stop" + assert output_messages[1]["parts"][0]["content"] == "Second choice" + assert output_messages[1]["finish_reason"] == "length" + + +@pytest.mark.anyio +async def test_lm_concurrent_history_isolation( + tracer_provider: TracerProvider, + logger_provider: LoggerProvider, + meter_provider: MeterProvider, + span_exporter, +) -> None: + import asyncio + + class InterleavedLM(dspy.LM): + def __init__(self) -> None: + super().__init__( + model="openai/gpt-4o", + api_key="fake-api-key", + model_type="chat", + ) + + async def aforward(self, *args: Any, **kwargs: Any) -> Any: + call_id = kwargs.get("call_id") + delay = 0.05 if call_id == "call-1" else 0.01 + await asyncio.sleep(delay) + + resp = mock.MagicMock() + c = mock.MagicMock() + c.message.content = f"Result for {call_id}" + c.message.reasoning_content = None + c.message.tool_calls = None + c.finish_reason = "stop" + + resp.choices = [c] + resp.model = f"gpt-4o-{call_id}" + resp.id = f"id-{call_id}" + resp.usage = { + "prompt_tokens": 5 if call_id == "call-1" else 10, + "completion_tokens": 5, + } + return resp + + with instrument( + DSPyInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + lm = InterleavedLM() + res1, res2 = await asyncio.gather( + lm.acall(prompt="Prompt 1", call_id="call-1"), + lm.acall(prompt="Prompt 2", call_id="call-2"), + ) + + assert res1 is not None + assert res2 is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 2 + + # Map spans by their prompt content + span_by_prompt = {} + for s in spans: + input_msgs = json.loads(s.attributes[GenAI.GEN_AI_INPUT_MESSAGES]) + prompt_text = input_msgs[0]["parts"][0]["content"] + span_by_prompt[prompt_text] = s + + span1 = span_by_prompt["Prompt 1"] + span2 = span_by_prompt["Prompt 2"] + + assert span1.attributes[GenAI.GEN_AI_RESPONSE_MODEL] == "gpt-4o-call-1" + assert span1.attributes[GenAI.GEN_AI_USAGE_INPUT_TOKENS] == 5 + out1 = json.loads(span1.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert out1[0]["parts"][0]["content"] == "Result for call-1" + + assert span2.attributes[GenAI.GEN_AI_RESPONSE_MODEL] == "gpt-4o-call-2" + assert span2.attributes[GenAI.GEN_AI_USAGE_INPUT_TOKENS] == 10 + out2 = json.loads(span2.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]) + assert out2[0]["parts"][0]["content"] == "Result for call-2"