diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/939.added b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/939.added new file mode 100644 index 000000000..b572bd257 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/939.added @@ -0,0 +1 @@ +Link chat spans to the prompt template that produced their input when the prompt runs outside a chain or several prompts share a parent, set `gen_ai.prompt.version` from `prompt_version` or LangChain Hub metadata, and detect every `BasePromptTemplate` subclass diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_execution_context.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_execution_context.py index 5b1ec971d..ea73a889d 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_execution_context.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_execution_context.py @@ -18,6 +18,9 @@ from wrapt import wrap_function_wrapper from opentelemetry.context import attach, detach, get_current +from opentelemetry.instrumentation.genai.langchain._prompt_input import ( + _wrap_prompt_input, +) from opentelemetry.instrumentation.genai.langchain._run_context import ( _astart_run, _RunScope, @@ -230,6 +233,26 @@ def _patch( ) def instrument(self) -> None: + # Innermost on the streams, so the stream boundary below chooses the + # run id first and the prompt input reuses it. + for method, stream, asynchronous in ( + ("invoke", False, False), + ("ainvoke", False, True), + ("stream", True, False), + ("astream", True, True), + ): + self._patch( + "langchain_core.language_models.chat_models", + "BaseChatModel", + method, + partial( + _wrap_prompt_input, + invocations=self._invocations, + stream=stream, + asynchronous=asynchronous, + ), + ) + for module_name, class_name, method in _METHODS: self._patch(module_name, class_name, method, self._wrap) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_prompt_input.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_prompt_input.py new file mode 100644 index 000000000..ebd9a5976 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/_prompt_input.py @@ -0,0 +1,132 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Link a chat model call to the prompt template that produced its input. + +The run tree alone cannot always tell: a prompt formatted outside any chain +has no parent run to publish to, and two prompts formatted under one parent +reach their models in any order. The callback handler records the prompt value +each prompt run returns; these wrappers find that value in a chat model's +input and record its template under the run id chosen here for the call. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable, Iterator +from functools import wraps +from typing import Any +from uuid import UUID + +from langchain_core.prompt_values import PromptValue + +from opentelemetry.instrumentation.genai.langchain._run_context import ( + _new_run_id, + _signature, +) +from opentelemetry.instrumentation.genai.langchain.invocation_manager import ( + _InvocationManager, +) + +__all__ = ["_wrap_prompt_input"] + +_Call = tuple[UUID | None, tuple[Any, ...], dict[str, Any]] + + +def _wrap_prompt_input( + original: Callable[..., Any], + invocations: _InvocationManager, + stream: bool, + asynchronous: bool, +) -> Callable[..., Any]: + parameters = _signature(original) + + def prepare( + instance: Any, args: tuple[Any, ...], kwargs: dict[str, Any] + ) -> _Call: + bound = parameters.bind(instance, *args, **kwargs) + value = bound.arguments.get("input") + prompt_context = ( + invocations.get_prompt_value_context(value) + if isinstance(value, PromptValue) + else None + ) + if prompt_context is None: + return None, args, kwargs + # The model starts its run with this id, so the handler finds the + # template when the run starts. + config = dict(bound.arguments.get("config") or {}) + run_id: UUID = config.get("run_id") or _new_run_id() + config["run_id"] = run_id + bound.arguments["config"] = config + invocations.set_model_prompt_context(run_id, prompt_context) + return run_id, bound.args[1:], bound.kwargs + + def forget(run_id: UUID | None) -> None: + # The handler takes the template when the run starts; this drops it + # when the run never started. + if run_id is not None: + invocations.pop_model_prompt_context(run_id) + + @wraps(original) + def invoke( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + run_id, args, kwargs = prepare(instance, args, kwargs) + try: + return wrapped(*args, **kwargs) + finally: + forget(run_id) + + @wraps(original) + async def ainvoke( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + run_id, args, kwargs = prepare(instance, args, kwargs) + try: + return await wrapped(*args, **kwargs) + finally: + forget(run_id) + + # The stream wrappers are generators, so the input is read on the first + # advancement, like the model's own stream, and an unread stream records + # nothing. + @wraps(original) + def sync_stream( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Iterator[Any]: + run_id, args, kwargs = prepare(instance, args, kwargs) + try: + yield from wrapped(*args, **kwargs) + finally: + forget(run_id) + + @wraps(original) + async def async_stream( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> AsyncIterator[Any]: + run_id, args, kwargs = prepare(instance, args, kwargs) + chunks = wrapped(*args, **kwargs) + try: + async for chunk in chunks: + yield chunk + finally: + forget(run_id) + close = getattr(chunks, "aclose", None) + if close is not None: + await close() + + if stream: + return async_stream if asynchronous else sync_stream + return ainvoke if asynchronous else invoke 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 06f3566dc..50275f0e5 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 @@ -16,6 +16,7 @@ GenerationChunk, LLMResult, ) +from langchain_core.prompt_values import PromptValue from opentelemetry.instrumentation.genai.langchain.agent_context import ( claim_agent, @@ -75,14 +76,17 @@ _PROMPT_TEMPLATE_TYPES = {"ChatPromptTemplate", "PromptTemplate"} -def _prompt_template_type(runnable: Mapping[str, Any]) -> str | None: +def _class_name(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 + runnable_id = cast(Sequence[object], runnable_id) + class_name = runnable_id[-1] if runnable_id else None + return class_name if isinstance(class_name, str) else None + + +def _prompt_template_type(runnable: Mapping[str, Any]) -> str | None: + template_type = _class_name(runnable) return template_type if template_type in _PROMPT_TEMPLATE_TYPES else None @@ -94,15 +98,26 @@ def _string_sequence(value: object) -> list[str]: ] +def _string_metadata( + metadata: Mapping[str, Any] | None, *keys: str +) -> str | None: + for key in keys: + value = (metadata or {}).get(key) + if isinstance(value, (str, int)) and not isinstance(value, bool): + return str(value) + return None + + def _extract_prompt_context( runnable: Mapping[str, Any], inputs: object, metadata: Mapping[str, Any] | None, *, capture_variables: bool = True, + run_type: str | None = None, ) -> _PromptContext | None: - template_type = _prompt_template_type(runnable) - if template_type is None: + # LangChain reports every BasePromptTemplate subclass with run_type "prompt". + if _prompt_template_type(runnable) is None and run_type != "prompt": return None raw_template_kwargs = runnable.get("kwargs") @@ -114,17 +129,27 @@ def _extract_prompt_context( name = (metadata or {}).get("prompt_name") if not isinstance(name, str): name = template_kwargs.get("name") + if not isinstance(name, str): + # Prompts pulled from LangChain Hub carry their repo and commit. + name = _string_metadata(metadata, "lc_hub_repo") if not isinstance(name, str): runnable_name = runnable.get("name") + # An unnamed template reports its class name, which names nothing. name = ( runnable_name if isinstance(runnable_name, str) and runnable_name not in _PROMPT_TEMPLATE_TYPES + and runnable_name != _class_name(runnable) else None ) + version = ( + _string_metadata(metadata, "prompt_version", "lc_hub_commit_hash") + if name is not None + else None + ) if not capture_variables: - return _PromptContext(name=name, variables={}) + return _PromptContext(name=name, variables={}, version=version) required_variables = _string_sequence( template_kwargs.get("input_variables") @@ -172,7 +197,9 @@ def _extract_prompt_context( continue effective_variables[key] = value - return _PromptContext(name=name, variables=effective_variables) + return _PromptContext( + name=name, variables=effective_variables, version=version + ) def _conversation_id(metadata: dict[str, Any] | None) -> str | None: @@ -410,6 +437,7 @@ def on_chain_start( inputs, metadata, capture_variables=capture_content, + run_type=kwargs.get("run_type"), ): self._invocation_manager.set_prompt_context(run_id, prompt_context) @@ -421,7 +449,14 @@ def on_chain_end( parent_run_id: UUID | None = None, **kwargs: Any, ) -> Any: - self._invocation_manager.publish_prompt_context(run_id) + prompt_context = self._invocation_manager.publish_prompt_context( + run_id + ) + if prompt_context is not None and isinstance(outputs, PromptValue): + # The value carries its template to whichever model it reaches. + self._invocation_manager.set_prompt_value_context( + outputs, prompt_context + ) invocation = self._invocation_manager.get_invocation(run_id=run_id) if invocation is None or not isinstance( invocation, (WorkflowInvocation, LocalAgentInvocation) @@ -553,8 +588,14 @@ 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 + # Consume the sibling prompt either way, so a later model call under the + # same parent cannot take it; the model's own input wins when known. + sibling_prompt_context = ( + self._invocation_manager.consume_prompt_context(parent_run_id) + ) + prompt_context = ( + self._invocation_manager.pop_model_prompt_context(run_id) + or sibling_prompt_context ) llm_invocation = self._telemetry_handler.inference( provider, @@ -565,6 +606,7 @@ def on_chat_model_start( ) if prompt_context is not None: llm_invocation.prompt_name = prompt_context.name + llm_invocation.prompt_version = prompt_context.version llm_invocation.prompt_variables = prompt_context.variables llm_invocation.input_messages = input_messages llm_invocation.top_p = top_p diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py index 022107199..ec0dd7f4f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py +++ b/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 +import weakref from collections.abc import Mapping from dataclasses import dataclass, field from typing import TypeAlias @@ -30,6 +31,7 @@ class _PromptContext: name: str | None variables: Mapping[str, object] + version: str | None = None @dataclass @@ -50,6 +52,12 @@ def __init__(self) -> None: # Map from run_id -> _InvocationState, to keep track of invocations and parent/child relationships # TODO: TTL cache to avoid memory leaks in long-running processes. self._invocations: dict[UUID, _InvocationState] = {} + # Prompt contexts of the prompt values that prompt runs returned, keyed + # by object id while the value is alive. + self._prompt_values: dict[int, _PromptContext] = {} + # Prompt contexts of chat model runs whose input is one of those values, + # keyed by the run id chosen before the run starts. + self._model_prompt_contexts: dict[UUID, _PromptContext] = {} def add_invocation_state( self, @@ -98,10 +106,11 @@ def set_prompt_context( if invocation_state is not None: invocation_state.prompt_context = prompt_context - def publish_prompt_context(self, run_id: UUID) -> None: + def publish_prompt_context(self, run_id: UUID) -> _PromptContext | None: invocation_state = self._invocations.get(run_id) if invocation_state is None or invocation_state.prompt_context is None: - return + return None + prompt_context = invocation_state.prompt_context parent_run_id = invocation_state.parent_run_id parent_state = ( self._invocations.get(parent_run_id) @@ -109,10 +118,33 @@ def publish_prompt_context(self, run_id: UUID) -> None: else None ) if parent_state is not None: - parent_state.pending_prompt_contexts.append( - invocation_state.prompt_context - ) + parent_state.pending_prompt_contexts.append(prompt_context) invocation_state.prompt_context = None + return prompt_context + + def set_prompt_value_context( + self, prompt_value: object, prompt_context: _PromptContext + ) -> None: + """Remember which template produced ``prompt_value``, while it is alive.""" + key = id(prompt_value) + try: + weakref.finalize(prompt_value, self._prompt_values.pop, key, None) + except TypeError: + return # not weak-referenceable, so its id could be reused + self._prompt_values[key] = prompt_context + + def get_prompt_value_context( + self, prompt_value: object + ) -> _PromptContext | None: + return self._prompt_values.get(id(prompt_value)) + + def set_model_prompt_context( + self, run_id: UUID, prompt_context: _PromptContext + ) -> None: + self._model_prompt_contexts[run_id] = prompt_context + + def pop_model_prompt_context(self, run_id: UUID) -> _PromptContext | None: + return self._model_prompt_contexts.pop(run_id, None) def consume_prompt_context( self, parent_run_id: UUID | None diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_invocation_manager.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_invocation_manager.py index bb0524c93..5a29e4122 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_invocation_manager.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_invocation_manager.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 # tests/test_invocation_manager.py +import gc import uuid from unittest import mock @@ -319,3 +320,37 @@ def test_unpublished_prompt_context_is_discarded_on_delete(invocation_manager): invocation_manager.delete_invocation_state(prompt_id) assert invocation_manager.consume_prompt_context(parent_id) is None + + +def test_prompt_value_context_lives_as_long_as_the_value(invocation_manager): + class Value: + pass + + value = Value() + context = _PromptContext("greeting", {}, "3") + invocation_manager.set_prompt_value_context(value, context) + + assert invocation_manager.get_prompt_value_context(value) == context + del value + gc.collect() + assert invocation_manager._prompt_values == {} + + +def test_prompt_value_context_skips_values_without_weak_references( + invocation_manager, +): + value = ("not", "weak-referenceable") + invocation_manager.set_prompt_value_context( + value, _PromptContext("greeting", {}) + ) + + assert invocation_manager.get_prompt_value_context(value) is None + + +def test_model_prompt_context_is_taken_once(invocation_manager): + run_id = uuid.uuid4() + context = _PromptContext("greeting", {}) + invocation_manager.set_model_prompt_context(run_id, context) + + assert invocation_manager.pop_model_prompt_context(run_id) == context + assert invocation_manager.pop_model_prompt_context(run_id) is None diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_prompt_template.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_prompt_template.py new file mode 100644 index 000000000..2f746b379 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_prompt_template.py @@ -0,0 +1,169 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import asyncio + +import pytest +from langchain_core.language_models.fake_chat_models import ( + FakeMessagesListChatModel, +) +from langchain_core.messages import AIMessage +from langchain_core.prompts import ChatPromptTemplate, PromptTemplate +from langchain_core.runnables import RunnableLambda + +from opentelemetry.instrumentation.genai.langchain import ( + LangChainInstrumentor, +) +from opentelemetry.semconv._incubating.attributes import gen_ai_attributes +from opentelemetry.test_util_genai.instrumentor import instrument + +_PROMPT_VERSION = "gen_ai.prompt.version" + + +@pytest.fixture +def instrumented(tracer_provider, meter_provider, logger_provider): + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + content_capture="SPAN_ONLY", + ): + yield + + +def _model() -> FakeMessagesListChatModel: + return FakeMessagesListChatModel(responses=[AIMessage(content="ok")]) + + +def _call(model_name: str) -> dict: + return {"metadata": {"ls_model_name": model_name}} + + +def _prompt_attributes(span_exporter) -> dict[str, dict]: + return { + span.attributes[gen_ai_attributes.GEN_AI_REQUEST_MODEL]: { + key: value + for key, value in span.attributes.items() + if key.startswith("gen_ai.prompt.") + } + for span in span_exporter.get_finished_spans() + if span.attributes.get(gen_ai_attributes.GEN_AI_OPERATION_NAME) + == "chat" + } + + +def _greeting() -> ChatPromptTemplate: + prompt = ChatPromptTemplate.from_messages([("human", "Say hi to {who}.")]) + prompt.name = "greeting" + prompt.metadata = {"prompt_version": "3"} + return prompt + + +_GREETING = { + gen_ai_attributes.GEN_AI_PROMPT_NAME: "greeting", + _PROMPT_VERSION: "3", + "gen_ai.prompt.variable.who": "Ada", +} + + +@pytest.mark.usefixtures("instrumented") +def test_prompt_formatted_outside_a_chain(span_exporter): + prompt_value = _greeting().invoke({"who": "Ada"}) + _model().invoke(prompt_value, config=_call("invoke")) + list(_model().stream(prompt_value, config=_call("stream"))) + + assert _prompt_attributes(span_exporter) == { + "invoke": _GREETING, + "stream": _GREETING, + } + + +@pytest.mark.usefixtures("instrumented") +def test_async_prompt_formatted_outside_a_chain(span_exporter): + async def run() -> None: + prompt_value = await _greeting().ainvoke({"who": "Ada"}) + await _model().ainvoke(prompt_value, config=_call("ainvoke")) + async for _ in _model().astream(prompt_value, config=_call("astream")): + pass + + asyncio.run(run()) + + assert _prompt_attributes(span_exporter) == { + "ainvoke": _GREETING, + "astream": _GREETING, + } + + +@pytest.mark.usefixtures("instrumented") +def test_each_model_call_gets_the_prompt_that_built_its_input(span_exporter): + summary = PromptTemplate.from_template("Summarize: {text}", name="summary") + + def node(_: dict) -> str: + # Both prompts run under this node before either model call. + greeting_value = _greeting().invoke({"who": "Ada"}) + summary_value = summary.invoke({"text": "OTel"}) + _model().invoke(summary_value, config=_call("summary-call")) + _model().invoke(greeting_value, config=_call("greeting-call")) + return "done" + + RunnableLambda(node).invoke({}) + + assert _prompt_attributes(span_exporter) == { + "summary-call": { + gen_ai_attributes.GEN_AI_PROMPT_NAME: "summary", + "gen_ai.prompt.variable.text": "OTel", + }, + "greeting-call": _GREETING, + } + + +@pytest.mark.usefixtures("instrumented") +def test_hub_prompt_reports_repo_and_commit(span_exporter): + prompt = ChatPromptTemplate.from_messages( + [("human", "Weather in {city}?")] + ) + prompt.metadata = { + "lc_hub_owner": "acme", + "lc_hub_repo": "weather-now", + "lc_hub_commit_hash": "9f2c1ab", + } + + (prompt | _model()).invoke({"city": "Oslo"}, config=_call("hub")) + + assert _prompt_attributes(span_exporter)["hub"] == { + gen_ai_attributes.GEN_AI_PROMPT_NAME: "weather-now", + _PROMPT_VERSION: "9f2c1ab", + "gen_ai.prompt.variable.city": "Oslo", + } + + +class _ReviewPrompt(ChatPromptTemplate): + pass + + +@pytest.mark.usefixtures("instrumented") +def test_prompt_template_subclass_is_detected(span_exporter): + named = _ReviewPrompt.from_messages([("human", "Review {code}")]) + named.name = "code-review" + unnamed = _ReviewPrompt.from_messages([("human", "Review {code}")]) + unnamed.metadata = {"prompt_version": "2"} + + (named | _model()).invoke({"code": "x = 1"}, config=_call("named")) + (unnamed | _model()).invoke({"code": "x = 1"}, config=_call("unnamed")) + + attributes = _prompt_attributes(span_exporter) + assert attributes["named"] == { + gen_ai_attributes.GEN_AI_PROMPT_NAME: "code-review", + "gen_ai.prompt.variable.code": "x = 1", + } + # A version names nothing without a prompt name. + assert attributes["unnamed"] == {"gen_ai.prompt.variable.code": "x = 1"} + + +@pytest.mark.usefixtures("instrumented") +def test_model_input_not_built_from_a_prompt(span_exporter): + _greeting().invoke({"who": "Ada"}) + _model().invoke("Plain message.", config=_call("plain")) + + assert _prompt_attributes(span_exporter) == {"plain": {}}