Dify 中级实验（20）：综合实战——自动化报告生成流水线如何从数据到周报一步到位？

1. 业务场景

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

一家公司的 CEO 每周一早上要一份上周业务运营报告：销售对比、客户数据、客服数据、异常预警、趋势分析。以前这份报告全靠分析师手工做——从各个系统导出数据、在 Excel 里对账、写分析、排版成文档，每周一早上雷打不动要耗 3 个小时，而且周一上午往往是分析师最忙的时候：数据还没整理完，CEO 已经在催了。数据要得急、口径还不统一，这个月销售算的是含税收入、下个月又不含了，报告里的数字对不上，CEO 在会上当面质疑。

我们第一次接这类需求时，第一反应也是「写个脚本定时跑不就行了」。真正梳理完整条链路才发现——脚本只能把数据采回来，报告还得有人写：对账、分析、排版、推送，每一步都卡着人工，而这条链路恰恰是流水线化改造的最佳对象。

这不是个例。任何「周期性、重复性、数据源固定的报告」场景都是这个模式：周报、月报、日报、经营分析会材料、销售战报——只要数据来源稳定、格式固定，就是流水线化改造的最佳对象。

2. 场景痛点

这个流程的痛点，在分析师的周一早上体现得最直接：

• 手工 3 小时重复劳动：每周导出、对账、写分析、排版，同样的流程重复几十次，分析师的价值被耗在复制粘贴上，没有时间做真正的深度分析。
• 数据口径不统一：销售、客户、客服数据分散在几个系统，各自格式不一，手工对账容易错——报告里的数字前后对不上，被 CEO 当众质疑。
• 异常发现滞后：营收环比下降 10% 这种关键异常，靠人肉比对往往要拖到周二才发现，错过了及时干预的窗口。
• 报告格式不固定：每次排版全凭分析师手感，这周表格下周图表，格式变来变去，阅读成本高。

本质上，「从数据采集到报告推送」是一条完全可以自动化的流水线——多路采集、聚合、清洗、分析、模板、分支、Webhook 串起来，人只负责看结果、做决策。

3. 方案：为什么是「全自动报告流水线」

本实验是系列 1 的收官检验：综合运用 DIFY-102 全部进阶知识（多路采集、聚合、清洗、分析、模板、分支、Webhook），构建一条从数据采集到报告推送的全自动流水线。

选它的理由：

• 多路并行采集 + 聚合：销售/客户/客服三路数据并行采集，聚合节点统一汇总，数据口径一次定死；
• 清洗与异常检测内置：空值兜底、环比计算、异常计数全部代码化，异常不再靠人肉发现；
• 零 Token 模板渲染 + 按需推送：周报正文用模板转换（Jinja2）渲染，不消耗 Token；输出格式可选，异常时 Webhook 自动推送到群/系统。

这篇文章我们就用它搭一个「周报自动生成器」：输入 report_week，自动完成三路数据采集 → 聚合 → 清洗与异常检测 → 趋势分析 → 按格式输出 Markdown 周报或 JSON 报告，异常时自动推送告警。

4. 整体架构

链路很清晰：入口收报告周 → 三路并行采集 → 聚合 → 清洗与异常检测 → 趋势分析 → 按格式分支输出 → 异常时 Webhook 告警。关键设计是「检查类分支必须消费判断结果」——告警条件不消费，告警就是死代码。

5. 模块设计

5.1 三路数据采集（Code）

三个采集节点并行（拓扑上从开始直接分 3 路），各自返回模拟数据——这是多路并行 + 聚合的标准形态：

def main(report_week: str) -> dict:

    # 模拟从数据库 A 拉取销售数据

    import json

    sales = [

        {"week": "上上周", "revenue": 480000, "orders": 960},

        {"week": "上周", "revenue": 520000, "orders": 1040},

    ]

    return {"sales": json.dumps(sales, ensure_ascii=False)}

💡 参数名必须与函数签名一致（report_week），code 节点按名传参——Dify 1.16 实测：参数名不一致直接报错。

5.2 数据清洗与异常检测（Code）

核心清洗节点，接收三路聚合结果 + force_refresh 开关：

def main(sales: str, customers: str, support: str, force_refresh: str) -> dict:

    import json

    # 解析 + 空值兜底

    def parse(s):

        try:

            d = json.loads(s or "[]")

            return d if isinstance(d, list) else []

        except Exception:

            return []

    sl, cu, sp = parse(sales), parse(customers), parse(support)

    # 环比计算（上周 vs 上上周）

    def change(data, key):

        if len(data) < 2:

            return 0.0

        cur, prev = data[-1].get(key, 0), data[-2].get(key, 0)

        return round((cur - prev) / prev * 100, 1) if prev else 0.0

    anomaly_count = 0

    if change(sl, "revenue") < 0: anomaly_count += 1

    if change(cu, "active_users") < 0: anomaly_count += 1

    return {

        "revenue_change": str(change(sl, "revenue")),

        "customer_change": str(change(cu, "active_users")),

        "anomaly_count": str(anomaly_count),

        "cleaned": json.dumps({"sales": sl, "customers": cu, "support": sp}, ensure_ascii=False)

    }

5.3 Markdown 周报模板（模板转换）

Jinja2 渲染周报正文（零 Token），空值保护 + 条件渲染：

# 📊 周报 {{ report_week }}

## 一、核心指标总览

{% set rv = revenue_current if revenue_current is not none else 0 %}

{% set rv_prev = revenue_previous if revenue_previous is not none else 0 %}

- 营收：{{ rv }} 元（环比 {{ revenue_change }}%）

{% if anomaly_count | int > 0 %}

## 四、异常预警

共发现 {{ anomaly_count }} 项异常，请关注：

{% for a in anomalies %}

- {{ a }}

{% endfor %}

{% endif %}

💡 模板里所有变量都要有默认值保护（is not none else 0）——否则缺失字段直接渲染失败。

5.4 格式选择 + Webhook 告警（分支 + 通知）

output_format 决定走「结束」（返回报告）还是「Webhook 推送」（推送到群/系统）；告警条件（IF-ELSE）判断 anomaly_count > 0 才发通知——检查类分支必须消费判断结果，否则告警是死代码。

6. 运行验证

（本实验已在 TR4 验收中 4/4 PASS：确定性/内容/语义/边界全过）

7. 实战坑

8. 实验文档及源码获取

• 实验文档（完整操作步骤）：DIFY-21：综合实战（下）——自动化报告生成流水线.md
• 源码（可直接导入）：dify102_21_周报自动生成器.yml

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

系列 1（中级 20 篇）完——从参数提取到全自动流水线，Dify 工作流核心能力全部覆盖。敬请期待系列 2：高级实验（综合场景实战：智能客服、对话式 BI、企业级平台……）。