Введение в веб-трансляцию

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

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

API веб-каналов постепенно становится краеугольным камнем основных веб-платформ, включая браузеры, Node.js и Deno. В этой статье мы рассмотрим, что такое веб-каналы, как они работают, их преимущества и инструменты, построенные вокруг них.

Два основных преимущества веб-каналов:

1. Мгновенная обработка данных: данные можно обрабатывать по мере их поступления, нет необходимости ждать всю полезную нагрузку. Это может значительно повысить скорость загрузки больших данных, особенно при медленном соединении.
2. Полный контроль данных: веб-каналы позволяют разработчикам считывать и обрабатывать данные в зависимости от потребностей приложений и сценариев использования.

Веб-каналы можно разделить на 3 основных типа: WritableStream, ReadableStream И TransformStream. У каждого из них своя роль:

• WritableStream: записывает данные (но не читает) в любое место с помощью «писателя»;
• ReadableStream: асинхронно читает данные (но не записывает) с помощью «читателя» (reader);
• TransformStream: Манипулирует или преобразует данные в пути с помощью «преобразователя».

Веб-потоки можно объединять в цепочки (chain) — последовательность шагов обработки данных, что повышает читаемость и «ремонтопригодность» кода, а также облегчает создание сложных конвейеров обработки данных (pipelines).

коренастый

Блоки — это основные единицы данных в веб-потоках, часто представляемые в виде строк (текстовые потоки) или Uint8Array (двоичные потоки). Кусочки могут иметь разную форму и размер в зависимости от таких факторов, как:

1. Источник данных. Если мы читаем данные из файла, файловая система может читать их блоками определенного размера, что повлияет на размер кусков.
2. Потоковая реализация. Данные можно буферизовать и отправлять большими порциями или данные можно отправлять небольшими порциями.
3. Локальная среда разработки. При локальном росте размер фрагмента не зависит от состояния сети.
4. Сеть. Размер куска может быть ограничен максимальная единица передачи (MTU). Физическое расстояние между клиентом и сервером может привести к фрагментация чанков.

Наш код должен быть готов обрабатывать куски любого размера, так как это трудно предсказать, и мы часто не можем это контролировать.

Возьмем пример.

const decoder = new TextDecoder();
const encoder = new TextEncoder();

const readableStream = new ReadableStream({
  start(controller) {
    const text = "Stream me!";
    controller.enqueue(encoder.encode(text));
    controller.close();
  },
});

const transformStream = new TransformStream({
  transform(chunk, controller) {
    const text = decoder.decode(chunk);
    controller.enqueue(encoder.encode(text.toUpperCase()));
  },
});

const writableStream = new WritableStream({
  write(chunk) {
    console.log(decoder.decode(chunk));
  },
});

readableStream
   .pipeThrough(transformStream)
   .pipeTo(writableStream); // STREAM ME!

Сначала кодируем (кодируем) строку "Stream me!" и поставить его в очередь (enqueue) в ReadableStream. Этот ReadableStream становится источником данных, который может использоваться другими потоками.

Для передачи данных из ReadableStream другой поток использует метод pipeThrough. Данные передаются TransformStreamкоторый получает блоки данных, декодирует их в строки текста, делает строку заглавной, кодирует ее и помещает преобразованный блок для использования в следующем потоке.

Данные передаются WritableStream используя метод pipeTo. На самом деле, WritableStream — это конечная точка, которая позволяет вам потреблять данные нестандартным способом. В этом случае мы декодируем блок данных и выводим его на консоль.

---

Одним из основных способов взаимодействия с потоками веб-чата является метод getReaderпредоставлено Получить API. Этот метод позволяет последовательно считывать блоки данных из тела запроса по мере их поступления, что позволяет эффективно обрабатывать большие объемы данных.

const decoder = new TextDecoder();

const response = await fetch('/api/stream');
const reader = response.body.getReader();

let done = false;

while (!done) {
  const { value, done: doneReading } = await reader.read();
  done = doneReading;
  const data = JSON.parse(decoder.decode(value));
  // Работаем с данными
}

против давления

Одной из самых мощных функций, предоставляемых веб-потоками, является встроенное управление обратным давлением, которое возникает, когда данные генерируются быстрее, чем потребляются, например, когда высокоскоростной сервер отправляет данные клиенту с медленным соединением.

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

Управление противодавлением может быть сложным из-за баланса между скоростью генерации данных и потреблением данных. Неконтролируемое создание данных может привести к проблемам с памятью из-за буферизации данных. С другой стороны, приостановка генерации данных может привести к простою их генератора, если доступны возможности обработки данных.

Управление обратным давлением с помощью веб-новостей

Веб-каналы управляют обратным давлением посредством управления потоком. Когда поток находится в состоянии readable, данные свободно передаются между производителем и потребителем. Если поступающие данные начинают превышать возможности потребителя, поток переходит в состояние backpressure. Это состояние является сигналом для генератора приостановить производство данных.

Когда потребитель освобождается, поток возвращается в состояние readable, и генерация данных продолжается. Этот автоматический механизм защищает потребителя от переполнения данных, а генератор от простоя.

В ReadableStream противодавление применяется косвенно через привод. Когда считыватель извлекает фрагменты из потока, он сигнализирует, что готов к обработке новых данных. Если накопитель перестал восстанавливать данные (метод read перестал вызываться), поток приходит к выводу, что читатель занят обработкой существующих данных и что генерация новых данных должна быть приостановлена.

В WritableStream обратное давление применяется непосредственно к писателю. Метод write возвращает обещание, которое разрешается только тогда, когда поток готов для новых данных. Таким образом, невыполненное обещание служит индикатором обратного давления.

const stream = new WritableStream(...)

async function writeData(data) {
    const writer = stream.getWriter();
    for (const chunk of data) {
        // Ждем разрешения промиса для записи следующего чанка
        await writer.ready;
        writer.write(chunk);
    }
    writer.close();
}

Использование ключевых слов await заставляет выполнение кода приостанавливаться до тех пор, пока обещание не будет разрешено. Это гарантирует, что мы не будем записывать данные быстрее, чем поток записи может их обработать.

События, отправленные сервером

События, отправленные сервером (SSE), что можно примерно перевести как «события, отправляемые сервером», — это популярный способ доставки обновлений с сервера клиенту в режиме реального времени. В то время как веб-каналы в основном используются для обработки данных, SSE поддерживает открытое соединение с сервером, позволяя передавать данные по мере их появления.

Потоки веб-чата закрывают соединение после передачи всех данных. SSE используют долгоживущее HTTP-соединение, которое может использоваться сервером для передачи новых данных. ESS может быть запрошен в приложениях, где новые данные генерируются в режиме реального времени, включая поставщиков искусственного интеллекта (искусственный интеллект — искусственный интеллект, ИИ), таких как Открытый ИИ.

Для обработки ответа SSE, содержащего обычный текст, вы можете использовать библиотеку парсер источника событий для разбора переданных фрагментированных чанков с помощью функции feed:

import { createParser } from "eventsource-parser"

export function OpenAITextStream(
  res: Response,
): ReadableStream {
  const encoder = new TextEncoder()
  const decoder = new TextDecoder()

  let counter = 0

  const stream = new ReadableStream({
    async start(controller): Promise<void> {
      function onParse(event: ParsedEvent | ReconnectInterval): void {
        if (event.type === 'event') {
          const data = event.data
          if (data === '[DONE]') {
            controller.close()
            return
          }
          try {
            const json = JSON.parse(data)
            const text =
              json.choices[0]?.delta?.content ?? json.choices[0]?.text ?? ''

            if (counter < 2 && (text.match(/\n/) || []).length) {
              return
            }

            const queue = encoder.encode(`${JSON.stringify(text)}\n`)
            controller.enqueue(queue)

            counter++
          } catch (e) {
            controller.error(e)
          }
        }
      }

      const parser = createParser(onParse)
      // [Асинхронно перебираем]( тело ответа
      for await (const chunk of res.body as any) {
        parser.feed(decoder.decode(chunk))
      }
    }
  })

  return stream
}

Полный пример можно найти в этот репозиторий.

Примечание. перевод : мы поговорим о возможностях потоковой передачи через Интернет, предоставляемых Версель. Если вы не планируете использовать эту платформу, вы можете закончить чтение статьи на этом этапе.

Поток данных в Vercel

Vercel поддерживает веб-потоки в пограничных и бессерверных средах выполнения, обеспечивая эффективную обработку данных в реальном времени. Это дает возможность «стримить» данные в формате JSON клиенту или даже постепенно рендерить части UI (user interface — пользовательский интерфейс).

export const config = {
  runtime: "edge",
};

const delay = (ms) => new Promise((res) => setTimeout(res, ms));

export default async function handler() {
  const encoder = new TextEncoder();

  const readable = new ReadableStream({
    async start(controller) {
      controller.enqueue(encoder.encode("<html><body>"));
      await delay(500);
      controller.enqueue(encoder.encode("<ul><li>List Item 1</li>"));
      await delay(500);
      controller.enqueue(encoder.encode("<li>List Item 2</li>"));
      await delay(500);
      controller.enqueue(encoder.encode("<li>List Item 3</li></ul>"));
      await delay(500);
      controller.enqueue(encoder.encode("</body></html>"));
      controller.close();
    },
  });

  return new Response(readable, {
    headers: { "Content-Type": "text/html; charset=utf-8" },
  });
}

Прогрессивный рендеринг можно использовать для улучшения взаимодействия с пользователем при медленном соединении. Тот же принцип применяется к данным JSON, когда мы хотим отправить части большого набора данных, как только они будут готовы.

Версель SDK для искусственного интеллекта

Создание пользовательского интерфейса на основе данных LLM (большая языковая модель) стало очень популярным на фоне растущей популярности поставщиков ИИ. Потоковая передача ответов LLM привела к значительным изменениям в способах разработки приложений.

Решение об использовании потоковой передачи данных зависит от таких факторов, как размер языковой модели, скорость и длина ответа. Большие модели могут генерировать сложные ответы, но реагируют медленно из-за сложных вычислений. Меньшие модели реагируют быстро, но их ответы намного проще, чего может быть недостаточно для удовлетворения потребностей приложения.

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

import { OpenAIStream, StreamingTextResponse } from 'ai'

export const runtime="edge"

export async function POST(req: Request) {
  const response = await openai.createChatCompletion({
    model: 'gpt-3.5-turbo',
    stream: true,
    messages: [...]
  })
  // Преобразуем ответ в дружелюбный текстовый поток
  const stream = OpenAIStream(response)
  // Отвечает потоком
  return new StreamingTextResponse(stream)
}

Эти потоковые ответы могут быть использованы крючки useChat И useCompletion:

'use client'

import { useChat } from 'ai'

export default function Chat() {
  const { messages, input, handleInputChange, handleSubmit } = useChat()

  return ...
}

Заключение

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

Возможность автоматического управления обратным давлением и доставки данных блоками может значительно повысить производительность приложений. Преобразование данных в процессе передачи открывает целый ряд возможностей для обработки и манипулирования данными.