RAILWISE-CLI 工作流编排指南

RAILWISE-CLI 工作流编排指南

目标读者：需要构建复杂自动化流程的技术负责人、测绘项目工程师
预计阅读时间：25 分钟
前置要求：熟悉 cli-quickstart.md 基础用法与 cli-agents-configuration.md 智能体配置

---

1. 工作流核心概念

工作流（Workflow） 是 RAILWISE-CLI 的核心编排单元，它将多个智能体的操作串联为可重复执行的自动化管道。一个工作流由以下要素组成：

要素	说明	必需
name	工作流名称	是
description	工作流描述	否
variables	全局变量定义	否
steps	执行步骤序列	是
on_error	错误处理策略	否
output	全局输出配置	否

---

2. 基础语法

2.1 最小可用工作流

name: "数据导入与质检"
steps:
  - name: "导入"
    agent: "surveyor"
    action: "import"
    input: "data/raw.gsi"

  - name: "质检"
    agent: "inspector"
    action: "check"

2.2 完整结构示例

name: "轨道交通保护区自动化监测日报"
description: "每日自动执行数据导入、平差、监测分析、报告生成并邮件推送"

variables:
  project_code: "NB-M3-2025-001"
  monitoring_date: "{{today}}"         # 内置变量：今日日期
  data_dir: "data/{{monitoring_date}}"
  output_dir: "output/{{monitoring_date}}"

steps:
  # ... 步骤定义

on_error:
  strategy: "notify_and_continue"      # 错误处理策略
  notify:
    channel: "email"
    recipients: ["tech@railwise.cn"]

output:
  consolidate: true                    # 合并所有步骤输出
  destination: "{{output_dir}}/daily_report.md"

---

3. 步骤（Step）详解

3.1 步骤基本结构

steps:
  - name: "步骤名称"                   # 显示名称，用于日志和报告
    id: "step_01"                      # 唯一标识，用于引用和条件判断
    agent: "surveyor"                  # 执行智能体
    action: "import"                   # 智能体动作
    input: "data/file.gsi"             # 输入文件/目录（支持通配符）
    input_var: "previous_output"       # 或引用变量作为输入
    output: "output/step01.json"       # 输出文件路径
    output_var: "step01_result"        # 或存入变量供后续使用
    params:                            # 动作参数
      format: "gsi"
      encoding: "utf-8"
    condition: "{{variables.force_run}} == true"  # 执行条件
    timeout: 300                       # 超时时间（秒）
    retries: 2                         # 失败重试次数

3.2 输入输出模式

模式一：文件路径传递

- name: "导入"
  agent: "surveyor"
  action: "import"
  input: "data/*.gsi"                  # 支持 glob 通配符
  output: "output/raw_data.json"

- name: "平差"
  agent: "adjuster"
  action: "adjust"
  input: "output/raw_data.json"        # 引用上一步输出文件

模式二：变量传递（推荐）

- name: "导入"
  agent: "surveyor"
  action: "import"
  input: "data/*.gsi"
  output_var: "raw_data"

- name: "平差"
  agent: "adjuster"
  action: "adjust"
  input_var: "raw_data"                # 直接引用变量，无需文件IO
  output_var: "adjusted_data"

提示

变量模式在内存中传递数据，避免频繁的磁盘读写，执行效率提升 30%~50%，尤其适用于包含大量中间步骤的复杂工作流。

---

4. 控制结构

4.1 条件执行（Condition）

支持基于变量、步骤状态、文件存在性等条件控制步骤执行：

steps:
  - name: "检查数据文件"
    agent: "surveyor"
    action: "check_exists"
    input: "data/today"
    output_var: "data_exists"

  - name: "执行导入（仅当数据存在）"
    agent: "surveyor"
    action: "import"
    input: "data/today"
    condition: "{{data_exists}} == true"

  - name: "使用备用数据（当主数据不存在）"
    agent: "surveyor"
    action: "import"
    input: "data/backup"
    condition: "{{data_exists}} == false"

支持的运算符：

运算符	说明	示例
==	等于	{{status}} == "success"
!=	不等于	{{count}} != 0
> < >= <=	数值比较	{{error_count}} > 5
in	包含于	{{type}} in ["settlement", "horizontal"]
exists	文件存在	exists("data/file.gsi")
and or not	逻辑组合	{{a}} == true and {{b}} > 3

4.2 循环执行（Loop）

对列表或文件集合进行批量处理：

steps:
  - name: "获取监测点列表"
    agent: "surveyor"
    action: "list_points"
    output_var: "point_list"           # 假设返回 ["P01", "P02", "P03", ...]

  - name: "逐点分析"
    agent: "monitor"
    action: "analyze_point"
    loop:
      over: "{{point_list}}"           # 迭代变量
      as: "point_id"
    params:
      point_id: "{{point_id}}"
      days: 30
    output_var: "analysis_results"
    # 循环结束后，analysis_results 为结果数组

文件批量处理：

steps:
  - name: "批量导入水准数据"
    agent: "surveyor"
    action: "import"
    loop:
      over: "glob:data/leveling/*.gsi"
      as: "file_path"
    input: "{{file_path}}"
    output_var: "imported_batches"

4.3 并行执行（Parallel）

多个独立步骤同时执行，缩短总耗时：

steps:
  - name: "数据准备"
    agent: "surveyor"
    action: "prepare"
    output_var: "prepared_data"

  - name: "并行分析"
    parallel:
      - name: "沉降分析"
        agent: "monitor"
        action: "analyze"
        input_var: "prepared_data"
        params:
          type: "settlement"
        output_var: "settlement_result"

      - name: "水平位移分析"
        agent: "monitor"
        action: "analyze"
        input_var: "prepared_data"
        params:
          type: "horizontal"
        output_var: "horizontal_result"

      - name: "收敛分析"
        agent: "monitor"
        action: "analyze"
        input_var: "prepared_data"
        params:
          type: "convergence"
        output_var: "convergence_result"

  - name: "汇总报告"
    agent: "surveyor"
    action: "merge_report"
    params:
      sections:
        - "{{settlement_result}}"
        - "{{horizontal_result}}"
        - "{{convergence_result}}"
    output: "output/comprehensive_report.md"

1. 并行步骤之间不能有数据依赖（不能引用同一步骤的输出变量）
2. 并行步骤的 output_var 名称必须唯一
3. 系统并发上限受 max_concurrent_agents 配置约束
4. 并行步骤中任一失败，默认触发 on_error 策略

---

5. 错误处理

5.1 错误处理策略

on_error:
  strategy: "notify_and_stop"        # 可选值见下表
  notify:
    channel: "email"                  # 或 webhook / slack / 企业微信
    recipients: ["alert@railwise.cn"]
  retry:
    max_attempts: 3
    backoff: "exponential"             # 固定间隔 / 线性增长 / 指数退避

策略	行为
stop	立即停止工作流，返回错误状态
continue	记录错误，继续执行后续步骤
skip_step	跳过当前步骤，继续执行后续步骤
notify_and_stop	发送通知后停止
notify_and_continue	发送通知后继续执行
fallback	执行备用步骤（需定义 fallback 块）

5.2 步骤级错误处理

steps:
  - name: "数据导入"
    agent: "surveyor"
    action: "import"
    input: "data/main.gsi"
    retries: 2
    timeout: 120
    on_error:
      strategy: "fallback"
      fallback:
        - name: "使用备用数据"
          agent: "surveyor"
          action: "import"
          input: "data/backup.gsi"

---

6. 高级特性

6.1 模板与变量

内置变量：

变量	说明	示例
{{today}}	当前日期（YYYY-MM-DD）	2025-01-15
{{now}}	当前时间（ISO 8601）	2025-01-15T09:30:00+08:00
{{workflow_id}}	工作流实例ID	wf-abc123
{{step_count}}	总步骤数	5
{{project_dir}}	项目根目录	/path/to/project

自定义变量运算：

variables:
  base_dir: "data"
  date_str: "{{today}}"
  full_path: "{{base_dir}}/{{date_str}}"     # 字符串拼接
  yesterday: "{{today | offset_days(-1)}}"   # 日期运算

6.2 子工作流调用

复用已定义的工作流：

steps:
  - name: "执行标准质检流程"
    workflow: "workflows/standard-inspection.yaml"  # 子工作流路径
    params:
      input_data: "{{raw_data}}"
      standard: "GB 50497"
    output_var: "inspection_result"

6.3 钩子（Hooks）

在关键节点插入自定义操作：

hooks:
  pre_workflow:
    - name: "检查磁盘空间"
      command: "df -h"

  post_step:
    - name: "记录步骤耗时"
      script: |
        echo "{{step.name}} 耗时 {{step.duration}}ms" >> timing.log

  post_workflow:
    - name: "清理临时文件"
      command: "rm -rf .railwise/tmp/*"

---

7. 完整示例：盾构隧道监测日报工作流

name: "盾构隧道监测日报工作流"
description: "每日自动处理全站仪监测数据，生成日报并推送预警"

variables:
  project: "绕城高速管廊1标"
  line: "轨道交通3号线"
  data_dir: "data/{{today}}"
  output_dir: "output/daily/{{today}}"

steps:
  - name: "1. 数据导入"
    id: "import"
    agent: "surveyor"
    action: "import"
    input: "{{data_dir}}/*.gsi"
    params:
      format: "gsi"
      merge_sessions: true
    output_var: "raw_observations"

  - name: "2. 数据质检"
    id: "inspect"
    agent: "inspector"
    action: "check"
    input_var: "raw_observations"
    params:
      checks: ["限差", "粗差", "闭合差"]
      standard: "GB 50911"
    output_var: "quality_report"

  - name: "3. 平差计算"
    id: "adjust"
    agent: "adjuster"
    action: "adjust"
    input_var: "raw_observations"
    condition: "{{quality_report.passed}} == true"
    params:
      method: "间接平差"
      weight_scheme: "按距离定权"
    output_var: "adjusted_coordinates"

  - name: "4. 变形分析（并行）"
    id: "analysis"
    parallel:
      - name: "桥墩沉降"
        agent: "monitor"
        action: "analyze"
        input_var: "adjusted_coordinates"
        params:
          type: "settlement"
          points: ["P01", "P02", "P03"]
          reference_epoch: "2025-01-01"
        output_var: "settlement_analysis"

      - name: "桥墩水平位移"
        agent: "monitor"
        action: "analyze"
        input_var: "adjusted_coordinates"
        params:
          type: "horizontal"
          points: ["P01", "P02", "P03"]
        output_var: "horizontal_analysis"

  - name: "5. 预警判断"
    id: "alert"
    agent: "monitor"
    action: "evaluate_alerts"
    params:
      analyses:
        - "{{settlement_analysis}}"
        - "{{horizontal_analysis}}"
      method: "综合法"
    output_var: "alert_summary"

  - name: "6. 生成日报"
    id: "report"
    agent: "surveyor"
    action: "generate_report"
    params:
      template: "templates/daily_report.md"
      sections:
        project_info: "{{project}}"
        quality: "{{quality_report}}"
        settlement: "{{settlement_analysis}}"
        horizontal: "{{horizontal_analysis}}"
        alerts: "{{alert_summary}}"
    output: "{{output_dir}}/daily_report_{{today}}.md"

  - name: "7. 推送预警（如有）"
    id: "notify"
    agent: "surveyor"
    action: "send_notification"
    condition: "{{alert_summary.has_alert}} == true"
    params:
      channel: "enterprise-wechat"
      recipients: ["项目群"]
      message: "{{alert_summary.summary_text}}"

on_error:
  strategy: "notify_and_stop"
  notify:
    channel: "email"
    recipients: ["tech@railwise.cn"]
    subject: "[{{project}}] 工作流执行失败 — {{workflow_id}}"

---

8. 调试与优化

8.1 调试模式

# 启用详细日志
railwise run workflow.yaml --verbose

# 单步执行（进入交互式调试）
railwise run workflow.yaml --step-by-step

# 仅验证工作流语法，不执行
railwise validate workflow.yaml

# 生成执行计划图（Mermaid 格式）
railwise run workflow.yaml --dry-run --graph

8.2 性能分析

# 生成性能报告
railwise run workflow.yaml --profile

# 输出示例：
# ┌─────────────┬──────────┬──────────┬──────────┐
# │ 步骤        │ 耗时     │ 内存峰值 │ 状态     │
# ├─────────────┼──────────┼──────────┼──────────┤
# │ 数据导入    │ 2.3s     │ 156MB    │ ✓ 成功  │
# │ 数据质检    │ 4.1s     │ 89MB     │ ✓ 成功  │
# │ 平差计算    │ 12.5s    │ 423MB    │ ✓ 成功  │
# │ 变形分析    │ 8.7s     │ 312MB    │ ✓ 成功  │
# │ 生成报告    │ 1.2s     │ 67MB     │ ✓ 成功  │
# └─────────────┴──────────┴──────────┴──────────┘
# 总计：28.8s

---

9. 相关文档

• cli-quickstart.md — 快速入门指南
• cli-agents-configuration.md — 智能体配置详解
• cli-data-formats.md — 数据格式与导入
• cli-troubleshooting.md — 常见问题排查

---

元数据标签