fix: agent streaming cumulative tokens instead of partial messages

src/backend/base/langflow/base/agents/agent.py

@@ -4,12 +4,11 @@
 
 from langchain.agents import AgentExecutor, BaseMultiActionAgent, BaseSingleActionAgent
 from langchain.agents.agent import RunnableAgent
-from langchain_core.messages import HumanMessage
+from langchain_core.messages import BaseMessage, HumanMessage
 from langchain_core.runnables import Runnable
 
 from langflow.base.agents.callback import AgentAsyncHandler
 from langflow.base.agents.events import ExceptionWithMessageError, process_agent_events
-from langflow.base.agents.utils import data_to_messages
 from langflow.custom.custom_component.component import Component, _get_component_toolkit
 from langflow.field_typing import Tool
 from langflow.inputs.inputs import InputTypes, MultilineInput
@@ -23,9 +22,7 @@
 from langflow.utils.constants import MESSAGE_SENDER_AI
 
 if TYPE_CHECKING:
-    from langchain_core.messages import BaseMessage
-
-    from langflow.schema.log import SendMessageFunctionType
+    from langflow.schema.log import OnTokenFunctionType, SendMessageFunctionType
 
 
 DEFAULT_TOOLS_DESCRIPTION = "A helpful assistant with access to the following tools:"
@@ -118,6 +115,24 @@ def get_chat_history_data(self) -> list[Data] | None:
         # might be overridden in subclasses
         return None
 
+    def _data_to_messages_skip_empty(self, data: list[Data]) -> list[BaseMessage]:
+        """Convert data to messages, filtering only empty text while preserving non-text content.
+
+        Note: added to fix issue with certain providers failing when given empty text.
+        """
+        messages = []
+        for value in data:
+            # Only skip if the message has a text attribute that is empty/whitespace
+            text = getattr(value, "text", None)
+            if isinstance(text, str) and not text.strip():
+                # Skip only messages with empty/whitespace-only text strings
+                continue
+
+            lc_message = value.to_lc_message()
+            messages.append(lc_message)
+
+        return messages
+
     async def run_agent(
         self,
         agent: Runnable | BaseSingleActionAgent | BaseMultiActionAgent | AgentExecutor,
@@ -137,49 +152,52 @@ async def run_agent(
                 max_iterations=max_iterations,
             )
         # Convert input_value to proper format for agent
-        if hasattr(self.input_value, "to_lc_message") and callable(self.input_value.to_lc_message):
+        lc_message = None
+        if isinstance(self.input_value, Message):
             lc_message = self.input_value.to_lc_message()
-            input_text = lc_message.content if hasattr(lc_message, "content") else str(lc_message)
+            input_dict: dict[str, str | list[BaseMessage] | BaseMessage] = {"input": lc_message}
         else:
-            lc_message = None
-            input_text = self.input_value
-
-        input_dict: dict[str, str | list[BaseMessage]] = {
-            "input": self.input_value.to_lc_message() if isinstance(self.input_value, Message) else self.input_value
-        }
+            input_dict = {"input": self.input_value}
 
         if hasattr(self, "system_prompt"):
             input_dict["system_prompt"] = self.system_prompt
 
+        # Handle chat_history conversion
         if hasattr(self, "chat_history") and self.chat_history:
-            if isinstance(self.chat_history, Data):
-                input_dict["chat_history"] = data_to_messages(self.chat_history)
-            # Handle both lfx.schema.message.Message and langflow.schema.message.Message types
-            if all(hasattr(m, "to_data") and callable(m.to_data) and "text" in m.data for m in self.chat_history):
-                input_dict["chat_history"] = data_to_messages(self.chat_history)
-            if all(isinstance(m, Message) for m in self.chat_history):
-                input_dict["chat_history"] = data_to_messages([m.to_data() for m in self.chat_history])
-
-        if (
-            hasattr(lc_message, "content")
-            and isinstance(lc_message.content, list)
-            and all(isinstance(m, Message) for m in self.chat_history)
-        ):
-            input_dict["chat_history"] = data_to_messages([m.to_data() for m in self.chat_history])
-
+            if isinstance(self.chat_history, Data) or all(
+                hasattr(m, "to_data") and callable(m.to_data) and "text" in m.data for m in self.chat_history
+            ):
+                input_dict["chat_history"] = self._data_to_messages_skip_empty(self.chat_history)
+            elif all(isinstance(m, Message) for m in self.chat_history):
+                input_dict["chat_history"] = self._data_to_messages_skip_empty([m.to_data() for m in self.chat_history])
+
+        # Handle multimodal input (images + text)
+        # Note: Agent input must be a string, so we extract text and move images to chat_history
         if lc_message and hasattr(lc_message, "content") and isinstance(lc_message.content, list):
-            # ! Because the input has to be a string, we must pass the images in the chat_history
-
+            # Extract images and text content
             image_dicts = [item for item in lc_message.content if item.get("type") == "image"]
-            lc_message.content = [item for item in lc_message.content if item.get("type") != "image"]
-
-            if "chat_history" not in input_dict:
-                input_dict["chat_history"] = []
-            if isinstance(input_dict["chat_history"], list):
-                input_dict["chat_history"].extend(HumanMessage(content=[image_dict]) for image_dict in image_dicts)
-            else:
-                input_dict["chat_history"] = [HumanMessage(content=[image_dict]) for image_dict in image_dicts]
-        input_dict["input"] = input_text
+            text_content = [item for item in lc_message.content if item.get("type") != "image"]
+
+            # Extract text strings from text content items
+            text_strings = [
+                item.get("text", "")
+                for item in text_content
+                if item.get("type") == "text" and item.get("text", "").strip()
+            ]
+
+            # Set input to concatenated text or empty string
+            input_dict["input"] = " ".join(text_strings) if text_strings else ""
+
+            # Add images to chat_history
+            if image_dicts:
+                if "chat_history" not in input_dict:
+                    input_dict["chat_history"] = []
+                if isinstance(input_dict["chat_history"], list):
+                    input_dict["chat_history"].extend(HumanMessage(content=[img]) for img in image_dicts)
+        elif lc_message:
+            # Simple text message - extract string content
+            input_dict["input"] = lc_message.content if hasattr(lc_message, "content") else str(lc_message)
+
         if hasattr(self, "graph"):
             session_id = self.graph.session_id
         elif hasattr(self, "_session_id"):
@@ -194,6 +212,13 @@ async def run_agent(
             content_blocks=[ContentBlock(title="Agent Steps", contents=[])],
             session_id=session_id,
         )
+
+        # Create token callback if event_manager is available
+        # This wraps the event_manager's on_token method to match OnTokenFunctionType Protocol
+        on_token_callback: OnTokenFunctionType | None = None
+        if self._event_manager:
+            on_token_callback = cast("OnTokenFunctionType", self._event_manager.on_token)
+
         try:
             result = await process_agent_events(
                 runnable.astream_events(
@@ -203,6 +228,7 @@ async def run_agent(
                 ),
                 agent_message,
                 cast("SendMessageFunctionType", self.send_message),
+                on_token_callback,
             )
         except ExceptionWithMessageError as e:
             if hasattr(e, "agent_message") and hasattr(e.agent_message, "id"):