跳到主要内容

创建自定义 Python 组件

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

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

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

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

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

Langflow 中的自定义组件基于以下内容构建：

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

类级别属性​

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


_10
class MyCsvReader(Component)
_10
display_name = "CSV Reader" # Shown in node header
_10
description = "Reads CSV files" # Tooltip text
_10
icon = "file-text" # Visual identifier
_10
name = "CSVReader" # Unique internal ID
_10
documentation = "http://docs.example.com/csv_reader" # Optional

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

自定义组件的结构​

一个 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
]

常见的内部模式​

_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/ # Base directory set by LANGFLOW_COMPONENTS_PATH
_10
└── category_name/ # Required category subfolder that determines menu name
_10
└── custom_component.py # Component file

组件必须放在类别文件夹内，而不是直接放在基本目录中。类别文件夹名称决定了组件在 UI 菜单中的位置。

例如，要将组件添加到助手菜单，请将其放在 helpers 子文件夹中：


_10
/app/custom_components/ # LANGFLOW_COMPONENTS_PATH
_10
└── helpers/ # Displayed within the "Helpers" menu
_10
└── custom_component.py # Your component

您可以有多个类别文件夹，将组件组织到不同的菜单中：


_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 # Won't be loaded - missing category folder!

自定义组件输入和输出​

输入和输出定义了数据如何通过组件流动、组件在 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（如果您需要类型/颜色一致性，则不推荐）。


_37
from langflow.custom import Component
_37
In this example, the DataToDataFrame component defines its output using the outputs list. The df_out output is linked to the build_df method, so when connected in the UI, Langflow calls this method and passes its returned DataFrame to the next node. This demonstrates how each output maps to a method that generates the actual output data.
_37
from langflow.io import DataInput, Output
_37
_37
from langflow.schema import Data, DataFrame
_37
class DataToDataFrame(Component)
_37
display_name = "Data to DataFrame"
_37
description = "Convert multiple Data objects into a DataFrame"
_37
icon = "table"
_37
_37
inputs = [
_37
name = "DataToDataFrame"
_37
DataInput(
_37
name="items",
_37
display_name="Data Items",
_37
info="List of Data objects to convert",
_37
)
_37
]
_37
_37
outputs = [
_37
Output(
_37
is_list=True
_37
name="df_out",
_37
display_name="DataFrame Output",
_37
)
_37
]
_37
_37
method="build_df"
_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
_37
rows.append(row_dict)
_37
df = DataFrame(rows)
_37
self.status = f"Built DataFrame with {len(rows)} rows."

工具模式​

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

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 兼容的输入。

数据

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


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

在 UI 中，仅连接到 DataInput。

DataFrame

用于表格数据


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

在 UI 中，仅连接到 DataFrameInput。

基本类型（str, int, bool）

允许返回基本类型，但为了更好的 UI 一致性，建议封装在 Data 或 Message 中。


_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。

在此示例中，操作符字段通过 `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: Set build_config["some_field"]["required"] = True/False

  • advanced: Set 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") # Tells Langflow no data flows
    _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