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/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 e9b91f54a..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
@@ -7,7 +7,8 @@
import inspect
import sys
-from collections.abc import Awaitable, Callable, Sequence
+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
@@ -21,13 +22,22 @@
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 (
+ InferenceInvocation,
LocalAgentInvocation,
RetrievalInvocation,
ToolInvocation,
@@ -40,6 +50,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
@@ -50,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]
@@ -217,6 +233,26 @@ 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),
+ )
+ if hasattr(dspy.LM, "update_history"):
+ _wrap_function(
+ lm_module,
+ f"{lm_name}.update_history",
+ _lm_update_history(),
+ )
+
def unpatch_dspy() -> None:
"""Remove patches from DSPy classes."""
@@ -238,6 +274,226 @@ 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")
+ if hasattr(dspy.LM, "update_history"):
+ unwrap(dspy.LM, "update_history")
+
+
+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 _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:
+ 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. 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 = _get_field(entry, "usage")
+ if isinstance(usage, Mapping):
+ apply_usage_to_invocation(
+ invocation, cast(Mapping[str, Any], usage)
+ )
+
+ resp_obj: Any = _get_field(entry, "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:
+ 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_reasons=choice_finish_reasons
+ )
+
+
+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)
+ # 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
+
+
+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)
+ # 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
+
def _extract_tool_arguments(
instance: Tool,
@@ -430,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 8224b3d3e..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
@@ -7,26 +7,676 @@
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, OverflowError):
+ 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, OverflowError):
+ 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
+
+ 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
-def extract_input_content(input_args: dict[str, Any]) -> str:
+ 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,
+ 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=str(fr) if fr else None,
+ )
+ ]
+
+ if not isinstance(result, Sequence):
+ msgs: list[OutputMessage] = []
+ for idx, out in enumerate(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)
+ 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(
+ role="assistant",
+ parts=parts,
+ finish_reason=str(fr) if fr else None,
+ )
+ )
+ return msgs
+
+ msgs: list[OutputMessage] = []
+ 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=str(fallback_fr) if fallback_fr else None,
+ )
+ )
+ 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 fallback_fr
+ 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 +693,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 +731,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 +749,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..2c6a784ea
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/conformance/lm.py
@@ -0,0 +1,95 @@
+# 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..67841209f
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_lm.py
@@ -0,0 +1,1066 @@
+# 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,
+ )
+
+ 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")
+
+
+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"