feat: capture langchain message content by default

Author: nagkumar91

instrumentation-genai/opentelemetry-instrumentation-langchain/CHANGELOG.md

@@ -11,3 +11,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3665](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3665))
 - Align LangChain instrumentation with GenAI schema 1.37.0 and add unit coverage for updated metadata.
   ([#3813](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3813))
+- Capture GenAI input/output messages on spans by default with opt-out control.
+  ([#3813](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3813))

instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/__init__.py

@@ -36,6 +36,7 @@
 ---
 """
 
+import os
 from typing import Any, Callable, Collection
 
 from langchain_core.callbacks import BaseCallbackHandler  # type: ignore
@@ -72,6 +73,7 @@ def _instrument(self, **kwargs: Any):
         Enable Langchain instrumentation.
         """
         tracer_provider = kwargs.get("tracer_provider")
+        capture_messages = self._resolve_capture_messages(kwargs)
         tracer = get_tracer(
             __name__,
             __version__,
@@ -81,6 +83,7 @@ def _instrument(self, **kwargs: Any):
 
         otel_callback_handler = OpenTelemetryLangChainCallbackHandler(
             tracer=tracer,
+            capture_messages=capture_messages,
         )
 
         wrap_function_wrapper(
@@ -95,6 +98,18 @@ def _uninstrument(self, **kwargs: Any):
         """
         unwrap("langchain_core.callbacks.base.BaseCallbackManager", "__init__")
 
+    def _resolve_capture_messages(self, kwargs: dict[str, Any]) -> bool:
+        if "capture_messages" in kwargs:
+            return bool(kwargs["capture_messages"])
+
+        env_value = os.getenv(
+            "OTEL_INSTRUMENTATION_LANGCHAIN_CAPTURE_MESSAGES"
+        )
+        if env_value is not None:
+            return env_value.lower() in ("1", "true", "yes", "on")
+
+        return True
+
 
 class _BaseCallbackManagerInitWrapper:
     """

instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any
 from urllib.parse import urlparse
 from uuid import UUID
@@ -68,12 +69,14 @@ class OpenTelemetryLangChainCallbackHandler(BaseCallbackHandler):  # type: ignor
     def __init__(
         self,
         tracer: Tracer,
+        capture_messages: bool,
     ) -> None:
         super().__init__()  # type: ignore
 
         self.span_manager = _SpanManager(
             tracer=tracer,
         )
+        self._capture_messages = capture_messages
 
     def on_chat_model_start(
         self,
@@ -113,6 +116,13 @@ def on_chat_model_start(
 
         self._apply_request_attributes(span, params, metadata)
 
+        if self._capture_messages and messages:
+            serialized_messages = self._serialize_input_messages(messages)
+            span.set_attribute(
+                GenAI.GEN_AI_INPUT_MESSAGES,
+                self._serialize_to_json(serialized_messages),
+            )
+
     def _resolve_provider(
         self, llm_name: str | None, metadata: dict[str, Any] | None
     ) -> str | None:
@@ -295,6 +305,59 @@ def _extract_output_type(self, params: dict[str, Any]) -> str | None:
 
         return mapping.get(lowered)
 
+    def _serialize_input_messages(
+        self, messages: list[list[BaseMessage]]
+    ) -> list[dict[str, Any]]:
+        serialized: list[dict[str, Any]] = []
+        for conversation in messages:
+            for message in conversation:
+                serialized.append(self._serialize_message(message))
+        return serialized
+
+    def _serialize_output_messages(
+        self, response: LLMResult
+    ) -> list[dict[str, Any]]:
+        serialized: list[dict[str, Any]] = []
+        generations = getattr(response, "generations", [])  # type: ignore
+        for generation in generations:
+            for item in generation:
+                message = getattr(item, "message", None)
+                if message is not None:
+                    serialized.append(self._serialize_message(message))
+        return serialized
+
+    def _serialize_message(self, message: BaseMessage) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            "type": getattr(message, "type", message.__class__.__name__),
+            "content": getattr(message, "content", None),
+        }
+        for attr in (
+            "additional_kwargs",
+            "response_metadata",
+            "tool_call_id",
+            "tool_calls",
+            "usage_metadata",
+            "id",
+            "name",
+        ):
+            value = getattr(message, attr, None)
+            if value:
+                payload[attr] = value
+        return payload
+
+    def _serialize_to_json(self, payload: Any) -> str:
+        return json.dumps(payload, default=self._json_default)
+
+    @staticmethod
+    def _json_default(value: Any) -> Any:
+        if isinstance(value, (str, int, float, bool)) or value is None:
+            return value
+        if isinstance(value, dict):
+            return value
+        if isinstance(value, (list, tuple)):
+            return list(value)
+        return getattr(value, "__dict__", str(value))
+
     def on_llm_end(
         self,
         response: LLMResult,  # type: ignore [reportUnknownParameterType]
@@ -379,6 +442,14 @@ def on_llm_end(
                     OPENAI_RESPONSE_SYSTEM_FINGERPRINT, system_fingerprint
                 )
 
+        if self._capture_messages:
+            serialized_outputs = self._serialize_output_messages(response)
+            if serialized_outputs:
+                span.set_attribute(
+                    GenAI.GEN_AI_OUTPUT_MESSAGES,
+                    self._serialize_to_json(serialized_outputs),
+                )
+
         # End the LLM span
         self.span_manager.end_span(run_id)
 

instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_callback_handler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from dataclasses import dataclass
 from typing import Any
 from uuid import uuid4
@@ -24,12 +25,13 @@
 )
 
 
-def _create_handler():
+def _create_handler(capture_messages: bool = True):
     exporter = InMemorySpanExporter()
     provider = TracerProvider()
     provider.add_span_processor(SimpleSpanProcessor(exporter))
     handler = OpenTelemetryLangChainCallbackHandler(
-        tracer=provider.get_tracer(__name__)
+        tracer=provider.get_tracer(__name__),
+        capture_messages=capture_messages,
     )
     return handler, exporter
 
@@ -77,6 +79,22 @@ class _DummyLLMResult:
     llm_output: dict[str, Any]
 
 
+@dataclass
+class _DummyGeneration:
+    message: Any
+    generation_info: dict[str, Any] | None = None
+
+
+@dataclass
+class _FakeMessage:
+    content: str
+    type: str
+    additional_kwargs: dict[str, Any] | None = None
+    response_metadata: dict[str, Any] | None = None
+    usage_metadata: dict[str, Any] | None = None
+    id: str | None = None
+
+
 def test_llm_end_sets_response_metadata():
     handler, exporter = _create_handler()
     run_id = uuid4()
@@ -134,3 +152,83 @@ def test_choice_count_not_set_when_one():
     handler.span_manager.end_span(run_id)
     span = exporter.get_finished_spans()[0]
     assert GenAI.GEN_AI_REQUEST_CHOICE_COUNT not in span.attributes
+
+
+def test_capture_messages_sets_attributes_by_default():
+    handler, exporter = _create_handler()
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={"name": "ChatOpenAI"},
+        messages=[
+            [
+                _FakeMessage(
+                    content="hello",
+                    type="human",
+                    id="m1",
+                )
+            ]
+        ],
+        run_id=run_id,
+        tags=None,
+        parent_run_id=None,
+        metadata={"ls_model_name": "gpt-4"},
+        invocation_params={"params": {"model": "gpt-4"}},
+    )
+
+    handler.on_llm_end(
+        _DummyLLMResult(
+            generations=[
+                [
+                    _DummyGeneration(
+                        message=_FakeMessage(
+                            content="result",
+                            type="ai",
+                            id="m2",
+                        )
+                    )
+                ]
+            ],
+            llm_output={},
+        ),
+        run_id=run_id,
+        parent_run_id=None,
+    )
+
+    span = exporter.get_finished_spans()[0]
+    input_payload = json.loads(span.attributes[GenAI.GEN_AI_INPUT_MESSAGES])
+    output_payload = json.loads(span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES])
+    assert input_payload[0]["content"] == "hello"
+    assert output_payload[0]["content"] == "result"
+
+
+def test_capture_messages_can_be_disabled():
+    handler, exporter = _create_handler(capture_messages=False)
+    run_id = uuid4()
+
+    handler.on_chat_model_start(
+        serialized={"name": "ChatOpenAI"},
+        messages=[
+            [
+                _FakeMessage(
+                    content="hello",
+                    type="human",
+                )
+            ]
+        ],
+        run_id=run_id,
+        tags=None,
+        parent_run_id=None,
+        metadata={"ls_model_name": "gpt-4"},
+        invocation_params={"params": {"model": "gpt-4"}},
+    )
+
+    handler.on_llm_end(
+        _DummyLLMResult(generations=[], llm_output={}),
+        run_id=run_id,
+        parent_run_id=None,
+    )
+
+    span = exporter.get_finished_spans()[0]
+    assert GenAI.GEN_AI_INPUT_MESSAGES not in span.attributes
+    assert GenAI.GEN_AI_OUTPUT_MESSAGES not in span.attributes

instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_llm_call.py

@@ -188,7 +188,8 @@ def test_azure_chat_sets_provider_and_server_attributes():
     provider = TracerProvider()
     provider.add_span_processor(SimpleSpanProcessor(exporter))
     handler = OpenTelemetryLangChainCallbackHandler(
-        provider.get_tracer(__name__)
+        provider.get_tracer(__name__),
+        capture_messages=True,
     )
 
     run_id = uuid4()