Definiowanie przepływów pracy z użyciem AI

Podstawą funkcji AI w aplikacji są żądania dotyczące modeli generatywnych, ale rzadko zdarza się, aby można było po prostu pobrać dane wejściowe od użytkownika, przekazać je do modelu i wyświetlić wyniki. Zazwyczaj przed wywołaniem modelu i po nim należy wykonać czynności wstępne i końcowe. Przykład:

  • Pobieranie informacji kontekstowych do wysłania wraz z wywołaniem modelu
  • Pobieranie historii bieżącej sesji użytkownika, np. w aplikacji do czatu
  • Użycie jednego modelu do przeformatowania danych wejściowych użytkownika w sposób odpowiedni do przekazania do innego modelu.
  • Ocenianie „bezpieczeństwa” danych wyjściowych modelu przed ich przedstawieniem użytkownikowi.
  • Łączenie wyników kilku modeli

Aby zadanie związane z AI zakończyło się sukcesem, wszystkie kroki tego procesu muszą ze sobą współdziałać.

W Genkit logikę o ścisłych powiązaniach reprezentujesz za pomocą konstrukcji zwanej przepływem. Przepływy są tworzone tak samo jak funkcje, czyli przy użyciu zwykłego kodu TypeScript, ale oferują dodatkowe możliwości ułatwiające tworzenie funkcji AI:

  • Bezpieczeństwo typów: schematy danych wejściowych i wyjściowych zdefiniowane za pomocą Zod, które zapewniają sprawdzanie typów zarówno w czasie statycznego, jak i w czasie wykonywania.
  • Integracja z interfejsem dla deweloperów: debugowanie procesów niezależnie od kodu aplikacji za pomocą interfejsu dla deweloperów. W interfejsie dla deweloperów możesz uruchamiać przepływy i wyświetlać ścieżki poszczególnych kroków przepływu.
  • Uproszczone wdrażanie: wdróż przepływy bezpośrednio jako punkty końcowe interfejsu API internetowego za pomocą funkcji Cloud Functions for Firebase lub dowolnej platformy, która może hostować aplikację internetową.

W odróżnieniu od podobnych funkcji w innych frameworkach przepływy Genkit są lekkie i nieinwazyjne, a Twoja aplikacja nie musi być zgodna z żadną konkretną abstrakcją. Cała logika przepływu jest napisana w standardowym TypeScriptie, a kod wewnątrz przepływu nie musi być świadomy przepływu.

Definiowanie i wywoływanie przepływów

W najprostszej postaci przepływ danych tylko otacza funkcję. W tym przykładzie funkcja, która wywołuje funkcję generate(), jest opakowana: 

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

Wystarczy, że spróbujesz umieścić wywołania generate() w ten sposób, aby dodać pewną funkcjonalność: dzięki temu możesz uruchomić przepływ z Genkit CLI i z interfejsu dla deweloperów. Jest to wymagane w przypadku kilku funkcji Genkit, w tym wdrażania i obserwowalności (te tematy omawiamy w późniejszych sekcjach).

Schematy danych wejściowych i wyjściowych

Jedną z najważniejszych zalet przepływów Genkit w porównaniu z bezpośrednim wywołaniem interfejsu API modelu jest bezpieczeństwo typów danych wejściowych i wyjściowych. Podczas definiowania przepływów możesz definiować dla nich schematy za pomocą Zod w podobny sposób jak schemat wyjściowy wywołania generate(). W odróżnieniu od generate() możesz jednak określić też schemat wejściowy.

Oto dopracowany przykład z ostatniego przykładu, który definiuje przepływ, przyjmujący na wejściu ciąg znaków, a na wyjściu zwracający obiekt:

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

Pamiętaj, że schemat przepływu nie musi być zgodny ze schematem wywołań generate() w ramach przepływu (przepływ może nawet nie zawierać wywołań generate()). Oto wariant przykładu, który przekazuje schema do generate(), ale używa uporządkowanych danych do sformatowania prostego ciągu znaków, który zwraca przepływ.

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

Procesy wywoływania

Po zdefiniowaniu przepływu możesz go wywołać z kodu Node.js:

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

Argument przepływu musi być zgodny ze schematem danych wejściowych (jeśli go zdefiniujesz).

Jeśli zdefiniujesz schemat wyjściowy, odpowiedź przepływu będzie się do niego stosować. Jeśli na przykład ustawisz schemat wyjściowy na MenuItemSchema, dane wyjściowe przepływu będą zawierać jego właściwości:

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

Przepływy strumieniowe

Procesy obsługują strumieniowanie za pomocą interfejsu podobnego do interfejsu strumieniowania generate(). Transmisja strumieniowa jest przydatna, gdy przepływ generuje dużą ilość danych wyjściowych, ponieważ możesz wyświetlać dane wyjściowe użytkownikowi w miarę ich generowania, co zwiększa postrzeganą szybkość reakcji aplikacji. Przykładem jest interfejs LLM oparty na czacie, który często przesyła użytkownikowi odpowiedzi w miarę ich generowania.

Oto przykład przepływu, który obsługuje strumieniowanie:

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,
    };
  }
);
  • Opcja streamSchema określa typ wartości przepływów danych. Nie musi to być ten sam typ co outputSchema, czyli typ pełnego wyjścia przepływu.
  • streamingCallback to funkcja wywołania zwrotnego, która przyjmuje 1 parametr typu określonego przez streamSchema. Za każdym razem, gdy dane staną się dostępne w Twoim przepływie, wysyłaj je do strumienia wyjściowego, wywołując tę funkcję. Pamiętaj, że streamingCallback jest zdefiniowany tylko wtedy, gdy wywołujący Twój przepływ danych poprosił o wyjście strumieniowe, więc przed wywołaniem musisz sprawdzić, czy jest zdefiniowany.

W powyższym przykładzie wartości przesyłane przez przepływ są bezpośrednio powiązane z wartościami przesyłanymi przez wywołanie funkcji generate() wewnątrz przepływu. Chociaż tak się często dzieje, nie musi tak być: możesz przekazywać wartości do strumienia za pomocą funkcji wywołania zwrotnego tak często, jak jest to przydatne w Twoim przypadku.

Przebiegi strumieniowania połączeń

Strumieniowe przepływy danych można też wywoływać, ale zwracają one od razu obiekt odpowiedzi, a nie obietnicę:

const response = menuSuggestionStreamingFlow('Danube');

Obiekt odpowiedzi ma właściwość stream, której możesz użyć do iteracji nad strumieniem danych generowanym przez przepływ:

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

Podobnie jak w przypadku strumienia nieprzerywanego, możesz też uzyskać pełny wynik przepływu:

const output = await response.output;

Pamiętaj, że dane wyjściowe strumieniowe przepływu mogą być innego typu niż dane wyjściowe pełne. Dane wyjściowe strumieniowe są zgodne z streamSchema, a dane wyjściowe pełne są zgodne z outputSchema.

Uruchamianie procesów z poziomu wiersza poleceń

Możesz uruchamiać przepływy z poziomu wiersza poleceń za pomocą narzędzia wiersza poleceń Genkit:

genkit flow:run menuSuggestionFlow '"French"'

W przypadku przepływów strumieniowych możesz wydrukować dane wyjściowe strumienia na konsoli, dodając flagę -s:

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

Uruchamianie przepływu z poziomu wiersza poleceń jest przydatne do testowania przepływu lub do uruchamiania przepływów, które wykonują zadania potrzebne w razie potrzeby – na przykład do uruchamiania przepływu, który przetwarza dokument w bazie danych wektorów.

Debugowanie przepływów

Jedną z zalet umieszczania logiki AI w ramach przepływu jest możliwość testowania i debugowania przepływu niezależnie od aplikacji za pomocą interfejsu programisty Genkit.

Aby uruchomić interfejs dla deweloperów, uruchom te polecenia w katalogu projektu:

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

Na karcie Uruchom w interfejsie dla programistów możesz uruchomić dowolny z procesów zdefiniowanych w Twoim projekcie:

Zrzut ekranu pokazujący proces wykonywania

Po uruchomieniu przepływu możesz sprawdzić ślad wywołania przepływu, klikając Wyświetl ślad lub korzystając z karty Sprawdź.

W przeglądarce śladów możesz zobaczyć szczegóły dotyczące wykonania całego procesu, a także szczegóły dotyczące poszczególnych kroków w tym procesie. Rozważ na przykład ten proces, który zawiera kilka żądań generowania:

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

Gdy uruchomisz ten proces, w przeglądarce śladów zobaczysz szczegóły dotyczące każdego żądania generowania, w tym jego dane wyjściowe:

Zrzut ekranu pokazujący inspekcję śledzenia

Etapy przepływu

W ostatnim przykładzie każda wywołana funkcja generate() była widoczna w przeglądarce śladów jako osobny krok. Każde z podstawowych działań w Genkit pojawia się jako osobny krok w przebiegu:

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

Jeśli chcesz uwzględnić w śladach kod inny niż wymieniony powyżej, możesz to zrobić, otaczając go wywołaniem run(). Możesz to zrobić w przypadku wywołań bibliotek zewnętrznych, które nie są obsługiwane przez Genkit, lub dowolnej krytycznej sekcji kodu.

Oto np. przepływ danych z 2 krokami: pierwszy krok pobiera menu za pomocą nieokreślonej metody, a drugi zawiera menu jako kontekst wywołania funkcjigenerate().

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

Ponieważ krok wyszukiwania jest opakowany w wywołanie run(), jest on uwzględniony jako krok w przeglądarce śladu:

Zrzut ekranu przedstawiający wyraźnie zdefiniowany krok w inspektorze śladu

Wdrażanie przepływów

Możesz wdrażać przepływy bezpośrednio jako punkty końcowe interfejsu API sieci Web, które możesz wywoływać z klientów aplikacji. Wdrożenie jest szczegółowo omawiane na kilku innych stronach, ale w tej sekcji znajdziesz krótkie omówienie opcji wdrożenia.

Cloud Functions dla Firebase

Aby wdrażać przepływy za pomocą Cloud Functions dla Firebase, użyj wtyczki firebase. W definicjach przepływów zastąp defineFlow wartością onFlow i dodaj parametr 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) => {
    // ...
  }
);

Więcej informacji znajdziesz na tych stronach:

Express.js

Aby wdrażać przepływy za pomocą dowolnej platformy hostingowej Node.js, np. Cloud Run, zdefiniuj przepływy za pomocą defineFlow(), a następnie wywołaj startFlowServer():

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

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

Domyślnie startFlowServer będzie obsługiwać wszystkie przepływy zdefiniowane w kodzie źródłowym jako punkty końcowe HTTP (np. http://localhost:3400/menuSuggestionFlow). Możesz wywołać przepływ z żądaniem POST w ten sposób:

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

W razie potrzeby możesz dostosować serwer przepływów, aby wyświetlał określoną listę przepływów, jak pokazano poniżej. Możesz też podać niestandardowy port (użyje on zmiennej środowiskowej PORT, jeśli jest ustawiona) lub podać ustawienia 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: '*',
  },
});

Informacje o wdrażaniu na konkretne platformy znajdziesz w artykułach Wdrażanie za pomocą Cloud Run i Wdrażanie przepływów na dowolnej platformie Node.js.