הגדרת תהליכי עבודה של AI

הליבה של תכונות ה-AI באפליקציה היא בקשות למודלים גנרטיביים, אבל לרוב אי אפשר פשוט לקבל קלט מהמשתמש, להעביר אותו למודל ולהציג את הפלט של המודל בחזרה למשתמש. בדרך כלל, יש שלבים של עיבוד מקדים ועיבוד לאחרי הקריאה למודל. לדוגמה:

• אחזור מידע לפי הקשר לשליחה עם קריאת המודל
• אחזור ההיסטוריה של הסשן הנוכחי של המשתמש, למשל באפליקציית צ'אט
• שימוש במודל אחד כדי לעצב מחדש את הקלט של המשתמש באופן שמתאים להעברה למודל אחר
• הערכת 'הבטיחות' של הפלט של מודל לפני הצגתו למשתמש
• שילוב הפלט של כמה מודלים

כל שלב בתהליך העבודה הזה חייב לפעול יחד כדי שכל משימה שקשורה ל-AI תצליח.

ב-Genkit, מייצגים את הלוגיקה המקושרת הזו באמצעות מבנה שנקרא תהליך. תהליכים נכתבים בדיוק כמו פונקציות, באמצעות קוד TypeScript רגיל, אבל הם מוסיפים יכולות נוספות שנועדו להקל על פיתוח תכונות AI:

• בטיחות סוגים: סכימות קלט ופלט שמוגדרות באמצעות Zod, שמספק בדיקת סוגים גם סטטית וגם בסביבת זמן הריצה
• שילוב עם ממשק משתמש למפתחים: ניפוי באגים בתהליכים בנפרד מקוד האפליקציה באמצעות ממשק המשתמש למפתחים. בממשק המשתמש למפתחים אפשר להריץ תהליכים ולראות את הטרייסים של כל שלב בתהליך.
• פריסה פשוטה יותר: פריסה של תהליכים ישירות כנקודות קצה ל-API אינטרנט, באמצעות Cloud Functions for Firebase או כל פלטפורמה שיכולה לארח אפליקציית אינטרנט.

בניגוד לתכונות דומות במסגרות אחרות, התהליכים של 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() מוסיפה פונקציונליות מסוימת: כך תוכלו להריץ את התהליך מ-CLI של Genkit ומממשק המשתמש למפתחים, וזו גם דרישה לשימוש בכמה מהתכונות של Genkit, כולל פריסה וניטור (הנושאים האלה נדונים בקטעים הבאים).

סכימות קלט ופלט

אחד מהיתרונות החשובים ביותר של תהליכי Genkit על פני קריאה ישירה ל-API של מודל הוא בטיחות הסוגים של הקלט והפלט. כשמגדירים תהליכים, אפשר להגדיר להם סכימות באמצעות 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');

תהליכי עבודה בסטרימינג

זרמים תומכים בסטרימינג באמצעות ממשק שדומה לממשק הסטרימינג של 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, שהוא הסוג של הפלט המלא של התהליך.
• streamingCallback היא פונקציית קריאה חוזרת שמקבלת פרמטר יחיד מהסוג שצוין ב-streamSchema. בכל פעם שנתונים יהיו זמינים בתהליך, צריך לשלוח את הנתונים לזרם הפלט באמצעות קריאה לפונקציה הזו. חשוב לזכור ש-streamingCallback מוגדר רק אם מבצע הקריאה של הזרימה ביקש פלט סטרימינג, לכן צריך לבדוק שהוא מוגדר לפני שמפעילים אותו.

בדוגמה שלמעלה, הערכים שמשודרים על ידי הזרימה מקושרים ישירות לערכים שמשודרים על ידי הקריאה ל-generate() בתוך הזרימה. בדרך כלל זה המצב, אבל זה לא חייב להיות כך: אפשר להפיק ערכים לסטרימינג באמצעות הפונקציה הלא פוליקאלית (callback) בתדירות שמתאימה לתהליך שלכם.

קריאה לזרמי סטרימינג

אפשר גם להפעיל את הזרמים בסטרימינג, אבל הם מחזירים באופן מיידי אובייקט תגובה במקום הבטחה:

const response = menuSuggestionStreamingFlow('Danube');

לאובייקט התגובה יש מאפיין stream, שאפשר להשתמש בו כדי לבצע איטרציה על הפלט בסטרימינג של התהליך בזמן שהוא נוצר:

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.

כדי להפעיל את ממשק המשתמש למפתחים, מריצים את הפקודות הבאות מהספרייה של הפרויקט:

genkit start -- tsx --watch src/your-code.ts

בכרטיסייה 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) => {
    // ...
  }
);

מידע נוסף זמין בדפים הבאים:

• פריסה באמצעות Firebase
• הרשאה ואימות
• הפלאגין של Firebase

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"}'

אם צריך, אפשר להתאים אישית את שרת ה-flows כך שיציג רשימה ספציפית של flows, כפי שמתואר בהמשך. אפשר גם לציין יציאה בהתאמה אישית (אם משתנה הסביבה 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.