跳到主要内容

创建自定义 Python 组件

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

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

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

自定义组件快速入门​

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

创建 Python 文件​

  1. Create a Python file for your component, such as dataframe_processor.py.

  2. Write your component as an object of the Component class. Create a new class that inherits from Component and override the base class's methods.

    Backwards compatibility

    The lfx import path replaced the import from langflow.custom import Component in Langflow 1.7, but the original input is still compatible and works the same way.


    _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
    """A component that processes pandas DataFrames with various operations."""

  3. Define class attributes to provide information about your custom component:


    _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
    """A component that processes pandas DataFrames with various operations."""
    _13
    _13
    display_name: str = "DataFrame Processor"
    _13
    description: str = "Process and transform pandas DataFrames with various operations like filtering, sorting, and aggregation."
    _13
    documentation: str = "https://docs.langflow.org/components-dataframe-processor"
    _13
    icon: str = "DataframeIcon"
    _13
    priority: int = 100
    _13
    name: str = "dataframe_processor"

    • display_name: A user-friendly name shown in the visual editor.
    • description: A brief description of what your component does.
    • documentation: A link to detailed documentation.
    • icon: An emoji or icon identifier for visual representation. Langflow uses Lucide for icons. To assign an icon to your component, set the icon attribute to the name of a Lucide icon as a string, such as icon = "file-text". Langflow renders icons from the Lucide library automatically. For more information, see Contributing bundles.
    • priority: An optional integer to control display order. Lower numbers appear first.
    • name: An optional internal identifier that defaults to class name.

  4. Define the component's interface by specifying its inputs, outputs, and the method that will process them. The method name must match the method field in your outputs list, as this is how Langflow knows which method to call to generate each output.

    This example creates a minimal custom component skeleton.


    _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
    """A component that processes pandas DataFrames with various operations."""
    _21
    _21
    display_name: str = "DataFrame Processor"
    _21
    description: str = "Process and transform pandas DataFrames with various operations like filtering, sorting, and aggregation."
    _21
    documentation: str = "https://docs.langflow.org/components-dataframe-processor"
    _21
    icon: str = "DataframeIcon"
    _21
    priority: int = 100
    _21
    name: str = "dataframe_processor"
    _21
    _21
    # input and output lists
    _21
    inputs = []
    _21
    outputs = []
    _21
    _21
    # method
    _21
    def some_output_method(self):
    _21
    return ...

保存自定义组件​

将自定义组件保存在 Langflow 目录中，以便 UI 能够发现并加载它。

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

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

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

类别文件夹名称决定了组件在 Langflow 核心组件 (Core components) 菜单中出现的位置。 例如，要将示例 DataFrameProcessor 组件添加到 Data 类别中，请将其放在 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
│ └── 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 组件

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


_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"module '{__name__}' has no attribute '{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 组件现在可以在可视化编辑器的 核心组件 (Core components) 菜单下的 Data 类别中找到。

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. 实例化 (Instantiation)：创建组件并初始化内部结构。
  2. 分配输入 (Assigning Inputs)：将来自可视化编辑器或连接的值分配给组件字段。
  3. 验证与设置 (Validation and Setup)：可选钩子，如 _pre_run_setup。
  4. 生成输出 (Outputs Generation)：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"已处理 {self.ctx['processed_items']}"})
    _15
    _15
    def get_summary(self) -> Data:
    _15
    # 在不同方法中访问计数器
    _15
    total = self.ctx["processed_items"]
    _15
    return Data(data={"summary": f"总共处理了 {total} 个项目"})

输入与输出​

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

输入 (Inputs)​

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

输入通常是来自 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)​

输出定义在类级的 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 子类)
  • 原始类型 (Primitive types)：str、int、bool（不推荐用于保持类型一致性）

关联方法​

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

例如，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)​

支持 工具模式 (Tool Mode) 的组件可以用作独立组件（不 处于 工具模式 时），也可以用作具有 Tools 输入的其他组件（如 代理 (Agent) 组件）的工具。

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


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

类型注解 (Typed annotations)​

在 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 方法实时调整输入的可见性或属性，从而创建一种上下文感知的可视化编辑器体验，仅根据用户的选择暴露相关字段。

在此示例中，operator 字段通过 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 = "正则路由 (Regex Router)"
_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="如果运算符='正则表达式'则使用",
_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 == "正则表达式":
_10
build_config["regex_pattern"]["show"] = True
_10
else:
_10
build_config["regex_pattern"]["show"] = False
_10
return build_config

除了 show 和 hide 之外，您还可以在 update_build_config 中修改其他字段属性。

  • required：动态设置字段为必填或可选


    _10
    if field_value == "正则表达式":
    _10
    build_config["regex_pattern"]["required"] = True
    _10
    else:
    _10
    build_config["regex_pattern"]["required"] = False

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


    _10
    if field_value == "experimental":
    _10
    build_config["regex_pattern"]["advanced"] = True
    _10
    return build_config


_10

Search