fix: prevent streaming response duplication with custom OpenAI endpoints

When streaming, _process_chunk emits both an add_message event (to
initialise the chat bubble) and a token event for each chunk.  Consumers
that process both event types — such as the OpenAI Responses API
endpoint — end up yielding the same content twice: once from the
add_message text and once from the token data.

Root-cause fix: set the first-chunk add_message text to an empty string
so streamed content is delivered exclusively through token events.

Defensive fix: track whether token events have been received in the
OpenAI Responses streaming handler and skip add_message text content
when tokens are active, preventing duplication even if the root cause
fix is bypassed.

Fixes #10719

Author: skylarkoo7

src/backend/base/langflow/api/v1/openai_responses.py

@@ -132,6 +132,7 @@ async def openai_stream_generator() -> AsyncGenerator[str, None]:
                 tool_call_counter = 0
                 processed_tools = set()  # Track processed tool calls to avoid duplicates
                 previous_content = ""  # Track content already sent to calculate deltas
+                has_token_events = False  # Track whether token events are being received
 
                 async for event_data in consume_and_yield(asyncio_queue, asyncio_queue_client_consumed):
                     if event_data is None:
@@ -161,6 +162,7 @@ async def openai_stream_generator() -> AsyncGenerator[str, None]:
                                 # Handle add_message events
                                 if event_type == "token":
                                     token_data = data.get("chunk", "")
+                                    has_token_events = True
                                     await logger.adebug(
                                         "[OpenAIResponses][stream] token: token_data=%s",
                                         token_data,
@@ -195,11 +197,18 @@ async def openai_stream_generator() -> AsyncGenerator[str, None]:
                                         message_state,
                                     )
 
-                                    # Skip processing text content if state is "complete"
-                                    # All content has already been streamed via token events
-                                    if message_state == "complete":
+                                    # Skip processing text content when token events are
+                                    # the authoritative source of streamed content.  The
+                                    # component emits both an add_message and a token
+                                    # event for the first chunk, so processing both would
+                                    # cause the same content to appear twice in the
+                                    # response stream (#10719).
+                                    if message_state == "complete" or has_token_events:
                                         await logger.adebug(
-                                            "[OpenAIResponses][stream] skipping add_message with state=complete"
+                                            "[OpenAIResponses][stream] skipping add_message text "
+                                            "(state=%s, has_token_events=%s)",
+                                            message_state,
+                                            has_token_events,
                                         )
                                         # Still process content_blocks for tool calls, but skip text content
                                         text = ""

src/lfx/src/lfx/base/models/model.py

@@ -271,16 +271,24 @@ async def _get_chat_result(
     async def _handle_stream(self, runnable, inputs):
         """Handle streaming responses from the language model.
 
+        Creates a Message whose ``text`` is the async iterator returned by the
+        model's ``astream`` method and passes it to ``send_message``.  During
+        streaming the component emits *token* events (one per chunk) and a
+        single *add_message* event to initialise the message bubble in the
+        frontend.  The ``add_message`` event intentionally carries **empty**
+        text so that streamed content is delivered exclusively through token
+        events — this avoids duplication in API consumers that process both
+        event types (see #10719).
+
         Args:
-            runnable: The language model configured for streaming
-            inputs: The inputs to send to the model
+            runnable: The language model configured for streaming.
+            inputs: The inputs to send to the model.
 
         Returns:
             tuple: (Message object if connected to chat output, model result)
         """
         lf_message = None
         if self.is_connected_to_chat_output():
-            # Add a Message
             if hasattr(self, "graph"):
                 session_id = self.graph.session_id
             elif hasattr(self, "_session_id"):

src/lfx/src/lfx/custom/custom_component/component.py

@@ -1824,9 +1824,13 @@ async def _process_chunk(
         complete_message += chunk
         if self._event_manager:
             if first_chunk:
-                # Send the initial message only on the first chunk
+                # Send the initial message event to create the message
+                # bubble in the frontend.  Use empty text so the content
+                # is delivered exclusively through token events — sending
+                # the same text via both channels causes duplication in
+                # consumers like the OpenAI Responses endpoint (#10719).
                 msg_copy = message.model_copy()
-                msg_copy.text = complete_message
+                msg_copy.text = ""
                 await self._send_message_event(msg_copy, id_=message_id)
             await asyncio.to_thread(
                 self._event_manager.on_token,