Ядром функций искусственного интеллекта вашего приложения являются запросы генеративной модели, но редко можно просто взять вводимые пользователем данные, передать их в модель и отобразить выходные данные модели обратно пользователю. Обычно вызов модели должен сопровождаться этапами предварительной и постобработки. Например:
- Получение контекстной информации для отправки с вызовом модели.
- Получение истории текущего сеанса пользователя, например, в приложении чата.
- Использование одной модели для переформатирования вводимых пользователем данных таким образом, чтобы их можно было передать в другую модель.
- Оценка «безопасности» выходных данных модели перед представлением их пользователю.
- Объединение выпуска нескольких моделей.
Каждый шаг этого рабочего процесса должен работать вместе, чтобы любая задача, связанная с ИИ, была успешной.
В Genkit вы представляете эту тесно связанную логику с помощью конструкции, называемой потоком. Потоки пишутся так же, как и функции, с использованием обычного кода Go, но они добавляют дополнительные возможности, призванные облегчить разработку функций ИИ:
- Безопасность типов : схемы ввода и вывода, обеспечивающие как статическую проверку типов, так и проверку типов во время выполнения.
- Интеграция с пользовательским интерфейсом разработчика . Отладка выполняется независимо от кода вашего приложения с использованием пользовательского интерфейса разработчика. В пользовательском интерфейсе разработчика вы можете запускать потоки и просматривать трассировки для каждого шага потока.
- Упрощенное развертывание . Развертывайте потоки непосредственно в качестве конечных точек веб-API, используя любую платформу, на которой может размещаться веб-приложение.
Потоки Genkit легкие и ненавязчивые и не заставляют ваше приложение соответствовать какой-либо конкретной абстракции. Вся логика потока написана на стандартном Go, и код внутри потока не обязательно должен поддерживать поток.
Определение и вызов потоков
В своей простейшей форме поток просто оборачивает функцию. Следующий пример оборачивает функцию, которая вызывает 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
})
Просто обертывая вызовы genkit.Generate() таким образом, вы добавляете некоторую функциональность: это позволяет запускать поток из CLI Genkit и из пользовательского интерфейса разработчика, а также является требованием для некоторых функций Genkit, включая развертывание и возможность наблюдения (эти темы обсуждаются в последующих разделах).
Схемы ввода и вывода
Одним из наиболее важных преимуществ потоков Genkit по сравнению с прямым вызовом API модели является безопасность типов как входных, так и выходных данных. При определении потоков вы можете определять схемы почти так же, как вы определяете схему вывода вызова genkit.Generate() ; однако, в отличие от genkit.Generate() , вы также можете указать входную схему.
Вот уточнение последнего примера, который определяет поток, который принимает строку в качестве входных данных и выводит объект:
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),
)
})
Обратите внимание, что схема потока не обязательно должна совпадать со схемой вызовов genkit.Generate() внутри потока (на самом деле поток может даже не содержать вызовов genkit.Generate() ). Вот вариант примера, который передает схему в genkit.Generate() , но использует структурированный вывод для форматирования простой строки, которую возвращает поток.
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
})
Потоки вызовов
Определив поток, вы можете вызвать его из кода Go:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
Аргумент потока должен соответствовать входной схеме.
Если вы определили схему вывода, ответ потока будет соответствовать ей. Например, если вы установите схему вывода MenuItem , выходные данные потока будут содержать его свойства:
item, err := menuSuggestionFlow.Run(ctx, "bistro")
if err != nil {
log.Fatal(err)
}
log.Println(item.DishName)
log.Println(item.Description)
Потоковые потоки
Потоки поддерживают потоковую передачу, используя интерфейс, аналогичный интерфейсу потоковой передачи genkit.Generate() . Потоковая передача полезна, когда ваш поток генерирует большой объем выходных данных, поскольку вы можете представить выходные данные пользователю по мере их создания, что улучшает восприятие реакции вашего приложения. В качестве знакомого примера можно привести интерфейсы LLM на основе чата, которые часто передают пользователю ответы по мере их создания.
Вот пример потока, поддерживающего потоковую передачу:
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
})
Тип string в StreamCallback[string] определяет тип значений потоков вашего потока. Это не обязательно должен быть тот же тип, что и тип возвращаемого значения, который является типом полного вывода потока ( Menu в этом примере).
В этом примере значения, передаваемые потоком, напрямую связаны со значениями, передаваемыми вызовом genkit.Generate() внутри потока. Хотя это часто бывает, это не обязательно: вы можете выводить значения в поток, используя обратный вызов, так часто, как это полезно для вашего потока.
Вызов потоковых потоков
Потоковые потоки можно запускать как непотоковые потоки с помощью menuSuggestionFlow.Run(ctx, "bistro") или их можно передавать в потоковом режиме:
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)
}
}
Запуск потоков из командной строки
Вы можете запускать потоки из командной строки с помощью инструмента Genkit CLI:
genkit flow:run menuSuggestionFlow '"French"' Для потоков потоковой передачи вы можете вывести выходные данные потоковой передачи на консоль, добавив флаг -s :
genkit flow:run menuSuggestionFlow '"French"' -sЗапуск потока из командной строки полезен для тестирования потока или для запуска потоков, выполняющих задачи, необходимые на разовой основе — например, для запуска потока, который принимает документ в базу данных векторов.
Отладка потоков
Одним из преимуществ инкапсуляции логики искусственного интеллекта в потоке является то, что вы можете тестировать и отлаживать поток независимо от вашего приложения, используя пользовательский интерфейс разработчика Genkit.
Пользовательский интерфейс разработчика полагается на то, что приложение Go продолжает работать, даже если логика завершена. Если вы только начинаете и Genkit не является частью более широкого приложения, добавьте select {} в последнюю строку main() чтобы предотвратить завершение работы приложения, чтобы вы могли проверить его в пользовательском интерфейсе.
Чтобы запустить пользовательский интерфейс разработчика, выполните следующую команду из каталога вашего проекта:
genkit start -- go run .На вкладке «Выполнить» пользовательского интерфейса разработчика вы можете запустить любой из потоков, определенных в вашем проекте:
После запуска потока вы можете проверить трассировку вызова потока, щелкнув «Просмотреть трассировку» или просмотрев вкладку «Проверка» .
Развертывание потоков
Вы можете развернуть свои потоки непосредственно в качестве конечных точек веб-API, готовых к вызову из клиентов вашего приложения. Развертывание подробно обсуждается на нескольких других страницах, но в этом разделе даются краткие обзоры вариантов развертывания.
net/http сервер
Чтобы развернуть поток с помощью любой хостинговой платформы Go, например Cloud Run, определите свой поток с помощью DefineFlow() и запустите сервер net/http с предоставленным обработчиком потока:
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() — это дополнительная вспомогательная функция, которая запускает сервер и управляет его жизненным циклом, включая перехват сигналов прерываний для упрощения локальной разработки, но вы можете использовать свой собственный метод.
Чтобы обслуживать все потоки, определенные в вашей кодовой базе, вы можете использовать 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))
Вы можете вызвать конечную точку потока с помощью запроса POST следующим образом:
curl -X POST "http://localhost:3400/menuSuggestionFlow" \
-H "Content-Type: application/json" -d '{"data": "banana"}'Другие серверные платформы
Вы также можете использовать другие серверные платформы для развертывания своих потоков. Например, вы можете использовать Gin всего в несколько строк:
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"))
Информацию о развертывании на конкретных платформах см. в разделе Genkit с Cloud Run .