Genkit Next.js 集成实战：使用 `@genkit-ai/next` 将流式 Flow 暴露为 Next.js API 路由

Genkit Next.js 集成实战：使用@genkit-ai/next将流式 Flow 暴露为 Next.js API 路由

【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit

本篇技术指南以 Genkit 官方仓库中的 Next.js 集成示例应用 为主体，完整讲解如何将 Genkit Flow 通过@genkit-ai/next插件暴露为 Next.js App Router 的 API Route，并实现逐 token 流式响应。读完本文，你将掌握示例应用从环境准备、Flow 定义、路由挂载到浏览器端流式消费的完整链路，并能理解appRoute底层的 SSE（Server-Sent Events）实现与 OTel 根 Span 检测禁用的原因，为在真实 Next.js 项目中接入 Genkit 提供可直接复用的工程方案。

示例应用概览：一个可运行的 "讲笑话" 演示

仓库中的 js/testapps/next 是一个名为nextjs-sample的示例应用，其用途正如示例根目录 README 所述：用于测试@genkit-ai/next插件，演示将 Genkit Flow 作为 Next.js API 路由使用并支持流式输出。它不是一个抽象的概念演示，而是一个依赖workspace:*本地包（genkit、@genkit-ai/google-genai、@genkit-ai/next）的真实可运行工程，其依赖与脚本配置见 js/testapps/next/package.json。

示例应用演示的核心功能可归纳为下表：

环境准备与 API Key 配置

前置要求

• Node.js（v18 或更高版本）
• pnpm包管理器（仓库采用 pnpm workspace 管理多语言、多包依赖）

配置 Gemini API Key

示例默认使用 Google Gemini 模型（具体为gemini-flash-latest，见 js/testapps/next/src/genkit/index.ts），因此需要先导出 API Key 环境变量：

export GEMINI_API_KEY='<your-api-key>'

构建与安装依赖

示例通过 pnpm workspace 引用仓库内的本地包，因此需从仓库根目录执行安装与初始化：

pnpm install pnpm run setup

pnpm install会按 pnpm-workspace.yaml 的声明一次性安装所有 workspace 包；setup则负责完成各子包的构建/初始化步骤，确保@genkit-ai/next等本地依赖处于可用状态。如果你只想单独验证 Next.js 示例，js/testapps/next/package.json中还提供了vendor脚本，可将../../plugins/next下的插件打包为 tgz 并安装到本示例中。

运行示例

pnpm run dev

该命令等价于next dev，启动后 Next.js 开发服务器监听在http://localhost:3000。

浏览器打开首页后，你可以在输入框中填写笑话类型（如cat、dad），点击Run走普通 JSON 请求，点击Stream体验逐 token 流式打字机效果——这正是验证 Genkit 流式能力的最直观方式。

代码结构剖析：从 Flow 定义到路由挂载

示例应用的前端代码规模很小但结构清晰，全部源码位于 js/testapps/next/src：

src/ ├── app/ │ ├── api/joke/route.ts # API 路由：将 tellJoke Flow 暴露为 POST 接口 │ ├── layout.tsx / page.tsx # 前端页面与客户端流式消费逻辑 │ └── globals.css / page.module.css └── genkit/ ├── index.ts # Genkit 实例（AI 配置、插件、模型） └── joke.ts # tellJoke 流式 Flow 定义

第一步：配置 Genkit 实例

js/testapps/next/src/genkit/index.ts 是服务端 Genkit 配置的入口：

import { googleAI } from '@genkit-ai/google-genai'; import { genkit } from 'genkit'; import { disableOTelRootSpanDetection } from 'genkit/tracing'; disableOTelRootSpanDetection(); export const ai = genkit({ plugins: [googleAI()], model: googleAI.model('gemini-flash-latest'), });

两个关键点：

1. 模型与插件：通过googleAI()插件接入 Google AI 服务，并将默认模型设为gemini-flash-latest（使用 Gemini 快速闪存模型）。示例中的模型 ID 以仓库实际配置为准，生产环境建议显式指定具体的模型版本。
2. disableOTelRootSpanDetection()：这一行代码在示例 README 中被特别强调——Next.js Edge 运行时要求禁用 OTel 根 Span 检测。其源码实现位于 js/core/src/tracing/instrumentation.ts：它通过设置全局标记global['__genkit_disableRootSpanDetection'] = true，让 Genkit 不再自定义 OTel 根 Span，而是保留 OpenTelemetry 默认根 Span，从而兼容 Edge Runtime 对 trace 结构的要求。该函数在源码注释中明确标注为 "unstable"，未来实现细节可能变化，但在 Next.js 场景下这一行配置是必要的。

第二步：定义流式 Flow

js/testapps/next/src/genkit/joke.ts 定义了核心的tellJokeFlow：

import { z } from 'genkit'; import { ai } from '.'; export const tellJoke = ai.defineFlow( { name: 'tellJoke', inputSchema: z.string().nullable(), outputSchema: z.string(), streamSchema: z.string(), }, async (type: string | null, { sendChunk }) => { const { response, stream } = ai.generateStream( `Tell me a ${type || 'dad'} joke` ); for await (const chunk of stream) { sendChunk(chunk.text); } return (await response).text; } );

值得注意的细节：

• 三个 Schema 齐备：inputSchema接受可空字符串（不填类型时默认生成 "dad joke"），outputSchema声明最终返回字符串，streamSchema声明每个流式分片的数据形态。这正是 Genkit 对可流式 Action 的完整约定——streamSchema的存在让框架能够为流式请求做类型校验与序列化。
• 流式生成的核心写法：ai.generateStream()返回{ response, stream }二元组。循环内对stream逐块取出chunk.text并通过回调参数sendChunk推送；循环结束后await response拿到完整最终文本并作为 Flow 返回值。sendChunk与返回值在appRoute中分别对应 SSE 的data: {message}事件与data: {result}事件（详见下文底层原理）。

第三步：用appRoute()挂载为 API 路由

js/testapps/next/src/app/api/joke/route.ts 全文件只有三行有效逻辑：

import { tellJoke } from '@/genkit/joke'; import { appRoute } from '@genkit-ai/next'; export const POST = appRoute(tellJoke);

这里遵循 Next.js App Router 的 Route Handler 约定：导出POST函数即声明该路由接受 POST 请求。appRoute(tellJoke)返回一个接收NextRequest、返回NextResponse的标准路由处理器，Genkit 自动完成请求体解析、Schema 校验、Flow 执行与响应序列化。也就是说，一个 Flow 从定义到成为 HTTP 接口，只需要这一行挂载代码。

第四步：前端页面消费 Flow

js/testapps/next/src/app/page.tsx 展示了客户端如何调用服务端暴露的 Flow：

• 普通调用：使用runFlow，POST 到/api/joke，input传入笑话类型（空串映射为null），一次性拿到完整结果：

const resp = await runFlow<typeof tellJoke>({ url: '/api/joke', input: type === '' ? null : type, }); setResponse(resp);

• 流式调用：使用streamFlow，对返回的stream做for await遍历，把每个分片累加到界面上实现打字机效果；output是 Promise，需要await才能拿到最终完整结果：

const { stream, output } = streamFlow<typeof tellJoke>({ url: '/api/joke', input: type === '' ? null : type, }); for await (const chunk of stream) { accum = accum + chunk; setResponse(accum); } setResponse(await output);

<typeof tellJoke>是类型层面的关键技巧：runFlow/streamFlow都接收 Action 类型参数，从而让input、chunk、output的 TypeScript 类型全部从 Flow 定义中自动推导，前后端共享同一份类型契约。

深入appRoute：SSE 协议与流式响应原理

appRoute的实现位于 js/plugins/next/src/index.ts，理解它能让你在排查问题时得心应手。其核心逻辑如下：

1. 根据Accept头区分流式与非流式请求

const streamId = req.headers.get('x-genkit-stream-id'); if (req.headers.get('accept') !== 'text/event-stream') { // 非流式：直接执行 action.run，返回 JSON const resp = await action.run(input, { context, abortSignal: req.signal, init }); return NextResponse.json({ result: resp.result }); } // 流式：走 SSE 响应分支

• 客户端请求头Accept: text/event-stream时，服务端返回SSE 流（Content-Type: text/event-stream），分片以data: {"message": ...}事件推送，结束时发送data: {"result": ...}与END标记；
• 否则返回标准application/json，响应体为{ result: ... }。
• 插件测试 js/plugins/next/tests/index_test.ts 中专门构造了两种请求对比验证（如 "supports scalars" 用例），并提供了把 SSE 文本解析为结构化chunks的工具函数，可作为理解协议格式的参考。

2. 流式响应的实现细节

流式分支使用 Web Streams API 的TransformStream逐块写入：

const encoder = new TextEncoder(); const { readable, writable } = new TransformStream(); const writer = writable.getWriter(); // ... onChunk: (chunk) => { writer.write(encoder.encode(`data: ${JSON.stringify({ message: chunk })}\n\n`)); }, onDone: (output) => { writer.write(encoder.encode(`data: ${JSON.stringify({ result: output })}\n\n`)); writer.write(encoder.encode('END')); writer.close(); },

响应头固定包含：

Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive Transfer-Encoding: chunked

3. 错误处理策略：对外隐藏内部错误

实现中有两处刻意设计的错误边界，非常值得生产项目借鉴：

• 用户可见错误（UserFacingError，如INVALID_ARGUMENT）以 JSON 或 SSEerror:事件原样返回，并映射正确的 HTTP 状态码（非流式 400，流式仍为 200 但事件内携带错误）；
• 内部错误（普通Error）只console.error记录日志，响应统一为INTERNAL / Internal Error，测试用例 "hides internal errors" 明确验证了这一点——防止敏感堆栈信息泄露给客户端。

4. 请求取消与上下文传递

action.run接收abortSignal: req.signal，将 Next.js 请求的取消信号桥接给 Genkit 执行引擎，客户端断开时服务端可及时终止模型生成。同时，appRoute支持通过opts.contextProvider注入genkit/context的上下文提供器（如apiKey()鉴权），在 Flow 执行前从请求头提取认证等信息，测试用例 "supports context providers" 覆盖了鉴权成功、鉴权失败（403）以及流式场景下鉴权失败返回 SSE 错误事件的完整路径。

进阶能力：Durable Streaming（持久化流式）

示例 README 聚焦于基础流式，而@genkit-ai/next插件还支持Durable Streaming（Beta），详见 js/plugins/next/README.md：客户端断线后可以凭streamId重新连接到同一流，不丢失已产生的状态。

启用方式是在appRoute选项中传入streamManager：

import { myFlow } from '@/genkit/myFlow'; import { appRoute } from '@genkit-ai/next'; import { InMemoryStreamManager } from 'genkit/beta'; export const POST = appRoute(myFlow, { streamManager: new InMemoryStreamManager(), });

• InMemoryStreamManager适用于开发与测试；生产环境应改用持久化实现，例如@genkit-ai/firebase插件提供的FirestoreStreamManager/RtdbStreamManager，或自行实现StreamManager接口。
• 客户端通过streamFlow的返回结果拿到streamId保存，断线后携带该 ID 重新发起streamFlow即可续接：

const result = streamFlow({ url: '/api/myDurableFlow', input: 'tell me a long story' }); const streamId = await result.streamId; // 保存此 ID // ... 之后需要重连时 ... const reconnectedResult = streamFlow({ url: '/api/myDurableFlow', streamId });

从 js/plugins/next/src/index.ts 源码可以看到重连机制的实现：请求头携带x-genkit-stream-id时，appRoute会调用streamManager.subscribe(streamId)订阅既有流；若流不存在则返回204 No Content（对应测试用例 "returns 204 for unknown streamId"）。首次发起流式请求时，服务端会生成新的 UUID 作为streamId并通过x-genkit-stream-id响应头返回给客户端。

初始化数据（Init Data）支持

对于定义了initSchema的 Flow 或 Action，客户端可通过init选项传递初始化数据，例如会话 ID、模型参数等：

const result = await runFlow<typeof myFlow>({ url: '/api/myFlow', input: 'say hello', init: { sessionId: 'abc123', config: { temperature: 0.7 } }, });

请求体中init与data一同发送，服务端会在 Flow 执行前用initSchema校验（js/plugins/next/tests/index_test.ts 中的 "rejects init that does not conform to initSchema" 用例验证了不合规的init会返回400 INVALID_ARGUMENT，且校验失败发生在 Flow 运行之前），校验通过后才将init注入 Action 上下文。

验证与预期行为

示例 README 给出了两条验证途径：

1. 通过 API 路由测试 joke flow：直接以POST请求/api/joke（路径取决于你的路由配置），带Accept: text/event-stream头观察 SSE 流，或不带头观察 JSON 结果。
2. 预期行为（可在本地实测确认）：
   • Genkit Flow 可以无缝充当 Next.js API Route Handler；
   • 流式响应逐 token 增量送达（浏览器页面点击 Stream 可见打字机效果）；
   • 已禁用 OTel 根 Span 检测（Next.js Edge 运行时环境的必要条件）。

小结

本示例虽小，却串联起了 Genkit + Next.js 集成的全部关键链路：genkit()实例配置 →defineFlow定义可流式 Flow（streamSchema+sendChunk）→appRoute()一行挂载为 Route Handler → 前端runFlow/streamFlow类型安全消费。在此基础上，appRoute的 SSE 实现、错误隐藏策略、上下文注入、Durable Streaming 与 Init Data 支持构成了将 Genkit 深度嵌入 Next.js 应用的能力底座。你可以以此为模板，将tellJoke替换为检索增强生成、Agent 对话等真实业务 Flow，快速搭建具备流式体验的 AI 应用。

进一步阅读：插件源码 js/plugins/next/src/index.ts 与客户端库 js/plugins/next/src/client.ts，以及覆盖了鉴权、标量/对象入参、错误语义、持久化流式等场景的测试 js/plugins/next/tests/index_test.ts。

【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit