工作流节点类型和组合模式——顺序执行、条件分支、重复和人机循环实用指南

Dify 的 Workflow 是一个强大的功能，可以让您直观地构建以 LLM 为中心的业务逻辑。在本文中，我们将系统地解释每种节点类型的特征以及实践中经常出现的组合模式。重点不是节点的“功能列表”，而是“在哪些场景下如何组合它们”。

1.工作流的基本概念

1.1 聊天流和工作流的区别

Dify 有两种流程构建模式。

1.2 工作流执行模型

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 ノードの入力変数定義例
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节点提示描述示例：

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

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

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

2.3 代码节点

运行 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请求节点

调用外部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 问题分类器节点

使用 LLM 将用户输入分类为预定义类别的节点。允许基于自然语言的分支，这比 IF/ELSE 更灵活。

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

2.7 迭代节点

对数组数据重复执行相同处理的节点。

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

2.8 循环节点

重复处理直到满足条件的节点。迭代是数组遍历，而循环是基于条件的迭代。

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

2.9 人在环（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 ノード設定例
outputs:
  - name: final_result
    type: string
    value: "{{processed_output}}"
  - name: metadata
    type: object
    value: "{{processing_metadata}}"

3. 组合模式

3.1 模式1：顺序执行

最简单的模式。处理从输入到输出呈线性流动。

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

适用场景：

• 内部常见问题解答
• 文件摘要
• 生成标准报告

实施示例：文档摘要工作流程

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：迭代

对多个数据项重复相同过程的模式。

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

适用场景：

• 多个文件的批量分析
• 批量处理（发票处理、合同检查）
• 每条记录的单独评估和汇总

带有代码节点的预处理示例：

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：人在环流程

一种涉及人类对人工智能处理结果的认可的模式。

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

适用场景：

• 合同和法律文件的审查
• 大额订单审批流程
• 当对人工智能判断的信心较低时的回退
• 有合规要求的企业

3.5 模式 5：使用循环提高质量（迭代细化）

一种自动评估LLM输出质量并在不符合标准时重新生成的模式。

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

代码节点中质量评估示例：

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. 工作流程设计最佳实践

5.1 节点设计原则

• 单一职责：将一个进程分配给一个节点。避免将多个任务塞进 LLM 节点
• 显式变量名称：使用有意义的名称，例如 contract_risk_analysis_result 而不是 output
• 错误处理：使用 IF/ELSE 处理每个节点的故障情况
• 可测试性：允许中间节点的输出发送到末端节点进行验证。

5.2 性能考虑因素

5.3 调试技巧

1. 一步一步：用Dify的“一步一步运行”功能检查每个节点的输入和输出
2. 查看日志：在执行日志中查看各节点的处理时间和输出
3. 测试数据：为正常系统和异常系统准备测试数据。
4. 部分执行：仅提取并测试有问题节点周围的区域

6.常见问题及解决方法

7. 总结

设计工作流的关键不是记住各个节点的功能，而是理解节点如何组合在一起以满足业务需求的模式。

我们建议采用分步方法，从顺序执行模式开始了解基本结构，然后根据业务需求添加条件分支和迭代。