input output messages and updated example

Author: wrisa

instrumentation-genai/opentelemetry-instrumentation-langchain/examples/agent/main.py

@@ -10,10 +10,10 @@
 
 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.prebuilt import create_react_agent
 
 from opentelemetry import _logs, metrics, trace
 from opentelemetry.exporter.otlp.proto.grpc._log_exporter import (
@@ -80,9 +80,7 @@ def main():
     )
 
     session_id = str(uuid4())
-    agent = create_agent(
-        llm, tools=[multiply, add], name="coordinator"
-    ).with_config(
+    agent = create_react_agent(llm, tools=[multiply, add]).with_config(
         {
             "metadata": {
                 "agent_name": "coordinator",

instrumentation-genai/opentelemetry-instrumentation-langchain/examples/workflow/main.py

@@ -2,11 +2,17 @@
 # SPDX-License-Identifier: Apache-2.0
 
 """
-LangGraph StateGraph example with an LLM node.
+LangGraph StateGraph example with two LLM nodes.
 
-Similar to the manual example (../manual/main.py) but uses LangGraph's StateGraph
-with a node that calls ChatOpenAI. OpenTelemetry LangChain instrumentation traces
-the LLM calls made from within the graph node.
+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
@@ -37,8 +43,9 @@
 
 # Configure tracing
 trace.set_tracer_provider(TracerProvider())
-span_processor = BatchSpanProcessor(OTLPSpanExporter())
-trace.get_tracer_provider().add_span_processor(span_processor)
+trace.get_tracer_provider().add_span_processor(
+    BatchSpanProcessor(OTLPSpanExporter())
+)
 
 # Configure logging
 _logs.set_logger_provider(LoggerProvider())
@@ -49,66 +56,87 @@
 # Configure metrics
 metrics.set_meter_provider(
     MeterProvider(
-        metric_readers=[
-            PeriodicExportingMetricReader(
-                OTLPMetricExporter(),
-            ),
-        ]
+        metric_readers=[PeriodicExportingMetricReader(OTLPMetricExporter())]
     )
 )
 
 
 class GraphState(TypedDict):
-    """State for the graph; messages are accumulated with add_messages."""
+    """State shared across all graph nodes."""
 
     messages: Annotated[list, add_messages]
+    research: str
 
 
 def build_graph(llm: ChatOpenAI):
-    """Build a StateGraph with a single LLM node."""
-
-    def llm_node(state: GraphState) -> dict:
-        """Node that invokes the LLM with the current messages."""
-        response = llm.invoke(state["messages"])
+    """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("llm", llm_node)
-    builder.add_edge(START, "llm")
-    builder.add_edge("llm", END)
+    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():
-    # Set up instrumentation (traces LLM calls from within graph nodes)
     LangChainInstrumentor().instrument()
 
-    # ChatOpenAI setup
     llm = ChatOpenAI(
         model="gpt-3.5-turbo",
         temperature=0.1,
-        max_tokens=100,
-        top_p=0.9,
-        frequency_penalty=0.5,
-        presence_penalty=0.5,
-        stop=["\n", "Human:", "AI:"],
-        seed=100,
+        max_tokens=200,
+        seed=42,
     )
 
     graph = build_graph(llm)
 
-    initial_messages = [
-        SystemMessage(content="You are a helpful assistant!"),
-        HumanMessage(content="What is the capital of France?"),
-    ]
+    question = "What is the capital of France?"
+    print(f"Question: {question}\n")
+
+    result = graph.invoke(
+        {
+            "messages": [HumanMessage(content=question)],
+            "research": "",
+        }
+    )
 
-    result = graph.invoke({"messages": initial_messages})
+    print("Research output:")
+    print(f"  {result['research']}\n")
 
-    print("LangGraph output (messages):")
-    for msg in result.get("messages", []):
-        print(f"  {type(msg).__name__}: {msg.content}")
+    print("Final summary:")
+    print(f"  {result['messages'][-1].content}")
 
-    # Un-instrument after use
     LangChainInstrumentor().uninstrument()
 
 

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

@@ -26,7 +26,7 @@ classifiers = [
 ]
 dependencies = [
   "opentelemetry-instrumentation ~= 0.60b0",
-  "opentelemetry-util-genai >= 0.4b0.dev",
+  "opentelemetry-util-genai >= 0.4b0",
 ]
 
 [project.optional-dependencies]

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

@@ -18,6 +18,10 @@
     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,
@@ -65,6 +69,7 @@ def on_chain_start(
             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
             )
@@ -92,6 +97,7 @@ def on_chain_start(
                         else "unknown",
                     )
                     agent.agent_name = suggested_agent_name
+                    agent.input_messages = make_input_message(inputs)
 
                     if metadata:
                         agent.agent_id = metadata.get("agent_id")
@@ -146,6 +152,9 @@ def on_chain_end(
             self._invocation_manager.delete_invocation_state(run_id)
             return
 
+        if isinstance(invocation, (WorkflowInvocation, AgentInvocation)):
+            invocation.output_messages = make_last_output_message(outputs)
+
         invocation.stop()
 
         if not invocation.span.is_recording():

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

@@ -0,0 +1,93 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from typing import Any, Optional
+
+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 []
+    input_messages: list[InputMessage] = []
+    messages = data.get("messages")
+    if messages is not None:
+        for msg in messages:
+            content = getattr(msg, "content", "")
+            if content:
+                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 = {
+        k: v
+        for k, v in data.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."""
+    if not isinstance(data, dict):
+        return []
+    output_messages: list[OutputMessage] = []
+    messages = data.get("messages")
+    if messages is None:
+        return []
+    for msg in messages:
+        content = getattr(msg, "content", "")
+        if content:
+            if isinstance(msg, AIMessage):
+                output_message = OutputMessage(
+                    role="assistant",
+                    parts=[Text(msg.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