企业官网建设流程全解析

Genkit Reflection 协议 V2 深度解析：基于 WebSocket 与 JSON-RPC 2.0 的双向反射架构

【免费下载链接】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 仓库中的 docs/reflection-v2-protocol.md 为骨架，结合 genkit-tools（CLI/Manager）、JS 与 Go 运行时中的 V2 实现源码，系统讲解 Genkit Reflection V2 协议的设计动机、传输层规范、JSON-RPC 2.0 消息格式、流式扩展、五大协议方法（register / listActions / listValues / runAction / cancelAction）的完整参数语义、健康检查机制，以及从 V1 到 V2 的迁移要点。读完本文，你将能够理解 Genkit CLI 与用户应用（Runtime）之间基于 WebSocket 的反射通道是如何建立、保活、断线重连并承载流式执行的，也能据此在自己的 Runtime 实现中正确实现或对接 V2 协议。

一、为什么需要 V2：从「Runtime 开 HTTP 服务」到「CLI 起 WebSocket 服务」

Reflection（反射）是 Genkit 开发者工具链的核心机制：CLI（Runtime Manager）需要枚举用户应用中注册的 Action（Flow、Model、Tool 等）、执行它们、读取输出，从而驱动 Dev UI、genkit start等交互。V1 与 V2 的架构对比，官方文档给出了精炼的对照：

V2 的核心变化是连接方向反转：

• Server 角色：Genkit CLI（RuntimeManagerV2）启动一个 WebSocket 服务器；
• Client 角色：Genkit Runtime（用户应用）作为 WebSocket 客户端连接到 CLI 的服务器。

这一设计带来两个直接收益：

1. 一个 CLI 管理多个 Runtime：对于多服务（multi-service）项目，多个 Runtime 可以各自建立一条 WebSocket 连接，CLI 侧通过runtimeId区分与路由（见下文listRuntimes/getRuntimeById）；
2. Runtime 不再需要自行管理 HTTP 服务与端口：反射能力不再依赖 Runtime 进程内额外的监听端口，降低了用户应用的部署复杂度。

二、传输层与消息结构

2.1 传输规范

所有消息都遵循 JSON-RPC 2.0 规范：每条消息都携带jsonrpc: "2.0"版本字段，并依据是否携带id区分请求/响应与通知。

2.2 Request（请求）

Manager 向 Runtime 发起调用时发送请求，id由发送方（Manager）生成：

{ "jsonrpc": "2.0", "method": "methodName", "params": { ... }, "id": 1 }

id生成规则（源码可证）：id可以是数字（自增计数）或字符串（UUID），在同一个 WebSocket 会话内对每个 pending 请求必须唯一。查看 manager-v2.ts 的实现：RuntimeManagerV2内部维护pendingRequests: Map<number | string, ...>，通过requestIdCounter自增生成请求 ID（(++this.requestIdCounter).toString()）；而 Go 运行时 reflection_v2.go 使用atomic.Uint64自增序列生成字符串 ID。

2.3 Response（成功响应）

{ "jsonrpc": "2.0", "result": { ... }, "id": 1 }

响应通过id与请求对应。Manager 侧收到响应后，会依据请求方法对result做 Zod Schema 校验，例如listActions会经过ReflectionListActionsResponseSchema校验，甚至对 Go 运行时可能携带的metadata: null做归一化处理（见 manager-v2.ts 的normalizeListActionsResult）。

2.4 Response（错误响应）

{ "jsonrpc": "2.0", "error": { "code": -32000, "message": "Error message", "data": { "code": 13, "message": "Error message", "details": { "traceId": "...", "stack": "..." } } }, "id": 1 }

error.data中是一个与 V1 API 对齐的Status对象，包含：

• code：Genkit 规范状态码（例如 13 表示 INTERNAL、3 表示 INVALID_ARGUMENT）；
• message：错误信息；
• details：附加上下文，包括traceId与stack堆栈。

JSON-RPC 2.0 标准错误码在三端实现中保持一致（见 reflection_v2.go）：

以runAction失败为例，Go 运行时的sendRunActionError（reflection_v2.go）会把底层错误转换为status.Error并装配上述Status形状的data：若错误链中存在context.Canceled，则强制将状态码置为status.Cancelled（方便 Dev UI 区分「用户主动取消」与「执行失败」），并尽力提取traceId与stack写入details。

2.5 Notification（通知）

不带id的请求即为通知，发送方不期待任何响应：

{ "jsonrpc": "2.0", "method": "methodName", "params": { ... } }

Runtime 侧的sendNotification（reflection_v2.go）发送的消息ID字段留空，Manager 侧的sendNotification（manager-v2.ts）同样构造不带id的请求帧——这正是流式扩展得以实现的基础。

三、流式扩展：用 Notification 弥补 JSON-RPC 2.0 的短板

JSON-RPC 2.0 原生不支持流式响应。V2 协议通过在 Runtime -> Manager 方向上使用与特定 Request ID 关联的 Notification来扩展流式能力：

3.1 Stream Chunk Notification

{ "jsonrpc": "2.0", "method": "streamChunk", "params": { "requestId": 1, "chunk": { ... } } }

Manager 侧收到后通过streamCallbacks映射找到对应的StreamingCallback并调用之（manager-v2.ts）。JS 运行时在runAction的流式分支中，把action.run的onChunk回调逐块转换为streamChunk通知（reflection-v2.ts）。

3.2 Run Action State Notification

{ "jsonrpc": "2.0", "method": "runActionState", "params": { "requestId": 1, "state": { "traceId": "..." } } }

该通知用于在 span 启动时尽早把 trace ID 推送给 Manager，使 Dev UI 能在 action 尚未结束时就开始关联追踪数据。Go 实现的runActionTelemetry.callback（reflection_v2.go）在根 span 启动回调中记录 traceId、注册可取消的 active action，并立即发送runActionState通知；Manager 侧通过traceIdCallbacks把 traceId 回传给调用方（manager-v2.ts）。

四、协议方法总览

除此之外，从双端实现源码可以看到还有一个文档方法表之外的通知方法configure（Manager -> Runtime，携带telemetryServerUrl，用于在注册之后动态下发遥测服务器地址，见 manager-v2.ts 与 reflection-v2.ts），以及与双向流式输入相关的sendInputStreamChunk/endInputStream（Manager -> Runtime，见本文第六节）。

五、详细 API 规范

5.1 Registration（注册）

方向：Runtime -> Manager类型：Request

参数：

结果：

源码佐证（参数组装）：

• JS 运行时在连接建立后立即注册（reflection-v2.ts）：id取自process.env.GENKIT_RUNTIME_ID或默认的${process.pid}[-index]，pid为process.pid，genkitVersion为nodejs/${GENKIT_VERSION}，envs默认['dev']；若注册响应的telemetryServerUrl非空且环境变量GENKIT_TELEMETRY_SERVER未设置，则调用setTelemetryServerUrl完成遥测握手。
• Go 运行时（reflection_v2.go）的register同样组装这些字段：runtimeID取自GENKIT_RUNTIME_ID环境变量，缺省时回退为os.Getpid()；genkitVersion为"go/" + internal.Version；reflectionApiSpecVersion使用internal.GENKIT_REFLECTION_API_SPEC_VERSION。
• 协议版本号reflectionApiSpecVersion当前取值为1，在 manager.ts 与 version.go 中均有常量定义。
• Manager 侧handleRegister（manager-v2.ts）用 Zod SchemaReflectionRegisterParamsSchema（reflection.ts）校验参数，构造RuntimeInfo存入runtimesMap，触发RuntimeEvent.ADD事件，并把telemetryServerUrl作为成功结果返回。

5.2 List Actions（列出 Action）

方向：Manager -> Runtime类型：Request

参数：void

结果：

源码佐证：JS 运行时handleListActions（reflection-v2.ts）从 registry 获取所有可解析 Action，将key / name / description / metadata / inputSchema / outputSchema（JSON Schema 由toJsonSchema转换）逐项写入响应；Go 运行时则用listResolvableActions枚举后原样组装（reflection_v2.go）。Manager 侧收到后会补全缺失的action.key（manager-v2.ts）。

5.3 List Values（列出 Value）

方向：Manager -> Runtime类型：Request

参数：

结果：

实现限制（以当前源码为准）：虽然文档示例中type可以是"model"、"prompt"、"schema"，但当前 JS 与 Go 运行时实现均只接受"defaultModel"与"middleware"两种取值，其余类型返回-32602Invalid params（见 reflection-v2.ts 与 reflection_v2.go）。Go 注册表目前不按类型细分，type参数被接受但忽略，仅用于保持与 JS 侧一致的错误形状。

5.4 Run Action（执行 Action）

方向：Manager -> Runtime类型：Request

参数：

结果（非流式）：

Manager 侧构造请求时会自动推导这两个布尔标志：stream: !!streamingCallback、streamInput: !!inputStream（manager-v2.ts）。运行时侧用ReflectionRunActionParamsSchema（reflection.ts，即RunActionRequestSchema.extend({ stream, streamInput })）校验入参。

流式流程（Streaming Flow）：

1. Runtime 可选地发送runActionState通知；
2. Runtime 发送streamChunk通知；
3. Runtime 发送携带result的最终响应（结构与非流式一致）。

JS 运行时在流式分支执行完毕后会先await flushTracing()再发送最终响应，确保追踪数据已落盘（reflection-v2.ts）。

双向流式流程（Bidirectional Streaming Flow，streamInput: true）：

1. Manager 发送sendInputStreamChunk通知；
2. Manager 发送endInputStream通知；
3. Runtime 按上述流式流程继续。

这两个输入方向的 Notification 参数定义为{ requestId, chunk }与{ requestId }（reflection.ts）。Go 运行时的实现尤为细致（reflection_v2.go）：readLoop会在分发runAction之前预注册bidi session（探测到streamInput: true即注册），使得先于 action 初始化完成的输入块被缓冲而非丢弃；bidiSession内部使用sync.Cond+ 事件队列按到达顺序把块投递给api.BidiJSONConnection，队列无界但从不阻塞 WebSocket 读循环，以保证cancelAction始终可被及时处理。

5.5 Cancel Action（取消 Action）

方向：Manager -> Runtime类型：Request

参数：

结果：

源码佐证：JS 运行时维护activeActions: Map<traceId, { abortController, startTime }>（reflection-v2.ts），handleCancelAction依据 traceId 找到对应条目后调用abortController.abort()并返回{ message: 'Action cancelled' }，找不到则返回-32602错误（reflection-v2.ts）。Go 运行时等价实现见handleCancelAction（reflection_v2.go），取消通过action.cancel()（即 runAction 的context.CancelFunc）触发。

六、健康检查与连接保活

连接生命周期管理在实现中体现得比文档描述更完整：

• 断线清理：Manager 侧handleDisconnect（manager-v2.ts）在连接关闭时移除对应 Runtime 并触发RuntimeEvent.REMOVE；Go 运行时drainPending（reflection_v2.go）会让所有未决请求以「connection closed」错误立即失败，避免调用方无限阻塞。
• 指数退避重连：JS 客户端以 500ms 为基数、5s 为上限做指数退避重连（baseDelayMs = 500、maxDelayMs = 5000，见 reflection-v2.ts）；Go 客户端使用完全一致的重连参数，并通过reconnectBaseDelay << attempt计算退避间隔（reflection_v2.go）。
• 中断的输入流兜底：连接意外断开时，JS 运行时closeInputStreams/failInputStreams会关闭/报错所有进行中的输入 Channel，防止 action 体永久挂起（reflection-v2.ts）；Go 运行时closeBidiSessions也会为每个在途 bidi session 入队结束标记，让 action 优雅收尾（reflection_v2.go）。
• 请求超时：Manager 侧每个 pending 请求默认 30 秒超时，超时后从pendingRequests移除并 reject（manager-v2.ts）。

七、如何启用 V2（以当前仓库为准）

V2 仍处于实验阶段，CLI 通过--experimental-reflection-v2开关启用（start.ts）：

genkit start --experimental-reflection-v2 -- <your-app-command>

启用后的关键行为（见 manager-utils.ts）：

1. CLI 通过getPort({ port: makeRange(3200, 3400) })在 3200~3400 区间挑选一个空闲端口启动 WebSocket 服务器；
2. 以环境变量GENKIT_REFLECTION_V2_SERVER=ws://localhost:<port>注入被启动的 Runtime 进程（RuntimeManagerV2.startWebSocketServer会在启动时打印Starting reflection server: ws://localhost:<port>，见 manager-v2.ts）；
3. Runtime 进程启动时读取该环境变量并作为 WebSocket 客户端连入（JS 侧见 reflection-v2.ts），连接建立后立即发起register；
4. 若未指定端口，RuntimeManagerV2 同样使用 3200~3400 区间自动选择端口（manager-v2.ts）。

此外 CLI 还提供--write-env-file <file>选项，把上述环境变量以.env格式写入指定文件，便于在外部启动的 Runtime 进程中手动注入（start.ts）。

八、双端实现速览与源码地图

V2 协议在仓库中的实现横跨「工具链（Manager）」与「运行时（Runtime）」两侧，相关源码路径如下：

• Manager（CLI 侧）：
  • RuntimeManagerV2 核心实现：WebSocket 服务器、register/streamChunk/runActionState处理、pending 请求与流式回调路由、bidi 输入转发；
  • 协议参数与响应 Schema：全部方法入参/出参的 Zod 定义；
  • 启动命令与端口分配、环境变量注入；
  • Manager 单测 与 start 命令测试。
• JS Runtime：reflection-v2.ts（WebSocket 客户端、注册、方法分发、流式与 bidi 输入 Channel），配套测试 reflection-v2_test.ts。
• Go Runtime：reflection_v2.go（连接生命周期、readLoop分发、bidi session 事件队列、错误归一化），配套测试 reflection_v2_test.go。

九、兼容性与迁移注意事项

V1 与 V2 的关键差异可归结为一点：谁是 Server、谁是 Client。V1 中 Runtime 启动 HTTP Server 供 CLI 轮询；V2 中 CLI 启动 WebSocket Server，Runtime 建立持久连接。

官方文档明确指出：CLI 将根据配置决定使用哪种模式（例如--experimental-reflection-v2开关）。这意味着在 V2 正式化之前，V1 仍作为默认/兜底通道存在，两端需要保持对reflectionApiSpecVersion的协商能力（当前协议版本为1）。迁移到 V2 的实现者需要注意：

• Runtime 需要实现主动外连 + 自动重连逻辑，而非被动监听端口；
• 流式输出不再是 HTTP 分块响应，而是与 Request ID 绑定的streamChunk/runActionState通知；
• 双向流式（bidi）依赖sendInputStreamChunk/endInputStream通知通道，并需要考虑连接中断时输入流的兜底收尾；
• 错误上报必须携带Status形状的data（code/message/details），以便 Dev UI 统一展示规范状态码与堆栈。

十、结语

Reflection V2 通过「连接方向反转 + WebSocket 持久连接 + JSON-RPC 2.0 流式扩展」三项设计，把 Genkit 开发者工具的反射通道从「Runtime 被动暴露 HTTP 接口」升级为「CLI 主动聚合多 Runtime 的会话式通道」，为多服务项目编排、流式生成预览、双向 Agent 会话等场景提供了统一且低开销的传输基础。无论是实现新的 Runtime 语言绑定，还是理解genkit start与 Dev UI 背后的通信机制，本文梳理的协议规范与源码映射都将是可靠的起点。

【免费下载链接】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