ワークフロー
データの取り込み・変換・出力をステップで組み合わせて自動化します。複雑な統合ツールではなく、シンプルな UI で AI とのデータ連携に特化しています。
仕組み
- ステップを上から順に実行します。データは**行の集まり（list[dict]）**として各ステップを流れます。
- ステップは大きく 取得(source)→ 加工(transform)→ 出力(load / action) に分かれます。
- 外部サービスの認証情報は 接続管理(ApiConnection) に保存し、ステップから参照します（接続は作成した本人のみが利用できます。管理者も他人の接続は使えません）。
- ステップの利用にロール制限はありません。シェルコマンド・rsync・Python 関数を含め、すべてのステップを全ユーザーが利用できます。
- 値の差し込みにはテンプレートが使えます:
{{ $json.列名 }}/{{ $today }}/{{ $workflow.name }}/{{ $env.変数名 }}/{{ $node["ステップ名"].json[0].列 }}。プレビュー実行後は、各入力欄の{{ }}ボタンから上流ステップの実列名を型付きで選んで挿入できます。 - テンプレートが式 1 個だけ（地の文なし）のとき、「値をセット」では list / dict が文字列化されず構造のまま列に入ります。文と混ぜると JSON 文字列になります。
- ステップ名を変更しても、他ステップの
$node["…"]参照は自動で追随します。
共通の設定（全ステップ）
各ステップには、失敗時の挙動を制御する共通オプションがあります。
| 項目 | 既定 | 説明 |
|---|---|---|
| リトライ | off | 失敗時に再試行する |
| 最大リトライ回数 | 3 | 再試行の回数 |
| リトライ間隔(ms) | 1000 | 再試行までの待ち時間 |
| エラー時の挙動 | stop | stop（中断）/ continue（無視して継続）/ continue_with_error_output（エラー行を出力して継続） |
オペレータ一覧
ファイル・API 連携
| 表示名 | キー | 役割 |
|---|---|---|
| ファイル取得 | source_file | CSV / Excel / JSON / JSONL / PDF / TSV / DOCX / PPTX などを読込（フォルダ・glob・接続 URI・仮想パス対応） |
| ファイル保存 | load_file | CSV / TSV / JSON / JSONL / MD / Excel / gzip で書き出し |
| ファイル差分取得 | source_file_landed | 追加・変更されたファイルだけを取得（差分トリガ）。ファイルごとのスナップショット比較で判定するため、時計ズレや mtime を保持したコピーでも取りこぼしません。「削除も検知」を有効にすると消えたファイルを削除イベントとして出力（「RAG に保存」の upsert が削除を反映）。マイファイル / 共有 / 接続フォルダ（SFTP / SMB / S3 等）に対応 |
| RSS 取得 | source_rss | RSS / Atom フィードを取得 |
| REST API 取得 | source_rest | REST をページング・認証つきで取得（保存済みの「REST API」接続を利用可） |
| REST API 送信 | action_http_request | 各行ごとに HTTP リクエストを送信し、応答を行に付与 |
| Webhook 受信 | source_webhook | Webhook 受信を有効化 |
| Webhook 送信 | action_webhook | Webhook URL へ署名・リトライつき送信 |
| rsync 取得 / 送信 | source_rsync / load_rsync | リモートと rsync で同期 |
| メール送信 | action_email | SMTP でメール送信 |
| チャット通知送信 | action_notify | Slack / Teams / Chatwork / LINE WORKS / LINE / Discord / Telegram へ送信（送信先はプロバイダ選択式） |
SQL・データ蓄積
| 表示名 | キー | 役割 |
|---|---|---|
| SQL 取得 (SELECT) | source_sql | SELECT / WITH のみ。クエリ直書き or テーブル＋条件。差分取得対応 |
| SQL 保存 (INSERT) | load_sql | append / replace / merge。テーブル自動作成・バッチ投入（PostgreSQL / MySQL / MariaDB 対応） |
| SQL 更新 (UPDATE) | update_sql | 主キーで行単位 UPDATE（PostgreSQL / MySQL / MariaDB 対応） |
| SQL 実行 (任意 SQL) | execute_sql | 任意の DDL / DML を実行 |
| BigQuery 取得 / 書込 | source_bigquery / load_bigquery | BigQuery をクエリ / ストリーム挿入（SA JSON 認証） |
| ClickHouse 取得 / 書込 | source_clickhouse / load_clickhouse | ClickHouse の読み書き |
| データセットから取得 | source_dataset | 内蔵のデータセットを読み込み（別ワークフローや AI 分析の保存結果を入力に） |
| データセットに保存 | load_postgres_dataset | 内蔵のデータセットに保存（ダッシュボード や BI キーで参照） |
| RAG に保存 | load_to_rag | 行をまとめて 1 文書としてナレッジに埋め込み |
| RAG から検索 | source_rag_retrieve | ナレッジから類似チャンクを取得 |
データ加工
| 表示名 | キー | 役割 |
|---|---|---|
| 行の絞り込み (WHERE) | transform_filter | 条件で行を残す / 除く |
| 列の編集 (SET) | transform_set | 列の設定 / 改名 / 削除 / 指定列だけ残す |
| 列名の変更 (AS) | transform_rename_keys | 全行のキーを一括改名 |
| 並べ替え (ORDER BY) | transform_sort | 昇順 / 降順 / ランダム（重複除去・件数制限も） |
| 件数制限 (LIMIT) | transform_limit | 先頭 / 末尾 N 件 |
| 重複削除 (DISTINCT) | transform_dedup | 指定列で重複除去 |
| 行に分割 (UNNEST) | transform_split | 1 行を複数行に展開 |
| 集計・グループ化 (GROUP BY) | transform_aggregate | グループ化 + 集約関数 |
| データ結合 (JOIN) | transform_merge | append / inner / left / outer 結合 |
| データ比較 (EXCEPT) | transform_compare | 2 つのデータの差分検出 |
| 型変換 (CAST) | transform_cast | integer / float / boolean / date / datetime / string |
| データ検証 (CHECK) | transform_validate | ルール検証（flag / reject） |
ユーティリティ（AI・コード）
| 表示名 | キー | 役割 |
|---|---|---|
| AI エージェント | ai_agent | map / filter / reduce / classify を LLM で実行 |
| VLM画像取得 | source_image_ai | 画像を VLM（失敗時 OCR）で説明文に変換 |
| Python 関数 | transform_code | サンドボックスで transform(data) を実行 |
| シェルコマンド | transform_execute_command | コマンド実行し標準出力を行に付与 |
| 日時操作 | transform_datetime | フォーマット / 加減算 / 抽出 / 差分 |
| 暗号化・ハッシュ | transform_crypto | hash / base64 / URL encode / hmac |
| Markdown / HTML | transform_markdown | Markdown ↔ HTML 変換 |
| XML 処理 | transform_xml | XML ↔ dict 変換 |
条件分岐・繰り返し
| 表示名 | キー | 役割 |
|---|---|---|
| 条件分岐 (IF) | flow_if | true / false に振り分け |
| 分岐 (SWITCH) | flow_switch | 名前つき出力へ多分岐 |
| ループ (FOR EACH) | flow_loop | バッチ分割で逐次 / 並列処理 |
| 待機 (WAIT) | transform_wait | 指定秒スリープ |
| 終了 (BREAK) | flow_break | 条件成立でループを抜ける |
外部システム連携
業務システム（ERP / SFA）・監視・製造 IoT と連携します。接続情報は接続管理で管理します。
業務システム（ERP / SFA）
| 表示名 | キー | 役割 |
|---|---|---|
| Salesforce 取得 / 登録 | source_salesforce / load_salesforce | SOQL 取得 / Composite API 書込（insert / upsert） |
| HubSpot 取得 | source_hubspot | CRM オブジェクト取得 |
| kintone 取得 / 登録 | source_kintone / load_kintone | バルク読込 / insert・upsert |
| freee 取得 | source_freee | freee 会計リソース |
| SmartHR 取得 | source_smarthr | 従業員 / 部署など |
| Sansan 取得 | source_sansan | 名刺 / 連絡先 |
| Money Forward 取得 | source_moneyforward | クラウド会計リソース |
| Backlog 取得 | source_backlog | issue / project / wiki / user |
監視・ログ
| 表示名 | キー | 役割 |
|---|---|---|
| Elasticsearch 取得 / 書込 | source_elasticsearch / load_elasticsearch | クエリ検索 / _bulk 書込 |
| Zabbix 取得 | source_zabbix | 監視データ取得 |
| ログ取得 | source_logs | アプリケーションログを取得 |
| 各種メトリクス | source_pipeline_metrics ほか | パイプライン / チャット / ヘルプデスク / システムの統計 |
製造・IoT（FA 連携） — 工場ラインの PLC / センサに直結します。
| 表示名 | キー | 役割 |
|---|---|---|
| OPC-UA 取得 / 書込 | source_opcua / load_opcua | OPC-UA サーバのノード値を読み書き（PLC データ取得の標準） |
| Modbus 取得 / 書込 | source_modbus / load_modbus | Modbus TCP のレジスタ / コイルを読み書き |
| MQTT 受信 / 送信 | source_mqtt / load_mqtt | MQTT ブローカを購読（有界収集）/ publish |
| SLMP / 三菱 取得 / 書込 | source_slmp / load_slmp | 三菱 PLC（MC プロトコル 3E / 4E）のデバイス読み書き ※実験的 |
| FINS / オムロン 取得 / 書込 | source_fins / load_fins | オムロン PLC（FINS/UDP）のメモリエリア読み書き ※実験的 |
補足
S3 / GCS / SFTP / FTP / SMB / Box / Google Drive / OneDrive / Dropbox / Azure Blob / SharePoint などのファイル転送は、独立オペレータではなく ファイル取得 / 保存（source_file / load_file）に接続 URI を指定して扱います。接続の作り方はファイル管理を参照してください。
主要オペレータのパラメータ
AI エージェント（ai_agent）
| 項目 | 既定 | 説明 |
|---|---|---|
| 操作 | map | map（各行を変換）/ filter（残す行を判定）/ reduce（グループ集約）/ classify（分類） |
| モデル | 文脈の既定 | 使う LLM |
| プロンプト / システムプロンプト | — | 指示文 |
| 出力列 | ai_output（classify は _label） | 結果を入れる列 |
| 分類カテゴリ | — | classify のラベル候補。multi_label で複数付与 |
| temperature ほか | 0.7（classify は 0.1） | チャットと同じ生成パラメータ |
| 行データの自動添付 | オン | 行 JSON をプロンプトに自動添付（オフなら {{ $json.列名 }} 差し込みのみ） |
| 添付する列 | 全列 | 添付する列の絞り込み（トークン削減・列の除外） |
Python 関数（transform_code）
transform(data) を定義します。data は行の配列です。mode は all（全行を一度に）/ each（1 行ずつ）。サンドボックスで実行され、ファイル・ネットワーク・危険な属性アクセスは禁止、timeout_seconds（既定 30）で打ち切られます。
ファイル取得 / 保存（source_file / load_file）
取得は file_path（フォルダ末尾 /・glob・接続 URI・仮想パス myfile/・shared/・external/<label>）、format（未指定なら拡張子で判定）、encoding、delimiter、max_rows など。保存は file_path（必須）、format（既定 csv）、append、compress(.gz)。
SQL 取得 / 保存（source_sql / load_sql）
取得は接続（connectionId）＋ query（SELECT / WITH）または table ＋ columns / where / order_by / limit、差分取得（incremental）。保存は table（必須）、write_disposition（append / replace / merge）、primary_key（merge 用）、batch_size（既定 1000）。
RAG に保存 / 検索（load_to_rag / source_rag_retrieve）
保存は bot_id（必須）、mode（append / replace_all）、chunk_size（600）、chunk_overlap（100）。検索は bot_id（必須）、query、top_k（5）、similarity_threshold。
条件の演算子（filter / IF / break 共通）
== != > >= < <= / contains not_contains starts_with ends_with / regex / exists not_exists / is_empty is_not_empty。match で AND（all）/ OR（any）、case_sensitive の切り替え。
集約関数（transform_aggregate）
count / sum / avg / min / max / count_unique / concat / first / last / collect。group_by 空で全体集約。
代表的なパイプライン例
- 社内 DB をナレッジ化: SQL 取得 → AI エージェント（要約） → RAG に保存
- ファイル取込 → 蓄積: ファイル取得（CSV glob） → 型変換 → データ検証 → データセットに保存
- API → BigQuery: REST API 取得 → 絞り込み → 集計 → BigQuery 書込
- 差分監視 → 画像読取 → 通知: ファイル差分取得 → VLM画像取得 → チャット通知送信
- CRM 連携: kintone 取得 → データ比較（差分） → 条件分岐 → Salesforce 登録
- 監視 → 分類 → メール: メトリクス取得 → AI エージェント（障害分類） → 絞り込み → メール送信