How to make custom components support streaming output and connect to chat output components?

[singhwang]

In langflow components have openai component than can stream its output into chat output. I want to use this custom component code as a framework to achieve similar data streaming output and connect it to the chat output component. How can I achieve this?

from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema.message import Message


class CustomComponent(Component):
    display_name = "Custom Component"
    description = "Use as a template to create your own component."
    documentation: str = "https://docs.langflow.org/components-custom-components"
    icon = "code"
    name = "CustomComponent"

    inputs = [
        MessageTextInput(
            name="input_value",
            display_name="Input Value",
            info="This is a custom component Input",
            value="Hello, World!",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="build_output"),
    ]

    def build_output(self) -> Message:
        message = Message(text=self.input_value)
        self.status = message
        return message

[dosubot[bot]]

To make your custom component support streaming output and connect it to the Chat Output component in Langflow, you need to:

1. Implement a streaming method in your component
   Your component should provide a method (commonly called stream) that yields data chunks (e.g., tokens or message parts) as they are generated. This is similar to how the OpenAI component and tests for streaming are structured. For example, you can use an async generator to yield message chunks:

   from langflow.custom import Component
   from langflow.io import MessageTextInput, Output
   from langflow.schema.message import Message
   
   class CustomComponent(Component):
       display_name = "Custom Component"
       description = "Streams output as chunks."
       icon = "code"
       name = "CustomComponent"
   
       inputs = [
           MessageTextInput(
               name="input_value",
               display_name="Input Value",
               info="This is a custom component Input",
               value="Hello, World!",
               tool_mode=True,
           ),
       ]
   
       outputs = [
           Output(display_name="Output", name="output", method="build_output"),
       ]
   
       async def stream(self):
           # Example: stream each word as a chunk
           for word in self.input_value.split():
               yield word + " "
               await asyncio.sleep(0.1)  # Simulate delay
   
       def build_output(self) -> Message:
           # Return a Message whose text is an async generator
           return Message(text=self.stream())

   This approach is inspired by the streaming test and how the OpenAI component streams output using an event handler and yields text chunks as they arrive [test example].

2. Expose the streaming method
   Langflow expects a stream method on the component for streaming support. The backend checks for this method to determine if the component supports streaming [chat.py backend].

3. Connect to the Chat Output component
   The Chat Output component is designed to accept streaming data. When you connect your custom component’s output to the Chat Output, Langflow will handle the streaming and update the chat UI as new chunks arrive [ChatOutput implementation].

4. How it works in the UI
   The frontend listens for streaming events and updates the chat message as each chunk is received, so your users will see the message appear incrementally [chatView frontend].

Summary of steps:

• Implement an async generator (stream method) in your component that yields message chunks.
• Return a Message object with its text set to the generator.
• Connect your component’s output to the Chat Output component in your flow.

References:

• Streaming test example
• ChatOutput component
• Backend streaming logic
• Frontend streaming handling

If you need a more advanced example (e.g., streaming from an external API or handling errors), let me know!

To continue talking to Dosu, mention @dosu-bot.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[singhwang]

Using the sample code you provided, the error is as follows:
An error occured in the Chat Output Component, stopping your flow. See below for more details.
Error details:
AttributeError
Details: 'str' object has no attribute 'content'

[dosubot[bot]]

The error 'str' object has no attribute 'content' occurs because the Message class in Langflow does not have a content attribute—its main attribute for message data is text. The Chat Output component expects a Message object with a text attribute (which can be a string, iterator, or async iterator for streaming), not a content attribute or a plain string with a content property [Message class definition].

To fix this and enable streaming, your custom component should:

• Ensure the Message object is constructed with the text attribute set to either a string or an async generator (for streaming).
• Not use or expect a content attribute anywhere in your code.

Here’s a corrected example for a streaming custom component:

import asyncio
from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema.message import Message

class CustomComponent(Component):
    display_name = "Custom Component"
    description = "Streams output as chunks."
    icon = "code"
    name = "CustomComponent"

    inputs = [
        MessageTextInput(
            name="input_value",
            display_name="Input Value",
            info="This is a custom component Input",
            value="Hello, World!",
            tool_mode=True,
        ),
    ]

    outputs = [
        Output(display_name="Output", name="output", method="build_output"),
    ]

    async def stream(self):
        # Stream each word as a chunk
        for word in self.input_value.split():
            yield word + " "
            await asyncio.sleep(0.1)  # Simulate delay

    def build_output(self) -> Message:
        # Set text to the async generator for streaming
        return Message(text=self.stream())

This approach ensures the Message object is constructed correctly and is compatible with the Chat Output component, which expects the text attribute for streaming [Chat Output implementation].

If you still encounter issues, double-check that all connections in your flow are passing Message objects (not plain strings) and that you are not referencing a content attribute anywhere in your custom or output components.

To continue talking to Dosu, mention @dosu-bot.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[singhwang]

Same issue, no change in error. The key here is whether the custom component can cooperate with the chat output component to complete the streaming output.

[singhwang]

I didn't wait for a valid reply, I figured out a way to solve it myself, so I'll close this Q&A for now.