フロー構築

データの取り込み・変換・出力をステップで組み合わせて自動化します。複雑な統合ツールではなく、シンプルな UI で AI とのデータ連携に特化しています。

仕組み

• ステップを上から順に実行します。データは**行の集まり（list[dict]）**として各ステップを流れます。
• ステップは大きく 取得(source)→ 加工(transform)→ 出力(load / action) に分かれます。
• 外部サービスの認証情報は 接続管理(ApiConnection) に保存し、ステップから参照します（接続は作成した本人のみが利用できます。管理者も他人の接続は使えません）。
• ステップの利用にロール制限はありません。シェルコマンド・rsync・Python 関数を含め、すべてのステップを全ユーザーが利用できます。
• 値の差し込みにはテンプレートが使えます（Jinja / GitHub Actions と同じ見た目）: 今の行の列は {{ 列名 }} / {{ order.customer.name }} / {{ items[0].id }}、前のステップの出力は {{ steps["ステップ名"][0].列名 }}、実行情報は {{ sys.today }} / {{ sys.now }} / {{ sys.count }} / {{ workflow.name }} / {{ workflow.run_id }} / {{ env.NAME }} / {{ loop.index }}。steps / workflow / env / sys / loop / data は予約語で、行の列より優先されます。旧形式（{{ $json.列名 }} / {{ $node["…"].json }} / {{ $today }} など）も引き続き同じ意味で使えます。
• テンプレートが式 1 個だけ（地の文なし）のとき、「値をセット」では list / dict が文字列化されず構造のまま列に入ります。文と混ぜると JSON 文字列になります。
• ステップ名を変更しても、他ステップの steps["…"] 参照は自動で追随します。
• AI 作成: エディタの「AI 作成」パネルに言葉で指示すると、パイプライン全体（ステップと接続線）を AI が下書き・修正します。変更案を承認するとエディタに反映され、保存で確定します。

共通の設定（全ステップ）

各ステップには、失敗時の挙動を制御する共通オプションがあります。

オペレータ一覧

ファイル連携

API 連携

SQL・データ蓄積

データ処理

ユーティリティ（AI・コード）

条件分岐・繰り返し

外部システム連携

業務システム（ERP / SFA）・監視・製造 IoT と連携します。接続情報は接続管理で管理します。

業務システム（ERP / SFA / CRM）

専用ステップは持たず、「REST API 取得」「REST API 送信」+ 保存済みの「REST API」接続で繋ぎます（認証ヘッダ・OAuth2 client credentials・ページングに対応）。設定例:

書き込みは「REST API 送信」で行単位に POST / PATCH します。

監視・ログ

製造・IoT（FA 連携） — 工場ラインの PLC / センサに直結します。

主要オペレータのパラメータ

AI エージェント（ai_agent）

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）。差分取得は 2 方式: カーソル（cursor_column の前回最大値より大きい行だけを DB に絞らせる。lookback で基準を少し戻して遅れて届いた行を拾い直せる。更新日時や連番のある表向け、削除は取れない）と スナップショット（primary_key と行のハッシュを台帳に持ち、追加・変更・削除を突き合わせる。emit_deleted で消えた行を削除イベントとして出力し、「RAG に保存」の upsert が反映する。更新日時の無い表や削除を伝えたい表向け、SQL_SNAPSHOT_MAX_ROWS 以下の行数）。スナップショットの台帳はファイル差分取得と同じ仕組みで、失敗した実行では確定しません。保存は 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 の切り替え。

比較は値の型を見て行います。両辺が数値なら数値（"1,000" > 999 は真）、日付なら日付（2026/09/04 == 2026-09-04 は真）、それ以外は文字列（既定は大文字小文字を無視）。片側だけが数値・日付の場合は従来どおり文字列として比べ、その旨を判定理由に残します。type（number / date / text / bool）で型を固定でき、value には {{ $json.列名 }} などの差し込み、field には a.b.c のネスト指定が使えます。実行履歴の各ステップに「条件の判定」（先頭 10 行の判定根拠）が表示されます。

集約関数（transform_aggregate）

count / sum / avg / min / max / count_unique / concat / first / last / collect。group_by 空で全体集約。

ステップ実行（プレビュー）の挙動

編集画面の「ステップ実行」は上流から順に実行して結果を表示しますが、外部に書き込む・送信するステップ（SQL 保存、ファイル保存、メール、Webhook 送信、通知、SaaS 登録など）は実行しません。該当ステップは「書込のためプレビューでは実行しません」と表示され、入力行がそのまま次のステップに渡ります。実際の書込は「実行」で行ってください。プレビューは PIPELINE_PREVIEW_TIMEOUT_SEC（既定 120 秒）で打ち切られます。

共通の詳細設定（全ステップ）

各ステップの設定下部「詳細設定」で、失敗時の再試行（回数・待ち時間）、失敗時の扱い（停止 / 続行 / エラー行を付けて続行）、入力 0 行でも実行するか、を設定できます。外部に書き込むステップでは二重書込を防ぐため再試行は行われません。「続行」で握った失敗があると、その実行では差分取得のカーソルやファイルのスナップショットは確定せず、次回に持ち越されます。

無人運転の挙動

• 停止中の取りこぼしを取り直す: 外部ディレクトリ監視は、サーバ停止中に置かれたファイルを起動時に走査して発火します。スケジュール実行も、停止中に逃した直近 1 回分を起動時に実行します。同じ内容・同じ配送を二重に実行しないよう、実行には冪等キーが付きます。
• 中断された実行は再実行: 再起動などで中断した実行は自動で待機列に戻ります（PIPELINE_RUN_MAX_ATTEMPTS、既定 2 回まで）。
• 連続失敗は自動停止: 同じワークフローが連続で失敗すると（PIPELINE_AUTO_PAUSE_AFTER_FAILURES、既定 5 回）自動起動（イベント / スケジュール / Webhook）を一時停止し、一覧に「自動停止中」と表示します。原因を直して手動実行すると解除されます。
• 差分は根拠つきで残る: ファイル着荷の差分（追加 / 更新 / 削除と前後の size・更新時刻）、SQL 差分取得のカーソル範囲、RAG 同期で削除したキーは実行履歴の「差分の根拠」で確認できます。処理済みの記録（カーソル・スナップショット）は、実行が成功しエラーのあるステップが無い場合にだけ確定します。失敗した分は次回に持ち越されます。
• 同サイズ・同時刻の上書き: ファイル着荷の content_hash を有効にすると、内容のハッシュで変更を検知します（既定は無効）。

実行先（worker pool）— 重いワークフローを別プロセス / 別マシンへ

ワークフローの実行は本体に内蔵された常駐 worker が行いますが、AI 推論や大量データ処理など重いものは、別プロセスや別マシンの worker に逃がせます。仕組みは「待機列（DB）から worker が取りに来る」方式で、worker にポートはありません。

• 各ワークフローには 実行先（pool 名） があり、既定は default（内蔵 worker）。変更できるのは管理者だけです（資源の配分のため。管理画面 > システム > 「ワークフローの実行先」で、稼働中の worker を見ながら付け替えます）。
• 外部 worker は本体と同じ .env（同じ DB）で digitalbase worker --pool <名前> を起動します（.env の PIPELINE_WORKER_POOL でも可）。名前が一致したワークフローの実行だけを取ります。
• その pool の worker が居ないと実行は「待機中」のまま止まります（管理画面に警告が出ます）。worker の生存は heartbeat（PIPELINE_WORKER_HEARTBEAT_SEC）で判定し、別マシンの worker が落ちた実行も自動で待機列に戻ります。
• 外部 worker は SIGTERM で「実行中のものを終えてから」停止します。

代表的なパイプライン例

1. 社内 DB をナレッジ化: SQL 取得 → AI エージェント（要約） → RAG に保存
2. ファイル取込 → 蓄積: ファイル取得（CSV glob） → 型変換 → データ検証 → データセットに保存
3. API → BigQuery: REST API 取得 → 絞り込み → 集計 → BigQuery 書込
4. 差分監視 → 画像読取 → 通知: ファイル差分取得 → VLM画像取得 → チャット通知送信
5. CRM 連携: REST API 取得（kintone） → データ比較（差分） → 条件分岐 → REST API 送信（Salesforce）
6. 監視 → 分類 → メール: メトリクス取得 → AI エージェント（障害分類） → 絞り込み → メール送信

関連

• ファイル — 接続 URI とファイルの扱い
• ダッシュボード — データセットと BigQuery
• RAG設定