AI 워크플로 정의

앱의 AI 기능의 핵심은 생성형 모델 요청이지만 단순히 사용자 입력을 받아 모델에 전달하고 모델 출력을 사용자에게 다시 표시할 수 있는 경우는 드뭅니다. 일반적으로 모델 호출과 함께 사전 처리 및 사후 처리 단계가 있습니다. 예를 들면 다음과 같습니다.

• 모델 호출과 함께 전송할 컨텍스트 정보 가져오기
• 사용자의 현재 세션 기록을 가져옵니다(예: 채팅 앱).
• 한 모델을 사용하여 다른 모델에 전달하는 데 적합한 방식으로 사용자 입력의 형식을 변경합니다.
• 사용자에게 모델의 출력을 표시하기 전에 모델 출력의 '안전'을 평가합니다.
• 여러 모델의 출력 결합

AI 관련 작업이 성공하려면 이 워크플로의 모든 단계가 함께 작동해야 합니다.

Genkit에서는 흐름이라는 구성을 사용하여 이러한 긴밀하게 연결된 로직을 나타냅니다. 흐름은 일반 TypeScript 코드를 사용하여 함수와 마찬가지로 작성되지만 AI 기능의 개발을 용이하게 하기 위한 추가 기능이 추가됩니다.

• 유형 안전성: 정적 및 런타임 유형 검사를 모두 제공하는 Zod를 사용하여 정의된 입력 및 출력 스키마
• 개발자 UI와 통합: 개발자 UI를 사용하여 애플리케이션 코드와는 별개로 흐름을 디버그합니다. 개발자 UI에서 흐름을 실행하고 흐름의 각 단계에 대한 트레이스를 볼 수 있습니다.
• 간소화된 배포: Firebase용 Cloud Functions 또는 웹 앱을 호스팅할 수 있는 플랫폼을 사용하여 흐름을 웹 API 엔드포인트로 직접 배포합니다.

다른 프레임워크의 유사한 기능과 달리 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() 호출을 이와 같이 래핑하기만 하면 몇 가지 기능이 추가됩니다. 이렇게 하면 Genkit CLI 및 개발자 UI에서 흐름을 실행할 수 있으며, 배포 및 관측 가능성을 비롯한 Genkit의 여러 기능에 대한 요구사항입니다 (이후 섹션에서 이러한 주제를 다룹니다).

입력 및 출력 스키마

모델 API를 직접 호출하는 것에 비교하여 Genkit 흐름이 갖는 가장 중요한 이점 중 하나는 입력 및 출력 모두의 유형 안전성입니다. 흐름을 정의할 때는 generate() 호출의 출력 스키마를 정의하는 것과 거의 동일한 방식으로 Zod를 사용하여 흐름의 스키마를 정의할 수 있습니다. 그러나 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

명령줄에서 흐름을 실행하면 흐름을 테스트하거나 임시로 필요한 작업을 실행하는 흐름을 실행하는 데 유용합니다(예: 문서를 벡터 데이터베이스로 처리하는 흐름 실행).

디버깅 흐름

흐름 내에 AI 로직을 캡슐화하는 한 가지 이점은 Genkit 개발자 UI를 사용하여 앱과 별개로 흐름을 테스트하고 디버그할 수 있다는 것입니다.

개발자 UI를 시작하려면 프로젝트 디렉터리에서 다음 명령어를 실행합니다.

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

개발자 UI의 Run 탭에서 프로젝트에 정의된 흐름을 실행할 수 있습니다.

흐름을 실행한 후 트레이스 보기를 클릭하거나 검사 탭을 확인하여 흐름 호출의 트레이스를 검사할 수 있습니다.

트레이스 뷰어에서는 전체 흐름 실행에 관한 세부정보와 흐름 내 각 개별 단계에 관한 세부정보를 확인할 수 있습니다. 예를 들어 여러 생성 요청이 포함된 다음 흐름을 고려해 보겠습니다.

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용 Cloud Functions로 흐름을 배포하려면 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 플러그인

Express.js

Cloud Run과 같은 Node.js 호스팅 플랫폼을 사용하여 흐름을 배포하려면 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 플랫폼에 흐름 배포를 참고하세요.