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

1.20.0 起工作流 SDK 并进 Semantic，叫 Automate（照 Palantir）：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";

三个概念：

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", "usage_kg", "scrap_rate", "process"]
  },
  {
    "name": "logistics_rate",
    "kind": "get",
    "title": "物流包装费率",
    "input": { "destination": { "type": "text", "required": true } },
    "type": "demo.logistics_rate",
    "pk": "=input.destination"
  }
]

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

自动提取

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

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.rfq_no" } }
    },
    "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(usage_kg * (1 + scrap_rate) * 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": { "rfq_no": "=input.rfq_no", "product": "=steps.rfq.row.product", "currency": "=steps.rfq.row.currency",
                  "unit_price": "=steps.material.summary.total", "ai_summary": "=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": "报价单号" }]
  }
}

步骤

每一步都可以写 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。
• 安全：没有赋值、循环、方法调用和自定义函数；属性只读自有的（碰不到原型链）；计算量有上限。

逐行计算里，行的属性直接用名字（usage_kg），关联上的行用 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 / 智能体）

谁能跑：

• 模型按工作流应用计费（清单 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.mtd_usd > object.budget_usd * 0.8", "cooldown": "12h",
              "input": { "account_key": "=object.account_key" } },
  "schedule": { "every": "1d", "at": "01:30" }
}

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

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

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

为什么触发条件放在自动化里（照 Palantir 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 自己的应用还包括平台员工），邮箱已验证。别的地址不发，结果里写明原因——平台不做垃圾邮件中继。
• 去重与冷却：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

上限

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

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