Action 原语与注册表:一切皆 Action

30 秒导读: Genkit 里没有"模型对象""工具对象""流程对象"这些各自为政的类型。它把它们全部压成同一个东西——Action:一个普通异步函数,外面包了四层能力(输入/输出 Zod 校验、每次调用开一个 OpenTelemetry span、可流式吐 chunk、可注入运行时 context)。再配一张按 /类型/名字 寻址的注册表。读懂这一章 = 拿到理解整个项目的钥匙:后面每一章的主角(generate、tool、flow、dotprompt……)都只是"某个 actionType 的 Action"。

---

1. 这是什么(零基础也能懂)

一句话定义: Action 是 Genkit 的地基抽象——一个把普通函数升级成"自描述、会校验、可观测、可远程调用"的可寻址单元。

它要解决的问题。 一个 AI 框架里有太多"可调用的东西":调大模型、跑一个工具、执行一段业务流程、渲染一个 prompt、查一个向量库。如果每种都单独设计一套类型、一套调用约定、一套可观测性接线,框架会迅速长成一堆互不相通的孤岛。

Genkit 的答案是归一化:与其做五种对象,不如只做一种原语,让这五种都变成"填了不同 actionType 的同一个 Action"。

它们共享的四样能力——这正是 action() 工厂免费送给每个被包装函数的东西:

能力	白话	谁提供
输入/输出校验	进来的参数、出去的结果都按 Zod schema 卡一遍,不合格直接抛错	parseSchema
可观测性	每次调用自动开一个 OpenTelemetry span,记下输入、输出、trace/span id	runInNewSpan
流式	函数体内可以边算边 sendChunk(...) 往外吐增量结果	ActionFnArg.sendChunk
context 注入	调用方可塞入 auth 等"侧信道"数据,函数体内用 context 读到	runWithContext

用起来什么样。 下面是一个最小的真实感示例,注意:你只是写了个普通函数,却立刻得到了校验 + 追踪 + 流式:

// 示意,非源码:一个 action 的最小样子
const upper = action(
  {
    actionType: 'custom',
    name: 'upper',
    inputSchema: z.string(),   // 进来必须是 string
    outputSchema: z.string(),  // 出去必须是 string
    streamSchema: z.string(),  // 流式 chunk 也是 string
  },
  async (input, { sendChunk, streamingRequested }) => {
    if (streamingRequested) sendChunk('working...'); // 想流式就吐 chunk
    return input.toUpperCase();
  }
);

await upper('hello');        // => 'HELLO'(且自动开了一个 span)
const { stream, output } = upper.stream('hello'); // 拿到流 + 最终结果

一句话直觉。 把 Action 想成"带说明书的 USB 设备":插进注册表这个"USB 集线器"后,谁都能按一个标准地址(/custom/upper)找到它、调用它、看它的输入输出规格、监控它的运行——不管它内部是模型、工具还是流程。

---

2. 顶层全景(它大概怎么转)

怎么读这张图: 从左到右是"定义 → 注册 → 调用"三步。上半部分是创建期(action() 造出一个 Action,defineAction() 顺手登记进注册表);下半部分是调用期(一次调用穿过校验、span、context、流式四道关)。

                     创建期
  你的普通函数 ──► action(config, fn) ──► Action(可调用函数 + __action 元数据)
        (fn)          ▲                         │
                      │                         │ defineAction 还会顺手:
       config: actionType / name /             ▼
       inputSchema / outputSchema        registry.registerAction
       / streamSchema                    键 = /<actionType>/<name>
                                                │
                                                ▼
                                    ┌────────────────────────┐
                                    │  Registry (actionsById) │  按 /type/name 寻址
                                    │  /model/googleai/gemini │
                                    │  /tool/myTool           │
                                    │  /flow/myFlow           │
                                    └────────────────────────┘
 ─────────────────────────────────────────────────────────────────────────
                     调用期:  await action(input, options)
                                                │
   input ─►①parseSchema(input) ─►②runInNewSpan ─►③runWithContext ─►④fn(input, args)
                 校验入参          开 span          注入 context      你的逻辑
                                  记 trace/span                      args.sendChunk 流式
                                                                      │
   output ◄─ ⑤parseSchema(output) ◄──────────────────────────────────┘
                 校验出参

部件一句话职责:

部件	干什么	文件
action()	工厂:把 (config, fn) 包成一个 Action(带 .run / .stream)	js/core/src/action.ts:450
defineAction()	action() + 立刻注册进 Registry	js/core/src/action.ts:692
defineActionAsync()	延迟(lazy)版:注册一个"将来才 resolve"的 Action	js/core/src/action.ts:835
Registry	存/查所有 Action、schema、plugin 的中央表	js/core/src/registry.ts:152
parseSchema()	用 Zod/JSON schema 校验数据,不合格抛 ValidationError	js/core/src/schema.ts:297
runInNewSpan()	开一个 OpenTelemetry span 包住这次调用	js/core/src/tracing/instrumentation.ts:62
runWithContext()	把 context 放进 AsyncLocalStorage,供调用栈里读取	js/core/src/context.ts:37

主线走一遍(高层): 你写一个函数 → action()/defineAction() 把它包成 Action 并(可选)登记进注册表 → 别处用 /type/name 从注册表查出它 → 调用时依次穿过"校验入参、开 span、注入 context、执行你的逻辑、校验出参",最后返回结果(以及一份 telemetry)。

---

3. 核心原理(逐个机制,由浅入深)

3.1 action():把函数包成 Action 的那层壳

它要解决的小问题。 怎么让"写业务逻辑"和"获得校验+追踪+流式"解耦?答案是:你只管写最内层的 fn,action() 在外面包一层调度壳。

思路。 工厂返回的不是一个对象,而是一个可直接调用的函数,同时在它身上挂了几个属性:__action(元数据)、.run(返回结果+telemetry)、.stream(返回流)。直接 await action(input) 其实是 .run 的薄封装,只取 result:

// 真实源码 js/core/src/action.ts:481-486
const actionFn = (async (input?, options?) => {
  return (await actionFn.run(input, options)).result;
}) as Action<...>;
actionFn.__action = { ...actionMetadata };

元数据里最关键的一行:注册键的约定。 名字被规范成 /<actionType>/<name>,这就是后面注册表寻址用的 key:

// 真实源码 js/core/src/action.ts:462-467
const actionName =
  typeof config.name === 'string'
    ? config.name
    : `${config.name.pluginId}/${config.name.actionId}`;
const actionMetadata = {
  key: `/${config.actionType}/${actionName}`,   // 例:/model/googleai/gemini-2.5-flash
  ...
};

调用期的完整管线都在 actionFn.run 里(js/core/src/action.ts:489)。它按顺序做五件事,下面几个小节逐一拆开。先看它的骨架:

run(input, options):
  ① parseSchema(input)             ← 3.2 校验入参(action.ts:495-515)
  ② runInNewSpan({...}, async () =>{  ← 3.3 开 span(action.ts:528)
       ③ 组装 ActionFnArg + runWithContext ← 3.4 context 注入(action.ts:564-588)
       ④ fn(input, { sendChunk, streamingRequested, context, trace, ... }) ← 3.5 流式
     })
  ⑤ parseSchema(output)            ← 校验出参(action.ts:600)
  return { result, telemetry:{traceId,spanId} }

3.2 Zod 校验:进出都卡一遍

它要解决的小问题。 大模型和外部世界给的数据经常"形状不对"。Action 在边界上把关:入参不合规就别进函数,出参不合规就别返回。

原理演示:

// 示意,非源码:校验的心智模型
function guardedRun(input) {
  input = parseSchema(input, { schema: inputSchema });   // 不合格 → 抛 ValidationError
  const output = realWork(input);
  return parseSchema(output, { schema: outputSchema });  // 出参也校验
}

真实实现。 入参校验发生在 run 开头(注意:流式输入走的是逐项校验的分支);出参校验在拿到结果之后:

// 真实源码 js/core/src/action.ts:495-500(入参)
if (config.inputSchema || config.inputJsonSchema) {
  if (!options?.inputStream) {
    input = parseSchema(input, {
      schema: config.inputSchema,
      jsonSchema: config.inputJsonSchema,
    });
  } else { /* 逐项校验流式输入 */ }
}

// 真实源码 js/core/src/action.ts:600-603(出参)
output = parseSchema(output, {
  schema: config.outputSchema,
  jsonSchema: config.outputJsonSchema,
});

parseSchema(js/core/src/schema.ts:297)本身是"先 validateSchema,不合格就抛 ValidationError,合格就原样返回"。校验器实际用 Ajv 或 @cfworker/json-schema(为了在 Cloudflare Workers 等边缘环境也能跑),并对编译后的 validator 做了 WeakMap 缓存。

关键细节: 校验只在声明了 schema 时才发生——没给 inputSchema 就直接放行。这让"随手糊一个无 schema 的 Action"也能用,校验是渐进增强而非强制。

3.3 OpenTelemetry span:每次调用自动可观测

它要解决的小问题。 Agent 系统的调用是嵌套的(flow 调 model,model 触发 tool……),出了问题要能顺着 trace 一层层看。Action 的做法是:每一次调用都自动开一个 span,天然形成调用树。

思路。 用 runInNewSpan 把真正的执行包起来;在 span 内把 input、output、subtype(即 actionType)、以及注册键写进 span 的 metadata / labels,并把 traceId/spanId 抓出来随结果返回。

真实实现(节选骨架):

// 真实源码 js/core/src/action.ts:528-561(节选)
let output = await runInNewSpan(
  {
    metadata: { name: actionName },
    labels: {
      [SPAN_TYPE_ATTR]: 'action',
      'genkit:metadata:subtype': config.actionType,
      ...(genkitKey ? { 'genkit:key': genkitKey } : {}),
    },
  },
  async (metadata, span) => {
    traceId = span.spanContext().traceId;
    spanId = span.spanContext().spanId;
    if (options?.onTraceStart) options.onTraceStart({ traceId, spanId });
    metadata.input = input;
    // ... 执行 fn,拿到 output 后:
    metadata.output = JSON.stringify(output);
    return output;
  }
);

关键细节:

• onTraceStart(ActionRunOptions,action.ts:137)让调用方在执行开始的一瞬间就拿到 trace/span id,不必等 Action 跑完——对"边跑边把 trace 链接展示给用户"很有用。
• 出错时也不放过:catch 里把 traceId 挂到 error 上再抛(action.ts:592-596),保证异常也能被追溯到具体 trace。

3.4 context 注入:auth 等侧信道数据怎么进函数体

它要解决的小问题。 一个 tool 想知道"当前是谁在调我"(auth),但 auth 不该混进业务输入 input 里。Genkit 用一条侧信道 context:通过 ActionRunOptions.context 传入,在函数体内通过 ActionFnArg.context 读到。

它怎么流动。 底层用 AsyncLocalStorage(runWithContext,context.ts:37)把 context 塞进当前异步调用栈——这样嵌套调用的子 Action 能自动继承父级 context,不用手动层层透传。

优先级规则(读代码才看得出的精华): 组装传给 fn 的 context 时,合并顺序是"注册表默认 context → 显式传入的 context 或从上游继承的 context",后者覆盖前者:

// 真实源码 js/core/src/action.ts:568-588(节选)
const actFn = () =>
  fn(input, {
    ...options,
    context: {
      ...actionFn.__registry?.context,          // 注册表级默认
      ...(options?.context ?? getContext()),     // 显式传入 > 上游继承
    },
    streamingRequested: !!options?.onChunk && options.onChunk !== sentinelNoopStreamingCallback,
    sendChunk: options?.onChunk ?? sentinelNoopStreamingCallback,
    trace: { traceId, spanId },
    registry: actionFn.__registry,
    abortSignal: options?.abortSignal ?? makeNoopAbortSignal(),
    ...
  });
// 显式传了 context 就用它跑,否则让上游 context 继续贯穿
const output = await runWithContext(options?.context, actFn);

ActionFnArg(action.ts:159)就是函数体收到的第二个参数,把上面这些能力打包成一个对象:streamingRequested、sendChunk、context、trace、abortSignal、registry、inputStream、init。

3.5 流式:sendChunk 与 streamingRequested

它要解决的小问题。 大模型是边生成边出字的。Action 要让函数体能"边算边吐",同时又能被"只想要最终结果"的调用方无缝调用。

两个配套开关:

• streamingRequested——布尔量,告诉函数体"调用方到底想不想要流"。它的判断是"有没有传真正的 onChunk"(排除掉那个哨兵空回调 sentinelNoopStreamingCallback)。
• sendChunk——函数体调它往外吐一个 chunk;没人要流时它就是个空操作哨兵,函数体照调不误、零成本。

同一个 Action,两种调用姿势:

// 示意,非源码
await myAction(input);                       // 只要最终结果,streamingRequested=false
const { stream, output } = myAction.stream(input); // 要流:遍历 stream,再 await output

.stream() 的实现精华(action.ts:613-677): 它内部造一个 ReadableStream,把 onChunk 接到这个流的 controller 上,然后调 run;函数体每次 sendChunk 就 enqueue 一个 chunk。返回一个 { stream, output }:stream 是异步生成器(可 for await),output 是最终结果的 Promise。跑完/出错时自动 close 掉流。

关于底层 StreamManager。 js/core/src/streaming.ts 还提供了一套"存储/订阅"抽象(ActionStreamInput、StreamManager、内存实现 InMemoryStreamManager),用于跨进程/断线重连场景:一端 open(streamId).write(chunk),另一端 subscribe(streamId, {...}) 收 chunk——迟到的订阅者还能补看已缓存的历史 chunk。它服务的是远程/持久化流式,和函数体内的 sendChunk 是不同层次,这里不展开。

---

4. 深入实现:注册表(Registry)

Action 造出来后要能"被找到"。这就是 Registry(js/core/src/registry.ts:152)——一张按 /type/name 寻址的中央表。

4.1 键的约定与解析

注册键格式统一是 /<actionType>/<name>,而 name 里可以再嵌插件名和路径,于是出现这几种真实形态:

键样例	含义
/util/generate	内建工具 generate
/model/googleai/gemini-2.5-flash	googleai 插件下的 model
/prompt/my-plugin/folder/my-prompt	插件下带子路径的 prompt
mcp-host:tool/my-tool	动态 Action Provider(如 MCP host)下的工具

parseRegistryKey(registry.ts:99)负责把这些字符串拆回结构化的 { actionType, pluginName?, actionName, dynamicActionHost? }。它先特判 /dynamic-action-provider 前缀(用 : 分 host),否则按 / 分词:tokens.length >= 4 认为"第 3 段是插件名、其余是动作名",否则是 /type/name 两段式。

actionType 是一个封闭枚举(ACTION_TYPES,registry.ts:40),这也直接回答了"Action 到底能是哪几种":

custom, dynamic-action-provider, embedder, evaluator, executable-prompt,
flow, indexer, model, background-model, check-operation, cancel-operation,
prompt, reranker, retriever, tool, tool.v2, util, resource,
agent, agent-snapshot, agent-abort

看到这个列表就明白了本章标题的分量:model / tool / flow / prompt / retriever / embedder…… 全是 actionType 的取值,而它们的载体是同一个 Action。

4.2 注册与查找

注册(registerAction,registry.ts:278)做三件事:校验传入的 type 与 Action 自带的 actionType 一致、按 /${type}/${name} 拼出键、存进 actionsById,并把 action.__registry 反向指回自己(context 继承就靠这个反向指针)。键冲突会打错误日志并覆盖:

// 真实源码 js/core/src/registry.ts:301-312(节选)
const key = `/${type}/${action.__action.name}`;
if (this.actionsById.hasOwnProperty(key)) {
  logger.error(`ERROR: ${key} already has an entry in the registry. Overwriting.`);
}
this.actionsById[key] = action;
action.__registry = this;

查找(lookupAction,registry.ts:222)不是简单查表——它带惰性解析:如果键里带插件名且该插件还没初始化,先 initializePlugin;如果表里仍没有,再让插件的 resolver 动态注册一个;还处理动态 Action Provider(如 MCP)的即时取用;最后回退到 parent 注册表。

父子注册表。 Registry 可以 withParent 叠加(registry.ts:195)。查找、listActions、lookupSchema 等都会在本层查不到时回退父层——这让"全局注册表 + 每次请求的临时覆盖层"成为可能。

4.3 defineAction vs defineActionAsync:同步登记 vs 延迟登记

这是本章最后一块拼图:两种把 Action 放进注册表的方式。

defineAction(action.ts:692)= action() + 立即 registerAction。 它多做两件保护/接线:

1. 禁止在运行时定义 Action——如果当前正处在某个 Action 的执行栈里(isInRuntimeContext()),直接抛错。这防止"一边跑一边偷偷改注册表"这类难以追踪的副作用。
2. 包裹的 fn 里会先 registry.initializeAllPlugins(),再在 runInActionRuntimeContext 里跑——于是执行期间 isInRuntimeContext() 为真,上面的保护得以生效。

// 真实源码 js/core/src/action.ts:705-717(节选)
if (isInRuntimeContext()) {
  throw new Error('Cannot define new actions at runtime.\n' + ...);
}
const act = action(config, async (i, options) => {
  await registry.initializeAllPlugins();
  return await runInActionRuntimeContext(() => fn(i, options));
});
act.__action.actionType = config.actionType;
registry.registerAction(config.actionType, act);
return act;

defineActionAsync(action.ts:835)= 注册一个"将来才 resolve"的 Action。 它接收的是一个 config 的 PromiseLike,用 lazy(...)(js/core/src/async.ts)包成"首次被 await 时才真正构造 Action"的惰性 promise,然后通过 registerActionAsync(registry.ts:318)把这个 promise 存进 actionsById。

这解决的是"定义 Action 需要异步准备"的场景:比如插件要先去远端拉一份模型清单,才能知道该注册哪些 model。查找时 lookupAction 里的 await this.actionsById[key] 会自然地把 promise 兑现成真正的 Action。

AsyncProvider<T>(registry.ts:35,即 () => Promise<T>)就是这类"异步产出物"的通用签名。

两者对比:

维度	defineAction	defineActionAsync
构造时机	立即	惰性,首次查找时
存进注册表的东西	Action 本身	Action 的 PromiseLike
适用场景	config 当场就齐全	需要异步准备(如拉远端清单)
底层注册方法	registerAction	registerActionAsync

---

5. 巧妙之处(可借鉴的技术)

• 哨兵空回调让"流式"零分支成本。 sentinelNoopStreamingCallback(action.ts:878)既当"没人订阅时的 sendChunk 默认值",又当"判断 streamingRequested 的标记":onChunk !== 哨兵 才算真要流。函数体永远可以无脑 sendChunk,不用先判断有没有人听。

• context 走 AsyncLocalStorage 自动贯穿,而非层层传参。 runWithContext(context.ts:37)把 auth 等塞进异步上下文,子 Action 用 getContext() 就能继承——嵌套调用不必手动透传。且 context 为 undefined 时是无副作用的直通(context.ts:41-43),不为它开销。

• "运行时禁止 defineAction"用同一套 ALS 开关实现。 runInActionRuntimeContext / isInRuntimeContext(action.ts:913-922)靠一个 AsyncLocalStorage 标志位,既标记"正在执行 Action",又被 defineAction 用来拒绝运行期定义——一套机制两用。

• 校验器按 schema 做 WeakMap 缓存(schema.ts:37-38),同一个 Zod schema 只编译一次 Ajv/cfworker validator,重复调用不重复编译。

• 注册表可叠父层(withParent),把"全局稳定注册表"和"每请求临时覆盖"优雅分层。

---

6. 边界与局限

• 无 schema = 不校验。 只有声明了 inputSchema/outputSchema 才卡关(action.ts:495、:600);不给就完全放行。图省事不写 schema 就享受不到边界保护。

• 运行期不能 defineAction。 处于 Action 执行栈内调用 defineAction 会直接抛错(action.ts:705)。要在运行中动态产出 Action,得走插件 resolver / 动态 Action Provider 这条路,而不是 defineAction。

• 注册键冲突是"覆盖 + 打日志",不是硬报错。 同键重复注册只会 logger.error 然后覆盖(registry.ts:303-306),不会阻止——重名的两个 Action 谁后注册谁生效,容易埋坑。

• schema 注解在递归 schema(z.lazy)上可能失效。 applyAnnotations 明确不解析 JSON schema 的 $ref(schema.ts:107-112 的注释),递归定义上的注解可能落不到引用的定义里。

---

7. 横向对比

同属 ai-agent-reference 货架的多数框架把 model、tool、workflow 做成各自独立的类 + 各自的执行器。Genkit 的取舍是极致归一:一个 Action 原语 + 一个 actionType 枚举 + 一张注册表,承载全部能力种类。好处是可观测性、校验、流式、context 这些横切能力只实现一次、人人复用;代价是所有东西都被 Zod schema 和 /type/name 键约束住,灵活性让位于一致性。

本章只讲原语本身。它如何被真正用起来,见同组后续章节:

• Plugin 系统与 genkit() 门面 —— 插件如何把一批 Action 装进注册表。
• generate:模型调用与自动工具循环 —— generate 这个 Action 怎么驱动 agentic 循环。
• 工具执行、中断与 human-in-the-loop —— tool 类 Action 的执行与中断。
• Dotprompt 与结构化输出 —— prompt 类 Action 与 schema 校验的联动。
• 会话状态与可观测性 —— span 如何汇成 Dev UI 里的调用树。

---

8. 代码地图(导航索引)

主题	文件路径	符号名
Action 工厂(调用管线核心)	js/core/src/action.ts	action、actionFn.run、actionFn.stream
Action 类型定义	js/core/src/action.ts	Action、BidiAction
元数据结构	js/core/src/action.ts	ActionMetadata、ActionMetadataSchema
函数体收到的参数	js/core/src/action.ts	ActionFnArg(sendChunk/streamingRequested/context/trace)
调用方选项(侧信道)	js/core/src/action.ts	ActionRunOptions(onChunk/context/onTraceStart)
同步定义并注册	js/core/src/action.ts	defineAction
延迟定义并注册	js/core/src/action.ts	defineActionAsync
运行时上下文开关	js/core/src/action.ts	isInRuntimeContext、runInActionRuntimeContext
哨兵空回调	js/core/src/action.ts	sentinelNoopStreamingCallback
注册表	js/core/src/registry.ts	Registry、registerAction、lookupAction、registerActionAsync
actionType 枚举	js/core/src/registry.ts	ACTION_TYPES、ActionType
键解析	js/core/src/registry.ts	parseRegistryKey
异步产出物签名	js/core/src/registry.ts	AsyncProvider
schema 校验	js/core/src/schema.ts	parseSchema、validateSchema、toJsonSchema
流式存储/订阅抽象	js/core/src/streaming.ts	StreamManager、ActionStreamInput、InMemoryStreamManager
context 注入	js/core/src/context.ts	runWithContext、getContext、ActionContext
开 span	js/core/src/tracing/instrumentation.ts	runInNewSpan、SPAN_TYPE_ATTR、setCustomMetadataAttributes