[opentelemetry-instrumentation-genai-langchain] Capture prompt template name and variables on chat

instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/730.added

@@ -0,0 +1 @@
+Capture prompt template name - `gen_ai.prompt.name` and variables - `gen_ai.prompt.variable.<key>` on chat

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

@@ -22,6 +22,7 @@
 )
 from opentelemetry.instrumentation.genai.langchain.invocation_manager import (
     _InvocationManager,
+    _PromptContext,
 )
 from opentelemetry.instrumentation.genai.langchain.operation_mapping import (
     OperationName,
@@ -71,6 +72,96 @@
     "conversation_id",
 )
 
+_PROMPT_TEMPLATE_TYPES = {"ChatPromptTemplate", "PromptTemplate"}
+
+
+def _prompt_template_type(runnable: Mapping[str, Any]) -> str | None:
+    runnable_id = runnable.get("id")
+    if not isinstance(runnable_id, Sequence) or isinstance(runnable_id, str):
+        return None
+    runnable_id = cast(Sequence[str], runnable_id)
+    template_type = runnable_id[-1] if runnable_id else None
+    if not isinstance(template_type, str):
+        return None
+    return template_type if template_type in _PROMPT_TEMPLATE_TYPES else None
+
+
+def _string_sequence(value: object) -> list[str]:
+    if not isinstance(value, Sequence) or isinstance(value, str):
+        return []
+    return [
+        item for item in cast(Sequence[object], value) if isinstance(item, str)
+    ]
+
+
+def _extract_prompt_context(
+    runnable: Mapping[str, Any],
+    inputs: object,
+    metadata: Mapping[str, Any] | None,
+    *,
+    capture_variables: bool = True,
+) -> _PromptContext | None:
+    template_type = _prompt_template_type(runnable)
+    if template_type is None:
+        return None
+
+    raw_template_kwargs = runnable.get("kwargs")
+    if not isinstance(raw_template_kwargs, Mapping):
+        template_kwargs: Mapping[str, object] = {}
+    else:
+        template_kwargs = cast(Mapping[str, object], raw_template_kwargs)
+
+    name = (metadata or {}).get("prompt_name")
+    if not isinstance(name, str):
+        name = template_kwargs.get("name")
+    if not isinstance(name, str):
+        runnable_name = runnable.get("name")
+        name = (
+            runnable_name
+            if isinstance(runnable_name, str)
+            and runnable_name not in _PROMPT_TEMPLATE_TYPES
+            else None
+        )
+
+    if not capture_variables:
+        return _PromptContext(name=name, variables={})
+
+    required_variables = _string_sequence(
+        template_kwargs.get("input_variables")
+    )
+    declared_variables = set(required_variables)
+    declared_variables.update(
+        _string_sequence(template_kwargs.get("optional_variables"))
+    )
+
+    partial_variables = template_kwargs.get("partial_variables")
+    if isinstance(partial_variables, Mapping):
+        partial_variables = cast(Mapping[object, object], partial_variables)
+        declared_variables.update(
+            key for key in partial_variables if isinstance(key, str)
+        )
+
+    if isinstance(inputs, Mapping):
+        runtime_variables: Mapping[str, object] = cast(
+            Mapping[str, object], inputs
+        )
+    elif len(required_variables) == 1:
+        runtime_variables = {required_variables[0]: inputs}
+    else:
+        runtime_variables = {}
+
+    effective_variables: dict[str, object] = {}
+    for source in (partial_variables, runtime_variables):
+        if not isinstance(source, Mapping):
+            continue
+        source = cast(Mapping[object, object], source)
+        for key, value in source.items():
+            if not isinstance(key, str) or key not in declared_variables:
+                continue
+            effective_variables[key] = value
+
+    return _PromptContext(name=name, variables=effective_variables)
+
 
 def _conversation_id(metadata: dict[str, Any] | None) -> str | None:
     """Return the conversation id from a run's own metadata."""
@@ -180,7 +271,7 @@ def __init__(
 
     def on_chain_start(
         self,
-        serialized: dict[str, Any],
+        serialized: dict[str, Any] | None,
         inputs: dict[str, Any],
         *,
         run_id: UUID,
@@ -189,6 +280,7 @@ def on_chain_start(
         metadata: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> Any:
+        serialized = serialized or {}
         parent_agent_name, ancestor_agent_names = self._find_agent_context(
             parent_run_id
         )
@@ -300,6 +392,14 @@ def on_chain_start(
                 run_id, parent_run_id, None
             )
 
+        if prompt_context := _extract_prompt_context(
+            serialized,
+            inputs,
+            metadata,
+            capture_variables=capture_content,
+        ):
+            self._invocation_manager.set_prompt_context(run_id, prompt_context)
+
     def on_chain_end(
         self,
         outputs: dict[str, Any],
@@ -308,6 +408,7 @@ def on_chain_end(
         parent_run_id: UUID | None = None,
         **kwargs: Any,
     ) -> Any:
+        self._invocation_manager.publish_prompt_context(run_id)
         invocation = self._invocation_manager.get_invocation(run_id=run_id)
         if invocation is None or not isinstance(
             invocation, (WorkflowInvocation, LocalAgentInvocation)
@@ -437,13 +538,19 @@ def on_chat_model_start(
         parent_context = self._invocation_manager.get_parent_context(
             parent_run_id
         )
+        prompt_context = self._invocation_manager.consume_prompt_context(
+            parent_run_id
+        )
         llm_invocation = self._telemetry_handler.inference(
             provider,
             request_model=request_model,
             context=parent_context,
             conversation_id=_conversation_id(metadata),
             _attach_to_context=self._attach_to_context,
         )
+        if prompt_context is not None:
+            llm_invocation.prompt_name = prompt_context.name
+            llm_invocation.prompt_variables = prompt_context.variables
         llm_invocation.input_messages = input_messages
         llm_invocation.top_p = top_p
         llm_invocation.top_k = top_k

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

@@ -1,6 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import TypeAlias
 from uuid import UUID
@@ -14,7 +15,7 @@
     WorkflowInvocation,
 )
 
-__all__ = ["_InvocationManager"]
+__all__ = ["_InvocationManager", "_PromptContext"]
 
 _AnyInvocation: TypeAlias = (
     InferenceInvocation
@@ -25,13 +26,23 @@
 )
 
 
+@dataclass(frozen=True)
+class _PromptContext:
+    name: str | None
+    variables: Mapping[str, object]
+
+
 @dataclass
 class _InvocationState:
     invocation: _AnyInvocation | None
     children: list[UUID] = field(default_factory=lambda: list())
     parent_run_id: UUID | None = None
     ended: bool = False
     agent_name: str | None = None
+    prompt_context: _PromptContext | None = None
+    pending_prompt_contexts: list[_PromptContext] = field(
+        default_factory=lambda: list[_PromptContext]()
+    )
 
 
 class _InvocationManager:
@@ -80,6 +91,44 @@ def get_parent_context(self, parent_run_id: UUID | None) -> Context | None:
             current = self.get_parent_run_id(current)
         return None
 
+    def set_prompt_context(
+        self, run_id: UUID, prompt_context: _PromptContext
+    ) -> None:
+        invocation_state = self._invocations.get(run_id)
+        if invocation_state is not None:
+            invocation_state.prompt_context = prompt_context
+
+    def publish_prompt_context(self, run_id: UUID) -> None:
+        invocation_state = self._invocations.get(run_id)
+        if invocation_state is None or invocation_state.prompt_context is None:
+            return
+        parent_run_id = invocation_state.parent_run_id
+        parent_state = (
+            self._invocations.get(parent_run_id)
+            if parent_run_id is not None
+            else None
+        )
+        if parent_state is not None:
+            parent_state.pending_prompt_contexts.append(
+                invocation_state.prompt_context
+            )
+        invocation_state.prompt_context = None
+
+    def consume_prompt_context(
+        self, parent_run_id: UUID | None
+    ) -> _PromptContext | None:
+        current_run_id = parent_run_id
+        while current_run_id is not None:
+            invocation_state = self._invocations.get(current_run_id)
+            if invocation_state is None:
+                return None
+            if invocation_state.pending_prompt_contexts:
+                prompt_context = invocation_state.pending_prompt_contexts[-1]
+                invocation_state.pending_prompt_contexts.clear()
+                return prompt_context
+            current_run_id = invocation_state.parent_run_id
+        return None
+
     def delete_invocation_state(self, run_id: UUID) -> None:
         invocation_state = self._invocations.get(run_id)
         if not invocation_state: