El núcleo de las funciones de IA de tu app son las solicitudes de modelos generativos, pero es raro que puedas simplemente tomar la entrada del usuario, pasarla al modelo y mostrarle el resultado del modelo al usuario. Por lo general, hay pasos de procesamiento previo y posterior que deben acompañar a la llamada al modelo. Por ejemplo:
- Recupera información contextual para enviarla con la llamada al modelo.
- Recuperar el historial de la sesión actual del usuario, por ejemplo, en una app de chat
- Usar un modelo para cambiar el formato de la entrada del usuario de una manera adecuada para pasarla a otro modelo.
- Evaluar la “seguridad” del resultado de un modelo antes de presentarlo al usuario.
- Combinar el resultado de varios modelos.
Cada paso de este flujo de trabajo debe funcionar en conjunto para que cualquier tarea relacionada con la IA sea exitosa.
En Genkit, representas esta lógica estrechamente vinculada con una construcción llamada flujo. Los flujos se escriben de la misma manera que las funciones, con código Go normal, pero agregan capacidades adicionales destinadas a facilitar el desarrollo de funciones de IA:
- Seguridad de tipos: Esquemas de entrada y salida, que proporcionan verificación de tipos estáticos y de tiempo de ejecución.
- Integración con la IU para desarrolladores: Depurar flujos independientemente del código de tu aplicación con la IU para desarrolladores. En la IU para desarrolladores, puedes ejecutar flujos y ver los seguimientos de cada paso del flujo.
- Implementación simplificada: Implementa flujos directamente como extremos de API web con cualquier plataforma que pueda alojar una app web.
Los flujos de Genkit son ligeros y discretos, y no obligan a tu app a cumplir con ninguna abstracción específica. Toda la lógica del flujo está escrita en Go estándar, y el código dentro de un flujo no necesita ser consciente del flujo.
Define y llama a flujos
En su forma más sencilla, un flujo solo une una función. En el siguiente ejemplo, se une una función que llama a 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
})
Si unes tus llamadas a genkit.Generate() de esta manera, agregas algunas funciones: esto te permite ejecutar el flujo desde la CLI de Genkit y desde la IU para desarrolladores, y es un requisito para varias de las funciones de Genkit, incluidas las funciones de observabilidad y de implementación (en secciones posteriores, se analizan estos temas).
Esquemas de entrada y salida
Una de las ventajas más importantes que tienen los flujos de Genkit en comparación con llamar directamente a una API de modelo es la seguridad de tipos de entrada y salida. Cuando defines flujos, puedes definir esquemas, de la misma manera que defines el esquema de salida de una llamada a genkit.Generate(). Sin embargo, a diferencia de genkit.Generate(), también puedes especificar un esquema de entrada.
Este es un perfeccionamiento del último ejemplo, que define un flujo que toma una cadena como entrada y genera un objeto:
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),
)
})
Ten en cuenta que el esquema de un flujo no tiene que alinearse necesariamente con el esquema de las llamadas a genkit.Generate() dentro del flujo (de hecho, un flujo ni siquiera puede contener llamadas a genkit.Generate()). Esta es una variación del ejemplo que pasa un esquema a genkit.Generate(), pero usa el resultado estructurado para dar formato a una cadena simple, que devuelve el flujo.
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
})
Flujos de llamadas
Una vez que hayas definido un flujo, puedes llamarlo desde tu código Go:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
El argumento del flujo debe cumplir con el esquema de entrada.
Si definiste un esquema de salida, la respuesta del flujo se ajustará a él. Por ejemplo, si configuras el esquema de salida como MenuItem, el resultado del flujo contendrá sus propiedades:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
if err != nil {
log.Fatal(err)
}
log.Println(item.DishName)
log.Println(item.Description)
Flujos de transmisión
Los flujos admiten la transmisión a través de una interfaz similar a la interfaz de transmisión de genkit.Generate(). La transmisión es útil cuando tu flujo genera una gran cantidad de resultados, ya que puedes presentarlos al usuario a medida que se generan, lo que mejora la capacidad de respuesta percibida de tu app. Como ejemplo conocido, las interfaces de LLM basadas en chat suelen transmitir sus respuestas al usuario a medida que se generan.
Este es un ejemplo de un flujo que admite la transmisión:
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
})
El tipo string en StreamCallback[string] especifica el tipo de valores que transmite tu flujo. No es necesario que sea del mismo tipo que el tipo de datos que se devuelve, que es el tipo del resultado completo del flujo (Menu en este ejemplo).
En este ejemplo, los valores que transmite el flujo se acoplan directamente a los valores que transmite la llamada a genkit.Generate() dentro del flujo.
Si bien este suele ser el caso, no tiene por qué serlo: puedes enviar valores al flujo con la devolución de llamada con la frecuencia que sea útil para tu flujo.
Llamadas a flujos de transmisión
Los flujos de transmisión se pueden ejecutar como flujos que no son de transmisión con menuSuggestionFlow.Run(ctx, "bistro") o se pueden transmitir de la siguiente manera:
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)
}
}
Ejecuta flujos desde la línea de comandos
Puedes ejecutar flujos desde la línea de comandos con la herramienta de la CLI de Genkit:
genkit flow:run menuSuggestionFlow '"French"'Para los flujos de transmisión, puedes imprimir el resultado de la transmisión en la consola si agregas la marca -s:
genkit flow:run menuSuggestionFlow '"French"' -sEjecutar un flujo desde la línea de comandos es útil para probarlo o para ejecutar flujos que realizan tareas necesarias de forma ad hoc, por ejemplo, para ejecutar un flujo que transfiera un documento a tu base de datos de vectores.
Flujos de depuración
Una de las ventajas de encapsular la lógica de IA dentro de un flujo es que puedes probar y depurar el flujo de forma independiente de tu app con la IU para desarrolladores de Genkit.
La IU del desarrollador depende de que la app para Go siga ejecutándose, incluso si la lógica se completó. Si recién comienzas y Genkit no forma parte de una app más amplia, agrega select {} como la última línea de main() para evitar que la app se cierre y puedas inspeccionarla en la IU.
Para iniciar la IU para desarrolladores, ejecuta el siguiente comando desde el directorio de tu proyecto:
genkit start -- go run .En la pestaña Ejecutar de la IU para desarrolladores, puedes ejecutar cualquiera de los flujos definidos en tu proyecto:
Después de ejecutar un flujo, puedes inspeccionar un registro de la invocación del flujo haciendo clic en Ver registro o en la pestaña Inspeccionar.
Implementa flujos
Puedes implementar tus flujos directamente como extremos de API web, listos para que los llames desde los clientes de tu app. La implementación se analiza en detalle en varias otras páginas, pero en esta sección se proporcionan breves descripciones generales de tus opciones de implementación.
Servidor net/http
Para implementar un flujo con cualquier plataforma de hosting de Go, como Cloud Run, define tu flujo con DefineFlow() y, luego, inicia un servidor net/http con el controlador de flujo proporcionado:
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() es una función auxiliar opcional que inicia el servidor y administra su ciclo de vida, incluida la captura de indicadores de interrupción para facilitar el desarrollo local, pero puedes usar tu propio método.
Para entregar todos los flujos definidos en tu base de código, puedes usar 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))
Puedes llamar a un extremo de flujo con una solicitud POST de la siguiente manera:
curl -X POST "http://localhost:3400/menuSuggestionFlow" \
-H "Content-Type: application/json" -d '{"data": "banana"}'Otros frameworks de servidor
También puedes usar otros frameworks de servidor para implementar tus flujos. Por ejemplo, puedes usar Gin con solo unas pocas líneas:
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"))
Para obtener información sobre la implementación en plataformas específicas, consulta Genkit con Cloud Run.