Определение рабочих процессов ИИ

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

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

Каждый шаг этого рабочего процесса должен работать вместе, чтобы любая задача, связанная с ИИ, была успешной.

В Genkit вы представляете эту тесно связанную логику с помощью конструкции, называемой потоком. Потоки пишутся так же, как и функции, с использованием обычного кода TypeScript, но они добавляют дополнительные возможности, призванные облегчить разработку функций ИИ:

• Безопасность типов : схемы ввода и вывода, определенные с помощью Zod, который обеспечивает как статическую проверку типов, так и проверку типов во время выполнения.
• Интеграция с пользовательским интерфейсом разработчика . Отладка выполняется независимо от кода вашего приложения с использованием пользовательского интерфейса разработчика. В пользовательском интерфейсе разработчика вы можете запускать потоки и просматривать трассировки для каждого шага потока.
• Упрощенное развертывание . Развертывайте потоки непосредственно в качестве конечных точек веб-API, используя Cloud Functions для Firebase или любую платформу, на которой может размещаться веб-приложение.

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

Определение и вызов потоков

В своей простейшей форме поток просто оборачивает функцию. В следующем примере обертывается функция, которая вызывает generate() :

export const menuSuggestionFlow = ai.defineFlow(
  {
    name: 'menuSuggestionFlow',
  },
  async (restaurantTheme) => {
    const { text } = await ai.generate({
      model: gemini15Flash,
      prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
    });
    return text;
  }
);

Просто обертывая generate() таким образом, вы добавляете некоторую функциональность: это позволяет запускать поток из CLI Genkit и из пользовательского интерфейса разработчика, а также является требованием для некоторых функций Genkit, включая развертывание и возможность наблюдения (последующие разделы). обсудить эти темы).

Схемы ввода и вывода

Одним из наиболее важных преимуществ потоков Genkit по сравнению с прямым вызовом API модели является безопасность типов как входных, так и выходных данных. При определении потоков вы можете определить для них схемы с помощью Zod, почти так же, как вы определяете схему вывода вызова generate() ; однако, в отличие от generate() , вы также можете указать входную схему.

Вот уточнение последнего примера, который определяет поток, который принимает строку в качестве входных данных и выводит объект:

const MenuItemSchema = z.object({
  dishname: z.string(),
  description: z.string(),
});

export const menuSuggestionFlowWithSchema = ai.defineFlow(
  {
    name: 'menuSuggestionFlow',
    inputSchema: z.string(),
    outputSchema: MenuItemSchema,
  },
  async (restaurantTheme) => {
    const { output } = await ai.generate({
      model: gemini15Flash,
      prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
      output: { schema: MenuItemSchema },
    });
    if (output == null) {
      throw new Error("Response doesn't satisfy schema.");
    }
    return output;
  }
);

Обратите внимание, что схема потока не обязательно должна совпадать со схемой вызовов метода generate() внутри потока (фактически поток может даже не содержать вызовов generate() ). Ниже приведен вариант примера, который передает схему в generate() , но использует структурированный вывод для форматирования простой строки, которую возвращает поток.

export const menuSuggestionFlowMarkdown = ai.defineFlow(
  {
    name: 'menuSuggestionFlow',
    inputSchema: z.string(),
    outputSchema: z.string(),
  },
  async (restaurantTheme) => {
    const { output } = await ai.generate({
      model: gemini15Flash,
      prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
      output: { schema: MenuItemSchema },
    });
    if (output == null) {
      throw new Error("Response doesn't satisfy schema.");
    }
    return `**${output.dishname}**: ${output.description}`;
  }
);

Потоки вызовов

Определив поток, вы можете вызвать его из кода Node.js:

const { text } = await menuSuggestionFlow('bistro');

Аргумент потока должен соответствовать входной схеме, если вы ее определили.

Если вы определили схему вывода, ответ потока будет соответствовать ей. Например, если вы установите выходную схему MenuItemSchema , выходные данные потока будут содержать ее свойства:

const { dishname, description } =
  await menuSuggestionFlowWithSchema('bistro');

Потоковые потоки

Потоки поддерживают потоковую передачу, используя интерфейс, аналогичный интерфейсу потоковой передачи generate() . Потоковая передача полезна, когда ваш поток генерирует большой объем выходных данных, поскольку вы можете представить выходные данные пользователю по мере их создания, что улучшает восприятие реакции вашего приложения. В качестве знакомого примера можно привести интерфейсы LLM на основе чата, которые часто передают пользователю ответы по мере их создания.

Вот пример потока, поддерживающего потоковую передачу:

export const menuSuggestionStreamingFlow = ai.defineStreamingFlow(
  {
    name: 'menuSuggestionFlow',
    inputSchema: z.string(),
    streamSchema: z.string(),
    outputSchema: z.object({ theme: z.string(), menuItem: z.string() }),
  },
  async (restaurantTheme, streamingCallback) => {
    const response = await ai.generateStream({
      model: gemini15Flash,
      prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
    });

    if (streamingCallback) {
      for await (const chunk of response.stream) {
        // Here, you could process the chunk in some way before sending it to
        // the output stream via streamingCallback(). In this example, we output
        // the text of the chunk, unmodified.
        streamingCallback(chunk.text);
      }
    }

    return {
      theme: restaurantTheme,
      menuItem: (await response.response).text,
    };
  }
);

• streamSchema указывает тип значений потоков вашего потока. Это не обязательно должен быть тот же тип, что и outputSchema , который является типом полного вывода потока.
• streamingCallback — это функция обратного вызова, которая принимает один параметр типа, указанного streamSchema . Всякий раз, когда данные становятся доступными в вашем потоке, отправьте их в выходной поток, вызвав эту функцию. Обратите внимание, streamingCallback определяется только в том случае, если вызывающая сторона вашего потока запросила потоковый вывод, поэтому вам необходимо проверить, что он определен, прежде чем вызывать его.

В приведенном выше примере значения, передаваемые потоком, напрямую связаны со значениями, передаваемыми generate() внутри потока. Хотя это часто бывает, это не обязательно: вы можете выводить значения в поток, используя обратный вызов, так часто, как это полезно для вашего потока.

Вызов потоковых потоков

Потоковые потоки также можно вызывать, но они немедленно возвращают объект ответа, а не обещание:

const response = menuSuggestionStreamingFlow('Danube');

У объекта ответа есть свойство потока, которое можно использовать для перебора потокового вывода потока по мере его создания:

for await (const chunk of response.stream) {
  console.log('chunk', chunk);
}

Вы также можете получить полный вывод потока, как и в случае с непотоковым потоком:

const output = await response.output;

Обратите внимание, что потоковый вывод потока может отличаться от типа полного вывода; потоковый вывод streamSchema , тогда как полный вывод соответствует outputSchema .

Запуск потоков из командной строки

Вы можете запускать потоки из командной строки с помощью инструмента Genkit CLI:

genkit flow:run menuSuggestionFlow '"French"'

Для потоков потоковой передачи вы можете вывести выходные данные потоковой передачи на консоль, добавив флаг -s :

genkit flow:run menuSuggestionFlow '"French"' -s

Запуск потока из командной строки полезен для тестирования потока или для запуска потоков, выполняющих задачи, необходимые на разовой основе — например, для запуска потока, который принимает документ в базу данных векторов.

Отладка потоков

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

Чтобы запустить пользовательский интерфейс разработчика, выполните следующие команды из каталога вашего проекта:

genkit start -- tsx --watch src/your-code.ts

На вкладке «Выполнить» пользовательского интерфейса разработчика вы можете запустить любой из потоков, определенных в вашем проекте:

После запуска потока вы можете проверить трассировку вызова потока, щелкнув «Просмотреть трассировку» или просмотрев вкладку «Проверка» .

В средстве просмотра трассировки вы можете просмотреть подробные сведения о выполнении всего потока, а также сведения о каждом отдельном шаге потока. Например, рассмотрим следующий поток, который содержит несколько запросов на создание:

const PrixFixeMenuSchema = z.object({
  starter: z.string(),
  soup: z.string(),
  main: z.string(),
  dessert: z.string(),
});

export const complexMenuSuggestionFlow = ai.defineFlow(
  {
    name: 'complexMenuSuggestionFlow',
    inputSchema: z.string(),
    outputSchema: PrixFixeMenuSchema,
  },
  async (theme: string): Promise<z.infer<typeof PrixFixeMenuSchema>> => {
    const chat = ai.chat({ model: gemini15Flash });
    await chat.send('What makes a good prix fixe menu?');
    await chat.send(
      'What are some ingredients, seasonings, and cooking techniques that ' +
        `would work for a ${theme} themed menu?`
    );
    const { output } = await chat.send({
      prompt:
        `Based on our discussion, invent a prix fixe menu for a ${theme} ` +
        'themed restaurant.',
      output: {
        schema: PrixFixeMenuSchema,
      },
    });
    if (!output) {
      throw new Error('No data generated.');
    }
    return output;
  }
);

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

Этапы потока

В последнем примере вы видели, что каждый вызов generate() отображается как отдельный шаг в средстве просмотра трассировки. Каждое из фундаментальных действий Genkit отображается как отдельные шаги потока:

• generate()
• Chat.send()
• embed()
• index()
• retrieve()

Если вы хотите включить в свои трассировки код, отличный от приведенного выше, вы можете сделать это, обернув код в вызов run() . Вы можете сделать это для вызовов сторонних библиотек, не поддерживающих Genkit, или для любого критического раздела кода.

Например, вот поток, состоящий из двух шагов: первый шаг извлекает меню, используя некоторый неопределенный метод, а второй шаг включает меню в качестве контекста для вызова метода generate() .

export const menuQuestionFlow = ai.defineFlow(
  {
    name: 'menuQuestionFlow',
    inputSchema: z.string(),
    outputSchema: z.string(),
  },
  async (input: string): Promise<string> => {
    const menu = await run('retrieve-daily-menu', async (): Promise<string> => {
      // Retrieve today's menu. (This could be a database access or simply
      // fetching the menu from your website.)

      // ...

      return menu;
    });
    const { text } = await ai.generate({
      model: gemini15Flash,
      system: "Help the user answer questions about today's menu.",
      prompt: input,
      docs: [{ content: [{ text: menu }] }],
    });
    return text;
  }
);

Поскольку шаг извлечения заключен в вызов run() , он включается как шаг в средство просмотра трассировки:

Развертывание потоков

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

Облачные функции для Firebase

Чтобы развернуть потоки с помощью Cloud Functions для Firebase, используйте плагин firebase . В определениях потоков замените defineFlow на onFlow и включите authPolicy .

import { firebaseAuth } from '@genkit-ai/firebase/auth';
import { onFlow } from '@genkit-ai/firebase/functions';

export const menuSuggestion = onFlow(
  ai,
  {
    name: 'menuSuggestionFlow',
    authPolicy: firebaseAuth((user) => {
      if (!user.email_verified) {
        throw new Error('Verified email required to run flow');
      }
    }),
  },
  async (restaurantTheme) => {
    // ...
  }
);

Для получения дополнительной информации см. следующие страницы:

• Развертывание с помощью Firebase
• Авторизация и целостность
• Плагин Firebase

Экспресс.js

Чтобы развернуть потоки с помощью любой хостинговой платформы Node.js, например Cloud Run, определите свои потоки с помощью defineFlow() , а затем вызовите startFlowServer() :

export const menuSuggestionFlow = ai.defineFlow(
  {
    name: 'menuSuggestionFlow',
  },
  async (restaurantTheme) => {
    // ...
  }
);

ai.startFlowServer({
  flows: [menuSuggestionFlow],
});

По умолчанию startFlowServer будет обслуживать все потоки, определенные в вашей кодовой базе, как конечные точки HTTP (например, http://localhost:3400/menuSuggestionFlow ). Вы можете вызвать поток с помощью POST-запроса следующим образом:

curl -X POST "http://localhost:3400/menuSuggestionFlow" \
  -H "Content-Type: application/json"  -d '{"data": "banana"}'

При необходимости вы можете настроить сервер потоков для обслуживания определенного списка потоков, как показано ниже. Вы также можете указать собственный порт (он будет использовать переменную среды PORT, если она установлена) или указать настройки CORS.

export const flowA = ai.defineFlow({ name: 'flowA' }, async (subject) => {
  // ...
});

export const flowB = ai.defineFlow({ name: 'flowB' }, async (subject) => {
  // ...
});

ai.startFlowServer({
  flows: [flowB],
  port: 4567,
  cors: {
    origin: '*',
  },
});

Информацию о развертывании на конкретных платформах см. в разделах «Развертывание с помощью Cloud Run» и «Потоки развертывания на любой платформе Node.js» .