Flux

Les flux sont des fonctions encapsulées qui présentent certaines caractéristiques supplémentaires par rapport aux appels directs : elles sont fortement typées, peuvent être diffusées, peuvent être appelées localement et à distance, et sont entièrement observables. Firebase Genkit fournit des outils de CLI et d'UI pour les développeurs pour exécuter et déboguer des flux.

Définir des flux

Dans sa forme la plus simple, un flux encapsule simplement une fonction :

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

Cela vous permet d'exécuter la fonction à partir de la CLI Genkit et de l'UI du développeur. Cette étape est obligatoire pour de nombreuses fonctionnalités de Genkit, y compris le déploiement et l'observabilité.

Les flux Genkit présentent un avantage important par rapport à l'appel direct d'une API de modèle : la sécurité de type des entrées et des sorties. Les types d'arguments et de résultats d'un flux peuvent être des valeurs simples ou structurées. Genkit génère des schémas JSON pour ces valeurs à l'aide de invopop/jsonschema.

Le flux suivant prend un string en entrée et renvoie un 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
	},
)

Flux en cours d'exécution

Pour exécuter un flux dans votre code:

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

Vous pouvez également utiliser la CLI pour exécuter des flux :

genkit flow:run menuSuggestionFlow '"French"'

Diffusée

Voici un exemple simple d'un flux pouvant diffuser des valeurs en continu :

// 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
	},
)

Notez que le rappel de streaming peut être non défini. Il n'est défini que si le client à l'origine de l'appel demande une réponse en streaming.

Pour appeler un flux en mode streaming :

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

Si le flux n'implémente pas le streaming, StreamFlow() se comporte de la même manière que RunFlow().

Vous pouvez également utiliser la CLI pour diffuser des flux:

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

Déployer des flux

Si vous souhaitez pouvoir accéder à votre flux via HTTP, vous devez d'abord le déployer. Pour déployer des flux à l'aide de Cloud Run et de services similaires, définissez vos flux, puis appelez 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 démarre un serveur net/http qui expose vos flux en tant que points de terminaison HTTP (par exemple, http://localhost:3400/menuSuggestionFlow).

Le deuxième paramètre est un Options facultatif qui spécifie les éléments suivants:

• FlowAddr : adresse et port à écouter. Si ce n'est pas le cas, le serveur écoute sur le port spécifié par la variable d'environnement PORT. Si cette variable est vide, le port 3400 est utilisé par défaut.
• Flows: flux à diffuser. Si ce paramètre n'est pas spécifié, Init traite tous vos flux définis.

Si vous souhaitez diffuser des flux sur le même hôte et le même port que d'autres points de terminaison, vous pouvez définir FlowAddr sur - et appeler NewFlowServeMux() à la place pour obtenir un gestionnaire pour vos flux Genkit, que vous pouvez multiplexer avec vos autres gestionnaires de routage :

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

Observabilité des flux

Lorsque vous utilisez des SDK tiers qui ne sont pas instrumentés pour l'observabilité, vous pouvez parfois les afficher en tant qu'étape de traçage distincte dans l'interface utilisateur du développeur. Il vous suffit d'encapsuler le code dans la fonction 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
	})