diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/CHANGELOG.md b/instrumentation-genai/opentelemetry-instrumentation-langchain/CHANGELOG.md index 767dfcc7ed..9878790377 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/CHANGELOG.md +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +- Add LangChain workflow and agent span support. Also refactor LLM invocation. + ([#4449](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4449)) - Fix compatibility with wrapt 2.x by using positional arguments in `wrap_function_wrapper()` calls ([#4445](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4445)) - Added span support for genAI langchain llm invocation. diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/main.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/main.py new file mode 100644 index 0000000000..bbbedf5056 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/main.py @@ -0,0 +1,104 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +""" +ReAct agent example built with LangGraph. + +Uses LangGraph's prebuilt create_react_agent with simple calculator tools. +OpenTelemetry LangChain instrumentation traces the LLM calls made by the agent. +""" + +from uuid import uuid4 + +from langchain_core.messages import HumanMessage +from langchain_core.tools import tool +from langchain_openai import ChatOpenAI +from langgraph.prebuilt import create_react_agent + +from opentelemetry import _logs, metrics, trace +from opentelemetry.exporter.otlp.proto.grpc._log_exporter import ( + OTLPLogExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( + OTLPMetricExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter, +) +from opentelemetry.instrumentation.langchain import LangChainInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +# Configure tracing +trace.set_tracer_provider(TracerProvider()) +span_processor = BatchSpanProcessor(OTLPSpanExporter()) +trace.get_tracer_provider().add_span_processor(span_processor) + +# Configure logging +_logs.set_logger_provider(LoggerProvider()) +_logs.get_logger_provider().add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter()) +) + +# Configure metrics +metrics.set_meter_provider( + MeterProvider( + metric_readers=[ + PeriodicExportingMetricReader( + OTLPMetricExporter(), + ), + ] + ) +) + + +@tool +def multiply(a: float, b: float) -> float: + """Multiply two numbers together.""" + return a * b + + +@tool +def add(a: float, b: float) -> float: + """Add two numbers together.""" + return a + b + + +def main(): + LangChainInstrumentor().instrument() + + llm = ChatOpenAI( + model="gpt-3.5-turbo", + temperature=0.1, + max_tokens=100, + top_p=0.9, + seed=100, + ) + + session_id = str(uuid4()) + agent = create_react_agent(llm, tools=[multiply, add]).with_config( + { + "metadata": { + "agent_name": "coordinator", + "session_id": session_id, + }, + } + ) + + result = agent.invoke( + {"messages": [HumanMessage(content="What is (3 * 4) + 7?")]} + ) + + print("Agent output:") + for msg in result["messages"]: + print(f" {type(msg).__name__}: {msg.content}") + + LangChainInstrumentor().uninstrument() + + +if __name__ == "__main__": + main() diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/requirements.txt b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/requirements.txt new file mode 100644 index 0000000000..88f3320d42 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/requirements.txt @@ -0,0 +1,5 @@ +langchain==0.3.21 +langchain_openai +langgraph +opentelemetry-sdk>=1.39.0 +opentelemetry-exporter-otlp-proto-grpc>=1.39.0 \ No newline at end of file diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/single_node_agent.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/single_node_agent.py new file mode 100644 index 0000000000..72a9b5dcb7 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/single_node_agent.py @@ -0,0 +1,122 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +""" +Single-node agent example built with StateGraph. + +A single ReAct agent answers arithmetic questions using calculator tools. +OpenTelemetry LangChain instrumentation traces all LLM calls. +""" + +from uuid import uuid4 + +from langchain.agents import create_agent +from langchain_core.messages import HumanMessage +from langchain_core.tools import tool +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, MessagesState, StateGraph + +from opentelemetry import _logs, metrics, trace +from opentelemetry.exporter.otlp.proto.grpc._log_exporter import ( + OTLPLogExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( + OTLPMetricExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter, +) +from opentelemetry.instrumentation.langchain import LangChainInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +# Configure tracing +trace.set_tracer_provider(TracerProvider()) +trace.get_tracer_provider().add_span_processor( + BatchSpanProcessor(OTLPSpanExporter()) +) + +# Configure logging +_logs.set_logger_provider(LoggerProvider()) +_logs.get_logger_provider().add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter()) +) + +# Configure metrics +metrics.set_meter_provider( + MeterProvider( + metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter())] + ) +) + + +# --- Tools ---------------------------------------------------------------- + + +@tool +def multiply(a: float, b: float) -> float: + """Multiply two numbers.""" + return a * b + + +@tool +def add(a: float, b: float) -> float: + """Add two numbers.""" + return a + b + + +# --- Graph ---------------------------------------------------------------- + + +def build_single_node_graph(llm: ChatOpenAI): + session_id = str(uuid4()) + + agent = create_agent( + llm, tools=[multiply, add], name="math_agent" + ).with_config( + { + "metadata": { + "agent_name": "math_agent", + "session_id": session_id, + }, + } + ) + + def run_agent(state: MessagesState) -> dict: + result = agent.invoke({"messages": state["messages"]}) + return {"messages": result["messages"]} + + builder = StateGraph(MessagesState) + builder.add_node("math_agent", run_agent) + builder.add_edge(START, "math_agent") + builder.add_edge("math_agent", END) + + return builder.compile() + + +def main(): + LangChainInstrumentor().instrument() + + llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.1, seed=100) + graph = build_single_node_graph(llm) + + questions = [ + "What is 12 multiplied by 7?", + "What is 15 plus 27?", + ] + + for question in questions: + print(f"\nQuestion: {question}") + result = graph.invoke({"messages": [HumanMessage(content=question)]}) + last = result["messages"][-1] + print(f" Answer: {last.content}") + + LangChainInstrumentor().uninstrument() + + +if __name__ == "__main__": + main() diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/main.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/main.py new file mode 100644 index 0000000000..b7de4c3b59 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/main.py @@ -0,0 +1,144 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +""" +LangGraph StateGraph example with two LLM nodes. + +Graph topology: + + START → researcher → summariser → END + +Steps: + 1. *researcher* – gathers factual background on the user's question. + 2. *summariser* – condenses the researcher's output into a concise answer. + +OpenTelemetry LangChain instrumentation traces both LLM calls. +""" + +from typing import Annotated + +from langchain_core.messages import HumanMessage, SystemMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph +from langgraph.graph.message import add_messages +from typing_extensions import TypedDict + +from opentelemetry import _logs, metrics, trace +from opentelemetry.exporter.otlp.proto.grpc._log_exporter import ( + OTLPLogExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( + OTLPMetricExporter, +) +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter, +) +from opentelemetry.instrumentation.langchain import LangChainInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +# Configure tracing +trace.set_tracer_provider(TracerProvider()) +trace.get_tracer_provider().add_span_processor( + BatchSpanProcessor(OTLPSpanExporter()) +) + +# Configure logging +_logs.set_logger_provider(LoggerProvider()) +_logs.get_logger_provider().add_log_record_processor( + BatchLogRecordProcessor(OTLPLogExporter()) +) + +# Configure metrics +metrics.set_meter_provider( + MeterProvider( + metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter())] + ) +) + + +class GraphState(TypedDict): + """State shared across all graph nodes.""" + + messages: Annotated[list, add_messages] + research: str + + +def build_graph(llm: ChatOpenAI): + """Build a StateGraph with a researcher node and a summariser node.""" + + def researcher(state: GraphState) -> dict: + """Gather factual background on the last user message.""" + response = llm.invoke( + [ + SystemMessage( + content="You are a research assistant. Provide 2-3 factual sentences." + ), + HumanMessage(content=state["messages"][-1].content), + ] + ) + return { + "research": response.content, + "messages": [response], + } + + def summariser(state: GraphState) -> dict: + """Condense the researcher's output into one concise sentence.""" + response = llm.invoke( + [ + SystemMessage( + content="You are an expert summariser. Condense the text below into one clear sentence." + ), + HumanMessage(content=state["research"]), + ] + ) + return {"messages": [response]} + + builder = StateGraph(GraphState) + builder.add_node("researcher", researcher) + builder.add_node("summariser", summariser) + + builder.add_edge(START, "researcher") + builder.add_edge("researcher", "summariser") + builder.add_edge("summariser", END) + + return builder.compile() + + +def main(): + LangChainInstrumentor().instrument() + + llm = ChatOpenAI( + model="gpt-3.5-turbo", + temperature=0.1, + max_tokens=200, + seed=42, + ) + + graph = build_graph(llm) + + question = "What is the capital of France?" + print(f"Question: {question}\n") + + result = graph.invoke( + { + "messages": [HumanMessage(content=question)], + "research": "", + } + ) + + print("Research output:") + print(f" {result['research']}\n") + + print("Final summary:") + print(f" {result['messages'][-1].content}") + + LangChainInstrumentor().uninstrument() + + +if __name__ == "__main__": + main() diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/requirements.txt b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/requirements.txt new file mode 100644 index 0000000000..656212ca31 --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/requirements.txt @@ -0,0 +1,5 @@ +langchain==0.3.21 +langchain_openai +langgraph +opentelemetry-sdk>=1.39.0 +opentelemetry-exporter-otlp-proto-grpc>=1.39.0 diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml b/instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml index 80a406b7da..a6c61cc0ce 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/pyproject.toml @@ -25,7 +25,8 @@ classifiers = [ "Programming Language :: Python :: 3.14", ] dependencies = [ - "opentelemetry-instrumentation ~= 0.57b0", + "opentelemetry-instrumentation ~= 0.60b0", + "opentelemetry-util-genai >= 0.4b0", ] [project.optional-dependencies] diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py index a024ca530b..e181739c64 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/callback_handler.py @@ -13,11 +13,23 @@ from opentelemetry.instrumentation.langchain.invocation_manager import ( _InvocationManager, ) +from opentelemetry.instrumentation.langchain.operation_mapping import ( + OperationName, + classify_chain_run, + resolve_agent_name, +) +from opentelemetry.instrumentation.langchain.utils import ( + make_input_message, + make_last_output_message, +) from opentelemetry.util.genai.handler import TelemetryHandler +from opentelemetry.util.genai.invocation import ( + AgentInvocation, + InferenceInvocation, + WorkflowInvocation, +) from opentelemetry.util.genai.types import ( - Error, InputMessage, - LLMInvocation, # TODO: migrate to InferenceInvocation MessagePart, OutputMessage, Text, @@ -34,6 +46,139 @@ def __init__(self, telemetry_handler: TelemetryHandler) -> None: self._telemetry_handler = telemetry_handler self._invocation_manager = _InvocationManager() + def on_chain_start( + self, + serialized: dict[str, Any], + inputs: dict[str, Any], + *, + run_id: UUID, + parent_run_id: Optional[UUID] = None, + tags: Optional[list[str]] = None, + metadata: Optional[dict[str, Any]] = None, + **kwargs: Any, + ) -> Any: + operation = classify_chain_run( + serialized, metadata, kwargs, parent_run_id + ) + + if operation == OperationName.INVOKE_WORKFLOW: + workflow_name = kwargs.get("name") or serialized.get("name") + workflow_name_override = ( + metadata.get("workflow_name") if metadata else None + ) + workflow = self._telemetry_handler.start_workflow( + name=workflow_name_override or workflow_name + ) + workflow.input_messages = make_input_message(inputs) + self._invocation_manager.add_invocation_state( + run_id, parent_run_id, workflow + ) + elif operation == OperationName.INVOKE_AGENT: + # agent name passed by the user + suggested_agent_name = resolve_agent_name( + serialized, metadata, kwargs + ) + # find if there is an agent already + agent_invocation = self._find_nearest_agent(parent_run_id) + agent_invocation_name = ( + agent_invocation.agent_name if agent_invocation else None + ) + if suggested_agent_name: + suggested_agent_name_lower = suggested_agent_name.lower() + agent_invocation_name_lower = ( + agent_invocation_name.lower() + if agent_invocation_name + else None + ) + if suggested_agent_name_lower != agent_invocation_name_lower: + agent = self._telemetry_handler.start_invoke_local_agent( + provider=metadata.get("ls_provider", "unknown") + if metadata + else "unknown", + ) + agent.agent_name = suggested_agent_name + agent.input_messages = make_input_message(inputs) + + if metadata: + agent.agent_id = metadata.get("agent_id") + agent.agent_description = metadata.get( + "agent_description" + ) + + for key in ( + "thread_id", + "session_id", + "conversation_id", + ): + conv_id = metadata.get(key) + if conv_id: + agent.conversation_id = conv_id + break + + self._invocation_manager.add_invocation_state( + run_id, parent_run_id, agent + ) + else: + # We create invoke_agent span for the initial chain for agent. All follow-up chains invoked for agent invocation will not create agent span. + self._invocation_manager.add_invocation_state( + run_id, parent_run_id, None + ) + else: + # No agent name could be resolved; still register the run_id so that + # parent-child traversal (e.g. _find_nearest_agent) is not broken for + # any children of this node. + self._invocation_manager.add_invocation_state( + run_id, parent_run_id, None + ) + else: + # For unclassified chains, we still want to track them in the invocation manager to maintain the parent-child relationships, even though we won't create spans for them. + self._invocation_manager.add_invocation_state( + run_id, parent_run_id, None + ) + + def on_chain_end( + self, + outputs: dict[str, Any], + *, + run_id: UUID, + parent_run_id: Optional[UUID] = None, + **kwargs: Any, + ) -> Any: + invocation = self._invocation_manager.get_invocation(run_id=run_id) + if invocation is None or not isinstance( + invocation, (WorkflowInvocation, AgentInvocation) + ): + # If the invocation does not exist, we cannot set attributes or end it + self._invocation_manager.delete_invocation_state(run_id) + return + + invocation.output_messages = make_last_output_message(outputs) + + invocation.stop() + + if not invocation.span.is_recording(): + self._invocation_manager.delete_invocation_state(run_id) + + def on_chain_error( + self, + error: BaseException, + *, + run_id: UUID, + parent_run_id: Optional[UUID] = None, + **kwargs: Any, + ) -> Any: + invocation = self._invocation_manager.get_invocation(run_id=run_id) + if invocation is None or not isinstance( + invocation, (WorkflowInvocation, AgentInvocation) + ): + # If the invocation does not exist, we cannot set attributes or end it + self._invocation_manager.delete_invocation_state(run_id) + return + + invocation.fail(error) + if not invocation.span.is_recording(): + self._invocation_manager.delete_invocation_state(run_id=run_id) + def on_chat_model_start( self, serialized: dict[str, Any], @@ -129,25 +274,22 @@ def on_chat_model_start( ) ) - llm_invocation = LLMInvocation( + llm_invocation = self._telemetry_handler.start_inference( + provider, 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 ) + llm_invocation.input_messages = input_messages + llm_invocation.top_p = top_p + llm_invocation.frequency_penalty = frequency_penalty + llm_invocation.presence_penalty = presence_penalty + llm_invocation.stop_sequences = stop_sequences + llm_invocation.seed = seed + llm_invocation.temperature = temperature + llm_invocation.max_tokens = max_tokens self._invocation_manager.add_invocation_state( run_id=run_id, parent_run_id=parent_run_id, - invocation=llm_invocation, # pyright: ignore[reportArgumentType] + invocation=llm_invocation, ) def on_llm_end( @@ -161,7 +303,7 @@ def on_llm_end( llm_invocation = self._invocation_manager.get_invocation(run_id=run_id) if llm_invocation is None or not isinstance( llm_invocation, - LLMInvocation, + InferenceInvocation, ): # If the invocation does not exist, we cannot set attributes or end it return @@ -236,10 +378,8 @@ def on_llm_end( if response_id is not None: llm_invocation.response_id = str(response_id) - llm_invocation = self._telemetry_handler.stop_llm( - invocation=llm_invocation - ) - if llm_invocation.span and not llm_invocation.span.is_recording(): + llm_invocation.stop() + if not llm_invocation.span.is_recording(): self._invocation_manager.delete_invocation_state(run_id=run_id) def on_llm_error( @@ -253,14 +393,24 @@ def on_llm_error( llm_invocation = self._invocation_manager.get_invocation(run_id=run_id) if llm_invocation is None or not isinstance( llm_invocation, - LLMInvocation, + InferenceInvocation, ): # 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(): + llm_invocation.fail(error) + if not llm_invocation.span.is_recording(): self._invocation_manager.delete_invocation_state(run_id=run_id) + + def _find_nearest_agent( + self, run_id: Optional[UUID] + ) -> Optional[AgentInvocation]: + current = run_id + visited: set[UUID] = set() + while current is not None and current not in visited: + visited.add(current) + entity = self._invocation_manager.get_invocation(current) + if isinstance(entity, AgentInvocation): + return entity + current = self._invocation_manager.get_parent_run_id(current) + return None diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/invocation_manager.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/invocation_manager.py index 996b5069ed..9a340c0f4c 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/invocation_manager.py +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/invocation_manager.py @@ -12,8 +12,10 @@ @dataclass class _InvocationState: - invocation: GenAIInvocation + invocation: Optional[GenAIInvocation] children: List[UUID] = field(default_factory=lambda: list()) + parent_run_id: Optional[UUID] = None + ended: bool = False class _InvocationManager: @@ -28,23 +30,45 @@ def add_invocation_state( self, run_id: UUID, parent_run_id: Optional[UUID], - invocation: GenAIInvocation, - ): + invocation: Optional[GenAIInvocation], + ) -> None: invocation_state = _InvocationState(invocation=invocation) - self._invocations[run_id] = invocation_state if parent_run_id is not None and parent_run_id in self._invocations: + invocation_state.parent_run_id = parent_run_id + parent_invocation_state = self._invocations[parent_run_id] parent_invocation_state.children.append(run_id) + self._invocations[run_id] = invocation_state + def get_invocation(self, run_id: UUID) -> Optional[GenAIInvocation]: invocation_state = self._invocations.get(run_id) return invocation_state.invocation if invocation_state else None + def get_parent_run_id(self, run_id: UUID) -> Optional[UUID]: + invocation_state = self._invocations.get(run_id) + return invocation_state.parent_run_id if invocation_state else None + def delete_invocation_state(self, run_id: UUID) -> None: invocation_state = self._invocations.get(run_id) if not invocation_state: return - for child_id in list(invocation_state.children): - self._invocations.pop(child_id, None) + + invocation_state.ended = True + + # Defer removal if any children are still live, so upward traversal + # (e.g. _find_nearest_agent) can still walk through this node. + if any(c in self._invocations for c in invocation_state.children): + return + self._invocations.pop(run_id, None) + + # Propagate cleanup upward: if the parent has already ended and has no + # more live children, it can now be removed too. + if invocation_state.parent_run_id: + parent_state = self._invocations.get( + invocation_state.parent_run_id + ) + if parent_state is not None and parent_state.ended: + self.delete_invocation_state(invocation_state.parent_run_id) diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/operation_mapping.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/operation_mapping.py new file mode 100644 index 0000000000..45f0a879fc --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/operation_mapping.py @@ -0,0 +1,217 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Callback-to-semconv operation mapping for LangChain callbacks. + +Maps each LangChain callback to the correct GenAI semantic convention +operation name. Direct callbacks (``on_chat_model_start``, +``on_llm_start``, ``on_tool_start``, ``on_retriever_start``) have a +fixed 1-to-1 mapping. ``on_chain_start`` requires heuristic +classification because LangChain emits this callback for agents, +workflows, and internal plumbing alike. +""" + +from __future__ import annotations + +from typing import Any, Optional +from uuid import UUID + +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) + +__all__ = [ + "OperationName", + "classify_chain_run", + "resolve_agent_name", + "should_ignore_chain", +] + +# --------------------------------------------------------------------------- +# Operation name constants (sourced from the GenAI semconv enum where +# available, with string fallbacks for values not yet in the enum). +# --------------------------------------------------------------------------- + + +class OperationName: + """Canonical GenAI semantic convention operation names.""" + + INVOKE_AGENT: str = GenAI.GenAiOperationNameValues.INVOKE_AGENT.value + # invoke_workflow is not yet in the semconv enum; use the expected + # string value so the mapping is forward-compatible. + INVOKE_WORKFLOW: str = "invoke_workflow" + + +# --------------------------------------------------------------------------- +# LangGraph markers – names and prefixes produced by LangGraph that must +# be recognized when classifying ``on_chain_start`` callbacks. +# --------------------------------------------------------------------------- + +LANGGRAPH_NODE_KEY = "langgraph_node" +LANGGRAPH_START_NODE = "__start__" +MIDDLEWARE_PREFIX = "Middleware." +LANGGRAPH_IDENTIFIER = "LangGraph" + +# Metadata keys used by callers to override classification. +_META_AGENT_SPAN = "otel_agent_span" +_META_WORKFLOW_SPAN = "otel_workflow_span" +_META_AGENT_NAME = "agent_name" +_META_AGENT_TYPE = "agent_type" +_META_OTEL_TRACE = "otel_trace" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def resolve_agent_name( + serialized: dict[str, Any], + metadata: Optional[dict[str, Any]], + kwargs: dict[str, Any], +) -> Optional[str]: + """Derive the best-effort agent name from callback arguments. + + Checks (in priority order): + 1. ``metadata["agent_name"]`` + 2. ``kwargs["name"]`` + 3. ``serialized["name"]`` + 4. ``metadata["langgraph_node"]`` (if present and not a start node) + """ + if metadata: + name = metadata.get(_META_AGENT_NAME) + if name: + return str(name) + + name = kwargs.get("name") + if name: + return str(name) + + name = serialized.get("name") if serialized else None + if name: + return str(name) + + if metadata: + node = metadata.get(LANGGRAPH_NODE_KEY) + if node and node != LANGGRAPH_START_NODE: + return str(node) + + return None + + +def _has_agent_signals(metadata: Optional[dict[str, Any]]) -> bool: + """Return True when metadata contains any signal that the chain is an agent.""" + if not metadata: + return False + return bool( + metadata.get(_META_AGENT_SPAN) + or metadata.get(_META_AGENT_NAME) + or metadata.get(_META_AGENT_TYPE) + ) + + +def _looks_like_workflow( + serialized: dict[str, Any], + metadata: Optional[dict[str, Any]], + parent_run_id: Optional[UUID], +) -> bool: + """Return True if the chain looks like a top-level workflow/graph.""" + if parent_run_id is not None: + return False + + # An explicit workflow override is authoritative. + if metadata and metadata.get(_META_WORKFLOW_SPAN): + return True + + # Heuristic: check for LangGraph identifier in the serialized repr. + if serialized: + name = serialized.get("name", "") + graph_id = ( + serialized.get("graph", {}).get("id", "") + if isinstance(serialized.get("graph"), dict) + else "" + ) + return LANGGRAPH_IDENTIFIER in name or LANGGRAPH_IDENTIFIER in graph_id + + return True + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def should_ignore_chain( + metadata: Optional[dict[str, Any]], + agent_name: Optional[str], + kwargs: dict[str, Any], +) -> bool: + """Return True if the chain callback should be silently suppressed. + + Suppression happens when: + * The node is the LangGraph ``__start__`` node. + * The name carries the ``Middleware.`` prefix. + * ``metadata["otel_trace"]`` is explicitly ``False``. + * ``metadata["otel_agent_span"]`` is explicitly ``False`` and no other + agent signals are present. + """ + if metadata: + node = metadata.get(LANGGRAPH_NODE_KEY) + if node == LANGGRAPH_START_NODE: + return True + + if metadata.get(_META_OTEL_TRACE) is False: + return True + + if ( + metadata.get(_META_AGENT_SPAN) is False + and not metadata.get(_META_AGENT_NAME) + and not metadata.get(_META_AGENT_TYPE) + ): + return True + + if agent_name and agent_name.startswith(MIDDLEWARE_PREFIX): + return True + + name_from_kwargs = kwargs.get("name", "") + if isinstance(name_from_kwargs, str) and name_from_kwargs.startswith( + MIDDLEWARE_PREFIX + ): + return True + + return False + + +def classify_chain_run( + serialized: dict[str, Any], + metadata: Optional[dict[str, Any]], + kwargs: dict[str, Any], + parent_run_id: Optional[UUID] = None, +) -> Optional[str]: + """Classify a ``on_chain_start`` callback into a semconv operation. + + Returns one of the :class:`OperationName` constants, or ``None`` when + the chain should be suppressed (no span emitted). + + Classification order: + 1. Check for explicit suppression signals. + 2. Check for agent signals → ``invoke_agent``. + 3. Check for workflow signals → ``invoke_workflow``. + 4. Default: ``None`` (suppress – unclassified chains are not emitted). + """ + agent_name = resolve_agent_name(serialized, metadata, kwargs) + + # 1. Suppress known noise. + if should_ignore_chain(metadata, agent_name, kwargs): + return None + + # 2. Agent detection. + if _has_agent_signals(metadata): + return OperationName.INVOKE_AGENT + + # 3. Workflow / orchestration detection. + if _looks_like_workflow(serialized, metadata, parent_run_id): + return OperationName.INVOKE_WORKFLOW + + # 4. Default: suppress unclassified chains. + return None diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/utils.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/utils.py new file mode 100644 index 0000000000..b5de9323bf --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/src/opentelemetry/instrumentation/langchain/utils.py @@ -0,0 +1,91 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import json +from typing import Any, Optional, cast + +from langchain_core.messages import AIMessage + +from opentelemetry.util.genai.types import ( + InputMessage, + OutputMessage, + Text, +) + + +def make_input_message(data: Any) -> list[InputMessage]: + """Create structured input message with full data as JSON.""" + if not isinstance(data, dict): + return [] + data_dict = cast(dict[str, Any], data) + input_messages: list[InputMessage] = [] + messages: Any = data_dict.get("messages") + if messages is not None: + for msg in messages: + content: Any = getattr(msg, "content", "") + if content and isinstance(content, str): + input_message = InputMessage( + role="user", parts=[Text(content)] + ) + input_messages.append(input_message) + return input_messages + # Fallback: serialize non-message state fields as input. + # Common in LangGraph where nodes use structured state fields + # (e.g., user_query) rather than a message list. + exclude_keys = {"messages", "intermediate_steps"} + input_data: dict[str, Any] = { + k: v + for k, v in data_dict.items() + if k not in exclude_keys and v is not None + } + if input_data: + serialized = serialize(input_data) + if serialized: + return [InputMessage(role="user", parts=[Text(serialized)])] + return input_messages + + +def make_output_message(data: dict[str, Any]) -> list[OutputMessage]: + """Create structured output message with full data as JSON.""" + output_messages: list[OutputMessage] = [] + messages: Any = data.get("messages") + if messages is None: + return [] + for msg in messages: + content: Any = getattr(msg, "content", "") + if content and isinstance(msg, AIMessage) and isinstance(content, str): + output_message = OutputMessage( + role="assistant", + parts=[Text(content)], + finish_reason="stop", + ) + output_messages.append(output_message) + return output_messages + + +def make_last_output_message(data: dict[str, Any]) -> list[OutputMessage]: + """Extract only the last AI message as the output. + + For Workflow and AgentInvocation spans, the final AI message best represents + the actual output. Intermediate AI messages (e.g., tool-call decisions) are + already captured in child LLM invocation spans. + """ + all_messages = make_output_message(data) + if all_messages: + return [all_messages[-1]] + return [] + + +def serialize(obj: Any) -> Optional[str]: + """Serialize object to JSON string. + + Uses default=str to handle non-JSON-serializable objects (like LangChain + message objects) by converting them to their string representation while + keeping the overall structure as valid JSON. + """ + if obj is None: + return None + try: + return json.dumps(obj, ensure_ascii=False, default=str) + except (TypeError, ValueError): + return None diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/conftest.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/conftest.py index ceb4f247a1..6a9106ac79 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/conftest.py +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/conftest.py @@ -211,9 +211,10 @@ def deserialize(cassette_string): return yaml.load(cassette_string, Loader=yaml.Loader) -@pytest.fixture(scope="module", autouse=True) +@pytest.fixture(scope="function", autouse=True) def fixture_vcr(vcr): - vcr.register_serializer("yaml", PrettyPrintJSONBody) + if vcr is not None: + vcr.register_serializer("yaml", PrettyPrintJSONBody) return vcr diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_callback_handler.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_callback_handler.py new file mode 100644 index 0000000000..26bbcffd7e --- /dev/null +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_callback_handler.py @@ -0,0 +1,995 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for on_chain_start / on_chain_end / on_chain_error in +OpenTelemetryLangChainCallbackHandler. + +All TelemetryHandler interactions are mocked so that these tests exercise only +the callback-handler logic and the invocation-manager bookkeeping. +""" + +import uuid +from unittest import mock + +from langchain_core.messages import AIMessage, HumanMessage + +from opentelemetry.instrumentation.langchain.callback_handler import ( + OpenTelemetryLangChainCallbackHandler, +) +from opentelemetry.instrumentation.langchain.utils import ( + make_input_message, + make_last_output_message, + make_output_message, + serialize, +) +from opentelemetry.util.genai.invocation import ( + AgentInvocation, + WorkflowInvocation, +) +from opentelemetry.util.genai.types import InputMessage, OutputMessage, Text + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_handler(): + """Return a handler wired to a MagicMock TelemetryHandler.""" + telemetry = mock.MagicMock() + + # start_workflow returns a mock WorkflowInvocation + workflow_inv = mock.MagicMock(spec=WorkflowInvocation) + workflow_inv.span = mock.MagicMock() + workflow_inv.span.is_recording.return_value = False + telemetry.start_workflow.return_value = workflow_inv + + # start_invoke_local_agent returns a mock AgentInvocation + agent_inv = mock.MagicMock(spec=AgentInvocation) + agent_inv.span = mock.MagicMock() + agent_inv.span.is_recording.return_value = False + telemetry.start_invoke_local_agent.return_value = agent_inv + + handler = OpenTelemetryLangChainCallbackHandler(telemetry) + return handler, telemetry, workflow_inv, agent_inv + + +def _run_id(): + return uuid.uuid4() + + +# --------------------------------------------------------------------------- +# on_chain_start – INVOKE_WORKFLOW +# --------------------------------------------------------------------------- + + +class TestOnChainStartWorkflow: + def test_workflow_span_created(self): + handler, telemetry, workflow_inv, _ = _make_handler() + run_id = _run_id() + + # LangGraph graph serialized dict triggers workflow classification + handler.on_chain_start( + serialized={"name": "LangGraph", "id": ["langgraph"]}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + telemetry.start_workflow.assert_called_once() + assert ( + handler._invocation_manager.get_invocation(run_id) is workflow_inv + ) + + def test_workflow_name_from_serialized(self): + handler, telemetry, _, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "MyLangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + telemetry.start_workflow.assert_called_once_with(name="MyLangGraph") + + def test_workflow_name_overridden_by_metadata(self): + handler, telemetry, _, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "MyLangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"workflow_name": "custom_workflow"}, + ) + + telemetry.start_workflow.assert_called_once_with( + name="custom_workflow" + ) + + def test_workflow_registered_in_invocation_manager(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + assert ( + handler._invocation_manager.get_invocation(run_id) is workflow_inv + ) + + +# --------------------------------------------------------------------------- +# on_chain_start – INVOKE_AGENT +# --------------------------------------------------------------------------- + + +class TestOnChainStartAgent: + def test_new_agent_span_created(self): + handler, telemetry, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent", "ls_provider": "openai"}, + ) + + telemetry.start_invoke_local_agent.assert_called_once_with( + provider="openai" + ) + assert agent_inv.agent_name == "math_agent" + assert handler._invocation_manager.get_invocation(run_id) is agent_inv + + def test_agent_metadata_set(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={ + "agent_name": "math_agent", + "agent_id": "agent-123", + "agent_description": "does math", + "thread_id": "thread-abc", + }, + ) + + assert agent_inv.agent_id == "agent-123" + assert agent_inv.agent_description == "does math" + assert agent_inv.conversation_id == "thread-abc" + + def test_conversation_id_prefers_thread_id_over_session_id(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={ + "agent_name": "math_agent", + "thread_id": "t1", + "session_id": "s1", + }, + ) + + assert agent_inv.conversation_id == "t1" + + def test_duplicate_agent_name_does_not_create_new_span(self): + """When the nearest ancestor already has the same agent name, no new + AgentInvocation span is created; the run is still tracked with None.""" + handler, telemetry, _, agent_inv = _make_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + # Register the parent agent + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + telemetry.start_invoke_local_agent.reset_mock() + + # A child chain with the same agent name should NOT create a new span + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"agent_name": "math_agent"}, + ) + + telemetry.start_invoke_local_agent.assert_not_called() + assert handler._invocation_manager.get_invocation(child_run_id) is None + + def test_different_agent_name_creates_new_span(self): + """A child chain with a different agent name creates a new AgentInvocation.""" + handler, telemetry, _, _ = _make_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + # First agent + first_agent_inv = mock.MagicMock(spec=AgentInvocation) + first_agent_inv.span = mock.MagicMock() + first_agent_inv.span.is_recording.return_value = False + telemetry.start_invoke_local_agent.return_value = first_agent_inv + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + # Second agent with a different name + second_agent_inv = mock.MagicMock(spec=AgentInvocation) + second_agent_inv.span = mock.MagicMock() + second_agent_inv.span.is_recording.return_value = False + telemetry.start_invoke_local_agent.return_value = second_agent_inv + + handler.on_chain_start( + serialized={"name": "weather_agent"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"agent_name": "weather_agent"}, + ) + + assert ( + handler._invocation_manager.get_invocation(child_run_id) + is second_agent_inv + ) + assert second_agent_inv.agent_name == "weather_agent" + + def test_agent_name_comparison_is_case_insensitive(self): + handler, telemetry, _, _ = _make_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + parent_agent_inv = mock.MagicMock(spec=AgentInvocation) + parent_agent_inv.span = mock.MagicMock() + parent_agent_inv.span.is_recording.return_value = False + telemetry.start_invoke_local_agent.return_value = parent_agent_inv + + handler.on_chain_start( + serialized={"name": "Math_Agent"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"agent_name": "Math_Agent"}, + ) + telemetry.start_invoke_local_agent.reset_mock() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"agent_name": "math_agent"}, + ) + + # Same name (case-insensitive) → no new span + telemetry.start_invoke_local_agent.assert_not_called() + + def test_no_agent_name_registers_none_invocation(self): + """When resolve_agent_name returns None the run_id must still be + registered so that child traversal works.""" + handler, telemetry, _, _ = _make_handler() + run_id = _run_id() + + # metadata has otel_agent_span=True so classify_chain_run → INVOKE_AGENT, + # but no agent_name / kwargs name / serialized name, so resolve_agent_name + # returns None. + handler.on_chain_start( + serialized={}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"otel_agent_span": True}, + ) + + telemetry.start_invoke_local_agent.assert_not_called() + # run_id must still be registered (with None invocation) so traversal works + assert run_id in handler._invocation_manager._invocations + assert handler._invocation_manager.get_invocation(run_id) is None + + def test_no_agent_name_child_can_still_find_ancestor_agent(self): + """Even when an intermediate node has no agent name, a deeper child + must still be able to walk up and find a grandparent AgentInvocation.""" + handler, telemetry, _, agent_inv = _make_handler() + grandparent_id = _run_id() + parent_id = _run_id() + + # Grandparent: a known agent + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=grandparent_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + # Parent: INVOKE_AGENT but no resolvable name → registers None + telemetry.start_invoke_local_agent.reset_mock() + handler.on_chain_start( + serialized={}, + inputs={}, + run_id=parent_id, + parent_run_id=grandparent_id, + metadata={"otel_agent_span": True}, + ) + + # Child: should find the grandparent agent via _find_nearest_agent + found = handler._find_nearest_agent(parent_id) + assert found is agent_inv + + +# --------------------------------------------------------------------------- +# on_chain_start – unclassified +# --------------------------------------------------------------------------- + + +class TestOnChainStartUnclassified: + def test_unclassified_chain_registers_none_and_no_span(self): + handler, telemetry, _, _ = _make_handler() + run_id = _run_id() + parent_run_id = _run_id() + + # Register the parent first so that the child links correctly + handler._invocation_manager.add_invocation_state( + parent_run_id, None, None + ) + + handler.on_chain_start( + serialized={"name": "SomeInternalChain"}, + inputs={}, + run_id=run_id, + parent_run_id=parent_run_id, + ) + + telemetry.start_workflow.assert_not_called() + telemetry.start_invoke_local_agent.assert_not_called() + assert run_id in handler._invocation_manager._invocations + assert handler._invocation_manager.get_invocation(run_id) is None + + +# --------------------------------------------------------------------------- +# on_chain_end +# --------------------------------------------------------------------------- + + +class TestOnChainEnd: + def test_workflow_invocation_stopped_on_chain_end(self): + handler, telemetry, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + handler.on_chain_end(outputs={}, run_id=run_id) + + workflow_inv.stop.assert_called_once() + + def test_agent_invocation_stopped_on_chain_end(self): + handler, telemetry, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + handler.on_chain_end(outputs={}, run_id=run_id) + + agent_inv.stop.assert_called_once() + + def test_none_invocation_on_chain_end_does_not_raise(self): + """on_chain_end for a run registered with None invocation (unclassified + or duplicate agent) must silently do nothing.""" + handler, _, _, _ = _make_handler() + run_id = _run_id() + + handler._invocation_manager.add_invocation_state(run_id, None, None) + + # Must not raise + handler.on_chain_end(outputs={}, run_id=run_id) + + def test_unknown_run_id_on_chain_end_does_not_raise(self): + handler, _, _, _ = _make_handler() + handler.on_chain_end(outputs={}, run_id=_run_id()) + + def test_invocation_state_cleaned_up_after_chain_end(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + # span.is_recording() returns False → should be cleaned up + workflow_inv.span.is_recording.return_value = False + handler.on_chain_end(outputs={}, run_id=run_id) + + assert run_id not in handler._invocation_manager._invocations + + +# --------------------------------------------------------------------------- +# on_chain_error +# --------------------------------------------------------------------------- + + +class TestOnChainError: + def test_workflow_invocation_failed_on_chain_error(self): + handler, telemetry, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + err = RuntimeError("something went wrong") + handler.on_chain_error(error=err, run_id=run_id) + + workflow_inv.fail.assert_called_once_with(err) + + def test_agent_invocation_failed_on_chain_error(self): + handler, telemetry, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + err = ValueError("agent failed") + handler.on_chain_error(error=err, run_id=run_id) + + agent_inv.fail.assert_called_once_with(err) + + def test_none_invocation_on_chain_error_does_not_raise(self): + handler, _, _, _ = _make_handler() + run_id = _run_id() + + handler._invocation_manager.add_invocation_state(run_id, None, None) + + handler.on_chain_error(error=RuntimeError("boom"), run_id=run_id) + + def test_unknown_run_id_on_chain_error_does_not_raise(self): + handler, _, _, _ = _make_handler() + handler.on_chain_error(error=RuntimeError("boom"), run_id=_run_id()) + + def test_invocation_state_cleaned_up_after_chain_error(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + workflow_inv.span.is_recording.return_value = False + handler.on_chain_error(error=RuntimeError("boom"), run_id=run_id) + + assert run_id not in handler._invocation_manager._invocations + + +# --------------------------------------------------------------------------- +# _find_nearest_agent +# --------------------------------------------------------------------------- + + +class TestFindNearestAgent: + def test_returns_none_when_no_agent_in_ancestry(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + assert handler._find_nearest_agent(run_id) is None + + def test_finds_direct_parent_agent(self): + handler, telemetry, _, agent_inv = _make_handler() + parent_id = _run_id() + child_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=parent_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + # Register the child as unclassified so it links to the parent + handler._invocation_manager.add_invocation_state( + child_id, parent_id, None + ) + + found = handler._find_nearest_agent(child_id) + assert found is agent_inv + + def test_finds_grandparent_agent(self): + handler, telemetry, _, agent_inv = _make_handler() + grandparent_id = _run_id() + parent_id = _run_id() + child_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=grandparent_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + handler._invocation_manager.add_invocation_state( + parent_id, grandparent_id, None + ) + handler._invocation_manager.add_invocation_state( + child_id, parent_id, None + ) + + found = handler._find_nearest_agent(child_id) + assert found is agent_inv + + +# --------------------------------------------------------------------------- +# utils.make_input_message +# --------------------------------------------------------------------------- + + +class TestMakeInputMessage: + def test_returns_empty_list_for_non_dict(self): + assert make_input_message("not a dict") == [] + assert make_input_message(None) == [] + assert make_input_message(42) == [] + + def test_empty_dict_returns_empty_list(self): + assert make_input_message({}) == [] + + def test_messages_key_with_human_message(self): + msg = HumanMessage(content="Hello") + result = make_input_message({"messages": [msg]}) + + assert len(result) == 1 + assert isinstance(result[0], InputMessage) + assert result[0].role == "user" + assert len(result[0].parts) == 1 + assert isinstance(result[0].parts[0], Text) + assert result[0].parts[0].content == "Hello" + + def test_messages_key_skips_empty_content(self): + msg_empty = HumanMessage(content="") + msg_valid = HumanMessage(content="Hi") + result = make_input_message({"messages": [msg_empty, msg_valid]}) + + assert len(result) == 1 + assert result[0].parts[0].content == "Hi" + + def test_messages_key_multiple_messages(self): + msgs = [HumanMessage(content="First"), HumanMessage(content="Second")] + result = make_input_message({"messages": msgs}) + + assert len(result) == 2 + assert result[0].parts[0].content == "First" + assert result[1].parts[0].content == "Second" + + def test_messages_key_takes_priority_over_other_fields(self): + msg = HumanMessage(content="hello") + result = make_input_message( + {"messages": [msg], "user_query": "should be ignored"} + ) + + assert len(result) == 1 + assert result[0].parts[0].content == "hello" + + def test_fallback_serializes_non_message_state_fields(self): + result = make_input_message({"user_query": "what is 2+2?"}) + + assert len(result) == 1 + assert result[0].role == "user" + # The content should be a JSON serialization of the dict + assert "user_query" in result[0].parts[0].content + assert "what is 2+2?" in result[0].parts[0].content + + def test_fallback_excludes_intermediate_steps_key(self): + # messages key absent → fallback path runs; intermediate_steps excluded + result = make_input_message( + { + "user_query": "hi", + "intermediate_steps": [("tool", "result")], + } + ) + + assert len(result) == 1 + content = result[0].parts[0].content + assert "intermediate_steps" not in content + assert "user_query" in content + + def test_messages_key_present_skips_fallback_even_with_other_fields(self): + # messages key present (even empty list) → early return, fallback not reached + result = make_input_message( + { + "user_query": "ignored", + "messages": [], + "intermediate_steps": [("tool", "result")], + } + ) + + assert result == [] + + def test_fallback_returns_empty_when_all_values_are_none(self): + result = make_input_message({"user_query": None, "context": None}) + assert result == [] + + def test_fallback_returns_empty_when_only_excluded_keys(self): + result = make_input_message( + {"messages": None, "intermediate_steps": None} + ) + assert result == [] + + def test_messages_key_with_empty_list(self): + # messages key present but empty → return empty list (no fallback) + result = make_input_message({"messages": [], "user_query": "ignored"}) + assert result == [] + + +# --------------------------------------------------------------------------- +# utils.make_output_message / make_last_output_message +# --------------------------------------------------------------------------- + + +class TestMakeOutputMessage: + def test_returns_empty_list_for_non_dict(self): + assert make_output_message("not a dict") == [] + assert make_output_message(None) == [] + + def test_returns_empty_list_when_no_messages_key(self): + assert make_output_message({"output": "hi"}) == [] + + def test_returns_empty_list_when_messages_is_none(self): + assert make_output_message({"messages": None}) == [] + + def test_ai_message_produces_assistant_output(self): + ai_msg = AIMessage(content="The answer is 42") + result = make_output_message({"messages": [ai_msg]}) + + assert len(result) == 1 + assert isinstance(result[0], OutputMessage) + assert result[0].role == "assistant" + assert result[0].finish_reason == "stop" + assert result[0].parts[0].content == "The answer is 42" + + def test_non_ai_message_skipped(self): + human_msg = HumanMessage(content="Hello") + result = make_output_message({"messages": [human_msg]}) + assert result == [] + + def test_ai_message_with_empty_content_skipped(self): + ai_msg = AIMessage(content="") + result = make_output_message({"messages": [ai_msg]}) + assert result == [] + + def test_multiple_ai_messages_all_returned(self): + msgs = [ + AIMessage(content="First response"), + AIMessage(content="Second response"), + ] + result = make_output_message({"messages": msgs}) + + assert len(result) == 2 + assert result[0].parts[0].content == "First response" + assert result[1].parts[0].content == "Second response" + + def test_mixed_messages_only_ai_returned(self): + msgs = [ + HumanMessage(content="question"), + AIMessage(content="answer"), + HumanMessage(content="follow-up"), + ] + result = make_output_message({"messages": msgs}) + + assert len(result) == 1 + assert result[0].parts[0].content == "answer" + + +class TestMakeLastOutputMessage: + def test_returns_only_last_ai_message(self): + msgs = [ + AIMessage(content="intermediate"), + AIMessage(content="final answer"), + ] + result = make_last_output_message({"messages": msgs}) + + assert len(result) == 1 + assert result[0].parts[0].content == "final answer" + + def test_returns_empty_when_no_ai_messages(self): + result = make_last_output_message( + {"messages": [HumanMessage(content="hi")]} + ) + assert result == [] + + def test_returns_empty_for_empty_outputs(self): + assert make_last_output_message({}) == [] + + def test_single_ai_message_returned(self): + ai_msg = AIMessage(content="only response") + result = make_last_output_message({"messages": [ai_msg]}) + + assert len(result) == 1 + assert result[0].parts[0].content == "only response" + + +# --------------------------------------------------------------------------- +# utils.serialize +# --------------------------------------------------------------------------- + + +class TestSerialize: + def test_none_returns_none(self): + assert serialize(None) is None + + def test_dict_serialized_to_json(self): + result = serialize({"key": "value"}) + assert result == '{"key": "value"}' + + def test_list_serialized_to_json(self): + result = serialize([1, 2, 3]) + assert result == "[1, 2, 3]" + + def test_non_serializable_falls_back_to_str(self): + class Custom: + def __str__(self): + return "custom_repr" + + result = serialize({"obj": Custom()}) + assert result is not None + assert "custom_repr" in result + + def test_string_value(self): + result = serialize("hello") + assert result == '"hello"' + + +# --------------------------------------------------------------------------- +# input_messages / output_messages set on invocations via callback handler +# --------------------------------------------------------------------------- + + +class TestInputMessagesOnInvocations: + def test_workflow_input_messages_set_from_messages_key(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + msg = HumanMessage(content="What is the weather?") + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={"messages": [msg]}, + run_id=run_id, + parent_run_id=None, + ) + + assigned = workflow_inv.input_messages + assert len(assigned) == 1 + assert assigned[0].role == "user" + assert assigned[0].parts[0].content == "What is the weather?" + + def test_workflow_input_messages_set_from_state_fallback(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={"user_query": "plan a trip"}, + run_id=run_id, + parent_run_id=None, + ) + + assigned = workflow_inv.input_messages + assert len(assigned) == 1 + assert "user_query" in assigned[0].parts[0].content + assert "plan a trip" in assigned[0].parts[0].content + + def test_workflow_input_messages_empty_for_empty_inputs(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + assert workflow_inv.input_messages == [] + + def test_agent_input_messages_set_from_messages_key(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + msg = HumanMessage(content="Solve x+2=5") + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={"messages": [msg]}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + assigned = agent_inv.input_messages + assert len(assigned) == 1 + assert assigned[0].parts[0].content == "Solve x+2=5" + + def test_agent_input_messages_set_from_state_fallback(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={"problem": "integrate x^2"}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + assigned = agent_inv.input_messages + assert len(assigned) == 1 + assert "integrate x^2" in assigned[0].parts[0].content + + +class TestOutputMessagesOnInvocations: + def test_workflow_output_messages_set_on_chain_end(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + ai_msg = AIMessage(content="The final answer is 42") + handler.on_chain_end( + outputs={"messages": [ai_msg]}, + run_id=run_id, + ) + + assigned = workflow_inv.output_messages + assert len(assigned) == 1 + assert assigned[0].role == "assistant" + assert assigned[0].parts[0].content == "The final answer is 42" + + def test_workflow_output_messages_only_last_ai_message(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + msgs = [ + AIMessage(content="intermediate tool call"), + AIMessage(content="final answer"), + ] + handler.on_chain_end(outputs={"messages": msgs}, run_id=run_id) + + assigned = workflow_inv.output_messages + assert len(assigned) == 1 + assert assigned[0].parts[0].content == "final answer" + + def test_workflow_output_messages_empty_when_no_ai_messages(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + handler.on_chain_end( + outputs={"messages": [HumanMessage(content="hi")]}, + run_id=run_id, + ) + + assert workflow_inv.output_messages == [] + + def test_workflow_output_messages_empty_for_empty_outputs(self): + handler, _, workflow_inv, _ = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + + handler.on_chain_end(outputs={}, run_id=run_id) + + assert workflow_inv.output_messages == [] + + def test_agent_output_messages_set_on_chain_end(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + ai_msg = AIMessage(content="x = 3") + handler.on_chain_end(outputs={"messages": [ai_msg]}, run_id=run_id) + + assigned = agent_inv.output_messages + assert len(assigned) == 1 + assert assigned[0].parts[0].content == "x = 3" + + def test_agent_output_messages_only_last_ai_message(self): + handler, _, _, agent_inv = _make_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "math_agent"}, + ) + + msgs = [ + AIMessage(content="let me think..."), + AIMessage(content="the answer is 7"), + ] + handler.on_chain_end(outputs={"messages": msgs}, run_id=run_id) + + assigned = agent_inv.output_messages + assert len(assigned) == 1 + assert assigned[0].parts[0].content == "the answer is 7" diff --git a/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_invocation_manager.py b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_invocation_manager.py index c120af1042..8e977ae65a 100644 --- a/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_invocation_manager.py +++ b/instrumentation-genai/opentelemetry-instrumentation-langchain/tests/test_invocation_manager.py @@ -104,7 +104,73 @@ def test_delete_invocation_state(invocation_manager, mock_invocation): assert run_id not in invocation_manager._invocations -def test_delete_invocation_state_with_children(invocation_manager): +def test_delete_invocation_state_deferred_while_children_live( + invocation_manager, +): + """Deleting a parent while children are still live defers its removal.""" + parent_id = uuid.uuid4() + child_id = uuid.uuid4() + + parent_invocation = mock.Mock(spec=GenAIInvocation) + child_invocation = mock.Mock(spec=GenAIInvocation) + + invocation_manager.add_invocation_state( + run_id=parent_id, parent_run_id=None, invocation=parent_invocation + ) + invocation_manager.add_invocation_state( + run_id=child_id, parent_run_id=parent_id, invocation=child_invocation + ) + + # Delete the parent while the child is still live + invocation_manager.delete_invocation_state(parent_id) + + # Parent should still be present (deferred) because child is live + assert parent_id in invocation_manager._invocations + assert invocation_manager._invocations[parent_id].ended is True + + # After the child is deleted, the parent should also be cleaned up + invocation_manager.delete_invocation_state(child_id) + + assert child_id not in invocation_manager._invocations + assert parent_id not in invocation_manager._invocations + + +def test_delete_invocation_state_propagates_upward(invocation_manager): + """When the last child is removed, an already-ended parent is cleaned up.""" + grandparent_id = uuid.uuid4() + parent_id = uuid.uuid4() + child_id = uuid.uuid4() + + for run_id, parent in [ + (grandparent_id, None), + (parent_id, grandparent_id), + (child_id, parent_id), + ]: + invocation_manager.add_invocation_state( + run_id=run_id, + parent_run_id=parent, + invocation=mock.Mock(spec=GenAIInvocation), + ) + + # Mark grandparent and parent as ended (deferred) + invocation_manager.delete_invocation_state(grandparent_id) + invocation_manager.delete_invocation_state(parent_id) + + assert grandparent_id in invocation_manager._invocations # deferred + assert parent_id in invocation_manager._invocations # deferred + + # Removing the last live node should cascade upward + invocation_manager.delete_invocation_state(child_id) + + assert child_id not in invocation_manager._invocations + assert parent_id not in invocation_manager._invocations + assert grandparent_id not in invocation_manager._invocations + + +def test_delete_invocation_state_with_multiple_children_defers_until_last( + invocation_manager, +): + """Parent removal is deferred until all children are gone.""" parent_id = uuid.uuid4() child1_id = uuid.uuid4() child2_id = uuid.uuid4() @@ -113,32 +179,63 @@ def test_delete_invocation_state_with_children(invocation_manager): child1_invocation = mock.Mock(spec=GenAIInvocation) child2_invocation = mock.Mock(spec=GenAIInvocation) - # Add parent and children + invocation_manager.add_invocation_state( + run_id=parent_id, parent_run_id=None, invocation=parent_invocation + ) + invocation_manager.add_invocation_state( + run_id=child1_id, parent_run_id=parent_id, invocation=child1_invocation + ) + invocation_manager.add_invocation_state( + run_id=child2_id, parent_run_id=parent_id, invocation=child2_invocation + ) + + # Delete parent while both children live → deferred + invocation_manager.delete_invocation_state(parent_id) + assert parent_id in invocation_manager._invocations + + # Remove first child → parent still deferred (child2 is live) + invocation_manager.delete_invocation_state(child1_id) + assert parent_id in invocation_manager._invocations + + # Remove last child → parent is now cleaned up + invocation_manager.delete_invocation_state(child2_id) + assert parent_id not in invocation_manager._invocations + + +def test_get_parent_run_id_returns_none_for_unknown(invocation_manager): + assert invocation_manager.get_parent_run_id(uuid.uuid4()) is None + + +def test_get_parent_run_id_returns_registered_parent(invocation_manager): + parent_id = uuid.uuid4() + child_id = uuid.uuid4() + invocation_manager.add_invocation_state( run_id=parent_id, parent_run_id=None, - invocation=parent_invocation, + invocation=mock.Mock(spec=GenAIInvocation), ) invocation_manager.add_invocation_state( - run_id=child1_id, + run_id=child_id, parent_run_id=parent_id, - invocation=child1_invocation, + invocation=mock.Mock(spec=GenAIInvocation), ) + + assert invocation_manager.get_parent_run_id(child_id) == parent_id + assert invocation_manager.get_parent_run_id(parent_id) is None + + +def test_none_invocation_can_be_stored_and_retrieved(invocation_manager): + """Nodes with no associated span (None invocation) must still be tracked.""" + run_id = uuid.uuid4() + invocation_manager.add_invocation_state( - run_id=child2_id, - parent_run_id=parent_id, - invocation=child2_invocation, + run_id=run_id, parent_run_id=None, invocation=None ) - # Verify initial state - assert len(invocation_manager._invocations) == 3 - assert len(invocation_manager._invocations[parent_id].children) == 2 + assert run_id in invocation_manager._invocations + assert invocation_manager.get_invocation(run_id) is None - # Delete parent - invocation_manager.delete_invocation_state(parent_id) - # Verify parent and all children were removed - assert parent_id not in invocation_manager._invocations - assert child1_id not in invocation_manager._invocations - assert child2_id not in invocation_manager._invocations - assert len(invocation_manager._invocations) == 0 +def test_delete_nonexistent_run_id_does_not_raise(invocation_manager): + invocation_manager.delete_invocation_state(uuid.uuid4()) # must not raise diff --git a/uv.lock b/uv.lock index fd482cb747..3bca158f14 100644 --- a/uv.lock +++ b/uv.lock @@ -3571,6 +3571,7 @@ name = "opentelemetry-instrumentation-langchain" source = { editable = "instrumentation-genai/opentelemetry-instrumentation-langchain" } dependencies = [ { name = "opentelemetry-instrumentation" }, + { name = "opentelemetry-util-genai" }, ] [package.optional-dependencies] @@ -3582,6 +3583,7 @@ instruments = [ requires-dist = [ { name = "langchain", marker = "extra == 'instruments'", specifier = ">=0.3.21" }, { name = "opentelemetry-instrumentation", editable = "opentelemetry-instrumentation" }, + { name = "opentelemetry-util-genai", editable = "util/opentelemetry-util-genai" }, ] provides-extras = ["instruments"]