定義 AI 工作流程

應用程式 AI 功能的核心是生成式模型要求，但鮮少只擷取使用者輸入內容、傳遞至模型，並將模型輸出內容傳回給使用者。通常必須有模型呼叫的前置和後續處理步驟。例如：

  • 擷取情境資訊，與模型呼叫一起傳送。
  • 擷取使用者目前工作階段的記錄，例如在即時通訊應用程式中。
  • 以適合傳遞至其他模型的方式，使用一個模型將使用者輸入內容重新格式化。
  • 向使用者呈現模型輸出內容前，請先評估模型輸出內容的「安全性」。
  • 合併多個模型的輸出內容。

此工作流程的每個步驟都必須協同合作，才能成功實現任何 AI 相關任務。

在 Genkit 中，您可以使用稱為流程的結構來表示這個緊密連結的邏輯。資料流的編寫方式與函式類似，會使用一般的 Go 程式碼，但新增以下功能，可簡化 AI 功能的開發作業：

  • 類型安全：輸入和輸出結構定義，同時提供靜態和執行階段類型檢查。
  • 與開發人員 UI 整合：使用開發人員 UI，單獨對應用程式程式碼進行偵錯流程。在開發人員 UI 中，您可以執行流程並查看流程中每個步驟的追蹤記錄。
  • 簡化部署：使用任何可託管網頁應用程式的平台，直接將流程部署為網路 API 端點。

Genkit 資料流既輕又不突兀，且不會強制應用程式符合任何特定抽象化機制。流程的所有邏輯都是以標準 Go 編寫，而流程中的程式碼不需要具備流量感知特性。

定義及呼叫流程

以最簡單的形式來說，資料流只會納入一個函式。以下範例會納入呼叫 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
    })

只要像這樣包裝 genkit.Generate() 呼叫，即可新增一些功能：透過 Genkit CLI 和開發人員 UI 執行資料流，而且對於一些 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.GenerateData() 的範例變化，但使用結構化輸出內容來設定資料流傳回的簡單字串格式。請注意，我們將 MenuItem 做為類型參數傳遞的方式，相當於傳遞 WithOutputType() 選項並在回應中取得該類型的值。

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

StreamCallback[string] 中的 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

從指令列執行流程對於測試流程或執行臨時所需工作的執行流程很有幫助，例如執行將文件擷取到向量資料庫的流程。

偵錯流程

在流程中封裝 AI 邏輯的一大優點，就是您可以使用 Genkit 開發人員 UI，單獨測試應用程式流程並進行偵錯。

開發人員 UI 仰賴 Go 應用程式繼續執行，即使邏輯已完成也一樣。如果您才剛開始使用，而且 Genkit 不屬於廣泛應用程式，請新增 select {} 做為 main() 的最後一行，以防止應用程式關閉，以便在 UI 中檢查。

如要啟動開發人員 UI，請從專案目錄執行下列指令：

genkit start -- go run .

在開發人員 UI 的「Run」分頁中，您可以執行專案中定義的任何流程：

Flow 執行器的螢幕截圖

執行流程後，只要按一下「View trace」或「Inspect」分頁，即可檢查流程叫用的追蹤記錄。

部署流程

您可以直接將流程部署為網路 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"))

如要瞭解如何部署至特定平台，請參閱「搭配 Cloud Run 使用 Genkit」一文。