Dify 实验 · 中级

Dify 中级实验（04）：迭代进阶——如何批量处理数据并守住性能边界？

1. 业务场景

先讲一个我们实际遇到的场景。

一家做产品的公司，用户反馈散落在多个渠道：App Store 评论、Google Play 评论、官方论坛帖子。产品团队每周要汇总分析一次，反馈里还混着大量垃圾内容——一句话灌水、纯标点、重复字符，直接送进情感分析会把结论带偏。人工逐条看，几百条评论看到眼瞎。

我们第一次接这类需求时，第一反应也是「在代码节点里写个 for 循环，一次跑完」。真正动手才发现——循环好写，逐条分流、嵌套遍历、输出收集全要自己造，数组一大还撞上限。后来翻 Dify 的节点列表才发现：平台原生的 Iteration（迭代） 就是干这个的——输入数组自动逐条处理，支持嵌套迭代、逐条条件分流、输出结构设计。

这不是个例。任何「拿到一个数组、逐条处理」的业务场景都是这个模式：批量取数、逐条审核、分页采集、批处理流水线——数组进来了，每条都要过一道，还要控制每条的走向。

2. 场景痛点

这个流程的痛点，在产品团队身上体现得最直接：

逐条处理靠人肉循环：数据源多、条数多，人工一条条看、一条条记，一天时间就耗在重复劳动上——处理动作不产生任何增量价值。
垃圾评论污染分析：低质评论混进情感分析，结论被灌水内容带偏，产品决策基于失真数据——比没数据更危险。
串行处理慢：一个数据源一个数据源地跑，整体耗时长，反馈时效性差——用户上周的抱怨，这周才被看见。
性能边界不可控：数组一大就撞上迭代上限/超时，上线前没人知道边界在哪，生产环境才炸——边界不是报错那一刻才知道的，是要提前设计出来的。

本质上，批量数据处理的核心不是「能不能循环」，而是「循环的边界和每一条的质量」——这两点控制不住，批处理就是灾难。

3. 方案：为什么是迭代节点

选迭代节点的理由，我们实际对比过：

原生数组处理：输入数组自动逐条迭代，嵌套迭代支持「数据源 → 评论」两级结构，一条链跑完所有；
逐条分流：迭代内可挂 IF-ELSE——质量过滤后再送 LLM，低质评论根本不进模型，省 Token 且结论干净；
边界透明：30 个元素上限、10 分钟超时、并行数限制都是真实约束，提前设计不踩雷。

这篇文章我们就用它搭一个「多源反馈批量分析器」：外层迭代遍历数据源，内层迭代逐条做质量过滤，只有合格的评论才送 LLM 做情感分析。

4. 整体架构

graph TD start["开始：sources JSON"] parse["解析数据源配置：Code json.loads + 截断 [:20]"] subgraph iter_outer["外层迭代：遍历数据源（iter_sources）"] fetch["模拟获取数据源评论：Code 按 limit 生成 mock + 0.2s 延迟"] subgraph iter_inner["内层迭代：逐条质量过滤与分析（iter_reviews）"] filter["质量过滤：Code quality_ok 判定"] rule{"质量合格判断：IF-ELSE"} sentiment["情感分析：LLM"] collect["汇聚分析结果：Code"] end end summary["汇总统计：Code"] end1["结束"] start --> parse --> iter_outer iter_outer --> fetch --> iter_inner iter_inner --> filter --> rule rule -- "case_ok" --> sentiment --> collect rule -- "case_bad" --> collect collect --> summary --> end1

链路很清晰：入口收数组 → 外层迭代遍历数据源 → 内层迭代逐条过滤与分析 → 汇总统计。两级迭代是架构核心，质量过滤卡在 LLM 之前，保证分析只吃高质量数据。

5. 模块设计

5.1 开始节点变量

variables:

- label: 数据源配置（JSON 数组，每项含 name/url/limit）

  required: true

  type: paragraph        # 长 JSON 粘贴，paragraph 而非 text-input

  variable: sources

5.2 外层迭代容器

迭代输入数组有 30 个元素上限，解析节点先截断留余量：

def main(sources_text: str) -> dict:

    import json

    try:

        data = json.loads(sources_text or "[]")

        if not isinstance(data, list):

            data = []

    except Exception:

        data = []

    data = data[:20]              # 迭代上限 30，提前截断

    return {"source_items": data, "total": len(data)}

外层迭代容器配置（节选）：

- data:

    error_handle_mode: terminated

    is_parallel: false

    iterator_selector: [cd_parse, source_items]

    output_selector: [iter_reviews, output]   # 内层迭代的输出

    output_type: array[string]

    parallel_nums: 10

    start_node_id: itstart0

    title: 外层迭代：遍历数据源

    type: iteration

  id: iter_sources

5.3 质量过滤（Code）

低质判定逻辑输出 quality_ok——注意布尔值展平为字符串：变量选择器看不到 boolean 类型，下游 if-else 也只用字符串比较：

def main(review: dict) -> dict:

    import re

    review = review or {}

    content = str(review.get("content", "") or "")

    quality_issues = []

    if len(content) < 3:

        quality_issues.append("内容过短")

    if re.match(r"^[!！.。?？，,。]+$", content):

        quality_issues.append("无实质内容")

    if len(set(content)) <= 2 and len(content) > 1:

        quality_issues.append("疑似垃圾评论")

    return {

        "quality_ok": "true" if len(quality_issues) == 0 else "false",  # 字符串，不是布尔

        "quality_issues": quality_issues,

        ...

    }

5.4 内层分支与 LLM

IF-ELSE 用 is 比较字符串（不能用 =）：

cases:

- case_id: case_ok

  conditions:

  - comparison_operator: is

    value: "true"

    variable_selector: [cd_quality, quality_ok]

  logical_operator: and

- case_id: case_bad

  conditions:

  - comparison_operator: is

    value: "false"

    variable_selector: [cd_quality, quality_ok]

  logical_operator: and

迭代内 LLM 引用当前元素用 {{#iter_reviews.item.字段#}}，且必须显式 reasoning_format: separated——否则思考过程混入 text，整个迭代输出数组都被污染：

prompt_template:

- id: p_analyze

  role: system

  text: |

    分析以下用户评价：

    平台：{{#iter_reviews.item.platform#}}

    用户：{{#iter_reviews.item.user#}}

    评分：{{#iter_reviews.item.rating#}}/5

    评价内容：{{#iter_reviews.item.content#}}

    输出分析结论（50字以内）：

    1. 情感倾向（正面/中性/负面）

    2. 核心关注点

    3. 可采取的行动建议

reasoning_format: separated

6. 运行验证

输入上方 sources 测试数据，观察：

检查项	预期	实测
外层迭代次数	3（3 个数据源）	3
内层迭代总次数	9（3+2+4）	9
质量过滤拦截	低质评论被拦，不走 LLM	与预期一致
汇总输出	扫描 9 条 / 有效 N 条 / 拦截 M 条	与预期一致

进阶对比：把外层迭代 is_parallel 打开（并行数 3）再跑一次，对比串行/并行耗时——本实验 mock 延迟 0.2s，数据量小看不出差距，数据量大了差异明显。

7. 实战坑

坑	现象	修复
迭代输入数组超 30 个	运行报 `then length of var "item" must be less than 30 elements`	上游代码节点 `data[:20]` 提前截断留余量
output_selector 选 array[object]	迭代输出为空数组，下游取不到	内部代码节点 `json.dumps` 序列化为 string，`output_type: array[string]`
布尔输出给下游判断	变量选择器看不到 boolean，取不到值	输出 `"true"`/`"false"` 字符串，if-else 用 `is` 比较
嵌套迭代内部节点 parentId 写错层级	校验报「parentId=None 不是 iteration 节点」	parentId 是节点外层字段（与 id 同级），不是 data 内
迭代内 LLM 缺 reasoning_format	思考过程混入 text，判断逻辑全乱	显式 `reasoning_format: separated`

💡 嵌套迭代格式本身能通过校验，但 Dify 运行层对「迭代套迭代」支持有限——如果你的场景不需要逐源再逐条的两级结构，优先用「串联迭代 + 展平代码节点」更稳。

8. 实验文档及源码获取

实验文档（完整操作步骤）：DIFY-05：迭代进阶——批量处理的边界与陷阱.md
源码（可直接导入）：dify102_05_批量反馈分析器.yml

文章聚焦核心配置与采坑点；实验的完整分步操作（节点搭建/参数表/调试指引）见实验文档原文。

本系列 · Dify 实验 · 中级