看 WorkflowEntry.__init__ 的入参（api/core/workflow/workflow_entry.py:163），它列出了跑一个工作流需要的全部"零件"：

构造函数核心就是造一个 GraphEngine 并给它挂"层"（api/core/workflow/workflow_entry.py:213）：

# api/core/workflow/workflow_entry.py:213
self.graph_engine = GraphEngine(
    workflow_id=workflow_id,
    graph=graph,
    graph_runtime_state=graph_runtime_state,
    command_channel=command_channel,
    config=GraphEngineConfig(                          # 并发配置：几个 worker 并行跑节点
        min_workers=dify_config.GRAPH_ENGINE_MIN_WORKERS,
        max_workers=dify_config.GRAPH_ENGINE_MAX_WORKERS,
        scale_up_threshold=dify_config.GRAPH_ENGINE_SCALE_UP_THRESHOLD,
        scale_down_idle_time=dify_config.GRAPH_ENGINE_SCALE_DOWN_IDLE_TIME,
    ),
    child_engine_builder=self._child_engine_builder,   # 嵌套子工作流用
)

if dify_config.DEBUG:                                  # ① 调试模式 → 挂"详细日志层"
    self.graph_engine.layer(DebugLoggingLayer(...))

limits_layer = ExecutionLimitsLayer(                   # ② ★挂"执行限制层"：熔断
    max_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS,
    max_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME)
self.graph_engine.layer(limits_layer)
self.graph_engine.layer(LLMQuotaLayer(tenant_id=tenant_id))   # ③ 挂"LLM 额度层"

if dify_config.ENABLE_OTEL or is_instrument_flag_enabled():
    self.graph_engine.layer(ObservabilityLayer())      # ④ 挂"可观测层"(OTel)

装配好后，执行就一句 run()（api/core/workflow/workflow_entry.py:250）：

# api/core/workflow/workflow_entry.py:250
def run(self) -> Generator[GraphEngineEvent, None, None]:   # ★返回类型是"事件生成器"
    graph_engine = self.graph_engine
    try:
        # 在 Graphon 引擎的事件流上，套一层 Dify 的"响应流"兼容过滤
        generator = iter_dify_graph_engine_events(graph_engine, self._response_stream_filter)
        yield from generator                                 # ① 引擎每产一个事件就 yield 出去
    except GenerateTaskStoppedError:
        pass                                                 # ② 用户主动停止 → 静默收尾
    except Exception as e:
        logger.exception("Unknown Error when workflow entry running")
        yield GraphRunFailedEvent(error=str(e))              # ③ 出错也变成一个"失败事件"吐出
        return

图要跑，先得懂图的结构。graph_topology.py 把 JSON 解析成可查询的拓扑（api/core/workflow/graph_topology.py:16）：

# api/core/workflow/graph_topology.py:16
class WorkflowGraphTopology:
    @classmethod
    def from_graph(cls, graph):                       # ① 从 {nodes, edges} 解析
        node_ids = cls._node_ids_from_graph(graph)    #    收集所有节点 id
        edges = graph.get("edges")
        for edge in edges:                            #    每条边: source → target
            source = edge.get("source"); target = edge.get("target")
            incoming[target].append(source)           #    记录"target 的上游有谁"
        return cls(node_ids=node_ids, incoming=incoming)

    def is_upstream(self, *, source_node_id, target_node_id):   # ② 判断 A 是不是 B 的上游
        if source_node_id == target_node_id: return ...
        queue = deque(self._incoming.get(target_node_id, ()))   #    从 B 往上游 BFS
        while queue:
            candidate = queue.popleft()
            if candidate == source_node_id: return True         #    找到 A → 是上游
            ...

画布上你常需要"只测这一个节点对不对"，不想跑整张图。WorkflowEntry.single_step_run（api/core/workflow/workflow_entry.py:265）就干这个：

# api/core/workflow/workflow_entry.py:265
def single_step_run(cls, *, workflow, node_id, user_id, user_inputs, variable_pool, ...):
    node_config = workflow.get_node_config_by_id(node_id)       # ① 只取这一个节点的配置
    node_type = node_config["data"].type
    node_version = str(node_config["data"].version)
    node_cls = resolve_workflow_node_class(                     # ② 解析出它的节点类(Day09)
        node_type=node_type, node_version=node_version)
    run_context = build_dify_run_context(                       # ③ 造运行上下文(DEBUGGER 模式)
        tenant_id=workflow.tenant_id, app_id=workflow.app_id,
        user_id=user_id, invoke_from=InvokeFrom.DEBUGGER)
    graph_runtime_state = GraphRuntimeState(variable_pool=variable_pool, start_at=..., ...)
    if is_start_node_type(node_type):                           # ④ 若是开始节点，把用户输入灌进池子
        add_node_inputs_to_pool(variable_pool, node_id=node_id, inputs=user_inputs)
    ...