フロー構築

データの取り込み・加工・出力をステップの組み合わせで自動化する。AI とのデータ連携に使う。

仕組み

• ステップは上から順に実行される。データは行の集まりとして次のステップへ流れる。流れは 取得 → 加工 → 出力（保存・送信）。
• 外部サービスの認証情報は 外部接続 に保存する。接続は作成者のほか、タグで共有された人も使える。
• 文字列の設定には値を差し込める。旧形式（{{ $json.列名 }} など）も使える。
  • 行の列: {{ 列名 }}
  • 前のステップ: {{ steps["ステップ名"][0].列名 }}
  • 実行情報: {{ sys.today }} / {{ sys.now }} / {{ workflow.run_id }} / {{ env.NAME }} / {{ loop.index }}

フローを作る

1. サイドバーの「フロー構築」を開く。
2. 新規作成 を選ぶ。
3. 名前を入れる。 → 編集画面が開く。
4. ステップ追加 でステップを選ぶ。
5. ステップを設定する。
6. ステップの ステップ実行 で、そのステップまでの結果を確かめる。
7. 保存 する。
8. 実行 を選ぶ。 → 結果は 実行履歴 に残る。

AIで編集 を選ぶと、パネルに指示を書くだけで AI がフロー全体を下書き・修正する。変更内容を確かめて 許可 すると編集画面に反映される。保存 で確定する。

ステップでできること

ステップを選ぶ画面の分類ごとに、できることをまとめる。各ステップの設定項目と既定値はステップ設定リファレンスを見る。

業務システム（ERP / SFA / CRM）とつなぐ

専用のステップは無い。REST API 取得 / REST API 送信 と、保存済みの「REST API」接続を組み合わせる。

クライアント証明書（mTLS）・独自 CA は、REST 接続の「TLS / クライアント証明書」欄で設定する。

• クライアント証明書: PEM（証明書 + 秘密鍵）または PKCS#12（.p12 / .pfx）+ パスワード。oauth2 の token 取得にも使う。
• CA 証明書（PEM）: 社内 CA が発行したサーバー証明書の検証に使う。
• 保存時に検証する。証明書・鍵は暗号化して保存する。

AI エージェントのステップ

AI エージェント は、APIアクセスキーで作ったアシスタント（モデル・ナレッジ・ツールの設定）に、行のデータについて指示を出すステップ。

1. API Access Key で使うアシスタントのキーを選ぶ。
2. 処理モード を選ぶ。
   • 各行ごと — 行ごとに 1 回実行する。行のデータは自動で添付される。
   • 全行まとめて1回 — 全部の行をまとめて 1 回実行する。要約・集計に向く。
3. プロンプト に指示を書く。{{ 列名 }} で行の値を差し込める。
4. 出力 を選ぶ。
   • テキスト — 結果を 出力カラム（既定 ai_result）に書く。
   • 構造化 — 出力フィールド に列名・型・説明を並べると、結果をその列に分けて書く。

詳細設定 では次を選べる。

• 添付する列の絞り込み
• ファイル添付列（行にあるファイルを添付して読ませる）
• 並列数（最大 8）
• 行が失敗した時の扱い

自動で動かす

フローは 実行 による手動実行のほか、決まった時刻・外部からの呼び出し・社内の出来事をきっかけに動かせる。

編集画面上部の トリガ を選ぶと トリガ設定 が開く。3 つのタブで設定と状態を確かめられる。

決まった時刻に動かす（スケジュール）

1. フローを保存する。
2. 編集画面上部の トリガ を選ぶ。 → トリガ設定 の スケジュール タブが開く。
3. 一覧から間隔を選ぶ（5分ごと / 15分ごと / 30分ごと / 1時間ごと / 毎朝 9:00 / 平日 9:00 / 毎日 18:00 / 毎日 0:00 / 毎週月曜 0:00 / 毎月1日 0:00）。 一覧に無い時は Custom (cron) に「分 時 日 月 曜日」の形で入れる（例 0 9 * * 1-5）。@daily などの別名や、@every 30s のような間隔（10 秒以上）も使える。
4. スケジュールを有効にする をオンにする。
5. スケジュール設定 を選ぶ。 → 指定した時刻に動く。フロー一覧の「スケジュール済み」に数えられる。

止める時は スケジュールを有効にする をオフにするか、スケジュール解除 を選ぶ。

外部のシステムから呼び出す（Webhook）

1. フローの先頭に Webhook 受信 ステップを置く。
2. ステップの Secret token でトークンを作る。
3. フローを保存する。
4. トリガ を選ぶ。
5. Webhook 受信 タブを開く。 → 受信 URL と curl の例が表示される。
6. 呼び出す側から受信 URL へ POST する。ヘッダ X-Webhook-Token にトークンを付ける。 → 送った内容を入力としてフローが動く。

トークンが無いと呼び出しは拒否される。同じ配送 ID の再送では二重に動かない。

社内の出来事をきっかけに動かす（イベント）

1. フローの先頭に イベント受信 ステップを置く。

2. イベント で、きっかけにする出来事を選ぶ。

   イベント	動く時	対象で絞り込めるもの
   ファイル作成・上書き時	ファイルがアップロード・保存された時	フォルダやファイルのパス（例 myfile/inbox/ や myfile/inbox/*.csv）
   ファイル削除時	ファイルが削除された時	同上
   ファイル名変更・移動時	ファイルの名前が変わった・移動された時	変更前のパス
   パイプライン完了時	他のフローが成功して終わった時	フロー
   パイプライン失敗時	他のフローが失敗した時	フロー
   データセット更新時	データセットに書き込まれた時	データセット
   RAG 取込完了時	ナレッジへの取り込みが終わった時	ナレッジ

3. 必要なら 対象 で絞り込む（空にするとすべてが対象）。

4. フローを保存する。 → トリガ の イベント タブに、設定したイベントと「有効」が表示される。

一時的に止める時は、ステップの 有効 をオフにする。

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

ステップ実行 は上流から順に実行して結果を表示する。外部への書き込み・送信をするステップは実行せず、入力行をそのまま次へ渡す。

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

各ステップの 再試行とエラー時の動作 で、次を設定できる。外部に書き込むステップは再試行しない。

無人運転の挙動

• 実行履歴の「差分の根拠」 — 差分取得したステップで、追加・更新・削除の内訳を確かめられる。処理済みの記録は、実行が成功した時だけ確定する。
• 連続失敗で自動停止 — 管理者が設定すると、続けて失敗したフローの自動起動が止まる。一覧に「自動停止中」と表示される。手動で実行して成功すると解除される。
• 停止中に逃したファイル・スケジュールの回収や、中断した実行の再実行は、管理者が設定する（環境変数）。

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

重いフローを、別のプロセスや別のマシンの worker で動かせる。設定は ADMIN / SUPER が行う。

1. フロー構築の一覧で 実行先 を選ぶ。 → ワークフローの実行先 が開く。

2. フローごとに実行先（pool 名、既定 default）を選ぶ。

3. 外部の worker を、本体と同じ .env で起動する。

   db worker --pool <名前>
   

   pip / uv で入れた場合は次のコマンドにする。

   digitalbase worker --pool <名前>
   

   常駐させる時は systemd などに登録する。

その pool の worker がいないと、実行は「待機中」のままになる。

代表的なフローの例

1. 社内 DB をナレッジにする: SQL 取得 → AI エージェント（要約） → データを RAG に保存
2. ファイルを取り込んで蓄積する: ファイル取得（CSV をまとめて） → 型変換 → データ検証 → データセットに保存
3. API から BigQuery へ: REST API 取得 → 行の絞り込み → 集計 → BigQuery 書込
4. 届いた画像を読み取って通知する: ファイル差分取得 → 画像・動画の読み込み → チャット通知送信
5. CRM 連携: REST API 取得（kintone） → データ比較 → 条件分岐 → REST API 送信（Salesforce）
6. ログを分類してメールする: Elasticsearch 取得 → AI エージェント（障害分類） → 行の絞り込み → メール送信
7. 複数ファイルを集計して SQL・BI で見る: ファイル取得（CSV をまとめて） → SQL で変換 → データセットに保存（ダッシュボード・BI から参照）

関連

• ステップ設定リファレンス — 各ステップの設定項目
• フォルダ — 接続フォルダとファイルの扱い
• ダッシュボード — データセットの参照
• RAG設定