Workflow ノードタイプと組合せパターン – 順次実行・条件分岐・反復・Human-in-the-Loop の実践ガイド

Dify の Workflow は、LLM を中心としたビジネスロジックをビジュアルに構築できる強力な機能です。本記事では、各ノードタイプの特性と、実務で頻出する組合せパターンを体系的に解説します。ノードの「機能一覧」ではなく、「どのシーンでどう組み合わせるか」に重点を置いています。

1. Workflow の基本概念

1.1 Chatflow と Workflow の違い

Dify には2種類のフロー構築モードがあります。

1.2 Workflow の実行モデル

flowchart LR
    START[Start ノード] --> PROCESS[処理ノード群]
    PROCESS --> END[End ノード]
    
    subgraph PROCESS[処理フロー]
        direction TB
        A[LLM] --> B[Code]
        B --> C[IF/ELSE]
        C --> D[HTTP Request]
    end

2. 主要ノードタイプの詳細

2.1 Start ノード

Workflow のエントリーポイント。入力変数を定義します。

# Start ノードの入力変数定義例
variables:
  - name: document_text
    type: string
    required: true
    description: "処理対象のドキュメントテキスト"
  - name: language
    type: select
    options: ["ja", "en", "zh"]
    default: "ja"
  - name: file
    type: file
    required: false
    allowed_types: ["pdf", "docx"]

2.2 LLM ノード

LLM を呼び出してテキスト生成・分類・抽出などを行う中核ノード。

設定項目:

LLM ノードの Prompt 記述例:

以下のテキストを分析し、JSON形式で結果を返してください。

入力テキスト:
{{document_text}}

出力形式:
{
  "summary": "100文字以内の要約",
  "category": "contract | invoice | report | other",
  "risk_level": "high | medium | low",
  "key_entities": ["抽出されたエンティティのリスト"]
}

2.3 Code ノード

Python または JavaScript コードを実行するノード。データ変換、計算、フォーマット処理に使用します。

# Code ノード例: LLM出力のJSON解析と検証
import json

def main(llm_output: str) -> dict:
    """LLMの出力をパースして構造化データに変換"""
    try:
        result = json.loads(llm_output)
        # バリデーション
        required_keys = ["summary", "category", "risk_level"]
        for key in required_keys:
            if key not in result:
                result[key] = "N/A"
        # risk_level の正規化
        valid_levels = {"high", "medium", "low"}
        if result.get("risk_level") not in valid_levels:
            result["risk_level"] = "medium"
        return {"result": result, "is_valid": True}
    except json.JSONDecodeError:
        return {"result": None, "is_valid": False}

2.4 HTTP Request ノード

外部 API を呼び出すノード。社内システム連携や外部サービスとの統合に使用します。

# HTTP Request ノード設定例
method: POST
url: "https://api.internal.example.com/tickets"
headers:
  Content-Type: "application/json"
  Authorization: "Bearer {{api_token}}"
body:
  type: json
  content: |
    {
      "title": "{{ticket_title}}",
      "description": "{{llm_output}}",
      "priority": "{{risk_level}}",
      "assignee": "auto"
    }
timeout: 30
retry:
  max_retries: 3
  retry_interval: 5

2.5 IF/ELSE ノード（条件分岐）

条件に基づいてフローを分岐させるノード。

# IF/ELSE 条件設定例
conditions:
  - if: "{{category}} == 'contract' AND {{risk_level}} == 'high'"
    then: "法務レビューフローへ"
  - elif: "{{category}} == 'contract' AND {{risk_level}} != 'high'"
    then: "自動処理フローへ"
  - else: "一般処理フローへ"

条件式で使用可能な演算子:

2.6 Question Classifier ノード

LLM を使ってユーザー入力を事前定義したカテゴリに分類するノード。IF/ELSE よりも柔軟な自然言語ベースの分岐が可能です。

# Question Classifier 設定例
model: gpt-4o-mini
classes:
  - name: "技術サポート"
    description: "製品の技術的な問題、エラー、設定に関する質問"
  - name: "料金・契約"
    description: "プラン、料金、契約更新、解約に関する質問"  
  - name: "一般問い合わせ"
    description: "その他の質問、フィードバック、要望"

2.7 Iteration（反復）ノード

配列データに対して同一処理を繰り返し実行するノード。

# Iteration ノード設定例
input: "{{document_list}}"  # 配列型の変数
# 配列の各要素に対して内部フローを実行
# 内部フローでは {{item}} で現在の要素を参照
output: "{{results}}"  # 処理結果の配列
parallel_mode: false  # 順次実行（true で並列実行）
max_iterations: 50    # 最大反復回数
error_handling: continue  # エラー時も継続

2.8 Loop（ループ）ノード

条件が満たされるまで処理を繰り返すノード。Iteration が配列走査なのに対し、Loop は条件ベースの繰り返しです。

# Loop ノード設定例
max_iterations: 10
break_condition: "{{quality_score}} >= 0.9 OR {{iteration_count}} >= 5"
# ループ内で LLM による生成 → Code で品質スコア計算 → 条件判定

2.9 Human-in-the-Loop（HITL）ノード

人間の判断・承認を挟むノード。高リスクな処理や低確信度の結果に対して人間のレビューを求めます。

# HITL ノード設定例
title: "契約書レビュー承認"
description: "AIが高リスクと判定した契約書です。内容を確認して承認してください。"
display_variables:
  - "{{contract_summary}}"
  - "{{risk_points}}"
  - "{{recommended_action}}"
actions:
  - label: "承認"
    value: "approved"
  - label: "差し戻し"
    value: "rejected"  
  - label: "法務部にエスカレーション"
    value: "escalated"
timeout: 86400  # 24時間（秒）

2.10 End ノード

Workflow の出力を定義する終端ノード。

# End ノード設定例
outputs:
  - name: final_result
    type: string
    value: "{{processed_output}}"
  - name: metadata
    type: object
    value: "{{processing_metadata}}"

3. 組合せパターン

3.1 パターン1: 順次実行（Sequential）

最もシンプルなパターン。入力から出力まで直線的に処理が流れます。

flowchart LR
    S[Start] --> L1[LLM: 分類]
    L1 --> K[Knowledge Base 検索]
    K --> L2[LLM: 回答生成]
    L2 --> E[End]

適用シーン:

• 社内 FAQ 応答
• ドキュメント要約
• 定型レポート生成

実装例: ドキュメント要約ワークフロー

3.2 パターン2: 条件分岐（Branching）

入力内容や中間結果に応じて処理パスを切り替えるパターン。

flowchart TD
    S[Start] --> QC[Question Classifier]
    QC -->|技術サポート| KB1[Knowledge: 技術文書]
    QC -->|料金・契約| KB2[Knowledge: 契約情報]
    QC -->|一般| L1[LLM: 一般回答]
    KB1 --> L2[LLM: 技術回答生成]
    KB2 --> L3[LLM: 契約回答生成]
    L1 --> E[End]
    L2 --> E
    L3 --> E

適用シーン:

• マルチドメインの問い合わせ対応
• 言語による処理分岐
• リスクレベルに応じた処理変更

3.3 パターン3: 反復処理（Iteration）

複数のデータ項目に対して同一の処理を繰り返すパターン。

flowchart TD
    S[Start: 文書リスト] --> SP[Code: リスト分割]
    SP --> IT[Iteration]
    subgraph IT[反復処理]
        direction LR
        I1[LLM: 分析] --> I2[Code: スコア算出]
    end
    IT --> AG[Code: 結果集約]
    AG --> E[End]

適用シーン:

• 複数ファイルの一括分析
• バッチ処理（請求書処理、契約書チェック）
• 各レコードの個別評価と集約

Code ノードでの前処理例:

def main(raw_text: str) -> dict:
    """テキストを段落リストに分割"""
    paragraphs = [p.strip() for p in raw_text.split("\n\n") if p.strip()]
    return {"items": paragraphs, "count": len(paragraphs)}

3.4 パターン4: Human-in-the-Loop 付きフロー

AI の処理結果に対して人間の承認を挟むパターン。

flowchart TD
    S[Start] --> L1[LLM: 契約書分析]
    L1 --> IF{リスクレベル判定}
    IF -->|高リスク| HITL[Human Review]
    IF -->|低リスク| AUTO[自動承認]
    HITL -->|承認| PROC[処理実行]
    HITL -->|差し戻し| L1
    HITL -->|エスカレーション| ESC[法務部通知]
    AUTO --> PROC
    PROC --> HTTP[HTTP: 社内システム登録]
    HTTP --> E[End]
    ESC --> E

適用シーン:

• 契約書・法務ドキュメントのレビュー
• 高額な発注の承認フロー
• AI 判定の確信度が低い場合のフォールバック
• コンプライアンス要件がある業務

3.5 パターン5: ループによる品質向上（Iterative Refinement）

LLM の出力品質を自動評価し、基準に満たない場合は再生成するパターン。

flowchart TD
    S[Start] --> L1[LLM: 初回生成]
    L1 --> C1[Code: 品質スコア算出]
    C1 --> IF{スコア >= 閾値?}
    IF -->|Yes| E[End: 出力]
    IF -->|No| L2[LLM: フィードバック付き再生成]
    L2 --> C1

Code ノードでの品質評価例:

import json
import re

def main(llm_output: str, expected_format: str) -> dict:
    """LLM出力の品質を評価"""
    score = 0.0
    feedback = []
    
    # JSON 形式の妥当性チェック
    try:
        parsed = json.loads(llm_output)
        score += 0.4
    except json.JSONDecodeError:
        feedback.append("出力がJSON形式ではありません")
        return {"score": score, "feedback": "; ".join(feedback)}
    
    # 必須フィールドの存在チェック
    required_fields = ["summary", "category", "risk_level"]
    present = sum(1 for f in required_fields if f in parsed)
    score += 0.3 * (present / len(required_fields))
    
    missing = [f for f in required_fields if f not in parsed]
    if missing:
        feedback.append(f"不足フィールド: {', '.join(missing)}")
    
    # 要約の長さチェック
    if "summary" in parsed and 10 <= len(parsed["summary"]) <= 200:
        score += 0.3
    else:
        feedback.append("要約の長さが不適切です（10-200文字）")
    
    return {
        "score": round(score, 2),
        "feedback": "; ".join(feedback) if feedback else "OK",
        "is_acceptable": score >= 0.8
    }

4. 典型的なビジネスシーン別の推奨構成

5. Workflow 設計のベストプラクティス

5.1 ノード設計の原則

• 単一責任: 1ノードに1つの処理を割り当てる。LLM ノードに複数タスクを詰め込まない
• 明示的な変数名: output ではなく contract_risk_analysis_result のように意味のある名前を使う
• エラーハンドリング: 各ノードの失敗ケースを IF/ELSE で処理する
• テスト容易性: 中間ノードの出力を End ノードに出して検証できるようにする

5.2 パフォーマンスの考慮

5.3 デバッグ手法

1. ステップ実行: Dify の「Run step by step」機能で各ノードの入出力を確認
2. ログ確認: 実行ログで各ノードの処理時間と出力を確認
3. テストデータ: 正常系・異常系の両方のテストデータを準備
4. 部分実行: 問題のあるノード周辺だけを切り出してテスト

6. よくある問題と対策

7. まとめ

Workflow の設計で重要なのは、個々のノードの機能を暗記することではなく、ビジネス要件に対してどのノードをどう組み合わせるかというパターンを理解することです。

まずは順次実行パターンで基本構造を理解し、その後ビジネス要件に応じて条件分岐や反復処理を追加していく段階的なアプローチを推奨します。