跳轉到主要內容
文件建構工作流程變數與表達式

變數與表達式

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"