在 LangFlow 中，**节点（Node）是构成工作流的核心基本单元**
在 LangFlow 中，节点（Node）是构成工作流的核心基本单元，每个节点封装了特定功能（如模型调用、数据处理、条件判断等），并通过输入/输出端口（Port）与其他节点连接，最终形成可执行的 AI 工作流。理解节点的结构是自定义节点、调试工作流的关键，其核心结构可分为 元数据（Metadata）、配置项（Config）、输入/输出端口（Ports）、执行逻辑（Run Logic） 四大核心部分，以下是详细拆解：
一、节点的核心结构总览
LangFlow 的节点本质是一个可序列化、可配置、可执行的 Python 类，基于 LangFlow 提供的基类（如 BaseNode）封装，其结构可抽象为：
class CustomNode(BaseNode):# 1. 元数据：节点的“身份信息”metadata: NodeMetadata = NodeMetadata(...)# 2. 配置项：节点的可自定义参数（暴露给前端界面）config: Dict[str, ConfigField] = {"param1": ConfigField(...),"param2": ConfigField(...)}# 3. 输入端口：接收其他节点的输出数据inputs: List[Port] = [InputPort(name="input1", type="str", ...),InputPort(name="input2", type="list", ...)]# 4. 输出端口：向其他节点传递处理后的数据outputs: List[Port] = [OutputPort(name="output1", type="dict", ...),OutputPort(name="output2", type="str", ...)]# 5. 执行逻辑：节点的核心功能实现def run(self, inputs: Dict[str, Any]) -> Dict[str, Any]:# 处理输入数据（结合 config 配置）result1 = self.process_data(inputs["input1"], self.config["param1"].value)result2 = self.format_result(result1)# 返回输出数据（与 outputs 端口对应）return {"output1": result1, "output2": result2}
下面逐个拆解每个核心部分的作用、格式和约束：
二、1. 元数据（Metadata）：节点的“身份卡”
元数据用于定义节点的基础信息，供 LangFlow 前端识别、分类和展示，核心字段如下（通过 NodeMetadata 类定义）：
| 字段名 | 类型 | 作用说明 |
|---|---|---|
name | str | 节点唯一标识（英文驼峰/下划线，如 CustomLLMCall），不可重复 |
display_name | str | 前端显示的节点名称（中文/英文，如“自定义 LLM 调用”） |
description | str | 节点功能描述（如“调用自定义 API 实现 LLM 对话生成”） |
category | str | 节点分类（如 LLM、Tools、Data Processing、Condition），决定前端菜单位置 |
icon | str（可选） | 前端显示的图标（使用 Font Awesome 图标名，如 fa-brain） |
documentation | str（可选） | 节点文档链接（如 GitHub 文档 URL） |
version | str（可选） | 节点版本（如 1.0.0） |
示例：
from langflow.base.node import NodeMetadatametadata = NodeMetadata(name="WeatherQuery",display_name="天气查询",description="调用天气 API 查询指定城市的实时天气",category="Tools",icon="fa-cloud-sun",version="1.0.0"
)
三、2. 配置项（Config）：节点的“自定义参数”
配置项是节点的可配置参数（暴露在 LangFlow 前端右侧面板），用户可通过界面调整参数值（如 API 密钥、模型温度、超时时间等）。
核心定义：ConfigField 类
每个配置项通过 ConfigField 封装，关键参数如下：
| 参数名 | 类型 | 作用说明 |
|---|---|---|
value | Any | 配置项默认值（如 0.7、"sk-xxx"、30） |
display_name | str | 前端显示的参数名称（如“温度”、“API 密钥”） |
description | str（可选） | 参数描述（如“控制模型输出的随机性，值越大越随机”） |
type | str | 参数类型（约束输入格式，如 str、int、float、bool、secret） |
required | bool | 是否必填（默认 False，必填项前端会标红） |
options | List[Any]（可选） | 下拉选择框选项（如 ["gpt-3.5-turbo", "gpt-4"]） |
placeholder | str（可选） | 输入框提示文本（如“请输入 API 密钥”） |
配置项字典格式
config 是一个字典，key 为参数唯一标识（Python 变量名格式），value 为 ConfigField 实例：
示例：
from langflow.base.node import ConfigFieldconfig = {"api_key": ConfigField(value="",display_name="API 密钥",description="天气 API 的访问密钥",type="secret", # 密码类型，输入时隐藏required=True),"city": ConfigField(value="北京",display_name="查询城市",type="str",placeholder="请输入城市名称（如上海、广州）"),"timeout": ConfigField(value=30,display_name="超时时间",description="API 请求超时时间（秒）",type="int",options=[10, 20, 30, 60] # 下拉选择),"include_forecast": ConfigField(value=False,display_name="包含未来3天预报",type="bool" # 复选框)
}
四、3. 输入/输出端口（Ports）：节点的“数据接口”
端口是节点与其他节点进行数据通信的通道，输入端口（InputPort）接收上游节点的输出，输出端口（OutputPort）向下游节点传递数据。端口的核心约束是 类型匹配（下游输入端口类型需与上游输出端口类型一致，否则无法连接）。
端口基类 Port 的核心字段
InputPort 和 OutputPort 均继承自 Port，核心参数如下：
| 参数名 | 类型 | 作用说明 |
|---|---|---|
name | str | 端口唯一标识（如 input_text、weather_data） |
display_name | str | 前端显示的端口名称（如“输入文本”、“天气结果”） |
type | str | 数据类型（如 str、dict、list、LLMResponse、ToolResult） |
description | str（可选） | 端口描述（如“接收用户查询的文本”） |
is_required | bool | 是否必填端口（输入端口专用，默认 False，必填端口未连接时工作流无法运行） |
1. 输入端口（InputPort）
定义节点需要接收的上游数据，放在 inputs 列表中。
示例：
from langflow.base.node import InputPortinputs = [InputPort(name="user_query",display_name="用户查询",type="str",description="用户输入的天气查询文本（如“北京今天天气怎么样”）",is_required=True),InputPort(name="extra_params",display_name="额外参数",type="dict",description="可选的额外查询参数（如{'lang': 'zh'}）")
]
2. 输出端口（OutputPort）
定义节点处理后输出的数据，放在 outputs 列表中，返回值需与端口名一一对应（通过 run 方法的返回字典指定）。
示例：
from langflow.base.node import OutputPortoutputs = [OutputPort(name="weather_result",display_name="天气结果",type="dict",description="包含温度、湿度、风速的天气数据"),OutputPort(name="error_msg",display_name="错误信息",type="str",description="请求失败时的错误提示（成功时为None）")
]
端口类型匹配规则
- 基础类型（
str、int、float、bool、dict、list）：严格匹配（如int端口不能接收str类型）。 - 自定义类型（如
LLMResponse、ToolResult）：需继承自 LangFlow 的基础数据类型，或通过type字符串自定义（需确保上下游节点约定一致）。 - 特殊类型
Any：可接收任意类型数据（不推荐，易导致数据格式错误）。
五、4. 执行逻辑（Run Logic）：节点的“核心功能”
run 方法是节点的核心执行入口，接收上游节点传递的输入数据（通过 inputs 参数），结合节点配置项（self.config），执行业务逻辑后返回输出数据（与 outputs 端口对应）。
关键说明：
- 输入数据获取：
inputs字典的key与输入端口的name一致，直接通过inputs["端口名"]获取数据（如inputs["user_query"]）。 - 配置项获取：通过
self.config["配置项key"].value获取用户在前端设置的参数值（如self.config["api_key"].value）。 - 输出数据返回：返回字典的
key必须与输出端口的name完全一致，否则下游节点无法获取对应数据。 - 异常处理：建议在
run方法中捕获异常（如 API 请求失败、数据格式错误），并通过输出端口返回错误信息（如error_msg）。
示例：run 方法实现
import requestsdef run(self, inputs: Dict[str, Any]) -> Dict[str, Any]:# 1. 获取输入数据和配置项user_query = inputs["user_query"]extra_params = inputs.get("extra_params", {}) # 可选输入，默认空字典api_key = self.config["api_key"].valuecity = self.config["city"].valuetimeout = self.config["timeout"].valueinclude_forecast = self.config["include_forecast"].value# 2. 执行核心逻辑（调用天气 API）try:url = f"https://api.weather.com/v3/weather/now?city={city}&key={api_key}&include_forecast={include_forecast}"response = requests.get(url, timeout=timeout, params=extra_params)response.raise_for_status() # 抛出 HTTP 错误weather_data = response.json()# 3. 构造输出数据return {"weather_result": weather_data,"error_msg": None}except Exception as e:# 4. 异常处理：返回错误信息return {"weather_result": None,"error_msg": f"请求失败：{str(e)}"}
六、节点的额外特性（可选）
1. 动态端口（Dynamic Ports）
支持根据配置项动态添加/删除端口（如“多输入文本”节点，用户可通过配置项增加输入端口），需重写 get_dynamic_inputs 或 get_dynamic_outputs 方法。
2. 验证逻辑（Validation）
重写 validate 方法，对输入数据或配置项进行校验（如 API 密钥格式、城市名称合法性），校验失败时抛出 ValidationError，前端会显示错误提示。
3. 序列化与反序列化
LangFlow 自动将节点配置、连接关系序列化为 JSON 格式（工作流文件 .langflow），无需手动实现，核心依赖元数据、配置项、端口的标准化定义。
七、总结：节点结构的核心逻辑
LangFlow 节点的设计遵循 “配置化定义 + 模块化执行” 思想：
- 通过 元数据 定义节点的“身份”和展示方式；
- 通过 配置项 暴露用户可调整的参数；
- 通过 端口 定义数据输入/输出的接口和类型约束；
- 通过 run 方法 封装核心业务逻辑，实现“输入 → 处理 → 输出”的闭环。
无论是使用内置节点还是自定义节点，理解这四大核心部分，就能快速上手 LangFlow 工作流的搭建、调试和扩展。