Definizione dei flussi di lavoro di IA

Il cuore delle funzionalità di IA della tua app sono le richieste del modello generativo, ma è raro che tu possa semplicemente prendere l'input dell'utente, passarlo al modello e visualizzare l'output del modello all'utente. Di solito, la chiamata del modello prevede alcune fasi di pre- e post-elaborazione. Ad esempio:

  • Recupero delle informazioni contestuali da inviare con la chiamata del modello.
  • Recupero della cronologia della sessione corrente dell'utente, ad esempio in un'app di chat.
  • Utilizzare un modello per riformattare l'input dell'utente in modo da poter essere trasferito a un altro modello.
  • Valutazione della "sicurezza" dell'output di un modello prima di presentarlo all'utente.
  • Combinazione dell'output di diversi modelli.

Ogni passaggio di questo flusso di lavoro deve funzionare assieme per il successo di qualsiasi attività correlata all'IA.

In Genkit, rappresenti questa logica strettamente collegata utilizzando una costruzione chiamata flusso. I flussi sono scritti proprio come le funzioni, utilizzando il normale codice Go, ma aggiungono ulteriori funzionalità destinate a facilitare lo sviluppo delle funzionalità di IA:

  • Sicurezza dei tipi: schemi di input e output, che forniscono il controllo del tipo sia statico che di runtime.
  • Integrazione con l'interfaccia utente per sviluppatori: i flussi di debug sono indipendenti dal codice dell'applicazione utilizzando l'interfaccia utente per sviluppatori. Nella UI per sviluppatori puoi eseguire flussi e visualizzare tracce per ogni passaggio del flusso.
  • Deployment semplificato: il deployment dei flussi avviene direttamente come endpoint delle API web, utilizzando qualsiasi piattaforma in grado di ospitare un'app web.

I flussi di Genkit sono leggeri e discreti e non obbligano la tua app a conformarsi a nessuna astrazione specifica. Tutta la logica del flusso è scritta in Go standard e il codice all'interno di un flusso non deve essere sensibile al flusso.

Definizione e chiamata dei flussi

Nella sua forma più semplice, un flusso aggrega una funzione. L'esempio seguente aggrega una funzione che chiama Generate():

menuSuggestionFlow := genkit.DefineFlow(g, "menuSuggestionFlow",
    func(ctx context.Context, theme string) (string, error) {
        resp, err := genkit.Generate(ctx, g,
            ai.WithPrompt("Invent a menu item for a %s themed restaurant.", theme),
        )
        if err != nil {
            return "", err
        }

        return resp.Text(), nil
    })

È sufficiente eseguire il wrapping delle chiamate genkit.Generate() in questo modo per aggiungere alcune funzionalità: in questo modo, puoi eseguire il flusso dall'interfaccia a riga di comando di Genkit e dall'interfaccia utente per sviluppatori. È inoltre un requisito per diverse funzionalità di Genkit, tra cui deployment e osservabilità (le sezioni successive trattano questi argomenti).

Schemi di input e output

Uno dei vantaggi più importanti dei flussi Genkit rispetto alla chiamata diretta di un'API del modello è la sicurezza dei tipi sia di input che di output. Durante la definizione dei flussi, puoi definire gli schemi, proprio come faresti con 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 prende 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() all'interno del flusso (in effetti, un flusso potrebbe non contenere nemmeno chiamate genkit.Generate()). Ecco una variante dell'esempio che chiama genkit.GenerateData(), ma utilizza l'output strutturato per formattare una stringa semplice, che il flusso restituisce. Nota come passiamo MenuItem come parametro di tipo; questo equivale a passare l'opzione WithOutputType() e a ricevere un valore di quel tipo in risposta.

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

Una volta definito un flusso, puoi chiamarlo dal tuo 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 sue proprietà:

item, err := menuSuggestionFlow.Run(ctx, "bistro")
if err != nil {
    log.Fatal(err)
}

log.Println(item.DishName)
log.Println(item.Description)

Flussi di flussi di dati

I flussi supportano i flussi di dati utilizzando un'interfaccia simile all'interfaccia di flusso di genkit.Generate(). Il flusso di dati è utile quando il flusso genera una grande quantità di output, perché puoi presentare l'output all'utente mentre viene generato, il che migliora la reattività percepita della tua app. Ad esempio, le interfacce LLM basate su chat spesso inviano le risposte all'utente man mano che vengono generate.

Ecco un esempio di flusso che supporta i flussi di dati:

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 trasmessi dal flusso. Non è necessario che sia dello stesso tipo del flusso restituito, che è il tipo di output completo del flusso (Menu in questo esempio).

In questo esempio, i valori trasmessi dal flusso sono direttamente associati ai valori trasmessi dalla chiamata genkit.Generate() all'interno del flusso. Sebbene ciò avvenga spesso, non è necessario che lo sia: puoi inviare valori al flusso di dati utilizzando il callback tutte le volte che ritieni utile per il tuo flusso.

Chiamata ai flussi di streaming

I flussi di flussi di dati possono essere eseguiti come flussi non di flusso con menuSuggestionFlow.Run(ctx, "bistro") oppure possono essere trasmessi in flusso:

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

Esecuzione di flussi dalla riga di comando

Puoi eseguire flussi dalla riga di comando utilizzando lo strumento dell'interfaccia a riga di comando di Genkit:

genkit flow:run menuSuggestionFlow '"French"'

Per i flussi di flussi di dati, puoi stampare l'output del flusso di dati sulla console aggiungendo il flag -s:

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

L'esecuzione di un flusso dalla riga di comando è utile per testare un flusso o per eseguire flussi che eseguono attività necessarie ad hoc, ad esempio per eseguire un flusso che importa un documento nel database vettoriale.

Flussi di debug

Uno dei vantaggi di incapsulare la logica di AI all'interno di un flusso è che puoi testare ed eseguire il debug del flusso in modo indipendente dalla tua app utilizzando l'interfaccia utente per sviluppatori Genkit.

L'UI per sviluppatori si basa sull'esecuzione dell'app Go, anche se la logica è stata completata. Se hai appena iniziato e Genkit non fa parte di un'app più ampia, aggiungi select {} come ultima riga di main() per evitare che l'app si chiuda in modo da poterla esaminare nella UI.

Per avviare l'interfaccia utente dello sviluppatore, esegui il comando seguente dalla directory del progetto:

genkit start -- go run .

Dalla scheda Esegui dell'interfaccia utente sviluppatore, puoi eseguire uno qualsiasi dei flussi definiti nel tuo progetto:

Screenshot dell'esecutore Flow

Dopo aver eseguito un flusso, puoi ispezionare una traccia della chiamata del flusso facendo clic su Visualizza traccia o osservando la scheda Ispeziona.

Deployment dei flussi

Puoi eseguire il deployment dei flussi direttamente come endpoint API web, pronti per essere chiamato dai client della tua app. Il deployment è discusso in dettaglio in molte altre pagine, ma questa sezione offre una breve panoramica delle opzioni di deployment.

Server net/http

Per eseguire il deployment di un flusso utilizzando qualsiasi piattaforma di hosting Go, ad esempio Cloud Run, definisci il flusso utilizzando DefineFlow() e avvia un server net/http con il gestore di 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 helper facoltativa che avvia il server e ne gestisce il ciclo di vita, inclusa l'acquisizione di indicatori di interruzione per facilitare lo sviluppo locale, ma puoi utilizzare il tuo metodo.

Per gestire tutti i flussi definiti nel tuo codebase, 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 del server

Puoi anche utilizzare altri framework del 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.