def _process_transformer_chain(
    dag: schemas.SyncDag,
    entity_dag: schemas.SyncDagCreate,
    current_node: schemas.DagNode,
    destination_node_create: schemas.DagNodeCreate,
) -> None:
    """Process the chain of transformers until reaching a destination."""
    while True:
        edges_from_current = dag.get_edges_from_node(current_node.id)
        if not edges_from_current:
            raise HTTPException(
                status_code=400, detail=f"Node '{current_node.name}' has no outgoing edges"
            )

        if len(edges_from_current) > 1:
            raise HTTPException(
                status_code=400,
                detail=(
                    f"Node '{current_node.name}' has more than one outgoing edge. "
                    f"Only one path is allowed."
                ),
            )

        edge_from_current = edges_from_current[0]
        edge_from_current_create = convert_to_edge_create(edge_from_current)
        entity_dag.edges.append(edge_from_current_create)

        next_node = dag.get_node(edge_from_current.to_node_id)

        if next_node.type == NodeType.destination:
            # Reached destination, we're done
            break
        elif next_node.type == NodeType.transformer:
            # Add transformer node
            next_node_create = convert_to_node_create(next_node)
            entity_dag.nodes.append(next_node_create)

            # Check if this is a web fetcher (which should continue the chain)
            # or a regular chunker (which should end the chain)
            if "web_fetcher" in next_node.name.lower() or "web fetcher" in next_node.name.lower():
                # Web fetcher continues the chain (outputs FileEntity)
                current_node = next_node
            else:
                # Regular transformers (like file_chunker) end the chain
                # After a transformer, connect directly to destination
                # (skip parent/chunk entities for UI simplification)
                transformer_to_dest_edge = schemas.DagEdgeCreate(
                    from_node_id=next_node.id, to_node_id=destination_node_create.id
                )
                entity_dag.edges.append(transformer_to_dest_edge)
                break
        elif next_node.type == NodeType.entity:
            # Add entity node and continue following the chain
            next_node_create = convert_to_node_create(next_node)
            entity_dag.nodes.append(next_node_create)
            current_node = next_node
        else:
            raise HTTPException(
                status_code=400,
                detail=(
                    f"Node ({next_node.name}) after '{current_node.name}' "
                    f"has unsupported type ({next_node.type})."
                ),
            )
Process the chain of transformers until reaching a destination.
backend/airweave/api/v1/endpoints/dag.py