跳转到主要内容
文档构建工作流变量与表达式

变量与表达式

workflow 级参数、上游步骤输出、classifier 路由、凭据引用 —— 把数据在 step 之间流动起来。

Braidrun 里所有动态值都走"变量引用"。语法只有一种形式：双花括号。

yaml
{{var:name}}                     # workflow 级变量
{{steps.fetch.data.user.name}}   # 上游步骤的 JSON 输出
{{classifier.category}}          # 分类器路由结果
{{credentials.openai_api_key}}   # 凭据库里的密钥

四种变量源

1. var: — workflow 参数

工作流顶部的 variables / params 段，或者调度 / Webhook 预设的输入。

yaml
variables:
  target_date:
    type: string
    default: "yesterday"
  top_n:
    type: number
    default: 10

steps:
  - id: fetch
    type: code
    code: |
      return { date: "{{var:target_date}}", top: {{var:top_n}} };

2. steps.id.* — 上游步骤的输出

每个 step 运行完都有一份结构化输出。最常用的几个子字段：

  • steps.id.output — 纯文本或原始返回值。
  • steps.id.data — extract 之后的结构化 JSON。
  • steps.id.artifacts[0].url — 产物文件链接。
  • steps.id.tokens / cost / duration_ms — 执行统计。
yaml
steps:
  - id: fetch
    type: code
    extract:
      articles: $.items
      total: $.meta.total

  - id: summarize
    type: single
    agent: writer
    task: |
      我有 {{steps.fetch.data.total}} 条新闻，这是第一条：
      {{steps.fetch.data.articles[0].title}}

3. classifier.* — classifier 的路由结果

上一个 classifier step 的结果作为一级引用，用于 condition / branching。

yaml
steps:
  - id: triage
    type: classifier
    agent: planner
    task: "{{var:user_message}}"
    categories: [billing, technical, other]

  - id: respond
    type: single
    condition: "{{classifier.category}} == 'billing'"
    agent: writer
    task: "客户关于账单的问题是：{{var:user_message}}"

等价于 steps.<classifier_step_id>.category —— 只是更短、更语义。

4. credentials.* — 凭据引用

解析链路：用户命名空间 → 团队命名空间 → 系统命名空间；详见 凭据管理。

JSONPath 表达式

大括号里可以写简单的路径语法 —— 基本兼容 JSONPath 的子集：

  • {{steps.fetch.data.articles[0].title}} — 数组索引
  • {{steps.fetch.data.articles[*].title}} — 数组所有元素（返回数组）
  • {{steps.fetch.data.articles[?(@.score>5)]}} — 条件过滤
  • {{steps.fetch.data | length}} — filter：length / upper / lower / default / trim

extract：把 step 输出结构化

LLM 或 code 返回的数据往往需要"挑出几个字段"给下游用。用 extract：

yaml
- id: fetch_user
  type: code
  code: |
    const res = await fetch('https://api.example.com/me');
    return await res.json();
  extract:
    name: $.profile.fullName
    company: $.profile.org.name
    is_admin: $.permissions[?(@=='admin')] | length > 0

之后 steps.fetch_user.data.name / .company 就直接可用。

condition：条件执行

任何 step 都可以带 condition —— 表达式为 false 时这一步被跳过（execution 状态标为 SKIPPED）。

yaml
- id: summarize
  type: single
  agent: writer
  condition: "{{classifier.category}} == 'ai' && length({{steps.fetch.data.articles}}) > 0"
  task: "..."

支持的运算符：

  • 比较：== != < > <= >=
  • 逻辑：&& || !
  • 字符串：contains(x, y) startsWith(x, y) endsWith(x, y) matches(x, /re/)
  • 数组：length(x) any(list, cond) all(list, cond)
  • 判空：isEmpty(x) isNotEmpty(x)
condition 被跳过 vs 失败

SKIPPED 是"正常没跑"，下游可以用 {{steps.x.status}} == 'SKIPPED' 检测。FAILED 则是运行时出错，会传导到 execution.status。两者不要混用。

retry：重试策略

yaml
- id: fetch
  type: code
  code: "..."
  retry:
    max_attempts: 3
    backoff: exponential     # fixed | linear | exponential
    initial_delay_ms: 1000
    max_delay_ms: 30000
    retry_on: [network, rate_limit, http_5xx]

仅在 transient（网络 / 限速 / 5xx）错误上重试，业务异常（condition 失败、LLM 拒绝）不会被重试。

aggregate：多输入合并

当一个 step 从多个上游收数据时用 aggregate。比如 group_chat 后汇总发言：

yaml
- id: combine
  type: code
  aggregate:
    comments: concat_lines({{steps.group_chat_1.messages}})
    total_tokens: sum({{steps.*.tokens}})
  code: "return { comments, total_tokens };"

可用聚合函数：concat / concat_lines / json_array / join(sep) / first / last / max / min / avg。

DTO 命名映射

YAML 用 snake_case（group_chat、agent_based），JSON API / DTO 用 camelCase（groupChat、agentBased）。平台自动双向转换，你不需要关心。

常见坑

  • 引用未运行的 step — 如果 condition 跳过了上游 step，引用它的下游会拿到 null/undefined。用 | default(...) 兜底。
  • 大字段直接塞 task — task: "{{steps.x.output}}" 如果上游输出 100KB，LLM 会爆 token。先 extract 压缩再传。
  • YAML 里的引号 — 包含冒号或特殊字符的表达式要用双引号包起来： condition: "{{x}} == 1"