Firebase Genkit

何时使用此技能

• AI 工作流编排：构建具有类型安全输入/输出的多步骤 AI 管道
• 基于流程的 API：将 LLM 调用封装成可部署的 HTTP 端点
• 工具调用 / 代理：为模型配备自定义工具并实现代理循环
• RAG 管道：使用向量数据库（Pinecone, pgvector, Firestore, Chroma 等）进行检索增强生成
• 多代理系统：协调多个专门的 AI 代理
• 流式响应：为聊天或长篇内容提供实时的逐词元输出
• Firebase/Cloud Run 部署：将 AI 函数部署到 Google Cloud
• 提示词管理：使用 Dotprompt 将提示词作为版本化的 .prompt 文件进行管理

安装与设置

步骤 1：安装 Genkit CLI

# npm (推荐用于 JavaScript/TypeScript)
npm install -g genkit-cli

# macOS/Linux 二进制文件
curl -sL cli.genkit.dev | bash

步骤 2：创建一个 TypeScript 项目

mkdir my-genkit-app && cd my-genkit-app
npm init -y
npm pkg set type=module
npm install -D typescript tsx
npx tsc --init
mkdir src && touch src/index.ts

步骤 3：安装 Genkit 核心和一个模型插件

# 核心 + Google AI (Gemini) — 免费层级，无需信用卡
npm install genkit @genkit-ai/google-genai

# 或者：Vertex AI (需要 GCP 项目)
npm install genkit @genkit-ai/vertexai

# 或者：OpenAI
npm install genkit genkitx-openai

# 或者：Anthropic (Claude)
npm install genkit genkitx-anthropic

# 或者：Ollama (本地模型)
npm install genkit genkitx-ollama

步骤 4：配置 API 密钥

# Google AI (Gemini)
export GEMINI_API_KEY=your_key_here

# OpenAI
export OPENAI_API_KEY=your_key_here

# Anthropic
export ANTHROPIC_API_KEY=your_key_here

核心概念

初始化 Genkit

import { googleAI } from '@genkit-ai/google-genai';
import { genkit } from 'genkit';

const ai = genkit({
  plugins: [googleAI()],
  model: googleAI.model('gemini-2.5-flash'), // 默认模型
});

定义流程

流程是核心原语：类型安全、可观察、可部署的 AI 函数。

import { genkit, z } from 'genkit';
import { googleAI } from '@genkit-ai/google-genai';

const ai = genkit({ plugins: [googleAI()] });

// 使用 Zod 定义输入/输出模式
const SummaryInputSchema = z.object({
  text: z.string().describe('要总结的文本'),
  maxWords: z.number().optional().default(100),
});

const SummaryOutputSchema = z.object({
  summary: z.string(),
  keyPoints: z.array(z.string()),
});

export const summarizeFlow = ai.defineFlow(
  {
    name: 'summarizeFlow',
    inputSchema: SummaryInputSchema,
    outputSchema: SummaryOutputSchema,
  },
  async ({ text, maxWords }) => {
    const { output } = await ai.generate({
      model: googleAI.model('gemini-2.5-flash'),
      prompt: `用最多 ${maxWords} 个词总结以下文本并提取关键点：\n\n${text}`,
      output: { schema: SummaryOutputSchema },
    });

    if (!output) throw new Error('未生成输出');
    return output;
  }
);

// 调用流程
const result = await summarizeFlow({
  text: '这里是长文章内容...',
  maxWords: 50,
});
console.log(result.summary);

生成内容

// 简单文本生成
const { text } = await ai.generate({
  model: googleAI.model('gemini-2.5-flash'),
  prompt: '用一句话解释量子计算。',
});

// 结构化输出
const { output } = await ai.generate({
  prompt: '列出 3 种编程语言及其用例',
  output: {
    schema: z.object({
      languages: z.array(z.object({
        name: z.string(),
        useCase: z.string(),
      })),
    }),
  },
});

// 带系统提示词
const { text: response } = await ai.generate({
  system: '你是一位资深的 TypeScript 工程师。回答要简洁。',
  prompt: 'TypeScript 中 interface 和 type 的区别是什么？',
});

// 多模态（图像 + 文本）
const { text: description } = await ai.generate({
  prompt: [
    { text: '这张图片里有什么？' },
    { media: { url: 'https://example.com/image.jpg', contentType: 'image/jpeg' } },
  ],
});

流式流程

export const streamingFlow = ai.defineFlow(
  {
    name: 'streamingFlow',
    inputSchema: z.object({ topic: z.string() }),
    streamSchema: z.string(),        // 每个数据块的类型
    outputSchema: z.object({ full: z.string() }),
  },
  async ({ topic }, { sendChunk }) => {
    const { stream, response } = ai.generateStream({
      prompt: `写一篇关于 ${topic} 的详细文章。`,
    });

    for await (const chunk of stream) {
      sendChunk(chunk.text);         // 将每个词元流式传输到客户端
    }

    const { text } = await response;
    return { full: text };
  }
);

// 客户端消费
const stream = streamingFlow.stream({ topic: 'AI 伦理' });
for await (const chunk of stream.stream) {
  process.stdout.write(chunk);
}
const finalOutput = await stream.output;

工具调用（代理）

import { z } from 'genkit';

// 定义工具
const getWeatherTool = ai.defineTool(
  {
    name: 'getWeather',
    description: '获取某个城市的当前天气',
    inputSchema: z.object({ city: z.string() }),
    outputSchema: z.object({ temp: z.number(), condition: z.string() }),
  },
  async ({ city }) => {
    // 调用真实的天气 API
    return { temp: 22, condition: '晴朗' };
  }
);

const searchWebTool = ai.defineTool(
  {
    name: 'searchWeb',
    description: '在网络上搜索信息',
    inputSchema: z.object({ query: z.string() }),
    outputSchema: z.string(),
  },
  async ({ query }) => {
    // 调用搜索 API
    return `关于 "${query}" 的搜索结果`;
  }
);

// 带有工具的代理流程
export const agentFlow = ai.defineFlow(
  {
    name: 'agentFlow',
    inputSchema: z.object({ question: z.string() }),
    outputSchema: z.string(),
  },
  async ({ question }) => {
    const { text } = await ai.generate({
      prompt: question,
      tools: [getWeatherTool, searchWebTool],
      returnToolRequests: false, // 自动执行工具
    });
    return text;
  }
);

使用 Dotprompt 管理提示词

将提示词作为版本化的 .prompt 文件进行管理：

# src/prompts/summarize.prompt
---
model: googleai/gemini-2.5-flash
input:
  schema:
    text: string
    style?: string
output:
  schema:
    summary: string
    sentiment: string
---
用 {{style, default: "专业"}} 的语气总结以下文本：

{{text}}

返回包含摘要和情感（正面/负面/中性）的 JSON。

// 加载并使用 dotprompt
const summarizePrompt = ai.prompt('summarize');
const { output } = await summarizePrompt({
  text: '此处是文章内容...',
  style: '随意',
});

RAG — 检索增强生成

import { devLocalVectorstore } from '@genkit-ai/dev-local-vectorstore';
import { textEmbedding004 } from '@genkit-ai/google-genai';

const ai = genkit({
  plugins: [
    googleAI(),
    devLocalVectorstore([{
      indexName: 'documents',
      embedder: textEmbedding004,
    }]),
  ],
});

// 索引文档
await ai.index({
  indexer: devLocalVectorstoreIndexer('documents'),
  docs: [
    { content: [{ text: '文档 1 内容...' }], metadata: { source: 'doc1' } },
    { content: [{ text: '文档 2 内容...' }], metadata: { source: 'doc2' } },
  ],
});

// RAG 流程
export const ragFlow = ai.defineFlow(
  {
    name: 'ragFlow',
    inputSchema: z.object({ question: z.string() }),
    outputSchema: z.string(),
  },
  async ({ question }) => {
    // 检索相关文档
    const docs = await ai.retrieve({
      retriever: devLocalVectorstoreRetriever('documents'),
      query: question,
      options: { k: 3 },
    });

    // 基于检索到的文档生成答案
    const { text } = await ai.generate({
      system: '仅使用提供的上下文回答问题。',
      prompt: question,
      docs,
    });

    return text;
  }
);

聊天会话

export const chatFlow = ai.defineFlow(
  {
    name: 'chatFlow',
    inputSchema: z.object({ message: z.string(), sessionId: z.string() }),
    outputSchema: z.string(),
  },
  async ({ message, sessionId }) => {
    const session = ai.loadSession(sessionId) ?? ai.createSession({ sessionId });
    const chat = session.chat({
      system: '你是一个乐于助人的助手。',
    });

    const { text } = await chat.send(message);
    return text;
  }
);

多代理系统

// 专业代理
const researchAgent = ai.defineFlow(
  { name: 'researchAgent', inputSchema: z.string(), outputSchema: z.string() },
  async (query) => {
    const { text } = await ai.generate({
      system: '你是一位研究专家。收集事实并引用来源。',
      prompt: query,
      tools: [searchWebTool],
    });
    return text;
  }
);

const writerAgent = ai.defineFlow(
  { name: 'writerAgent', inputSchema: z.string(), outputSchema: z.string() },
  async (brief) => {
    const { text } = await ai.generate({
      system: '你是一位专业作家。写出清晰、引人入胜的内容。',
      prompt: brief,
    });
    return text;
  }
);

// 编排器委托给专业代理
export const contentPipelineFlow = ai.defineFlow(
  {
    name: 'contentPipelineFlow',
    inputSchema: z.object({ topic: z.string() }),
    outputSchema: z.string(),
  },
  async ({ topic }) => {
    const research = await researchAgent(`研究：${topic}`);
    const article = await writerAgent(`根据以下内容写一篇文章：${research}`);
    return article;
  }
);

开发者工具

CLI 命令

# 启动开发者 UI 并连接到您的应用
genkit start -- npx tsx --watch src/index.ts
genkit start -o -- npx tsx src/index.ts    # 自动打开浏览器

# 从 CLI 运行特定流程
genkit flow:run summarizeFlow '{"text": "Hello world", "maxWords": 10}'

# 以流式输出运行
genkit flow:run streamingFlow '{"topic": "AI"}' -s

# 评估一个流程
genkit eval:flow ragFlow --input eval-inputs.json

# 查看所有命令
genkit --help

# 禁用分析遥测
genkit config set analyticsOptOut true

开发者 UI

开发者 UI 运行在 http://localhost:4000，提供以下功能：

• 流程运行器：使用自定义 JSON 输入执行流程
• 追踪检查器：可视化每个步骤（生成、嵌入、检索、工具调用）
• 提示词演练场：交互式测试提示词
• 模型测试器：比较不同模型的输出
• 评估器：对流程运行评估数据集

# 为方便添加 npm 脚本
# package.json
"scripts": {
  "genkit:dev": "genkit start -- npx tsx --watch src/index.ts"
}

npm run genkit:dev

部署

Firebase Cloud Functions

import { onCallGenkit } from 'firebase-functions/https';
import { defineSecret } from 'firebase-functions/params';

const apiKey = defineSecret('GOOGLE_AI_API_KEY');

export const summarize = onCallGenkit(
  { secrets: [apiKey] },
  summarizeFlow
);

firebase deploy --only functions

Express.js 服务器

import express from 'express';
import { expressHandler } from 'genkit/express';

const app = express();
app.use(express.json());

app.post('/summarize', expressHandler(summarizeFlow));
app.post('/chat', expressHandler(chatFlow));

app.listen(3000, () => console.log('服务器运行在端口 3000'));

Cloud Run

# 构建并部署
gcloud run deploy genkit-app \
  --source . \
  --region us-central1 \
  --set-env-vars GEMINI_API_KEY=$GEMINI_API_KEY

支持的插件

模型提供商

向量数据库

最佳实践

1. 始终定义输入/输出模式 — 使用 Zod 对象为开发者 UI 提供带标签的字段，并确保 API 安全
2. 对所有 AI 逻辑使用流程 — 即使是简单的调用；流程免费为您提供追踪和部署能力
3. 将 API 密钥存储在环境变量中 — 绝不硬编码；对于生产环境使用 Firebase Secrets
4. 使用 ai.run() 追踪自定义步骤 — 将非 Genkit 代码包裹在 ai.run() 中以获得追踪可见性
5. 流式处理长内容 — 使用带有 streamSchema + sendChunk 的 defineFlow 以获得更好的用户体验
6. 用代理分离关注点 — 专业的子流程 > 一个庞大的流程
7. 对团队提示词使用 Dotprompt — .prompt 文件支持版本控制、审查和重用

约束条件

必须做

• 为所有流程输入和输出定义模式
• 处理 generate() 返回的 null 输出 — 抛出有意义的错误
• 在与开发服务器分开运行流程时，设置 GENKIT_ENV=dev
• 部署到 Firebase 时，使用 onCallGenkit（而不是原始的 Cloud Functions）

绝不能做

• 绝不在源代码中硬编码 API 密钥
• 如果需要追踪/可观测性，不要在流程外使用 generate()
• 不要不带命令地调用 genkit start — 始终传递 -- <your-run-command>
• 避免在工具处理程序中阻塞事件循环 — 使用 async/await

参考链接

• 官方文档
• 入门指南
• 开发者工具
• 流程参考
• 工具调用
• RAG 指南
• 多代理系统
• Dotprompt
• GitHub 仓库
• API 参考

示例

示例 1：最小化流程

import { googleAI } from '@genkit-ai/google-genai';
import { genkit, z } from 'genkit';

const ai = genkit({ plugins: [googleAI()] });

export const helloFlow = ai.defineFlow(
  {
    name: 'helloFlow',
    inputSchema: z.object({ name: z.string() }),
    outputSchema: z.string(),
  },
  async ({ name }) => {
    const { text } = await ai.generate(`用创意的方式向 ${name} 打招呼。`);
    return text;
  }
);

// 运行它
const greeting = await helloFlow({ name: 'World' });
console.log(greeting);

示例 2：完整的 RAG + 代理管道

import { googleAI, textEmbedding004 } from '@genkit-ai/google-genai';
import { devLocalVectorstore } from '@genkit-ai/dev-local-vectorstore';
import { genkit, z } from 'genkit';

const ai = genkit({
  plugins: [
    googleAI(),
    devLocalVectorstore([{ indexName: 'kb', embedder: textEmbedding004 }]),
  ],
});

// 索引知识库文档
const indexKnowledgeBase = ai.defineFlow(
  { name: 'indexKB', inputSchema: z.array(z.string()) },
  async (texts) => {
    await ai.index({
      indexer: devLocalVectorstoreIndexer('kb'),
      docs: texts.map(text => ({ content: [{ text }] })),
    });
  }
);

// 使用 RAG 回答问题
export const answerFlow = ai.defineFlow(
  {
    name: 'answerFlow',
    inputSchema: z.object({ question: z.string() }),
    outputSchema: z.object({ answer: z.string(), sources: z.number() }),
  },
  async ({ question }) => {
    const docs = await ai.retrieve({
      retriever: devLocalVectorstoreRetriever('kb'),
      query: question,
      options: { k: 5 },
    });

    const { text } = await ai.generate({
      system: '仅从提供的上下文中回答。如果不确定，请说明。',
      prompt: question,
      docs,
    });

    return { answer: text, sources: docs.length };
  }
);

示例 3：多模型比较

import { googleAI } from '@genkit-ai/google-genai';
import { openAI } from 'genkitx-openai';
import { genkit, z } from 'genkit';

const ai = genkit({ plugins: [googleAI(), openAI()] });

export const compareModelsFlow = ai.defineFlow(
  {
    name: 'compareModelsFlow',
    inputSchema: z.object({ prompt: z.string() }),
    outputSchema: z.object({ gemini: z.string(), gpt4o: z.string() }),
  },
  async ({ prompt }) => {
    const [geminiResult, gptResult] = await Promise.all([
      ai.generate({ model: googleAI.model('gemini-2.5-flash'), prompt }),
      ai.generate({ model: 'openai/gpt-4o', prompt }),
    ]);

    return {
      gemini: geminiResult.text,
      gpt4o: gptResult.text,
    };
  }
);