diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/775.changed b/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/775.changed new file mode 100644 index 000000000..82a22a205 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/.changelog/775.changed @@ -0,0 +1 @@ +Use the shared RetrievalDocument model for retrieval results, with null IDs and scores for text-only passages, instead of capturing passage text. 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 ac393519b..4b5ccebc0 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 @@ -34,6 +34,7 @@ from opentelemetry.util.genai.types import ( InputMessage, OutputMessage, + RetrievalDocument, TextPart, ) from opentelemetry.util.genai.utils import bind_arguments @@ -485,7 +486,8 @@ def _set_retrieval_invocation_documents( if passages is None: return - invocation.documents = [{"content": str(psg)} for psg in passages] + # Retrieve returns passage text, without document IDs or scores. + invocation.documents = [RetrievalDocument() for _ in passages] def _retrieve_forward( diff --git a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_retrieve.py b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_retrieve.py index 6bb01b79a..7fa2056ce 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_retrieve.py +++ b/instrumentation/opentelemetry-instrumentation-genai-dspy/tests/test_retrieve.py @@ -8,11 +8,15 @@ import copy import json from typing import Any +from unittest.mock import Mock import dspy import pytest from opentelemetry.instrumentation.genai.dspy import DSPyInstrumentor +from opentelemetry.instrumentation.genai.dspy.patch import ( + _set_retrieval_invocation_documents, +) from opentelemetry.sdk._logs import LoggerProvider from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.trace import TracerProvider @@ -25,6 +29,9 @@ from opentelemetry.semconv.attributes import error_attributes from opentelemetry.test_util_genai.instrumentor import instrument from opentelemetry.trace import StatusCode +from opentelemetry.util.genai.handler import TelemetryHandler +from opentelemetry.util.genai.invocation import RetrievalInvocation +from opentelemetry.util.genai.types import RetrievalDocument _GEN_AI_RETRIEVAL_TOP_K = "gen_ai.retrieval.top_k" @@ -49,11 +56,13 @@ def __call__(self, query: str, k: int = 3, **kwargs: Any) -> list[Any]: return [_DummyPassage(f"Passage {i} for {query}") for i in range(k)] +@pytest.mark.parametrize("k", [0, 1, 3]) def test_sync_retrieve_execution( tracer_provider: TracerProvider, logger_provider: LoggerProvider, meter_provider: MeterProvider, span_exporter: InMemorySpanExporter, + k: int, ) -> None: rm = _DummyRM() dspy.settings.configure(rm=rm) @@ -65,9 +74,9 @@ def test_sync_retrieve_execution( meter_provider=meter_provider, content_capture="SPAN_ONLY", ): - retrieve = dspy.Retrieve(k=3) + retrieve = dspy.Retrieve(k=k) res = retrieve("What is OpenTelemetry?") - assert len(res.passages) == 3 + assert len(res.passages) == k spans = span_exporter.get_finished_spans() assert len(spans) == 1 @@ -77,7 +86,7 @@ def test_sync_retrieve_execution( assert span.status.status_code == StatusCode.UNSET attrs = span.attributes or {} assert attrs.get(GenAI.GEN_AI_OPERATION_NAME) == "retrieval" - assert attrs.get(_GEN_AI_RETRIEVAL_TOP_K) == 3 + assert attrs.get(_GEN_AI_RETRIEVAL_TOP_K) == k assert isinstance(attrs.get(_GEN_AI_RETRIEVAL_TOP_K), int) assert ( attrs.get(GenAI.GEN_AI_RETRIEVAL_QUERY_TEXT) @@ -87,10 +96,22 @@ def test_sync_retrieve_execution( docs_attr = attrs.get(GenAI.GEN_AI_RETRIEVAL_DOCUMENTS) assert isinstance(docs_attr, str) docs = json.loads(docs_attr) - assert len(docs) == 3 - assert docs[0] == {"content": "Passage 0 for What is OpenTelemetry?"} - assert docs[1] == {"content": "Passage 1 for What is OpenTelemetry?"} - assert docs[2] == {"content": "Passage 2 for What is OpenTelemetry?"} + assert docs == [{"id": None, "score": None}] * k + assert res.passages == [ + f"Passage {i} for What is OpenTelemetry?" for i in range(k) + ] + + +def test_retrieval_documents_do_not_stringify_passages() -> None: + class Passage: + def __str__(self) -> str: + raise AssertionError("passage text must not be read") + + handler = Mock(spec=TelemetryHandler) + handler.should_capture_content.return_value = True + invocation = Mock(spec=RetrievalInvocation) + _set_retrieval_invocation_documents(handler, invocation, [Passage()]) + assert invocation.documents == [RetrievalDocument()] def test_retrieve_forward_direct_call( diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/775.changed b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/775.changed new file mode 100644 index 000000000..bcbe6a7e7 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/775.changed @@ -0,0 +1 @@ +Use the shared RetrievalDocument model to capture only document IDs and scores, with null for unavailable fields, instead of document text. diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/README.rst b/instrumentation/opentelemetry-instrumentation-genai-langchain/README.rst index e6d8dd870..57685a6c2 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/README.rst @@ -17,7 +17,7 @@ application: agent name, id, description, and conversation/session id when available. * **Tool spans** for tool calls made during a run. * **Retrieval spans** for retriever invocations, capturing the query and retrieved - documents (including id, content, and relevance scores when available). + document IDs and relevance scores when available. The spans nest to reflect the graph, so a single graph invocation produces a workflow span with the agent, tool, and model calls it triggered as children. @@ -102,11 +102,14 @@ Retrieval Spans and Document Scores ----------------------------------- When invoking LangChain retrievers (e.g., vectorstores, knowledge bases, or contextual compression retrievers), -retrieval spans are recorded with the query and retrieved document metadata. +retrieval spans are recorded with the query and retrieved document IDs and scores. When message content capture is enabled (``OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=SPAN_ONLY`` or ``SPAN_AND_EVENT``), the retrieved documents are serialized into the -``gen_ai.retrieval.documents`` span attribute as a JSON array of objects with ``id`` and ``content``. +``gen_ai.retrieval.documents`` span attribute using the shared ``RetrievalDocument`` +model, as a JSON array of objects with only ``id`` and ``score``. +Document text (previously recorded as ``content``) and metadata are not captured. +Query text capture is unchanged. When available, relevance and similarity scores are captured in each document object under ``score``: @@ -117,7 +120,8 @@ When available, relevance and similarity scores are captured in each document ob * **Duck-typed / custom documents**: extracted from a top-level ``score`` attribute or mapping key. If a document has no score, or if the score is non-numeric or non-finite (``NaN``, ``Infinity``), -the ``score`` key is omitted to ensure RFC 8259 JSON compliance. +``score`` is recorded as JSON ``null`` to ensure RFC 8259 JSON compliance. +Missing document IDs are also recorded as ``null``. Configuration ------------- diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py index 62027de49..07776230e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py @@ -56,6 +56,7 @@ InputMessage, MessagePart, OutputMessage, + RetrievalDocument, Role, TextPart, ToolCallRequestPart, @@ -142,32 +143,18 @@ def _extract_document_score(doc: Any) -> float | int | None: return None -def _document_to_dict(doc: Any) -> dict[str, Any]: - """Convert a Document, duck-typed document object, or Mapping to a dict. - - Extracts content (checking page_content first, then content), id, - and conditionally score if present and numeric. - """ +def _document_to_retrieval_document(doc: object) -> RetrievalDocument: + """Extract only the standard document ID and relevance score.""" if isinstance(doc, Mapping): - doc_map = cast(Mapping[str, Any], doc) - content = doc_map.get("page_content") - if content is None: - content = doc_map.get("content") + doc_map = cast(Mapping[str, object], doc) doc_id = doc_map.get("id") else: - content = getattr(doc, "page_content", None) - if content is None: - content = getattr(doc, "content", None) doc_id = getattr(doc, "id", None) - doc_dict: dict[str, Any] = { - "content": content, - "id": doc_id, - } - score = _extract_document_score(doc) - if score is not None: - doc_dict["score"] = score - return doc_dict + return RetrievalDocument( + id=doc_id if isinstance(doc_id, str) else None, + score=_extract_document_score(doc), + ) class OpenTelemetryLangChainCallbackHandler(BaseCallbackHandler): @@ -851,7 +838,7 @@ def on_retriever_end( if self._telemetry_handler.should_capture_content(): invocation.documents = [ - _document_to_dict(doc) for doc in documents + _document_to_retrieval_document(doc) for doc in documents ] invocation.stop() self._invocation_manager.delete_invocation_state(run_id) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py index 595595703..b77907741 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py @@ -38,7 +38,7 @@ from opentelemetry.instrumentation.genai.langchain.callback_handler import ( OpenTelemetryLangChainCallbackHandler, - _document_to_dict, + _document_to_retrieval_document, _extract_document_score, ) from opentelemetry.instrumentation.genai.langchain.utils import ( @@ -64,6 +64,7 @@ FilePart, InputMessage, OutputMessage, + RetrievalDocument, TextPart, ToolCallRequestPart, UriPart, @@ -1635,7 +1636,7 @@ def test_invocation_stopped(self): retrieval_inv.stop.assert_called_once() - def test_documents_set_from_page_content(self): + def test_documents_use_shared_model_without_content(self): handler, _, retrieval_inv = _make_handler_with_retrieval() run_id = _run_id() @@ -1648,10 +1649,8 @@ def test_documents_set_from_page_content(self): handler.on_retriever_end(documents=docs, run_id=run_id) assigned = retrieval_inv.documents - assert len(assigned) == 2 - assert assigned[0]["content"] == "doc one" - assert "source" not in assigned[0] - assert assigned[1]["content"] == "doc two" + assert assigned == [RetrievalDocument(), RetrievalDocument()] + assert all(isinstance(doc, RetrievalDocument) for doc in assigned) def test_document_id_included_when_present(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1662,7 +1661,7 @@ def test_document_id_included_when_present(self): handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[doc], run_id=run_id) - assert retrieval_inv.documents[0]["id"] == "doc-123" + assert retrieval_inv.documents[0].id == "doc-123" def test_document_id_none_when_absent(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1673,7 +1672,7 @@ def test_document_id_none_when_absent(self): handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[doc], run_id=run_id) - assert retrieval_inv.documents[0]["id"] is None + assert retrieval_inv.documents[0].id is None def test_state_cleaned_up_after_end(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1704,7 +1703,7 @@ def test_documents_set_when_content_enabled(self): handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=docs, run_id=run_id) - assert retrieval_inv.documents[0]["content"] == "visible" + assert retrieval_inv.documents == [RetrievalDocument()] def test_unknown_run_id_does_not_raise(self): handler, _, _ = _make_handler_with_retrieval() @@ -1723,7 +1722,7 @@ class DuckDoc: handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[DuckDoc()], run_id=run_id) - assert retrieval_inv.documents[0]["score"] == 0.85 + assert retrieval_inv.documents[0].score == 0.85 def test_document_score_from_metadata(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1736,7 +1735,7 @@ def test_document_score_from_metadata(self): handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[doc], run_id=run_id) - assert retrieval_inv.documents[0]["score"] == 0.92 + assert retrieval_inv.documents[0].score == 0.92 def test_document_score_precedence(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1751,7 +1750,7 @@ class DuckDoc: handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[DuckDoc()], run_id=run_id) - assert retrieval_inv.documents[0]["score"] == 0.9 + assert retrieval_inv.documents[0].score == 0.9 def test_document_score_fallback_to_metadata_when_attr_is_none(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1766,7 +1765,7 @@ class DuckDoc: handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[DuckDoc()], run_id=run_id) - assert retrieval_inv.documents[0]["score"] == 0.77 + assert retrieval_inv.documents[0].score == 0.77 def test_document_score_zero_preserved(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1786,10 +1785,8 @@ class DuckDoc: documents=[DuckDoc(), doc_meta], run_id=run_id ) - assert "score" in retrieval_inv.documents[0] - assert retrieval_inv.documents[0]["score"] == 0.0 - assert "score" in retrieval_inv.documents[1] - assert retrieval_inv.documents[1]["score"] == 0 + assert retrieval_inv.documents[0].score == 0.0 + assert retrieval_inv.documents[1].score == 0 @pytest.mark.parametrize( "invalid_score", @@ -1804,7 +1801,7 @@ def test_document_score_non_numeric_ignored(self, invalid_score): handler.on_retriever_start(serialized={}, query="q", run_id=run_id) handler.on_retriever_end(documents=[doc], run_id=run_id) - assert "score" not in retrieval_inv.documents[0] + assert retrieval_inv.documents[0].score is None @pytest.mark.parametrize( "non_finite_score", @@ -1836,7 +1833,7 @@ class DuckDocAttr: ) for item in retrieval_inv.documents: - assert "score" not in item + assert item.score is None def test_document_score_plain_dict_and_duck_typed(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1887,38 +1884,14 @@ class CustomDuckDocNonePageContentFallback: run_id=run_id, ) - assigned = retrieval_inv.documents - assert len(assigned) == 6 - assert assigned[0] == { - "content": "dict content 1", - "id": "dict-1", - "score": 0.88, - } - assert assigned[1] == { - "content": "dict content 2", - "id": None, - "score": 0.72, - } - assert assigned[2] == { - "content": "dict content fallback when page_content is None", - "id": "dict-3", - "score": 0.64, - } - assert assigned[3] == { - "content": "duck content", - "id": "duck-1", - "score": 0.95, - } - assert assigned[4] == { - "content": "duck content fallback", - "id": "duck-2", - "score": 0.81, - } - assert assigned[5] == { - "content": "duck content fallback when page_content is None", - "id": "duck-3", - "score": 0.55, - } + assert retrieval_inv.documents == [ + RetrievalDocument(id="dict-1", score=0.88), + RetrievalDocument(score=0.72), + RetrievalDocument(id="dict-3", score=0.64), + RetrievalDocument(id="duck-1", score=0.95), + RetrievalDocument(id="duck-2", score=0.81), + RetrievalDocument(id="duck-3", score=0.55), + ] def test_document_score_non_mapping_metadata(self): handler, _, retrieval_inv = _make_handler_with_retrieval() @@ -1949,7 +1922,7 @@ class DuckDocNoneMeta: ) for item in retrieval_inv.documents: - assert "score" not in item + assert item.score is None class TestExtractDocumentScore: @@ -2082,62 +2055,78 @@ def test_no_score(self): assert _extract_document_score({"metadata": "str"}) is None -class TestDocumentToDict: - def test_document_with_page_content_and_score(self): +class TestDocumentToRetrievalDocument: + def test_document_with_id_and_score(self): doc = Document( page_content="doc content", id="d1", metadata={"score": 0.85} ) - assert _document_to_dict(doc) == { - "content": "doc content", - "id": "d1", - "score": 0.85, - } + assert _document_to_retrieval_document(doc) == RetrievalDocument( + id="d1", score=0.85 + ) - def test_duck_typed_with_content_fallback_missing_page_content(self): + def test_duck_typed_content_is_ignored(self): class DuckNoPageContent: content = "fallback content" id = "d2" score = 0.9 - assert _document_to_dict(DuckNoPageContent()) == { - "content": "fallback content", - "id": "d2", - "score": 0.9, - } + assert _document_to_retrieval_document( + DuckNoPageContent() + ) == RetrievalDocument(id="d2", score=0.9) - def test_duck_typed_with_content_fallback_none_page_content(self): + def test_duck_typed_none_page_content_is_ignored(self): class DuckNonePageContent: page_content = None content = "fallback content when page_content is None" id = "d3" score = 0.75 - assert _document_to_dict(DuckNonePageContent()) == { - "content": "fallback content when page_content is None", - "id": "d3", - "score": 0.75, - } + assert _document_to_retrieval_document( + DuckNonePageContent() + ) == RetrievalDocument(id="d3", score=0.75) - def test_mapping_with_content_fallback_missing_page_content(self): + def test_mapping_content_is_ignored(self): doc_map = {"content": "mapping fallback", "id": "m1", "score": 0.8} - assert _document_to_dict(doc_map) == { - "content": "mapping fallback", - "id": "m1", - "score": 0.8, - } + assert _document_to_retrieval_document(doc_map) == RetrievalDocument( + id="m1", score=0.8 + ) - def test_mapping_with_content_fallback_none_page_content(self): + def test_mapping_none_page_content_is_ignored(self): doc_map = { "page_content": None, "content": "mapping fallback when page_content is None", "id": "m2", "score": 0.7, } - assert _document_to_dict(doc_map) == { - "content": "mapping fallback when page_content is None", - "id": "m2", - "score": 0.7, - } + assert _document_to_retrieval_document(doc_map) == RetrievalDocument( + id="m2", score=0.7 + ) + + def test_does_not_access_document_content(self) -> None: + class LazyDocument: + id = "lazy" + score = 0.0 + + @property + def page_content(self) -> str: + raise AssertionError("document content must not be read") + + @property + def content(self) -> str: + raise AssertionError("document content must not be read") + + assert _document_to_retrieval_document( + LazyDocument() + ) == RetrievalDocument(id="lazy", score=0.0) + + @pytest.mark.parametrize( + "doc_id", [None, 42, True, ["doc"], {"id": "doc"}] + ) + def test_non_string_ids_are_not_recorded(self, doc_id: object) -> None: + assert ( + _document_to_retrieval_document({"id": doc_id}) + == RetrievalDocument() + ) def test_non_finite_scores_omitted(self): doc_nan = Document(page_content="c", metadata={"score": math.nan}) @@ -2146,9 +2135,11 @@ def test_non_finite_scores_omitted(self): page_content="c", metadata={"score": float("-inf")} ) - assert _document_to_dict(doc_nan) == {"content": "c", "id": None} - assert _document_to_dict(doc_inf) == {"content": "c", "id": None} - assert _document_to_dict(doc_neginf) == {"content": "c", "id": None} + assert _document_to_retrieval_document(doc_nan) == RetrievalDocument() + assert _document_to_retrieval_document(doc_inf) == RetrievalDocument() + assert ( + _document_to_retrieval_document(doc_neginf) == RetrievalDocument() + ) class TestOnRetrieverError: diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_retriever.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_retriever.py index 27375f3b7..3a999bee5 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_retriever.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_retriever.py @@ -140,10 +140,11 @@ def test_retrieval_span_attributes( == "What is the capital of France?" ) docs_attr = attrs[gen_ai_attributes.GEN_AI_RETRIEVAL_DOCUMENTS] - assert docs_attr is not None - assert "Paris is the capital of France." in docs_attr - assert "doc-1" in docs_attr - assert "Berlin is the capital of Germany." in docs_attr + assert type(docs_attr) is str + assert json.loads(docs_attr) == [ + {"id": "doc-1", "score": None}, + {"id": None, "score": None}, + ] else: assert gen_ai_attributes.GEN_AI_RETRIEVAL_QUERY_TEXT not in attrs assert gen_ai_attributes.GEN_AI_RETRIEVAL_DOCUMENTS not in attrs @@ -369,7 +370,7 @@ def test_document_without_id_in_span_content( docs_attr = spans[0].attributes[ gen_ai_attributes.GEN_AI_RETRIEVAL_DOCUMENTS ] - assert "no id here" in docs_attr + assert json.loads(docs_attr) == [{"id": None, "score": None}] instrumentor.uninstrument() @@ -406,7 +407,7 @@ def test_document_metadata_not_in_span_content( docs_attr = spans[0].attributes[ gen_ai_attributes.GEN_AI_RETRIEVAL_DOCUMENTS ] - assert "text" in docs_attr + assert "content" not in docs_attr assert "wiki" not in docs_attr assert "alice" not in docs_attr finally: @@ -476,7 +477,7 @@ def test_retriever_documents_with_attribute_score( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert parsed[0]["content"] == "text" + assert set(parsed[0]) == {"id", "score"} assert parsed[0]["id"] == "doc-1" assert parsed[0]["score"] == 0.85 finally: @@ -515,7 +516,7 @@ def test_retriever_documents_with_metadata_score( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert parsed[0]["content"] == "text" + assert set(parsed[0]) == {"id", "score"} assert parsed[0]["id"] == "doc-2" assert parsed[0]["score"] == 0.92 finally: @@ -630,7 +631,7 @@ def test_retriever_documents_without_score( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert "score" not in parsed[0] + assert parsed[0]["score"] is None finally: instrumentor.uninstrument() @@ -711,7 +712,7 @@ def test_retriever_documents_with_non_finite_scores( parsed = json.loads(docs_attr) assert len(parsed) == 4 for item in parsed: - assert "score" not in item + assert item["score"] is None finally: instrumentor.uninstrument() @@ -752,7 +753,7 @@ def test_retriever_documents_with_metadata_relevance_score( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert parsed[0]["content"] == "contextual content" + assert set(parsed[0]) == {"id", "score"} assert parsed[0]["id"] == "doc-rerank-1" assert parsed[0]["score"] == 0.88 finally: @@ -807,6 +808,7 @@ async def test_async_retriever_documents_with_scores( ] parsed = json.loads(docs_attr) assert len(parsed) == 3 + assert all(set(doc) == {"id", "score"} for doc in parsed) assert parsed[0]["id"] == "doc-kb" assert parsed[0]["score"] == 0.95 @@ -815,7 +817,7 @@ async def test_async_retriever_documents_with_scores( assert parsed[1]["score"] == 0.82 assert parsed[2]["id"] == "doc-plain" - assert "score" not in parsed[2] + assert parsed[2]["score"] is None finally: instrumentor.uninstrument() @@ -868,10 +870,9 @@ async def test_retriever_grounded_knowledge_base_sync_and_async( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert ( - parsed[0]["content"] - == "Amazon Bedrock Knowledge Bases provides managed RAG." - ) + assert set(parsed[0]) == {"id", "score"} + assert type(parsed[0]["id"]) is str + assert type(parsed[0]["score"]) is float assert parsed[0]["id"] == "kb-result-1" assert parsed[0]["score"] == 0.89 finally: @@ -926,7 +927,9 @@ async def test_retriever_grounded_contextual_compression_sync_and_async( ] parsed = json.loads(docs_attr) assert len(parsed) == 1 - assert parsed[0]["content"] == "High relevance chunk after reranking." + assert set(parsed[0]) == {"id", "score"} + assert type(parsed[0]["id"]) is str + assert type(parsed[0]["score"]) is float assert parsed[0]["id"] == "rerank-1" assert parsed[0]["score"] == 0.94 finally: diff --git a/instrumentation/opentelemetry-instrumentation-genai-llama-index/.changelog/775.changed b/instrumentation/opentelemetry-instrumentation-genai-llama-index/.changelog/775.changed new file mode 100644 index 000000000..d838afb52 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-llama-index/.changelog/775.changed @@ -0,0 +1 @@ +Use the shared RetrievalDocument model to capture only document IDs and scores, with null for unavailable scores, instead of node text. diff --git a/instrumentation/opentelemetry-instrumentation-genai-llama-index/src/opentelemetry/instrumentation/genai/llama_index/_handler.py b/instrumentation/opentelemetry-instrumentation-genai-llama-index/src/opentelemetry/instrumentation/genai/llama_index/_handler.py index 967b9ab6d..8d6e41581 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-llama-index/src/opentelemetry/instrumentation/genai/llama_index/_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-llama-index/src/opentelemetry/instrumentation/genai/llama_index/_handler.py @@ -5,6 +5,8 @@ import contextvars import inspect +import logging +import math from base64 import b64decode from binascii import Error as BinasciiError from collections.abc import Callable, Mapping, MutableMapping, Sequence @@ -61,6 +63,7 @@ Modality, OutputMessage, ReasoningPart, + RetrievalDocument, Role, SystemInstructionPart, TextPart, @@ -69,6 +72,8 @@ UriPart, ) +_logger = logging.getLogger(__name__) + _ToolExecutionAttributes = tuple[str, str | None] _AGENT_TOOL_ATTRIBUTES: ContextVar[ dict[str, _ToolExecutionAttributes] | None @@ -327,25 +332,27 @@ def _retrieval_top_k(retriever: BaseRetriever) -> int | None: def _retrieval_documents( result: object, -) -> list[dict[str, Any]] | None: +) -> list[RetrievalDocument] | None: """Convert retrieved LlamaIndex nodes to semconv document objects.""" if not isinstance(result, Sequence): return None candidates = cast(Sequence[object], result) - documents: list[dict[str, Any]] = [] + documents: list[RetrievalDocument] = [] for candidate in candidates: if not isinstance(candidate, NodeWithScore): continue try: - document: dict[str, Any] = { - "id": candidate.node_id, - "content": candidate.node.get_content(), - } - if candidate.score is not None: - document["score"] = candidate.score - documents.append(document) + document_id = candidate.node_id + score = candidate.score + if score is not None and not math.isfinite(score): + score = None except BaseException: + _logger.warning( + "Failed to extract retrieval document attributes", + exc_info=True, + ) continue + documents.append(RetrievalDocument(id=document_id, score=score)) # Preserve [] for a genuine empty result, but omit the attribute when a # non-empty result could not be converted into semantic-convention docs. if documents: diff --git a/instrumentation/opentelemetry-instrumentation-genai-llama-index/tests/test_retrieval.py b/instrumentation/opentelemetry-instrumentation-genai-llama-index/tests/test_retrieval.py index 65687f25f..d7cb1083b 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-llama-index/tests/test_retrieval.py +++ b/instrumentation/opentelemetry-instrumentation-genai-llama-index/tests/test_retrieval.py @@ -4,6 +4,7 @@ from __future__ import annotations import json +import logging import pytest from llama_index.core.base.base_retriever import BaseRetriever @@ -19,6 +20,7 @@ error_attributes as ErrorAttributes, ) from opentelemetry.trace import SpanKind, StatusCode +from opentelemetry.util.genai.types import RetrievalDocument _GEN_AI_RETRIEVAL_TOP_K = "gen_ai.retrieval.top_k" @@ -28,21 +30,45 @@ def test_unconvertible_retrieval_results_are_omitted() -> None: assert _retrieval_documents([object()]) is None +@pytest.mark.parametrize("score", [None, 0.0, 0.9]) +def test_retrieval_documents_use_shared_model_without_reading_content( + score: float | None, +) -> None: + class LazyTextNode(TextNode): + def get_content(self, *args: object, **kwargs: object) -> str: + raise AssertionError("node content must not be read") + + nodes = [NodeWithScore(node=LazyTextNode(id_="doc-1"), score=score)] + assert _retrieval_documents(nodes) == [ + RetrievalDocument(id="doc-1", score=score) + ] + + class _Retriever(BaseRetriever): - def __init__(self, error: BaseException | None = None) -> None: + def __init__( + self, + error: BaseException | None = None, + *, + nodes: list[NodeWithScore] | None = None, + ) -> None: super().__init__() self.similarity_top_k = 2 self._error = error + self._nodes = ( + nodes + if nodes is not None + else [ + NodeWithScore( + node=TextNode(id_="doc-1", text="Paris is in France."), + score=0.9, + ) + ] + ) def _retrieve(self, query_bundle: QueryBundle) -> list[NodeWithScore]: if self._error: raise self._error - return [ - NodeWithScore( - node=TextNode(id_="doc-1", text="Paris is in France."), - score=0.9, - ) - ] + return self._nodes def _span(exporter): @@ -51,6 +77,108 @@ def _span(exporter): return spans[0] +@pytest.mark.parametrize("attribute", ["node_id", "score"]) +@pytest.mark.parametrize("error_type", [RuntimeError, BaseException]) +def test_retrieval_documents_skip_broken_accessors( + attribute: str, + error_type: type[BaseException], + caplog: pytest.LogCaptureFixture, +) -> None: + class BrokenNodeWithScore(NodeWithScore): + def __getattribute__(self, name: str) -> object: + if name == attribute: + raise error_type("broken document accessor") + return super().__getattribute__(name) + + broken = BrokenNodeWithScore(node=TextNode(id_="broken"), score=0.5) + valid = NodeWithScore(node=TextNode(id_="valid"), score=0.0) + with caplog.at_level(logging.WARNING): + assert _retrieval_documents([valid, broken, valid]) == [ + RetrievalDocument(id="valid", score=0.0), + RetrievalDocument(id="valid", score=0.0), + ] + assert _retrieval_documents([broken]) is None + assert len(caplog.records) == 2 + assert all( + record.message == "Failed to extract retrieval document attributes" + for record in caplog.records + ) + + +@pytest.mark.parametrize("is_async", [False, True]) +@pytest.mark.asyncio +async def test_broken_document_does_not_change_retrieval_result( + span_exporter, instrument_llama_index_with_content, is_async: bool +) -> None: + class BrokenNodeWithScore(NodeWithScore): + @property + def node_id(self) -> str: + raise RuntimeError("broken document ID") + + broken = BrokenNodeWithScore(node=TextNode(id_="broken"), score=0.5) + valid = NodeWithScore(node=TextNode(id_="valid"), score=0.9) + retriever = _Retriever(nodes=[broken, valid]) + result = ( + await retriever.aretrieve("query") + if is_async + else retriever.retrieve("query") + ) + + assert len(result) == 2 + assert result[0] is broken + assert result[1] is valid + span = _span(span_exporter) + assert span.status.status_code == StatusCode.UNSET + assert ErrorAttributes.ERROR_TYPE not in span.attributes + assert json.loads( + span.attributes[GenAIAttributes.GEN_AI_RETRIEVAL_DOCUMENTS] + ) == [{"id": "valid", "score": 0.9}] + + +@pytest.mark.parametrize("is_async", [False, True]) +@pytest.mark.parametrize( + "score, expected", + [ + (float("nan"), None), + (float("inf"), None), + (float("-inf"), None), + (None, None), + (0.0, 0.0), + (-0.5, -0.5), + (0.9, 0.9), + ], +) +@pytest.mark.asyncio +async def test_retrieval_scores_serialize_as_valid_json( + span_exporter, + instrument_llama_index_with_content, + is_async: bool, + score: float | None, + expected: float | None, +) -> None: + node = NodeWithScore(node=TextNode(id_="doc-1"), score=score) + retriever = _Retriever(nodes=[node]) + result = ( + await retriever.aretrieve("query") + if is_async + else retriever.retrieve("query") + ) + assert result[0] is node + span = _span(span_exporter) + raw = span.attributes[GenAIAttributes.GEN_AI_RETRIEVAL_DOCUMENTS] + assert type(raw) is str + documents = json.loads( + raw, + parse_constant=lambda value: pytest.fail( + f"Non-standard JSON constant: {value}" + ), + ) + assert documents == [{"id": "doc-1", "score": expected}] + if expected is not None: + assert type(documents[0]["score"]) is float + assert span.status.status_code == StatusCode.UNSET + + def test_retrieval_captures_documents_and_query( span_exporter, instrument_llama_index_with_content ) -> None: @@ -68,9 +196,9 @@ def test_retrieval_captures_documents_and_query( assert type(documents) is str assert top_k == 2 assert query_text == "Where is Paris?" - assert json.loads(documents) == [ - {"id": "doc-1", "content": "Paris is in France.", "score": 0.9} - ] + assert json.loads(documents) == [{"id": "doc-1", "score": 0.9}] + assert type(json.loads(documents)[0]["id"]) is str + assert type(json.loads(documents)[0]["score"]) is float def test_retrieval_query_bundle_captures_query( @@ -120,9 +248,7 @@ async def test_async_retrieval_captures_query_and_documents( assert type(query_text) is str assert type(documents) is str assert query_text == "Where is Paris?" - assert json.loads(documents) == [ - {"id": "doc-1", "content": "Paris is in France.", "score": 0.9} - ] + assert json.loads(documents) == [{"id": "doc-1", "score": 0.9}] def test_sync_retrieval_error_is_unchanged( diff --git a/util/opentelemetry-util-genai/.changelog/775.added b/util/opentelemetry-util-genai/.changelog/775.added new file mode 100644 index 000000000..644bb7f54 --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/775.added @@ -0,0 +1 @@ +Add the shared RetrievalDocument model with optional id and score fields; retain legacy mapping support on retrieval invocations. diff --git a/util/opentelemetry-util-genai/README.rst b/util/opentelemetry-util-genai/README.rst index 0f13f46fe..729621246 100644 --- a/util/opentelemetry-util-genai/README.rst +++ b/util/opentelemetry-util-genai/README.rst @@ -63,6 +63,26 @@ also accept raw strings should use ``Modality | str`` rather than ``Modality`` alone. Custom modalities remain strings, not additional enum members. +Retrieval Documents +------------------- + +Set ``RetrievalInvocation.documents`` using +``opentelemetry.util.genai.types.RetrievalDocument`` objects: + +.. code-block:: python + + from opentelemetry.util.genai.types import RetrievalDocument + + with handler.retrieval(data_source_id="my-index") as invocation: + invocation.documents = [RetrievalDocument(id="doc-1", score=0.9)] + +The model contains only the optional ``id`` and ``score`` fields; unset +fields serialize as JSON ``null``. Documents are recorded in +``gen_ai.retrieval.documents`` only in ``SPAN_ONLY`` or ``SPAN_AND_EVENT`` +content-capture mode. Passing dictionaries is deprecated, but existing +dictionary payloads continue to serialize unchanged. + + Environment Variables --------------------- diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_retrieval_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_retrieval_invocation.py index b9ab5d662..0b250ec47 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_retrieval_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_retrieval_invocation.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from typing import Any, Final +from typing import Final from opentelemetry._logs import Logger from opentelemetry.context import Context @@ -16,6 +16,7 @@ from opentelemetry.util.genai._instruments import _Instruments from opentelemetry.util.genai._invocation import Error, GenAIInvocation from opentelemetry.util.genai.completion_hook import CompletionHook +from opentelemetry.util.genai.types import RetrievalDocument from opentelemetry.util.genai.utils import ( ContentCapturingMode, gen_ai_json_dumps, @@ -96,7 +97,14 @@ def __init__( self._server_port: int | None = server_port self.top_k: int | None = None self.query_text: str | None = None - self.documents: Sequence[Mapping[str, Any]] | None = None + self.documents: ( + Sequence[RetrievalDocument | Mapping[str, object]] | None + ) = None + """Retrieved document models, captured only in span content modes. + + Passing mappings is deprecated; use ``RetrievalDocument`` instead. + Legacy mappings are still serialized unchanged. + """ def _get_metric_attributes(self) -> dict[str, AttributeValue]: # data_source_id intentionally excluded — high cardinality diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py index 7cc07bc4f..aae56434a 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py @@ -350,6 +350,18 @@ class OutputMessage: name: str | None = None +@dataclass() +class RetrievalDocument: + """Represents a document retrieved from a vector database or search system. + + Mirrors the `GenAI retrieval Python model - RetrievalDocument + `__. + """ + + id: str | None = None + score: float | None = None + + # Callback an instrumentor may supply to derive the error.type attribute from a # provider exception. # Returns None to fall back to the exception's fully qualified type name. diff --git a/util/opentelemetry-util-genai/tests/test_handler_retrieval.py b/util/opentelemetry-util-genai/tests/test_handler_retrieval.py index 02ab4a4fd..33161e008 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_retrieval.py +++ b/util/opentelemetry-util-genai/tests/test_handler_retrieval.py @@ -26,7 +26,7 @@ ) from opentelemetry.util.genai.handler import TelemetryHandler from opentelemetry.util.genai.invocation import RetrievalInvocation -from opentelemetry.util.genai.types import Error +from opentelemetry.util.genai.types import Error, RetrievalDocument class _RetrievalTestBase(TestCase): @@ -253,6 +253,99 @@ def test_stop_sets_documents_when_content_capture_enabled(self) -> None: self.assertIsInstance(raw, str) self.assertEqual(json.loads(raw), docs) + @patch.dict( + os.environ, + { + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "SPAN_ONLY", + }, + ) + def test_typed_documents_on_success_and_failure(self) -> None: + for failed in (False, True): + with self.subTest(failed=failed): + self.span_exporter.clear() + handler = TelemetryHandler( + tracer_provider=self.tracer_provider + ) + invocation = handler.retrieval() + invocation.documents = [ + RetrievalDocument(id="doc-1", score=0.95), + RetrievalDocument(id="doc-2", score=0.0), + RetrievalDocument(id="doc-3"), + RetrievalDocument(score=0.5), + RetrievalDocument(), + ] + if failed: + invocation.fail(ValueError("retrieval failed")) + else: + invocation.stop() + + span = self._get_finished_spans()[0] + raw = span.attributes[GenAI.GEN_AI_RETRIEVAL_DOCUMENTS] + self.assertIsInstance(raw, str) + documents = json.loads(raw) + self.assertEqual( + documents, + [ + {"id": "doc-1", "score": 0.95}, + {"id": "doc-2", "score": 0.0}, + {"id": "doc-3", "score": None}, + {"id": None, "score": 0.5}, + {"id": None, "score": None}, + ], + ) + self.assertIsInstance(documents[0]["id"], str) + self.assertIsInstance(documents[0]["score"], float) + if failed: + self.assertEqual(span.status.status_code, StatusCode.ERROR) + self.assertEqual( + span.attributes["error.type"], "ValueError" + ) + else: + self.assertEqual(span.status.status_code, StatusCode.UNSET) + + @patch.dict( + os.environ, + { + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "SPAN_ONLY", + }, + ) + def test_documents_preserve_legacy_mappings_alongside_models(self) -> None: + legacy = { + "id": "legacy", + "content": "text", + "metadata": {"source": "db"}, + } + handler = TelemetryHandler(tracer_provider=self.tracer_provider) + invocation = handler.retrieval() + invocation.documents = [legacy, RetrievalDocument(id="typed")] + invocation.stop() + + raw = self._get_finished_spans()[0].attributes[ + GenAI.GEN_AI_RETRIEVAL_DOCUMENTS + ] + self.assertEqual( + json.loads(raw), [legacy, {"id": "typed", "score": None}] + ) + + @patch.dict( + os.environ, + { + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "SPAN_ONLY", + }, + ) + def test_empty_documents_are_distinct_from_absent_documents(self) -> None: + handler = TelemetryHandler(tracer_provider=self.tracer_provider) + with handler.retrieval() as invocation: + invocation.documents = [] + with handler.retrieval(): + pass + + empty, absent = self._get_finished_spans() + self.assertEqual( + empty.attributes[GenAI.GEN_AI_RETRIEVAL_DOCUMENTS], "[]" + ) + self.assertNotIn(GenAI.GEN_AI_RETRIEVAL_DOCUMENTS, absent.attributes) + @patch.dict( os.environ, { @@ -263,7 +356,7 @@ def test_stop_suppresses_query_text_and_docs_in_event_only_mode( self, ) -> None: handler = TelemetryHandler(tracer_provider=self.tracer_provider) - docs = [{"id": "doc_1", "score": 0.95}] + docs = [RetrievalDocument(id="doc_1", score=0.95)] invocation = handler.retrieval() assert invocation.should_capture_content is True invocation.query_text = "What is the capital of France?" @@ -279,7 +372,7 @@ def test_stop_suppresses_query_text_and_docs_in_event_only_mode( def test_stop_suppresses_documents_when_content_capture_disabled( self, ) -> None: - docs = [{"id": "doc_1", "score": 0.95}] + docs = [RetrievalDocument(id="doc_1", score=0.95)] invocation = self.handler.retrieval() invocation.documents = docs invocation.stop()