Il nucleo delle funzionalità di IA della tua app sono le richieste di modelli generativi, ma è raro che tu possa semplicemente prendere l'input dell'utente, passarlo al modello e mostrare nuovamente all'utente l'output del modello. In genere, esistono passaggi di pre- e post-elaborazione che devono accompagnare la chiamata del modello. Ad esempio:
- Recupero delle informazioni contestuali da inviare con la chiamata al modello
- Recupero della cronologia della sessione corrente dell'utente, ad esempio in un'app di chat
- Utilizzo di un modello per riformattare l'input dell'utente in modo che sia adatto per essere passato a un altro modello
- Valutare la "sicurezza" dell'output di un modello prima di presentarlo all'utente
- Combinare l'output di più modelli
Ogni passaggio di questo flusso di lavoro deve funzionare insieme per il buon esito di qualsiasi attività correlata all'IA.
In Genkit, questa logica strettamente collegata viene rappresentata utilizzando una struttura chiamata flusso. I flussi vengono scritti come le funzioni, utilizzando il normale codice TypeScript, ma aggiungono funzionalità aggiuntive volte a semplificare lo sviluppo delle funzionalità di IA:
- Sicurezza del tipo: schemi di input e output definiti utilizzando Zod, che fornisce il controllo dei tipi sia statico che di runtime
- Integrazione con l'interfaccia utente per sviluppatori: esegui il debug dei flussi indipendentemente dal codice dell'applicazione utilizzando l'interfaccia utente per sviluppatori. Nell'interfaccia utente per gli sviluppatori, puoi eseguire i flussi e visualizzare le tracce per ogni passaggio.
- Deployment semplificato: esegui il deployment dei flussi direttamente come endpoint API web utilizzando Cloud Functions per Firebase o qualsiasi piattaforma in grado di ospitare un'app web.
A differenza di funzionalità simili in altri framework, i flussi di Genkit sono leggeri e discreti e non forzano la tua app a conformarsi a un'astrazione specifica. Tutta la logica del flusso è scritta in TypeScript standard e il codice all'interno di un flusso non deve essere consapevole del flusso.
Definizione e chiamata dei flussi
Nella sua forma più semplice, un flusso racchiude semplicemente una funzione. Nell'esempio riportato di seguito viene eseguita la wrapping di una funzione che chiama 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;
}
);
Se inserisci le chiamate generate() in questo modo, aggiungi alcune funzionalità:
in questo modo puoi eseguire il flusso dalla CLI di Genkit e dall'interfaccia utente per gli sviluppatori,
e questo è un requisito per diverse funzionalità di Genkit, tra cui il deployment e
l'osservabilità (questi argomenti sono trattati nelle sezioni successive).
Schemi di input e output
Uno dei vantaggi più importanti dei flussi Genkit rispetto all'uso diretto di un'API di modello è la sicurezza di tipo sia degli input che degli output. Quando definisci i flussi, puoi definire schemi per loro utilizzando Zod, in modo molto simile a come definisci lo schema di output di una chiamata generate(). Tuttavia, a differenza di generate(), puoi anche specificare uno schema di input.
Ecco un perfezionamento dell'ultimo esempio, che definisce un flusso che riceve una stringa come input e restituisce un oggetto:
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;
}
);
Tieni presente che lo schema di un flusso non deve necessariamente essere in linea con lo schema delle chiamate generate() all'interno del flusso (infatti, un flusso potrebbe persino non contenere chiamate generate()). Ecco una variante dell'esempio che passa uno schema a generate(), ma utilizza l'output strutturato per formattare una semplice stringa restituita dal flusso.
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}`;
}
);
Flussi di chiamata
Una volta definito un flusso, puoi richiamarlo dal codice Node.js:
const { text } = await menuSuggestionFlow('bistro');
L'argomento del flusso deve essere conforme allo schema di input, se ne hai definito uno.
Se hai definito uno schema di output, la risposta del flusso sarà conforme. Ad esempio, se imposti lo schema di output su MenuItemSchema, l'output del flusso conterrà le relative proprietà:
const { dishname, description } =
await menuSuggestionFlowWithSchema('bistro');
Flussi in streaming
I flussi supportano lo streaming utilizzando un'interfaccia simile a quella di generate(). Lo streaming è utile quando il flusso genera una grande quantità di output, perché puoi presentarlo all'utente man mano che viene generato, migliorando la reattività percepita della tua app. Come esempio familiare, le interfacce LLM basate su chat spesso trasmettono le risposte all'utente man mano che vengono generate.
Ecco un esempio di flusso che supporta lo streaming:
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,
};
}
);
- L'opzione
streamSchemaspecifica il tipo di valori trasmessi dal flusso. Non deve necessariamente essere dello stesso tipo dioutputSchema, che è il tipo dell'output completo del flusso. streamingCallbackè una funzione di callback che accetta un singolo parametro, del tipo specificato dastreamSchema. Ogni volta che i dati diventano disponibili all'interno del flusso, inviali allo stream di output chiamando questa funzione. Tieni presente chestreamingCallbackè definito solo se chi ha chiamato il flusso ha richiesto l'output in streaming, quindi devi verificare che sia definito prima di invocarlo.
Nell'esempio precedente, i valori in streaming del flusso sono accoppiati direttamente ai valori in streaming della chiamata generate() all'interno del flusso. Sebbene spesso sia così, non è necessario: puoi inviare valori allo stream utilizzando il callback tutte le volte che è utile per il tuo flusso.
Flussi di streaming delle chiamate
Anche i flussi di streaming sono richiamabili, ma restituiscono immediatamente un oggetto di risposta anziché una promessa:
const response = menuSuggestionStreamingFlow('Danube');
L'oggetto response ha una proprietà stream, che puoi utilizzare per eseguire l'iterazione sull'output in streaming del flusso man mano che viene generato:
for await (const chunk of response.stream) {
console.log('chunk', chunk);
}
Puoi anche ottenere l'output completo del flusso, come con un flusso non in streaming:
const output = await response.output;
Tieni presente che l'output in streaming di un flusso potrebbe non essere dello stesso tipo dell'output completo. L'output in streaming è conforme a streamSchema, mentre l'output completo è conforme a outputSchema.
Eseguire i flussi dalla riga di comando
Puoi eseguire i flussi dalla riga di comando utilizzando lo strumento a riga di comando Genkit:
genkit flow:run menuSuggestionFlow '"French"'Per i flussi in streaming, puoi stampare l'output in streaming nella console aggiungendo il flag -s:
genkit flow:run menuSuggestionFlow '"French"' -sL'esecuzione di un flusso dalla riga di comando è utile per testarlo o per eseguire flussi che svolgono le attività necessarie su base ad hoc, ad esempio per eseguire un flusso che importa un documento nel database di vettori.
Flussi di debug
Uno dei vantaggi dell'incapsulamento della logica di IA all'interno di un flusso è che puoi testare e eseguire il debug del flusso indipendentemente dalla tua app utilizzando l'interfaccia utente per sviluppatori di Genkit.
Per avviare l'interfaccia utente per gli sviluppatori, esegui i seguenti comandi dalla directory del progetto:
genkit start -- tsx --watch src/your-code.tsNella scheda Esegui dell'interfaccia utente per gli sviluppatori, puoi eseguire uno qualsiasi dei flussi definiti nel tuo progetto:
Dopo aver eseguito un flusso, puoi ispezionare una traccia dell'invocazione del flusso facendo clic su Visualizza traccia o consultando la scheda Controlla.
Nel visualizzatore della traccia puoi visualizzare i dettagli sull'esecuzione dell'intero flusso, nonché i dettagli di ciascun singolo passaggio al suo interno. Ad esempio, considera il seguente flusso, che contiene diverse richieste di generazione:
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;
}
);
Quando esegui questo flusso, lo strumento di visualizzazione della traccia mostra i dettagli di ogni richiesta di generazione, incluso l'output:
Passaggi del flusso
Nell'ultimo esempio, hai visto che ogni chiamata generate() veniva visualizzata come un
passaggio distinto nel visualizzatore di tracce. Ogni azione fondamentale di Genkit viene visualizzata come
passaggio distinto di un flusso:
generate()Chat.send()embed()index()retrieve()
Se vuoi includere nelle tracce codice diverso da quello riportato sopra, puoi farlo inserendo il codice in una chiamata run(). Potresti farlo per le chiamate alle librerie di terze parti che non supportano Genkit o per qualsiasi sezione critica di codice.
Ad esempio, di seguito è riportato un flusso con due passaggi: il primo recupera un menu utilizzando un metodo non specificato e il secondo include il menu come contesto per una chiamata 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;
}
);
Poiché il passaggio di recupero è racchiuso in una chiamata run(), è incluso come passaggio nel visualizzatore della traccia:
Deployment dei flussi
Puoi eseguire il deployment dei flussi direttamente come endpoint API web, pronti per essere chiamati dai client delle app. Il deployment è descritto in dettaglio in diverse altre pagine, ma questa sezione fornisce brevi panoramiche delle opzioni di deployment.
Cloud Functions for Firebase
Per eseguire il deployment dei flussi con Cloud Functions for Firebase, utilizza il plug-in firebase. Nelle tue definizioni di flusso, sostituisci defineFlow con onFlow e includi un 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) => {
// ...
}
);
Per ulteriori informazioni, consulta le seguenti pagine:
Express.js
Per eseguire il deployment dei flussi utilizzando qualsiasi piattaforma di hosting Node.js, come Cloud Run, definisci
i flussi utilizzando defineFlow() e poi chiama startFlowServer():
export const menuSuggestionFlow = ai.defineFlow(
{
name: 'menuSuggestionFlow',
},
async (restaurantTheme) => {
// ...
}
);
ai.startFlowServer({
flows: [menuSuggestionFlow],
});
Per impostazione predefinita, startFlowServer pubblicherà tutti i flussi definiti nel codice base come endpoint HTTP (ad esempio http://localhost:3400/menuSuggestionFlow). Puoi chiamare un flusso con una richiesta POST come segue:
curl -X POST "http://localhost:3400/menuSuggestionFlow" \
-H "Content-Type: application/json" -d '{"data": "banana"}'Se necessario, puoi personalizzare il server di flussi in modo che pubblichi un elenco specifico di flussi, come mostrato di seguito. Puoi anche specificare una porta personalizzata (verrà utilizzata la variabile di ambiente PORT, se impostata) o le impostazioni 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: '*',
},
});
Per informazioni sul deployment su piattaforme specifiche, consulta Eseguire il deployment con Cloud Run e Eseguire il deployment dei flussi su qualsiasi piattaforma Node.js.