fix(langchain): record dict-form assistant messages on workflow and agent spans (#881)

* fix(langchain): record dict-form assistant messages on workflow and agent spans

to_output_messages kept only AIMessage instances, so role/content dicts
and (role, content) tuples returned in a workflow or agent's `messages`
were dropped from gen_ai.output.messages, or an earlier AIMessage was
reported as the final output instead.

Convert entries with convert_to_messages one at a time in a shared
helper used for both inputs and outputs, so a single unconvertible entry
no longer drops the other convertible ones.

Fixes #388

Assisted-by: Claude Opus 5.5

* Add changelog fragment for #881

Assisted-by: Claude Opus 5.5

* fix(langchain): convert messages in one batch before falling back per entry

Try convert_to_messages on the whole list first and only convert entries
one at a time when that fails. In the fallback, BaseMessage instances are
kept as they are instead of going through convert_to_messages again.

Assisted-by: Claude Opus 5.5

Author: Thanatos9404

instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/881.fixed

@@ -0,0 +1 @@
+Record assistant messages given as role/content dicts or `(role, content)` tuples in workflow and agent `gen_ai.output.messages` instead of dropping them, and keep convertible messages when another entry in the list cannot be converted.

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

@@ -14,6 +14,7 @@
     ChatMessage,
     FunctionMessage,
     HumanMessage,
+    MessageLikeRepresentation,
     SystemMessage,
     ToolMessage,
     convert_to_messages,
@@ -337,25 +338,36 @@ def _message_parts(message: BaseMessage) -> list[MessagePart]:
     return _content_to_parts(message.content)
 
 
+def _normalize_messages(
+    messages: Iterable[MessageLikeRepresentation],
+) -> list[BaseMessage]:
+    """Convert entries to ``BaseMessage`` s, skipping unconvertible ones."""
+    materialized = list(messages)
+    try:
+        return convert_to_messages(materialized)
+    except Exception:  # pylint: disable=broad-except
+        normalized: list[BaseMessage] = []
+        for message in materialized:
+            if isinstance(message, BaseMessage):
+                normalized.append(message)
+                continue
+            try:
+                normalized.extend(convert_to_messages([message]))
+            except Exception:  # pylint: disable=broad-except
+                continue
+        return normalized
+
+
 def to_input_messages(
-    messages: Iterable[Any],
+    messages: Iterable[MessageLikeRepresentation],
 ) -> list[InputMessage]:
     """Convert LangChain messages into spec-conformant ``InputMessage`` s.
 
     Called only when content capture is enabled
     (``TelemetryHandler.should_capture_content()``).
     """
-    materialized = list(messages)
-    try:
-        normalized_messages: Iterable[BaseMessage] = convert_to_messages(
-            materialized
-        )
-    except Exception:  # pylint: disable=broad-except
-        normalized_messages = [
-            m for m in materialized if isinstance(m, BaseMessage)
-        ]
     result: list[InputMessage] = []
-    for message in normalized_messages:
+    for message in _normalize_messages(messages):
         parts = _message_parts(message)
         if not parts and not _has_content(message):
             continue
@@ -370,22 +382,21 @@ def to_input_messages(
 
 
 def to_output_messages(
-    messages: Iterable[BaseMessage],
+    messages: Iterable[MessageLikeRepresentation],
     *,
     finish_reason: str = "",
 ) -> list[OutputMessage]:
-    """Convert LangChain ``AIMessage`` instances into ``OutputMessage`` s.
+    """Convert LangChain messages into ``OutputMessage`` s.
 
-    Non-``AIMessage`` entries are skipped: only assistant turns are recorded
-    as ``gen_ai.output.messages``. Tool execution results belong on the
-    *input* side of the next inference call, not the output side of the
-    previous one.
+    Only assistant turns are recorded as ``gen_ai.output.messages``. Tool
+    execution results belong on the *input* side of the next inference call,
+    not the output side of the previous one.
 
     Called only when content capture is enabled
     (``TelemetryHandler.should_capture_content()``).
     """
     result: list[OutputMessage] = []
-    for message in messages:
+    for message in _normalize_messages(messages):
         if not isinstance(message, AIMessage):
             continue
         parts = _ai_message_parts(message)
@@ -476,7 +487,9 @@ def make_input_message(data: Any) -> list[InputMessage]:
             messages, Iterable
         ):
             return []
-        return to_input_messages(cast(Iterable[BaseMessage], messages))
+        return to_input_messages(
+            cast(Iterable[MessageLikeRepresentation], 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.
@@ -500,7 +513,7 @@ def make_input_message(data: Any) -> list[InputMessage]:
 def make_output_message(data: Any) -> list[OutputMessage]:
     """Build ``OutputMessage`` s from a workflow/agent output mapping.
 
-    Only ``AIMessage`` entries become outputs. ``finish_reason`` is left
+    Only assistant messages become outputs. ``finish_reason`` is left
     empty: the underlying per-LLM-call finish reasons are recorded on child
     inference spans, and util-genai filters empty values out of
     ``gen_ai.response.finish_reasons``.
@@ -518,7 +531,9 @@ def make_output_message(data: Any) -> list[OutputMessage]:
         or not isinstance(messages, Iterable)
     ):
         return []
-    return to_output_messages(cast(Iterable[BaseMessage], messages))
+    return to_output_messages(
+        cast(Iterable[MessageLikeRepresentation], messages)
+    )
 
 
 def make_last_output_message(data: Any) -> list[OutputMessage]:

instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py

@@ -339,8 +339,14 @@ async def test_chain_classification_and_completion(
                 AIMessage(content="final answer"),
             ]
         },
+        {
+            "messages": [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "final answer"},
+            ]
+        },
     ],
-    ids=["empty", "no-assistant", "last-assistant"],
+    ids=["empty", "no-assistant", "last-assistant", "dict-assistant"],
 )
 async def test_chain_input_and_output_capture(
     asynchronous: bool,
@@ -364,10 +370,10 @@ async def test_chain_input_and_output_capture(
         assert json.loads(captured[0]["parts"][0]["content"]) == inputs
     else:
         assert GenAI.GEN_AI_INPUT_MESSAGES not in attrs
-    if (
-        content_capture == ContentCapturingMode.SPAN_ONLY
-        and outputs.get("messages")
-        and isinstance(outputs["messages"][-1], AIMessage)
+    last = (outputs.get("messages") or [None])[-1]
+    if content_capture == ContentCapturingMode.SPAN_ONLY and (
+        isinstance(last, AIMessage)
+        or (isinstance(last, dict) and last.get("role") == "assistant")
     ):
         captured = json.loads(attrs[GenAI.GEN_AI_OUTPUT_MESSAGES])
         assert len(captured) == 1
@@ -1453,6 +1459,19 @@ def test_messages_key_malformed_entry_does_not_raise(self, bad_entry):
         assert len(result) == 1
         assert result[0].parts[0].content == "Hi"
 
+    def test_messages_key_malformed_entry_keeps_convertible_entries(self):
+        result = make_input_message(
+            {
+                "messages": [
+                    {"role": "user", "content": "Hello"},
+                    ("human",),
+                    HumanMessage(content="Hi"),
+                ]
+            }
+        )
+
+        assert [m.parts[0].content for m in result] == ["Hello", "Hi"]
+
     def test_messages_key_all_malformed_returns_empty_without_raising(self):
         result = make_input_message(
             {"messages": [("human",), None, {"content": "x"}]}
@@ -1548,6 +1567,36 @@ def test_ai_message_produces_assistant_output(self):
         assert result[0].finish_reason == ""
         assert result[0].parts[0].content == "The answer is 42"
 
+    @pytest.mark.parametrize(
+        "assistant_entry",
+        [
+            {"role": "assistant", "content": "Hello"},
+            ("ai", "Hello"),
+        ],
+    )
+    def test_message_like_assistant_entry_produces_assistant_output(
+        self, assistant_entry
+    ):
+        result = make_output_message({"messages": [assistant_entry]})
+
+        assert len(result) == 1
+        assert result[0].role == "assistant"
+        assert result[0].parts[0].content == "Hello"
+
+    def test_malformed_entry_keeps_convertible_assistant_entries(self):
+        result = make_output_message(
+            {
+                "messages": [
+                    {"role": "not_a_real_role", "content": "dropped"},
+                    None,
+                    {"role": "assistant", "content": "kept"},
+                ]
+            }
+        )
+
+        assert len(result) == 1
+        assert result[0].parts[0].content == "kept"
+
     def test_non_ai_message_skipped(self):
         human_msg = HumanMessage(content="Hello")
         result = make_output_message({"messages": [human_msg]})
@@ -1626,6 +1675,16 @@ def test_returns_only_last_ai_message(self):
         assert len(result) == 1
         assert result[0].parts[0].content == "final answer"
 
+    def test_returns_last_assistant_message_in_dict_form(self):
+        msgs = [
+            AIMessage(content="intermediate"),
+            {"role": "assistant", "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")]}