数据截至 (上游 commit 7a975c596eca)
Langflow 可视化工作流构建器 — 架构与原理
30 秒导读： Langflow 是一个把 AI 工作流画出来的工具——你在浏览器画布上拖节点、连线，后端把这张图变成一串 Python 组件实例并按依赖顺序跑完，跑的过程实时回推到画布上。它最核心的一件事是**「一个 Python 类 = 画布上一个节点」的双向反射**：类里的
inputs/outputs声明被反射成 UI 用的 template JSON，画布存回来的 JSON 又被反射回组件实例。
1. 这是什么（零基础也能懂）
- 一句话定义： 一个可视化的 AI 工作流构建器——画布上画流程，后端把流程当程序执行，并且这条流程天生就是一个 HTTP API 和一个 MCP 工具。
1.1 解决谁的什么问题
假设你要做一个「用户提问 → 查向量库 → 拼 prompt → 调 LLM → 回答」的链路。
不用 Langflow，你要自己写胶水代码、自己管并发和顺序、自己做前端调试界面、自己再包一层 API 给别人调。
Langflow 把这四件事一起接管了：编排、执行、调试界面、对 外出口。
1.2 它能做什么
- 画布上拖拽组件、连线，组成一条 flow（一张有向图，允许有环）。
- 每个组件就是一个 Python 类，UI 上可以直接看到并改它的源码（
code字段随流程一起存）。 - 内置 300+ 组件，覆盖各家 LLM、向量库、文档加载器、工具、流程控制（依据：
src/lfx/src/lfx/_assets/component_index.json的metadata记录num_components: 354、num_modules: 95）。 - 边跑边看：每个节点跑完立刻把结果推回画布，token 逐个流式吐出。
- 跑完的 flow 可以直接被
POST /api/v1/run/{flow_id}调用，或作为 MCP 工具暴露给别的 agent。
1.3 用起来什么样
最小的一次真实调用——把一个流程 JSON 直接在命令行跑起来（依据：src/lfx/README.md:480）：
uv run lfx run simple-agent-flow.json "Hello world"
同一条流程在 Python 里可以完全不碰画布地搭出来（依据：src/lfx/tests/unit/graph/graph/test_cycles.py:37-55）：
# 示意，非源码：真实测试里的搭图写法
chat_input = ChatInput(_id="chat_input")
chat_output = ChatOutput(_id="chat_output")
chat_output.set(input_value=chat_input.message_response) # 传的是「方法」，不是值
graph = Graph(chat_input, chat_output) # 两端一给，图就建好了
重点看第 3 行：chat_input.message_response 是一个方法对象，不是调用结果。Langflow 用「你把哪个 output 方法交给了哪个 input」来推断出一条边——这跟你在画布上从右边小圆点拉一条线到左边小圆点，是完全等价的两种输入方式（依据：src/lfx/src/lfx/custom/custom_component/component.py:977-1003 的 _connect_to_component / _add_edge）。
1.4 一句话直觉
把它想成 「Excel 的公式依赖图」搬到了 AI 上：每个格子（节点）声明自己依赖哪 些格子，引擎算出谁先算、谁能同时算，改一个格子就沿着依赖往下重算——只不过这里的「格子」是一次 LLM 调用或一次向量检索，而且允许格子之间成环。
1.5 读之前先知道的边界
| 它不做什么 | 说明 |
|---|---|
| 不是纯 DAG 引擎 | 图里可以有环，所以不能无脑拓扑排序；环必须靠 max_iterations 兜底（依据：src/lfx/src/lfx/graph/graph/base.py:566 的 if self.is_cyclic and max_iterations is None 硬性要求） |
| 不是无状态纯函数编排 | 组件可以往 graph.context 写共享状态，Loop 的迭代计数就存在那里（依据：src/lfx/src/lfx/components/flow_controls/loop.py:68-74） |
| 不做沙箱隔离 | 节点里的 code 字段就是会被执行的 Python 源码，所以要在建图前做设置校验（依据：src/lfx/src/lfx/graph/graph/base.py:1812 调用、定义在 src/lfx/src/lfx/utils/flow_validation.py:892 的 validate_flow_for_current_settings） |
2. 顶层全景（它大概怎么转）
2.1 一个组件的四种形态
怎么读这张图： 顺时针走一圈。上排是「设计时」——类怎么变成画布上能拖的东西；下排是「运行时」——存下来的 JSON 怎么变回能跑的 Python 对象。同一个组件在四个格子里换了四层皮。
设 计 时 ─────────────────────────────────────────────►
┌────────────────┐ 反射 ┌──────────────┐ GET /all ┌──────────────┐
│ ① Python 组件类 │ ────────► │ ② template │ ─────────► │ ③ 画布节点 │
│ inputs/outputs │ │ JSON │ │ 拖拽 + 连线 │
│ + 方法 │ │ (含源码) │ │ │
└────────────────┘ └──────────────┘ └──────┬───────┘
│ 保存
运 行 时 ◄───────────────────────────────────────────── ▼
┌────────────────┐ 实例化 ┌──────────────┐ 解析 ┌──────────────┐
│ ⑥ 组件实例 │ ◄──────── │ ⑤ Vertex │ ◄───────── │ ④ flow JSON │
│ 跑 output 方法 │ │ 参数已填好 │ │ nodes/edges │
└────────────────┘ └──────────────┘ └──────────────┘
2.2 部件一句话职责
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
Component | 组件基类：声明 inputs/outputs，跑输出方法，也是搭图 DSL 的入口 | src/lfx/src/lfx/custom/custom_component/component.py |
| 模板反射层 | 把类的属性抠出来变成前端 template JSON | src/lfx/src/lfx/custom/utils.py、src/lfx/src/lfx/custom/attributes.py |
Graph | 装 Vertex/Edge，排层、剪枝、发起并发、缓存快照 | src/lfx/src/lfx/graph/graph/base.py |
Vertex | 一个节点的运行时壳：填参数、实例化组件、收结果 | src/lfx/src/lfx/graph/vertex/base.py |
RunnableVerticesManager | 记「谁还欠谁」，判断某个节点现在能不能跑 | src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py |
EventManager | 把每一步进展塞进异步队列，最终变成 SSE 事件 | src/lfx/src/lfx/events/event_manager.py |
| 构建端点 | POST /build/{flow_id}/flow + GET /build/{job_id}/events | src/backend/base/langflow/api/v1/chat.py |
2.3 主线走一遍（点一下 Run 之后）
① 点 Run ─► ② 建图 ─► ③ 排层 ─► ④ 取第一层 ─► ⑤ 并发 build ─► ⑥ 算下一批 ─► ⑦ SSE 推回画布
▲ │
└───────────────────────────────┘
队列没空就继续下一轮
分步说明：
- 点 Run — 前端
POST /api/v1/build/{flow_id}/flow，立刻拿到一个job_id（依据：src/backend/base/langflow/api/v1/chat.py:256、:367）。 - 建图 —
Graph.from_payload吃下{nodes, edges}，做迁移、做安全校验，造出 Vertex 和 Edge（依据：src/lfx/src/lfx/graph/graph/base.py:1468）。 - 排层 —
prepare()→sort_vertices()算出第一层和后续层，把第一层灌进运行队列（依据：src/lfx/src/lfx/graph/graph/base.py:2664、:2370）。 - 取第一层 — 队列里每个 vertex_id 起一个 asyncio 任务。
- 并发 build —
build_vertex实例化组件、跑输出方法、收结果（依据：src/lfx/src/lfx/graph/graph/base.py:2049）。 - 算下一批 — 每个节点跑完，问
RunnableVerticesManager「它的后继现在欠债还清了吗」，还清的进下一批（依据：src/lfx/src/lfx/graph/graph/base.py:2268）。 - SSE 推回 — 每个节点的结果连同「下一批是谁」「哪些被剪掉了」一起发成事件，前端据此点亮节点（依据：
src/backend/base/langflow/api/build.py:867、src/frontend/src/utils/buildUtils.ts:686）。
关键认知：这里没有全局拓扑序在指挥。 调度是事件驱动的滚雪球：跑完谁，才知道接下来能跑谁。正因如此，图里有环也能转起来。
3. 阅读地图
六章按「静态 → 动态」排，建议顺序读；只想解决某个具体问题就直接跳。
| 顺序 | 章节 | 什么时候读它 |
|---|---|---|
| 1 | 组件模型 — 一个 Python 类如何长成画布上的节点 | 想写自定义组件，或想搞懂 template JSON 里每个字段从哪来 |
| 2 | 从 JSON 到图 — 画布存的那坨数据怎么变成可执行对象 | 想读懂 flow JSON 的结构，或调试「连了线但参数没传过去」 |
| 3 | 调度引擎 — 谁先跑、谁能并行、结果怎么往下传 | 关心执行顺序、并发度、缓存与 frozen 节点 |
| 4 | 环、循环与条件路由 — 这张图为什么不是 DAG | 用到 If-Else / Loop / Agent 自循环，或撞上「分支两边都跑了」 |
| 5 | 运行时与事件流 — 点一下 Run 之后前后端之间发生了什么 | 排查流式输出、SSE 断流、构建卡住、取消构建 |
| 6 | 组件生态与对外出口 — 上百个集成怎么不拖垮启动，一条流怎么变成别人能调的工具 | 关心启动性能、懒加载，或要把 flow 暴露成 API / MCP 工具 |
先读哪一章的快速判断：
- 你要改组件 → 第 1 章。
- 你要改执行行为 → 第 3、4 章。
- 你要接前端或排查线上问题 → 第 5 章。
- 你要做部署与集成 → 第 6 章。
4. 巧妙之处（值得抄走的技术）
4.1 「声明即 UI」：类属性直接反射成前端表单
组件类里写的 display_name、icon、inputs、outputs 这些属性，被一张固定的映射表逐个抠出来、按类型清洗，拼成 template JSON。
妙在哪： 前端不需要为每个组件写一份表单定义，也没有第二份 schema 要同步— —UI 的形状只有一个真相来源，就是 Python 类本身。
依据：src/lfx/src/lfx/custom/attributes.py:77（ATTR_FUNC_MAPPING 列出全部会被反射的属性）、src/lfx/src/lfx/custom/custom_component/base_component.py:89（get_template_config）、src/lfx/src/lfx/custom/utils.py:464（build_custom_component_template_from_inputs）。
4.2 输出的类型是从方法返回注解里读出来的
Output(name="message", method="message_response") 只写了方法名，没写类型。类型是运行时 get_type_hints(method) 拿返回注解算出来的。
妙在哪： 端口能不能连（类型兼容性检查）这件事，直接由 Python 的类型注解兜底，写组件的人不必再手抄一遍类型。
依据：src/lfx/src/lfx/custom/custom_component/component.py:1119（_get_method_return_type）、:1163-1167（to_frontend_node 里回填 output.types）。
4.3 只算「连出去了的」输出，没连的方法根本不跑
一个组件可以有多个 output 方法。运行时并不是全跑一遍，而是先看这个节点在画布上从哪几个端口连了线出去，只跑对应的方法。
妙在哪： 一个「向量库」组件同时提供 search_documents 和 as_dataframe 两个出口，你只连了前者，后者的开销就是零。画布的连线在这里当了惰性求值的开关。
依据：src/lfx/src/lfx/custom/custom_component/component.py:1348（_should_process_output，看 self._vertex.edges_source_names）、:1294（_build_results 只遍历 _get_outputs_to_process()）。
4.4 两套剪枝机制并存：一套会重置，一套不会
If-Else 这类路由组件要「只让一条分支跑」，但它可能身处一个环里、每轮条件还不一样。Langflow 为此并行维护了两套标记：
| 机制 | 作用 | 生命周期 |
|---|---|---|
ACTIVE / INACTIVE | 环内单轮的通断 | 每轮迭代后被重置 |
conditionally_excluded_vertices | 条件路由的持久剪枝 | 只有同一个源节点再次决策时才会被清掉重算 |
妙在哪： 如果只有前者，环转到下一圈时剪枝就失效、两条分支都会跑；如果只有后者，环就永远没法翻篇。两套叠加才同时满足「环能继续转」和「分支只走一条」。
另有一 个细节很值得学：剪枝时会先算出「从这个节点的其他输出端口仍然可达的下游」，把它们保护起来不剪——这样 If-Else 两条分支下游的那个汇合节点不会被误伤。
依据：src/lfx/src/lfx/graph/graph/base.py:1318（exclude_branch_conditionally 的 docstring 明说这两套的区别）、:1015（_get_vertices_reachable_from_other_outputs）、src/lfx/src/lfx/components/flow_controls/conditional_router.py:131（iterate_and_stop_once 同时调 stop() 和 exclude_branch_conditionally()）。
4.5 环里的「第一次执行」被特批放行
正常规则是「所有前驱都跑完了才能跑」。但环里的节点永远有一个前驱还没跑（就是环本身），死锁。
Langflow 的解法是给环内节点开一个窄口子：首次执行时，只要还欠着的前驱全都在同一个环里，且这是个 loop 节点，就放行；一旦跑过一次（进了 ran_at_least_once），立刻恢复严格规则。
妙在哪： 用「跑没跑过」这一个布尔位，就把「破冰」和「防止乱序」两个矛盾需求分开了，不需要引入单独的环调度器。
依据：src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py:67-100（are_all_predecessors_fulfilled）。
4.6 环上的边是另一个类，靠「契约」传值
建边时会先算出哪些顶点在环上；只要边的任一端在环上，造出来的就不是 Edge 而是 CycleEdge。CycleEdge 多了一个 is_fulfilled 标志和一个 honor() 方法：取源节点已经建好的结果塞进目标节点的参数，并且明确拒绝为了取值去重建源节点。
妙在哪： 把「环上的取值必须是只读的」这条规矩写死在数据结构里，而不是靠调度器每次小心翼翼地判断。
依据：src/lfx/src/lfx/graph/graph/base.py:2602-2616（build_edge 二选一）、src/lfx/src/lfx/graph/edge/base.py:295-318（honor，源节点没建好直接抛错）。
4.7 启动不 import 组件，读一份预构建索引
354 个组件散在 95 个模块里，每个都可能拖进 langchain、boto3、各种 SDK。全 import 一遍，启动会非常慢。
Langflow 的做法是把所有组件的 template（连源码一起）预先烘焙成一个 6 MB 的 JSON 随包发布，启动时只读这个文件；带 SHA256 校验和版本号，对不上就回退到动态构建并把结果写进用户缓存。
组件真正被 import 的时机推迟到「这个节点要跑了」。
妙在哪： 三层回退（内置索引 → 用户缓存 → 动态构建）让 快路径极快，慢路径永远可用；而且索引里存的就是 template JSON 本身，前端 GET /all 几乎是零成本。
依据：src/lfx/src/lfx/interface/components.py:156（_read_component_index，含 SHA256 与版本校验）、:517（_load_production_mode 的回退链）、:1296（ensure_component_loaded 按需补齐）、src/lfx/src/lfx/components/__init__.py:342（模块级 __getattr__ 惰性导入）。
4.8 流程 JSON 自带源码 —— 便利与风险是同一件事
template 里有一个 code 字段，装的是这个组件类的完整 Python 源码，跟着 flow JSON 一起存、一起导出。这就是「在 UI 里改组件源码」和「导出的 flow 换台机器也能跑」的实现基础。
代价是：加载一个 flow 等于准备执行别人给的代码。所以 from_payload 里放了一道兜底校验，注释里明说「理想情况这该只在 API 边界做，但漏一个端点就是任意代码执行，所以这里再拦一次」。
妙在哪： 一个诚实的纵深防御案例——把安全检查放在唯一的必经之路上，而不是相信每个调用方都记得检查。
依据：src/lfx/src/lfx/custom/custom_component/component.py:1171-1187（to_frontend_node 塞入 code 字段）、src/lfx/src/lfx/graph/graph/base.py:1805-1812（兜底校验及其注释）。
4.9 一条 flow 免费获得三个对外出口
同一份 flow JSON，不改一行，同时是：
| 出口 | 入口符号 | 做了什么 |
|---|---|---|
| HTTP API | simplified_run_flow | POST /api/v1/run/{flow_id_or_name}，把 flow 当函数调 |
| MCP 工具 | handle_list_tools | 遍历标了 input 的节点，把它们的 template 字段翻译成 JSON Schema |
| Agent 的工具 | Component.to_toolkit | 把组件本身包成 LangChain Tool 给 agent 用 |
妙在哪： MCP 的 inputSchema 不是手写的，而是从「哪些字段 show 且非 advanced」自动推出来的——UI 上显示给人填的字段，正好就是给模型填的字段。
依据：src/backend/base/langflow/api/v1/endpoints.py:906、src/backend/base/langflow/api/v1/mcp_utils.py:434、src/backend/base/langflow/helpers/flow.py:540（json_schema_from_flow）、src/lfx/src/lfx/custom/custom_component/component.py:1573。
5. 代码地图（导航索引）
按主题给出文件路径 + 真实符号名。符号名比行号抗漂移，上游更新后用符号 grep 即可重新定位。
5.1 组件模型
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 组件基类、搭图 DSL | src/lfx/src/lfx/custom/custom_component/component.py | Component、Component.set、_connect_to_component、_add_edge |
| 输入/输出登记 | src/lfx/src/lfx/custom/custom_component/component.py | map_inputs、map_outputs |
| 类 → 前端节点 | src/lfx/src/lfx/custom/custom_component/component.py | to_frontend_node、_get_method_return_type |
| 只跑连了线的输出 | src/lfx/src/lfx/custom/custom_component/component.py | _should_process_output、_get_outputs_to_process、_build_results |
| 可反射属性白名单 | src/lfx/src/lfx/custom/attributes.py | ATTR_FUNC_MAPPING |
| 模板构建入口 | src/lfx/src/lfx/custom/utils.py | build_custom_component_template、build_custom_component_template_from_inputs、build_component_metadata |
| 模板配置抽取 | src/lfx/src/lfx/custom/custom_component/base_component.py | get_template_config、build_template_config |
stop() / start() 剪枝 API | src/lfx/src/lfx/custom/custom_component/custom_component.py | CustomComponent.stop、CustomComponent.start |
5.2 图的构建
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 从 JSON 建图 | src/lfx/src/lfx/graph/graph/base.py | Graph.from_payload、add_nodes_and_edges、_build_graph |
| 造顶点 / 选顶点类 | src/lfx/src/lfx/graph/graph/base.py | _build_vertices、_create_vertex、_get_vertex_class |
| 造边 / 环边判定 | src/lfx/src/lfx/graph/graph/base.py | build_edge、_build_edges、cycle_vertices |
| 边的类型校验与取值 | src/lfx/src/lfx/graph/edge/base.py | Edge.validate_handles、CycleEdge.honor、get_result_from_source |
| 顶点参数装配 | src/lfx/src/lfx/graph/vertex/base.py | Vertex.build_params、update_raw_params |
| 字段级参数处理 | src/lfx/src/lfx/graph/vertex/param_handler.py | ParameterHandler.process_edge_parameters、process_field_parameters |
5.3 调度与执行
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 准备与排层 | src/lfx/src/lfx/graph/graph/base.py | Graph.prepare、sort_vertices、initialize |
| 分层拓扑排序 | src/lfx/src/lfx/graph/graph/utils.py | get_sorted_vertices、layered_topological_sort、sort_chat_inputs_first |
| 单步执行（生成器式） | src/lfx/src/lfx/graph/graph/base.py | Graph.astep、async_start、start |
| 按层并发执行 | src/lfx/src/lfx/graph/graph/base.py | Graph.process、_execute_tasks、arun、_run |
| 建单个节点 + 冻结缓存 | src/lfx/src/lfx/graph/graph/base.py | Graph.build_vertex |
| 节点自身的构建流程 | src/lfx/src/lfx/graph/vertex/base.py | Vertex.build、_build、finalize_build、build_inactive |
| 「谁能跑」判定 | src/lfx/src/lfx/graph/graph/runnable_vertices_manager.py | RunnableVerticesManager.is_vertex_runnable、are_all_predecessors_fulfilled |
| 下一批可跑节点 | src/lfx/src/lfx/graph/graph/base.py | get_next_runnable_vertices、find_next_runnable_vertices |
5.4 环与分支
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 找环 | src/lfx/src/lfx/graph/graph/utils.py | find_cycle_vertices、find_all_cycle_edges、has_cycle |
| 迭代上限保护 | src/lfx/src/lfx/graph/graph/utils.py | should_continue |
| ACTIVE/INACTIVE 剪枝 | src/lfx/src/lfx/graph/graph/base.py | mark_branch、_mark_branch、mark_all_vertices |
| 条件路由持久剪枝 | src/lfx/src/lfx/graph/graph/base.py | exclude_branch_conditionally、exclude_branches_conditionally、_replace_conditional_exclusions |
| If-Else 组件 | src/lfx/src/lfx/components/flow_controls/conditional_router.py | ConditionalRouterComponent、iterate_and_stop_once、evaluate_condition |
| Loop 组件 | src/lfx/src/lfx/components/flow_controls/loop.py | LoopComponent、initialize_data、get_loop_body_vertices |
5.5 运行时与事件
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 构建端点 | src/backend/base/langflow/api/v1/chat.py | build_flow、get_build_events、cancel_build |
| 构建主流程 | src/backend/base/langflow/api/build.py | start_flow_build、generate_flow_events、build_vertices、cancel_flow_build |
| 事件分发 | src/lfx/src/lfx/events/event_manager.py | EventManager.send_event、register_event、create_default_event_manager |
| 节点构建事件 回推 | src/lfx/src/lfx/graph/utils.py | emit_vertex_build_event、log_vertex_build |
| 前端接事件 | src/frontend/src/utils/buildUtils.ts | buildFlowVertices、onEvent、processEndVertexEvent、BATCHABLE_EVENTS |
| 流式消息与 token | src/lfx/src/lfx/custom/custom_component/component.py | send_message、_stream_message、_process_chunk |
5.6 生态与出口
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 组件索引读取与回退 | src/lfx/src/lfx/interface/components.py | _read_component_index、_load_production_mode、_load_from_index_or_cache |
| 全量类型字典缓存 | src/lfx/src/lfx/interface/components.py | get_and_cache_all_types_dict、import_langflow_components |
| 按需补齐组件 | src/lfx/src/lfx/interface/components.py | ensure_component_loaded、load_single_component |
| 惰性导入 | src/lfx/src/lfx/components/__init__.py、src/lfx/src/lfx/components/_importing.py | 模块级 __getattr__、_dynamic_imports、import_mod |
| 预构建索引数据 | src/lfx/src/lfx/_assets/component_index.json | entries、metadata、sha256、version |
| 运行 flow 的 API | src/backend/base/langflow/api/v1/endpoints.py | simplified_run_flow、simple_run_flow、webhook_run_flow |
| MCP 出口 | src/backend/base/langflow/api/v1/mcp.py、mcp_utils.py | handle_list_tools、handle_call_tool |
| flow → JSON Schema | src/backend/base/langflow/helpers/flow.py | json_schema_from_flow |
| 组件 → Agent 工具 | src/lfx/src/lfx/custom/custom_component/component.py | to_toolkit、_get_tools、_handle_tool_mode |
关于本文的引用
- 所有
path:line均相对克隆根目录，锚定 commit26dc6fd3bc3a49178022b81c58e44a7d0a659a34（langflowv1.10.1）。 - 注意目录布局：
src/backend/base/langflow/graph、custom、interface等大多是再导出的薄壳，真实实现在src/lfx/src/lfx/下（依据：src/backend/base/langflow/graph/__init__.py:1-4全部from lfx.graph...转发）。查代码时直接去src/lfx/。