跳到主要内容

创建自定义 Python 组件

创建您自己的自定义组件，以向 Langflow 添加您需要的任何功能，从 API 集成到数据处理。

在 Langflow 的基于节点的开发环境中，每个节点都是执行离散功能的“组件”。Langflow 中的自定义组件基于以下内容构建：

使用自定义组件快速入门将示例组件添加到 Langflow，然后使用随后的参考指南进行更高级的组件自定义。

自定义组件快速入门​

通过创建 Python 文件，将其保存在正确的文件夹中，包含 __init__.py 文件，并将其加载到 Langflow 中来创建自定义 DataFrameProcessor 组件。

创建 Python 文件​

  1. 为您的组件创建一个 Python 文件，例如 dataframe_processor.py。

  2. 将您的组件编写为 Component 类的对象。创建一个继承自 Component 的新类并覆盖基类的方法。

    向后兼容性

    在 Langflow 1.7 中，lfx 导入路径取代了 import from langflow.custom import Component，但原始输入仍然兼容并以相同方式工作。


    _10
    from typing import Any, Dict, Optional
    _10
    import pandas as pd
    _10
    from lfx.custom.custom_component.component import Component
    _10
    _10
    class DataFrameProcessor(Component)
    _10
    """一个使用各种操作处理 pandas DataFrame 的组件。"""

  3. 定义类属性以提供有关您的自定义组件的信息


    _13
    from typing import Any, Dict, Optional
    _13
    import pandas as pd
    _13
    from lfx.custom.custom_component.component import Component
    _13
    _13
    class DataFrameProcessor(Component)
    _13
    """一个使用各种操作处理 pandas DataFrame 的组件。"""
    _13
    _13
    display_name: str = "DataFrame 处理器"
    _13
    description: str = "使用各种操作（如过滤、排序和聚合）处理和转换 pandas DataFrame。"
    _13
    documentation: str = "https://docs.langflow.org.cn/components-dataframe-processor"
    _13
    icon: str = "DataframeIcon"
    _13
    priority: int = 100
    _13
    name: str = "dataframe_processor"

    • display_name：在可视化编辑器中显示的用户友好名称。
    • description：您的组件功能的简要描述。
    • documentation：指向详细文档的链接。
    • icon：用于视觉表示的表情符号或图标标识符。Langflow 使用 Lucide 图标。要为组件分配图标，请将 icon 属性设置为 Lucide 图标的名称字符串，例如 icon = "file-text"。Langflow 会自动从 Lucide 库渲染图标。有关更多信息，请参阅贡献包。
    • priority：控制显示顺序的可选整数。数字越小，越先显示。
    • name：可选的内部标识符，默认为类名。

  4. 通过指定组件的输入、输出以及处理它们的方法来定义组件的接口。方法名称必须与输出列表中的 method 字段匹配，因为这是 Langflow 知道调用哪个方法来生成每个输出的方式。

    此示例创建了一个最小的自定义组件骨架。


    _21
    from typing import Any, Dict, Optional
    _21
    import pandas as pd
    _21
    from lfx.custom.custom_component.component import Component
    _21
    _21
    class DataFrameProcessor(Component)
    _21
    """一个使用各种操作处理 pandas DataFrame 的组件。"""
    _21
    _21
    display_name: str = "DataFrame 处理器"
    _21
    description: str = "使用各种操作（如过滤、排序和聚合）处理和转换 pandas DataFrame。"
    _21
    documentation: str = "https://docs.langflow.org.cn/components-dataframe-processor"
    _21
    icon: str = "DataframeIcon"
    _21
    priority: int = 100
    _21
    name: str = "dataframe_processor"
    _21
    _21
    # 输入和输出列表
    _21
    inputs = []
    _21
    outputs = []
    _21
    _21
    # 方法
    _21
    def some_output_method(self)
    _21
    return ...

保存自定义组件​

将自定义组件保存在 Langflow 目录中，UI 将在此处发现并加载它。

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

将组件保存在默认目录中时，组件必须以特定的目录结构组织才能在可视化编辑器中正确加载和显示。

组件必须放在类别文件夹中，而不是直接放在基本目录中。

类别文件夹名称决定了组件在 Langflow 中的显示位置。 核心组件菜单。例如，要将示例 DataFrameProcessor 组件添加到数据类别，请将其放在 data 子文件夹中。


_10
src/lfx/src/lfx/components/
_10
└── data/ # 类别文件夹（决定菜单位置）
_10
├── __init__.py # 必需 - 使其成为 Python 包
_10
└── dataframe_processor.py # 您的自定义组件文件

如果您使用 LANGFLOW_COMPONENTS_PATH 环境变量在其他位置创建自定义组件，则组件必须以类似的目录结构组织才能在可视化编辑器中显示。


_10
/your/custom/components/path/ # 由 LANGFLOW_COMPONENTS_PATH 设置的基本目录
_10
└── category_name/
_10
├── __init__.py
_10
└── custom_component.py

您可以拥有多个类别文件夹来将组件组织到不同的类别中，每个文件夹中可以有多个组件。


_10
/app/custom_components/
_10
├── data/
_10
│ ├── __init__.py
_10
│ ├── custom_component.py
_10
│ └── dataframe_processor.py
_10
└── tools/
_10
├── __init__.py
_10
└── custom_tool.py

创建 __init__.py 文件​

每个类别目录必须包含一个 __init__.py 文件，以便 Langflow 正确识别和加载组件。这是 Python 包的要求，可确保目录被视为模块。

要包含 DataFrameProcessor 组件，请在组件目录中创建一个名为 __init__.py 的文件，其内容如下。


_10
from .dataframe_processor import DataFrameProcessor
_10
_10
__all__ = ["DataFrameProcessor"]

延迟加载 DataFrameProcessor 组件

或者，您可以延迟加载组件，这有利于性能，但稍微复杂一些。


_32
from __future__ import annotations
_32
_32
from typing import TYPE_CHECKING, Any
_32
_32
from lfx.components._importing import import_mod
_32
_32
if TYPE_CHECKING
_32
from lfx.components.data.dataframe_processor import DataFrameProcessor
_32
_32
_dynamic_imports = {
_32
"DataFrameProcessor": "dataframe_processor",
_32
}
_32
_32
__all__ = [
_32
"DataFrameProcessor",
_32
]
_32
_32
def __getattr__(attr_name: str) -> Any
_32
"""在属性访问时延迟导入数据组件。"""
_32
if attr_name not in _dynamic_imports
_32
msg = f"模块 '{__name__}' 没有属性 '{attr_name}'"
_32
raise AttributeError(msg)
_32
try
_32
result = import_mod(attr_name, _dynamic_imports[attr_name], __spec__.parent)
_32
except (ModuleNotFoundError, ImportError, AttributeError) as e
_32
msg = f"无法从 '{__name__}' 导入 '{attr_name}'：{e}"
_32
raise AttributeError(msg) from e
_32
globals()[attr_name] = result
_32
return result
_32
_32
def __dir__() -> list[str]
_32
return list(__all__)

有关延迟加载的其他示例，请参阅 FAISS 组件。

加载您的组件​

确保应用程序构建您的组件。

  1. 要重建后端和前端，请运行 make install_frontend && make build_frontend && make install_backend && uv run langflow run --port 7860。

  2. 刷新前端应用程序。您的新 DataFrameProcessor 组件在可视化编辑器中数据类别下的 核心组件菜单中可用。

Docker 部署​

在 Docker 中运行 Langflow 时，挂载您的自定义组件目录并在 docker run 命令中设置 LANGFLOW_COMPONENTS_PATH 环境变量以指向自定义组件目录。


_10
docker run -d \
_10
--name langflow \
_10
-p 7860:7860 \
_10
-v ./custom_components:/app/custom_components \
_10
-e LANGFLOW_COMPONENTS_PATH=/app/custom_components \
_10
langflowai/langflow:latest

创建与保存自定义组件示例中相同的自定义组件目录结构。


_10
/app/custom_components/ # LANGFLOW_COMPONENTS_PATH
_10
└── data/
_10
├── __init__.py
_10
└── dataframe_processor.py

组件如何执行​

Langflow 的引擎管理

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

您可以通过在自定义组件代码中覆盖这些可选钩子来自定义执行。

  • _pre_run_setup() - 在验证和设置期间使用。将此方法添加到您的组件类中，以在执行开始前初始化组件状态。


    _10
    class MyComponent(Component)
    _10
    # ... 您的输入、输出和其他属性 ...
    _10
    _10
    def _pre_run_setup(self)
    _10
    if not hasattr(self, "_initialized")
    _10
    self._initialized = True
    _10
    self.iteration = 0

  • 覆盖 run 或 _run - 在输出生成期间使用。将此方法添加到您的组件类中以自定义主执行逻辑。


    _10
    class MyComponent(Component)
    _10
    _10
    async def_run(self)
    _10
    # 自定义执行逻辑在此处
    _10
    # 这将代替默认输出方法调用运行
    _10
    pass

  • 将数据存储在 self.ctx 中。在您的任何组件方法中使用 self.ctx 在方法调用之间共享数据。


    _15
    class MyComponent(Component)
    _15
    _15
    def _pre_run_setup(self)
    _15
    # 在设置中初始化计数器
    _15
    self.ctx["processed_items"] = 0
    _15
    _15
    def process_data(self) -> Data
    _15
    # 在处理期间增加计数器
    _15
    self.ctx["processed_items"] += 1
    _15
    return Data(data={"item": f"processed {self.ctx['processed_items']}"})
    _15
    _15
    def get_summary(self) -> Data
    _15
    # 在不同方法中访问计数器
    _15
    total = self.ctx["processed_items"]
    _15
    return Data(data={"summary": f"Processed {total} items total"})

输入和输出​

输入和输出是类级配置，它们定义了数据如何流经组件，它如何在可视化编辑器中显示，以及如何验证与其他组件的连接。

输入​

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

输入通常是 lfx.io 中某个类的实例（例如 StrInput、DataInput 或 MessageTextInput）。

例如，此组件有三个输入：一个文本字段 (StrInput)、一个布尔开关 (BoolInput) 和一个下拉选择 (DropdownInput)。


_10
from lfx.io import StrInput, BoolInput, DropdownInput
_10
_10
inputs = [
_10
StrInput(name="title", display_name="标题"),
_10
BoolInput(name="enabled", display_name="已启用", value=True),
_10
DropdownInput(name="mode", display_name="模式", options=["快速", "安全", "实验性"], value="安全")
_10
]

StrInput 创建一个用于输入文本的单行文本字段。name="title" 参数表示您可以使用 self.title 在组件方法中访问此值，而 display_name="标题" 在可视化编辑器中显示“标题”作为标签。

BoolInput 创建一个布尔开关，默认情况下使用 value=True 启用。用户可以打开或关闭它，您可以使用 self.enabled 访问当前状态。

DropdownInput 提供一个包含三个预定义选项（“快速”、“安全”和“实验性”）的选择菜单。value="安全" 将“安全”设置为默认选择，您可以使用 self.mode 访问用户的选择。

有关所有可用参数的列表，请参阅 Langflow 代码库中的 BaseInputMixin 定义。

有关所有可用输入类型的列表，请参阅 Langflow 代码库中的输入类型定义。


_10
from lfx.io import StrInput, DataInput, MultilineInput, IntInput, BoolInput, DropdownInput, FileInput, CodeInput, ModelInput, HandleInput, Output

输出​

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

输出通常是 lfx.io 中 Output 的实例。

例如，此组件有一个返回 DataFrame 的 output。


_16
from lfx.io import Output
_16
from lfx.schema import DataFrame
_16
_16
outputs = [
_16
Output(
_16
name="df_out",
_16
display_name="DataFrame 输出",
_16
method="build_df"
_16
)
_16
]
_16
_16
def build_df(self) -> DataFrame
_16
# 处理数据并返回 DataFrame
_16
df = DataFrame({"col1": [1, 2], "col2": [3, 4]})
_16
self.status = f"已使用 {len(df)} 行构建 DataFrame。"
_16
return df

Output 在可视化编辑器中创建了一个名为DataFrame 输出的连接点。name="df_out" 参数标识此输出，而 display_name="DataFrame 输出" 在 UI 中显示标签。method="build_df" 参数告诉 Langflow 当此输出连接到另一个组件时调用 build_df 方法。

build_df 方法处理数据并返回一个 DataFrame。-> DataFrame 类型注解有助于 Langflow 验证连接并在可视化编辑器中提供颜色编码。您还可以设置 self.status 以在 UI 中显示进度消息。

有关所有可用参数的完整列表，请参阅 Langflow 代码库中的 Output 类定义。常见参数包括

附加返回类型

  • Message：结构化聊天消息
  • Data：具有 .data 和可选 .text 的灵活对象
  • DataFrame：表格数据 (pandas DataFrame 子类)
  • 基本类型：str、int、bool，不建议用于类型一致性

关联方法​

每个输出都链接到一个方法，其中输出方法名称必须与方法名称匹配。该方法通常返回 Message、Data 或 DataFrame 等对象，并且可以使用 self. 访问输入。

例如，Output 定义了一个名为 file_contents 的连接点，当连接时将调用 read_file 方法。read_file 方法使用 self.filename 访问文件名输入，读取文件内容，设置状态消息，并将内容包装在 Data 对象中返回。


_12
Output(
_12
name="file_contents",
_12
display_name="文件内容",
_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"已从 {path} 读取 {len(content)} 个字符"
_12
return Data(data={"content": content})

具有多个输出的组件​

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

例如：


_10
outputs = [
_10
Output(display_name="已处理数据", name="processed_data", method="process_data"),
_10
Output(display_name="调试信息", name="debug_info", method="provide_debug_info"),
_10
]

默认情况下，Langflow 中生成多个输出的组件在可视化编辑器中只允许选择一个输出。组件将只有一个输出端口，用户可以在其中选择首选的输出类型。

此行为由 group_outputs 参数控制

  • group_outputs=False (默认)：当组件有多个输出且 group_outputs 为 false 或未设置时，输出在可视化编辑器中分组，用户必须选择一个。

    当组件在流程中使用时只期望返回一种输出类型时，使用此选项。

  • group_outputs=True：所有输出在可视化编辑器中同时可用。组件为每个输出提供一个输出端口，用户可以将零个或多个输出连接到其他组件。

    当组件期望返回多个值并由下游组件或进程并行使用时，使用此选项。

在此示例中，可视化编辑器提供了一个输出端口，用户可以从中选择一个输出。由于 group_outputs=False 是默认行为，因此无需在组件中显式设置，如本例所示。


_12
outputs = [
_12
Output(
_12
name="structured_output",
_12
display_name="结构化输出",
_12
method="build_structured_output",
_12
),
_12
Output(
_12
name="dataframe_output",
_12
display_name="DataFrame 输出",
_12
method="build_structured_dataframe",
_12
),
_12
]

工具模式​

支持工具模式的组件可以作为独立组件使用（当不处于工具模式时）或作为具有工具输入的其他组件（如代理组件）的工具。

您可以通过设置 tool_mode=True 来允许自定义组件支持工具模式。


_10
inputs = [
_10
MessageTextInput(
_10
name="message",
_10
display_name="消息",
_10
info="输入将由工具直接处理的消息",
_10
tool_mode=True,
_10
),
_10
]

类型化注解​

在 Langflow 中，类型化注解允许 Langflow 在视觉上引导用户并保持流程一致性。始终使用 -> Data、-> Message 或 -> DataFrame 等返回类型注解您的输出方法，以启用正确的视觉编辑器颜色编码和验证。使用 Data、Message 或 DataFrame 包装器而不是返回普通结构，以获得更好的一致性。保持组件中的类型一致，以使流程可预测且更易于构建。

类型化注解提供颜色编码，其中 -> Data 或 -> Message 等输出获得不同的颜色，自动验证会阻止不兼容的连接，并提高了用户快速理解组件之间数据流的可读性。

常见返回类型​

用于聊天式输出。连接到多个兼容 Message 的输入之一。


_10
def produce_message(self) -> Message
_10
return Message(text="你好！来自类型化方法！", sender="系统")

启用动态字段​

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

在此示例中，运算符字段使用 real_time_refresh=True 触发更新。regex_pattern 字段最初是隐藏的，并由 dynamic=True 控制。


_23
from lfx.custom import Component
_23
from lfx.io import DropdownInput, StrInput
_23
_23
class RegexRouter(Component)
_23
display_name = "正则表达式路由器"
_23
description = "演示正则表达式输入的动态字段。"
_23
_23
inputs = [
_23
DropdownInput(
_23
name="operator",
_23
display_name="运算符",
_23
options=["等于", "包含", "正则表达式"],
_23
value="等于",
_23
real_time_refresh=True,
_23
),
_23
StrInput(
_23
name="regex_pattern",
_23
display_name="正则表达式模式",
_23
info="如果 operator='regex'，则使用",
_23
dynamic=True,
_23
show=False,
_23
),
_23
]

根据用户选择显示或隐藏字段​

当用户更改带有 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 中修改其他字段属性，而不仅仅是 show 和 hide。

  • required：动态使字段必填或可选


    _10
    if field_value == "regex"
    _10
    build_config["regex_pattern"]["required"] = True
    _10
    else
    _10
    build_config["regex_pattern"]["required"] = False

  • advanced：将字段移动到“高级”部分


    _10
    if field_value == "实验性"
    _10
    build_config["regex_pattern"]["advanced"] = False # 在主部分显示
    _10
    else
    _10
    build_config["regex_pattern"]["advanced"] = True # 在高级部分隐藏

  • options：根据其他选择更改下拉选项


    _10
    if field_value == "regex"
    _10
    build_config["operator"]["options"] = ["正则表达式", "包含", "以...开头"]
    _10
    else
    _10
    build_config["operator"]["options"] = ["等于", "包含", "不等于"]

错误处理和日志记录​

当验证失败时，您可以引发标准 Python 异常（例如 ValueError）或专门的异常（例如 ToolException）。Langflow 会自动捕获这些异常并在可视化编辑器中显示适当的错误消息，帮助用户快速识别出错的原因。


_10
def compute_result(self) -> str
_10
if not self.user_input
_10
raise ValueError("未提供输入。")
_10
# ...

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


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

Langflow 提供了几种工具来帮助您调试和管理组件执行。您可以使用 self.status 在可视化编辑器中直接显示有关执行结果的简短消息，从而使用户更容易进行故障排除。


_10
def parse_data(self) -> Data
_10
# ...
_10
self.status = f"成功解析了 {len(rows)} 行。"
_10
return Data(data={"rows": rows})

当某些条件失败时，您可以使用 self.stop() 停止单个输出路径，而无需停止同一组件的其他输出。

如果用户输入为空，此示例将停止输出，从而防止组件处理无效数据。


_10
def some_output(self) -> Data
_10
if not self.user_input or len(self.user_input.strip()) == 0
_10
self.stop("some_output")
_10
return Data(data={"error": "提供了空输入"})

您可以使用 self.log() 记录组件内部的关键执行详细信息。这些日志以结构化数据存储并显示在组件详细信息视图的“日志”或“事件”部分中，以后可以通过可视化编辑器中的日志按钮或导出的文件访问。

组件日志与 Langflow 的主应用程序日志系统不同。self.log() 创建组件特定的日志，这些日志显示在 UI 中，而 Langflow 的主日志系统使用 structlog 进行应用程序级日志记录，该日志记录输出到 langflow.log 文件。有关更多信息，请参阅日志。

当组件开始处理文件时，此示例会记录一条消息。


_10
def process_file(self, file_path: str)
_10
self.log(f"正在处理文件 {file_path}")

向 Langflow 贡献自定义组件​

要将您的自定义组件贡献给 Langflow 项目，请参阅贡献组件。

Search