這個默認的項目模版中定義了一個默認的 flow，名叫 menuSuggestionFlow。我們可以選擇這個 flow 來，來進行測試。比如如下圖所示，我們輸入 apple，讓 AI 幫我推薦一個菜品：這里我們選擇了 Google AI 驅動的 Go 語言模版，并且生成了一個示例的 flow。我們接下來就可以直接使用 npx genkit-cli@latest start 來啟動開發服務器。
npx genkit@0.5.17 start
Starting app at ....
Genkit Tools API: http://localhost:4000/api
time=2024-11-14T09:18:25.771+08:00 level=INFO msg="googleai.Init: Google AI requires setting GOOGLE_GENAI_API_KEY or GOOGLE_API_KEY in the environment. You can get an API key at https://ai.google.dev"
exit status 1
App process exited with code 1, signal null這里指的注意的是我們需要設置 Google AI 的 API Key，我們可以通過 export GOOGLE_GENAI_API_KEY=<your-api-key> 來設置。如果不設置，則會出現上面的錯誤。這個 API Key 可以通過 Google AI[11] 獲取。由于目前 Gemini 提供了每天的免費額度，所以我們日常的開發和測試可以在免費額度內完成。不過需要注意的是，主流的國際 AI 服務都限制了中國地區的訪問，請根據你的情況選擇可以訪問的方式。
export GOOGLE_GENAI_API_KEY=
npx genkit-cli@latest start
Starting app at ....
Genkit Tools API: http://localhost:4000/api
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=model name=googleai/gemini-1.5-flash
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=model name=googleai/gemini-1.0-pro
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=model name=googleai/gemini-1.5-pro
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=embedder name=googleai/text-embedding-004
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=embedder name=googleai/embedding-001
time=2024-11-14T09:22:58.907+08:00 level=INFO msg=RegisterAction type=flow name=menuSuggestionFlow
time=2024-11-14T09:22:58.907+08:00 level=INFO msg="starting reflection server"
time=2024-11-14T09:22:58.908+08:00 level=INFO msg="starting flow server"
time=2024-11-14T09:22:58.908+08:00 level=INFO msg="all servers started successfully"
time=2024-11-14T09:22:58.908+08:00 level=INFO msg="server listening" addr=127.0.0.1:3100
time=2024-11-14T09:22:58.908+08:00 level=INFO msg="server listening" addr=127.0.0.1:3400
time=2024-11-14T09:22:58.933+08:00 level=INFO msg="request start" reqID=1 method=GET path=/api/__health
time=2024-11-14T09:22:58.933+08:00 level=INFO msg="request end" reqID=1
Genkit Tools UI: http://localhost:4000通過上面的輸出，我們可以看到 Genkit 的開發服務器已經啟動，并且我們可以通過 http://localhost:4000 訪問 Genkit 的 UI 界面。這是 Genkit 提供的工具界面，通過這個界面，我們可以方便的調試和測試我們的 AI Agent。
這個默認的項目模版中定義了一個默認的 flow，名叫 menuSuggestionFlow。我們可以選擇這個 flow 來，來進行測試。比如如下圖所示，我們輸入 apple，讓 AI 幫我推薦一個菜品：
注意一下，在這個頁面的最下方包含了一個 View Trace 的按鈕，這個按鈕可以幫助我們查看 AI Agent 的執行 trace，這對于我們調試和優化 AI Agent 非常重要。這個追蹤借助了 OpenTelemetry[12] 的實現，也非常方便在你的代碼中進行擴展，方便追蹤 Agent 的執行過程。
接下來，我們看一下代碼是如何定義整個 AI Agent 的。
// 初始化 Genkit 并啟動流程服務器。此調用必須放在最后，在所有插件配置和流程定義之后。將 nil 配置項傳遞給 Init 時，Genkit 將啟動本地流程服務器，您可以使用開發者界面進行交互。
if err := genkit.Init(ctx, nil); err != nil {
log.Fatal(err)
}}
import (
"context"
"errors"
"fmt"
"log"
// 導入 Genkit 核心庫
"github.com/firebase/genkit/go/ai"
"github.com/firebase/genkit/go/genkit"
// 導入 Google AI 插件
"github.com/firebase/genkit/go/plugins/googleai"
)
func main() {
ctx := context.Background()
// 初始化 Google AI 插件。留空 apiKey 參數時，插件會從推薦使用的 GOOGLE_GENAI_[API](http://www.dlbhg.com/wiki/api/)_KEY 環境變量讀取值。
if err := googleai.Init(ctx, nil); err != nil {
log.Fatal(err)
}
// 定義一個簡單流程，提示大型語言模型 (LLM) 生成菜單建議。
genkit.DefineFlow("menuSuggestionFlow", func(ctx context.Context, input string) (string, error) {
// Google AI [API 提供訪問](http://www.dlbhg.com/blog/api-access-what-is-api-access)多個生成模型的功能。這里我們指定 gemini-1.5-flash。
m := googleai.Model("gemini-1.5-flash")
if m == nil {
return "", errors.New("menuSuggestionFlow: failed to find model")
}
// 構建請求并發送到[模型 API](http://www.dlbhg.com/blog/model-api-is-the-best-way-to-unlock-ai)。
resp, err := m.Generate(ctx,
ai.NewGenerateRequest(
&ai.GenerationCommonConfig{Temperature: 1},
ai.NewUserTextMessage(fmt.Sprintf("Suggest an item for the menu of a %s themed restaurant", input))),
nil)
if err != nil {
return "", err
}
// 處理來自模型 API 的響應。在這個例子中，我們只將其轉換為字符串，但更復雜的流程可能會將響應轉換為結構化輸出或將響應鏈接到另一個 LLM 調用等。
text := resp.Text()
return text, nil
})
// 初始化 Genkit 并啟動流程服務器。此調用必須放在最后，在所有插件配置和流程定義之后。將 nil 配置項傳遞給 Init 時，Genkit 將啟動本地流程服務器，您可以使用開發者界面進行交互。
if err := genkit.Init(ctx, nil); err != nil {
log.Fatal(err)
}
}這其中比較重要的流程在 genkit.DefineFlow 中定義。這個定義了流程名和對應的實現方式。在這個例子中，我們定義了一個名為 menuSuggestionFlow 的流程，這個流程接收一個字符串輸入，然后返回一個字符串輸出。在最開始我們初始化了一個 Google AI 的模型，然后通過這個模型來生成對應的輸出。當然，正如代碼注釋中提到的，正常的 AI Agent 要遠比這個 Hello World 級別的程序更加復雜。不過沒關系，接下來，我們就嘗試把這個應用修改一下，實現一個基礎的，你自己的 ChatPDF 工具。
如何實現一個自己的 ChatPDF 工具？思路當然很簡單，我們需要一個流程，這個流程接收一個 PDF 文件然后進行保存。同時，我們還需要另外一個流程，接收一個用戶的問題，查詢保存的數據，然后返回一個答案。在第一個過程中，我們需要解析這個 PDF 文件，然后根據這個 PDF 文件的內容來回答用戶的問題。在這個過程中我們會需要用到一些 AI 的能力，比如文本的摘要，文本的問答，文本的總結等等。
所以首先，我們需要定義一個提取 PDF 文件內容的方法，這個功能需要使用到 github.com/ledongthuc/pdf[13] 這個庫，我們可以偷懶直接復制它的示例代碼：
func readPdf(path string) (string, error) {
f, r, err := pdf.Open(path)
// remember close file
defer f.Close()
if err != nil {
return "", err
}
var buf bytes.Buffer
b, err := r.GetPlainText()
if err != nil {
return "", err
}
buf.ReadFrom(b)
return buf.String(), nil
}由于 AI 上下文長度的限制，我們很難把整個 PDF 文件的內容都交給 AI 來作文上下文（不過雖然我們在用的 Gemini 的 API 上下文長度確實足以滿足這個需求，但是我們現在討論的是一種更通用的解決方案），所以我們需要對 PDF 文件拆分進行向量化，僅把需要的文本內容交給 AI 來作文上下文。這里解析到文本內容后，我們還需要使用 LangChainGo 中的 github.com/tmc/langchaingo/textsplitter[14] 這個庫來對文本內容進行裁剪，確保文本長度不會超過 AI 的上下文限制。
splitter := textsplitter.NewRecursiveCharacter(
textsplitter.WithChunkSize(200),
textsplitter.WithChunkOverlap(20),
)內容基本完成，現在我們可以定義一個單獨的工作流，這個工作流接受一個 PDF 文件路徑，并且拆分解析這個 PDF 中的文本，并將其向量化后保存，以供后續使用。為了方便，我們在這里使用一個調試用 VectorDB，這個 DB 是基于本地文件使用的 github.com/firebase/genkit/go/plugins/localvec[15]。請注意，我這里使用的 LangchainGo 庫的版本為 v0.1.13-pre.0，可能會和你的使用 API 有一些出入。
if err := localvec.Init(); err != nil {
log.Fatal(err)
}
pdfIndexer, _, err := localvec.DefineIndexerAndRetriever(
"pdfIndexer",
localvec.Config{
Embedder: googleai.Embedder("text-embedding-004"),
},
)
if err != nil {
log.Fatal(err)
} splitter := textsplitter.NewRecursiveCharacter(
textsplitter.WithChunkSize(200),
textsplitter.WithChunkOverlap(20),
) genkit.DefineFlow("indexPDF", func(ctx context.Context, input string) (string, error) {
pdfText, err := genkit.Run(ctx, "readPdf", func() (string, error) {
return readPdf(input)
})
if err != nil {
return "", err
} chunks, err := genkit.Run(ctx, "chunk", func() ([]*ai.Document, error) {
chunks, err := splitter.SplitText(pdfText)
if err != nil {
return nil, err
} var docs []*ai.Document
for _, chunk := range chunks {
docs = append(docs, ai.DocumentFromText(chunk, nil))
}
return docs, nil
})
if err != nil {
return "", err
} err = ai.Index(ctx, pdfIndexer, ai.WithIndexerDocs(chunks...))
return "", err
})如果這個時候你還在運行 Genkit 的開發服務器，那么它會自動嘗試構建這段代碼，這時候我們回到 Genkit 的 UI 界面，選擇我們剛剛定義的 indexPDF 工作流，然后輸入一個 PDF 文件路徑，點擊 Run 按鈕，就可以看到這個工作流的執行結果。注意上面的代碼中，genkit.Run 方法的第一個參數是 ctx，這個參數是 Genkit 提供的上下文，可以用于追蹤和日志記錄。為了方便檢查，我們也可以通過 View Trace 查看一下我們當前的執行調用情況：
這里我使用了 Nuki 這家公司的介紹故事 PDF 作為測試文件，你可以根據你的需要修改這個文件路徑。
在完成文件處理流程之后，我們就需要實現一個基礎的，面向問答的 Flow。畢竟，前面的文件解析流程對普通用戶來說只是前置步驟，我們最終的目的還是希望用戶可以上傳一個 PDF 文件，然后可以向這個文件提問，并得到答案。
接下來就是實現一個 chatPDF 的 Flow，這個 Flow 接收一個用戶的問題，然后根據這個問題的內容，從向量數據庫中檢索出相關的文本內容，然后交給 AI 來作文回答。這部分也就是我們熟悉的 RAG[16] 回答環節了。
// 我們要額外修改一個地方，確保我們能夠從向量數據庫中檢索出相關的文本內容。
pdfIndexer, pdfRetriever, err := localvec.DefineIndexerAndRetriever(
"pdfIndexer",
localvec.Config{
Embedder: googleai.Embedder("text-embedding-004"),
},
)
if err != nil {
log.Fatal(err)
}
// ...保持其他代碼不變... genkit.DefineFlow("qaPDF", func(ctx context.Context, question string) (string, error) {
model := googleai.Model("gemini-1.5-flash") // 從向量數據庫中檢索出相關的文本內容
docs, err := genkit.Run(ctx, "retrieve", func() (*ai.RetrieverResponse, error) {
return pdfRetriever.Retrieve(ctx, &ai.RetrieverRequest{
Document: ai.DocumentFromText(question, nil),
})
})
if err != nil {
return "", err
} // 在此之前我們聲明嵌入信息：
embededInfo := ai.NewSystemTextMessage("Here is the context:")
for _, doc := range docs.Documents {
embededInfo.Content = append(embededInfo.Content, doc.Content...)
} // 現在可以讓 AI 回答問題了
resp, err := ai.GenerateText(ctx, model, ai.WithMessages(
ai.NewSystemTextMessage(`You are acting as a helpful AI assistant that can answer questions about the provided context.
Use only the context provided to answer the question. If you don't know, do not make up an answer. Do not add or change anything in the context.`),
embededInfo,
ai.NewUserTextMessage(question),
))
return resp, err
})現在我們已經定義了一個完整的工具流程，剛剛的工作流用于處理文件內容，這個工作流用于檢索這些相關信息并進行回答。因為我使用的文件是 Nuki 公司的介紹文件，那么我的問題自然也和這家公司相關：
中到底取出了什么樣的數據，這會幫助我們決策是否需要優化 RAG 的實現：
在完成這個 Demo 之后，我們應該已經初步了解了 Genkit 的開發流程，并且可以基于這個框架開發出自己的 AI Agent 應用。雖然這個工具目前仍舊是在早期階段，但應對基礎的 Agent 開發已經基本足夠。當然，從另外一方面說，這個框架仍舊需要生態進一步完善，比如以 RAG 為例，我們剛剛使用的僅僅是最基礎的 RAG 功能， 為了提升 RAG 的成功率，業內還有很多其他的實現方式提升搜索結果準確率，比如 GraphRAG 等等，這部分功能暫時沒有開箱即用的方式，仍然需要進一步提升生態。
不過今天的文章就先到這里結束吧。
[1]Genkit: https://github.com/firebase/genkit
[2]Node.js 的實現: https://firebase.google.com/docs/genkit?hl=zh-cn
[3]Go 語言的實現: https://firebase.google.com/docs/genkit-go/get-started-go
[4]十五周年博客: https://go.dev/blog/15years
[5]LangChainGo: https://github.com/tmc/langchaingo
[6]Genkit: https://developers.googleblog.com/en/introducing-genkit-for-go-build-scalable-ai-powered-apps-in-go/
[7]LangChain: https://www.langchain.com/
[8]LangGraph: https://www.langchain.com/langgraph
[9]LangGraphGo: https://github.com/tmc/langgraphgo
[10]https://github.com/firebase/genkit/issues/1295
[11]Google AI: https://ai.google.dev
[12]OpenTelemetry: https://opentelemetry.io/
[13]github.com/ledongthuc/pdf: https://github.com/ledongthuc/pdf
[14]github.com/tmc/langchaingo/textsplitter: https://pkg.go.dev/github.com/tmc/langchaingo/textsplitter
[15]github.com/firebase/genkit/go/plugins/localvec: https://pkg.go.dev/github.com/firebase/genkit/go/plugins/localvec
[16]RAG: https://en.wikipedia.org/wiki/Retrieval-augmented_generation
文章轉自微信公眾號@白日異夢