흐름

흐름은 추가 특성이 있는 함수입니다. 강타입(strongly typed), 스트리밍이 가능하고, 로컬 및 원격으로 호출 가능하며, 완전히 관찰할 수 있습니다. Firebase Genkit은 흐름 작업(실행, 디버깅 등)을 위한 CLI 및 개발자 UI 도구를 제공합니다.

흐름 정의

import { defineFlow } from '@genkit-ai/flow';

export const menuSuggestionFlow = defineFlow(
  {
    name: 'menuSuggestionFlow',
  },
  async (restaurantTheme) => {
    const suggestion = makeMenuItemSuggestion(restaurantTheme);

    return suggestion;
  }
);

흐름의 입력 및 출력 스키마는 zod를 사용하여 정의할 수 있습니다.

import { defineFlow } from '@genkit-ai/flow';
import * as z from 'zod';

export const menuSuggestionFlow = defineFlow(
  {
    name: 'menuSuggestionFlow',
    inputSchema: z.string(),
    outputSchema: z.string(),
  },
  async (restaurantTheme) => {
    const suggestion = makeMenuItemSuggestion(input.restaurantTheme);

    return suggestion;
  }
);

스키마가 지정되면 Genkit가 입력 및 출력 스키마의 유효성을 검사합니다.

실행 중인 흐름

runFlow 함수를 사용하여 흐름을 실행합니다.

const response = await runFlow(menuSuggestionFlow, 'French');

CLI를 사용하여 흐름을 실행할 수도 있습니다.

genkit flow:run menuSuggestionFlow '"French"'

스트리밍됨

다음은 흐름에서 값을 스트리밍할 수 있는 흐름의 간단한 예입니다.

export const menuSuggestionFlow = defineFlow(
  {
    name: 'menuSuggestionFlow',
    streamSchema: z.string(),
  },
  async (restaurantTheme, streamingCallback) => {
    if (streamingCallback) {
      makeMenuItemSuggestionsAsync(restaurantTheme).subscribe((suggestion) => {
        streamingCallback(suggestion);
      });
    }
  }
);

streamingCallback는 정의되지 않을 수 있습니다. 이는 호출 클라이언트가 스트리밍된 응답을 요청하는 경우에만 정의됩니다.

스트리밍 모드에서 흐름을 호출하려면 streamFlow 함수를 사용합니다.

const response = streamFlow(menuSuggestionFlow, 'French');

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

흐름에서 스트리밍을 구현하지 않는 경우 streamFlow는 runFlow와 동일하게 작동합니다.

CLI를 사용하여 흐름을 스트리밍할 수도 있습니다.

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

흐름 배포

HTTP를 통해 흐름에 액세스하려면 먼저 흐름을 배포해야 합니다. Genkit는 Firebase용 Cloud Functions 및 Cloud Run과 같은 Express.js 호스트에 대한 통합을 제공합니다.

배포된 흐름은 로컬 흐름과 동일한 기능 (예: 스트리밍 및 관측 가능성)을 모두 지원합니다.

Firebase용 Cloud 함수

Firebase용 Cloud Functions에서 흐름을 사용하려면 firebase 플러그인을 사용하고 defineFlow를 onFlow로 바꾸고 authPolicy를 포함합니다.

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

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

Express.js

Cloud Run 및 유사한 서비스를 사용하여 흐름을 배포하려면 defineFlow를 사용하여 흐름을 정의한 후 startFlowsServer()를 호출합니다.

import { defineFlow, startFlowsServer } from '@genkit-ai/flow';

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

startFlowsServer();

기본적으로 startFlowsServer는 코드베이스에서 정의한 모든 흐름을 HTTP 엔드포인트 (예: http://localhost:3400/menuSuggestionFlow)로 제공합니다.

흐름 서버를 통해 노출할 흐름을 선택할 수 있습니다. 커스텀 포트를 지정할 수 있습니다 (설정된 경우 PORT 환경 변수가 사용됨). CORS 설정을 지정할 수도 있습니다.

import { defineFlow, startFlowsServer } from '@genkit-ai/flow';

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

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

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

흐름 관측 가능성

관측 가능성을 위해 계측되지 않은 서드 파티 SDK를 사용하는 경우 개발자 UI에서 별도의 트레이스 단계로 보고 싶은 경우가 있습니다. run 함수에 코드를 래핑하기만 하면 됩니다.

import { defineFlow, run } from '@genkit-ai/flow';

export const menuSuggestionFlow = defineFlow(
  {
    name: 'menuSuggestionFlow',
    outputSchema: z.array(s.string()),
  },
  async (restaurantTheme) => {
    const themes = await run('find-similar-themes', async () => {
      return await findSimilarRestaurantThemes(restaurantTheme);
    });

    const suggestions = makeMenuItemSuggestions(themes);

    return suggestions;
  }
);