應用程式 AI 功能的核心是生成式模型要求，但您很少會直接擷取使用者輸入內容、將其傳遞至模型，然後將模型輸出內容顯示給使用者。通常，模型呼叫必須搭配預處理和後處理步驟。例如：
- 擷取要透過模型呼叫傳送的背景資訊
- 擷取使用者目前工作階段的記錄，例如在即時通訊應用程式中
- 使用一個模型，以適合傳遞至其他模型的方式重新格式化使用者輸入內容
- 在向使用者顯示模型輸出內容前，先評估內容的「安全性」
- 結合多個模型的輸出內容
這個工作流程的每個步驟都必須相互配合，才能順利完成任何與 AI 相關的任務。
在 Genkit 中，您可以使用名為「流程」的結構來表示這種緊密連結的邏輯。流程的編寫方式與函式相同，使用一般 TypeScript 程式碼，但會加入額外功能，方便開發 AI 功能：
- 型別安全性：使用 Zod 定義的輸入和輸出結構定義，可提供靜態和執行階段型別檢查
- 與開發人員 UI 整合：使用開發人員 UI 時，偵錯流程與應用程式程式碼無關。您可以在開發人員 UI 中執行流程，並查看流程的每個步驟追蹤記錄。
- 簡化部署程序：使用 Firebase 專用 Cloud Functions 或任何可代管網路應用程式的平台，直接將流程部署為網路 API 端點。
與其他架構中的類似功能不同，Genkit 的流程輕巧且不顯眼，也不會強制應用程式符合任何特定抽象。所有流程邏輯都是以標準 TypeScript 編寫，因此流程內的程式碼不需要支援流程。
定義及呼叫流程
最簡單的情況下，流程只會包裝函式。以下範例會包裝呼叫 generate() 的函式：
export const menuSuggestionFlow = ai.defineFlow(
{
name: 'menuSuggestionFlow',
},
async (restaurantTheme) => {
const { text } = await ai.generate({
model: gemini15Flash,
prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
});
return text;
}
);
只要將 generate() 呼叫包裝成這樣，就能新增一些功能：這樣一來，您就能從 Genkit CLI 和開發人員 UI 執行流程，而且這是 Genkit 多項功能 (包括部署和可觀察性) 的必要條件 (後續章節會討論這些主題)。
輸入和輸出結構定義
相較於直接呼叫模型 API，Genkit 資料流程最重要的優點之一，就是輸入和輸出內容的型別安全性。定義資料流時，您可以使用 Zod 為資料流定義結構定義，這與定義 generate() 呼叫的輸出結構定義方式大致相同；不過，與 generate() 不同的是，您也可以指定輸入結構定義。
以下是上一個範例的改良版，其中定義了一個流程，可將字串做為輸入內容並輸出物件：
const MenuItemSchema = z.object({
dishname: z.string(),
description: z.string(),
});
export const menuSuggestionFlowWithSchema = ai.defineFlow(
{
name: 'menuSuggestionFlow',
inputSchema: z.string(),
outputSchema: MenuItemSchema,
},
async (restaurantTheme) => {
const { output } = await ai.generate({
model: gemini15Flash,
prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
output: { schema: MenuItemSchema },
});
if (output == null) {
throw new Error("Response doesn't satisfy schema.");
}
return output;
}
);
請注意，流程的結構定義不一定會與流程中 generate() 呼叫的結構定義一致 (事實上，流程可能根本不含 generate() 呼叫)。以下是這個範例的變化版本，它會將結構定義傳遞至 generate()，但會使用結構化輸出格式化流程傳回的簡單字串。
export const menuSuggestionFlowMarkdown = ai.defineFlow(
{
name: 'menuSuggestionFlow',
inputSchema: z.string(),
outputSchema: z.string(),
},
async (restaurantTheme) => {
const { output } = await ai.generate({
model: gemini15Flash,
prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
output: { schema: MenuItemSchema },
});
if (output == null) {
throw new Error("Response doesn't satisfy schema.");
}
return `**${output.dishname}**: ${output.description}`;
}
);
通話流程
定義流程後，您可以透過 Node.js 程式碼呼叫它：
const { text } = await menuSuggestionFlow('bistro');
如果您定義了輸入結構定義，則流程的引數必須符合該結構定義。
如果您定義了輸出結構定義，流程回應就會符合該定義。舉例來說，如果您將輸出結構定義設為 MenuItemSchema，流程輸出內容就會包含其屬性：
const { dishname, description } =
await menuSuggestionFlowWithSchema('bistro');
串流流程
Flows 支援使用類似 generate() 串流介面的串流功能。當流程產生大量輸出內容時，串流就很實用，因為您可以將輸出內容在產生時呈現給使用者，進而改善應用程式的回應速度。舉例來說，以聊天為基礎的 LLM 介面通常會在產生回應時，將回應串流傳送給使用者。
以下是支援串流的流程示例：
export const menuSuggestionStreamingFlow = ai.defineStreamingFlow(
{
name: 'menuSuggestionFlow',
inputSchema: z.string(),
streamSchema: z.string(),
outputSchema: z.object({ theme: z.string(), menuItem: z.string() }),
},
async (restaurantTheme, streamingCallback) => {
const response = await ai.generateStream({
model: gemini15Flash,
prompt: `Invent a menu item for a ${restaurantTheme} themed restaurant.`,
});
if (streamingCallback) {
for await (const chunk of response.stream) {
// Here, you could process the chunk in some way before sending it to
// the output stream via streamingCallback(). In this example, we output
// the text of the chunk, unmodified.
streamingCallback(chunk.text);
}
}
return {
theme: restaurantTheme,
menuItem: (await response.response).text,
};
}
);
streamSchema選項會指定資料流的值類型。這個值不一定需要與outputSchema相同，outputSchema是流程完整輸出的類型。streamingCallback是回呼函式，會使用streamSchema指定類型的單一參數。只要資料在流程中可用，請呼叫這個函式，將資料傳送至輸出串流。請注意，只有在流程的呼叫端要求串流輸出時，才會定義streamingCallback，因此您需要在呼叫前確認已定義該值。
在上述範例中，流程傳送的值會直接連結至流程內 generate() 呼叫傳送的值。雖然這通常是情況，但不一定如此：您可以使用回呼將值輸出至串流，頻率可視流程的用途而定。
呼叫串流流程
串流流程也是可呼叫的，但會立即傳回回應物件，而非承諾：
const response = menuSuggestionStreamingFlow('Danube');
回應物件具有串流屬性，可用於在產生時，逐一檢視流程的串流輸出內容：
for await (const chunk of response.stream) {
console.log('chunk', chunk);
}
您也可以取得流程的完整輸出內容，就像使用非串流流程一樣：
const output = await response.output;
請注意，流程的串流輸出內容可能與完整輸出內容的類型不同；串流輸出內容符合 streamSchema，而完整輸出內容符合 outputSchema。
透過指令列執行流程
您可以使用 Genkit CLI 工具，透過指令列執行流程：
genkit flow:run menuSuggestionFlow '"French"'如要針對串流流程輸出串流輸出內容，請新增 -s 旗標：
genkit flow:run menuSuggestionFlow '"French"' -s透過指令列執行流程，可用於測試流程，或執行流程以執行所需的臨時任務，例如執行將文件取入向量資料庫的流程。
偵錯流程
在流程中封裝 AI 邏輯的好處之一，就是您可以使用 Genkit 開發人員 UI 獨立測試及偵錯流程，不必透過應用程式。
如要啟動開發人員 UI，請在專案目錄中執行下列指令：
genkit start -- tsx --watch src/your-code.ts您可以透過開發人員 UI 的「Run」分頁，執行專案中定義的任何流程：
執行流程後，您可以按一下「View trace」或查看「Inspect」分頁，檢查流程叫用記錄。
您可以在追蹤檢視器中查看整個流程的執行詳細資料，以及流程中各個步驟的詳細資料。舉例來說，請考慮下列流程，其中包含多個產生要求：
const PrixFixeMenuSchema = z.object({
starter: z.string(),
soup: z.string(),
main: z.string(),
dessert: z.string(),
});
export const complexMenuSuggestionFlow = ai.defineFlow(
{
name: 'complexMenuSuggestionFlow',
inputSchema: z.string(),
outputSchema: PrixFixeMenuSchema,
},
async (theme: string): Promise<z.infer<typeof PrixFixeMenuSchema>> => {
const chat = ai.chat({ model: gemini15Flash });
await chat.send('What makes a good prix fixe menu?');
await chat.send(
'What are some ingredients, seasonings, and cooking techniques that ' +
`would work for a ${theme} themed menu?`
);
const { output } = await chat.send({
prompt:
`Based on our discussion, invent a prix fixe menu for a ${theme} ` +
'themed restaurant.',
output: {
schema: PrixFixeMenuSchema,
},
});
if (!output) {
throw new Error('No data generated.');
}
return output;
}
);
執行此流程時，追蹤檢視器會顯示每個產生要求的詳細資料，包括輸出內容：
流程步驟
在上一例中，您會發現每個 generate() 呼叫都會在追蹤檢視器中顯示為個別步驟。Genkit 的每個基本動作都會顯示為流程的個別步驟：
generate()Chat.send()embed()index()retrieve()
如果您想在追蹤記錄中加入上述以外的程式碼，可以將程式碼包裝在 run() 呼叫中。您可能會針對呼叫不支援 Genkit 的第三方程式庫，或任何程式碼的關鍵區段執行這項操作。
舉例來說，以下是包含兩個步驟的流程：第一個步驟會使用某種未指定的方法擷取選單，第二個步驟則會將選單納入 generate() 呼叫的內容。
export const menuQuestionFlow = ai.defineFlow(
{
name: 'menuQuestionFlow',
inputSchema: z.string(),
outputSchema: z.string(),
},
async (input: string): Promise<string> => {
const menu = await run('retrieve-daily-menu', async (): Promise<string> => {
// Retrieve today's menu. (This could be a database access or simply
// fetching the menu from your website.)
// ...
return menu;
});
const { text } = await ai.generate({
model: gemini15Flash,
system: "Help the user answer questions about today's menu.",
prompt: input,
docs: [{ content: [{ text: menu }] }],
});
return text;
}
);
由於擷取步驟會在 run() 呼叫中包裝，因此會在追蹤檢視器中列為步驟：
部署流程
您可以直接將流程部署為網路 API 端點，以便從應用程式用戶端呼叫。部署作業已在其他幾個頁面中詳細討論，但本節將簡要介紹部署選項。
Cloud Functions for Firebase
如要使用 Cloud functions for Firebase 部署流程，請使用 firebase 外掛程式。在流程定義中，將 defineFlow 替換為 onFlow，並加入 authPolicy。
import { firebaseAuth } from '@genkit-ai/firebase/auth';
import { onFlow } from '@genkit-ai/firebase/functions';
export const menuSuggestion = onFlow(
ai,
{
name: 'menuSuggestionFlow',
authPolicy: firebaseAuth((user) => {
if (!user.email_verified) {
throw new Error('Verified email required to run flow');
}
}),
},
async (restaurantTheme) => {
// ...
}
);
如需詳細資訊，請參閱以下頁面：
Express.js
如要使用任何 Node.js 代管平台 (例如 Cloud Run) 部署資料流，請使用 defineFlow() 定義資料流，然後呼叫 startFlowServer()：
export const menuSuggestionFlow = ai.defineFlow(
{
name: 'menuSuggestionFlow',
},
async (restaurantTheme) => {
// ...
}
);
ai.startFlowServer({
flows: [menuSuggestionFlow],
});
根據預設，startFlowServer 會將程式碼集中定義的所有流程做為 HTTP 端點 (例如 http://localhost:3400/menuSuggestionFlow) 提供。您可以使用 POST 要求呼叫流程，如下所示：
curl -X POST "http://localhost:3400/menuSuggestionFlow" \
-H "Content-Type: application/json" -d '{"data": "banana"}'如有需要，您可以自訂流程伺服器，以便提供特定的流程清單，如下所示。您也可以指定自訂連接埠 (如果已設定，則會使用 PORT 環境變數)，或指定 CORS 設定。
export const flowA = ai.defineFlow({ name: 'flowA' }, async (subject) => {
// ...
});
export const flowB = ai.defineFlow({ name: 'flowB' }, async (subject) => {
// ...
});
ai.startFlowServer({
flows: [flowB],
port: 4567,
cors: {
origin: '*',
},
});
如要瞭解如何部署至特定平台，請參閱「使用 Cloud Run 部署」和「將流程部署至任何 Node.js 平台」。