Reflex 企业版 React Flow 实战示例：动态连线建节点与自定义节点连接数限制

Reflex 企业版 React Flow 实战示例：动态连线建节点与自定义节点连接数限制

【免费下载链接】reflex🕸️ Web apps in pure Python 🐍项目地址: https://gitcode.com/GitHub_Trending/re/reflex

导读

examples.md 是 Reflex Enterprise 中rxe.flow（基于 React Flow 封装的流程图组件）的实战示例集，展示如何构建交互式流程图的两种典型能力：在连线拖放结束时自动创建新节点与边，以及为自定义节点的连接点（Handle）设置连接数量上限。读完本文，你将掌握on_connect_end与screen_to_flow_position的协作方式、rx.memo自定义节点与get_node_connections的用法，并能直接复用文中的两段可运行代码，为工作流编辑器、DAG 编排器等场景落地基础能力。

示例环境与核心概念

两段示例都遵循 Reflex Flow 的统一编程模型：

• 使用reflex_enterprise as rxe命名空间，流程图组件挂在rxe.flow之下，类型定义（Node、Edge、HandleType、Position、XYPosition、ConnectionInProgress、NoConnection等）从reflex_enterprise.components.flow.types导入；
• 节点与边以 Python 字典（类型别名Node/Edge）表示，存放在rx.State中；
• 画布渲染依赖rxe.flow组件，配套辅助组件包括rxe.flow.controls()（缩放/适配/锁定控制面板）、rxe.flow.mini_map()（缩略图）、rxe.flow.background()（背景网格）；
• 变更应用逻辑（rxe.flow.util.apply_node_changes/apply_edge_changes/add_edge）必须写在组件的事件回调 lambda 中，状态侧只保存结果列表——这一点在 interactivity.md 中有明确约定，示例中set_nodes/set_edges均为纯 setter。

关于流程图的基本概念（节点、Handle、边、连接线、视口）可参考 overview.md，节点与边的完整字段定义见 nodes.md 与 edges.md。

示例一：连线拖放到空白处自动生成新节点（Add Node on Edge Drop）

效果与适用场景

当用户从一个节点的连接点拖出连线、并松开在画布空白区域（连接未命中任何目标 Handle，即isValid为假）时，系统会在鼠标松开的位置自动创建一个新节点，同时补上一条从源节点指向新节点的边。这非常适合可视化编程、流程编排等需要"随手拉出节点"的交互场景。

状态定义

import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import ( ConnectionInProgress, Edge, NoConnection, Node, XYPosition, ) node_style = { "color": "#000000", } initial_nodes: list[Node] = [ { "id": "0", "type": "input", "data": {"label": "Node"}, "position": {"x": 0, "y": 50}, "style": node_style, }, ] class AddNodesOnEdgeDropState(rx.State): nodes: rx.Field[list[Node]] = rx.field(default_factory=lambda: initial_nodes) edges: rx.Field[list[Edge]] = rx.field(default_factory=list) node_id: int = 1 @rx.event def increment(self): self.node_id += 1 @rx.event def set_nodes(self, nodes: list[Node]): self.nodes = nodes @rx.event def set_edges(self, edges: list[Edge]): self.edges = edges @rx.event def handle_connect_end( self, connection_status: NoConnection | ConnectionInProgress, event: rx.event.PointerEventInfo, flow_position: XYPosition, ): if not connection_status["isValid"]: node_id = str(self.node_id) self.increment() self.nodes.append({ "id": node_id, "position": flow_position, "data": {"label": f"Node {node_id}"}, "origin": (0.5, 0.0), "style": node_style, }) self.edges.append({ "id": node_id, "source": connection_status["fromNode"]["id"], "target": node_id, "style": node_style, })

关键点拆解：

• nodes/edges使用rx.Field[list[Node]]+rx.field(default_factory=...)声明，保持与 Reflex 状态系统的响应式绑定；
• node_id作为自增计数器，保证每次新节点/新边的id唯一；
• handle_connect_end接收三个参数：连接状态对象（NoConnection | ConnectionInProgress联合类型）、指针事件信息rx.event.PointerEventInfo、以及由调用方换算好的流程坐标XYPosition。当connection_status["isValid"]为假（即连线未命中有效目标）时，读取connection_status["fromNode"]["id"]作为边的source；
• origin: (0.5, 0.0)表示以节点顶部中心为定位原点，与组件上node_origin=(0.5, 0.0)的设置保持一致，使新节点以鼠标松开点为中心生成。

画布与事件接线

def add_node_on_edge_drop(): return rx.box( rxe.flow.provider( rxe.flow( rxe.flow.controls(), rxe.flow.mini_map(), rxe.flow.background(), on_connect=lambda connection: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.add_edge(connection, AddNodesOnEdgeDropState.edges) ), on_connect_end=( lambda connection_status, event: ( AddNodesOnEdgeDropState.handle_connect_end( connection_status, event, rxe.flow.api.screen_to_flow_position( x=event.client_x, y=event.client_y, ), ) ) ), nodes=AddNodesOnEdgeDropState.nodes, edges=AddNodesOnEdgeDropState.edges, default_nodes=AddNodesOnEdgeDropState.nodes, default_edges=AddNodesOnEdgeDropState.edges, on_nodes_change=lambda changes: AddNodesOnEdgeDropState.set_nodes( rxe.flow.util.apply_node_changes( AddNodesOnEdgeDropState.nodes, changes ) ), on_edges_change=lambda changes: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.apply_edge_changes( AddNodesOnEdgeDropState.edges, changes ) ), fit_view=True, fit_view_options={"padding": 2}, node_origin=(0.5, 0.0), ) ), height="100vh", width="100vw", )

逐条说明：

1. 外层包裹rxe.flow.provider：FlowProvider 是上下文提供器，让rxe.flow.api下的 hooks（如screen_to_flow_position）可以在rxe.flow之外访问流程内部状态。provider 不接收任何事件处理器，所有事件都要挂在内部的rxe.flow上，详见 components.md。
2. on_connect：连线成功命中目标时触发，rxe.flow.util.add_edge(connection, edges)根据连接对象生成新边并追加到当前边列表（utils.md）。
3. on_connect_end：每次连线结束（无论成功与否）都会触发。这里把指针的屏幕像素坐标event.client_x/event.client_y交给rxe.flow.api.screen_to_flow_position换算成流程画布坐标，再连同连接状态一起传入状态事件。坐标换算发生在客户端，最终XYPosition以参数形式进入后端事件处理器（hooks.md 对此有专门说明，并指向本示例作为完整实现）。
4. 受控双绑：同时传入nodes/edges与default_nodes/default_edges，配合on_nodes_change/on_edges_change将画布拖拽、选中、删除等变更同步回状态——这正是 interactivity.md 中"受控流程"的标准接法：apply_node_changes/apply_edge_changes在客户端求值产生新列表，再由纯 setter 存回状态。
5. 视图适配：fit_view=True让流程自动缩放适配视口，fit_view_options={"padding": 2}控制留白；node_origin=(0.5, 0.0)统一节点定位原点。

示例二：自定义节点的连接数限制（Connection Limit on Custom Node）

效果与适用场景

该示例创建了一个名为custom的自定义节点，其左侧目标 Handle 通过is_connectable属性限制最多只允许 1 条连接——一旦已有连接，Handle 即不可再接受新连线。适用于端口数量受限的组件建模（如单一上游依赖、容量约束等场景）。

状态与节点布局

import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import Edge, HandleType, Node, Position node_style = { "color": "#000000", } class ConnectionLimitState(rx.State): nodes: rx.Field[list[Node]] = rx.field( default_factory=lambda: [ { "id": "1", "type": "input", "data": {"label": "Node 1"}, "position": {"x": 0, "y": 25}, "sourcePosition": "right", "style": node_style, }, { "id": "2", "type": "custom", "data": {}, "position": {"x": 250, "y": 50}, "style": node_style, }, { "id": "3", "type": "input", "data": {"label": "Node 2"}, "position": {"x": 0, "y": 100}, "sourcePosition": "right", "style": node_style, }, ] ) edges: rx.Field[list[Edge]] = rx.field(default_factory=list) @rx.event def set_nodes(self, nodes: list[Node]): self.nodes = nodes @rx.event def set_edges(self, edges: list[Edge]): self.edges = edges

三个节点中，"1"和"3"是input类型（仅含源 Handle，sourcePosition指向右侧），"2"是custom类型（data为空字典，渲染逻辑完全由自定义组件承担）。状态侧仍是标准的受控接线：set_nodes/set_edges两个纯 setter 配合apply_node_changes/apply_edge_changes使用。

带连接数限制的 Handle 组件

@rx.memo def custom_handle( type: rx.Var[HandleType], position: rx.Var[Position], connection_count: rx.Var[int] ) -> rxe.components.flow.Handle: connections = rxe.flow.api.get_node_connections() return rxe.flow.handle( type=type, position=position, connection_count=connection_count, is_connectable=connections.length() < connection_count.guess_type(), ) @rx.memo def custom_node() -> rx.el.Div: return rx.el.div( custom_handle(type="target", position="left", connection_count=1), rx.el.div("← Only one edge allowed"), class_name="border border-1 p-2 rounded-sm", border_color=rx.color_mode_cond("black", ""), color="black", bg="white", )

实现原理：

• @rx.memo装饰器：与 React.memo 语义对应，按声明式参数（type、position、connection_count）构建组件，参数类型用rx.Var[...]标注，是自定义节点的标准写法（可对比 nodes.md 中颜色选择器节点示例）；
• rxe.flow.api.get_node_connections()：返回当前节点上的连接数组（支持按节点、Handle 类型、Handle ID 过滤，见 hooks.md）。这里无参调用，取该节点全部连接；
• is_connectable动态判定：connections.length() < connection_count.guess_type()在客户端求值——连接数已满时返回假，Handle 自动禁用。connection_count作为rx.Var[int]传入，通过.guess_type()得到比较用的数值；调用custom_handle(type="target", position="left", connection_count=1)即把上限设为 1；
• rxe.flow.handle的connection_count属性：将上限值同步给底层 Handle 实现，与is_connectable配合完成"达到上限即不可再连"的限制逻辑。

渲染画布

def connection_limit(): return rx.box( rxe.flow( rxe.flow.background(), nodes=ConnectionLimitState.nodes, edges=ConnectionLimitState.edges, default_nodes=ConnectionLimitState.nodes, default_edges=ConnectionLimitState.edges, on_nodes_change=lambda changes: ConnectionLimitState.set_nodes( rxe.flow.util.apply_node_changes(ConnectionLimitState.nodes, changes) ), on_edges_change=lambda changes: ConnectionLimitState.set_edges( rxe.flow.util.apply_edge_changes(ConnectionLimitState.edges, changes) ), on_connect=lambda connection: ConnectionLimitState.set_edges( rxe.flow.util.add_edge(connection, ConnectionLimitState.edges) ), node_types={ "custom": rx.vars.function.ArgsFunctionOperation.create( (), custom_node() ) }, color_mode="light", fit_view=True, ), height="100vh", width="100vw", )

接线要点：

• node_types注册自定义节点：将"custom"类型映射到custom_node()渲染函数。此处通过rx.vars.function.ArgsFunctionOperation.create((), custom_node())将无参数组件封装为可序列化的函数操作对象，供前端按节点类型调用；
• on_connect成功连线时用add_edge追加边，新连线一旦到达custom节点，其目标 Handle 的连接数变为 1，is_connectable随即为假，后续拖线到该 Handle 将被拒绝；
• color_mode="light"固定浅色主题，fit_view=True自动适配视口。

两示例背后的通用模式

综合两个示例可以看出 Reflex Flow 交互式流程图的固定骨架：

1. 状态层：Node/Edge字典列表 + 自增 id 计数器 + 纯 setter 事件（set_nodes/set_edges）；
2. 变更应用：on_nodes_change/on_edges_change内用rxe.flow.util.apply_node_changes/apply_edge_changes在客户端求值，on_connect内用rxe.flow.util.add_edge生成新边——这些工具函数只能在组件代码中使用，严禁在@rx.event内部调用（interactivity.md 明确强调）；
3. 进阶交互：需要"画布级"操作（如按屏幕坐标建节点）时，用rxe.flow.api.screen_to_flow_position换算坐标，用rxe.flow.api.get_node_connections读取连接状态，这些 hooks 依赖外层rxe.flow.provider（hooks.md）；
4. 自定义渲染：@rx.memo+rx.Var参数构建节点/Handle/边组件，通过node_types/edge_types注册。

若需进一步扩展，可将示例一与示例二组合：在handle_connect_end中同时检查目标节点的剩余容量，实现"容量不足则拒绝、容量有余则落点建节点"的完整交互；边与路径的更多自定义（get_bezier_path、base_edge、edge_label_renderer）可参考 edges.md 中的按钮边示例。

【免费下载链接】reflex🕸️ Web apps in pure Python 🐍项目地址: https://gitcode.com/GitHub_Trending/re/reflex