흐름

흐름은 직접 호출에 비해 몇 가지 추가 특성이 있는 래핑된 함수입니다. 강력하게 유형이 지정되고, 스트리밍 가능하며, 로컬 및 원격으로 호출할 수 있고, 완전히 관찰 가능합니다. Firebase Genkit은 흐름을 실행하고 디버그하기 위한 CLI 및 개발자 UI 도구를 제공합니다.

흐름 정의

가장 간단한 형태의 경우 흐름이 함수를 래핑합니다.

menuSuggestionFlow := genkit.DefineFlow(
	"menuSuggestionFlow",
	func(ctx context.Context, restaurantTheme string) (string, error) {
		suggestion := makeMenuItemSuggestion(restaurantTheme)
		return suggestion, nil
	})

이렇게 하면 Genkit CLI 및 개발자 UI에서 함수를 실행할 수 있으며, 배포 및 관측 가능성을 포함한 Genkit의 많은 기능에 대한 요구사항입니다.

모델 API를 직접 호출하는 것에 비교하여 Genkit 흐름이 갖는 중요한 이점은 입력 및 출력 모두의 유형 안전성입니다. 흐름의 인수 및 결과 유형은 단순한 값일 수도 있고 구조화된 값일 수도 있습니다. Genkit는 invopop/jsonschema를 사용하여 이러한 값에 대한 JSON 스키마를 생성합니다.

다음 흐름은 string을 입력으로 사용하고 struct를 출력합니다.

type MenuSuggestion struct {
	ItemName    string `json:"item_name"`
	Description string `json:"description"`
	Calories    int    `json:"calories"`
}


menuSuggestionFlow := genkit.DefineFlow(
	"menuSuggestionFlow",
	func(ctx context.Context, restaurantTheme string) (MenuSuggestion, error) {
		suggestion := makeStructuredMenuItemSuggestion(restaurantTheme)
		return suggestion, nil
	},
)

흐름 실행

코드에서 흐름을 실행하려면 다음 안내를 따르세요.

suggestion, err := menuSuggestionFlow.Run(context.Background(), "French")

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

genkit flow:run menuSuggestionFlow '"French"'

스트리밍

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

// Types for illustrative purposes.
type InputType string
type OutputType string
type StreamType string


menuSuggestionFlow := genkit.DefineStreamingFlow(
	"menuSuggestionFlow",
	func(
		ctx context.Context,
		restaurantTheme InputType,
		callback func(context.Context, StreamType) error,
	) (OutputType, error) {
		var menu strings.Builder
		menuChunks := make(chan StreamType)
		go makeFullMenuSuggestion(restaurantTheme, menuChunks)
		for {
			chunk, ok := <-menuChunks
			if !ok {
				break
			}
			if callback != nil {
				callback(context.Background(), chunk)
			}
			menu.WriteString(string(chunk))
		}
		return OutputType(menu.String()), nil
	},
)

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

스트리밍 모드에서 흐름을 호출하려면 다음 단계를 따르세요.

menuSuggestionFlow.Stream(
	context.Background(),
	"French",
)(func(sfv *genkit.StreamFlowValue[OutputType, StreamType], err error) bool {
	if err != nil {
		// handle err
		return false
	}
	if !sfv.Done {
		fmt.Print(sfv.Stream)
		return true
	} else {
		fmt.Print(sfv.Output)
		return false
	}
})

흐름이 스트리밍을 구현하지 않으면 StreamFlow()는 RunFlow()와 동일하게 동작합니다.

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

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

흐름 배포

HTTP를 통해 흐름에 액세스하려면 먼저 배포해야 합니다. Cloud Run 및 유사한 서비스를 사용하여 흐름을 배포하려면 흐름을 정의한 다음 Init()을 호출합니다.

func main() {
	genkit.DefineFlow(
		"menuSuggestionFlow",
		func(ctx context.Context, restaurantTheme string) (string, error) {
			// ...
			return "", nil
		},
	)
	if err := genkit.Init(context.Background(), nil); err != nil {
		log.Fatal(err)
	}
}

Init는 흐름을 HTTP로 노출하는 net/http 서버를 시작합니다. 엔드포인트 (예: http://localhost:3400/menuSuggestionFlow)

두 번째 파라미터는 다음을 지정하는 Options(선택사항)입니다.

  • FlowAddr: 리슨할 주소 및 포트입니다. 지정하지 않으면 서버가 PORT 환경 변수로 지정한 포트에서 리슨합니다. 비어 있으면 기본값 포트 3400을 사용합니다.
  • Flows: 제공할 흐름입니다. 지정하지 않으면 Init에서 정의된 모든 흐름을 제공합니다.

다른 엔드포인트와 동일한 호스트 및 포트에서 흐름을 제공하려면 FlowAddr를 -로 설정하고 대신 NewFlowServeMux()를 호출하여 Genkit 흐름의 핸들러를 가져와 다른 경로 핸들러와 다중화할 수 있습니다.

mainMux := http.NewServeMux()
mainMux.Handle("POST /flow/", http.StripPrefix("/flow/", genkit.NewFlowServeMux(nil)))

흐름 관측 가능성

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

genkit.DefineFlow(
	"menuSuggestionFlow",
	func(ctx context.Context, restaurantTheme string) (string, error) {
		themes, err := genkit.Run(ctx, "find-similar-themes", func() (string, error) {
			// ...
			return "", nil
		})

		// ...
		return themes, err
	})