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

本教程全面讲解n8n AI工作流自动化平台的核心功能与实战应用，通过丰富的代码示例和完整的实战案例，帮助开发者从零开始掌握AI驱动的工作流自动化技术。

n8n概述与核心概念
n8n安装与部署
节点与触发器设计
AI Agent节点集成
LLM API调用
数据转换与映射
Webhook与定时任务
错误处理与重试机制
多系统集成
自定义代码节点
RAG知识库集成
企业级工作流架构
与Dify/Zapier对比
实战案例一：AI内容审核工作流
实战案例二：智能客服工作流
最佳实践与性能优化
总结

n8n概述与核心概念

什么是n8n

n8n（发音为"n-eight-n"）是一个开源的工作流自动化平台，它允许用户通过可视化的方式连接各种应用和服务，构建复杂的自动化工作流。与Zapier、Make等商业平台不同，n8n可以自托管部署，数据完全掌控在自己手中，同时支持代码级别的自定义扩展。

在AI时代，n8n已经发展成为构建AI工作流的强大平台。它原生支持AI Agent节点、LLM调用、向量数据库集成、RAG知识库等AI能力，使得开发者可以快速将AI能力嵌入到业务流程中。

n8n的核心架构

┌─────────────────────────────────────────────┐
│                  n8n 平台                     │
│  ┌───────────┐  ┌───────────┐  ┌──────────┐ │
│  │  触发器    │  │  执行引擎  │  │  存储层   │ │
│  │ (Trigger) │→│ (Executor) │→│ (Storage) │ │
│  └───────────┘  └───────────┘  └──────────┘ │
│       ↑              ↓              ↓        │
│  ┌───────────┐  ┌───────────┐  ┌──────────┐ │
│  │  Webhook  │  │   节点     │  │  数据库   │ │
│  │  定时器    │  │  (Nodes)  │  │  文件系统  │ │
│  │  手动触发  │  │  300+集成  │  │  外部存储  │ │
│  └───────────┘  └───────────┘  └──────────┘ │
└─────────────────────────────────────────────┘

核心概念

工作流（Workflow）：由多个节点组成的自动化流程
节点（Node）：工作流中的单个操作单元，如发送邮件、调用API、执行代码
触发器（Trigger）：启动工作流的事件，如Webhook、定时任务、文件变更
连接（Connection）：节点之间的数据传递关系
凭证（Credential）：访问第三方服务的认证信息
执行（Execution）：工作流的一次运行实例

n8n安装与部署

Docker一键部署（推荐）

Docker是最简单的部署方式，适合快速开始和生产环境。

# 创建docker-compose.yml
mkdir -p ~/n8n && cd ~/n8n

cat > docker-compose.yml << 'EOF'
version: '3.8'

services:
  n8n:
    image: docker.n8n.io/n8nio/n8n:latest
    restart: always
    ports:
      - "5678:5678"
    environment:
      # 基础配置
      - N8N_HOST=localhost
      - N8N_PORT=5678
      - N8N_PROTOCOL=http
      - WEBHOOK_URL=http://localhost:5678/
      
      # 数据库配置（使用SQLite，生产环境建议PostgreSQL）
      - DB_TYPE=sqlite
      - DB_SQLITE_DATABASE=/home/node/.n8n/database.sqlite
      
      # 安全配置
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=your-secure-password
      
      # AI相关配置
      - OPENAI_API_KEY=sk-your-openai-key
      - N8N_AI_ENABLED=true
      
      # 执行配置
      - EXECUTIONS_MODE=regular
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168
      
    volumes:
      - n8n_data:/home/node/.n8n
      
    # 健康检查
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:5678/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  n8n_data:
EOF

# 启动服务
docker compose up -d

# 查看日志
docker compose logs -f n8n

生产环境部署（Docker Compose + PostgreSQL）

# docker-compose.prod.yml
version: '3.8'

services:
  postgres:
    image: postgres:15-alpine
    restart: always
    environment:
      POSTGRES_DB: n8n
      POSTGRES_USER: n8n
      POSTGRES_PASSWORD: ${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U n8n"]
      interval: 10s
      timeout: 5s
      retries: 5

  n8n:
    image: docker.n8n.io/n8nio/n8n:latest
    restart: always
    depends_on:
      postgres:
        condition: service_healthy
    ports:
      - "5678:5678"
    environment:
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - WEBHOOK_URL=${WEBHOOK_URL}
      - N8N_ENCRYPTION_KEY=${ENCRYPTION_KEY}
    volumes:
      - n8n_data:/home/node/.n8n

  redis:
    image: redis:7-alpine
    restart: always
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  n8n_data:
  redis_data:

环境变量配置文件

# .env 文件
DB_PASSWORD=your-strong-db-password
N8N_USER=admin
N8N_PASSWORD=your-strong-n8n-password
WEBHOOK_URL=https://n8n.yourdomain.com/
ENCRYPTION_KEY=your-random-32-char-key-here
OPENAI_API_KEY=sk-your-openai-key
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key

npm安装（开发环境）

# 需要Node.js 18+
npm install n8n -g

# 启动n8n
n8n start

# 或者指定端口
N8N_PORT=5678 n8n start

访问n8n

部署完成后，访问 http://localhost:5678，首次访问需要设置管理员账号。

节点与触发器设计

触发器类型

n8n支持多种触发器，选择合适的触发器是设计工作流的第一步。

// n8n触发器类型速查
const triggerTypes = {
  // 时间触发
  "Schedule Trigger": {
    description: "定时执行，支持Cron表达式",
    useCase: "定时数据同步、报表生成、定期检查",
    example: "每天凌晨2点执行数据备份"
  },
  
  // Webhook触发
  "Webhook": {
    description: "接收HTTP请求触发",
    useCase: "接收外部系统回调、表单提交、API集成",
    example: "接收Stripe支付回调"
  },
  
  // 文件触发
  "Local File Trigger": {
    description: "监听文件系统变更",
    useCase: "文件上传处理、日志监控",
    example: "监控文件夹中的新CSV文件"
  },
  
  // 邮件触发
  "Email Trigger (IMAP)": {
    description: "监听新邮件",
    useCase: "邮件自动处理、工单创建",
    example: "收到支持邮件时自动创建工单"
  },
  
  // 消息队列触发
  "RabbitMQ Trigger": {
    description: "消费消息队列",
    useCase: "异步任务处理、事件驱动架构",
    example: "处理订单消息队列"
  },
  
  // 数据库触发
  "Postgres Trigger": {
    description: "监听数据库变更",
    useCase: "数据同步、实时通知",
    example: "新订单创建时触发通知"
  }
};

常用节点详解

// n8n核心节点速查
const coreNodes = {
  // 数据处理
  "Set": "设置/修改数据字段",
  "Function": "执行自定义JavaScript代码",
  "IF": "条件判断分支",
  "Switch": "多条件分支",
  "SplitInBatches": "分批处理大数据集",
  "Merge": "合并多路数据",
  
  // HTTP请求
  "HTTP Request": "发送HTTP请求调用API",
  
  // 数据库
  "Postgres": "PostgreSQL操作",
  "MySQL": "MySQL操作",
  "MongoDB": "MongoDB操作",
  
  // 消息通知
  "Slack": "发送Slack消息",
  "Email Send": "发送邮件",
  "Telegram": "Telegram机器人",
  
  // 文件操作
  "Read Binary File": "读取文件",
  "Write Binary File": "写入文件",
  
  // AI相关
  "AI Agent": "AI智能体节点",
  "OpenAI": "OpenAI API调用",
  "Vector Store": "向量数据库操作",
};

工作流设计模式

# 工作流设计的常见模式（用Python伪代码描述）

# 模式1：线性流水线
pipeline = """
触发器 → 数据获取 → 数据清洗 → AI处理 → 结果存储 → 通知
"""

# 模式2：条件分支
conditional = """
触发器 → 数据获取 → IF判断:
  ├─ 条件A → AI处理A → 结果A
  ├─ 条件B → AI处理B → 结果B
  └─ 默认 → 人工处理
"""

# 模式3：批量处理
batch = """
触发器 → 数据获取 → SplitInBatches → 循环:
  ├─ AI处理 → 结果收集
  └─ 等待/限速
→ 合并结果 → 通知
"""

# 模式4：错误重试
resilient = """
触发器 → 操作 → 成功?
  ├─ 是 → 继续
  └─ 否 → 等待 → 重试(最多3次)
    └─ 仍失败 → 错误通知 → 人工处理
"""

# 模式5：人工审批
human_in_loop = """
触发器 → AI分析 → 需要人工?
  ├─ 否 → 自动处理
  └─ 是 → 发送审批请求 → 等待审批
    ├─ 批准 → 执行
    └─ 拒绝 → 通知申请人
"""

AI Agent节点集成

n8n的AI Agent节点是其AI能力的核心，支持多种Agent类型和工具集成。

AI Agent节点配置

{
  "node": {
    "type": "@n8n/n8n-nodes-langchain.agent",
    "name": "AI Agent",
    "parameters": {
      "agent": "conversationalAgent",
      "text": "={{ $json.user_message }}",
      "systemMessage": "你是一个专业的技术客服，负责回答用户的技术问题。\n\n规则：\n1. 只回答技术相关问题\n2. 如果不确定，请说'我不确定，让我为您转接人工客服'\n3. 使用友好专业的语气",
      "maxIterations": 5,
      "returnIntermediateSteps": true
    }
  }
}

Agent类型选择

// n8n支持的Agent类型
const agentTypes = {
  "Conversational Agent": {
    description: "对话型Agent，适合多轮对话",
    useCase: "客服、咨询、问答",
    features: ["上下文记忆", "工具调用", "多轮推理"]
  },
  
  "Tools Agent": {
    description: "工具型Agent，专注于工具调用",
    useCase: "数据查询、API调用、文件操作",
    features: ["精确工具选择", "参数提取", "结果解析"]
  },
  
  "Plan and Execute Agent": {
    description: "规划型Agent，先规划再执行",
    useCase: "复杂任务分解、多步骤工作",
    features: ["任务规划", "步骤执行", "动态调整"]
  },
  
  "ReAct Agent": {
    description: "推理+行动型Agent",
    useCase: "需要推理的复杂查询",
    features: ["思维链推理", "工具调用", "自我纠正"]
  }
};

为Agent添加工具

{
  "tool_nodes": [
    {
      "type": "@n8n/n8n-nodes-langchain.toolCode",
      "name": "查询订单",
      "parameters": {
        "name": "query_order",
        "description": "根据订单号查询订单状态和物流信息",
        "code": "const orderId = $input.orderId;\n// 模拟数据库查询\nconst orders = {\n  '12345': { status: '已发货', tracking: 'SF1234567' },\n  '12346': { status: '待付款', tracking: null }\n};\nreturn orders[orderId] || { error: '订单未找到' };",
        "inputSchema": {
          "type": "object",
          "properties": {
            "orderId": {
              "type": "string",
              "description": "订单号"
            }
          },
          "required": ["orderId"]
        }
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.toolHttpRequest",
      "name": "搜索知识库",
      "parameters": {
        "name": "search_kb",
        "description": "在知识库中搜索相关信息",
        "method": "POST",
        "url": "http://kb-api:8080/search",
        "bodyParameters": {
          "parameters": [
            { "name": "query", "value": "={{ $input.query }}" }
          ]
        }
      }
    }
  ]
}

LLM API调用

通过HTTP Request节点调用OpenAI

{
  "node": {
    "type": "n8n-nodes-base.httpRequest",
    "name": "调用OpenAI",
    "parameters": {
      "method": "POST",
      "url": "https://api.openai.com/v1/chat/completions",
      "authentication": "genericCredentialType",
      "genericAuthType": "httpHeaderAuth",
      "sendHeaders": true,
      "headerParameters": {
        "parameters": [
          {
            "name": "Authorization",
            "value": "Bearer {{ $credentials.openaiApiKey }}"
          }
        ]
      },
      "sendBody": true,
      "bodyParameters": {
        "parameters": [
          {
            "name": "model",
            "value": "gpt-4o"
          },
          {
            "name": "messages",
            "value": "={{ JSON.stringify([{ role: 'system', content: '你是一个专业的翻译助手' }, { role: 'user', content: $json.input_text }]) }}"
          },
          {
            "name": "temperature",
            "value": "0.3"
          },
          {
            "name": "max_tokens",
            "value": "2000"
          }
        ]
      },
      "options": {
        "timeout": 60000
      }
    }
  }
}

通过HTTP Request节点调用Anthropic Claude

{
  "node": {
    "type": "n8n-nodes-base.httpRequest",
    "name": "调用Claude",
    "parameters": {
      "method": "POST",
      "url": "https://api.anthropic.com/v1/messages",
      "sendHeaders": true,
      "headerParameters": {
        "parameters": [
          { "name": "x-api-key", "value": "={{ $credentials.anthropicApiKey }}" },
          { "name": "anthropic-version", "value": "2023-06-01" },
          { "name": "content-type", "value": "application/json" }
        ]
      },
      "sendBody": true,
      "specifyBody": "json",
      "jsonBody": "={{ JSON.stringify({ model: 'claude-sonnet-4-20250514', max_tokens: 2000, messages: [{ role: 'user', content: $json.prompt }] }) }}"
    }
  }
}

使用n8n原生OpenAI节点

n8n提供原生的OpenAI节点，使用更简单：

{
  "node": {
    "type": "@n8n/n8n-nodes-langchain.openAi",
    "name": "OpenAI Chat",
    "parameters": {
      "resource": "chat",
      "model": "gpt-4o",
      "messages": {
        "values": [
          { "role": "system", "content": "你是一个专业的数据分析助手" },
          { "role": "user", "content": "={{ $json.question }}" }
        ]
      },
      "options": {
        "temperature": 0.5,
        "maxTokens": 2000,
        "responseFormat": "json"
      }
    },
    "credentials": {
      "openAiApi": {
        "id": "your-credential-id",
        "name": "OpenAI API"
      }
    }
  }
}

调用国产大模型

{
  "node": {
    "type": "n8n-nodes-base.httpRequest",
    "name": "调用通义千问",
    "parameters": {
      "method": "POST",
      "url": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
      "sendHeaders": true,
      "headerParameters": {
        "parameters": [
          { "name": "Authorization", "value": "Bearer {{ $credentials.dashscopeKey }}" }
        ]
      },
      "sendBody": true,
      "specifyBody": "json",
      "jsonBody": "={{ JSON.stringify({ model: 'qwen-max', messages: [{ role: 'system', content: '你是一个专业的AI助手' }, { role: 'user', content: $json.input }], temperature: 0.7, max_tokens: 2000 }) }}"
    }
  }
}

数据转换与映射

数据在节点间的传递

// n8n中数据的基本结构
const n8nDataStructure = {
  // 每个节点输出一个数组
  output: [
    {
      json: {           // JSON数据
        key1: "value1",
        key2: "value2",
        nested: {
          field: "value"
        }
      },
      binary: {         // 二进制数据（文件等）
        data: {
          data: "base64...",
          mimeType: "application/pdf",
          fileName: "document.pdf"
        }
      }
    }
  ]
};

// 表达式语法
const expressions = {
  // 引用上一个节点的数据
  "当前项": "{{ $json.fieldName }}",
  "上一个节点第一项": "{{ $('NodeName').first().json.fieldName }}",
  "上一个节点所有项": "{{ $('NodeName').all() }}",
  
  // 引用特定节点
  "指定节点": "{{ $node['NodeName'].json.fieldName }}",
  
  // 索引访问
  "第一项": "{{ $input.first().json.fieldName }}",
  "第N项": "{{ $input.all()[0].json.fieldName }}",
  
  // 内置方法
  "当前时间": "{{ $now }}",
  "工作流ID": "{{ $workflow.id }}",
  "执行ID": "{{ $execution.id }}",
};

Set节点数据映射

{
  "node": {
    "type": "n8n-nodes-base.set",
    "name": "数据映射",
    "parameters": {
      "mode": "manual",
      "assignments": {
        "assignments": [
          {
            "name": "user_name",
            "value": "={{ $json.firstName }} {{ $json.lastName }}",
            "type": "string"
          },
          {
            "name": "email",
            "value": "={{ $json.email.toLowerCase() }}",
            "type": "string"
          },
          {
            "name": "is_vip",
            "value": "={{ $json.totalSpent > 10000 }}",
            "type": "boolean"
          },
          {
            "name": "ai_prompt",
            "value": "={{ '用户 ' + $json.firstName + ' 的订单金额为 ' + $json.orderAmount + ' 元，请分析用户画像。' }}",
            "type": "string"
          }
        ]
      }
    }
  }
}

Function节点自定义转换

// n8n Function节点代码示例
// 将API响应转换为AI所需的输入格式

const items = $input.all();
const results = [];

for (const item of items) {
  const order = item.json;
  
  // 构建AI分析所需的结构化数据
  const aiInput = {
    user_id: order.userId,
    order_summary: {
      total_amount: order.amount,
      items_count: order.items.length,
      category: order.items.map(i => i.category).join(', '),
      order_date: new Date(order.createdAt).toISOString().split('T')[0],
    },
    user_history: {
      total_orders: order.user.totalOrders,
      avg_order_value: order.user.avgOrderValue,
      last_order_days_ago: Math.floor(
        (Date.now() - new Date(order.user.lastOrderDate)) / 86400000
      ),
    },
    // 构建给AI的分析提示词
    analysis_prompt: `请分析以下用户订单数据并给出建议：
用户ID: ${order.userId}
订单金额: ¥${order.amount}
商品类别: ${order.items.map(i => i.category).join(', ')}
用户历史订单数: ${order.user.totalOrders}
用户平均订单金额: ¥${order.user.avgOrderValue}

请分析：
1. 用户消费行为特征
2. 是否有流失风险
3. 推荐的营销策略`,
  };
  
  results.push({ json: aiInput });
}

return results;

Webhook与定时任务

Webhook配置

{
  "node": {
    "type": "n8n-nodes-base.webhook",
    "name": "接收Webhook",
    "parameters": {
      "path": "ai-process",
      "httpMethod": "POST",
      "responseMode": "responseNode",
      "options": {
        "rawBody": true
      }
    },
    "webhookId": "ai-webhook-001"
  }
}

// Webhook接收的数据结构
const webhookData = {
  headers: {
    "content-type": "application/json",
    "x-webhook-signature": "sha256=..."
  },
  body: {
    event: "new_review",
    data: {
      review_id: "REV-001",
      content": "这个产品非常好用，推荐购买！",
      user_id: "USER-123",
      product_id: "PROD-456"
    }
  },
  query: {},
};

Webhook安全验证

// Function节点：验证Webhook签名
const crypto = require('crypto');

const secret = 'your-webhook-secret';
const signature = $input.first().headers['x-webhook-signature'];
const body = JSON.stringify($input.first().body);

const expectedSig = 'sha256=' + crypto
  .createHmac('sha256', secret)
  .update(body)
  .digest('hex');

if (signature !== expectedSig) {
  throw new Error('Webhook签名验证失败');
}

// 签名验证通过，继续处理
return $input.all();

定时任务配置

{
  "node": {
    "type": "n8n-nodes-base.scheduleTrigger",
    "name": "每日定时",
    "parameters": {
      "rule": {
        "interval": [
          {
            "field": "cronExpression",
            "expression": "0 9 * * *"
          }
        ]
      }
    }
  }
}

// 常用Cron表达式
const cronExpressions = {
  "每天凌晨2点": "0 2 * * *",
  "每小时": "0 * * * *",
  "每5分钟": "*/5 * * * *",
  "工作日每天9点": "0 9 * * 1-5",
  "每周一上午10点": "0 10 * * 1",
  "每月1号凌晨": "0 0 1 * *",
};

错误处理与重试机制

错误处理节点

{
  "node": {
    "type": "n8n-nodes-base.errorTrigger",
    "name": "错误触发器",
    "parameters": {}
  }
}

工作流级错误处理

{
  "workflow": {
    "settings": {
      "errorWorkflow": "error-handler-workflow-id",
      "executionOrder": "v1"
    }
  }
}

节点级重试配置

{
  "node": {
    "type": "n8n-nodes-base.httpRequest",
    "name": "API调用（带重试）",
    "parameters": {
      "method": "POST",
      "url": "https://api.example.com/process",
      "options": {
        "timeout": 30000,
        "retry": {
          "maxRetries": 3,
          "retryInterval": 1000,
          "retryOn": [429, 500, 502, 503, 504]
        }
      }
    }
  }
}

自定义错误处理逻辑

// Function节点：带错误处理的API调用
const items = $input.all();
const results = [];
const errors = [];

for (const item of items) {
  try {
    // 模拟可能失败的操作
    const response = await this.helpers.httpRequest({
      method: 'POST',
      url: 'https://api.openai.com/v1/chat/completions',
      headers: {
        'Authorization': `Bearer ${$credentials.openaiApiKey}`,
        'Content-Type': 'application/json',
      },
      body: {
        model: 'gpt-4o',
        messages: [{ role: 'user', content: item.json.prompt }],
      },
    });
    
    results.push({
      json: {
        success: true,
        result: response.choices[0].message.content,
        usage: response.usage,
      }
    });
    
  } catch (error) {
    // 记录错误但继续处理
    errors.push({
      json: {
        success: false,
        error: error.message,
        input: item.json.prompt,
        timestamp: new Date().toISOString(),
      }
    });
    
    // 如果是速率限制错误，等待后重试
    if (error.statusCode === 429) {
      const retryAfter = error.response?.headers?.['retry-after'] || 60;
      await new Promise(resolve => setTimeout(resolve, retryAfter * 1000));
    }
  }
}

// 返回成功和失败的结果
return [...results, ...errors];

错误通知工作流

{
  "workflow": {
    "name": "错误处理工作流",
    "nodes": [
      {
        "type": "n8n-nodes-base.errorTrigger",
        "name": "错误触发",
        "parameters": {}
      },
      {
        "type": "n8n-nodes-base.set",
        "name": "格式化错误信息",
        "parameters": {
          "assignments": {
            "assignments": [
              {
                "name": "error_message",
                "value": "={{ '工作流执行失败\\n\\n工作流: ' + $json.workflow.name + '\\n节点: ' + $json.execution.lastNodeExecuted + '\\n错误: ' + $json.execution.error.message + '\\n时间: ' + $now }}",
                "type": "string"
              }
            ]
          }
        }
      },
      {
        "type": "n8n-nodes-base.slack",
        "name": "发送Slack通知",
        "parameters": {
          "channel": "#alerts",
          "text": "={{ $json.error_message }}",
          "otherOptions": {
            "includeLinkToWorkflow": true
          }
        }
      }
    ]
  }
}

多系统集成

Slack集成

{
  "node": {
    "type": "n8n-nodes-base.slack",
    "name": "发送Slack消息",
    "parameters": {
      "resource": "message",
      "operation": "send",
      "channel": "#ai-notifications",
      "text": "AI分析完成",
      "blocksUi": {
        "blocksValues": [
          {
            "type": "section",
            "text": {
              "type": "mrkdwn",
              "text": "*AI分析报告*\n\n用户: {{ $json.user_name }}\n分析结果: {{ $json.ai_result }}\n置信度: {{ $json.confidence }}%"
            }
          },
          {
            "type": "actions",
            "elements": [
              {
                "type": "button",
                "text": { "type": "plain_text", "text": "查看详情" },
                "url": "={{ $json.detail_url }}"
              }
            ]
          }
        ]
      }
    }
  }
}

邮件集成

{
  "node": {
    "type": "n8n-nodes-base.emailSend",
    "name": "发送邮件",
    "parameters": {
      "fromEmail": "ai@yourcompany.com",
      "toEmail": "={{ $json.recipient_email }}",
      "subject": "AI分析报告 - {{ $json.report_date }}",
      "html": "<h2>AI分析报告</h2><p>尊敬的{{ $json.user_name }}：</p><p>{{ $json.ai_summary }}</p><hr><p>详细结果：</p><pre>{{ $json.ai_detail }}</pre>",
      "options": {
        "attachments": "={{ $json.attachment_path }}"
      }
    }
  }
}

数据库集成

{
  "node": {
    "type": "n8n-nodes-base.postgres",
    "name": "保存AI结果",
    "parameters": {
      "operation": "insert",
      "table": "ai_analysis_results",
      "columns": "user_id, analysis_type, result, confidence, created_at",
      "values": "={{ $json.user_id }}, {{ $json.analysis_type }}, {{ $json.ai_result }}, {{ $json.confidence }}, NOW()"
    }
  }
}

向量数据库集成（Pinecone）

{
  "node": {
    "type": "@n8n/n8n-nodes-langchain.vectorStorePinecone",
    "name": "Pinecone向量存储",
    "parameters": {
      "operation": "search",
      "index": "knowledge-base",
      "query": "={{ $json.search_query }}",
      "topK": 5,
      "includeMetadata": true
    },
    "credentials": {
      "pineconeApi": {
        "id": "pinecone-cred-id",
        "name": "Pinecone API"
      }
    }
  }
}

自定义代码节点

JavaScript代码节点

// n8n Code节点 - JavaScript模式
// 处理AI响应并提取结构化数据

const items = $input.all();
const results = [];

for (const item of items) {
  const aiResponse = item.json.ai_response;
  
  // 尝试从AI响应中提取JSON
  let structuredData;
  try {
    // 查找JSON块
    const jsonMatch = aiResponse.match(/```json\n([\s\S]*?)\n```/);
    if (jsonMatch) {
      structuredData = JSON.parse(jsonMatch[1]);
    } else {
      // 尝试直接解析
      structuredData = JSON.parse(aiResponse);
    }
  } catch (e) {
    // 如果不是JSON，使用正则提取关键信息
    structuredData = {
      sentiment: aiResponse.includes('积极') ? 'positive' : 
                 aiResponse.includes('消极') ? 'negative' : 'neutral',
      keywords: (aiResponse.match(/关键词[：:]\s*(.*)/)?.[1] || '').split(/[,，、]/),
      summary: aiResponse.substring(0, 200),
    };
  }
  
  results.push({
    json: {
      ...item.json,
      structured_result: structuredData,
      processed_at: new Date().toISOString(),
    }
  });
}

return results;

Python代码节点（通过HTTP调用）

由于n8n原生不支持Python代码节点，可以通过HTTP请求调用Python微服务：

# python-service/app.py - Python AI处理微服务
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import openai
import json

app = FastAPI()

class AnalysisRequest(BaseModel):
    text: str
    analysis_type: str  # sentiment, summary, extract, translate
    language: Optional[str] = "zh"

class AnalysisResponse(BaseModel):
    result: dict
    confidence: float
    tokens_used: int

@app.post("/analyze", response_model=AnalysisResponse)
async def analyze(request: AnalysisRequest):
    """AI分析接口"""
    
    prompts = {
        "sentiment": f"分析以下文本的情感倾向（积极/中性/消极），返回JSON格式：\n{request.text}",
        "summary": f"用3句话总结以下内容：\n{request.text}",
        "extract": f"从以下文本中提取关键信息（人名、地点、时间、事件），返回JSON：\n{request.text}",
        "translate": f"将以下文本翻译为{'英文' if request.language == 'en' else '中文'}：\n{request.text}",
    }
    
    if request.analysis_type not in prompts:
        raise HTTPException(status_code=400, detail=f"不支持的分析类型: {request.analysis_type}")
    
    try:
        response = openai.ChatCompletion.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个专业的文本分析助手。请用JSON格式返回结果。"},
                {"role": "user", "content": prompts[request.analysis_type]}
            ],
            temperature=0.3,
        )
        
        result_text = response.choices[0].message.content
        try:
            result = json.loads(result_text)
        except:
            result = {"raw": result_text}
        
        return AnalysisResponse(
            result=result,
            confidence=0.85,
            tokens_used=response.usage.total_tokens
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/batch-analyze")
async def batch_analyze(texts: List[str], analysis_type: str):
    """批量分析接口"""
    results = []
    for text in texts:
        result = await analyze(AnalysisRequest(text=text, analysis_type=analysis_type))
        results.append(result.dict())
    return {"results": results}

n8n中调用Python微服务：

{
  "node": {
    "type": "n8n-nodes-base.httpRequest",
    "name": "调用Python分析服务",
    "parameters": {
      "method": "POST",
      "url": "http://python-service:8000/analyze",
      "sendBody": true,
      "specifyBody": "json",
      "jsonBody": "={{ JSON.stringify({ text: $json.content, analysis_type: 'sentiment' }) }}",
      "options": {
        "timeout": 30000
      }
    }
  }
}

RAG知识库集成

n8n RAG工作流架构

用户查询 → 向量化 → 向量检索 → 上下文组装 → LLM生成 → 返回结果
                ↑
知识文档 → 分块 → 向量化 → 存储到向量数据库

知识库导入工作流

{
  "workflow": {
    "name": "知识库导入",
    "nodes": [
      {
        "type": "n8n-nodes-base.readBinaryFile",
        "name": "读取文档",
        "parameters": {
          "filePath": "/data/knowledge-base/*.txt"
        }
      },
      {
        "type": "n8n-nodes-base.extractFromFile",
        "name": "提取文本",
        "parameters": {
          "operation": "text"
        }
      },
      {
        "type": "n8n-nodes-base.splitInBatches",
        "name": "分块处理",
        "parameters": {
          "batchSize": 10
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.textSplitterCharacterTextSplitter",
        "name": "文本分割",
        "parameters": {
          "chunkSize": 1000,
          "chunkOverlap": 200
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.embeddingsOpenAi",
        "name": "向量化",
        "parameters": {
          "model": "text-embedding-3-small"
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.vectorStorePinecone",
        "name": "存储到Pinecone",
        "parameters": {
          "operation": "insert",
          "index": "knowledge-base"
        }
      }
    ]
  }
}

RAG查询工作流

{
  "workflow": {
    "name": "RAG知识问答",
    "nodes": [
      {
        "type": "n8n-nodes-base.webhook",
        "name": "接收查询",
        "parameters": {
          "path": "rag-query",
          "httpMethod": "POST"
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.embeddingsOpenAi",
        "name": "查询向量化",
        "parameters": {
          "model": "text-embedding-3-small"
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.vectorStorePinecone",
        "name": "向量检索",
        "parameters": {
          "operation": "search",
          "index": "knowledge-base",
          "topK": 5
        }
      },
      {
        "type": "n8n-nodes-base.set",
        "name": "组装上下文",
        "parameters": {
          "assignments": {
            "assignments": [
              {
                "name": "context",
                "value": "={{ $json.matches.map(m => m.metadata.text).join('\\n\\n') }}",
                "type": "string"
              },
              {
                "name": "prompt",
                "value": "={{ '基于以下参考资料回答问题。\\n\\n参考资料：\\n' + $json.context + '\\n\\n用户问题：' + $('接收查询').first().json.body.question + '\\n\\n请用中文回答，如果参考资料中没有相关信息请说明。' }}",
                "type": "string"
              }
            ]
          }
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.openAi",
        "name": "AI回答",
        "parameters": {
          "resource": "chat",
          "model": "gpt-4o",
          "messages": {
            "values": [
              { "role": "system", "content": "你是一个专业的知识助手，基于提供的参考资料回答问题。" },
              { "role": "user", "content": "={{ $json.prompt }}" }
            ]
          }
        }
      }
    ]
  }
}

企业级工作流架构

微服务化部署

# docker-compose.enterprise.yml
version: '3.8'

services:
  # n8n主服务
  n8n-main:
    image: docker.n8n.io/n8nio/n8n:latest
    environment:
      - EXECUTIONS_MODE=queue
      - QUEUE_BULL_REDIS_HOST=redis
      - N8N_MULTI_MAIN_SETUP=true
    deploy:
      replicas: 2

  # n8n工作节点
  n8n-worker:
    image: docker.n8n.io/n8nio/n8n:latest
    command: worker
    environment:
      - EXECUTIONS_MODE=queue
      - QUEUE_BULL_REDIS_HOST=redis
    deploy:
      replicas: 3

  # Redis（队列）
  redis:
    image: redis:7-alpine

  # PostgreSQL（持久化）
  postgres:
    image: postgres:15-alpine

  # Python AI服务
  ai-service:
    build: ./ai-service
    deploy:
      replicas: 2

  # 向量数据库
  qdrant:
    image: qdrant/qdrant:latest

工作流模板化

{
  "template": {
    "name": "AI处理标准模板",
    "description": "标准的AI处理工作流模板，包含输入验证、AI处理、结果存储、通知",
    "category": "AI",
    "nodes": [
      {
        "type": "n8n-nodes-base.webhook",
        "name": "输入",
        "position": [250, 300]
      },
      {
        "type": "n8n-nodes-base.if",
        "name": "输入验证",
        "position": [450, 300],
        "parameters": {
          "conditions": {
            "string": [
              {
                "value1": "={{ $json.content }}",
                "operation": "isNotEmpty"
              }
            ]
          }
        }
      },
      {
        "type": "@n8n/n8n-nodes-langchain.openAi",
        "name": "AI处理",
        "position": [650, 250]
      },
      {
        "type": "n8n-nodes-base.postgres",
        "name": "保存结果",
        "position": [850, 250]
      },
      {
        "type": "n8n-nodes-base.slack",
        "name": "通知",
        "position": [1050, 250]
      },
      {
        "type": "n8n-nodes-base.respondToWebhook",
        "name": "返回错误",
        "position": [650, 400]
      }
    ]
  }
}

环境隔离

// n8n环境配置管理
const environments = {
  development: {
    n8n_url: "http://localhost:5678",
    ai_model: "gpt-4o-mini",
    vector_db: "local-chromadb",
    notification: "test-channel",
  },
  staging: {
    n8n_url: "https://n8n-staging.yourdomain.com",
    ai_model: "gpt-4o",
    vector_db: "pinecone-staging",
    notification: "staging-alerts",
  },
  production: {
    n8n_url: "https://n8n.yourdomain.com",
    ai_model: "gpt-4o",
    vector_db: "pinecone-production",
    notification: "production-alerts",
  }
};

与Dify/Zapier对比

功能对比

comparison = {
    "n8n": {
        "类型": "开源，可自托管",
        "AI能力": "原生AI Agent节点，支持多种LLM，RAG集成",
        "自定义能力": "支持JavaScript/Python代码节点，完全可扩展",
        "数据控制": "数据完全在自己服务器上",
        "价格": "免费（自托管）或云版付费",
        "学习曲线": "中等，需要了解节点和表达式",
        "适用场景": "需要深度定制的AI工作流、数据敏感场景",
        "优势": "开源免费、高度可定制、AI原生支持、社区活跃",
        "劣势": "需要自行部署维护、UI不如商业产品精致",
    },
    "Dify": {
        "类型": "开源，可自托管",
        "AI能力": "专注于AI应用开发，Agent/Chatbot/RAG内置",
        "自定义能力": "支持API扩展，但工作流能力较弱",
        "数据控制": "数据完全在自己服务器上",
        "价格": "免费（自托管）或云版付费",
        "学习曲线": "低，面向非技术用户设计",
        "适用场景": "快速构建AI聊天应用、知识库问答",
        "优势": "AI体验好、上手快、可视化强",
        "劣势": "通用自动化能力弱、第三方集成少",
    },
    "Zapier": {
        "类型": "商业SaaS平台",
        "AI能力": "AI功能有限，主要是调用外部API",
        "自定义能力": "支持Code by Zapier（JavaScript），但限制多",
        "数据控制": "数据存储在Zapier服务器",
        "价格": "按任务数付费，$19.99/月起",
        "学习曲线": "低，面向非技术用户",
        "适用场景": "简单的跨应用自动化、营销自动化",
        "优势": "集成最多（6000+）、无需部署、稳定",
        "劣势": "贵、数据不在自己手上、AI能力弱、自定义有限",
    }
}

选型建议

def recommend_platform(needs: dict) -> str:
    """根据需求推荐平台"""
    
    # 需要深度AI定制
    if needs.get("deep_ai_customization"):
        return "n8n"
    
    # 只需要AI聊天机器人
    if needs.get("chatbot_only"):
        return "Dify"
    
    # 需要大量第三方集成
    if needs.get("many_integrations", 0) > 100:
        if needs.get("budget") == "low":
            return "n8n"
        return "Zapier"
    
    # 数据安全要求高
    if needs.get("data_security") == "high":
        return "n8n"  # 自托管
    
    # 非技术用户
    if needs.get("technical_level") == "low":
        if needs.get("ai_focus"):
            return "Dify"
        return "Zapier"
    
    # 综合推荐
    return "n8n"

实战案例一：AI内容审核工作流

需求描述

构建一个自动化的内容审核系统：当有新内容提交时，通过AI进行内容审核，包括敏感词检测、情感分析、质量评分，审核通过的内容自动发布，不通过的发送人工复审通知。

工作流设计

Webhook接收内容 → 输入验证 → AI内容审核 → 审核结果判断
                                           ├─ 通过 → 自动发布 → 通知作者
                                           └─ 不通过 → 人工审核队列 → 通知审核员

完整工作流JSON

{
  "name": "AI内容审核工作流",
  "nodes": [
    {
      "type": "n8n-nodes-base.webhook",
      "name": "接收内容提交",
      "position": [250, 300],
      "parameters": {
        "path": "content-review",
        "httpMethod": "POST",
        "responseMode": "lastNode"
      }
    },
    {
      "type": "n8n-nodes-base.if",
      "name": "输入验证",
      "position": [450, 300],
      "parameters": {
        "conditions": {
          "string": [
            { "value1": "={{ $json.body.content }}", "operation": "isNotEmpty" },
            { "value1": "={{ $json.body.author_id }}", "operation": "isNotEmpty" }
          ]
        }
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.openAi",
      "name": "AI内容审核",
      "position": [650, 250],
      "parameters": {
        "resource": "chat",
        "model": "gpt-4o",
        "messages": {
          "values": [
            {
              "role": "system",
              "content": "你是一个专业的内容审核员。请对提交的内容进行全面审核，返回JSON格式的审核结果。\n\n审核维度：\n1. sensitive_content: 是否包含敏感内容（暴力、色情、政治敏感等），布尔值\n2. sentiment: 情感倾向（positive/neutral/negative）\n3. quality_score: 内容质量评分（1-10）\n4. spam_score: 垃圾内容概率（0-1）\n5. issues: 发现的问题列表\n6. recommendation: 审核建议（approve/reject/review）\n7. reason: 审核理由"
            },
            {
              "role": "user",
              "content": "请审核以下内容：\n\n标题：{{ $json.body.title }}\n内容：{{ $json.body.content }}\n作者ID：{{ $json.body.author_id }}"
            }
          ]
        },
        "options": {
          "temperature": 0.1,
          "responseFormat": "json_object"
        }
      }
    },
    {
      "type": "n8n-nodes-base.set",
      "name": "解析审核结果",
      "position": [850, 250],
      "parameters": {
        "assignments": {
          "assignments": [
            { "name": "review_result", "value": "={{ JSON.parse($json.message.content) }}", "type": "object" },
            { "name": "content_id", "value": "={{ $('接收内容提交').first().json.body.content_id }}", "type": "string" },
            { "name": "author_id", "value": "={{ $('接收内容提交').first().json.body.author_id }}", "type": "string" },
            { "name": "title", "value": "={{ $('接收内容提交').first().json.body.title }}", "type": "string" }
          ]
        }
      }
    },
    {
      "type": "n8n-nodes-base.if",
      "name": "审核结果判断",
      "position": [1050, 250],
      "parameters": {
        "conditions": {
          "string": [
            { "value1": "={{ $json.review_result.recommendation }}", "operation": "equals", "value2": "approve" }
          ]
        }
      }
    },
    {
      "type": "n8n-nodes-base.httpRequest",
      "name": "自动发布",
      "position": [1250, 150],
      "parameters": {
        "method": "POST",
        "url": "http://content-api:8080/api/content/publish",
        "sendBody": true,
        "specifyBody": "json",
        "jsonBody": "={{ JSON.stringify({ content_id: $json.content_id, status: 'published', review_score: $json.review_result.quality_score }) }}"
      }
    },
    {
      "type": "n8n-nodes-base.httpRequest",
      "name": "加入人工审核队列",
      "position": [1250, 350],
      "parameters": {
        "method": "POST",
        "url": "http://content-api:8080/api/review-queue",
        "sendBody": true,
        "specifyBody": "json",
        "jsonBody": "={{ JSON.stringify({ content_id: $json.content_id, issues: $json.review_result.issues, reason: $json.review_result.reason, priority: $json.review_result.sensitive_content ? 'high' : 'normal' }) }}"
      }
    },
    {
      "type": "n8n-nodes-base.slack",
      "name": "通知审核员",
      "position": [1450, 350],
      "parameters": {
        "channel": "#content-review",
        "text": "🚨 新内容需要人工审核\n\n标题：{{ $json.title }}\n问题：{{ $json.review_result.issues.join(', ') }}\n原因：{{ $json.review_result.reason }}\n优先级：{{ $json.review_result.sensitive_content ? '高' : '普通' }}"
      }
    },
    {
      "type": "n8n-nodes-base.respondToWebhook",
      "name": "返回结果",
      "position": [1450, 150],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ JSON.stringify({ success: true, status: $json.review_result.recommendation, score: $json.review_result.quality_score }) }}"
      }
    }
  ],
  "connections": {
    "接收内容提交": { "main": [[{ "node": "输入验证", "type": "main", "index": 0 }]] },
    "输入验证": { "main": [[{ "node": "AI内容审核", "type": "main", "index": 0 }]] },
    "AI内容审核": { "main": [[{ "node": "解析审核结果", "type": "main", "index": 0 }]] },
    "解析审核结果": { "main": [[{ "node": "审核结果判断", "type": "main", "index": 0 }]] },
    "审核结果判断": {
      "main": [
        [{ "node": "自动发布", "type": "main", "index": 0 }],
        [{ "node": "加入人工审核队列", "type": "main", "index": 0 }]
      ]
    },
    "自动发布": { "main": [[{ "node": "返回结果", "type": "main", "index": 0 }]] },
    "加入人工审核队列": { "main": [[{ "node": "通知审核员", "type": "main", "index": 0 }]] }
  }
}

测试数据

{
  "test_cases": [
    {
      "name": "正常内容",
      "data": {
        "content_id": "ART-001",
        "title": "Python入门教程",
        "content": "Python是一种简单易学的编程语言，适合初学者入门...",
        "author_id": "USER-001"
      },
      "expected": "approve"
    },
    {
      "name": "低质量内容",
      "data": {
        "content_id": "ART-002",
        "title": "asdfgh",
        "content": "买买买！！！点击链接 http://spam.com",
        "author_id": "USER-002"
      },
      "expected": "reject"
    },
    {
      "name": "需要复审",
      "data": {
        "content_id": "ART-003",
        "title": "关于某事件的看法",
        "content": "我认为这个事件需要从多个角度来看...",
        "author_id": "USER-003"
      },
      "expected": "review"
    }
  ]
}

实战案例二：智能客服工作流

需求描述

构建一个智能客服系统，能够自动回答常见问题，处理订单查询，当问题复杂时转接人工客服。系统需要维护对话上下文，支持多轮对话。

工作流设计

Webhook接收消息 → 用户识别 → 获取对话历史 → AI Agent处理
                                              ├─ 查询订单 → 订单API → 返回结果
                                              ├─ 常见问题 → 知识库检索 → AI回答
                                              └─ 复杂问题 → 转人工 → 通知客服

完整工作流JSON

{
  "name": "智能客服工作流",
  "nodes": [
    {
      "type": "n8n-nodes-base.webhook",
      "name": "接收用户消息",
      "position": [250, 300],
      "parameters": {
        "path": "customer-service",
        "httpMethod": "POST"
      }
    },
    {
      "type": "n8n-nodes-base.set",
      "name": "提取消息信息",
      "position": [450, 300],
      "parameters": {
        "assignments": {
          "assignments": [
            { "name": "user_id", "value": "={{ $json.body.user_id }}", "type": "string" },
            { "name": "message", "value": "={{ $json.body.message }}", "type": "string" },
            { "name": "channel", "value": "={{ $json.body.channel || 'web' }}", "type": "string" },
            { "name": "timestamp", "value": "={{ $now.toISO() }}", "type": "string" }
          ]
        }
      }
    },
    {
      "type": "n8n-nodes-base.httpRequest",
      "name": "获取对话历史",
      "position": [650, 300],
      "parameters": {
        "method": "GET",
        "url": "http://conversation-service:8080/api/conversations/{{ $json.user_id }}",
        "options": { "timeout": 5000 }
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.agent",
      "name": "AI客服Agent",
      "position": [850, 300],
      "parameters": {
        "agent": "conversationalAgent",
        "text": "用户消息：{{ $json.message }}",
        "systemMessage": "你是一个专业的客服助手，代表某电商平台为用户提供服务。\n\n你的能力：\n1. 查询订单状态和物流信息\n2. 解答商品相关问题\n3. 处理退换货咨询\n4. 提供购物流程指导\n\n规则：\n1. 始终保持友好、专业的语气\n2. 如果需要查询订单，请使用query_order工具\n3. 如果问题超出你的能力范围，建议转接人工客服\n4. 记住之前的对话上下文，避免重复询问\n5. 回答要简洁明了，避免过长",
        "maxIterations": 3
      }
    },
    {
      "type": "n8n-nodes-base.if",
      "name": "需要转人工?",
      "position": [1050, 300],
      "parameters": {
        "conditions": {
          "string": [
            { "value1": "={{ $json.output }}", "operation": "contains", "value2": "转接人工" }
          ]
        }
      }
    },
    {
      "type": "n8n-nodes-base.httpRequest",
      "name": "创建人工工单",
      "position": [1250, 400],
      "parameters": {
        "method": "POST",
        "url": "http://ticket-service:8080/api/tickets",
        "sendBody": true,
        "specifyBody": "json",
        "jsonBody": "={{ JSON.stringify({ user_id: $json.user_id, message: $json.message, priority: 'normal', source: 'ai_escalation', conversation_summary: $json.output }) }}"
      }
    },
    {
      "type": "n8n-nodes-base.httpRequest",
      "name": "保存对话记录",
      "position": [1250, 200],
      "parameters": {
        "method": "POST",
        "url": "http://conversation-service:8080/api/conversations/{{ $json.user_id }}/messages",
        "sendBody": true,
        "specifyBody": "json",
        "jsonBody": "={{ JSON.stringify({ role: 'assistant', content: $json.output, timestamp: $now.toISO() }) }}"
      }
    },
    {
      "type": "n8n-nodes-base.respondToWebhook",
      "name": "返回AI回复",
      "position": [1450, 200],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ JSON.stringify({ reply: $json.output, need_human: false }) }}"
      }
    },
    {
      "type": "n8n-nodes-base.respondToWebhook",
      "name": "返回转人工回复",
      "position": [1450, 400],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ JSON.stringify({ reply: '您的问题我需要转接人工客服为您处理，请稍候...', need_human: true, ticket_id: $json.ticket_id }) }}"
      }
    }
  ],
  "connections": {
    "接收用户消息": { "main": [[{ "node": "提取消息信息", "type": "main", "index": 0 }]] },
    "提取消息信息": { "main": [[{ "node": "获取对话历史", "type": "main", "index": 0 }]] },
    "获取对话历史": { "main": [[{ "node": "AI客服Agent", "type": "main", "index": 0 }]] },
    "AI客服Agent": { "main": [[{ "node": "需要转人工?", "type": "main", "index": 0 }]] },
    "需要转人工?": {
      "main": [
        [{ "node": "保存对话记录", "type": "main", "index": 0 }],
        [{ "node": "创建人工工单", "type": "main", "index": 0 }]
      ]
    },
    "保存对话记录": { "main": [[{ "node": "返回AI回复", "type": "main", "index": 0 }]] },
    "创建人工工单": { "main": [[{ "node": "返回转人工回复", "type": "main", "index": 0 }]] }
  }
}

最佳实践与性能优化

工作流性能优化

// 优化策略速查
const optimizationStrategies = {
  // 1. 批量处理
  batchProcessing: {
    description: "将多个单独的API调用合并为批量调用",
    example: "使用SplitInBatches节点，每批处理10-50条数据",
    benefit: "减少API调用次数，降低延迟和成本"
  },
  
  // 2. 并行执行
  parallelExecution: {
    description: "独立的分支并行执行",
    example: "AI审核和数据存储可以同时进行",
    benefit: "缩短总执行时间"
  },
  
  // 3. 缓存结果
  caching: {
    description: "缓存频繁查询的结果",
    example: "使用Redis缓存AI分析结果，相同输入不重复调用",
    benefit: "减少AI API调用，降低成本"
  },
  
  // 4. 限速控制
  rateLimiting: {
    description: "控制API调用频率",
    example: "使用Wait节点在批次间添加延迟",
    benefit: "避免触发API限流"
  },
  
  // 5. 条件执行
  conditionalExecution: {
    description: "只在需要时执行昂贵的操作",
    example: "先用规则引擎过滤明显不符合条件的数据，再调用AI",
    benefit: "减少不必要的AI调用"
  }
};

监控与告警

{
  "workflow": {
    "name": "n8n监控告警",
    "nodes": [
      {
        "type": "n8n-nodes-base.scheduleTrigger",
        "name": "每5分钟检查",
        "parameters": {
          "rule": { "interval": [{ "field": "cronExpression", "expression": "*/5 * * * *" }] }
        }
      },
      {
        "type": "n8n-nodes-base.httpRequest",
        "name": "获取执行统计",
        "parameters": {
          "method": "GET",
          "url": "http://localhost:5678/api/v1/executions?limit=100&status=error"
        }
      },
      {
        "type": "n8n-nodes-base.if",
        "name": "检查错误率",
        "parameters": {
          "conditions": {
            "number": [
              { "value1": "={{ $json.data.length }}", "operation": "larger", "value2": 5 }
            ]
          }
        }
      },
      {
        "type": "n8n-nodes-base.slack",
        "name": "发送告警",
        "parameters": {
          "channel": "#n8n-alerts",
          "text": "⚠️ n8n工作流告警\n\n最近5分钟内有 {{ $json.data.length }} 个执行失败\n请检查工作流状态。"
        }
      }
    ]
  }
}

安全最佳实践

security_checklist = {
    "凭证管理": [
        "使用n8n内置的凭证管理，不要硬编码密钥",
        "定期轮换API密钥",
        "为不同环境使用不同的凭证",
    ],
    "Webhook安全": [
        "验证Webhook签名",
        "使用HTTPS",
        "限制IP白名单（如可能）",
        "设置合理的超时时间",
    ],
    "数据安全": [
        "敏感数据不要记录到执行日志",
        "使用环境变量存储配置",
        "定期清理过期的执行数据",
    ],
    "访问控制": [
        "设置强密码",
        "启用MFA（如支持）",
        "限制用户权限",
        "审计用户操作",
    ],
    "网络安全": [
        "不要将n8n直接暴露在公网",
        "使用反向代理（Nginx/Caddy）",
        "配置防火墙规则",
    ]
}

总结

通过本教程的学习，你应该掌握了n8n AI工作流自动化的核心技术：

基础架构：n8n的安装部署、核心概念、节点与触发器设计
AI集成：AI Agent节点配置、LLM API调用、多模型支持
数据处理：数据转换与映射、表达式语法、Function节点自定义
可靠性：Webhook安全、定时任务、错误处理与重试机制
系统集成：Slack/Email/数据库/向量数据库的集成方法
代码扩展：JavaScript代码节点、Python微服务集成
RAG实现：知识库导入、向量检索、RAG查询工作流
企业级：微服务部署、环境隔离、工作流模板化
平台选型：n8n vs Dify vs Zapier的详细对比与选型建议
实战案例：AI内容审核和智能客服两个完整的工作流案例

n8n作为一个开源、可自托管的工作流自动化平台，在AI时代展现出了强大的生命力。它的AI原生支持、灵活的扩展能力和完全的数据控制权，使其成为构建AI自动化工作流的理想选择。

下一步学习建议：

动手实践：从简单的Webhook+AI工作流开始，逐步增加复杂度

探索社区：n8n社区有大量现成的工作流模板可以参考

深入Agent：学习更复杂的AI Agent架构，如Multi-Agent协作

生产部署：学习n8n的队列模式、高可用部署和监控告警

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

教程简介

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

目录

n8n概述与核心概念

什么是n8n

n8n的核心架构

核心概念

n8n安装与部署

Docker一键部署（推荐）

生产环境部署（Docker Compose + PostgreSQL）

环境变量配置文件

npm安装（开发环境）

访问n8n

节点与触发器设计

触发器类型

常用节点详解

工作流设计模式

AI Agent节点集成

AI Agent节点配置

Agent类型选择

为Agent添加工具

LLM API调用

通过HTTP Request节点调用OpenAI

通过HTTP Request节点调用Anthropic Claude

使用n8n原生OpenAI节点

调用国产大模型

数据转换与映射

数据在节点间的传递

Set节点数据映射

Function节点自定义转换

Webhook与定时任务

Webhook配置

Webhook安全验证

定时任务配置

错误处理与重试机制

错误处理节点

工作流级错误处理

节点级重试配置

自定义错误处理逻辑

错误通知工作流

多系统集成

Slack集成

邮件集成

数据库集成

向量数据库集成（Pinecone）

自定义代码节点

JavaScript代码节点

Python代码节点（通过HTTP调用）

RAG知识库集成

n8n RAG工作流架构

知识库导入工作流

RAG查询工作流

企业级工作流架构

微服务化部署

工作流模板化

环境隔离

与Dify/Zapier对比

功能对比

选型建议

实战案例一：AI内容审核工作流

需求描述

工作流设计

完整工作流JSON

测试数据

实战案例二：智能客服工作流

需求描述

工作流设计

完整工作流JSON

最佳实践与性能优化

工作流性能优化

监控与告警

安全最佳实践

总结

相关教程

Python入门完全教程

Java基础编程教程

Go语言实战教程

Python编程入门完全教程

目录