Definición de flujos de trabajo de IA

El núcleo de las funciones de IA de tu app son las solicitudes de modelos generativos, pero es raro que puedas simplemente tomar la entrada del usuario, pasarla al modelo y mostrarle al usuario el resultado del modelo. Por lo general, hay pasos de procesamiento previo y posterior que deben acompañar a la llamada al modelo. Por ejemplo:

  • Cómo recuperar información contextual para enviarla con la llamada al modelo
  • Recuperar el historial de la sesión actual del usuario, por ejemplo, en una app de chat
  • Usar un modelo para cambiar el formato de la entrada del usuario de una manera adecuada para pasarla a otro modelo
  • Evaluar la "seguridad" del resultado de un modelo antes de presentarlo al usuario
  • Combinación de los resultados de varios modelos

Cada paso de este flujo de trabajo debe funcionar en conjunto para que cualquier tarea relacionada con la IA sea exitosa.

En Genkit, representas esta lógica estrechamente vinculada con una construcción llamada flujo. Los flujos se escriben de la misma manera que las funciones, con código TypeScript normal, pero agregan capacidades adicionales destinadas a facilitar el desarrollo de funciones de IA:

  • Seguridad de tipos: Esquemas de entrada y salida definidos con Zod, que proporciona verificación de tipos estáticos y de tiempo de ejecución
  • Integración con la IU para desarrolladores: Depurar flujos independientemente del código de tu aplicación con la IU para desarrolladores En la IU para desarrolladores, puedes ejecutar flujos y ver los seguimientos de cada paso del flujo.
  • Implementación simplificada: Implementa flujos directamente como extremos de API web con Cloud Functions para Firebase o cualquier plataforma que pueda alojar una app web.

A diferencia de las funciones similares en otros frameworks, los flujos de Genkit son ligeros y discretos, y no obligan a tu app a cumplir con ninguna abstracción específica. Toda la lógica del flujo está escrita en TypeScript estándar, y el código dentro de un flujo no necesita ser consciente del flujo.

Cómo definir y llamar a flujos

En su forma más simple, un flujo solo une una función. En el siguiente ejemplo, se une una función que llama a 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;
  }
);

Si unes tus llamadas a generate() de esta manera, agregas algunas funciones: esto te permite ejecutar el flujo desde la CLI de Genkit y desde la IU para desarrolladores, y es un requisito para varias de las funciones de Genkit, incluidas las funciones de observabilidad y de implementación (en secciones posteriores, se analizan estos temas).

Esquemas de entrada y salida

Una de las ventajas más importantes que tienen los flujos de Genkit en comparación con llamar directamente a una API de modelo es la seguridad de tipos de entrada y salida. Cuando definas flujos, podrás definir esquemas para ellos con Zod, de la misma manera que defines el esquema de salida de una llamada a generate(). Sin embargo, a diferencia de generate(), también puedes especificar un esquema de entrada.

Este es un perfeccionamiento del último ejemplo, que define un flujo que toma una cadena como entrada y genera un objeto:

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;
  }
);

Ten en cuenta que el esquema de un flujo no tiene que alinearse necesariamente con el esquema de las llamadas a generate() dentro del flujo (de hecho, un flujo ni siquiera puede contener llamadas a generate()). Esta es una variación del ejemplo que pasa un esquema a generate(), pero usa el resultado estructurado para dar formato a una cadena simple, que muestra el flujo.

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}`;
  }
);

Flujos de llamadas

Una vez que hayas definido un flujo, puedes llamarlo desde tu código de Node.js:

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

El argumento del flujo debe cumplir con el esquema de entrada, si definiste uno.

Si definiste un esquema de salida, la respuesta del flujo se ajustará a él. Por ejemplo, si configuras el esquema de salida en MenuItemSchema, el resultado del flujo contendrá sus propiedades:

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

Flujos de transmisión

Los flujos admiten la transmisión a través de una interfaz similar a la interfaz de transmisión de generate(). La transmisión es útil cuando tu flujo genera una gran cantidad de resultados, ya que puedes presentarlos al usuario a medida que se generan, lo que mejora la capacidad de respuesta percibida de tu app. Como ejemplo conocido, las interfaces de LLM basadas en chat suelen transmitir sus respuestas al usuario a medida que se generan.

Este es un ejemplo de un flujo que admite la transmisión:

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,
    };
  }
);
  • La opción streamSchema especifica el tipo de valores que transmite tu flujo. No es necesario que sea del mismo tipo que outputSchema, que es el tipo del resultado completo del flujo.
  • streamingCallback es una función de devolución de llamada que toma un solo parámetro, del tipo que especifica streamSchema. Cada vez que los datos estén disponibles en tu flujo, llama a esta función para enviarlos al flujo de salida. Ten en cuenta que streamingCallback solo se define si el llamador de tu flujo solicitó una salida de transmisión, por lo que debes verificar que esté definido antes de llamarlo.

En el ejemplo anterior, los valores que transmite el flujo se acoplan directamente a los valores que transmite la llamada a generate() dentro del flujo. Si bien esto suele ser así, no tiene por qué serlo: puedes enviar valores al flujo con la devolución de llamada con la frecuencia que sea útil para tu flujo.

Llamadas a flujos de transmisión

Los flujos de transmisión también se pueden llamar, pero muestran de inmediato un objeto de respuesta en lugar de una promesa:

const response = menuSuggestionStreamingFlow('Danube');

El objeto de respuesta tiene una propiedad de flujo, que puedes usar para iterar sobre el resultado de transmisión del flujo a medida que se genera:

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

También puedes obtener el resultado completo del flujo, como puedes hacerlo con un flujo que no es de transmisión:

const output = await response.output;

Ten en cuenta que el resultado de transmisión continua de un flujo puede no ser del mismo tipo que el resultado completo. El resultado de transmisión continua cumple con streamSchema, mientras que el resultado completo cumple con outputSchema.

Ejecuta flujos desde la línea de comandos

Puedes ejecutar flujos desde la línea de comandos con la herramienta de CLI de Genkit:

genkit flow:run menuSuggestionFlow '"French"'

Para los flujos de transmisión, puedes imprimir el resultado de la transmisión en la consola si agregas la marca -s:

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

Ejecutar un flujo desde la línea de comandos es útil para probarlo o para ejecutar flujos que realizan tareas necesarias de forma ad hoc, por ejemplo, para ejecutar un flujo que transfiera un documento a tu base de datos de vectores.

Flujos de depuración

Una de las ventajas de encapsular la lógica de IA dentro de un flujo es que puedes probar y depurar el flujo de forma independiente de tu app con la IU para desarrolladores de Genkit.

Para iniciar la IU para desarrolladores, ejecuta los siguientes comandos desde el directorio de tu proyecto:

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

En la pestaña Run de la IU para desarrolladores, puedes ejecutar cualquiera de los flujos definidos en tu proyecto:

Captura de pantalla del ejecutor de flujo

Después de ejecutar un flujo, puedes inspeccionar un registro de la invocación del flujo haciendo clic en Ver registro o en la pestaña Inspeccionar.

En el visor de seguimiento, puedes ver detalles sobre la ejecución de todo el flujo, así como detalles de cada uno de los pasos individuales dentro del flujo. Por ejemplo, considera el siguiente flujo, que contiene varias solicitudes de generación:

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;
  }
);

Cuando ejecutas este flujo, el visor de seguimiento te muestra detalles sobre cada solicitud de generación, incluido su resultado:

Captura de pantalla del inspector de seguimiento

Pasos del flujo

En el último ejemplo, viste que cada llamada a generate() aparecía como un paso independiente en el visor de seguimiento. Cada una de las acciones fundamentales de Genkit aparece como pasos separados de un flujo:

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

Si deseas incluir código distinto del anterior en tus seguimientos, puedes unirlo en una llamada a run(). Puedes hacerlo para llamadas a bibliotecas de terceros que no sean compatibles con Genkit o para cualquier sección crítica de código.

Por ejemplo, este es un flujo con dos pasos: el primer paso recupera un menú con un método no especificado y el segundo paso incluye el menú como contexto para una llamada a 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;
  }
);

Como el paso de recuperación está unido en una llamada a run(), se incluye como un paso en el visor de seguimiento:

Captura de pantalla de un paso definido de forma explícita en el inspector de seguimiento

Implementa flujos

Puedes implementar tus flujos directamente como extremos de API web, listos para que los llames desde los clientes de tu app. La implementación se analiza en detalle en varias otras páginas, pero en esta sección se proporcionan breves descripciones generales de tus opciones de implementación.

Cloud Functions para Firebase

Para implementar flujos con Cloud Functions para Firebase, usa el complemento firebase. En las definiciones de flujo, reemplaza defineFlow por onFlow y, luego, incluye un 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) => {
    // ...
  }
);

Si deseas obtener más información, consulta las siguientes páginas:

Express.js

Para implementar flujos con cualquier plataforma de hosting de Node.js, como Cloud Run, define tus flujos con defineFlow() y, luego, llama a startFlowServer():

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

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

De forma predeterminada, startFlowServer publicará todos los flujos definidos en tu base de código como extremos HTTP (por ejemplo, http://localhost:3400/menuSuggestionFlow). Puedes llamar a un flujo con una solicitud POST de la siguiente manera:

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

Si es necesario, puedes personalizar el servidor de flujos para que entregue una lista específica de flujos, como se muestra a continuación. También puedes especificar un puerto personalizado (usará la variable de entorno PORT si está configurada) o especificar la configuración de 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: '*',
  },
});

Para obtener información sobre la implementación en plataformas específicas, consulta Implementa con Cloud Run y Implementa flujos en cualquier plataforma de Node.js.