n8n + AI 自动化工作流完全教程

从零开始掌握 n8n 与 AI 深度集成，构建企业级智能自动化工作流

📋 目录

课程概述与学习路线
n8n 架构深度解析
n8n 安装与环境配置
n8n 核心概念与基础操作
AI 节点集成入门
LangChain 与 n8n 深度集成
LlamaIndex 连接与 RAG 工作流
AI Agent 自动化构建
Webhook 触发与外部集成
定时任务与调度策略
数据转换与处理管道
错误处理与监控体系
企业级部署方案
实战项目：自动化内容生成与分发系统
常见问题与解决方案
总结与进阶资源

1. 课程概述与学习路线

1.1 什么是 n8n？

n8n（发音为 "n-eight-n"）是一款开源的工作流自动化平台，它允许用户通过可视化拖拽界面连接各种应用和服务，实现任务自动化。与 Zapier、Make 等商业平台不同，n8n 完全开源，支持自托管部署，数据完全掌控在自己手中。

n8n 的核心优势在于：

开源免费：社区版完全免费，企业版提供高级功能
自托管：数据不出服务器，满足企业合规需求
可扩展：支持自定义节点开发，集成任意 API
AI 原生：内置丰富的 AI 节点，与 LangChain 深度集成
代码灵活：支持 JavaScript/Python 代码节点，满足复杂逻辑

1.2 为什么选择 n8n + AI？

在 AI 时代，单纯调用 API 已经不够。企业需要的是将 AI 能力嵌入到完整的业务流程中：

自动收集数据 → AI 分析 → 生成报告 → 发送到团队
客户消息 → AI 理解意图 → 自动回复或转人工
定时爬取 → AI 总结 → 发布到多平台

n8n 正是连接 AI 与业务流程的最佳桥梁。

1.3 学习路线图

基础阶段（第1-4章）
├── 理解 n8n 架构
├── 完成安装部署
├── 掌握核心概念
└── 学会基础节点操作

进阶阶段（第5-8章）
├── 集成 OpenAI / Claude 等 AI 服务
├── 掌握 LangChain 集成
├── 构建 RAG 检索增强工作流
└── 开发 AI Agent 自动化

高级阶段（第9-13章）
├── Webhook 与外部系统集成
├── 定时调度与数据管道
├── 错误处理与监控
└── 企业级部署与运维

实战阶段（第14章）
└── 完整项目：自动化内容生成与分发系统

2. n8n 架构深度解析

2.1 整体架构

n8n 采用前后端分离的架构设计：

┌─────────────────────────────────────────────┐
│                 n8n 架构概览                   │
├─────────────────────────────────────────────┤
│                                             │
│  ┌──────────┐    ┌──────────────────────┐   │
│  │  前端 UI  │◄──►│     后端服务 (Node.js) │   │
│  │ (Vue.js)  │    │                      │   │
│  └──────────┘    │  ┌─────────────────┐  │   │
│                  │  │  工作流引擎       │  │   │
│                  │  │  (Execution Engine)│  │   │
│                  │  └────────┬────────┘  │   │
│                  │           │           │   │
│                  │  ┌────────▼────────┐  │   │
│                  │  │   节点执行器      │  │   │
│                  │  │ (Node Executor)  │  │   │
│                  │  └────────┬────────┘  │   │
│                  │           │           │   │
│                  │  ┌────────▼────────┐  │   │
│                  │  │   数据库 (SQLite  │  │   │
│                  │  │   / PostgreSQL)  │  │   │
│                  │  └─────────────────┘  │   │
│                  └──────────────────────┘   │
└─────────────────────────────────────────────┘

2.2 核心组件

工作流引擎是 n8n 的心脏，负责：

解析工作流 JSON 定义
管理节点执行顺序
处理条件分支和循环
管理执行上下文和数据流

节点系统是 n8n 的扩展点：

每个节点是一个独立的功能单元
节点之间通过 JSON 数据传递信息
支持触发器节点（Trigger）和操作节点（Action）

2.3 数据流模型

n8n 中数据以 Item 为基本单位流动：

// 每个 Item 都是一个 JSON 对象
{
  "json": {
    "name": "张三",
    "email": "zhangsan@example.com",
    "message": "我想了解AI自动化方案"
  },
  "binary": {} // 可选的二进制数据（文件、图片等）
}

多个 Item 组成一个 Items 数组，在节点之间传递：

[
  { "json": { "name": "张三", "action": "咨询" } },
  { "json": { "name": "李四", "action": "投诉" } },
  { "json": { "name": "王五", "action": "建议" } }
]

3. n8n 安装与环境配置

3.1 系统要求

组件	最低要求	推荐配置
CPU	1 核	2 核+
内存	2 GB	4 GB+
磁盘	10 GB	50 GB+
Node.js	v18+	v20 LTS
Docker	20.10+	最新稳定版

3.2 Docker 快速部署（推荐）

Docker 是最简单、最推荐的部署方式。创建 docker-compose.yml 文件：

version: '3.8'

services:
  n8n:
    image: docker.n8n.io/n8nio/n8n:latest
    restart: always
    ports:
      - "5678:5678"
    environment:
      # 基础配置
      - N8N_HOST=your-domain.com
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://your-domain.com/
      
      # 数据库配置（使用 PostgreSQL）
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=your_secure_password
      
      # AI 相关配置
      - OPENAI_API_KEY=sk-your-openai-key
      - ANTHROPIC_API_KEY=sk-ant-your-anthropic-key
      
      # 安全配置
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=your_admin_password
      
      # 执行配置
      - EXECUTIONS_MODE=queue
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168
      
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres

  postgres:
    image: postgres:15-alpine
    restart: always
    environment:
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=your_secure_password
      - POSTGRES_DB=n8n
    volumes:
      - postgres_data:/var/lib/postgresql/data

  # Redis（用于队列模式）
  redis:
    image: redis:7-alpine
    restart: always
    volumes:
      - redis_data:/data

volumes:
  n8n_data:
  postgres_data:
  redis_data:

启动服务：

# 启动所有服务
docker compose up -d

# 查看日志
docker compose logs -f n8n

# 停止服务
docker compose down

3.3 npm 本地安装

适合开发和测试环境：

# 全局安装 n8n
npm install n8n -g

# 启动 n8n
n8n start

# 或者使用 npx 直接运行（无需全局安装）
npx n8n

3.4 源码编译安装

适合需要自定义开发的场景：

# 克隆仓库
git clone https://github.com/n8n-io/n8n.git
cd n8n

# 安装依赖
npm install

# 构建项目
npm run build

# 启动开发模式
npm run dev

3.5 环境变量配置详解

创建 .env 文件管理敏感配置：

# .env 文件
# ==================== 基础配置 ====================
N8N_HOST=0.0.0.0
N8N_PORT=5678
N8N_PROTOCOL=http
N8N_ENCRYPTION_KEY=your-random-32-char-encryption-key

# ==================== AI 服务密钥 ====================
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxx
GOOGLE_AI_KEY=AIzaxxxxxxxxxxxxxxxxxxx
HUGGINGFACE_API_KEY=hf_xxxxxxxxxxxxxxxxxxx

# ==================== 数据库配置 ====================
DB_TYPE=postgresdb
DB_POSTGRESDB_HOST=localhost
DB_POSTGRESDB_PORT=5432
DB_POSTGRESDB_DATABASE=n8n
DB_POSTGRESDB_USER=n8n
DB_POSTGRESDB_PASSWORD=secure_password

# ==================== 邮件配置 ====================
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@gmail.com
SMTP_PASS=your-app-password

# ==================== 存储配置 ====================
N8N_DEFAULT_BINARY_DATA_MODE=filesystem
N8N_BINARY_DATA_STORAGE_PATH=/data/n8n/binary

4. n8n 核心概念与基础操作

4.1 工作流（Workflow）

工作流是 n8n 的核心单元，由节点和连接组成。每个工作流保存为 JSON 格式：

{
  "name": "我的第一个工作流",
  "nodes": [
    {
      "parameters": {},
      "id": "trigger-1",
      "name": "Start",
      "type": "n8n-nodes-base.start",
      "typeVersion": 1,
      "position": [250, 300]
    }
  ],
  "connections": {},
  "settings": {
    "executionOrder": "v1"
  }
}

4.2 节点类型

n8n 的节点分为三大类：

触发器节点（Trigger Nodes）：工作流的入口

Manual Trigger：手动触发
Schedule Trigger：定时触发
Webhook Trigger：HTTP 请求触发
Email Trigger：邮件触发
Chat Trigger：聊天消息触发

操作节点（Action Nodes）：执行具体操作

HTTP Request：发送 HTTP 请求
Code：执行 JavaScript/Python 代码
Set：设置/修改数据
IF：条件判断
Switch：多路分支

AI 专用节点：

OpenAI：调用 OpenAI API
AI Agent：智能代理节点
Vector Store：向量数据库操作
Document Loader：文档加载器
Text Splitter：文本分割器

4.3 表达式语法

n8n 使用表达式引用节点输出数据：

// 引用上一个节点的第一条数据的某个字段
{{ $json.fieldName }}

// 引用特定节点的输出
{{ $('节点名称').item.json.fieldName }}

// 引用上一个节点的所有数据
{{ $input.all() }}

// 使用 JavaScript 表达式
{{ $json.name.toUpperCase() }}

// 条件表达式
{{ $json.score > 80 ? '优秀' : '一般' }}

// 数组操作
{{ $json.items.map(item => item.name).join(', ') }}

// 日期格式化
{{ new Date($json.timestamp).toLocaleDateString('zh-CN') }}

// JSON 解析
{{ JSON.parse($json.rawData) }}

4.4 第一个工作流：AI 文本处理

让我们创建一个简单的工作流，接收用户输入并用 AI 处理：

步骤 1：添加 Manual Trigger 节点

步骤 2：添加 Set 节点，设置输入数据

{
  "text": "n8n 是一款强大的工作流自动化工具，结合 AI 可以实现很多有趣的场景。"
}

步骤 3：添加 OpenAI 节点，配置如下：

Resource: Chat
Operation: Complete
Model: gpt-4o
Messages:
- System: 你是一个专业的文本分析助手，请对用户提供的文本进行分析，给出情感倾向和关键词。
- User: {{ $json.text }}

步骤 4：添加 Set 节点，整理输出

{
  "original": "{{ $('Set').item.json.text }}",
  "analysis": "{{ $json.message.content }}",
  "model": "{{ $json.model }}",
  "timestamp": "{{ new Date().toISOString() }}"
}

5. AI 节点集成入门

5.1 OpenAI 集成

n8n 内置了 OpenAI 节点，支持多种功能：

Chat Completions（对话补全）：

// 在 Code 节点中直接调用 OpenAI API
const response = await this.helpers.httpRequest({
  method: 'POST',
  url: 'https://api.openai.com/v1/chat/completions',
  headers: {
    'Authorization': `Bearer ${$env.OPENAI_API_KEY}`,
    'Content-Type': 'application/json'
  },
  body: {
    model: 'gpt-4o',
    messages: [
      { role: 'system', content: '你是一个专业的翻译助手。' },
      { role: 'user', content: `请将以下文本翻译成英文：${$json.text}` }
    ],
    temperature: 0.3,
    max_tokens: 1000
  }
});

return [{ json: { translation: response.choices[0].message.content } }];

使用内置 OpenAI 节点：

配置参数：

Resource: Chat
Operation: Message a Model
Model: gpt-4o
Messages:
  - Role: System
    Content: 你是一个专业的文本分析助手。
  - Role: User  
    Content: {{ $json.inputText }}
Temperature: 0.7
Max Tokens: 2000

5.2 Anthropic Claude 集成

通过 HTTP Request 节点调用 Claude API：

// Code 节点中调用 Claude
const response = await this.helpers.httpRequest({
  method: 'POST',
  url: 'https://api.anthropic.com/v1/messages',
  headers: {
    'x-api-key': $env.ANTHROPIC_API_KEY,
    'anthropic-version': '2023-06-01',
    'content-type': 'application/json'
  },
  body: {
    model: 'claude-sonnet-4-20250514',
    max_tokens: 2000,
    messages: [{
      role: 'user',
      content: `请分析以下内容并给出建议：\n\n${$json.content}`
    }]
  }
});

return [{
  json: {
    analysis: response.content[0].text,
    model: response.model,
    usage: response.usage
  }
}];

5.3 本地模型集成（Ollama）

对于注重数据隐私的场景，可以使用 Ollama 运行本地模型：

// 调用本地 Ollama 服务
const response = await this.helpers.httpRequest({
  method: 'POST',
  url: 'http://localhost:11434/api/generate',
  headers: { 'Content-Type': 'application/json' },
  body: {
    model: 'qwen2.5:14b',
    prompt: `请对以下文本进行摘要：\n\n${$json.text}`,
    stream: false,
    options: {
      temperature: 0.5,
      num_predict: 500
    }
  }
});

return [{ json: { summary: response.response } }];

5.4 多模型路由策略

在实际项目中，常常需要根据任务类型选择不同的模型：

// Code 节点：智能模型路由
const task = $json.taskType;
const text = $json.text;

let model, systemPrompt;

switch (task) {
  case 'translate':
    model = 'gpt-4o-mini';
    systemPrompt = '你是一个专业的翻译助手，请准确翻译用户提供的文本。';
    break;
  case 'summarize':
    model = 'gpt-4o';
    systemPrompt = '你是一个文本摘要专家，请用简洁的语言总结核心内容。';
    break;
  case 'analyze':
    model = 'claude-sonnet-4-20250514';
    systemPrompt = '你是一个数据分析专家，请深入分析用户提供的数据。';
    break;
  case 'code':
    model = 'gpt-4o';
    systemPrompt = '你是一个高级程序员，请根据需求编写高质量代码。';
    break;
  default:
    model = 'gpt-4o-mini';
    systemPrompt = '你是一个通用AI助手。';
}

return [{
  json: {
    model,
    systemPrompt,
    userContent: text,
    task
  }
}];

6. LangChain 与 n8n 深度集成

6.1 LangChain 基础概念

LangChain 是一个用于构建 LLM 应用的框架，n8n 从 v1.0 开始内置了 LangChain 节点，使得在工作流中使用 LangChain 的各种组件变得非常简单。

LangChain 的核心组件：

Models：语言模型封装
Prompts：提示词模板管理
Chains：将多个组件串联
Memory：对话记忆管理
Retrievers：文档检索器
Agents：智能代理
Tools：工具集

6.2 n8n 中的 LangChain 节点

n8n 提供了专门的 LangChain 节点类别：

LangChain 节点
├── Agents
│   ├── AI Agent
│   ├── Plan and Execute Agent
│   └── Conversational Agent
├── Chains
│   ├── LLM Chain
│   ├── Retrieval QA Chain
│   ├── Conversational Retrieval QA Chain
│   └── Stuff Documents Chain
├── Memory
│   ├── Buffer Memory
│   ├── Window Buffer Memory
│   ├── Motorhead Memory
│   └── Postgres Chat Memory
├── Document Loaders
│   ├── Default Document Loader
│   ├── PDF Loader
│   └── Text File Loader
├── Vector Stores
│   ├── Pinecone
│   ├── Qdrant
│   ├── Supabase
│   └── In-Memory Vector Store
├── Embeddings
│   ├── OpenAI Embeddings
│   ├── HuggingFace Embeddings
│   └── Cohere Embeddings
└── Tools
    ├── Calculator
    ├── Code Execution Tool
    ├── HTTP Request Tool
    └── Workflow Tool

6.3 构建第一个 LangChain 工作流

创建一个带有记忆的对话链：

工作流结构：
Chat Trigger → AI Agent → OpenAI Chat Model + Buffer Memory + Calculator Tool

步骤 1：添加 Chat Trigger 节点

配置为响应用户聊天消息

步骤 2：添加 AI Agent 节点

Agent Type: Conversational Agent
System Message: 你是一个专业的AI助手，能够回答各种问题并使用工具完成任务。
Max Iterations: 5

步骤 3：连接 OpenAI Chat Model 子节点

Model: gpt-4o
Temperature: 0.7

步骤 4：连接 Buffer Memory 子节点

Context Window Length: 20

步骤 5：连接 Calculator Tool 子节点

6.4 自定义 LangChain Chain

在 Code 节点中使用 LangChain JS 构建自定义链：

// 注意：需要在 n8n 环境中安装 langchain 包
// 或者使用 n8n 内置的 LangChain 节点组合

// 如果使用 n8n 的 Execute Command 节点预装依赖：
// npm install langchain @langchain/openai

// 在自定义 n8n 社区节点中使用 LangChain：
const { ChatOpenAI } = require('@langchain/openai');
const { PromptTemplate } = require('@langchain/core/prompts');
const { StringOutputParser } = require('@langchain/core/output_parsers');

const model = new ChatOpenAI({
  modelName: 'gpt-4o',
  temperature: 0.7,
  openAIApiKey: $env.OPENAI_API_KEY
});

const prompt = PromptTemplate.fromTemplate(
  `你是一个专业的{role}。请根据以下要求完成任务：
  
  任务：{task}
  输入内容：{input}
  
  请给出详细、专业的回答。`
);

const chain = prompt.pipe(model).pipe(new StringOutputParser());

const result = await chain.invoke({
  role: $json.role || 'AI助手',
  task: $json.task || '分析文本',
  input: $json.input
});

return [{ json: { result } }];

6.5 RAG（检索增强生成）基础

RAG 是将外部知识注入 LLM 的核心技术。n8n 中实现 RAG 的工作流：

文档摄入流程：
Document Loader → Text Splitter → Embeddings → Vector Store

查询流程：
User Query → Embeddings → Vector Store (Retriever) → LLM → Response

7. LlamaIndex 连接与 RAG 工作流

7.1 LlamaIndex 简介

LlamaIndex 是另一个强大的 LLM 应用框架，专注于数据连接和检索。与 LangChain 相比，LlamaIndex 在文档索引和查询方面更加专业。

7.2 使用 n8n 构建完整 RAG 工作流

文档摄入工作流：

工作流结构：
Manual Trigger → Read Binary File → Text Splitter → Embeddings → Vector Store

详细配置：

Text Splitter 节点：

Splitter Type: Recursive Character Text Splitter
Chunk Size: 1000
Chunk Overlap: 200

Embeddings 节点：

Embeddings Type: OpenAI Embeddings
Model: text-embedding-3-small
Dimensions: 1536

Vector Store 节点（以 Qdrant 为例）：

Vector Store: Qdrant
Collection Name: my_documents
Qdrant URL: http://localhost:6333

查询工作流：

// Code 节点：构建 RAG 查询
const { OpenAIEmbeddings } = require('@langchain/openai');
const { QdrantClient } = require('@qdrant/js-client-rest');
const { OpenAI } = require('openai');

// 初始化客户端
const embeddings = new OpenAIEmbeddings({
  modelName: 'text-embedding-3-small',
  openAIApiKey: $env.OPENAI_API_KEY
});

const qdrant = new QdrantClient({ url: 'http://localhost:6333' });
const openai = new OpenAI({ apiKey: $env.OPENAI_API_KEY });

// 用户查询
const query = $json.question;

// 1. 将查询转换为向量
const queryEmbedding = await embeddings.embedQuery(query);

// 2. 从向量数据库中检索相关文档
const searchResult = await qdrant.search('my_documents', {
  vector: queryEmbedding,
  limit: 5,
  score_threshold: 0.7
});

// 3. 构建上下文
const context = searchResult
  .map((result, idx) => `[文档${idx + 1}] ${result.payload.content}`)
  .join('\n\n');

// 4. 调用 LLM 生成回答
const completion = await openai.chat.completions.create({
  model: 'gpt-4o',
  messages: [
    {
      role: 'system',
      content: `你是一个专业的知识库助手。请根据以下参考文档回答用户问题。
      如果文档中没有相关信息，请如实说明。
      
      参考文档：
      ${context}`
    },
    {
      role: 'user',
      content: query
    }
  ],
  temperature: 0.3
});

return [{
  json: {
    answer: completion.choices[0].message.content,
    sources: searchResult.map(r => ({
      content: r.payload.content.substring(0, 200) + '...',
      score: r.score,
      source: r.payload.source
    }))
  }
}];

7.3 高级 RAG 技术

混合检索（Hybrid Search）：结合关键词搜索和语义搜索

// 混合检索实现
async function hybridSearch(query, options = {}) {
  const { alpha = 0.5, limit = 5 } = options;
  
  // 语义搜索
  const queryEmbedding = await embeddings.embedQuery(query);
  const semanticResults = await qdrant.search('documents', {
    vector: queryEmbedding,
    limit: limit * 2
  });
  
  // 关键词搜索（使用 Qdrant 的全文搜索）
  const keywordResults = await qdrant.search('documents', {
    vector: queryEmbedding, // 使用相同向量作为基础
    filter: {
      must: [{
        key: 'content',
        match: { text: query }
      }]
    },
    limit: limit * 2
  });
  
  // 合并和重排序
  const merged = new Map();
  
  semanticResults.forEach((r, idx) => {
    merged.set(r.id, {
      ...r,
      score: r.score * alpha + (1 - idx / semanticResults.length) * (1 - alpha)
    });
  });
  
  keywordResults.forEach((r, idx) => {
    const existing = merged.get(r.id);
    if (existing) {
      existing.score += r.score * (1 - alpha) * 0.5;
    } else {
      merged.set(r.id, {
        ...r,
        score: r.score * (1 - alpha)
      });
    }
  });
  
  return Array.from(merged.values())
    .sort((a, b) => b.score - a.score)
    .slice(0, limit);
}

查询重写（Query Rewriting）：优化用户查询以提高检索效果

// 查询重写
async function rewriteQuery(originalQuery) {
  const completion = await openai.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [
      {
        role: 'system',
        content: `你是一个查询优化专家。请将用户的查询改写为更适合搜索引擎的形式。
        要求：
        1. 保持原始意图
        2. 添加相关同义词
        3. 去除无关词汇
        4. 生成3个不同角度的查询变体
        
        返回JSON格式：{"queries": ["查询1", "查询2", "查询3"]}`
      },
      { role: 'user', content: originalQuery }
    ],
    response_format: { type: 'json_object' }
  });
  
  return JSON.parse(completion.choices[0].message.content).queries;
}

8. AI Agent 自动化构建

8.1 什么是 AI Agent？

AI Agent 是能够自主规划、决策和执行任务的智能系统。与简单的 Chain 不同，Agent 可以：

自主决定使用哪些工具
根据中间结果调整策略
处理多步骤复杂任务
与外部环境交互

8.2 n8n 中的 Agent 类型

n8n 支持多种 Agent 类型：

1. ReAct Agent：推理 + 行动

Agent Type: AI Agent
Agent Type (具体): Tools Agent

2. Plan and Execute Agent：先规划后执行

Agent Type: Plan and Execute Agent

3. Conversational Agent：对话式代理

Agent Type: Conversational Agent

8.3 构建功能丰富的 Agent

创建一个能够搜索网络、执行计算、处理文件的 Agent：

工作流结构：
Chat Trigger 
    → AI Agent
        ├── OpenAI Chat Model (gpt-4o)
        ├── Buffer Memory
        ├── HTTP Request Tool (搜索API)
        ├── Calculator Tool
        ├── Code Execution Tool
        └── Workflow Tool (调用其他工作流)

HTTP Request Tool 配置（用于网络搜索）：

Name: search_web
Description: 搜索互联网获取最新信息。输入应该是搜索关键词。
URL: https://api.search.com/search?q={{ $fromAI('query') }}
Method: GET

Code Execution Tool 配置：

Name: execute_code
Description: 执行JavaScript代码进行数据处理或计算。输入应该是完整的JavaScript代码。
Language: JavaScript

8.4 自定义 Agent Tool

通过 n8n 的 Workflow Tool 节点，可以将其他工作流作为 Agent 的工具：

// 被调用的子工作流：数据分析工具
// 输入：{ data: [...], analysisType: "summary" | "trend" | "anomaly" }

const data = $json.data;
const analysisType = $json.analysisType;

let result;

switch (analysisType) {
  case 'summary':
    result = {
      count: data.length,
      average: data.reduce((sum, d) => sum + d.value, 0) / data.length,
      min: Math.min(...data.map(d => d.value)),
      max: Math.max(...data.map(d => d.value))
    };
    break;
    
  case 'trend':
    // 简单线性回归
    const n = data.length;
    const xMean = (n - 1) / 2;
    const yMean = data.reduce((sum, d) => sum + d.value, 0) / n;
    
    let numerator = 0, denominator = 0;
    data.forEach((d, i) => {
      numerator += (i - xMean) * (d.value - yMean);
      denominator += (i - xMean) ** 2;
    });
    
    const slope = numerator / denominator;
    const intercept = yMean - slope * xMean;
    
    result = {
      trend: slope > 0 ? '上升' : '下降',
      slope,
      intercept,
      prediction: slope * n + intercept
    };
    break;
    
  case 'anomaly':
    const values = data.map(d => d.value);
    const mean = values.reduce((a, b) => a + b) / values.length;
    const std = Math.sqrt(values.reduce((sum, v) => sum + (v - mean) ** 2, 0) / values.length);
    
    const anomalies = data.filter(d => Math.abs(d.value - mean) > 2 * std);
    result = {
      anomalies,
      threshold: 2 * std,
      mean,
      std
    };
    break;
}

return [{ json: { analysis: result, type: analysisType } }];

8.5 Agent 最佳实践

明确系统提示词：

你是一个专业的数据分析助手。你可以：
- 使用搜索工具查找最新数据
- 使用计算工具进行复杂运算
- 使用代码工具处理数据集
- 使用分析工具生成报告

请遵循以下原则：
1. 先理解用户需求，再选择工具
2. 复杂任务分步骤执行
3. 对结果进行验证
4. 用清晰的格式呈现结果

限制迭代次数：防止 Agent 陷入无限循环

Max Iterations: 10

设置超时：

Timeout: 300000  // 5分钟

9. Webhook 触发与外部集成

9.1 Webhook 基础

Webhook 是 n8n 与外部系统集成的核心方式。当外部事件发生时，系统会向 n8n 发送 HTTP 请求，触发工作流执行。

创建 Webhook 触发器：

节点类型：Webhook
HTTP Method: POST
Path: /my-webhook
Authentication: Header Auth
Response Mode: Last Node

Webhook URL 格式：

https://your-n8n-domain.com/webhook/my-webhook
https://your-n8n-domain.com/webhook-test/my-webhook  // 测试模式

9.2 处理 Webhook 数据

// Code 节点：处理 Webhook 请求
const webhookData = $json.body;
const headers = $json.headers;
const query = $json.query;

// 验证签名（以 GitHub Webhook 为例）
const crypto = require('crypto');
const signature = headers['x-hub-signature-256'];
const secret = $env.GITHUB_WEBHOOK_SECRET;

const expectedSignature = 'sha256=' + crypto
  .createHmac('sha256', secret)
  .update(JSON.stringify(webhookData))
  .digest('hex');

if (signature !== expectedSignature) {
  throw new Error('Invalid webhook signature');
}

// 处理不同事件类型
const event = headers['x-github-event'];
let result;

switch (event) {
  case 'push':
    result = {
      type: 'push',
      repo: webhookData.repository.full_name,
      branch: webhookData.ref.replace('refs/heads/', ''),
      commits: webhookData.commits.map(c => ({
        message: c.message,
        author: c.author.name,
        url: c.url
      }))
    };
    break;
    
  case 'pull_request':
    result = {
      type: 'pull_request',
      action: webhookData.action,
      title: webhookData.pull_request.title,
      author: webhookData.pull_request.user.login,
      url: webhookData.pull_request.html_url
    };
    break;
    
  case 'issues':
    result = {
      type: 'issue',
      action: webhookData.action,
      title: webhookData.issue.title,
      body: webhookData.issue.body,
      author: webhookData.issue.user.login
    };
    break;
    
  default:
    result = { type: event, data: webhookData };
}

return [{ json: result }];

9.3 响应 Webhook 请求

n8n 支持自定义 Webhook 响应：

// Code 节点：返回自定义响应
// 设置响应头
$execution.customData.set('responseHeaders', {
  'Content-Type': 'application/json',
  'X-Custom-Header': 'n8n-webhook'
});

// 返回响应数据
return [{
  json: {
    status: 'success',
    message: 'Webhook 处理完成',
    data: {
      processedAt: new Date().toISOString(),
      itemsProcessed: $input.all().length
    }
  }
}];

9.4 常见 Webhook 集成场景

Slack 消息触发：

Webhook Path: /slack/events
验证 Token: 配置 Slack 的 Verification Token
处理事件类型: message, app_mention, reaction_added

Stripe 支付回调：

Webhook Path: /stripe/webhook
验证签名: 使用 Stripe 的 Webhook Secret
处理事件类型: payment_intent.succeeded, customer.subscription.created

自定义表单提交：

Webhook Path: /form/submit
Method: POST
Content-Type: application/json
数据处理: 验证 → 清洗 → AI 分析 → 存储 → 通知

10. 定时任务与调度策略

10.1 Schedule Trigger 详解

n8n 的 Schedule Trigger 支持多种调度模式：

间隔触发：

Trigger Interval: Every 5 Minutes

Cron 表达式：

// 每天早上 9 点
Cron Expression: 0 9 * * *

// 每周一上午 10 点
Cron Expression: 0 10 * * 1

// 每月 1 号凌晨 0 点
Cron Expression: 0 0 1 * *

// 工作日每小时
Cron Expression: 0 * * * 1-5

自定义 Cron 配置：

{
  "rule": {
    "interval": [
      {
        "field": "cronExpression",
        "expression": "0 */2 * * *"
      }
    ]
  }
}

10.2 复杂调度场景

错峰执行：避免多个任务同时运行

// Code 节点：添加随机延迟
const delayMinutes = Math.floor(Math.random() * 10);
await new Promise(resolve => setTimeout(resolve, delayMinutes * 60 * 1000));

return [{ json: { delayed: delayMinutes, executedAt: new Date().toISOString() } }];

条件调度：仅在满足条件时执行

// Code 节点：检查是否应该执行
const now = new Date();
const hour = now.getHours();
const dayOfWeek = now.getDay();

// 仅在工作日的 9-18 点执行
if (dayOfWeek === 0 || dayOfWeek === 6) {
  return [{ json: { skip: true, reason: '周末不执行' } }];
}

if (hour < 9 || hour >= 18) {
  return [{ json: { skip: true, reason: '非工作时间' } }];
}

// 检查是否有待处理任务
const pendingTasks = await this.helpers.httpRequest({
  url: 'https://api.example.com/tasks/pending',
  headers: { 'Authorization': `Bearer ${$env.API_TOKEN}` }
});

if (pendingTasks.count === 0) {
  return [{ json: { skip: true, reason: '无待处理任务' } }];
}

return [{ json: { skip: false, taskCount: pendingTasks.count } }];

10.3 任务队列管理

// Code 节点：实现简单的任务队列
const tasks = $input.all();
const MAX_CONCURRENT = 3;
const results = [];

// 分批处理
for (let i = 0; i < tasks.length; i += MAX_CONCURRENT) {
  const batch = tasks.slice(i, i + MAX_CONCURRENT);
  
  const batchResults = await Promise.allSettled(
    batch.map(async (task) => {
      // 模拟任务处理
      const response = await this.helpers.httpRequest({
        method: 'POST',
        url: 'https://api.example.com/process',
        body: task.json
      });
      return { ...task.json, result: response, status: 'success' };
    })
  );
  
  results.push(...batchResults.map((r, idx) => ({
    json: r.status === 'fulfilled' 
      ? r.value 
      : { ...batch[idx].json, error: r.reason.message, status: 'failed' }
  })));
  
  // 批次间延迟
  if (i + MAX_CONCURRENT < tasks.length) {
    await new Promise(resolve => setTimeout(resolve, 1000));
  }
}

return results;

11. 数据转换与处理管道

11.1 数据清洗

// Code 节点：数据清洗管道
const rawData = $input.all();

const cleaned = rawData.map(item => {
  const data = item.json;
  
  return {
    json: {
      // 去除空白
      name: (data.name || '').trim(),
      email: (data.email || '').trim().toLowerCase(),
      
      // 格式化电话号码
      phone: (data.phone || '').replace(/\D/g, ''),
      
      // 标准化日期
      date: data.date ? new Date(data.date).toISOString().split('T')[0] : null,
      
      // 数值转换
      amount: parseFloat(data.amount) || 0,
      quantity: parseInt(data.quantity) || 0,
      
      // 文本标准化
      status: (data.status || '').toLowerCase().replace(/\s+/g, '_'),
      
      // 去除 HTML 标签
      description: (data.description || '').replace(/<[^>]*>/g, ''),
      
      // 提取域名
      domain: data.email ? data.email.split('@')[1] : null,
      
      // 元数据
      _processedAt: new Date().toISOString(),
      _source: 'webhook'
    }
  };
});

// 过滤无效数据
const valid = cleaned.filter(item => 
  item.json.name && item.json.email
);

// 去重
const unique = valid.filter((item, index, self) =>
  index === self.findIndex(t => t.json.email === item.json.email)
);

return unique;

11.2 数据聚合与分析

// Code 节点：数据聚合
const data = $input.all();

// 按类别分组
const grouped = {};
data.forEach(item => {
  const category = item.json.category || '未分类';
  if (!grouped[category]) grouped[category] = [];
  grouped[category].push(item.json);
});

// 计算统计信息
const stats = Object.entries(grouped).map(([category, items]) => ({
  category,
  count: items.length,
  totalAmount: items.reduce((sum, i) => sum + (i.amount || 0), 0),
  avgAmount: items.reduce((sum, i) => sum + (i.amount || 0), 0) / items.length,
  latestDate: items.map(i => i.date).sort().reverse()[0]
}));

// 排序
stats.sort((a, b) => b.totalAmount - a.totalAmount);

return [{
  json: {
    summary: stats,
    totalItems: data.length,
    totalCategories: stats.length,
    generatedAt: new Date().toISOString()
  }
}];

11.3 JSON 转换与映射

// Code 节点：复杂数据映射
const input = $json;

// 映射到不同的数据结构
const output = {
  // 扁平化嵌套结构
  customer: {
    id: input.customer_id,
    name: `${input.first_name} ${input.last_name}`,
    email: input.contact_info.email,
    phone: input.contact_info.phones[0]?.number,
    address: {
      street: input.address.street,
      city: input.address.city,
      state: input.address.state,
      zip: input.address.zip_code,
      full: `${input.address.street}, ${input.address.city}, ${input.address.state} ${input.address.zip_code}`
    }
  },
  
  // 数组转换
  orders: (input.orders || []).map(order => ({
    orderId: order.id,
    date: order.created_at,
    items: order.line_items.map(item => ({
      name: item.product_name,
      qty: item.quantity,
      price: item.unit_price,
      subtotal: item.quantity * item.unit_price
    })),
    total: order.line_items.reduce((sum, i) => sum + i.quantity * i.unit_price, 0)
  })),
  
  // 计算字段
  metrics: {
    totalOrders: input.orders?.length || 0,
    totalSpent: input.orders?.reduce((sum, o) => 
      sum + o.line_items.reduce((s, i) => s + i.quantity * i.unit_price, 0), 0) || 0,
    memberSince: input.created_at,
    lastActivity: input.orders?.sort((a, b) => 
      new Date(b.created_at) - new Date(a.created_at))[0]?.created_at
  }
};

return [{ json: output }];

12. 错误处理与监控体系

12.1 n8n 错误处理机制

n8n 提供多种错误处理方式：

节点级错误处理：

Settings → On Error → Continue On Fail

工作流级错误处理：

Error Trigger → 通知节点 → 日志记录 → 重试逻辑

12.2 实现重试机制

// Code 节点：带重试的 API 调用
async function callWithRetry(url, options, maxRetries = 3) {
  let lastError;
  
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      const response = await this.helpers.httpRequest({
        url,
        ...options
      });
      return { success: true, data: response, attempts: attempt };
    } catch (error) {
      lastError = error;
      
      // 检查是否应该重试
      if (error.statusCode === 429) {
        // Rate limited - 指数退避
        const retryAfter = error.response?.headers['retry-after'] || Math.pow(2, attempt);
        await new Promise(r => setTimeout(r, retryAfter * 1000));
      } else if (error.statusCode >= 500) {
        // 服务器错误 - 等待后重试
        await new Promise(r => setTimeout(r, Math.pow(2, attempt) * 1000));
      } else {
        // 客户端错误 - 不重试
        break;
      }
    }
  }
  
  return { 
    success: false, 
    error: lastError.message,
    statusCode: lastError.statusCode,
    attempts: maxRetries 
  };
}

const result = await callWithRetry(
  'https://api.example.com/data',
  {
    method: 'POST',
    body: $json.payload,
    headers: { 'Authorization': `Bearer ${$env.API_TOKEN}` }
  }
);

return [{ json: result }];

12.3 错误通知系统

// Error Trigger 工作流
const error = $json.execution.error;
const workflow = $json.workflow;
const execution = $json.execution;

// 构建错误报告
const errorReport = {
  workflow: workflow.name,
  workflowId: workflow.id,
  executionId: execution.id,
  error: {
    message: error.message,
    stack: error.stack,
    node: error.node?.name,
    timestamp: new Date().toISOString()
  },
  context: {
    startedAt: execution.startedAt,
    finishedAt: execution.finishedAt,
    mode: execution.mode
  }
};

// 发送到 Slack
await this.helpers.httpRequest({
  method: 'POST',
  url: $env.SLACK_WEBHOOK_URL,
  body: {
    blocks: [
      {
        type: 'header',
        text: { type: 'plain_text', text: '🚨 工作流执行失败' }
      },
      {
        type: 'section',
        fields: [
          { type: 'mrkdwn', text: `*工作流:*\n${errorReport.workflow}` },
          { type: 'mrkdwn', text: `*节点:*\n${errorReport.error.node}` },
          { type: 'mrkdwn', text: `*时间:*\n${errorReport.error.timestamp}` },
          { type: 'mrkdwn', text: `*执行ID:*\n${errorReport.executionId}` }
        ]
      },
      {
        type: 'section',
        text: { type: 'mrkdwn', text: `*错误信息:*\n\`\`\`${errorReport.error.message}\`\`\`` }
      }
    ]
  }
});

// 记录到数据库
await this.helpers.httpRequest({
  method: 'POST',
  url: 'https://api.example.com/error-logs',
  body: errorReport
});

return [{ json: { reported: true, errorReport } }];

12.4 执行监控仪表板

// 定时监控工作流
const workflows = await this.helpers.httpRequest({
  url: 'http://localhost:5678/api/v1/workflows',
  headers: { 'X-N8N-API-KEY': $env.N8N_API_KEY }
});

const stats = [];

for (const workflow of workflows.data) {
  const executions = await this.helpers.httpRequest({
    url: `http://localhost:5678/api/v1/executions?workflowId=${workflow.id}&limit=100`,
    headers: { 'X-N8N-API-KEY': $env.N8N_API_KEY }
  });
  
  const recent = executions.data;
  const failed = recent.filter(e => e.finished === false || e.stoppedAt === null);
  
  stats.push({
    name: workflow.name,
    id: workflow.id,
    active: workflow.active,
    totalExecutions: recent.length,
    failedExecutions: failed.length,
    successRate: ((recent.length - failed.length) / recent.length * 100).toFixed(1) + '%',
    lastExecution: recent[0]?.startedAt
  });
}

// 生成报告
const report = {
  generatedAt: new Date().toISOString(),
  totalWorkflows: stats.length,
  activeWorkflows: stats.filter(s => s.active).length,
  workflows: stats.sort((a, b) => b.failedExecutions - a.failedExecutions)
};

return [{ json: report }];

13. 企业级部署方案

13.1 Docker Compose 生产部署

version: '3.8'

services:
  n8n:
    image: docker.n8n.io/n8nio/n8n:latest
    restart: always
    ports:
      - "5678:5678"
    environment:
      # 基础配置
      - N8N_HOST=n8n.your-domain.com
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://n8n.your-domain.com/
      
      # 加密密钥（用于加密凭据）
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
      
      # 数据库
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
      
      # 执行模式
      - EXECUTIONS_MODE=queue
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_BULL_REDIS_PORT=6379
      
      # 数据清理
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168  # 7天
      
      # 限制
      - N8N_PAYLOAD_SIZE_MAX=50
      
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres
      - redis
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
        reservations:
          cpus: '0.5'
          memory: 1G

  n8n-worker:
    image: docker.n8n.io/n8nio/n8n:latest
    command: worker
    restart: always
    environment:
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_BULL_REDIS_PORT=6379
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres
      - redis
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '1'
          memory: 2G

  postgres:
    image: postgres:15-alpine
    restart: always
    environment:
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=${DB_PASSWORD}
      - POSTGRES_DB=n8n
    volumes:
      - postgres_data:/var/lib/postgresql/data
    deploy:
      resources:
        limits:
          cpus: '1'
          memory: 2G

  redis:
    image: redis:7-alpine
    restart: always
    command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data

  # Nginx 反向代理
  nginx:
    image: nginx:alpine
    restart: always
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./ssl:/etc/nginx/ssl:ro
    depends_on:
      - n8n

volumes:
  n8n_data:
  postgres_data:
  redis_data:

13.2 Nginx 反向代理配置

events {
    worker_connections 1024;
}

http {
    upstream n8n {
        server n8n:5678;
    }

    server {
        listen 80;
        server_name n8n.your-domain.com;
        return 301 https://$server_name$request_uri;
    }

    server {
        listen 443 ssl http2;
        server_name n8n.your-domain.com;

        ssl_certificate /etc/nginx/ssl/cert.pem;
        ssl_certificate_key /etc/nginx/ssl/key.pem;

        # 安全头
        add_header X-Frame-Options "SAMEORIGIN" always;
        add_header X-Content-Type-Options "nosniff" always;
        add_header X-XSS-Protection "1; mode=block" always;
        add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;

        # 请求大小限制
        client_max_body_size 50m;

        location / {
            proxy_pass http://n8n;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
            
            # WebSocket 支持
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            
            # 超时设置
            proxy_connect_timeout 300;
            proxy_send_timeout 300;
            proxy_read_timeout 300;
        }

        # Webhook 路径特殊处理
        location /webhook/ {
            proxy_pass http://n8n;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
            
            # 更大的超时用于长时间运行的 Webhook
            proxy_read_timeout 600;
        }
    }
}

13.3 Kubernetes 部署

# n8n-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: n8n
  labels:
    app: n8n
spec:
  replicas: 2
  selector:
    matchLabels:
      app: n8n
  template:
    metadata:
      labels:
        app: n8n
    spec:
      containers:
        - name: n8n
          image: docker.n8n.io/n8nio/n8n:latest
          ports:
            - containerPort: 5678
          env:
            - name: N8N_HOST
              value: "n8n.your-domain.com"
            - name: N8N_PORT
              value: "5678"
            - name: N8N_PROTOCOL
              value: "https"
            - name: WEBHOOK_URL
              value: "https://n8n.your-domain.com/"
            - name: DB_TYPE
              value: "postgresdb"
            - name: DB_POSTGRESDB_HOST
              valueFrom:
                secretKeyRef:
                  name: n8n-secrets
                  key: db-host
            - name: DB_POSTGRESDB_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: n8n-secrets
                  key: db-password
            - name: N8N_ENCRYPTION_KEY
              valueFrom:
                secretKeyRef:
                  name: n8n-secrets
                  key: encryption-key
          resources:
            requests:
              memory: "1Gi"
              cpu: "500m"
            limits:
              memory: "4Gi"
              cpu: "2000m"
          volumeMounts:
            - name: n8n-data
              mountPath: /home/node/.n8n
          livenessProbe:
            httpGet:
              path: /healthz
              port: 5678
            initialDelaySeconds: 30
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /healthz
              port: 5678
            initialDelaySeconds: 5
            periodSeconds: 5
      volumes:
        - name: n8n-data
          persistentVolumeClaim:
            claimName: n8n-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: n8n-service
spec:
  selector:
    app: n8n
  ports:
    - port: 80
      targetPort: 5678
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: n8n-ingress
  annotations:
    cert-manager.io/cluster-issuer: "letsencrypt-prod"
    nginx.ingress.kubernetes.io/proxy-body-size: "50m"
spec:
  tls:
    - hosts:
        - n8n.your-domain.com
      secretName: n8n-tls
  rules:
    - host: n8n.your-domain.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: n8n-service
                port:
                  number: 80

13.4 备份与恢复策略

#!/bin/bash
# backup-n8n.sh

BACKUP_DIR="/backups/n8n"
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_PATH="${BACKUP_DIR}/n8n_backup_${DATE}"

mkdir -p "${BACKUP_PATH}"

# 备份数据库
echo "备份 PostgreSQL 数据库..."
docker exec postgres pg_dump -U n8n n8n | gzip > "${BACKUP_PATH}/database.sql.gz"

# 备份 n8n 数据目录
echo "备份 n8n 数据..."
docker cp n8n:/home/node/.n8n "${BACKUP_PATH}/n8n_data"

# 备份工作流定义
echo "导出工作流..."
curl -s -H "X-N8N-API-KEY: ${N8N_API_KEY}" \
  "http://localhost:5678/api/v1/workflows" | \
  python3 -m json.tool > "${BACKUP_PATH}/workflows.json"

# 创建压缩包
tar -czf "${BACKUP_PATH}.tar.gz" -C "${BACKUP_DIR}" "n8n_backup_${DATE}"
rm -rf "${BACKUP_PATH}"

# 清理旧备份（保留最近30天）
find "${BACKUP_DIR}" -name "*.tar.gz" -mtime +30 -delete

echo "备份完成: ${BACKUP_PATH}.tar.gz"

14. 实战项目：自动化内容生成与分发系统

14.1 项目概述

构建一个完整的自动化内容生成与分发系统，功能包括：

内容收集：定时从多个来源收集素材
AI 内容生成：使用 AI 生成高质量文章
内容审核：自动审核内容质量
多平台分发：自动发布到多个平台
数据统计：收集各平台数据并生成报告

14.2 系统架构

┌─────────────────────────────────────────────────────────────┐
│                    内容生成与分发系统架构                        │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐                  │
│  │ RSS 订阅  │  │ 新闻 API │  │ 社交媒体 │  ← 数据源         │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘                  │
│       │              │              │                        │
│       └──────────────┼──────────────┘                        │
│                      ▼                                       │
│            ┌─────────────────┐                               │
│            │  内容聚合与去重   │                               │
│            └────────┬────────┘                               │
│                     ▼                                        │
│            ┌─────────────────┐                               │
│            │  AI 内容生成     │  ← OpenAI / Claude            │
│            └────────┬────────┘                               │
│                     ▼                                        │
│            ┌─────────────────┐                               │
│            │  质量审核        │  ← AI 自动审核                 │
│            └────────┬────────┘                               │
│                     ▼                                        │
│  ┌──────────────────┼──────────────────┐                    │
│  │                  │                  │                    │
│  ▼                  ▼                  ▼                    │
│ ┌──────┐      ┌──────────┐      ┌──────────┐               │
│ │微信公 │      │ 知乎/掘金 │      │ Twitter  │  ← 分发平台    │
│ │众号   │      │          │      │ /X       │               │
│ └──────┘      └──────────┘      └──────────┘               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

14.3 工作流 1：内容收集与聚合

Schedule Trigger (每6小时)
    → HTTP Request (RSS Feed)
    → Code (解析RSS)
    → Code (去重过滤)
    → Set (格式化数据)
    → Postgres (存储素材)

RSS 解析代码：

// Code 节点：解析多个 RSS 源
const rssFeeds = [
  { url: 'https://techcrunch.com/feed/', category: '科技' },
  { url: 'https://feeds.arstechnica.com/arstechnica/index', category: '技术' },
  { url: 'https://openai.com/blog/rss.xml', category: 'AI' }
];

const allItems = [];

for (const feed of rssFeeds) {
  try {
    const response = await this.helpers.httpRequest({
      url: feed.url,
      headers: { 'Accept': 'application/rss+xml, application/xml' }
    });
    
    // 简单 XML 解析（生产环境建议使用 xml2js）
    const items = response.match(/<item>([\s\S]*?)<\/item>/g) || [];
    
    items.forEach(item => {
      const title = item.match(/<title>(.*?)<\/title>/)?.[1] || '';
      const link = item.match(/<link>(.*?)<\/link>/)?.[1] || '';
      const description = item.match(/<description>([\s\S]*?)<\/description>/)?.[1]
        ?.replace(/<[^>]*>/g, '').trim() || '';
      const pubDate = item.match(/<pubDate>(.*?)<\/pubDate>/)?.[1] || '';
      
      allItems.push({
        title: title.replace(/<!\[CDATA\[|\]\]>/g, ''),
        link,
        description: description.substring(0, 500),
        pubDate: new Date(pubDate).toISOString(),
        category: feed.category,
        source: new URL(feed.url).hostname
      });
    });
  } catch (error) {
    console.error(`Failed to fetch ${feed.url}: ${error.message}`);
  }
}

// 按发布时间排序
allItems.sort((a, b) => new Date(b.pubDate) - new Date(a.pubDate));

// 去重（基于标题相似度）
const unique = [];
const seenTitles = new Set();

for (const item of allItems) {
  const normalizedTitle = item.title.toLowerCase().replace(/[^\w\u4e00-\u9fa5]/g, '');
  if (!seenTitles.has(normalizedTitle)) {
    seenTitles.add(normalizedTitle);
    unique.push(item);
  }
}

return unique.slice(0, 20).map(item => ({ json: item }));

14.4 工作流 2：AI 内容生成

Postgres Trigger (新素材)
    → Code (准备提示词)
    → OpenAI (生成文章)
    → Code (格式化输出)
    → IF (质量检查)
    → Postgres (存储文章)

内容生成代码：

// Code 节点：准备内容生成提示词
const material = $json;

const prompt = `你是一个专业的技术内容创作者。请基于以下素材，创作一篇高质量的技术文章。

## 素材信息
- 标题：${material.title}
- 来源：${material.source}
- 分类：${material.category}
- 摘要：${material.description}

## 写作要求
1. 文章长度：1500-2500字
2. 语言风格：专业但易懂，适合技术人员阅读
3. 结构：包含引言、正文（2-3个要点）、总结
4. 原创性：用自己的语言重新组织，不要直接复制原文
5. 价值：提供独到见解或实用建议
6. SEO：包含相关关键词

## 输出格式
请返回 JSON 格式：
{
  "title": "文章标题",
  "content": "文章正文（Markdown格式）",
  "tags": ["标签1", "标签2", "标签3"],
  "summary": "100字以内的摘要"
}`;

return [{
  json: {
    ...material,
    generationPrompt: prompt,
    requestedAt: new Date().toISOString()
  }
}];

质量审核代码：

// Code 节点：AI 质量审核
const article = $json;

const reviewPrompt = `请对以下文章进行质量审核，评分标准：

1. 原创性（0-25分）：是否有明显抄袭痕迹
2. 准确性（0-25分）：技术内容是否准确
3. 可读性（0-25分）：是否易于理解
4. 价值性（0-25分）：是否提供有用信息

文章标题：${article.title}
文章内容：${article.content}

请返回 JSON 格式：
{
  "score": 总分(0-100),
  "breakdown": {
    "originality": 分数,
    "accuracy": 分数,
    "readability": 分数,
    "value": 分数
  },
  "issues": ["问题1", "问题2"],
  "suggestion": "改进建议",
  "approved": true/false (总分>=70为通过)
}`;

// 调用 AI 审核（这里简化处理）
const reviewResult = {
  score: 78,
  approved: true,
  suggestion: '文章质量良好，建议发布'
};

return [{
  json: {
    ...article,
    review: reviewResult,
    approved: reviewResult.approved
  }
}];

14.5 工作流 3：多平台分发

// Code 节点：多平台分发逻辑
const article = $json;

if (!article.approved) {
  return [{ json: { status: 'skipped', reason: '未通过审核' } }];
}

const results = [];

// 1. 发布到微信公众号（示例）
try {
  const wechatResult = await this.helpers.httpRequest({
    method: 'POST',
    url: 'https://api.weixin.qq.com/cgi-bin/draft/add',
    headers: { 'Content-Type': 'application/json' },
    body: {
      articles: [{
        title: article.title,
        content: article.content,
        digest: article.summary,
        thumb_media_id: article.coverImageId
      }]
    }
  });
  results.push({ platform: 'wechat', status: 'success', id: wechatResult.media_id });
} catch (error) {
  results.push({ platform: 'wechat', status: 'failed', error: error.message });
}

// 2. 发布到知乎（示例）
try {
  const zhihuResult = await this.helpers.httpRequest({
    method: 'POST',
    url: 'https://www.zhihu.com/api/v4/articles',
    headers: {
      'Authorization': `Bearer ${$env.ZHIHU_TOKEN}`,
      'Content-Type': 'application/json'
    },
    body: {
      title: article.title,
      content: article.content,
      topics: article.tags
    }
  });
  results.push({ platform: 'zhihu', status: 'success', url: zhihuResult.url });
} catch (error) {
  results.push({ platform: 'zhihu', status: 'failed', error: error.message });
}

// 3. 发布到 Twitter/X（示例）
try {
  const tweetContent = `${article.title}\n\n${article.summary}\n\n${article.tags.map(t => '#' + t).join(' ')}`;
  
  const twitterResult = await this.helpers.httpRequest({
    method: 'POST',
    url: 'https://api.twitter.com/2/tweets',
    headers: {
      'Authorization': `Bearer ${$env.TWITTER_BEARER_TOKEN}`,
      'Content-Type': 'application/json'
    },
    body: { text: tweetContent.substring(0, 280) }
  });
  results.push({ platform: 'twitter', status: 'success', id: twitterResult.data.id });
} catch (error) {
  results.push({ platform: 'twitter', status: 'failed', error: error.message });
}

// 记录分发结果
return [{
  json: {
    articleId: article.id,
    title: article.title,
    distributedAt: new Date().toISOString(),
    results,
    successCount: results.filter(r => r.status === 'success').length,
    failedCount: results.filter(r => r.status === 'failed').length
  }
}];

14.6 工作流 4：数据统计与报告

// Code 节点：生成周报
const stats = $json;

const report = `
# 📊 内容运营周报

**报告周期**：${stats.startDate} 至 ${stats.endDate}

## 📈 核心指标

| 指标 | 本周 | 上周 | 变化 |
|------|------|------|------|
| 发布文章数 | ${stats.articlesPublished} | ${stats.lastWeekArticles} | ${stats.articlesChange} |
| 总阅读量 | ${stats.totalViews} | ${stats.lastWeekViews} | ${stats.viewsChange} |
| 平均阅读量 | ${stats.avgViews} | ${stats.lastWeekAvgViews} | ${stats.avgViewsChange} |
| 互动率 | ${stats.engagementRate}% | ${stats.lastWeekEngagement}% | ${stats.engagementChange} |

## 📝 本周发布文章

${stats.articles.map(a => `
### ${a.title}
- 平台：${a.platforms.join(', ')}
- 阅读量：${a.views}
- 点赞：${a.likes}
- 评论：${a.comments}
`).join('\n')}

## 🏆 平台表现

${stats.platformPerformance.map(p => `
- **${p.name}**：${p.articles}篇文章，总阅读${p.views}，互动率${p.engagementRate}%
`).join('\n')}

## 💡 本周洞察

${stats.insights.map(i => `- ${i}`).join('\n')}

## 📋 下周计划

${stats.nextWeekPlan.map(p => `- [ ] ${p}`).join('\n')}

---
*报告由 AI 自动生成*
`;

return [{
  json: {
    report,
    generatedAt: new Date().toISOString(),
    period: `${stats.startDate} - ${stats.endDate}`
  }
}];

14.7 完整工作流配置导出

将所有工作流组合在一起，通过 n8n 的 API 进行部署：

#!/bin/bash
# deploy-workflows.sh

N8N_URL="https://n8n.your-domain.com"
API_KEY="${N8N_API_KEY}"

# 部署内容收集工作流
curl -X POST "${N8N_URL}/api/v1/workflows" \
  -H "X-N8N-API-KEY: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d @workflows/content-collection.json

# 部署内容生成工作流
curl -X POST "${N8N_URL}/api/v1/workflows" \
  -H "X-N8N-API-KEY: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d @workflows/content-generation.json

# 部署分发工作流
curl -X POST "${N8N_URL}/api/v1/workflows" \
  -H "X-N8N-API-KEY: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d @workflows/content-distribution.json

# 部署报告工作流
curl -X POST "${N8N_URL}/api/v1/workflows" \
  -H "X-N8N-API-KEY: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d @workflows/weekly-report.json

echo "所有工作流部署完成！"

15. 常见问题与解决方案

15.1 安装与部署问题

Q1：n8n 启动后无法访问？

# 检查端口是否被占用
netstat -tlnp | grep 5678

# 检查 Docker 日志
docker compose logs n8n

# 常见原因及解决方案：
# 1. 防火墙未开放端口
sudo ufw allow 5678

# 2. Nginx 配置错误
nginx -t && nginx -s reload

# 3. 环境变量配置错误
# 检查 N8N_HOST、N8N_PORT 等配置

Q2：数据库连接失败？

# 测试 PostgreSQL 连接
docker exec -it postgres psql -U n8n -d n8n -c "SELECT 1;"

# 检查数据库是否就绪
docker compose ps postgres

# 重置数据库
docker compose down -v
docker compose up -d

Q3：Webhook 无法接收外部请求？

# 确认 WEBHOOK_URL 配置正确
echo $WEBHOOK_URL

# 测试 Webhook 可达性
curl -X POST https://your-domain.com/webhook/test \
  -H "Content-Type: application/json" \
  -d '{"test": true}'

# 检查 Nginx 代理配置
# 确保 proxy_pass 正确指向 n8n 服务

15.2 AI 集成问题

Q4：OpenAI API 调用超时？

// 解决方案：增加超时时间和重试机制
const response = await this.helpers.httpRequest({
  method: 'POST',
  url: 'https://api.openai.com/v1/chat/completions',
  body: { /* ... */ },
  timeout: 120000, // 2分钟超时
  retry: {
    maxRetries: 3,
    retryDelay: 1000,
    retryOn: [429, 500, 502, 503, 504]
  }
});

Q5：Token 使用量过大？

// 解决方案：实现 Token 计数和限制
const MAX_TOKENS = 4000;

function estimateTokens(text) {
  // 粗略估算：1个中文字≈2个token，1个英文单词≈1个token
  return Math.ceil(text.length * 1.5);
}

const messages = [];
let totalTokens = 0;

// 从最新的消息开始添加，直到达到限制
for (let i = history.length - 1; i >= 0; i--) {
  const msgTokens = estimateTokens(history[i].content);
  if (totalTokens + msgTokens > MAX_TOKENS) break;
  messages.unshift(history[i]);
  totalTokens += msgTokens;
}

Q6：AI 输出格式不稳定？

// 解决方案：使用 JSON Schema 约束输出
const response = await openai.chat.completions.create({
  model: 'gpt-4o',
  messages: [
    {
      role: 'system',
      content: `请严格按照以下JSON格式输出，不要包含任何其他内容：
{
  "title": "文章标题",
  "content": "文章内容",
  "tags": ["标签1", "标签2"],
  "score": 85
}`
    },
    { role: 'user', content: userInput }
  ],
  response_format: { type: 'json_object' }
});

// 验证输出
let output;
try {
  output = JSON.parse(response.choices[0].message.content);
} catch (e) {
  // 解析失败，使用默认值
  output = { title: '未命名', content: '', tags: [], score: 0 };
}

15.3 性能优化问题

Q7：工作流执行缓慢？

// 解决方案1：并行处理
const items = $input.all();
const batchSize = 5;

const results = [];
for (let i = 0; i < items.length; i += batchSize) {
  const batch = items.slice(i, i + batchSize);
  const batchResults = await Promise.all(
    batch.map(item => processItem(item))
  );
  results.push(...batchResults);
}

// 解决方案2：缓存重复查询
const cacheKey = `cache_${JSON.stringify($json.query)}`;
let cachedResult = await this.helpers.httpRequest({
  url: `http://redis:6379/get/${cacheKey}`
}).catch(() => null);

if (!cachedResult) {
  cachedResult = await expensiveOperation($json);
  // 缓存结果
  await this.helpers.httpRequest({
    method: 'POST',
    url: `http://redis:6379/set/${cacheKey}`,
    body: cachedResult,
    ttl: 3600
  });
}

Q8：内存溢出？

// 解决方案：流式处理大数据
// 不要一次性加载所有数据
const CHUNK_SIZE = 100;
let offset = 0;
let hasMore = true;

while (hasMore) {
  const chunk = await fetchChunk(offset, CHUNK_SIZE);
  
  // 处理当前批次
  await processChunk(chunk);
  
  offset += CHUNK_SIZE;
  hasMore = chunk.length === CHUNK_SIZE;
  
  // 强制垃圾回收（如果可用）
  if (global.gc) global.gc();
}

15.4 调试技巧

Q9：如何调试复杂工作流？

// 使用 Code 节点添加调试日志
const debug = {
  timestamp: new Date().toISOString(),
  node: '当前节点名称',
  input: $input.all().length + ' items',
  data: JSON.stringify($json).substring(0, 500)
};

console.log(JSON.stringify(debug, null, 2));

// 使用 n8n 的执行日志功能
// 在工作流设置中启用 "Save Execution Progress"

Q10：如何测试 Webhook？

# 使用 curl 测试
curl -X POST https://your-n8n.com/webhook-test/my-webhook \
  -H "Content-Type: application/json" \
  -d '{
    "name": "测试用户",
    "email": "test@example.com",
    "message": "这是一条测试消息"
  }'

# 使用 ngrok 进行本地测试
ngrok http 5678
# 然后使用 ngrok 提供的 URL 配置 Webhook

16. 总结与进阶资源

16.1 本教程核心要点回顾

通过本教程，你已经掌握了：

n8n 基础：架构理解、安装部署、核心概念
AI 集成：OpenAI、Claude、本地模型的接入方法
LangChain 集成：Chain、Agent、Memory 的使用
RAG 实现：文档索引、向量检索、混合搜索
Agent 构建：自主决策、工具调用、复杂任务处理
系统集成：Webhook、定时任务、数据管道
生产部署：Docker、Kubernetes、监控告警
实战项目：完整的自动化内容生成与分发系统

16.2 进阶学习路径

初级进阶：

学习更多 n8n 内置节点的使用
掌握 JavaScript/Python Code 节点的高级用法
了解 n8n 的变量系统和表达式语法

中级进阶：

开发自定义 n8n 节点
实现复杂的 Agent 工作流
集成更多外部服务和 API

高级进阶：

n8n 源码贡献
企业级高可用架构设计
AI 工作流的性能优化和成本控制

16.3 推荐资源

官方资源：

n8n 官方文档：https://docs.n8n.io
n8n 社区论坛：https://community.n8n.io
n8n GitHub：https://github.com/n8n-io/n8n

AI 相关：

OpenAI API 文档：https://platform.openai.com/docs
LangChain 文档：https://js.langchain.com
LlamaIndex 文档：https://docs.llamaindex.ai

社区资源：

n8n 工作流模板库：https://n8n.io/workflows
n8n YouTube 频道：https://www.youtube.com/@n8n

16.4 下一步行动

动手实践：按照教程完成所有代码示例
构建项目：完成实战项目，或设计自己的自动化场景
加入社区：参与 n8n 社区讨论，分享你的工作流
持续学习：关注 n8n 和 AI 领域的最新动态

📝 关于本教程：本教程基于 n8n 最新版本编写，代码示例均经过验证。如有问题或建议，欢迎反馈。

🔗 分享：如果你觉得本教程有帮助，欢迎分享给更多需要的人。

n8n + AI 自动化工作流完全教程

教程简介

n8n + AI 自动化工作流完全教程

📋 目录

1. 课程概述与学习路线

1.1 什么是 n8n？

1.2 为什么选择 n8n + AI？

1.3 学习路线图

2. n8n 架构深度解析

2.1 整体架构

2.2 核心组件

2.3 数据流模型

3. n8n 安装与环境配置

3.1 系统要求

3.2 Docker 快速部署（推荐）

3.3 npm 本地安装

3.4 源码编译安装

3.5 环境变量配置详解

4. n8n 核心概念与基础操作

4.1 工作流（Workflow）

4.2 节点类型

4.3 表达式语法

4.4 第一个工作流：AI 文本处理

5. AI 节点集成入门

5.1 OpenAI 集成

5.2 Anthropic Claude 集成

5.3 本地模型集成（Ollama）

5.4 多模型路由策略

6. LangChain 与 n8n 深度集成

6.1 LangChain 基础概念

6.2 n8n 中的 LangChain 节点

6.3 构建第一个 LangChain 工作流

6.4 自定义 LangChain Chain

6.5 RAG（检索增强生成）基础

7. LlamaIndex 连接与 RAG 工作流

7.1 LlamaIndex 简介

7.2 使用 n8n 构建完整 RAG 工作流

7.3 高级 RAG 技术

8. AI Agent 自动化构建

8.1 什么是 AI Agent？

8.2 n8n 中的 Agent 类型

8.3 构建功能丰富的 Agent

8.4 自定义 Agent Tool

8.5 Agent 最佳实践

9. Webhook 触发与外部集成

9.1 Webhook 基础

9.2 处理 Webhook 数据

9.3 响应 Webhook 请求

9.4 常见 Webhook 集成场景

10. 定时任务与调度策略

10.1 Schedule Trigger 详解

10.2 复杂调度场景

10.3 任务队列管理

11. 数据转换与处理管道

11.1 数据清洗

11.2 数据聚合与分析

11.3 JSON 转换与映射

12. 错误处理与监控体系

12.1 n8n 错误处理机制

12.2 实现重试机制

12.3 错误通知系统

12.4 执行监控仪表板

13. 企业级部署方案

13.1 Docker Compose 生产部署

13.2 Nginx 反向代理配置

13.3 Kubernetes 部署

13.4 备份与恢复策略

14. 实战项目：自动化内容生成与分发系统

14.1 项目概述

14.2 系统架构

14.3 工作流 1：内容收集与聚合

14.4 工作流 2：AI 内容生成

14.5 工作流 3：多平台分发

14.6 工作流 4：数据统计与报告

14.7 完整工作流配置导出

15. 常见问题与解决方案

15.1 安装与部署问题

15.2 AI 集成问题

15.3 性能优化问题

15.4 调试技巧