Helpers

Helper components provide utility functions to help manage data, tasks, and other components in your flow.

Create List

This component takes a list of text inputs and converts each text into a data object. These data objects are then collected into a list, which is returned as the output.

Parameters

Inputs
Name Display Name Info

texts

Texts

Enter one or more texts. This input accepts multiple text entries.

Outputs
Display Name Name Info

Data List

list

A list of data objects created from the input texts.

Component code

create_list.py
from langflow.custom import Component
from langflow.inputs import StrInput
from langflow.schema import Data
from langflow.template import Output


class CreateListComponent(Component):
    display_name = "Create List"
    description = "Creates a list of texts."
    icon = "list"
    name = "CreateList"
    legacy = True

    inputs = [
        StrInput(
            name="texts",
            display_name="Texts",
            info="Enter one or more texts.",
            is_list=True,
        ),
    ]

    outputs = [
        Output(display_name="Data List", name="list", method="create_list"),
    ]

    def create_list(self) -> list[Data]:
        data = [Data(text=text) for text in self.texts]
        self.status = data
        return data

Current Date

The Current Date component returns the current date and time in a selected timezone. This component provides a flexible way to obtain timezone-specific date and time information within a Langflow pipeline.

Parameters

Inputs
Name Display Name Info

timezone

Timezone

Select the timezone for the current date and time.

Outputs
Name Display Name Info

current_date

Current Date

The resulting current date and time in the selected timezone.

Component code

current_date.py
from datetime import datetime
from zoneinfo import ZoneInfo

from loguru import logger

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


class CurrentDateComponent(Component):
    display_name = "Current Date"
    description = "Returns the current date and time in the selected timezone."
    icon = "clock"
    name = "CurrentDate"

    inputs = [
        DropdownInput(
            name="timezone",
            display_name="Timezone",
            options=[
                "UTC",
                "US/Eastern",
                "US/Central",
                "US/Mountain",
                "US/Pacific",
                "Europe/London",
                "Europe/Paris",
                "Europe/Berlin",
                "Europe/Moscow",
                "Asia/Tokyo",
                "Asia/Shanghai",
                "Asia/Singapore",
                "Asia/Dubai",
                "Australia/Sydney",
                "Australia/Melbourne",
                "Pacific/Auckland",
                "America/Sao_Paulo",
                "America/Mexico_City",
                "America/Toronto",
                "America/Vancouver",
                "Africa/Cairo",
                "Africa/Johannesburg",
                "Atlantic/Reykjavik",
                "Indian/Maldives",
                "America/Bogota",
                "America/Lima",
                "America/Santiago",
                "America/Buenos_Aires",
                "America/Caracas",
                "America/La_Paz",
                "America/Montevideo",
                "America/Asuncion",
                "America/Cuiaba",
            ],
            value="UTC",
            info="Select the timezone for the current date and time.",
            tool_mode=True,
        ),
    ]
    outputs = [
        Output(display_name="Current Date", name="current_date", method="get_current_date"),
    ]

    def get_current_date(self) -> Message:
        try:
            tz = ZoneInfo(self.timezone)
            current_date = datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S %Z")
            result = f"Current date and time in {self.timezone}: {current_date}"
            self.status = result
            return Message(text=result)
        except Exception as e:  # noqa: BLE001
            logger.opt(exception=True).debug("Error getting current date")
            error_message = f"Error: {e}"
            self.status = error_message
            return Message(text=error_message)

Custom Component

Use this component as a template to create your custom component.

Component code

custom_component.py
# from langflow.field_typing import Data
from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema import Data


class CustomComponent(Component):
    display_name = "Custom Component"
    description = "Use as a template to create your own component."
    documentation: str = "http://docs.langflow.org/components/custom"
    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) -> Data:
        data = Data(value=self.input_value)
        self.status = data
        return data

Hierarchical Task

This component creates and manages hierarchical tasks for CrewAI agents in a Playground environment.

For more information, see the CrewAI documentation.

Parameters

Inputs
Name Display Name Info

task_description

Description

Descriptive text detailing task’s purpose and execution.

expected_output

Expected Output

Clear definition of expected task outcome.

tools

Tools

List of tools/resources limited for task execution. Uses the Agent tools by default.

Outputs
Name Display Name Info

task_output

Task

The built hierarchical task.

Component code

hierarchical_task.py
404: Not Found

ID Generator

This component generates a unique ID.

Parameters

Outputs
Name Display Name Info

value

Value

Unique ID generated.

Component code

id_generator.py
import uuid
from typing import Any

from typing_extensions import override

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


class IDGeneratorComponent(Component):
    display_name = "ID Generator"
    description = "Generates a unique ID."
    icon = "fingerprint"
    name = "IDGenerator"

    inputs = [
        MessageTextInput(
            name="unique_id",
            display_name="Value",
            info="The generated unique ID.",
            refresh_button=True,
        ),
    ]

    outputs = [
        Output(display_name="ID", name="id", method="generate_id"),
    ]

    @override
    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
        if field_name == "unique_id":
            build_config[field_name]["value"] = str(uuid.uuid4())
        return build_config

    def generate_id(self) -> Message:
        unique_id = self.unique_id or str(uuid.uuid4())
        self.status = f"Generated ID: {unique_id}"
        return Message(text=unique_id)

Message history

This component retrieves and manages chat messages from Langflow tables or an external memory.

This component was named Chat Memory prior to Langflow version 1.1.0.

Parameters

Inputs
Name Display Name Info

memory

External Memory

Retrieve messages from an external memory. If empty, it uses the Langflow tables.

sender

Sender Type

Filter by sender type.

sender_name

Sender Name

Filter by sender name.

n_messages

Number of Messages

Number of messages to retrieve.

session_id

Session ID

The session ID of the chat. If empty, the current session ID parameter is used.

order

Order

Order of the messages.

template

Template

The template to use for formatting the data. It can contain the keys {text}, {sender} or any other key in the message data.

Outputs
Name Display Name Info

messages

Messages (data object)

Retrieved messages as data objects.

messages_text

Messages (text)

Retrieved messages formatted as text.

lc_memory

Memory

The created LangChain-compatible memory object.

Component code

memory.py
from langchain.memory import ConversationBufferMemory

from langflow.custom import Component
from langflow.field_typing import BaseChatMemory
from langflow.helpers.data import data_to_text
from langflow.inputs import HandleInput
from langflow.io import DropdownInput, IntInput, MessageTextInput, MultilineInput, Output
from langflow.memory import LCBuiltinChatMemory, get_messages
from langflow.schema import Data
from langflow.schema.message import Message
from langflow.utils.constants import MESSAGE_SENDER_AI, MESSAGE_SENDER_USER


class MemoryComponent(Component):
    display_name = "Message History"
    description = "Retrieves stored chat messages from Langflow tables or an external memory."
    icon = "message-square-more"
    name = "Memory"

    inputs = [
        HandleInput(
            name="memory",
            display_name="External Memory",
            input_types=["BaseChatMessageHistory"],
            info="Retrieve messages from an external memory. If empty, it will use the Langflow tables.",
        ),
        DropdownInput(
            name="sender",
            display_name="Sender Type",
            options=[MESSAGE_SENDER_AI, MESSAGE_SENDER_USER, "Machine and User"],
            value="Machine and User",
            info="Filter by sender type.",
            advanced=True,
        ),
        MessageTextInput(
            name="sender_name",
            display_name="Sender Name",
            info="Filter by sender name.",
            advanced=True,
        ),
        IntInput(
            name="n_messages",
            display_name="Number of Messages",
            value=100,
            info="Number of messages to retrieve.",
            advanced=True,
        ),
        MessageTextInput(
            name="session_id",
            display_name="Session ID",
            info="The session ID of the chat. If empty, the current session ID parameter will be used.",
            advanced=True,
        ),
        DropdownInput(
            name="order",
            display_name="Order",
            options=["Ascending", "Descending"],
            value="Ascending",
            info="Order of the messages.",
            advanced=True,
        ),
        MultilineInput(
            name="template",
            display_name="Template",
            info="The template to use for formatting the data. "
            "It can contain the keys {text}, {sender} or any other key in the message data.",
            value="{sender_name}: {text}",
            advanced=True,
        ),
    ]

    outputs = [
        Output(display_name="Data", name="messages", method="retrieve_messages"),
        Output(display_name="Text", name="messages_text", method="retrieve_messages_as_text"),
    ]

    def retrieve_messages(self) -> Data:
        sender = self.sender
        sender_name = self.sender_name
        session_id = self.session_id
        n_messages = self.n_messages
        order = "DESC" if self.order == "Descending" else "ASC"

        if sender == "Machine and User":
            sender = None

        if self.memory:
            # override session_id
            self.memory.session_id = session_id

            stored = self.memory.messages
            # langchain memories are supposed to return messages in ascending order
            if order == "DESC":
                stored = stored[::-1]
            if n_messages:
                stored = stored[:n_messages]
            stored = [Message.from_lc_message(m) for m in stored]
            if sender:
                expected_type = MESSAGE_SENDER_AI if sender == MESSAGE_SENDER_AI else MESSAGE_SENDER_USER
                stored = [m for m in stored if m.type == expected_type]
        else:
            stored = get_messages(
                sender=sender,
                sender_name=sender_name,
                session_id=session_id,
                limit=n_messages,
                order=order,
            )
        self.status = stored
        return stored

    def retrieve_messages_as_text(self) -> Message:
        stored_text = data_to_text(self.template, self.retrieve_messages())
        self.status = stored_text
        return Message(text=stored_text)

    def build_lc_memory(self) -> BaseChatMemory:
        chat_memory = self.memory or LCBuiltinChatMemory(flow_id=self.flow_id, session_id=self.session_id)
        return ConversationBufferMemory(chat_memory=chat_memory)

Sequential Task

This component creates and manage sequential tasks for CrewAI agents. It builds a SequentialTask object with the provided description, expected output, and agent, allowing for the specification of tools and asynchronous execution.

For more information, see the CrewAI documentation.

Parameters

Inputs
Name Display Name Info

task_description

Description

Descriptive text detailing task’s purpose and execution.

expected_output

Expected Output

Clear definition of expected task outcome.

tools

Tools

List of tools/resources limited for task execution. Uses the Agent tools by default.

agent

Agent

CrewAI Agent that will perform the task.

task

Task

CrewAI Task that will perform the task.

async_execution

Async Execution

Boolean flag indicating asynchronous task execution.

Outputs
Name Display Name Info

task_output

Task

The built sequential task or list of tasks.

Component code

sequential_task.py
404: Not Found

Store Message

This component stores chat messages or text into Langflow tables or an external memory.

It provides flexibility in managing message storage and retrieval within a chat system.

Parameters

Inputs
Name Display Name Info

message

Message

The chat message to be stored. (Required)

memory

External Memory

The external memory to store the message. If empty, it will use the Langflow tables.

sender

Sender

The sender of the message. Can be Machine or User. If empty, the current sender parameter will be used.

sender_name

Sender Name

The name of the sender. Can be AI or User. If empty, the current sender parameter will be used.

session_id

Session ID

The session ID of the chat. If empty, the current session ID parameter will be used.

Outputs
Name Display Name Info

stored_messages

Stored Messages

The list of stored messages after the current message has been added.

Component code

store_message.py
from langflow.custom import Component
from langflow.inputs import HandleInput, MessageInput
from langflow.inputs.inputs import MessageTextInput
from langflow.memory import get_messages, store_message
from langflow.schema.message import Message
from langflow.template import Output
from langflow.utils.constants import MESSAGE_SENDER_AI, MESSAGE_SENDER_NAME_AI


class StoreMessageComponent(Component):
    display_name = "Store Message"
    description = "Stores a chat message or text into Langflow tables or an external memory."
    icon = "save"
    name = "StoreMessage"

    inputs = [
        MessageInput(name="message", display_name="Message", info="The chat message to be stored.", required=True),
        HandleInput(
            name="memory",
            display_name="External Memory",
            input_types=["BaseChatMessageHistory"],
            info="The external memory to store the message. If empty, it will use the Langflow tables.",
        ),
        MessageTextInput(
            name="sender",
            display_name="Sender",
            info="The sender of the message. Might be Machine or User. "
            "If empty, the current sender parameter will be used.",
            advanced=True,
        ),
        MessageTextInput(
            name="sender_name",
            display_name="Sender Name",
            info="The name of the sender. Might be AI or User. If empty, the current sender parameter will be used.",
            advanced=True,
        ),
        MessageTextInput(
            name="session_id",
            display_name="Session ID",
            info="The session ID of the chat. If empty, the current session ID parameter will be used.",
            value="",
            advanced=True,
        ),
    ]

    outputs = [
        Output(display_name="Stored Messages", name="stored_messages", method="store_message"),
    ]

    def store_message(self) -> Message:
        message = self.message

        message.session_id = self.session_id or message.session_id
        message.sender = self.sender or message.sender or MESSAGE_SENDER_AI
        message.sender_name = self.sender_name or message.sender_name or MESSAGE_SENDER_NAME_AI

        if self.memory:
            # override session_id
            self.memory.session_id = message.session_id
            lc_message = message.to_lc_message()
            self.memory.add_messages([lc_message])
            stored = self.memory.messages
            stored = [Message.from_lc_message(m) for m in stored]
            if message.sender:
                stored = [m for m in stored if m.sender == message.sender]
        else:
            store_message(message, flow_id=self.graph.flow_id)
            stored = get_messages(session_id=message.session_id, sender_name=message.sender_name, sender=message.sender)
        self.status = stored
        return stored

Structured Output

The Structured Output component transforms LLM responses into structured data formats.

Parameters

Inputs
Name Display Name Info

llm

Language Model

The language model to use to generate the structured output.

input_value

Input message

The input message to be processed by the language model.

schema_name

Schema Name

Provide a name for the output data schema.

output_schema

Output Schema

Define the structure and data types for the model’s output.

multiple

Generate Multiple

Set to True if the model should generate a list of outputs instead of a single output.

Outputs
Name Display Name Info

structured_output

Structured Output

The resulting structured output based on the defined schema.

Component code

structured_output.py
from typing import cast

from pydantic import BaseModel, Field, create_model

from langflow.base.models.chat_result import get_chat_result
from langflow.custom import Component
from langflow.field_typing.constants import LanguageModel
from langflow.helpers.base_model import build_model_from_schema
from langflow.io import BoolInput, HandleInput, MessageTextInput, Output, StrInput, TableInput
from langflow.schema.data import Data


class StructuredOutputComponent(Component):
    display_name = "Structured Output"
    description = (
        "Transforms LLM responses into **structured data formats**. Ideal for extracting specific information "
        "or creating consistent outputs."
    )
    icon = "braces"

    inputs = [
        HandleInput(
            name="llm",
            display_name="Language Model",
            info="The language model to use to generate the structured output.",
            input_types=["LanguageModel"],
        ),
        MessageTextInput(name="input_value", display_name="Input message"),
        StrInput(
            name="schema_name",
            display_name="Schema Name",
            info="Provide a name for the output data schema.",
        ),
        TableInput(
            name="output_schema",
            display_name="Output Schema",
            info="Define the structure and data types for the model's output.",
            value=[
                {
                    "name": "name",
                    "display_name": "Name",
                    "type": "str",
                    "description": "Specify the name of the output field.",
                },
                {
                    "name": "description",
                    "display_name": "Description",
                    "type": "str",
                    "description": "Describe the purpose of the output field.",
                },
                {
                    "name": "type",
                    "display_name": "Type",
                    "type": "str",
                    "description": (
                        "Indicate the data type of the output field " "(e.g., str, int, float, bool, list, dict)."
                    ),
                    "default": "text",
                },
                {
                    "name": "multiple",
                    "display_name": "Multiple",
                    "type": "boolean",
                    "description": "Set to True if this output field should be a list of the specified type.",
                    "default": "False",
                },
            ],
        ),
        BoolInput(
            name="multiple",
            display_name="Generate Multiple",
            info="Set to True if the model should generate a list of outputs instead of a single output.",
        ),
    ]

    outputs = [
        Output(name="structured_output", display_name="Structured Output", method="build_structured_output"),
    ]

    def build_structured_output(self) -> Data:
        if not hasattr(self.llm, "with_structured_output"):
            msg = "Language model does not support structured output."
            raise TypeError(msg)
        if not self.output_schema:
            msg = "Output schema cannot be empty"
            raise ValueError(msg)

        _output_model = build_model_from_schema(self.output_schema)
        if self.multiple:
            output_model = create_model(
                self.schema_name,
                objects=(list[_output_model], Field(description=f"A list of {self.schema_name}.")),  # type: ignore[valid-type]
            )
        else:
            output_model = _output_model
        try:
            llm_with_structured_output = cast(LanguageModel, self.llm).with_structured_output(schema=output_model)  # type: ignore[valid-type, attr-defined]

        except NotImplementedError as exc:
            msg = f"{self.llm.__class__.__name__} does not support structured output."
            raise TypeError(msg) from exc
        config_dict = {
            "run_name": self.display_name,
            "project_name": self.get_project_name(),
            "callbacks": self.get_langchain_callbacks(),
        }
        output = get_chat_result(runnable=llm_with_structured_output, input_value=self.input_value, config=config_dict)
        if isinstance(output, BaseModel):
            output_dict = output.model_dump()
        else:
            msg = f"Output should be a Pydantic BaseModel, got {type(output)} ({output})"
            raise TypeError(msg)
        return Data(data=output_dict)

Was this helpful?

Give Feedback

How can we improve the documentation?

© 2024 DataStax | Privacy policy | Terms of use

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Pulsar, Pulsar, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries. Kubernetes is the registered trademark of the Linux Foundation.

General Inquiries: +1 (650) 389-6000, info@datastax.com