Helpers
Helper components provide utility functions to help manage data, tasks, and other components in your flow.
Chat Memory
This component retrieves and manages chat messages from Langflow tables or an external memory.
Parameters
Name | Display Name | Info |
---|---|---|
memory |
External Memory |
Retrieve messages from an external memory. If empty, it will use 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 will be 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. |
Name | Display Name | Info |
---|---|---|
messages |
Messages (Data) |
Retrieved messages as Data objects. |
messages_text |
Messages (Text) |
Retrieved messages formatted as text. |
lc_memory |
Memory |
Built LangChain memory object. |
Component code
ChatMemory.py
404: Not Found
Combine Text
This component concatenates two text sources into a single text chunk using a specified delimiter.
Parameters
Name | Display Name | Info |
---|---|---|
first_text |
First Text |
The first text input to concatenate. |
second_text |
Second Text |
The second text input to concatenate. |
delimiter |
Delimiter |
A string used to separate the two text inputs. Defaults to a space. |
Component code
CombineText.py
from langflow.custom import Component
from langflow.io import MessageTextInput, Output
from langflow.schema.message import Message
class CombineTextComponent(Component):
display_name = "Combine Text"
description = "Concatenate two text sources into a single text chunk using a specified delimiter."
icon = "merge"
name = "CombineText"
inputs = [
MessageTextInput(
name="text1",
display_name="First Text",
info="The first text input to concatenate.",
),
MessageTextInput(
name="text2",
display_name="Second Text",
info="The second text input to concatenate.",
),
MessageTextInput(
name="delimiter",
display_name="Delimiter",
info="A string used to separate the two text inputs. Defaults to a whitespace.",
value=" ",
),
]
outputs = [
Output(display_name="Combined Text", name="combined_text", method="combine_texts"),
]
def combine_texts(self) -> Message:
combined = self.delimiter.join([self.text1, self.text2])
self.status = combined
return Message(text=combined)
Create List
This component dynamically creates a record with a specified number of fields.
Parameters
Name | Display Name | Info |
---|---|---|
n_fields |
Number of Fields |
Number of fields to be added to the record. |
text_key |
Text Key |
Key used as text. |
Component code
CreateList.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"
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
Custom Component
Use this component as a template to create your custom component.
Component code
CustomComponent.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 = "custom_components"
name = "CustomComponent"
inputs = [
MessageTextInput(name="input_value", display_name="Input Value", value="Hello, World!"),
]
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
Filter Data
This component converts LangChain documents into Data.
Component code
FilterData.py
from typing import List
from langflow.custom import Component
from langflow.io import DataInput, MessageTextInput, Output
from langflow.schema import Data
class FilterDataComponent(Component):
display_name = "Filter Data"
description = "Filters a Data object based on a list of keys."
icon = "filter"
beta = True
name = "FilterData"
inputs = [
DataInput(
name="data",
display_name="Data",
info="Data object to filter.",
),
MessageTextInput(
name="filter_criteria",
display_name="Filter Criteria",
info="List of keys to filter by.",
is_list=True,
),
]
outputs = [
Output(display_name="Filtered Data", name="filtered_data", method="filter_data"),
]
def filter_data(self) -> Data:
filter_criteria: List[str] = self.filter_criteria
data = self.data.data if isinstance(self.data, Data) else {}
# Filter the data
filtered = {key: value for key, value in data.items() if key in filter_criteria}
# Create a new Data object with the filtered data
filtered_data = Data(data=filtered)
self.status = filtered_data
return filtered_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
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. |
Name | Display Name | Info |
---|---|---|
task_output |
Task |
The built hierarchical task. |
Component code
HierarchicalTask.py
from langflow.base.agents.crewai.tasks import HierarchicalTask
from langflow.custom import Component
from langflow.io import HandleInput, MultilineInput, Output
class HierarchicalTaskComponent(Component):
display_name: str = "Hierarchical Task"
description: str = "Each task must have a description, an expected output and an agent responsible for execution."
icon = "CrewAI"
inputs = [
MultilineInput(
name="task_description",
display_name="Description",
info="Descriptive text detailing task's purpose and execution.",
),
MultilineInput(
name="expected_output",
display_name="Expected Output",
info="Clear definition of expected task outcome.",
),
HandleInput(
name="tools",
display_name="Tools",
input_types=["Tool"],
is_list=True,
info="List of tools/resources limited for task execution. Uses the Agent tools by default.",
required=False,
advanced=True,
),
]
outputs = [
Output(display_name="Task", name="task_output", method="build_task"),
]
def build_task(self) -> HierarchicalTask:
task = HierarchicalTask(
description=self.task_description,
expected_output=self.expected_output,
tools=self.tools or [],
)
self.status = task
return task
ID Generator
This component generates a unique ID.
Parameters
Name | Display Name | Info |
---|---|---|
value |
Value |
Unique ID generated. |
Component code
IDGenerator.py
import uuid
from typing import Any, Optional
from langflow.custom import CustomComponent
from langflow.schema.dotdict import dotdict
class IDGeneratorComponent(CustomComponent):
display_name = "ID Generator"
description = "Generates a unique ID."
name = "IDGenerator"
def update_build_config( # type: ignore
self, build_config: dotdict, field_value: Any, field_name: Optional[str] = None
):
if field_name == "unique_id":
build_config[field_name]["value"] = str(uuid.uuid4())
return build_config
def build_config(self):
return {
"unique_id": {
"display_name": "Value",
"refresh_button": True,
}
}
def build(self, unique_id: str) -> str:
return unique_id
Merge Data
Component code
MergeData.py
from langflow.custom import CustomComponent
from langflow.schema import Data
class MergeDataComponent(CustomComponent):
display_name = "Merge Data"
description = "Combines multiple data sources into a single unified Data object."
beta: bool = True
name = "MergeData"
field_config = {
"data": {"display_name": "Data"},
}
def build(self, data: list[Data]) -> Data:
if not data:
return Data()
if len(data) == 1:
return data[0]
merged_data = Data()
for value in data:
if merged_data is None:
merged_data = value
else:
merged_data += value
self.status = merged_data
return merged_data
Parse Data
The ParseData component converts Data objects into plain text using a specified template. This component transforms structured data into human-readable text formats, allowing for customizable output through the use of templates.
Parameters
Name | Display Name | Info |
---|---|---|
data |
Data |
The data to convert to text |
template |
Template |
The template to use for formatting the data. It can contain the keys {text}, {data} or any other key in the Data |
sep |
Separator |
The separator to use between multiple data items |
Name | Display Name | Info |
---|---|---|
text |
Text |
The resulting formatted text string as a Message object |
Component code
ParseData.py
from langflow.custom import Component
from langflow.helpers.data import data_to_text
from langflow.io import DataInput, MultilineInput, Output, StrInput
from langflow.schema.message import Message
class ParseDataComponent(Component):
display_name = "Parse Data"
description = "Convert Data into plain text following a specified template."
icon = "braces"
name = "ParseData"
inputs = [
DataInput(name="data", display_name="Data", info="The data to convert to text."),
MultilineInput(
name="template",
display_name="Template",
info="The template to use for formatting the data. It can contain the keys {text}, {data} or any other key in the Data.",
value="{text}",
),
StrInput(name="sep", display_name="Separator", advanced=True, value="\n"),
]
outputs = [
Output(display_name="Text", name="text", method="parse_data"),
]
def parse_data(self) -> Message:
data = self.data if isinstance(self.data, list) else [self.data]
template = self.template
result_string = data_to_text(template, data, sep=self.sep)
self.status = result_string
return Message(text=result_string)
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
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. |
Name | Display Name | Info |
---|---|---|
task_output |
Task |
The built sequential task or list of tasks. |
Component code
SequentialTask.py
from langflow.base.agents.crewai.tasks import SequentialTask
from langflow.custom import Component
from langflow.io import BoolInput, HandleInput, MultilineInput, Output
class SequentialTaskComponent(Component):
display_name: str = "Sequential Task"
description: str = "Each task must have a description, an expected output and an agent responsible for execution."
icon = "CrewAI"
inputs = [
MultilineInput(
name="task_description",
display_name="Description",
info="Descriptive text detailing task's purpose and execution.",
),
MultilineInput(
name="expected_output",
display_name="Expected Output",
info="Clear definition of expected task outcome.",
),
HandleInput(
name="tools",
display_name="Tools",
input_types=["Tool"],
is_list=True,
info="List of tools/resources limited for task execution. Uses the Agent tools by default.",
required=False,
advanced=True,
),
HandleInput(
name="agent",
display_name="Agent",
input_types=["Agent"],
info="CrewAI Agent that will perform the task",
required=True,
),
HandleInput(
name="task",
display_name="Task",
input_types=["SequentialTask"],
info="CrewAI Task that will perform the task",
),
BoolInput(
name="async_execution",
display_name="Async Execution",
value=True,
advanced=True,
info="Boolean flag indicating asynchronous task execution.",
),
]
outputs = [
Output(display_name="Task", name="task_output", method="build_task"),
]
def build_task(self) -> list[SequentialTask]:
tasks: list[SequentialTask] = []
task = SequentialTask(
description=self.task_description,
expected_output=self.expected_output,
tools=self.agent.tools,
async_execution=False,
agent=self.agent,
)
tasks.append(task)
self.status = task
if self.task:
if isinstance(self.task, list) and all(isinstance(task, SequentialTask) for task in self.task):
tasks = self.task + tasks
elif isinstance(self.task, SequentialTask):
tasks = [self.task] + tasks
return tasks
Split Text
This component splits text into chunks of a specified length.
Parameters
Name | Display Name | Info |
---|---|---|
texts |
Texts |
Texts to split. |
separators |
Separators |
Characters to split on. Defaults to a space. |
max_chunk_size |
Max Chunk Size |
The maximum length (in characters) of each chunk. |
chunk_overlap |
Chunk Overlap |
The amount of character overlap between chunks. |
recursive |
Recursive |
Whether to split recursively. |
Component code
SplitText.py
from typing import List
from langchain_text_splitters import CharacterTextSplitter
from langflow.custom import Component
from langflow.io import HandleInput, IntInput, MessageTextInput, Output
from langflow.schema import Data
from langflow.utils.util import unescape_string
class SplitTextComponent(Component):
display_name: str = "Split Text"
description: str = "Split text into chunks based on specified criteria."
icon = "scissors-line-dashed"
name = "SplitText"
inputs = [
HandleInput(
name="data_inputs",
display_name="Data Inputs",
info="The data to split.",
input_types=["Data"],
is_list=True,
),
IntInput(
name="chunk_overlap",
display_name="Chunk Overlap",
info="Number of characters to overlap between chunks.",
value=200,
),
IntInput(
name="chunk_size",
display_name="Chunk Size",
info="The maximum number of characters in each chunk.",
value=1000,
),
MessageTextInput(
name="separator",
display_name="Separator",
info="The character to split on. Defaults to newline.",
value="\n",
),
]
outputs = [
Output(display_name="Chunks", name="chunks", method="split_text"),
]
def _docs_to_data(self, docs):
data = []
for doc in docs:
data.append(Data(text=doc.page_content, data=doc.metadata))
return data
def split_text(self) -> List[Data]:
separator = unescape_string(self.separator)
documents = []
for _input in self.data_inputs:
if isinstance(_input, Data):
documents.append(_input.to_lc_document())
splitter = CharacterTextSplitter(
chunk_overlap=self.chunk_overlap,
chunk_size=self.chunk_size,
separator=separator,
)
docs = splitter.split_documents(documents)
data = self._docs_to_data(docs)
self.status = data
return data
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
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. |
Name | Display Name | Info |
---|---|---|
stored_messages |
Stored Messages |
The list of stored messages after the current message has been added. |
Component code
StoreMessage.py
from langflow.custom import Component
from langflow.inputs import MessageInput, StrInput, HandleInput
from langflow.schema.message import Message
from langflow.template import Output
from langflow.memory import get_messages, store_message
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.",
),
StrInput(
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,
),
StrInput(
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,
),
StrInput(
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="",
),
]
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