Рекурсивные типы. Часть 4/5. Схемы рекурсии

Содержание четвёртой части:

• Пример для мотивации

• Обобщение свёрток/развёрток

• Параморфизм и другие схемы рекурсии

• Хистоморфизм

• Футуморфизм

• Промежуточный итог

Другие части обзора

• Часть 1. Рекурсия.

• Часть 2. Неподвижные точки конструкторов типов.

• Часть 3. Свободные контейнеры.

• Часть 4. Схемы рекурсии (мы здесь).

• Часть 5. Занимательный матан.

Пример для мотивации

Свёртка и развёртка неподвижных точек конструкторов типов покрывает многие сценарии, где приходилось бы вручную писать рекурсивные функции, но иногда этого оказывается недостаточно. Рассмотрим пример.

В некоторой компании для каждого работника задан обобщённый перечь атрибутов:

type EmployerAttributes[A] = (Employer, List[A])

Одними из таких атрибутов являются подчинённые – такие же работники компании. Формируется иерархия сотрудников, которую можно записать посредством наименьшей неподвижной точки:

type Hierarchy = μ[EmployerAttributes]

Стоит задача – имея полную иерархию научиться находить для заданного работника всех его подчинённых. Такую функцию можно реализовать рекурсивно:

def findSubordinates(employer: Employer)(hierarchy: Hierarchy): Option[Hierarchy] =   val (empl, subrordinates) = outμ(hierarchy) //     отдаётся иерархия ↑↑↑↑↑↑↑↑↑ указанного сотрудника   Option //                 ↓↓↓↓↓↓↓↓↓ на каждом шаге требуется вся текущая иерархия     .when(empl == employer)(hierarchy)     .orElse(subrordinates.flatMap(findSubordinates(employer)).headOption)     //                            ↑↑↑↑рекурсия↑↑↑↑

Для вызова outμ потребуется доказательство ковариантности EmployerAttributes:

given Lift[EmployerAttributes] = [A, B] => (f: A => B) =>   (aLst: EmployerAttributes[A]) =>; aLst._1 -> aLst._2.map(f)

Пусть у нас есть такая структура компании:

type Employer = String  val hierarchy: Hierarchy = inμ("Бугор", //      Бугор   List(inμ("Петя",                      //       /     List(inμ("Ваня", Nil), inμ("Маша",  //     Петя       List(inμ("Вася", Nil))            //     /  \ ))                                  //  Ваня  Маша   ))                                    //       / )                                       //     Вася  val employerStringAlg: Algebra[EmployerAttributes][String] =   attrs => attrs._1 + attrs._2.mkString(": (", ", ", ")")  foldμ(employerStringAlg)(hierarchy)     // "Бугор: (Петя: (Ваня: (), Маша: (Вася: ())))"

Тогда найти подчинённых сотрудника можно так:

findSubordinates("Петя")(hierarchy)     // Some(Петя: (Ваня: (), Маша: (Вася: ()))) findSubordinates("Вася")(hierarchy)     // Some(Вася: ()) findSubordinates("Катя")(hierarchy)     // None

Применять для тех же целей функцию свёртки, описанную ранее, весьма не удобно. Проблема заключается в том, что простая свёртка оперирует F-алгеброй, которая в нашем случае распаковывает единственное значение типа Employer, в то время когда нам на каждом шаге в функции findSubordinates требуется как работник, так и вся иерархия его подчинённых. Другими словами, нам требуется другая свёртка, с более сложной алгеброй.

Обобщение свёрток/развёрток

Как можно обобщить F-алгебру для использования её при свёртке рекурсивных структур? Основная задача алгебры – распаковка значения некого типа X из контейнера F. Поэтому, пожалуй, в наиболее общем виде такими алгебрами будут конструкции вида

//              Algebra                   = [F[_]] =>> [X] =>>   F[  X ]  =>   X type GeneralizedAlgebra[W[_], G[_], M[_]] = [F[_]] =>> [X] =>> W[F[G[X]]] => M[X]

Но для простоты понимания можно декомпозировать эту конструкцию на три частных случая:

type     GAlgebra [G[_]] = [F[_]] =>> [X] =>> F[G[X]] =>   X type ElgotAlgebra [W[_]] = [F[_]] =>> [X] =>> W[F[X]] =>   X // алгебры Элгота type      AlgebraM[M[_]] = [F[_]] =>> [X] =>>   F[X]  => M[X]

В основе любой свёртки по-прежнему лежит всё тот же базовый алгоритм fold, описанный ранее. Секрет использования обобщённых алгебр в тех же целях заключается в их приведении к обычной F-алгебре, используемой в fold. Для реализации такого приведения необходимы некоторые возможности дополнительных контейнерных типов. Пожалуй, наиболее важной среди них является перестановка этих контейнеров с F[_]. Введём дополнительные обозначения:

type ~>[F[_], G[_]] = [X] => F[X] => G[X]   // естественное преобразование type ∘ [F[_], G[_]] = [X] =>> F[G[X]]       // композиция контейнеров type ⇄ [F[_], G[_]] = (F ∘ G) ~> (G ∘ F)    // собственно, перестановка контейнеров  def swap[F[_], G[_]](using sw: F ⇄ G) = sw // [X] => F[G[X]] => G[F[X]]

Также для удобства определим следующие классы типов:

type SwapOut[F[_]] = [G[_]] =>> F ⇄ G     // перестановка вложенного наружу type SwapIn [F[_]] = [G[_]] =>> G ⇄ F     // перестановка внешнего внутрь

Ещё нужны будут возможности «разматрёшивания» и «заматрёшивания» дополнительных контейнеров. Вообще, понятия «монады» и «комонады» заимствованы из теории категорий, и обоснование их использования выходит за рамки данного обзора. Поэтому тут введём их без комментариев:

case class Monad[F[_]: Lift]( // требуется доказательство коварианности F[_]   pure:         Id ~> F,   flatten: (F ∘ F) ~> F, ) { val lift = summon[Lift[F]] }  given liftFromMoand[F[_]](using m: Monad[F]): Lift[F] = m.lift def pure    [F[_]: Monad] = summon[Monad[F]].pure def flatten [F[_]: Monad] = summon[Monad[F]].flatten def flatFMap[F[_]: Monad] = [A, B] => (f: A => F[B]) => fmap[F](f) andThen flatten[F][B]  extension  [F[_]: Monad, A] (fa: F[A])   def flatMap[B] = (f: A => F[B]) => flatFMap(f)(fa) extension  [F[_]: Monad, A, B] (afb: A => F[B])   def andThenK[C] = (bfc: B => F[C]) => (a: A) => afb(a).flatMap(bfc)   case class Comonad[F[_]: Lift]( // требуется доказательство коварианности F[_]   extract:   F ~> Id,   coFlatten: F ~> (F ∘ F), ) { val fmap = summon[Lift[F]] }  given liftFromComoand[F[_]](using cm: Comonad[F]): Lift[F] = cm.fmap def extract  [F[_]: Comonad] = summon[Comonad[F]].extract def coFlatten[F[_]: Comonad] = summon[Comonad[F]].coFlatten def coFlatMap[F[_]: Comonad] = [A, B] => (f: F[A] => B) => coFlatten[F][A] andThen fmap[F](f)

Вооружившись перечисленными возможностями, вот так можно записать преобразование алгебр

def gAlgToFAlg[F[_]: Lift, G[_]: Comonad: SwapOut[F], X]:          AlgebraG[G][F][X] => Algebra[F][G[X]]    =   galg => fmap[F](coFlatten[G][X]) andThen swap[F, G][G[X]] andThen fmap[G](galg)  def elgotAlgToFAlg[F[_]: Lift, W[_]: Comonad: SwapOut[F], X]: ElgotAlgebra [W][F][X] => Algebra[F][W[F[X]]] =   elgotAlg => fmap(coFlatMap(elgotAlg)) andThen swap[F, W][X]  def algMToFAlg[F[_]: Lift, M[_]: Monad: SwapOut[F], X]:            AlgebraM[M][F][X] => Algebra[F][M[X]]    =   swap[F, M][X] andThen flatFMap(_)

и соответствующие им свёртки:

def gFold[Fix[_[_]]: Fold, F[_]: Lift, G[_]: Comonad: SwapOut[F], X]: AlgebraG[G][F][X] => Fix[F] => X =   gAlgToFAlg(_) pipe foldFix[Fix][F][G[X]] andThen extract[G][X]  def elgotFold[Fix[_[_]]: Fold, F[_]: Lift, W[_]: Comonad: SwapOut[F], X]: ElgotAlgebra[W][F][X] => Fix[F] => X =   elgotAlg => elgotAlgToFAlg(elgotAlg) pipe foldFix[Fix][F][W[F[X]]] andThen elgotAlg  def foldM[Fix[_[_]]: Fold, F[_]: Lift, M[_]: Monad: SwapOut[F], X]: AlgebraM[M][F][X] => Fix[F] => M[X] =   algMToFAlg andThen foldFix[Fix][F][M[X]]

Комбинатор pipe, передающий аргумент слева в функцию справа, доступен благодаря import scala.util.chaining.scalaUtilChainingOps.

Параморфизм и другие схемы рекурсии

Для решения задачи о подчинённых можно ввести обобщённую свёртку, называемую «параморфизм»:

type ParaW[Fix[_[_]]] = [F[_]] =>> [X] =>> (X, Fix[F]) // помимо одного элемента, доступна вся оставшаяся структура type ParaAlgebra[Fix[_[_]], F[_]] = [X] =>> GAlgebra[ParaW[Fix][F]][F][X]  given paraLift[Fix[_[_]], F[_]]: Lift[ParaW[Fix][F]] =   [A, B] => (f: A =>> B) => (pfixfa: ParaW[Fix][F][A]) => f(pfixfa._1) -> pfixfa._2  given paraComonad[Fix[_[_]], F[_]]: Comonad[ParaW[Fix][F]] = Comonad(   extract   = [X] => (px: (X, Fix[F])) => px._1,   coFlatten = [X] => (px: (X, Fix[F])) => px -> px._2, )  given paraSwapOut[Fix[_[_]]: Fold: InFix, F[_]: Lift]: F ⇄ ParaW[Fix][F] =   [X] => (fpara: F[ParaW[Fix][F][X]]) =>     fpara.map((_: ParaW[Fix][F][X])._1) -> inFix(fpara.map(_._2))  def para[Fix[_[_]]: Fold: InFix, F[_]: Lift, X]: ParaAlgebra[Fix, F][X] => Fix[F] => X =   gfold[Fix, F, ParaW[Fix][F], X]

Теперь в параморфизм можно спрятать всю рекурсию из нашей задаче:

def subordinatesParaAlg(employer: Employer): ParaAlgebra[μ, EmployerAttributes][Option[Hierarchy]] =   case (empl, subs) => Option     .when(empl == employer)(empl -> subs.map(_._2) pipe inFix[μ, EmployerAttributes])     .orElse(subs.flatMap(_._1).headOption)  def findSubordinates(employer: Employer): Hierarchy => Option[Hierarchy] =   para[μ, EmployerAttributes, Option[Hierarchy]](subordinatesParaAlg(employer))

Обычная свёртка обрабатывает каждый элемент рекурсивной структуры по-отдельности. В этом смысле её иногда называют итератором. В тоже время параморфизм рекурсивно проваливается внутрь, на каждом шагу давая доступ ко всей оставшейся структуре. Поэтому такую свёртку иногда называют простой рекурсией или рекурсором.

Параморфизм оказывается полезен для решения целого класса задач, и по этой причине ему присвоено собственное название. Также свои названия имеют и свёртки, основанные на некоторых других частных случаях обобщённых алгебр. Например, обычная свёртка fold традиционно называется катаморфизмом (cata). Названия зачастую заимствуются из греческого языка.

Схожим образом строятся и обобщённые развёртки. Соответствующие обобщённые коалгебры приводятся к обычным, для чего используются такие же монадные и комонадные возможности дополнительных контейнерных типов, а также возможность их перестановки с целевым F[_]. У многих обобщённых развёрток также устоялись собственные названия. Например, обычный unfold называется анаморфизмом (ana), а развёртка, дуальная параморфизму – апоморфизмом (apo).

Без своих имён не остались и некоторые пересвёртки (refold). В частности, катаморфизм после анаморфизма образуют хиломорфизм (hylo). Вот как через него выражается функция вычисления факториала:

def ana [Fix[_[_]]: Unfold] = unfoldFix[Fix]         //  анаморфизм  def cata[Fix[_[_]]:   Fold] =   foldFix[Fix]         // катаморфизм  def hylo[Fix[_[_]]: Unfold: Fold, F[_]: Lift, A, B]( // хиломорфизм   coalgebra: Coalgebra[F][A],   algebra: Algebra[F][B] ): A => B =   ana[Fix](coalgebra) andThen cata[Fix](algebra)      val natCoalg: Coalgebra[OptCell[Int]][Int ] = n => Option.when(n > 0)(n, n - 1) // список чисел [n .. 1] val factAlg:    Algebra[OptCell[Int]][Long] = _.fold(1L)(_ * _)                 // либо 1, либо n * prev  val fact = hylo(natCoalg, factAlg)  // Int => Long fact(10) // 3628800

Про ещё одну пересвёртку, динамоморфизм, будет рассказано в следующем разделе.

Опираясь на представленные выше примеры, можно собирать и более сложные алгебры/коалгебры – в них дополнительные контейнеры будут композицией контейнеров попроще. Имена таких свёрток/развёрток собираются из имён морфизмов, соответствующих алгебрам, на базе которых были собраны эти композитные конструкции. В частности, так строится ставший уже мемом зигогистоморфный препроморфизм (zygoHistoPrepro).

Хистоморфизм

Обычные списки List представляются в виде наименьших неподвижных точек. Для них важна прежде всего свёртка, когда сперва обрабатывается текущее значение head, а затем все последующие (tail). В свою очередь, потоки Stream или ленивые списки LazyList являются наибольшими неподвижными точками, нацеленными на развёртку, рост. На такие структуры можно смотреть как на пару текущего значения current и всех предыдущих (previous), которые были вычислены ранее.

В процессе свёртки наименьшей неподвижной точки мы шаг за шагом получаем промежуточные результаты и иногда встречаются задачи, когда оказывается полезным подсмотреть, что же было вычислено на предыдущих шагах. Заглянуть в прошлое нам поможет схема рекурсии с названием «хистоморфизм», использующая Cofree (см. предыдущую часть обзора):

extension [F[_]: Lift, X](cof: Cofree[F][X])     def current  = unapplyCofree[F, X](cof)._1     def previous = unapplyCofree[F, X](cof)._2  def cofreeAlgToFAlg[F[_]: Lift, X]: Algebra[F ∘ Cofree[F]][X] => Algebra[F][Cofree[F][X]] =   alg => fcof => pureCofree(alg(fcof), fcof) // outFix[𝛎]  def histo[Fix[_[_]]: Fold, F[_]: Lift, X]: Algebra[F ∘ Cofree[F]][X] => Fix[F] => X =   cofreeAlgToFAlg(_) pipe foldFix[Fix][F][Cofree[F][X]] andThen {_.current}

Хистоморфизм может быть использован и в чистом виде, но чаще он комбинируется с анаморфизмом, образуя такую пересвёртку:

def dynamo[Fix[_[_]]: Fold: Unfold, F[_]: Lift, A, B](   coalg     : Coalgebra[F][A],   histoAlg  : Algebra[F ∘ Cofree[F]][B] ): A => B =   unfoldFix[Fix][F][A](coalg) andThen histo(histoAlg) // ana >>> histo

Продемонстрируем работу этого динамоморфизма на примере вычисления последовательности Фибоначчи:

def fibonacci: Int => Int = dynamo[μ, Option, Int, Int](   i => Option.when(i > 1)(i - 1),  // коалгебра для развёртки последовательноти натуральных чисел   _.flatMap{ last => last.previous.map(prev => last.current + prev.current) }.getOrElse(1) ) //         если ранее уже посчитаны двое, то   последнее  +  предыдущее      иначе    1  fibonacci(10) // 55 - десятое число последовательности Фибоначчи

Таким образом, хистоморфизм может быть использован при вычислении членов последовательности, порождаемых рекурсивных выражениями, когда последующие члены зависят от предыдущих. Впрочем, применительно именно к последовательности Фибоначчи удобнее оказываются параморфизм, или даже обычная свёртка (катаморфизм).

Название «динамоморфизм» обусловлено тем, что изначально он предназначался для задач динамического программирования. При решении таких задачах сперва производится декомпозиция на мелкие подзадачки (развёртка), затем они решаются последовательно, с использованием результатов уже решённых подзадач (свёртка с Cofree – хистоморфизм). На Хабре есть неплохая статья о динамическом программировании с примерами на Python.

Футуморфизм

Схема рекурсии дуальная хистоморфизму носит название «футоморфимзм». Это развёртка, в алгебре которой используются свободный контейнер:

def futuCoalgToCoalg[F[_]: Lift, X]: Coalgebra[F ∘ Free[F]][X] => Coalgebra[F][Free[F][X]] =   coalg => outFix[μ, CoEnv[X, F]] andThen {_.fold(coalg, identity)}   //                           вычисляем сейчас - ↑↑↑↑↑  ↑↑↑↑↑↑↑↑ - вычислены ранее  def futu [Fix[_[_]]: Unfold: OutFix, F[_]: Lift, X]: Coalgebra[F ∘ Free[F]][X] => X => Fix[F] =   futuCoalgToCoalg andThen unfoldFix[Fix][F][Free[F][X]] andThen {pureFree[F][X] andThen _}

Хистоморфизм позволял заглядывать в прошлое, использовать результаты предыдущих шагов вычислений. Дуальный ему футуморфизм, наоборот даёт возможность оперировать значениями, которые будут вычислены лишь в будущем!

Пожалуй главной фишкой футуморфизма является возможность преобразования списков в деревья. На практике встречается нет так много задач, для решения которых было бы полезно столь интригующая особенность футуморфизма. Часто в пример приводят синтаксические анализаторы – последовательность лексем из исходного кода анализатор преобразует в абстрактное синтаксическое дерево (AST), причём какие-то сущности могут быть объявлены «в будущем», после того, как их идентификаторы встретятся в коде.

Здесь же приведём предложенный Адамом Вандерворстом пример работы с алгебраическими графами:

enum GraphSum[Vert, Gr]:   case Empty extends GraphSum[Nothing, Nothing]   case Vertex(a: Vert)   case Overlay(x: Gr, y: Gr)   case Connect(x: Gr, y: Gr) import GraphSum.*  type GraphBase[Vert] = [Gr] =>> GraphSum[Vert, Gr] type Graph    [Vert] = μ[GraphBase[Vert]]

Каждый такой ориентированный граф может быть:

• пустым Empty;

• единственной вершиной Vertex;

• наложением Overlay графов (множества вершин и рёбер графов объединяются);

• присоединением двух графов Connect (помимо эффекта наложения добавляются ребра от каждой вершины первого графа к каждой вершине второго).

Собирать граф будем из матрицы смежности, в которой указано, как каждая вершина соединена с другими:

type AdjacencyMatrix[Vert] = Iterable[(Vert, Set[Vert])]

По сути, каждый элемент такой последовательности представляет собой маленький граф и нам остаётся лишь правильно их склеить. Каждое ребро соединяет один граф с другим, но последовательно обрабатывая список рёбер нужно учитывать, что описание других частей графа появится лишь дальше в этом списке. Для этого используем оба конструктора свободного контейнера bindFree и pureFree:

def fromAdjCoalg[V]: Coalgebra[GraphBase[V] ∘ Free[GraphBase[V]]][AdjacencyMatrix[V]] =     case (vertex, connectedGraphs) :: tail => Overlay(       bindFree(Connect(         bindFree(Vertex(vertex)),         connectedGraphs.foldLeft[Free[GraphBase[V]][AdjacencyMatrix[V]]](           bindFree(Empty))(           (childGraph, childVertex) => bindFree(Overlay(             childGraph,             bindFree(Vertex(childVertex))           ))         )       )),       pureFree(tail), // ссылаемся на части графа, которые будут вычислены В БУДУЩЕМ!     )     case Nil => Empty  def fromAdj[V]: AdjacencyMatrix[V] => Graph[V] =     futu[μ, GraphBase[V], AdjacencyMatrix[V]](fromAdjCoalg[V])

Верхнеуровневая операция в коалгебре – это наложение двух графов Overlay(bindFree(...), pureFree(tail)). Первый граф строится на основе текущей строки матрицы смежности, но вот второй аргумент – это ссылка на вообще весь граф, который ещё предстоит вычислить!

Проверять результат будем приводя граф к строке посредством обычной свёртки:

def graphToString[V] = foldFix[μ][GraphBase[V]][String]:     case Empty         => "∅"     case Vertex(a)     => a.toString     case Overlay(l, r) =>       if l == "∅" then r else // упрощение выражения     if r == "∅" then l else // упрощение выражения         s"($l + $r)"     case Connect(l, r) =>     if l == "∅" then r else // упрощение выражения     if r == "∅" then l else // упрощение выражения       s"($l -> $r)"

В итоге получаем сточку, которую легко сверить с исходной матрицей смежности:

val adjacencyMatrix = Map(     4 -> Set(3),     2 -> Set(2),     3 -> Set(),     1 -> Set(1, 3),     4 -> Set(1, 2, 5),     5 -> Set(3),   )  graphToString(fromAdj(small_adj.toSeq)) // ((5 -> 3) + ((1 -> (1 + 3)) + ((2 -> 2) + (3 + (4 -> ((1 + 2) + 5))))))

Другие примеры использования схем рекурсии для такого графа можно подсмотреть у Адама по ссылке выше.

Промежуточный итог

Все перечисленные в этом разделе манипуляции с рекурсивными типами объединяются одним общим понятием – схемы рекурсии. Но сюда также относят и другие операции, например:

• свёртка двух, или более рекурсивных структур за раз;

• пересвёртки с промежуточным преобразованием рекурсивного контейнера посередине (map, flatMap);

• преобразования вида μ[F] => μ[G] (дерево в список и т.п.).

Но все эти схемы так или иначе опираются на всё те же универсальные свойства наибольшей и наименьшей неподвижных точек конструкторов типов – функции fold и unfold.

Ещё больше о схемах рекурсии можно узнать тут:

• Recursion Schemes in Scala — An Absolutely Elementary Introduction – простой доходчивый обзор восьми схем рекурсий Цзыяна Лю;

• An introduction to recursion schemes – большой обзор от Николя Ринаудо (+слайды);

• AST playground: recursion schemes and recursive data – большой обзор от Матеуша Кубужока;

• Introduction to Recursion Schemes with Matryoshka – большая статья некого Анатолия про рекурсивные типы и использование Scala-библиотеки Matryoshka;

• Awesome Recursion Schemes – библиография;

• про хронморфизмы на Haskell:

  • Recursion Schemes, Part IV: Time is of the Essence в блоге Патрика Томпсона (см. также другие части обзора его обзора схем рекурсии тут);

  • Time Traveling Recursion Schemes – путешествия во времени с Джаредом Тобином.