這里我們選擇了 Google AI 驅(qū)動(dòng)的 Go 語言模版，并且生成了一個(gè)示例的 flow。我們接下來就可以直接使用?npx genkit-cli@latest start?來啟動(dòng)開發(fā)服務(wù)器。

? 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

這里指的注意的是我們需要設(shè)置 Google AI 的 API Key，我們可以通過?export GOOGLE_GENAI_API_KEY=<your-api-key>?來設(shè)置。如果不設(shè)置，則會(huì)出現(xiàn)上面的錯(cuò)誤。這個(gè) API Key 可以通過?Google AI^[11]?獲取。由于目前 Gemini 提供了每天的免費(fèi)額度，所以我們?nèi)粘５拈_發(fā)和測試可以在免費(fèi)額度內(nèi)完成。不過需要注意的是，主流的國際 AI 服務(wù)都限制了中國地區(qū)的訪問，請(qǐng)根據(jù)你的情況選擇可以訪問的方式。

? export GOOGLE_GENAI_API_KEY=<your-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 的開發(fā)服務(wù)器已經(jīng)啟動(dòng)，并且我們可以通過?http://localhost:4000?訪問 Genkit 的 UI 界面。這是 Genkit 提供的工具界面，通過這個(gè)界面，我們可以方便的調(diào)試和測試我們的 AI Agent。

這個(gè)默認(rèn)的項(xiàng)目模版中定義了一個(gè)默認(rèn)的 flow，名叫?menuSuggestionFlow。我們可以選擇這個(gè) flow 來，來進(jìn)行測試。比如如下圖所示，我們輸入?apple，讓 AI 幫我推薦一個(gè)菜品：

注意一下，在這個(gè)頁面的最下方包含了一個(gè)?View Trace?的按鈕，這個(gè)按鈕可以幫助我們查看 AI Agent 的執(zhí)行 trace，這對(duì)于我們調(diào)試和優(yōu)化 AI Agent 非常重要。這個(gè)追蹤借助了?OpenTelemetry^[12]?的實(shí)現(xiàn)，也非常方便在你的代碼中進(jìn)行擴(kuò)展，方便追蹤 Agent 的執(zhí)行過程。

接下來，我們看一下代碼是如何定義整個(gè) AI Agent 的。

package main



import (

    "context"

    "errors"

    "fmt"

    "log"



    // 導(dǎo)入 Genkit 核心庫

    "github.com/firebase/genkit/go/ai"

    "github.com/firebase/genkit/go/genkit"



    // 導(dǎo)入 Google AI 插件

    "github.com/firebase/genkit/go/plugins/googleai"

)



func main() {

    ctx := context.Background()



    // 初始化 Google AI 插件。留空 apiKey 參數(shù)時(shí)，插件會(huì)從推薦使用的 GOOGLE_GENAI_API_KEY 環(huán)境變量讀取值。

    if err := googleai.Init(ctx, nil); err != nil {

        log.Fatal(err)

    }



    // 定義一個(gè)簡單流程，提示大型語言模型 (LLM) 生成菜單建議。

    genkit.DefineFlow("menuSuggestionFlow", func(ctx context.Context, input string) (string, error) {

        // Google AI API 提供訪問多個(gè)生成模型的功能。這里我們指定 gemini-1.5-flash。

        m := googleai.Model("gemini-1.5-flash")

        if m == nil {

            return "", errors.New("menuSuggestionFlow: failed to find model")

        }



        // 構(gòu)建請(qǐng)求并發(fā)送到模型 API。

        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 的響應(yīng)。在這個(gè)例子中，我們只將其轉(zhuǎn)換為字符串，但更復(fù)雜的流程可能會(huì)將響應(yīng)轉(zhuǎn)換為結(jié)構(gòu)化輸出或?qū)㈨憫?yīng)鏈接到另一個(gè) LLM 調(diào)用等。

        text := resp.Text()

        return text, nil

    })



    // 初始化 Genkit 并啟動(dòng)流程服務(wù)器。此調(diào)用必須放在最后，在所有插件配置和流程定義之后。將 nil 配置項(xiàng)傳遞給 Init 時(shí)，Genkit 將啟動(dòng)本地流程服務(wù)器，您可以使用開發(fā)者界面進(jìn)行交互。

    if err := genkit.Init(ctx, nil); err != nil {

        log.Fatal(err)

    }

}

這其中比較重要的流程在?genkit.DefineFlow?中定義。這個(gè)定義了流程名和對(duì)應(yīng)的實(shí)現(xiàn)方式。在這個(gè)例子中，我們定義了一個(gè)名為?menuSuggestionFlow?的流程，這個(gè)流程接收一個(gè)字符串輸入，然后返回一個(gè)字符串輸出。在最開始我們初始化了一個(gè) Google AI 的模型，然后通過這個(gè)模型來生成對(duì)應(yīng)的輸出。當(dāng)然，正如代碼注釋中提到的，正常的 AI Agent 要遠(yuǎn)比這個(gè) Hello World 級(jí)別的程序更加復(fù)雜。不過沒關(guān)系，接下來，我們就嘗試把這個(gè)應(yīng)用修改一下，實(shí)現(xiàn)一個(gè)基礎(chǔ)的，你自己的 ChatPDF 工具。

實(shí)現(xiàn) ChatPDF 工具

如何實(shí)現(xiàn)一個(gè)自己的 ChatPDF 工具？思路當(dāng)然很簡單，我們需要一個(gè)流程，這個(gè)流程接收一個(gè) PDF 文件然后進(jìn)行保存。同時(shí)，我們還需要另外一個(gè)流程，接收一個(gè)用戶的問題，查詢保存的數(shù)據(jù)，然后返回一個(gè)答案。在第一個(gè)過程中，我們需要解析這個(gè) PDF 文件，然后根據(jù)這個(gè) PDF 文件的內(nèi)容來回答用戶的問題。在這個(gè)過程中我們會(huì)需要用到一些 AI 的能力，比如文本的摘要，文本的問答，文本的總結(jié)等等。

所以首先，我們需要定義一個(gè)提取 PDF 文件內(nèi)容的方法，這個(gè)功能需要使用到?github.com/ledongthuc/pdf^[13]?這個(gè)庫，我們可以偷懶直接復(fù)制它的示例代碼：

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 上下文長度的限制，我們很難把整個(gè) PDF 文件的內(nèi)容都交給 AI 來作文上下文（不過雖然我們在用的 Gemini 的 API 上下文長度確實(shí)足以滿足這個(gè)需求，但是我們現(xiàn)在討論的是一種更通用的解決方案），所以我們需要對(duì) PDF 文件拆分進(jìn)行向量化，僅把需要的文本內(nèi)容交給 AI 來作文上下文。這里解析到文本內(nèi)容后，我們還需要使用 LangChainGo 中的?github.com/tmc/langchaingo/textsplitter^[14]?這個(gè)庫來對(duì)文本內(nèi)容進(jìn)行裁剪，確保文本長度不會(huì)超過 AI 的上下文限制。

splitter := textsplitter.NewRecursiveCharacter(

 textsplitter.WithChunkSize(200),

 textsplitter.WithChunkOverlap(20),

)

內(nèi)容基本完成，現(xiàn)在我們可以定義一個(gè)單獨(dú)的工作流，這個(gè)工作流接受一個(gè) PDF 文件路徑，并且拆分解析這個(gè) PDF 中的文本，并將其向量化后保存，以供后續(xù)使用。為了方便，我們在這里使用一個(gè)調(diào)試用 VectorDB，這個(gè) DB 是基于本地文件使用的?github.com/firebase/genkit/go/plugins/localvec^[15]。請(qǐng)注意，我這里使用的 LangchainGo 庫的版本為?v0.1.13-pre.0，可能會(huì)和你的使用 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

 })

如果這個(gè)時(shí)候你還在運(yùn)行 Genkit 的開發(fā)服務(wù)器，那么它會(huì)自動(dòng)嘗試構(gòu)建這段代碼，這時(shí)候我們回到 Genkit 的 UI 界面，選擇我們剛剛定義的?indexPDF?工作流，然后輸入一個(gè) PDF 文件路徑，點(diǎn)擊?Run?按鈕，就可以看到這個(gè)工作流的執(zhí)行結(jié)果。注意上面的代碼中，genkit.Run?方法的第一個(gè)參數(shù)是?ctx，這個(gè)參數(shù)是 Genkit 提供的上下文，可以用于追蹤和日志記錄。為了方便檢查，我們也可以通過?View Trace?查看一下我們當(dāng)前的執(zhí)行調(diào)用情況：

這里我使用了 Nuki 這家公司的介紹故事 PDF 作為測試文件，你可以根據(jù)你的需要修改這個(gè)文件路徑。

在完成文件處理流程之后，我們就需要實(shí)現(xiàn)一個(gè)基礎(chǔ)的，面向問答的 Flow。畢竟，前面的文件解析流程對(duì)普通用戶來說只是前置步驟，我們最終的目的還是希望用戶可以上傳一個(gè) PDF 文件，然后可以向這個(gè)文件提問，并得到答案。

接下來就是實(shí)現(xiàn)一個(gè)?chatPDF?的 Flow，這個(gè) Flow 接收一個(gè)用戶的問題，然后根據(jù)這個(gè)問題的內(nèi)容，從向量數(shù)據(jù)庫中檢索出相關(guān)的文本內(nèi)容，然后交給 AI 來作文回答。這部分也就是我們熟悉的?RAG^[16]?回答環(huán)節(jié)了。

 // 我們要額外修改一個(gè)地方，確保我們能夠從向量數(shù)據(jù)庫中檢索出相關(guān)的文本內(nèi)容。

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



  // 從向量數(shù)據(jù)庫中檢索出相關(guān)的文本內(nèi)容

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

  }



  // 現(xiàn)在可以讓 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

 })

現(xiàn)在我們已經(jīng)定義了一個(gè)完整的工具流程，剛剛的工作流用于處理文件內(nèi)容，這個(gè)工作流用于檢索這些相關(guān)信息并進(jìn)行回答。因?yàn)槲沂褂玫奈募?Nuki 公司的介紹文件，那么我的問題自然也和這家公司相關(guān)：

為了方便調(diào)試，我們還可以點(diǎn)擊?View Trace?按鈕，查看當(dāng)前的執(zhí)行調(diào)用情況，比如說我們從向量數(shù)據(jù)庫中到底取出了什么樣的數(shù)據(jù)，這會(huì)幫助我們決策是否需要優(yōu)化 RAG 的實(shí)現(xiàn)：

其他

在完成這個(gè) Demo 之后，我們應(yīng)該已經(jīng)初步了解了 Genkit 的開發(fā)流程，并且可以基于這個(gè)框架開發(fā)出自己的 AI Agent 應(yīng)用。雖然這個(gè)工具目前仍舊是在早期階段，但應(yīng)對(duì)基礎(chǔ)的 Agent 開發(fā)已經(jīng)基本足夠。當(dāng)然，從另外一方面說，這個(gè)框架仍舊需要生態(tài)進(jìn)一步完善，比如以 RAG 為例，我們剛剛使用的僅僅是最基礎(chǔ)的 RAG 功能， 為了提升 RAG 的成功率，業(yè)內(nèi)還有很多其他的實(shí)現(xiàn)方式提升搜索結(jié)果準(zhǔn)確率，比如 GraphRAG 等等，這部分功能暫時(shí)沒有開箱即用的方式，仍然需要進(jìn)一步提升生態(tài)。

不過今天的文章就先到這里結(jié)束吧。

參考資料

[1]Genkit:?https://github.com/firebase/genkit

[2]Node.js 的實(shí)現(xiàn):?https://firebase.google.com/docs/genkit?hl=zh-cn

[3]Go 語言的實(shí)現(xiàn):?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

文章轉(zhuǎn)自微信公眾號(hào)@白日異夢

#你可能也喜歡這些API文章!

如何快速實(shí)現(xiàn)REST API集成以優(yōu)化業(yè)務(wù)流程

使用FastAPI為Python構(gòu)建應(yīng)用程序

使用Django REST Framework構(gòu)建API

使用Flask、Google Cloud SQL和App Engine設(shè)置API

微服務(wù)為什么要用到 API 網(wǎng)關(guān)？

14個(gè)文本轉(zhuǎn)圖像AI API

什么是API定義？

修復(fù)API中損壞的訪問控制的指南

前端需要的免費(fèi)在線API接口