Switched langchain llm callbacks to use genai utils handler

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

@@ -7,5 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Added support to call genai utils handler for langchain LLM invocations.
+  ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3889](#3889))
+
 - Added span support for genAI langchain llm invocation.
   ([#3665](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3665))

instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml

@@ -25,9 +25,7 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
-  "opentelemetry-api >= 1.31.0",
   "opentelemetry-instrumentation ~= 0.57b0",
-  "opentelemetry-semantic-conventions ~= 0.57b0"
 ]
 
 [project.optional-dependencies]

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

@@ -38,18 +38,16 @@
 
 from typing import Any, Callable, Collection
 
-from langchain_core.callbacks import BaseCallbackHandler  # type: ignore
+from langchain_core.callbacks import BaseCallbackHandler
 from wrapt import wrap_function_wrapper  # type: ignore
 
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.langchain.callback_handler import (
     OpenTelemetryLangChainCallbackHandler,
 )
 from opentelemetry.instrumentation.langchain.package import _instruments
-from opentelemetry.instrumentation.langchain.version import __version__
 from opentelemetry.instrumentation.utils import unwrap
-from opentelemetry.semconv.schemas import Schemas
-from opentelemetry.trace import get_tracer
+from opentelemetry.util.genai.handler import get_telemetry_handler
 
 
 class LangChainInstrumentor(BaseInstrumentor):
@@ -72,15 +70,12 @@ def _instrument(self, **kwargs: Any):
         Enable Langchain instrumentation.
         """
         tracer_provider = kwargs.get("tracer_provider")
-        tracer = get_tracer(
-            __name__,
-            __version__,
-            tracer_provider,
-            schema_url=Schemas.V1_37_0.value,
-        )
 
+        telemetry_handler = get_telemetry_handler(
+            tracer_provider=tracer_provider
+        )
         otel_callback_handler = OpenTelemetryLangChainCallbackHandler(
-            tracer=tracer,
+            telemetry_handler=telemetry_handler
         )
 
         wrap_function_wrapper(
@@ -109,7 +104,7 @@ def __init__(
     def __call__(
         self,
         wrapped: Callable[..., None],
-        instance: BaseCallbackHandler,  # type: ignore
+        instance: BaseCallbackHandler,
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ):

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

@@ -17,36 +17,37 @@
 from typing import Any
 from uuid import UUID
 
-from langchain_core.callbacks import BaseCallbackHandler  # type: ignore
-from langchain_core.messages import BaseMessage  # type: ignore
-from langchain_core.outputs import LLMResult  # type: ignore
+from langchain_core.callbacks import BaseCallbackHandler
+from langchain_core.messages import BaseMessage
+from langchain_core.outputs import LLMResult
 
-from opentelemetry.instrumentation.langchain.span_manager import _SpanManager
-from opentelemetry.semconv._incubating.attributes import (
-    gen_ai_attributes as GenAI,
+from opentelemetry.instrumentation.langchain.invocation_manager import (
+    _InvocationManager,
+)
+from opentelemetry.util.genai.handler import TelemetryHandler
+from opentelemetry.util.genai.types import (
+    Error,
+    InputMessage,
+    LLMInvocation,
+    OutputMessage,
+    Text,
 )
-from opentelemetry.trace import Tracer
 
 
-class OpenTelemetryLangChainCallbackHandler(BaseCallbackHandler):  # type: ignore[misc]
+class OpenTelemetryLangChainCallbackHandler(BaseCallbackHandler):
     """
     A callback handler for LangChain that uses OpenTelemetry to create spans for LLM calls and chains, tools etc,. in future.
     """
 
-    def __init__(
-        self,
-        tracer: Tracer,
-    ) -> None:
-        super().__init__()  # type: ignore
-
-        self.span_manager = _SpanManager(
-            tracer=tracer,
-        )
+    def __init__(self, telemetry_handler: TelemetryHandler) -> None:
+        super().__init__()
+        self._telemetry_handler = telemetry_handler
+        self._invocation_manager = _InvocationManager()
 
     def on_chat_model_start(
         self,
         serialized: dict[str, Any],
-        messages: list[list[BaseMessage]],  # type: ignore
+        messages: list[list[BaseMessage]],
         *,
         run_id: UUID,
         tags: list[str] | None,
@@ -82,87 +83,87 @@ def on_chat_model_start(
         if request_model == "unknown":
             return
 
-        span = self.span_manager.create_chat_span(
-            run_id=run_id,
-            parent_run_id=parent_run_id,
-            request_model=request_model,
-        )
-
         if params is not None:
             top_p = params.get("top_p")
-            if top_p is not None:
-                span.set_attribute(GenAI.GEN_AI_REQUEST_TOP_P, top_p)
             frequency_penalty = params.get("frequency_penalty")
-            if frequency_penalty is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_REQUEST_FREQUENCY_PENALTY, frequency_penalty
-                )
             presence_penalty = params.get("presence_penalty")
-            if presence_penalty is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_REQUEST_PRESENCE_PENALTY, presence_penalty
-                )
             stop_sequences = params.get("stop")
-            if stop_sequences is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_REQUEST_STOP_SEQUENCES, stop_sequences
-                )
             seed = params.get("seed")
-            if seed is not None:
-                span.set_attribute(GenAI.GEN_AI_REQUEST_SEED, seed)
+
             # ChatOpenAI
             temperature = params.get("temperature")
-            if temperature is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_REQUEST_TEMPERATURE, temperature
-                )
+
             # ChatOpenAI
             max_tokens = params.get("max_completion_tokens")
-            if max_tokens is not None:
-                span.set_attribute(GenAI.GEN_AI_REQUEST_MAX_TOKENS, max_tokens)
 
+        provider = "unknown"
         if metadata is not None:
             provider = metadata.get("ls_provider")
-            if provider is not None:
-                span.set_attribute("gen_ai.provider.name", provider)
+
             # ChatBedrock
             temperature = metadata.get("ls_temperature")
-            if temperature is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_REQUEST_TEMPERATURE, temperature
-                )
+
             # ChatBedrock
             max_tokens = metadata.get("ls_max_tokens")
-            if max_tokens is not None:
-                span.set_attribute(GenAI.GEN_AI_REQUEST_MAX_TOKENS, max_tokens)
+
+        input_messages: list[InputMessage] = []
+        for sub_messages in messages:
+            for message in sub_messages:
+                content = message.content  # pyright: ignore[reportUnknownMemberType]
+                role = message.type
+                parts = [Text(content=content, type="text")]
+                input_messages.append(InputMessage(parts=parts, role=role))
+
+        llm_invocation = LLMInvocation(
+            request_model=request_model,
+            input_messages=input_messages,
+            provider=provider,
+            top_p=top_p,
+            frequency_penalty=frequency_penalty,
+            presence_penalty=presence_penalty,
+            stop_sequences=stop_sequences,
+            seed=seed,
+            temperature=temperature,
+            max_tokens=max_tokens,
+        )
+        llm_invocation = self._telemetry_handler.start_llm(
+            invocation=llm_invocation
+        )
+        self._invocation_manager.add_invocation_state(
+            run_id=run_id,
+            parent_run_id=parent_run_id,
+            invocation=llm_invocation,
+        )
 
     def on_llm_end(
         self,
-        response: LLMResult,  # type: ignore [reportUnknownParameterType]
+        response: LLMResult,
         *,
         run_id: UUID,
         parent_run_id: UUID | None,
         **kwargs: Any,
     ) -> None:
-        span = self.span_manager.get_span(run_id)
-
-        if span is None:
-            # If the span does not exist, we cannot set attributes or end it
+        llm_invocation = self._invocation_manager.get_invocation(run_id=run_id)
+        if llm_invocation is None or not isinstance(
+            llm_invocation, LLMInvocation
+        ):
+            # If the invocation does not exist, we cannot set attributes or end it
             return
 
-        finish_reasons: list[str] = []
-        for generation in getattr(response, "generations", []):  # type: ignore
+        output_messages: list[OutputMessage] = []
+        for generation in getattr(response, "generations", []):
             for chat_generation in generation:
+                # Get finish reason
                 generation_info = getattr(
                     chat_generation, "generation_info", None
                 )
                 if generation_info is not None:
                     finish_reason = generation_info.get(
                         "finish_reason", "unknown"
                     )
-                    if finish_reason is not None:
-                        finish_reasons.append(str(finish_reason))
+
                 if chat_generation.message:
+                    # Get finish reason if generation_info is None above
                     if (
                         generation_info is None
                         and chat_generation.message.response_metadata
@@ -172,46 +173,57 @@ def on_llm_end(
                                 "stopReason", "unknown"
                             )
                         )
-                        if finish_reason is not None:
-                            finish_reasons.append(str(finish_reason))
+
+                    # Get message content
+                    parts = [
+                        Text(
+                            content=chat_generation.message.content,
+                            type="text",
+                        )
+                    ]
+                    role = chat_generation.message.type
+                    output_message = OutputMessage(
+                        role=role,
+                        parts=parts,
+                        finish_reason=finish_reason,
+                    )
+                    output_messages.append(output_message)
+
+                    # Get token usage if available
                     if chat_generation.message.usage_metadata:
                         input_tokens = (
                             chat_generation.message.usage_metadata.get(
                                 "input_tokens", 0
                             )
                         )
+                        llm_invocation.input_tokens = input_tokens
+
                         output_tokens = (
                             chat_generation.message.usage_metadata.get(
                                 "output_tokens", 0
                             )
                         )
-                        span.set_attribute(
-                            GenAI.GEN_AI_USAGE_INPUT_TOKENS, input_tokens
-                        )
-                        span.set_attribute(
-                            GenAI.GEN_AI_USAGE_OUTPUT_TOKENS, output_tokens
-                        )
+                        llm_invocation.output_tokens = output_tokens
 
-        span.set_attribute(
-            GenAI.GEN_AI_RESPONSE_FINISH_REASONS, finish_reasons
-        )
+        llm_invocation.output_messages = output_messages
 
-        llm_output = getattr(response, "llm_output", None)  # type: ignore
+        llm_output = getattr(response, "llm_output", None)
         if llm_output is not None:
             response_model = llm_output.get("model_name") or llm_output.get(
                 "model"
             )
             if response_model is not None:
-                span.set_attribute(
-                    GenAI.GEN_AI_RESPONSE_MODEL, str(response_model)
-                )
+                llm_invocation.response_model_name = str(response_model)
 
             response_id = llm_output.get("id")
             if response_id is not None:
-                span.set_attribute(GenAI.GEN_AI_RESPONSE_ID, str(response_id))
+                llm_invocation.response_id = str(response_id)
 
-        # End the LLM span
-        self.span_manager.end_span(run_id)
+        llm_invocation = self._telemetry_handler.stop_llm(
+            invocation=llm_invocation
+        )
+        if llm_invocation.span and not llm_invocation.span.is_recording():
+            self._invocation_manager.delete_invocation_state(run_id=run_id)
 
     def on_llm_error(
         self,
@@ -221,4 +233,16 @@ def on_llm_error(
         parent_run_id: UUID | None,
         **kwargs: Any,
     ) -> None:
-        self.span_manager.handle_error(error, run_id)
+        llm_invocation = self._invocation_manager.get_invocation(run_id=run_id)
+        if llm_invocation is None or not isinstance(
+            llm_invocation, LLMInvocation
+        ):
+            # If the invocation does not exist, we cannot set attributes or end it
+            return
+
+        error_otel = Error(message=str(error), type=type(error))
+        llm_invocation = self._telemetry_handler.fail_llm(
+            invocation=llm_invocation, error=error_otel
+        )
+        if llm_invocation.span and not llm_invocation.span.is_recording():
+            self._invocation_manager.delete_invocation_state(run_id=run_id)