查看 Markdown

Semantic · 自动化（原工作流 SDK）

1.20.0 起工作流 SDK 并进 Semantic，叫 Automate：trigger.change = 对象新建 / 修改的条件，trigger.schedule = 时间条件，use / model / notify 步骤 = Action、AIP Logic、通知效果。SDK 是 semantic.automate（aidc semantic automate …）；本页的 workflow.*、aidc workflow 与清单 workflow 照样能用（清单 sdk 登记 semantic）。

把已经发布到 Nexus 的应用的能力组合成一个完整的流程：营业部的询价、设计中心的材料清单、采购部的行情、制造部的产能、财务部的核价口径……各自是独立的应用，工作流把它们串起来，算、分析，最后写回结果。可以手动运行、预演，也可以在语义层数据一变时自动运行。每一步用了谁的能力、输入从哪来、产出了什么，画布上看得清清楚楚。

import { workflow } from "/developer/sdk/v1/aidc.js";

三个概念：

概念	是什么	写在哪
能力（capability）	一个已发布应用对外提供的一件事：带参数的语义层查询 / 单条读取 / 聚合 / Action	提供方应用的清单 `exports`（或自动提取）
工作流（workflow）	把能力、计算、AI 分析连成的有向无环图，外加触发方式与结果	工作流应用的清单 `workflow`
运行（run）	工作流被执行一次：每一步的状态、耗时、输出、token	平台记录，`workflow.runs()` / 画布

1. 能力：已发布应用对外提供什么

显式导出（推荐）

在提供方应用的 aidc.app.json 里声明 exports：名字、参数、它做什么。条件里的值可以是字面量，也可以是 = 开头的表达式（只能用 input）。

"exports": [
  {
    "name": "bom_of",
    "kind": "query",
    "title": "零件的初始材料清单",
    "input": { "product": { "type": "text", "title": "零件", "required": true } },
    "type": "demo.bom_line",
    "where": { "product": "=input.product" },
    "select": ["material", "usageKg", "scrapRate", "process"]
  },
  {
    "name": "logistics_rate",
    "kind": "get",
    "title": "物流包装费率",
    "input": { "destination": { "type": "text", "required": true } },
    "type": "demo.logistics_rate",
    "pk": "=input.destination"
  }
]

kind	做什么	关键字段	输出
`query`	查一组对象	`type` `where` `select` `sort` `limit`	`{ rows, count, total }`（每行带 `_pk`）
`get`	按主键取一个对象	`type` `pk`	`{ row }`（没有为 `null`）
`aggregate`	分组汇总	`type` `where` `groupBy` `metrics`	`{ rows, count }`
`action`	执行一个 Action（写语义层）	`action` `object` `params`	`{ object, rev, event }`；预演时 `{ preview: true, plan }`

导出的类型 / Action 必须登记在本应用的 semantic.types / semantic.actions 里——能力不能超出应用自己的权限。

自动提取

已经发布的应用不用改一行就有能力：它登记的每个语义类型自动成为查询能力 <slug>/<类型 apiName>（参数 where sort limit），每个 Action 自动成为 <slug>/<Action apiName>（参数 = Action 的参数 + object）。apiName 按标准写法原样引用：aws-auto-stop/ec2Instance（camelCase 的 Object Type）、aws-auto-stop/auto-stop-ec2-instance（kebab-case 的 Action Type），旧写法 production-live/production.order_line 照旧。

名字分两种：工作流自己起的名字（输入、计算字段、输出、步骤 id、显式导出的能力名与输入名）是小写 snake_case；引用语义层的东西用它的 API name——属性与 Action 参数是 camelCase（where、select、sort、行上的字段 usageKg、object.rfqNo），调自动提取的 Action 能力时 with 的键就是 Action 的参数名（rfqNo）。

const caps = await workflow.capabilities();          // 公司的能力目录
// [{ ref: "quote-design/bom_of", kind: "query", auto: false, input: {...}, app: { slug, title, version, department } }, …]

aidc workflow capabilities                 # ● 显式导出  ○ 自动提取
aidc workflow capabilities --app quote-sales

谁的身份在执行

能力以「提供方应用 × 成员角色」执行：只读得到提供方登记过、且对全公司可见的类型，Action 过它自己的角色规则。所以工作流拿不到提供方自己拿不到的东西，成员触发的工作流也不会变成开发者权限。只有发布到正式通道的版本的能力对外可见。

2. 工作流：在清单里写 `workflow`

一个工作流就是一个应用：清单 sdk 登记 workflow，workflow 里写步骤。下面是报价演示的缩写（完整版见样板应用）：

{
  "slug": "quote-workflow",
  "sdk": ["workflow", "data", "semantic", "auth", "log"],
  "models": ["deepseek-flash"],
  "owner": { "department": "营业部", "team": "营业智能体-报价" },
  "workflow": {
    "input": { "rfq_no": { "type": "text", "title": "询价单号", "required": true } },
    "trigger": {
      "manual": { "roles": ["developer", "member"], "preview": "public" },
      "change": { "type": "demo.rfq", "when": "=object.status == '待报价'", "input": { "rfq_no": "=object.rfqNo" } }
    },
    "lanes": ["营业部", "设计中心", "采购部", "制造部", "财务部"],
    "steps": [
      { "id": "rfq", "kind": "use", "title": "询价单", "use": "quote-sales/rfq", "with": { "rfq_no": "=input.rfq_no" } },
      { "id": "bom", "kind": "use", "title": "初始材料清单", "use": "quote-design/bom_of", "with": { "product": "=steps.rfq.row.product" } },
      { "id": "prices", "kind": "use", "title": "原材料行情", "use": "quote-procurement/commodity_prices", "with": { "materials": "=pluck(steps.bom.rows, 'material')" } },
      { "id": "material", "kind": "compute", "title": "材料成本", "lane": "采购部",
        "from": "=steps.bom.rows", "join": [{ "rows": "=steps.prices.rows", "on": "material", "as": "p" }],
        "fields": { "cost": "=round(usageKg * (1 + scrapRate) * p.price, 3)" },
        "summary": { "total": "=round(sum(cost), 2)" } },
      { "id": "review", "kind": "model", "title": "AI 报价分析", "model": "deepseek-flash",
        "prompt": "询价：{{ steps.rfq.row }}\n材料成本：{{ steps.material.rows }}", "output": { "summary": "text", "risks": "text[]" } },
      { "id": "draft", "kind": "use", "title": "报价草稿", "use": "quote-sales/demo.propose_quote",
        "with": { "rfqNo": "=input.rfq_no", "product": "=steps.rfq.row.product", "currency": "=steps.rfq.row.currency",
                  "unitPrice": "=steps.material.summary.total", "aiSummary": "=steps.review.summary" } }
    ],
    "output": { "quote_no": "=steps.draft.object._pk", "material_cost": "=steps.material.summary.total" },
    "outputLabels": [{ "key": "material_cost", "title": "材料成本" }, { "key": "quote_no", "title": "报价单号" }]
  }
}

步骤

kind	做什么	字段
`use`	调一个能力	`use: "<应用 slug>/<能力名>"`、`with`（参数）
`compute`	计算：不写 `from` 求一组值；写了 `from` 逐行求字段，可 `join` 别的步骤的行（左关联），`filter`、`sort`，最后 `summary` 汇总	见下
`model`	AI 分析：提示词里用 `{{ 表达式 }}` 插入前面的结果，按 `output` 声明的字段返回	`model`（要登记在清单 `models`）、`system`、`prompt`、`output`、`maxTokens`
`notify`	通知：发一封邮件（正式运行才发，预演只给出内容），见下文通知	`to`（邮箱，1–5 个，写死在清单里）、`subject` / `body`（可用 `{{ }}`）、`dedupe`、`cooldown`

每一步都可以写 lane（画布泳道，缺省 = 能力提供方登记的部门）、when（条件为假就跳过，下游读到 null）、after（额外的先后依赖）。

依赖是自动的：表达式里引用了 steps.<id> 就依赖它。平台按依赖分层，同一层并行执行；有环、引用不存在的步骤，部署时就拒收。

表达式

以 = 开头的字符串是表达式（像 Excel 公式）；其他是字面量。

取值：input.x、steps.bom.rows、steps.rfq.row.product、trigger.pk、run.id、rows[0]、row["name"]
运算：+ - * / %、== != < <= > >=、&&（and）、||（or）、!（not）、a ? b : c、a ?? b
函数：round floor ceil abs min max sum avg count len pluck unique first last where find lookup join concat upper lower trim text number contains startsWith if coalesce isNull fixed percent now today hoursSince（距今多少小时，看门狗判断数据多久没更新）
缺值参与算术得 null（不会悄悄当 0，要兜底写 x ?? 0）；除以 0 得 null。
安全：没有赋值、循环、方法调用和自定义函数；属性只读自有的（碰不到原型链）；计算量有上限。

逐行计算里，行的属性直接用名字（usageKg），关联上的行用 as 起的名字（p.price），整行叫 row；字段之间可以互相引用，求值顺序按引用关系排，与书写顺序无关。summary 里每个字段名代表整列（sum(cost)），rows 是全部行，count 是行数。

结果

output 把最后要看的值取出来（{ 名字: 表达式 }），outputLabels 给出展示顺序和显示名（数组）。画布与 aidc workflow run 都按它展示。

3. 运行

// 开跑：立即返回（执行在服务端继续），用 watch 看每一步
const { run } = await workflow.run({ rfq_no: "RFQ-2609-002" }, { mode: "preview" });
const stop = workflow.watch(run.id, {
  onRun: (r) => render(r.steps),          // 每有一步变化推一次整份运行
  onDone: (r) => console.log(r.output),   // succeeded / failed
});

await workflow.runs({ limit: 20 });        // 最近的运行（摘要）
await workflow.getRun(run.id);             // 每一步的输出
await workflow.wait(run.id);               // 等结束（CLI / 智能体）

方式	说明
正式运行 `mode: "run"`	Action 真的写进语义层（留痕、广播给所有订阅端）
预演 `mode: "preview"`	读、算、AI 分析都真的跑，Action 只返回将要写入的计划
数据变化自动运行 `trigger.change`	语义层里这个类型的对象新建 / 修改后（`when` 为真）自动跑一次正式运行。发布到正式通道起开始生效；每条变化只触发一次；工作流自己写出的变化不回头触发自己；工作流写的数据可以再触发别的工作流（最多 3 层）；`cooldown` 内同一个对象只跑一次
定时运行 `trigger.schedule`	只给「时间本身就是条件」的事（日报、月底对账、数据断供的看门狗）。见下文触发

谁能跑：

调用方	正式运行	预演
开发者 Key（CLI / 智能体）	✓（可 `channel: "test"` 跑测试版）	✓
工作流应用里的访客：`trigger.manual.roles` 里的角色	✓	✓
其他本公司成员	—	✓
公开分享链接的访客	—	`trigger.manual.preview = "public"` 时 ✓

模型按工作流应用计费（清单 limits.dailyTokens 封顶）；公开访客预演每 10 分钟 6 次。
公开分享一个工作流 = 访客能预演，并看到每一步的输出与历史运行——包括提供方应用读出的数据（能力以提供方 × 成员身份执行）。所以 preview: "public" 只给演示数据或本来就可以公开的数据用；访客看运行记录时，是谁跑的只显示「公司成员 / 访客 / 数据变化」，不显示账号。
Idempotency-Key 相同返回同一次运行；x-aidc-dry-run: true 只校验输入、给出执行计划。
运行记录保留 180 天；每一步的输出截断到 200 行 / 48 KB（完整结果只在运行时传给下游）。超过 5 分钟没有进展的运行（执行它的函数被回收）会被标为「中断」，可以重跑；排队没开始的由平台补跑。

触发：数据变化优先，定时兜底

「什么时候跑」写在清单的 trigger 里，三种可以并存：

"trigger": {
  "manual": { "roles": ["developer", "member"], "preview": "members" },
  "change": { "type": "cloud.account", "when": "=object.mtdUsd > object.budgetUsd * 0.8", "cooldown": "12h",
              "input": { "account_key": "=object.accountKey" } },
  "schedule": { "every": "1d", "at": "01:30" }
}

先问：是不是「数据一变才需要做」？ 是 → 用 change：数据从 ERP / 云账单 / 任何数据源经数据流进语义层，对象一变就触发，不轮询、不写 cron、没变化时零成本。定时只留给「时间本身就是条件」的事——每天的汇总、月底的检查、数据该来却没来的看门狗（数据不来就没有变化可触发）。

写法	什么时候跑	说明
`change.type` + `when`	这个类型的对象新建 / 修改，且条件为真	条件里 `object` = 变化后的对象，`change` = `{ op, origin, actor, seq }`；条件为假不开跑、不计分钟
`change.cooldown`	同一个对象（主键）两次运行至少隔这么久	`30m` / `12h` / `1d`；冷却期内的变化只推进游标——告警类工作流一定要写
`change.input` 为空	工作流自己去查全部数据（不针对某个对象）	同一批变化只开一次运行：一次快照改了 N 个对象，不跑 N 遍同样的检查
`schedule.every`	每隔一段时间	`5m` 到 `7d`（整数 + `m`/`h`/`d`）；按 2026-01-01 00:00 UTC 起算对齐（`1h` = 每个整点）
`schedule.at`	按天的间隔在几点跑（UTC `HH:MM`）	只配 `1d`、`7d`…；`1d` + `01:30` = 每天 01:30 UTC（北京时间 09:30）
`schedule.input`	定时运行时给工作流的输入	字面量

定时的频率分级（部署时检查，aidc app check / deploy 打印告警）：

间隔	等级	处理
< 5 分钟	拒收	部署失败：要实时就用 `change`
5 分钟 – 1 小时	高频	一定告警：每月次数（每 5 分钟 = 8,640 次）与预计占用的计算分钟；先想想能不能用 `change`，能不能放宽
1 小时 – 1 天	中频	提示每月次数
≥ 1 天	正常	——

定时由平台已有的调度心跳执行（不为每个工作流另加 cron）：发布到正式通道时登记下一个时刻，换成没有定时的版本就取消；同一个时刻只跑一次；平台停过也不补跑错过的时刻（直接跳到下一个）。每次运行计入应用的计算分钟；本月计算分钟用完后，数据变化与定时都不再开跑，下月恢复。

为什么触发条件放在自动化里（Automate：条件 + 效果）：数据从哪来是数据流与数据源的事，数据存成什么、怎么变是 Ontology 的事（变化账），「变成什么样时、做什么」是流程——条件用的是工作流的表达式，做的事是工作流的步骤（查询、计算、AI 分析、写回、通知），运行记录、去重、链式上限也都在这里。所以触发与条件只有一份，在工作流里。

通知

notify 步骤发邮件——告警、日报、审批提醒：

{ "id": "mail", "kind": "notify", "title": "超预算告警",
  "when": "=len(steps.check.rows) > 0",
  "to": ["ops@example.com"],
  "subject": "云费用告警：{{ join(pluck(steps.check.rows, 'title'), '、') }}",
  "body": "{{ join(pluck(steps.check.rows, 'line'), '\n') }}",
  "dedupe": "=today()", "cooldown": "20h" }

收件人写死在清单里（部署时看得见发给谁），而且必须是本公司的账号（持本公司 License、本公司第一方账号；AIDC 自己组织 cell-aidc 的工作流还包括平台员工和客户组织的成员，即持任何一家组织有效 License 的人），邮箱已验证。客户组织之间不互通。发给客户组织的成员时，正文只放可以给对方看的内容：平台不按收件人的数据权限过滤正文，内容由写清单的人负责。别的地址不发，结果里写明原因——平台不做垃圾邮件中继。
去重与冷却：dedupe（缺省 = 渲染后的标题）相同的通知在 cooldown（缺省 6h）内只发一次；每个应用每天最多 limits.notificationsPerDay（缺省 20）封。告警风暴不会变成邮件风暴。
如实记录：每一封都进外发账——发出（sent）、被冷却或日上限拦下（suppressed）、邮件通道没配置（unconfigured）、失败（failed）；步骤输出里有状态与原因。没发出去不算步骤失败，收件人全不合格才失败。预演只返回将要发的内容，不发。
验证通道：aidc notify test（给自己发一封）；应用的通知记录：aidc app usage <slug>。邮件通道由平台配置（服务商 Resend，环境变量 RESEND_API_KEY）。

4. 画布

官方样板 workflow-canvas 是通用界面：泳道 = 部门，按依赖分层（部门比层数少时竖排，从上往下流），连线上的字是传过去的参数；点任意一步看它用了谁的能力、输入是怎么来的、公式、输出与提供方的应用卡片；运行实时推送，结束后可以回放。任何带 workflow 的应用都可以直接用这份界面，aidc app init <slug> --template workflow 生成的是最小起步版。

5. 部署与校验

aidc app deploy <目录> --dry-run     # 先查：结构、表达式、环、能力在公司目录里、参数名与必填
aidc app deploy <目录> && aidc app publish <slug>
aidc workflow run quote-workflow --param rfq_no=RFQ-2609-002 --preview

部署时平台对照公司的能力目录检查：每个 use 的提供方已发布到正式通道、能力存在、with 里的参数都是能力声明过的、必填的都给了；trigger.change 的类型在语义层里。所以要先发布提供方应用，再部署工作流。

CLI

aidc workflow capabilities [--app slug]
aidc workflow list
aidc workflow get <slug> [--channel test]
aidc workflow run <slug> [--input '{…}'] [--param 名=值 …] [--preview] [--channel test] [--no-wait]
aidc workflow runs <slug> | status <slug> <运行 id> | watch <slug> <运行 id>

API

方法	路径	说明
GET	`/api/v1/developer/capabilities/{命名空间}`	能力目录
GET	`/api/v1/developer/workflows/{命名空间}`	工作流列表
GET	`/api/v1/developer/workflows/{命名空间}/{slug}`	详情（编排计划、步骤、泳道、提供方卡片）
GET / POST	`/api/v1/developer/workflows/{命名空间}/{slug}/runs`	运行列表 / 开跑
GET	`/api/v1/developer/workflows/{命名空间}/{slug}/runs/{id}`	一次运行
GET	`/api/v1/developer/workflows/{命名空间}/{slug}/runs/{id}/events`	实时进度（SSE）

上限

步骤 ≤ 40；能力参数 ≤ 24；逐行计算 ≤ 5000 行；表达式 ≤ 1000 字；一次运行在一次函数调用内跑完（≤ 5 分钟）；数据变化触发链最多 3 层；定时最密每 5 分钟（≤ 1 小时告警）；通知收件人 ≤ 5、每天 ≤ limits.notificationsPerDay；运行时长计入应用的计算分钟（limits.computeMinutesPerMonth，缺省 2000 / 月）。

本页由 developer/docs/workflow.md 生成 · Markdown 原文 · llms.txt