Granite TimeSeries FlowState R1与工作流引擎n8n集成：自动化预测任务流水线

你是不是也遇到过这样的场景？每天都要手动从数据库里拉取最新的销售数据，然后运行预测模型，再盯着结果看有没有异常，一旦发现问题还得赶紧写邮件通知团队。整个过程繁琐、重复，还容易出错。要是能把这些步骤串起来，让机器自动完成就好了。

今天，我们就来聊聊怎么用一款叫n8n的开源自动化工具，把强大的时间序列预测模型Granite TimeSeries FlowState R1“编织”到你的日常工作流里。你不用写复杂的后端代码，只需要像搭积木一样，在可视化界面上拖拽几个节点，就能搭建一个从数据读取、模型预测到智能告警的完整自动化流水线。比如，让它每天凌晨自动预测服务器负载，一旦发现未来几小时可能超标，就提前发消息给运维同事。这样一来，你就能从重复劳动中解放出来，专注于更有价值的数据分析和策略制定。

1. 为什么需要自动化预测流水线？

在聊具体怎么做之前，我们先看看手动处理预测任务有哪些麻烦。首先，数据源可能分散在各个地方，每天手动提取和整理就是个体力活。其次，调用预测模型的API虽然就几行代码，但你要处理认证、错误重试、结果解析这些琐事。最关键的是，预测本身不是终点，你得根据预测结果做出反应。如果预测到销售额会下滑，你得通知市场部；如果预测到服务器压力会激增，你得提醒运维团队。这个“观察-判断-行动”的闭环如果靠人工，不仅慢，还可能因为疏忽而遗漏重要警报。

n8n这类工作流引擎，正好能解决这些问题。它就像一个可视化的“胶水”，能把不同的应用、服务和API连接起来。你把“从数据库取数”、“调用Granite模型”、“判断结果”、“发通知”这几个动作变成一个个节点，然后用线把它们连起来，定个时，一个自动化流水线就诞生了。整个过程几乎不需要写代码，维护和修改起来也特别直观。

2. 核心组件简介：Granite模型与n8n

在开始搭建之前，我们快速了解一下这次要用到的两个核心工具。

2.1 Granite TimeSeries FlowState R1：你的预测大脑

Granite TimeSeries FlowState R1是一个专门处理时间序列预测的模型。你可以把它理解为一个非常擅长从历史数据中找规律，并预测未来趋势的“大脑”。比如，你给它过去一年的每日销售额数据，它就能预测出接下来一周每天的销售额大概是多少。

它通常通过API的方式提供服务。你只需要按照要求，把历史数据整理成特定的格式（比如JSON），通过一个HTTP请求发送给它，它就会返回包含未来预测值的结果。这对于集成到自动化流程中来说非常友好，因为HTTP调用是标准操作，任何自动化工具都能轻松处理。

2.2 n8n：低代码的流程“编织器”

n8n是一个开源、可自托管的工作流自动化工具。它的最大特点是采用了“节点”（Node）的概念。每个节点代表一个独立的功能，比如“读取MySQL数据库”、“发送HTTP请求”、“判断条件”、“发送邮件”等等。

你的工作就是在n8n的图形化编辑器里，把这些节点拖到画布上，然后用线把它们按照逻辑顺序连接起来。节点之间可以传递数据，上一个节点的输出，可以作为下一个节点的输入。你还可以设置触发器，比如“每隔一小时”或者“每天上午9点”自动运行这个工作流。

对于不擅长编程的业务分析师或运维工程师来说，n8n大大降低了自动化门槛。你不需要知道如何用Python写一个完整的调度程序，只需要关注业务逻辑本身。

3. 实战：构建一个服务器负载预测告警流水线

光说不练假把式。我们假设一个运维中常见的场景：监控服务器集群的CPU使用率，并预测其未来趋势。如果预测到未来一小时内平均CPU使用率会超过85%，就自动发送告警邮件。

我们的自动化流水线将包含以下步骤：

1. 定时触发：每天上班前（比如早上8点）自动运行。
2. 获取历史数据：从监控数据库（如InfluxDB、Prometheus或普通MySQL）中拉取过去24小时的服务器CPU数据。
3. 调用预测模型：将历史数据发送给Granite TimeSeries FlowState R1的API，获取未来一小时的CPU使用率预测值。
4. 判断逻辑：检查预测的平均CPU使用率是否超过我们设定的阈值（85%）。
5. 执行动作：如果超过阈值，就自动发送一封告警邮件给运维团队；如果没超过，流程安静结束。

下面，我们一步步在n8n中实现它。

3.1 第一步：创建并触发工作流

首先，登录你的n8n实例，点击“创建工作流”。你会看到一个空白的画布。

我们需要一个起点，也就是触发器。在左侧节点面板搜索“Schedule Trigger”节点，把它拖到画布上。这个节点可以让工作流按计划执行。

双击它进行配置。在“Rule”栏，你可以设置类似Cron表达式的规则。比如，设置为“0 8 * * 1-5”就表示在每个工作日的早上8点运行。你也可以简单地选择“Every Day”（每天）、“Every Hour”（每小时）等预设选项。

配置好后，这个节点就会在你设定的时间点，自动触发整个工作流的运行。

3.2 第二步：从数据库获取历史监控数据

触发器节点之后，我们添加一个节点来获取数据。这里假设你的监控数据存在一个MySQL数据库里。

搜索并添加“MySQL”节点。首次使用需要配置数据库连接：填写主机地址、数据库名、用户名和密码。n8n会安全地保存这些凭证。

配置节点的操作，选择“Execute Query”。在SQL查询语句中，你可以写入类似下面的代码：

SELECT 
    `timestamp`,
    `cpu_usage_percent`
FROM `server_metrics`
WHERE `server_id` = 'web-01'
  AND `timestamp` >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
ORDER BY `timestamp` ASC;

这个查询会获取服务器web-01过去24小时内，每个时间点的CPU使用率。查询结果（一个包含timestamp和cpu_usage_percent的数据列表）会作为这个节点的输出，传递给下一个节点。

3.3 第三步：准备数据并调用Granite预测API

现在，我们拿到了历史数据，需要把它们转换成Granite模型API要求的格式，然后发起调用。

首先，你可能需要一个“Function”节点或“Code”节点来转换数据格式。假设Granite API要求一个JSON格式的请求体，像这样：{"series": [[timestamp1, value1], [timestamp2, value2], ...]}。

添加一个“Function”节点，连接到MySQL节点之后。在这个节点里，你可以写一段简单的JavaScript代码来处理上游传来的数据：

// 假设items是上游MySQL节点返回的数据数组
const historicalData = items[0].json.data; // 根据你的实际数据结构调整

// 将数据转换为模型需要的二维数组格式 [ [时间戳, 数值], ... ]
const formattedSeries = historicalData.map(row => {
  // 将时间转换为Unix时间戳（毫秒或秒，根据API要求）
  const timestamp = new Date(row.timestamp).getTime() / 1000;
  return [timestamp, row.cpu_usage_percent];
});

// 构建最终的请求体
const requestBody = {
  series: formattedSeries,
  forecast_horizon: 12 // 预测未来12个时间点（假设每5分钟一个点，即未来1小时）
};

// 将新数据赋值给输出项
return [{json: {graniteRequest: requestBody}}];

数据准备好后，添加一个“HTTP Request”节点。将其方法设置为“POST”，并填入你的Granite TimeSeries FlowState R1模型的API端点URL。

在“Headers”选项卡，添加必要的认证头，比如Authorization: Bearer YOUR_API_KEY。在“Body”选项卡，选择“JSON”，然后输入表达式{{ $json.graniteRequest }}，这样就会使用上一个Function节点准备好的数据作为请求体。

这个节点执行后，会收到模型的响应，其中就包含了未来一段时间的预测值。

3.4 第四步：解析结果并判断是否告警

HTTP Request节点返回的预测结果，我们需要从中提取出未来一小时的预测值，并计算平均值。

再添加一个“Function”节点来处理预测结果：

const predictionResponse = items[0].json; // 假设API返回的数据在json字段
// 这里需要根据Granite API的实际返回结构来解析
// 例如，假设预测值在 predictionResponse.forecast 数组里
const forecastValues = predictionResponse.forecast;

// 计算未来一小时预测值的平均值
const sum = forecastValues.reduce((a, b) => a + b, 0);
const averagePrediction = sum / forecastValues.length;

// 将平均值和原始数据一起传递下去
return [{
  json: {
    ...items[0].json, // 保留原有数据
    averageCpuPrediction: averagePrediction,
    forecastValues: forecastValues
  }
}];

现在，我们得到了预测的平均值averageCpuPrediction。接下来添加一个“IF”节点来做判断。

配置IF节点：设置条件为“averageCpuPrediction”“大于”“85”。这样，节点就会根据条件产生两条分支：一条为“真”（预测值>85），一条为“假”。

3.5 第五步：发送邮件告警

将IF节点的“真”分支，连接到一个“Email”节点（可能是“SendGrid”、“Gmail”或“SMTP”节点，取决于你配置的邮件服务）。

配置这个邮件节点：

• 收件人：填入运维团队的邮箱地址，如 ops-team@yourcompany.com。
• 主题：可以设置为动态内容，如 【服务器预警】Web-01 CPU预测值超标 - {{$now.format('YYYY-MM-DD HH:mm')}}。
• 正文：在正文中，你可以引用工作流中的数据，让告警信息更丰富。例如：

  预警！
  服务器：web-01
  预测时间窗口：未来一小时
  平均预测CPU使用率：{{$json.averageCpuPrediction.toFixed(2)}}%
  阈值：85%
  预测详细序列：{{$json.forecastValues}}
  请及时关注并处理。
  

这样，当预测值超过阈值时，一封包含详细信息的告警邮件就会自动发出。IF节点的“假”分支可以什么都不连，或者连接一个“No Operation”节点，表示无需任何操作。

至此，一个完整的自动化预测告警流水线就搭建完成了。你的画布上应该有类似这样的节点链条：Schedule Trigger -> MySQL -> Function（数据转换）-> HTTP Request -> Function（结果解析）-> IF -> Email。

4. 更多应用场景与进阶思路

上面这个例子只是一个开始。将Granite模型与n8n结合，能玩出很多花样。

• 电商库存预测补货：定时预测未来一周各SKU的销量，当预测库存低于安全线时，自动在ERP或采购系统中创建采购订单。
• 智能营销：预测下个月各渠道的流量和转化率，自动调整不同广告平台的每日预算分配。
• 能源管理：预测办公楼下一小时的用电负荷，自动调节空调系统的设定温度，实现节能。
• 批量预测与报告：每周一自动拉取上周所有门店的销售数据，批量调用模型预测下周业绩，然后将结果自动填入PPT模板，生成周度预测报告并邮件发送给管理层。

在进阶使用中，你还可以考虑：

• 错误处理：在HTTP Request节点后添加“Error Trigger”节点，当模型API调用失败时，发送通知给开发人员。
• 数据持久化：将预测结果写回数据库，便于后续分析和模型效果回溯。
• 复杂决策流：结合多个判断条件（比如预测值超过阈值且呈上升趋势），触发更复杂的联动操作。

5. 总结

通过n8n将Granite TimeSeries FlowState R1集成到工作流中，你实际上是在用可视化的方式，为预测模型装上了“眼睛”和“手脚”。眼睛负责定时抓取最新的数据，模型作为大脑进行分析思考，而手脚则负责根据思考结果执行具体的操作。这种模式极大地降低了时间序列预测技术落地的门槛，让业务人员也能直接参与构建智能化的决策响应系统。

整个过程就像搭乐高，核心是理解你的业务逻辑——数据从哪来，模型怎么用，结果怎么判断，判断后做什么。把这些环节想清楚，剩下的就是在n8n里找到对应的节点，把它们连起来。下次再遇到重复性的数据预测和监控任务时，不妨试试这个方法，给自己省出更多喝咖啡和思考战略的时间。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。