Vertex as the worker station of the graph

To reason about the design, it helps to picture Vertex as a worker station on an assembly line. The graph is the conveyor system; edges are belts; components are the workers doing the task; and the Vertex object supervises one station: it coordinates inputs, runs the worker, and hands off the outputs.

Project: langflow

src/
  lfx/
    graph/
      graph/
        base.py        (Graph orchestration)
      edge/
        base.py        (Edge routing between vertices)
      vertex/
        base.py  <---- (Vertex: wraps a component, manages params, build lifecycle)
        schema.py      (Node schemas)
    interface/
      initialize.py    (instantiate_class, get_instance_results)
      listing.py       (lazy_load_dict)
    schema/
      schema.py        (ResultData, OutputValue, build_output_logs)
      message.py       (Message)
      data.py          (Data)
      artifact.py      (ArtifactType)
    utils/
      schemas.py       (ChatOutputResponse)
      util.py          (sync_to_async)
    log/
      logger.py        (logger)

Each vertex knows four main things:

• Which component it wraps (vertex_type, base_type, custom_component).
• How it’s wired (incoming/outgoing edges, predecessors, successors).
• What inputs it needs and where they come from (templates, edges, runtime input).
• How to normalize the component’s output into shared schemas (ResultData, artifacts, logs, messages).

The key design choice: components don’t know about the graph. Vertex owns orchestration, keeping components focused on business logic instead of wiring and lifecycle.

The build lifecycle: from params to ResultData

Once we treat Vertex as a station supervisor, the core question becomes: what exactly happens when we tell it to run? That story is the build lifecycle: gather parameters, execute the component, normalize outputs, and wrap everything in a result schema.

The asynchronous entrypoint

The public API for executing a node is Vertex.build(...). It’s asynchronous, protected by a per-vertex lock, and handles more than just calling the component: it lazy-loads code, enforces state rules, injects chat inputs, runs a step pipeline, and logs the transaction.

async def build(
    self,
    user_id=None,
    inputs: dict[str, Any] | None = None,
    files: list[str] | None = None,
    requester: Vertex | None = None,
    event_manager: EventManager | None = None,
    **kwargs,
) -> Any:
    from lfx.interface.components import ensure_component_loaded
    from lfx.services.deps import get_settings_service

    settings_service = get_settings_service()
    if settings_service and settings_service.settings.lazy_load_components:
        component_name = self.id.split("-")[0]
        await ensure_component_loaded(self.vertex_type, component_name, settings_service)

    async with self.lock:
        if self.state == VertexStates.INACTIVE:
            self.build_inactive()
            return None

        is_loop_component = self.display_name == "Loop" or self.is_loop
        if self.frozen and self.built and not is_loop_component:
            return await self.get_requester_result(requester)
        if self.built and requester is not None:
            return await self.get_requester_result(requester)

        self._reset()

        if self.graph and self.graph.flow_id:
            await emit_build_start_event(self.graph.flow_id, self.id)

        # Session & chat input injection (simplified)
        if inputs and "session" in inputs and self.has_session_id:
            session_id_value = self.get_value_from_template_dict("session_id")
            if session_id_value == "":
                self.update_raw_params({"session_id": inputs["session"]}, overwrite=True)
        if self._is_chat_input() and (inputs or files):
            chat_input = {}
            ...
            self.update_raw_params(chat_input, overwrite=True)

        # Run configured steps (pipeline)
        for step in self.steps:
            if step not in self.steps_ran:
                await step(user_id=user_id, event_manager=event_manager, **kwargs)
                self.steps_ran.append(step)

        self.finalize_build()

        # Transaction logging (success path)
        flow_id = self.graph.flow_id
        if flow_id:
            outputs_dict = None
            if self.outputs_logs:
                outputs_dict = {
                    k: v.model_dump() if hasattr(v, "model_dump") else v
                    for k, v in self.outputs_logs.items()
                }
            await self._log_transaction_async(
                str(flow_id), source=self, target=None, status="success", outputs=outputs_dict
            )

    return await self.get_requester_result(requester)

The overall pattern is classic template method: higher-level orchestration logic in build, with self.steps (by default just self._build) providing the extensible core. That allows new behavior to be introduced as additional steps without rewriting the main control flow.

From wiring to actual arguments

Before a component can run, the vertex must gather all its inputs. Langflow separates two views of parameters to make this explicit: a wiring view (raw_params) and a runtime view (params).

build_params is responsible for the initial collection. It uses a ParameterHandler to combine:

• Field parameters – static config from the node template (default values, flags).
• Edge parameters – dynamic values flowing from upstream vertices.

def build_params(self) -> None:
    if self.graph is None:
        raise ValueError("Graph not found")

    if self.updated_raw_params:
        # Defer to _build_each_vertex_in_params_dict to reset
        return

    param_handler = ParameterHandler(self, storage_service=None)

    edge_params = param_handler.process_edge_parameters(self.edges)
    field_params, load_from_db_fields = param_handler.process_field_parameters()

    # Edge params override field params
    self.params = {**field_params, **edge_params}
    self.load_from_db_fields = load_from_db_fields
    self.raw_params = self.params.copy()

The critical distinction is:

• raw_params can contain vertices (single, lists, dicts). It represents how the graph is wired.
• params is what the component actually sees, after vertices have been resolved into concrete values.

The method _build_each_vertex_in_params_dict walks raw_params, calls get_result on any nested vertices, and writes the resolved values into params. The flag updated_raw_params ensures that when runtime data (like chat messages) mutates raw_params via update_raw_params, we don’t silently overwrite those values by rebuilding from the graph again.

Executing the component and normalizing outputs

With params ready, the private _build method executes the component. Here Vertex acts as a facade over Langflow’s component loader.

async def _build(
    self,
    fallback_to_env_vars,
    user_id=None,
    event_manager: EventManager | None = None,
) -> None:
    await logger.adebug(f"Building {self.display_name}")
    await self._build_each_vertex_in_params_dict()
    if self.base_type is None:
        raise ValueError(f"Base type for vertex {self.display_name} not found")

    if not self.custom_component:
        custom_component, custom_params = initialize.loading.instantiate_class(
            user_id=user_id, vertex=self, event_manager=event_manager
        )
    else:
        custom_component = self.custom_component
        if hasattr(self.custom_component, "set_event_manager"):
            self.custom_component.set_event_manager(event_manager)
        custom_params = initialize.loading.get_params(self.params)

    await self._build_results(
        custom_component=custom_component,
        custom_params=custom_params,
        fallback_to_env_vars=fallback_to_env_vars,
        base_type=self.base_type,
    )

    self._validate_built_object()
    self.built = True

Component execution can return different shapes (plain value, tuple with artifacts, etc.). _update_built_object_and_artifacts normalizes these into internal fields such as built_object, artifacts_raw, and artifacts_type. That keeps the rest of the class agnostic to the component’s exact return convention.

The final step, finalize_build, converts internal state into a single ResultData object, including logs, artifacts, messages, and aggregated token usage. This is the only thing the rest of the system needs to handle.

def finalize_build(self) -> None:
    result_dict = self.get_built_result()

    self.set_artifacts()  # hook, currently a no-op
    artifacts = self.artifacts_raw
    messages = self.extract_messages_from_artifacts(artifacts) if isinstance(artifacts, dict) else []
    token_usage = self._extract_token_usage()

    result_dict = ResultData(
        results=result_dict,
        artifacts=artifacts,
        outputs=self.outputs_logs,
        logs=self.logs,
        messages=messages,
        component_display_name=self.display_name,
        component_id=self.id,
        token_usage=token_usage,
    )
    self.set_result(result_dict)

The reusable pattern here is simple and powerful: collect parameters → execute component → normalize outputs → emit a single result schema. That’s the backbone of a maintainable workflow engine.

State, freezing, and concurrency

Once a single build works, runtime concerns appear: what if multiple requests hit the same vertex, when should work be skipped, and how do we reuse expensive results safely? Vertex answers these through per-node locking, lifecycle flags, and light coordination with the graph.

Per-vertex locking

Each vertex owns an asyncio.Lock. All mutations of build-related fields (self.built, self.params, self.result, etc.) occur inside this lock via build and get_result. Concurrent builds for the same vertex are serialized, while independent vertices can run in parallel.

This keeps internal state simple: most methods can assume a single logical build in progress and read/write instance attributes without fine-grained locking. The cost is that very hot vertices become serialized bottlenecks, but the trade-off is often worth the simpler reasoning.

Inactive, frozen, and loops

Vertex lifecycle is modeled through a state enum and a few flags: state, frozen, is_loop, and use_result. Together they control when work actually happens.

• INACTIVE: build short-circuits, marks the vertex as built, and returns None. This lets you disable nodes without rewiring the graph.
• Frozen: if a vertex is frozen and already built, subsequent builds reuse the existing result instead of re-running the component, unless it’s a loop.
• Loop components: identified by display_name == "Loop" or flags like allows_loop, they always execute even when frozen, because they iterate over data rather than compute a single cacheable value.

This logic lives in build: early exit for inactive nodes, cache hits for frozen nodes, and a special case for loops. It gives Langflow a mix of memoization and explicit turning off of subgraphs without scattering caching logic across components.

Graph-wide awareness without full coupling

Vertex also coordinates with the broader graph in a few places. The most direct is set_state, which updates graph.inactivated_vertices when nodes become inactive or active again, taking into account the node’s in-degree to avoid marking merge points too aggressively.

This is one of the points where the abstraction frays: vertex code reaches into graph.edges, graph.in_degree_map, and graph.inactivated_vertices instead of going through a dedicated graph API. It works, but it’s a hint that some responsibilities (like propagating inactive status) should live on the graph itself.

Token usage and observability

Correctness isn’t enough for a workflow engine running LLMs in production. We also need visibility: how many tokens are we spending, where are failures happening, and how long does each node take? Langflow treats these as first-class concerns inside Vertex.

Aggregating token usage across upstream nodes

Token usage isn’t tracked only per component. A vertex can compute token usage “up to this point” by walking all upstream vertices and summing their Usage values, plus its own component’s usage when available.

def _get_all_upstream_vertices(self) -> list[Vertex]:
    visited: set[str] = set()
    result: list[Vertex] = []
    stack = [edge.source_id for edge in self.graph.edges if edge.target_id == self.id]

    while stack:
        vid = stack.pop()
        if vid in visited:
            continue
        visited.add(vid)
        vertex = self.graph.get_vertex(vid)
        result.append(vertex)
        stack.extend(edge.source_id for edge in self.graph.edges if edge.target_id == vid)

    return result

def _accumulate_upstream_token_usage(self) -> Usage | None:
    predecessors = self._get_all_upstream_vertices()
    total_input = 0
    total_output = 0
    has_data = False

    for predecessor in predecessors:
        if predecessor.result and predecessor.result.token_usage:
            usage = predecessor.result.token_usage
            total_input += usage.input_tokens or 0
            total_output += usage.output_tokens or 0
            has_data = True

    if self.custom_component:
        own_usage = self.custom_component._token_usage
        if own_usage:
            total_input += own_usage.input_tokens or 0
            total_output += own_usage.output_tokens or 0
            has_data = True

    if not has_data:
        return None

    return Usage(
        input_tokens=total_input,
        output_tokens=total_output,
        total_tokens=total_input + total_output,
    )

Functionally, this gives the UI a meaningful number: “tokens consumed so far before and including this node.” Technically, it’s an O(E) traversal over graph.edges every time it runs, and it hardcodes knowledge of graph internals inside Vertex. The obvious next step is to move this traversal behind a graph-level API so it can be cached or optimized per topology.

Events, metrics, and transaction logs

Token usage is only one dimension of observability. Vertex is also the central place where execution is framed for the UI and logging systems.

• UI events: before_callback_event and after_callback_event produce structured events like StepStartedEvent and StepFinishedEvent, including raw metrics collected during execution.
• Transaction logging: _log_transaction_async records success or failure of each execution via log_transaction, capturing outputs in a structured way when available.
• Output logs: build_output_logs converts raw component outputs into OutputValue structures, which are easier to render and inspect.

The performance analysis around Vertex suggests a few concrete metrics that pair well with this design:

The pattern is consistent with the core lesson: the orchestrator is where you see both inputs and outputs. That makes it the right layer to emit events and metrics, instead of forcing every component to learn about observability concerns.

Design tension and refactoring pressure

The centralization of orchestration inside Vertex is deliberate and powerful, but it comes with tension: as more concerns accumulate (token accounting, chat formatting, state propagation, logging), the class risks turning into a god object.

Where the seams start to show

The analysis of Vertex highlights several specific smells:

• Multiple concerns intertwined: orchestration, observability, chat input handling, token aggregation, and state management all live in one class.
• Direct graph access: methods like _get_all_upstream_vertices and set_state reach into graph.edges and graph.inactivated_vertices instead of calling graph APIs.
• Magic strings: behavior keyed off display_name == "Loop" or "Text Output" instead of explicit capabilities on components.
• Placeholder hooks: hooks like set_artifacts are currently no-ops but still called, which can confuse readers about where artifacts are actually processed.

The recommended refactors are straightforward and generalize well beyond this codebase:

1. Move graph traversals into the graph layer – for token aggregation and inactivation propagation, expose narrow methods like graph.get_all_upstream_vertices(vertex) and keep Vertex as a consumer.
2. Replace magic names with capabilities – let components declare properties such as is_loop_component or supports_streaming, and interrogate those instead of hardcoding display names.
3. Clarify parameter mutation paths – simplify update_raw_params to avoid mutating caller mappings, and tighten the lifecycle of updated_raw_params so readers can see exactly when wiring-derived params are rebuilt.

Each of these moves peels one concern away from the central orchestrator or clarifies a boundary. That keeps the core idea—components stay pure, orchestration lives in one place—while making the class easier to maintain.

What holds up well

Despite its size, Vertex gets several important things right:

• Clear public surface: methods like build, get_result, build_params, update_raw_params, instantiate_component, set_state, and apply_on_outputs form a coherent API.
• Good error semantics: descriptive ValueError/TypeError messages and a dedicated ComponentBuildError that wraps tracebacks provide usable signals when builds fail.
• Strong result contracts: by always converging to ResultData, the rest of the system and the frontend can evolve without knowing the quirks of individual components.

The practical lesson is not “never have a big class”, but “if one object orchestrates everything, keep its seams clear so responsibilities can be pushed out over time without breaking the core contract.”

What to borrow for your own systems

Looking at Langflow’s Vertex as a whole, the central idea is consistent: components stay narrow and focused, while a single orchestration layer manages wiring, lifecycle, and observability. The implementation has rough edges, but the patterns are solid and reusable.

Actionable lessons

• Centralize orchestration, not business logic. Let components read parameters and return values. Keep graph awareness, lazy loading, freezing, and transaction logging in a dedicated orchestrator object.
• Separate wiring from runtime arguments. Maintain a “wiring view” of parameters (which may contain nodes) and a “runtime view” (raw values only). Make the transformation explicit and testable.
• Make observability a first-class concern of the orchestrator. Emit per-node metrics, structured logs, and UI events from the orchestration layer, where you see both inputs and outputs in context.
• Watch for god-object creep and extract early. When central classes start handling graph traversal, feature flags, and special cases, move those responsibilities into helpers or the graph/module where they belong.
• Design for concurrent execution with simple state. Use per-node locks and well-defined lifecycle flags (inactive, frozen, loop) so you can reason about behavior under load without scattering synchronization logic.

If you’re building your own AI workflow engine or any graph-based orchestrator, walking through a class like Vertex is a useful exercise. It shows how much leverage you get from a single well-placed abstraction—and how important it is to keep that abstraction clean as production concerns like observability and caching inevitably accumulate.