Il nucleo delle funzionalità di IA della tua app è costituito dalle richieste di modelli generativi, ma è raro che tu possa semplicemente prendere l'input dell'utente, passarlo al modello e mostrare nuovamente l'output del modello all'utente. In genere, esistono passaggi di pre- e post-elaborazione che devono accompagnare la chiamata del modello. Ad esempio:
- Recupero delle informazioni contestuali da inviare con la chiamata al modello.
- Recupero della cronologia della sessione corrente dell'utente, ad esempio in un'app chat.
- Utilizzo di un modello per riformattare l'input dell'utente in modo che sia adatto per essere passato a un altro modello.
- Valutare la "sicurezza" dell'output di un modello prima di presentarlo all'utente.
- Combinazione dell'output di più modelli.
Ogni passaggio di questo flusso di lavoro deve funzionare insieme per il buon esito di qualsiasi attività correlata all'IA.
In Genkit, questa logica strettamente collegata viene rappresentata utilizzando una struttura chiamata flusso. I flussi vengono scritti come le funzioni, utilizzando il normale codice Go, ma aggiungono funzionalità aggiuntive volte a semplificare lo sviluppo delle funzionalità di IA:
- Sicurezza del tipo: schemi di input e output che forniscono il controllo dei tipi sia statico che di runtime.
- Integrazione con l'interfaccia utente per sviluppatori: esegui il debug dei flussi indipendentemente dal codice dell'applicazione utilizzando l'interfaccia utente per sviluppatori. Nell'interfaccia utente per gli sviluppatori, puoi eseguire i flussi e visualizzare le tracce per ogni passaggio.
- Esecuzione di un deployment semplificata: esegui il deployment dei flussi direttamente come endpoint API web utilizzando qualsiasi piattaforma in grado di ospitare un'app web.
I flussi di Genkit sono leggeri e non invadenti e non forzano la tua app a conformarsi a un'astrazione specifica. Tutta la logica del flusso è scritta in Go standard e il codice all'interno di un flusso non deve essere consapevole del flusso.
Definizione e chiamata dei flussi
Nella sua forma più semplice, un flusso inserisce semplicemente una funzione. Nell'esempio seguente viene eseguita la wrapping di una funzione che chiama GenerateData():
menuSuggestionFlow := genkit.DefineFlow(g, "menuSuggestionFlow",
func(ctx context.Context, theme string) (string, error) {
resp, err := genkit.GenerateData(ctx, g,
ai.WithPrompt("Invent a menu item for a %s themed restaurant.", theme),
)
if err != nil {
return "", err
}
return resp.Text(), nil
})
Se inserisci le chiamate genkit.Generate() in questo modo, aggiungi alcune funzionalità:
in questo modo puoi eseguire il flusso dalla CLI di Genkit e dall'interfaccia utente per gli sviluppatori,
e questo è un requisito per diverse funzionalità di Genkit, tra cui il deployment e
l'osservabilità (le sezioni successive trattano questi argomenti).
Schemi di input e output
Uno dei vantaggi più importanti dei flussi Genkit rispetto all'uso diretto di un'API di modello è la sicurezza di tipo sia degli input che degli output. Quando definisci i flussi, puoi definire gli schemi, in modo molto simile a come definisci lo schema di output di una chiamata genkit.Generate(). Tuttavia, a differenza di genkit.Generate(), puoi anche specificare uno schema di input.
Ecco un perfezionamento dell'ultimo esempio, che definisce un flusso che riceve una stringa come input e restituisce un oggetto:
type MenuItem struct {
Name string `json:"name"`
Description string `json:"description"`
}
menuSuggestionFlow := genkit.DefineFlow(g, "menuSuggestionFlow",
func(ctx context.Context, theme string) (MenuItem, error) {
return genkit.GenerateData[MenuItem](ctx, g,
ai.WithPrompt("Invent a menu item for a %s themed restaurant.", theme),
)
})
Tieni presente che lo schema di un flusso non deve necessariamente essere in linea con lo schema delle chiamate genkit.Generate() al suo interno (infatti, un flusso potrebbe persino non contenere chiamate genkit.Generate()). Ecco una variante dell'esempio
che passa uno schema a genkit.Generate(), ma utilizza l'output strutturato per formattare una stringa semplice, che viene restituita dal flusso.
type MenuItem struct {
Name string `json:"name"`
Description string `json:"description"`
}
menuSuggestionMarkdownFlow := genkit.DefineFlow(g, "menuSuggestionMarkdownFlow",
func(ctx context.Context, theme string) (string, error) {
item, _, err := genkit.GenerateData[MenuItem](ctx, g,
ai.WithPrompt("Invent a menu item for a %s themed restaurant.", theme),
)
if err != nil {
return "", err
}
return fmt.Sprintf("**%s**: %s", item.Name, item.Description), nil
})
Flussi di chiamata
Dopo aver definito un flusso, puoi chiamarlo dal codice Go:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
L'argomento del flusso deve essere conforme allo schema di input.
Se hai definito uno schema di output, la risposta del flusso sarà conforme. Ad esempio, se imposti lo schema di output su MenuItem, l'output del flusso conterrà le relative proprietà:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
if err != nil {
log.Fatal(err)
}
log.Println(item.DishName)
log.Println(item.Description)
Flussi in streaming
I flussi supportano lo streaming utilizzando un'interfaccia simile a quella di genkit.Generate(). Lo streaming è utile quando il flusso genera una grande quantità di output, perché puoi presentarlo all'utente man mano che viene generato, migliorando la reattività percepita della tua app. Come esempio familiare, le interfacce LLM basate su chat spesso trasmettono le risposte all'utente man mano che vengono generate.
Ecco un esempio di flusso che supporta lo streaming:
type Menu struct {
Theme string `json:"theme"`
Items []MenuItem `json:"items"`
}
type MenuItem struct {
Name string `json:"name"`
Description string `json:"description"`
}
menuSuggestionFlow := genkit.DefineStreamingFlow(g, "menuSuggestionFlow",
func(ctx context.Context, theme string, callback core.StreamCallback[string]) (Menu, error) {
item, _, err := genkit.GenerateData[MenuItem](ctx, g,
ai.WithPrompt("Invent a menu item for a %s themed restaurant.", theme),
ai.WithStreaming(func(ctx context.Context, chunk *ai.ModelResponseChunk) error {
// Here, you could process the chunk in some way before sending it to
// the output stream using StreamCallback. In this example, we output
// the text of the chunk, unmodified.
return callback(ctx, chunk.Text())
}),
)
if err != nil {
return nil, err
}
return Menu{
Theme: theme,
Items: []MenuItem{item},
}, nil
})
Il tipo string in StreamCallback[string] specifica il tipo di valori degli stream di flusso. Non deve necessariamente essere dello stesso tipo del tipo di ritorno, ovvero del tipo dell'output completo del flusso (Menu in questo esempio).
In questo esempio, i valori in streaming del flusso sono accoppiati direttamente ai valori in streaming della chiamata genkit.Generate() all'interno del flusso.
Sebbene questo sia spesso il caso, non è necessario: puoi output valori
allo stream utilizzando il callback tutte le volte che è utile per il tuo flusso.
Flussi di streaming delle chiamate
I flussi in streaming possono essere eseguiti come flussi non in streaming con
menuSuggestionFlow.Run(ctx, "bistro") oppure possono essere in streaming:
streamCh, err := menuSuggestionFlow.Stream(ctx, "bistro")
if err != nil {
log.Fatal(err)
}
for result := range streamCh {
if result.Err != nil {
log.Fatal("Stream error: %v", result.Err)
}
if result.Done {
log.Printf("Menu with %s theme:\n", result.Output.Theme)
for item := range result.Output.Items {
log.Println(" - %s: %s", item.Name, item.Description)
}
} else {
log.Println("Stream chunk:", result.Stream)
}
}
Eseguire i flussi dalla riga di comando
Puoi eseguire i flussi dalla riga di comando utilizzando lo strumento a riga di comando Genkit:
genkit flow:run menuSuggestionFlow '"French"'Per i flussi in streaming, puoi stampare l'output in streaming nella console aggiungendo il flag -s:
genkit flow:run menuSuggestionFlow '"French"' -sL'esecuzione di un flusso dalla riga di comando è utile per testarlo o per eseguire flussi che svolgono le attività necessarie su base ad hoc, ad esempio per eseguire un flusso che importa un documento nel database di vettori.
Flussi di debug
Uno dei vantaggi dell'incapsulamento della logica di IA all'interno di un flusso è che puoi testare e eseguire il debug del flusso indipendentemente dalla tua app utilizzando l'interfaccia utente per sviluppatori di Genkit.
Per avviare l'interfaccia utente per gli sviluppatori, esegui il seguente comando dalla directory del progetto:
genkit start -- go run .Nella scheda Esegui dell'interfaccia utente per gli sviluppatori, puoi eseguire uno qualsiasi dei flussi definiti nel tuo progetto:
Dopo aver eseguito un flusso, puoi esaminare una traccia dell'invocazione del flusso facendo clic su Visualizza traccia o consultando la scheda Controlla.
Deployment dei flussi
Puoi eseguire il deployment dei flussi direttamente come endpoint API web, pronti per essere chiamati dai client delle app. Il deployment è descritto in dettaglio in diverse altre pagine, ma questa sezione fornisce brevi panoramiche delle opzioni di deployment.
Server net/http
Per eseguire il deployment di un flusso utilizzando qualsiasi piattaforma di hosting Go, come Cloud Run, definisci il flusso utilizzando DefineFlow() e avvia un server net/http con il gestore del flusso fornito:
import (
"context"
"log"
"net/http"
"github.com/firebase/genkit/go/genkit"
"github.com/firebase/genkit/go/plugins/googlegenai"
"github.com/firebase/genkit/go/plugins/server"
)
func main() {
ctx := context.Background()
g, err := genkit.Init(ctx, genkit.WithPlugins(&googlegenai.GoogleAI{}))
if err != nil {
log.Fatal(err)
}
menuSuggestionFlow := genkit.DefineFlow(g, "menuSuggestionFlow",
func(ctx context.Context, theme string) (MenuItem, error) {
// Flow implementation...
})
mux := http.NewServeMux()
mux.HandleFunc("POST /menuSuggestionFlow", genkit.Handler(menuSuggestionFlow))
log.Fatal(server.Start(ctx, "127.0.0.1:3400", mux))
}
server.Start() è una funzione di supporto facoltativa che avvia il server e ne gestisce il ciclo di vita, inclusa la cattura degli indicatori di interruzione per semplificare lo sviluppo locale, ma puoi utilizzare il tuo metodo.
Per pubblicare tutti i flussi definiti nel codice base, puoi utilizzare ListFlows():
mux := http.NewServeMux()
for _, flow := range genkit.ListFlows(g) {
mux.HandleFunc("POST /"+flow.Name(), genkit.Handler(flow))
}
log.Fatal(server.Start(ctx, "127.0.0.1:3400", mux))
Puoi chiamare un endpoint di flusso con una richiesta POST come segue:
curl -X POST "http://localhost:3400/menuSuggestionFlow" \
-H "Content-Type: application/json" -d '{"data": "banana"}'Altri framework di server
Puoi anche utilizzare altri framework di server per eseguire il deployment dei flussi. Ad esempio, puoi utilizzare Gin con poche righe:
router := gin.Default()
for _, flow := range genkit.ListFlows(g) {
router.POST("/"+flow.Name(), func(c *gin.Context) {
genkit.Handler(flow)(c.Writer, c.Request)
})
}
log.Fatal(router.Run(":3400"))
Per informazioni sul deployment su piattaforme specifiche, consulta Genkit con Cloud Run.