跳到主要内容

创建自定义 Python 组件

自定义组件通过继承自 Component 的 Python 类扩展 Langflow 的功能。这使得能够集成新功能、数据操作、外部服务和专用工具。

在 Langflow 基于节点的环境中，每个节点都是一个执行独立功能的"组件"。自定义组件是定义以下内容的 Python 类：

  • 输入 — 您的组件需要的数据或参数。
  • 输出 — 您的组件向下游节点提供的数据。
  • 逻辑 — 您如何处理输入以产生输出。

创建自定义组件的好处包括无限的可扩展性、可重用性、基于输入的自动 UI 字段生成，以及节点之间的类型安全连接。

创建自定义组件用于执行专门任务、调用 API 或添加高级逻辑。

Langflow 中的自定义组件基于：

  • 继承自 Component 的 Python 类。
  • 标识和描述组件的类级属性。
  • 确定数据流的输入和输出列表。
  • 用于日志记录和高级逻辑的内部变量。

类级属性​

定义这些属性来控制自定义组件的外观和行为：


_10
class MyCsvReader(Component):
_10
display_name = "CSV Reader" # 在节点标题中显示
_10
description = "Reads CSV files" # 工具提示文本
_10
icon = "file-text" # 视觉标识符
_10
name = "CSVReader" # 唯一的内部 ID
_10
documentation = "http://docs.example.com/csv_reader" # 可选

  • display_name：节点标题中的用户友好标签。
  • description：工具提示中显示的简要摘要。
  • icon：来自 Langflow 图标库的视觉标识符。
  • name：唯一的内部标识符。
  • documentation：指向外部文档的可选链接。
图标使用

Langflow 使用 Lucide 作为图标。要为您的组件分配图标，请将图标属性设置为 Lucide 图标的名称作为字符串，例如 icon = "file-text"。Langflow 自动从 Lucide 库渲染图标。

自定义组件的结构​

Langflow 自定义组件不仅仅是具有输入和输出的简单类。它包括具有可选生命周期步骤、输出生成、前端交互和逻辑组织的内部结构。

基本组件：

  • 继承自 langflow.custom.Component。
  • 声明元数据，如 display_name、description、icon 等。
  • 定义 inputs 和 outputs 列表。
  • 实现匹配输出规范的方法。

最小的自定义组件骨架包含以下内容：


_14
from langflow.custom import Component
_14
from langflow.template import Output
_14
_14
class MyComponent(Component):
_14
display_name = "My Component"
_14
description = "A short summary."
_14
icon = "sparkles"
_14
name = "MyComponent"
_14
_14
inputs = []
_14
outputs = []
_14
_14
def some_output_method(self):
_14
return ...

内部生命周期和执行流程​

Langflow 的引擎管理：

  • 实例化：创建组件并初始化内部结构。
  • 分配输入：来自 UI 或连接的值被分配给组件字段。
  • 验证和设置：可选钩子，如 _pre_run_setup。
  • 输出生成：run() 或 build_results() 触发输出方法。

可选钩子：

  • initialize_data 或 _pre_run_setup 可以在组件主要执行之前运行设置逻辑。
  • __call__、run() 或 _run() 可以被重写以自定义组件的调用方式或定义自定义执行逻辑。

输入和输出​

自定义组件输入使用以下属性定义：

  • name、display_name
  • 可选：info、value、advanced、is_list、tool_mode、real_time_refresh

例如：

  • StrInput：简单的文本输入。
  • DropdownInput：可选择的选项。
  • HandleInput：专门的连接。

自定义组件 Output 属性定义：

  • name、display_name、method
  • 可选：info

有关更多信息，请参阅自定义组件输入和输出。

关联方法​

每个输出都链接到一个方法：

  • 输出方法名称必须与方法名称匹配。
  • 该方法通常返回 Message、Data 或 DataFrame 等对象。
  • 该方法可以使用 self.<input_name> 访问输入。

例如：


_12
Output(
_12
display_name="File Contents",
_12
name="file_contents",
_12
method="read_file"
_12
)
_12
#...
_12
def read_file(self) -> Data:
_12
path = self.filename
_12
with open(path, "r") as f:
_12
content = f.read()
_12
self.status = f"Read {len(content)} chars from {path}"
_12
return Data(data={"content": content})

具有多个输出的组件​

一个组件可以定义多个输出。 每个输出可以有不同的对应方法。 例如：


_10
outputs = [
_10
Output(display_name="Processed Data", name="processed_data", method="process_data"),
_10
Output(display_name="Debug Info", name="debug_info", method="provide_debug_info"),
_10
]

使用 group_outputs 的输出分组行为​

默认情况下，在 Langflow 中定义多个输出的组件将在 UI 中将它们显示为下拉列表。此行为由 group_outputs 参数控制。

  • group_outputs=False（默认）： 当组件有多个输出且未指定 group_outputs（或设置为 False）时，输出会被分组到下拉列表中。用户一次只能从 UI 中选择一个输出。

  • group_outputs=True： 所有输出将同时在 UI 中显示。当组件预期返回应在下游并行使用的多个值时，这很有用。

示例：​

  1. group_outputs=False（默认行为）

_12
outputs = [
_12
Output(
_12
name="structured_output",
_12
display_name="Structured Output",
_12
method="build_structured_output",
_12
),
_12
Output(
_12
name="dataframe_output",
_12
display_name="DataFrame Output",
_12
method="build_structured_dataframe",
_12
),
_12
]

在此示例中，两个输出都将通过 UI 中的下拉选择可用。

注意：由于 group_outputs=False 是默认行为，因此不需要在组件中显式设置。

  1. group_outputs=True

_14
outputs = [
_14
Output(
_14
name="true_result",
_14
display_name="True",
_14
method="true_response",
_14
group_outputs=True,
_14
),
_14
Output(
_14
name="false_result",
_14
display_name="False",
_14
method="false_response",
_14
group_outputs=True,
_14
),
_14
]

在此示例中，两个输出都将独立显示并可在 UI 中直接选择。

何时使用​

  • 当组件预期根据流程逻辑仅返回其中一个输出时，使用 group_outputs=False。

  • 当组件应同时暴露多个输出时，使用 group_outputs=True，例如用于并行使用的结构化数据和表格。

常见内部模式​

_pre_run_setup()​

使用计数器设置初始化自定义组件：


_10
def _pre_run_setup(self):
_10
if not hasattr(self, "_initialized"):
_10
self._initialized = True
_10
self.iteration = 0

重写 run 或 _run​

您可以重写 async def _run(self): ... 来定义自定义执行逻辑，尽管基类的默认行为通常涵盖大多数情况。

在 self.ctx 中存储数据​

使用 self.ctx 作为跨组件执行流程的数据或计数器的共享存储：


_10
def some_method(self):
_10
count = self.ctx.get("my_count", 0)
_10
self.ctx["my_count"] = count + 1

目录结构要求​

默认情况下，Langflow 在 langflow/components 目录中查找自定义组件。

如果您使用 LANGFLOW_COMPONENTS_PATH 环境变量在不同位置创建自定义组件，组件必须按照特定的目录结构进行组织，以便正确加载并在 UI 中显示：


_10
/your/custom/components/path/ # LANGFLOW_COMPONENTS_PATH 设置的基础目录
_10
└── category_name/ # 决定菜单名称的必需分类子文件夹
_10
└── custom_component.py # 组件文件

组件必须放置在分类文件夹内，而不是直接放在基础目录中。 分类文件夹名称决定组件在 Langflow 组件菜单中的显示位置。

例如，要将组件添加到 Helpers 分类，请将其放在 helpers 子文件夹中：


_10
/app/custom_components/ # LANGFLOW_COMPONENTS_PATH
_10
└── helpers/ # 显示在"Helpers"分类中
_10
└── custom_component.py # 您的组件

您可以拥有多个分类文件夹来将组件组织到不同的分类中：


_10
/app/custom_components/
_10
├── helpers/
_10
│ └── helper_component.py
_10
└── tools/
_10
└── tool_component.py

这种文件夹结构是 Langflow 正确发现和加载自定义组件所必需的。直接放置在基础目录中的组件将不会被加载。


_10
/app/custom_components/ # LANGFLOW_COMPONENTS_PATH
_10
└── custom_component.py # 不会被加载 - 缺少分类文件夹！

自定义组件输入和输出​

输入和输出定义了数据如何流经组件、在 UI 中的显示方式以及如何验证与其他组件的连接。

输入​

输入在类级别的 inputs 列表中定义。当 Langflow 加载组件时，它使用此列表在 UI 中渲染组件字段和端口。用户或其他组件提供值或连接以填充这些输入。

输入通常是来自 langflow.io 的类的实例（如 StrInput、DataInput 或 MessageTextInput）。最常见的构造函数参数是：

  • name：内部变量名，通过 self.<name> 访问。
  • display_name：在 UI 中显示给用户的标签。
  • info （可选）：工具提示或简短描述。
  • value （可选）：默认值。
  • advanced （可选）：如果为 True，将字段移至"高级"部分。
  • required （可选）：如果为 True，强制用户提供值。
  • is_list （可选）：如果为 True，允许多个值。
  • input_types （可选）：限制允许的连接类型（例如，["Data"]、["LanguageModel"]）。

以下是最常用的输入类及其典型用法。

文本输入：用于简单的文本条目。

  • StrInput 创建单行文本字段。
  • MultilineInput 创建多行文本区域。

数值和布尔输入：确保用户只能输入有效的数值或布尔数据。

  • BoolInput、IntInput 和 FloatInput 为布尔值、整数和浮点值提供字段，确保类型一致性。

下拉菜单：用于从预定义选项中选择，对模式或级别很有用。

  • DropdownInput

密钥：用于敏感数据的专用输入，确保输入在 UI 中隐藏。

  • SecretStrInput 用于 API 密钥和密码。

专用数据输入：确保 UI 中的类型检查和颜色编码连接。

  • DataInput 期望一个 Data 对象（通常带有 .data 和可选的 .text）。
  • MessageInput 期望一个 Message 对象，用于聊天或基于代理的流程。
  • MessageTextInput 简化对 Message 的 .text 字段的访问。

基于句柄的输入：用于连接特定类型的输出，确保正确的管道连接。

  • HandleInput

文件上传：允许用户直接通过 UI 上传文件或从其他组件接收文件路径。

  • FileInput

列表：设置 is_list=True 以接受多个值，适用于批处理或分组操作。

此示例定义了三个输入：文本字段（StrInput）、布尔切换（BoolInput）和下拉选择（DropdownInput）。


_10
from langflow.io import StrInput, BoolInput, DropdownInput
_10
_10
inputs = [
_10
StrInput(name="title", display_name="Title"),
_10
BoolInput(name="enabled", display_name="Enabled", value=True),
_10
DropdownInput(name="mode", display_name="Mode", options=["Fast", "Safe", "Experimental"], value="Safe")
_10
]

输出​

输出在类级别的 outputs 列表中定义。当 Langflow 渲染组件时，每个输出成为 UI 中的连接点。当您将某些内容连接到输出时，Langflow 自动调用相应的方法并将返回的对象传递给下一个组件。

输出通常是来自 langflow.io 的 Output 实例，具有常见参数：

  • name：内部变量名。
  • display_name：在 UI 中显示的标签。
  • method：调用以产生输出的方法名称。
  • info （可选）：悬停时显示的帮助文本。

方法必须存在于类中，建议为其返回类型添加注释以进行更好的类型检查。 您还可以在方法内设置 self.status 消息以显示进度或日志。

常见返回类型：

  • Message：结构化聊天消息。
  • Data：具有 .data 和可选 .text 的灵活对象。
  • DataFrame：基于 Pandas 的表格（langflow.schema.DataFrame）。
  • 原始类型：str、int、bool（如果需要类型/颜色一致性，则不推荐）。

在此示例中，DataToDataFrame 组件使用输出列表定义其输出。df_out 输出链接到 build_df 方法，因此当在 UI 中连接时，Langflow 调用此方法并将其返回的 DataFrame 传递给下一个节点。这演示了每个输出如何映射到生成实际输出数据的方法。


_37
from langflow.custom import Component
_37
from langflow.io import DataInput, Output
_37
from langflow.schema import Data, DataFrame
_37
_37
class DataToDataFrame(Component):
_37
display_name = "Data to DataFrame"
_37
description = "Convert multiple Data objects into a DataFrame"
_37
icon = "table"
_37
name = "DataToDataFrame"
_37
_37
inputs = [
_37
DataInput(
_37
name="items",
_37
display_name="Data Items",
_37
info="List of Data objects to convert",
_37
is_list=True
_37
)
_37
]
_37
_37
outputs = [
_37
Output(
_37
name="df_out",
_37
display_name="DataFrame Output",
_37
method="build_df"
_37
)
_37
]
_37
_37
def build_df(self) -> DataFrame:
_37
rows = []
_37
for item in self.items:
_37
row_dict = item.data.copy() if item.data else {}
_37
row_dict["text"] = item.get_text() or ""
_37
rows.append(row_dict)
_37
_37
df = DataFrame(rows)
_37
self.status = f"Built DataFrame with {len(rows)} rows."
_37
return df

工具模式​

您可以通过设置参数 tool_mode=True 将自定义组件配置为工具。这允许组件在 Langflow 的工具模式工作流中使用，例如被代理组件使用。

Langflow 目前支持以下工具模式的输入类型：

  • DataInput
  • DataFrameInput
  • PromptInput
  • MessageTextInput
  • MultilineInput
  • DropdownInput

_10
inputs = [
_10
MessageTextInput(
_10
name="message",
_10
display_name="Mensage",
_10
info="Enter the message that will be processed directly by the tool",
_10
tool_mode=True,
_10
),
_10
]

类型注释​

在 Langflow 中，类型注释允许 Langflow 可视化地指导用户并保持流程一致性。

类型注释提供：

  • 颜色编码：像 -> Data 或 -> Message 这样的输出获得不同的颜色。
  • 验证：Langflow 自动阻止不兼容的连接。
  • 可读性：开发人员可以快速理解数据流。
  • 开发工具：在代码编辑器中提供更好的代码建议和错误检查。

常见返回类型​

Message

用于聊天风格的输出。


_10
def produce_message(self) -> Message:
_10
return Message(text="Hello! from typed method!", sender="System")

在 UI 中，仅连接到与 Message 兼容的输入。

Data

用于结构化数据，如字典或部分文本。


_10
def get_processed_data(self) -> Data:
_10
processed = {"key1": "value1", "key2": 123}
_10
return Data(data=processed)

在 UI 中，仅与 DataInput 连接。

DataFrame

用于表格数据


_10
def build_df(self) -> DataFrame:
_10
pdf = pd.DataFrame({"A": [1, 2], "B": [3, 4]})
_10
return DataFrame(pdf)

在 UI 中，仅连接到 DataFrameInput。

原始类型（str、int、bool）

返回原始类型是允许的，但建议包装在 Data 或 Message 中以获得更好的 UI 一致性。


_10
def compute_sum(self) -> int:
_10
return sum(self.numbers)

类型注释提示​

使用类型注释时，请考虑以下最佳实践：

  • 始终注释输出：指定返回类型，如 -> Data、-> Message 或 -> DataFrame，以启用正确的 UI 颜色编码和验证。
  • 包装原始数据：使用 Data、Message 或 DataFrame 包装器，而不是返回普通结构。
  • 谨慎使用原始类型：直接的 str 或 int 返回对于简单流程是可以的，但包装可以提高灵活性。
  • 也要注释辅助函数：即使是内部的，类型化也能提高可维护性和清晰度。
  • 处理边缘情况：需要时，优先返回带有错误字段的结构化 Data。
  • 保持一致性：在组件之间使用相同的类型，使流程可预测且更易于构建。

启用动态字段​

在 Langflow 中，动态字段允许输入根据用户交互而改变或显示。您可以通过设置 dynamic=True 使输入动态化。 可选地，设置 real_time_refresh=True 会触发 update_build_config 方法，以实时调整输入的可见性或属性，创建一个仅根据用户选择显示相关字段的上下文 UI。

在此示例中，operator 字段通过 real_time_refresh=True 触发更新。 regex_pattern 字段最初是隐藏的，通过 dynamic=True 控制。


_22
from langflow.io import DropdownInput, StrInput
_22
_22
class RegexRouter(Component):
_22
display_name = "Regex Router"
_22
description = "Demonstrates dynamic fields for regex input."
_22
_22
inputs = [
_22
DropdownInput(
_22
name="operator",
_22
display_name="Operator",
_22
options=["equals", "contains", "regex"],
_22
value="equals",
_22
real_time_refresh=True,
_22
),
_22
StrInput(
_22
name="regex_pattern",
_22
display_name="Regex Pattern",
_22
info="Used if operator='regex'",
_22
dynamic=True,
_22
show=False,
_22
),
_22
]

实现 update_build_config​

当修改具有 real_time_refresh=True 的字段时，Langflow 调用 update_build_config 方法，传递更新的字段名、值和组件配置，以根据用户输入动态调整其他字段的可见性或属性。

此示例将在用户选择不同的操作符时显示或隐藏 regex_pattern 字段。


_10
def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None) -> dict:
_10
if field_name == "operator":
_10
if field_value == "regex":
_10
build_config["regex_pattern"]["show"] = True
_10
else:
_10
build_config["regex_pattern"]["show"] = False
_10
return build_config

其他动态字段控制​

您还可以在 update_build_config 中修改其他属性，例如：

  • required：设置 build_config["some_field"]["required"] = True/False

  • advanced：设置 build_config["some_field"]["advanced"] = True

  • options：修改动态下拉选项。

管理动态字段的提示​

处理动态字段时，请考虑以下最佳实践以确保流畅的用户体验：

  • 最小化字段更改：只隐藏真正不相关的字段，以避免混淆用户。
  • 测试行为：确保添加或删除字段不会意外删除用户输入。
  • 保存数据：使用 build_config["some_field"]["show"] = False 隐藏字段而不丢失其值。
  • 澄清逻辑：添加 info 注释来解释字段为什么会根据条件出现或消失。
  • 保持可管理性：如果动态逻辑变得太复杂，考虑将其分解为更小的组件，除非它在单个节点中有明确的目的。

错误处理和日志记录​

在 Langflow 中，强大的错误处理确保您的组件行为可预测，即使在发生意外情况时也是如此，例如无效输入、外部 API 故障或内部逻辑错误。

错误处理技术​

  • 抛出异常： 如果发生严重错误，您可以抛出标准的 Python 异常，如 ValueError，或专门的异常，如 ToolException。Langflow 将自动捕获这些异常并在 UI 中显示适当的错误消息，帮助用户快速识别问题所在。

    _10
    def compute_result(self) -> str:
    _10
    if not self.user_input:
    _10
    raise ValueError("No input provided.")
    _10
    # ...

  • 返回结构化错误数据： 您可以返回包含"error"字段的 Data 对象，而不是突然停止流程。这种方法允许流程继续运行，并使下游组件能够检测和优雅地处理错误。

    _10
    def run_model(self) -> Data:
    _10
    try:
    _10
    # ...
    _10
    except Exception as e:
    _10
    return Data(data={"error": str(e)})

改进调试和流程管理​

  • 使用 self.status： 每个组件都有一个状态字段，您可以在其中存储有关执行结果的简短消息——例如成功摘要、部分进度或错误通知。这些直接出现在 UI 中，使用户更容易进行故障排除。


    _10
    def parse_data(self) -> Data:
    _10
    # ...
    _10
    self.status = f"Parsed {len(rows)} rows successfully."
    _10
    return Data(data={"rows": rows})

  • 使用 self.stop(...) 停止特定输出： 当某些条件失败时，您可以停止个别输出路径，而不影响整个组件。这在处理具有多个输出分支的组件时特别有用。


    _10
    def some_output(self) -> Data:
    _10
    if <some condition>:
    _10
    self.stop("some_output") # 告诉 Langflow 没有数据流
    _10
    return Data(data={"error": "Condition not met"})

  • 记录事件： 您可以在组件内记录关键执行详细信息。日志显示在组件详细视图的"日志"或"事件"部分，稍后可以通过流程的调试面板或导出文件访问，为更容易的调试提供清晰的组件行为轨迹。


    _10
    def process_file(self, file_path: str):
    _10
    self.log(f"Processing file {file_path}")
    _10
    # ...

错误处理和日志记录提示​

要构建更可靠的组件，请考虑以下最佳实践：

  • 早期验证输入：在开始时捕获缺失或无效的输入，以防止逻辑错误。
  • 使用 self.status 总结：使用简短的成功或错误摘要帮助用户快速理解结果。
  • 保持日志简洁：专注于有意义的消息，避免使 UI 混乱。
  • 返回结构化错误：适当时，返回 Data(data={"error": ...}) 而不是抛出异常，以允许下游处理。
  • 有选择地停止输出：仅在必要时使用 self.stop(...) 停止特定输出，以保持其他地方的正确流程行为。

向 Langflow 贡献自定义组件​

请参阅如何贡献以向 Langflow 贡献您的自定义组件。

Search