跳到主要内容

创建自定义Python组件

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

在Langflow的基于节点的环境中，每个节点都是一个执行离散功能的"组件"。自定义组件是Python类，它们定义：

  • Inputs — 组件所需的数据或参数。
  • Outputs — 组件提供给下游节点的数据。
  • Logic — 如何处理输入以产生输出。

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

创建自定义组件用于执行专门任务、调用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"
_10
documentation = "http://docs.example.com/csv_reader"

  • display_name: 在可视化编辑器中显示的用户友好标签。

  • description: 在工具提示中显示的简要摘要，添加到流程中时打印在组件名称下方。

  • icon: 来自Langflow图标库的装饰性图标，显示在名称旁边。

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

  • name: 唯一的内部标识符，通常与包含您的组件代码的文件夹名称相同。

  • documentation: 可选的外部文档链接，如API或产品文档。

自定义组件的结构​

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的引擎管理：

  • Instantiation: 创建组件并初始化内部结构。
  • Assigning Inputs: 将来自可视化编辑器或连接的值分配给组件字段。
  • Validation and Setup: 可选的钩子如 _pre_run_setup。
  • Outputs Generation: 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中产生多个输出的组件在可视化编辑器中只允许选择一个输出。 组件将只有一个输出端口，用户可以在其中选择首选的输出类型。

此行为由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="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
]

常见内部模式​

_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 在 /components 目录中查找自定义组件。

如果您使用 LANGFLOW_COMPONENTS_PATH 环境变量 在其他位置创建自定义组件，则必须按照特定的目录结构组织组件，以便在可视化编辑器中正确加载和显示：

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


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

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

类别文件夹名称决定了组件在 Langflow Components 菜单中的显示位置。 例如，要将组件添加到 Helpers 类别，请将其放在 helpers 子文件夹中：


_10
/app/custom_components/ # LANGFLOW_COMPONENTS_PATH
_10
└── helpers/ # 在 "Helpers" 类别中显示
_10
├── __init__.py # 必需
_10
└── custom_component.py # 您的组件

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


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

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


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

自定义组件的输入和输出​

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

输入​

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

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

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

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

文本输入: 用于简单的文本输入。

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

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

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

下拉菜单: 用于从预定义选项中选择，适用于模式或级别设置。

  • DropdownInput

密钥: 用于敏感数据的专用输入，确保在可视化编辑器中输入内容被隐藏。

  • SecretStrInput 用于 API 密钥和密码。

专用数据输入: 在可视化编辑器中确保类型检查和彩色编码的连接。

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

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

  • HandleInput

文件上传: 允许用户通过可视化编辑器直接上传文件或从其他组件接收文件路径。

  • 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 渲染组件时，每个输出都会成为可视化编辑器中的一个连接点。当您将某些内容连接到输出时，Langflow 会自动调用相应的方法并将返回的对象传递给下一个组件。

输出通常是 langflow.io 中的 Output 实例，具有以下常见参数：

  • name: 内部变量名。
  • display_name: 可视化编辑器中显示的标签。
  • method: 用于生成输出的方法名称。
  • info (可选): 悬停时显示的帮助文本。

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

常见返回类型:

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

在此示例中，DataToDataFrame 组件使用 outputs 列表定义其输出。df_out 输出链接到 build_df 方法，因此当连接到另一个组件（节点）时，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 = "将多个 Data 对象转换为 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 来允许自定义组件支持 工具模式：


_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 目前支持以下 工具模式 的输入类型：

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

类型注解​

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

类型注解提供：

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

常见返回类型​

  • Message: 用于聊天式输出。连接到任何几种兼容 Message 的输入。


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

  • Data: 用于字典或部分文本等结构化数据。仅连接到 DataInput（接受 Data 的端口）。


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

  • DataFrame: 用于表格数据。仅连接到 DataFrameInput（接受 DataFrame 的端口）。


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

  • 基本类型 (str, int, bool): 允许返回基本类型，但为了在可视化编辑器中获得更好的一致性，建议包装在 Data 或 Message 中。


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

类型注解的技巧​

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

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

启用动态字段​

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

在此示例中，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 会自动捕获这些异常并在可视化编辑器中显示适当的错误消息，帮助用户快速识别问题所在。


    _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：每个组件都有一个状态字段，您可以在其中存储关于执行结果的简短消息，例如成功摘要、部分进度或错误通知。这些消息会直接显示在可视化编辑器中，使用户更容易进行故障排除。


    _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 <some condition>:
    _10
    self.stop("some_output") # 告诉 Langflow 没有数据流动
    _10
    return Data(data={"error": "条件未满足"})

  • 记录事件：您可以在组件内部记录关键的执行细节。日志会显示在组件详情视图的"日志"或"事件"部分，之后可以通过流程的调试面板或导出的文件访问，为组件的行为提供清晰的跟踪，便于调试。


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

错误处理和日志记录的最佳实践​

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

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

为 Langflow 贡献自定义组件​

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

Search