n8n 工作流管理

n8n 工作流管理 是做什么的？

为 n8n 平台提供全面的工作流自动化管理功能，包括创建、测试、执行监控和性能优化。

⚠️ 关键：工作流创建规则

创建 n8n 工作流时，务必遵守以下规则：

1. ✅ 生成完整的工作流，包含所有功能节点
2. ✅ 包含实际的 HTTP 请求节点，用于 API 调用（如 ImageFX、Gemini、Veo、Suno 等）
3. ✅ 添加代码节点，用于数据转换和逻辑处理
4. ✅ 在所有节点之间创建正确的连接
5. ✅ 使用真实的节点类型（如 n8n-nodes-base.httpRequest、n8n-nodes-base.code、n8n-nodes-base.set）

绝对禁止：

• ❌ 创建“设置指南”占位节点
• ❌ 生成仅包含 TODO 注释的工作流
• ❌ 制作不完整、需要手动添加节点的工作流
• ❌ 使用纯文本节点替代实际功能

正确的工作流示例：

手动触发 → 设置配置 → HTTP 请求（API 调用） → 代码（解析） → 响应

错误的工作流示例：

手动触发 → 代码（“在此添加 HTTP 节点，配置 API...”）

请始终构建完整、可用的工作流，确保所有必要节点均已配置并连接。

设置

必需的环境变量：

• N8N_API_KEY — 您的 n8n API 密钥（在 n8n 界面中通过“设置 → API”获取）
• N8N_BASE_URL — 您的 n8n 实例 URL

通过 OpenClaw 设置配置凭据：

添加到 ~/.config/openclaw/settings.json：

{
  "skills": {
    "n8n": {
      "env": {
        "N8N_API_KEY": "your-api-key-here",
        "N8N_BASE_URL": "your-n8n-url-here"
      }
    }
  }
}

或者每次会话设置（不要将密钥持久保存到 shell rc 文件中）：

export N8N_API_KEY="your-api-key-here"
export N8N_BASE_URL="your-n8n-url-here"

验证连接：

python3 scripts/n8n_api.py list-workflows --pretty

**安全提示：**切勿将 API 密钥以明文形式存储在 shell 配置文件（~/.bashrc、~/.zshrc）中。请使用 OpenClaw 设置文件或安全的密钥管理器。

快速参考

工作流管理

列出工作流

python3 scripts/n8n_api.py list-workflows --pretty
python3 scripts/n8n_api.py list-workflows --active true --pretty

获取工作流详情

python3 scripts/n8n_api.py get-workflow --id <workflow-id> --pretty

创建工作流

# 从 JSON 文件创建
python3 scripts/n8n_api.py create --from-file workflow.json

激活/停用

python3 scripts/n8n_api.py activate --id <workflow-id>
python3 scripts/n8n_api.py deactivate --id <workflow-id>

测试与验证

验证工作流结构

# 验证现有工作流
python3 scripts/n8n_tester.py validate --id <workflow-id>

# 从文件验证
python3 scripts/n8n_tester.py validate --file workflow.json --pretty

# 生成验证报告
python3 scripts/n8n_tester.py report --id <workflow-id>

试运行测试

# 使用数据测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data '{"email": "[email protected]"}'

# 使用数据文件测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json

# 完整测试报告（验证 + 试运行）
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test.json --report

测试套件

# 运行多个测试用例
python3 scripts/n8n_tester.py test-suite --id <workflow-id> --test-suite test-cases.json

执行监控

列出执行记录

# 最近的执行（所有工作流）
python3 scripts/n8n_api.py list-executions --limit 10 --pretty

# 特定工作流的执行
python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 20 --pretty

获取执行详情

python3 scripts/n8n_api.py get-execution --id <execution-id> --pretty

手动执行

# 触发工作流
python3 scripts/n8n_api.py execute --id <workflow-id>

# 带数据执行
python3 scripts/n8n_api.py execute --id <workflow-id> --data '{"key": "value"}'

性能优化

分析性能

# 完整性能分析
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --pretty

# 分析特定时间段
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty

获取优化建议

# 按优先级排序的建议
python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty

生成优化报告

# 包含指标、瓶颈和建议的人类可读报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

获取工作流统计信息

# 执行统计
python3 scripts/n8n_api.py stats --id <workflow-id> --days 7 --pretty

Python API

基本用法

from scripts.n8n_api import N8nClient

client = N8nClient()

# 列出工作流
workflows = client.list_workflows(active=True)

# 获取工作流
workflow = client.get_workflow('workflow-id')

# 创建工作流
new_workflow = client.create_workflow({
    'name': 'My Workflow',
    'nodes': [...],
    'connections': {...}
})

# 激活/停用
client.activate_workflow('workflow-id')
client.deactivate_workflow('workflow-id')

# 执行
执行列表 = client.list_executions(workflow_id='workflow-id', limit=10)
执行详情 = client.get_execution('execution-id')

# 执行工作流
结果 = client.execute_workflow('workflow-id', data={'key': 'value'})

验证与测试

from scripts.n8n_api import N8nClient
from scripts.n8n_tester import WorkflowTester

client = N8nClient()
tester = WorkflowTester(client)

# 验证工作流
验证结果 = tester.validate_workflow(workflow_id='123')
print(f"是否有效: {验证结果['valid']}")
print(f"错误: {验证结果['errors']}")
print(f"警告: {验证结果['warnings']}")

# 试运行
试运行结果 = tester.dry_run(
    workflow_id='123',
    test_data={'email': '[email protected]'}
)
print(f"状态: {试运行结果['status']}")

# 测试套件
测试用例 = [
    {'name': '测试 1', 'input': {...}, 'expected': {...}},
    {'name': '测试 2', 'input': {...}, 'expected': {...}}
]
测试结果 = tester.test_suite('123', 测试用例)
print(f"通过: {测试结果['passed']}/{测试结果['total_tests']}")

# 生成报告
报告 = tester.generate_test_report(验证结果, 试运行结果)
print(报告)

性能优化

from scripts.n8n_optimizer import WorkflowOptimizer

optimizer = WorkflowOptimizer()

# 分析性能
分析结果 = optimizer.analyze_performance('workflow-id', days=7)
print(f"性能评分: {分析结果['performance_score']}/100")
print(f"健康状态: {分析结果['execution_metrics']['health']}")

# 获取建议
建议 = optimizer.suggest_optimizations('workflow-id')
print(f"优先操作: {len(建议['priority_actions'])}")
print(f"速效成果: {len(建议['quick_wins'])}")

# 生成报告
报告 = optimizer.generate_optimization_report(分析结果)
print(报告)

常见工作流程

1. 验证并测试工作流

# 验证工作流结构
python3 scripts/n8n_tester.py validate --id <workflow-id> --pretty

# 使用样本数据测试
python3 scripts/n8n_tester.py dry-run --id <workflow-id> \
  --data '{"email": "[email protected]", "name": "Test User"}'

# 如果测试通过，激活工作流
python3 scripts/n8n_api.py activate --id <workflow-id>

2. 调试失败的工作流

# 查看最近的执行记录
python3 scripts/n8n_api.py list-executions --id <workflow-id> --limit 10 --pretty

# 获取特定执行的详情
python3 scripts/n8n_api.py get-execution --id <execution-id> --pretty

# 验证工作流结构
python3 scripts/n8n_tester.py validate --id <workflow-id>

# 生成测试报告
python3 scripts/n8n_tester.py report --id <workflow-id>

# 检查优化问题
python3 scripts/n8n_optimizer.py report --id <workflow-id>

3. 优化工作流性能

# 分析当前性能
python3 scripts/n8n_optimizer.py analyze --id <workflow-id> --days 30 --pretty

# 获取可操作的建议
python3 scripts/n8n_optimizer.py suggest --id <workflow-id> --pretty

# 生成全面报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

# 查看执行统计
python3 scripts/n8n_api.py stats --id <workflow-id> --days 30 --pretty

# 通过试运行测试优化效果
python3 scripts/n8n_tester.py dry-run --id <workflow-id> --data-file test-data.json

4. 监控工作流健康状态

# 查看活动工作流
python3 scripts/n8n_api.py list-workflows --active true --pretty

# 查看最近的执行状态
python3 scripts/n8n_api.py list-executions --limit 20 --pretty

# 获取每个关键工作流的统计信息
python3 scripts/n8n_api.py stats --id <workflow-id> --pretty

# 生成健康报告
python3 scripts/n8n_optimizer.py report --id <workflow-id>

验证检查

测试模块执行全面的验证：

结构验证

• ✓ 必备字段存在（节点、连接）
• ✓ 所有节点都有名称和类型
• ✓ 连接目标存在
• ✓ 没有未连接的节点（警告）

配置验证

• ✓ 需要凭据的节点已配置
• ✓ 必要参数已设置
• ✓ HTTP 节点包含 URL
• ✓ Webhook 节点包含路径
• ✓ 邮件节点包含内容

流程验证

• ✓ 工作流包含触发节点
• ✓ 执行流程正常
• ✓ 无循环依赖
• ✓ 已识别结束节点

优化分析

优化器从多个维度进行分析：

执行指标

• 总执行次数
• 成功/失败率
• 健康状态（优秀/良好/一般/差）
• 错误模式

性能指标

• 节点数量与复杂度
• 连接模式
• 高开销操作（API 调用、数据库查询）
• 并行执行机会

瓶颈检测

• 串行高开销操作
• 高失败率
• 缺少错误处理
• 速率限制问题

优化机会

• **并行执行：**识别可以同时运行的节点
• **缓存：**建议对重复 API 调用进行缓存
• **批处理：**建议对大数据集进行批处理
• **错误处理：**添加错误恢复机制
• **复杂度降低：**拆分复杂工作流
• **超时设置：**配置执行限制

性能评分

工作流会获得一个性能评分（0-100），基于以下因素：

• **成功率：**越高越好（权重 50%）
• **复杂度：**越低越好（权重 30%）
• **瓶颈：**越少越好（严重：-20，高：-10，中：-5）
• **优化：**已实施的最佳实践（每个 +5）

分数解读：

• **90-100：**优秀——优化良好
• **70-89：**良好——可进行小幅改进
• **50-69：**一般——建议优化
• **0-49：**差——存在重大问题

最佳实践

开发

1. **规划结构：**在构建之前设计工作流节点和连接
2. **先验证：**部署前务必验证
3. **彻底测试：**使用多个测试用例进行试运行
4. **错误处理：**为可靠性添加错误节点
5. **文档记录：**在代码节点中注释复杂逻辑

测试

1. **样本数据：**创建逼真的测试数据文件
2. **边界情况：**测试边界条件和错误场景
3. **增量测试：**每添加一个节点就进行测试
4. **回归测试：**更改后重新测试
5. **生产环境模拟：**使用与生产环境类似的预发布环境

部署

1. **先停用：**以停用状态部署工作流
2. **逐步推广：**最初用有限流量测试
3. **密切监控：**仔细观察前几次执行
4. **快速回滚：**出现问题时可随时停用
5. **记录变更：**保存修改日志

优化

1. **基线指标：**在更改前记录性能数据
2. **一次只改一处：**隔离优化影响
3. **衡量结果：**对比更改前后的指标
4. **定期审查：**每月安排一次优化审查
5. **成本意识：**监控 API 使用量和执行成本

维护

1. **健康检查：**每周查看执行统计
2. **错误分析：**调查失败模式
3. **性能监控：**跟踪执行时间
4. **凭据轮换：**定期更新凭据
5. **清理：**归档或删除未使用的工作流

故障排除

身份验证错误

错误：环境中未找到 N8N_API_KEY

**解决方案：**设置环境变量：

export N8N_API_KEY="your-api-key"

连接错误

错误：HTTP 401：未授权

解决方案：

1. 验证 API 密钥是否正确
2. 检查 N8N_BASE_URL 是否正确设置
3. 确认 n8n 中已启用 API 访问

验证错误

验证失败：节点缺少“name”字段

**解决方案：**检查工作流 JSON 结构，确保所有必需字段都存在

执行超时

状态：超时——执行未完成

解决方案：

1. 检查工作流中是否存在无限循环
2. 减小测试数据集大小
3. 优化高开销操作
4. 在工作流设置中设置执行超时时间

速率限制

错误：HTTP 429：请求过多

解决方案：

1. 在 API 调用之间添加等待节点
2. 实现指数退避
3. 使用批处理
4. 检查 API 速率限制

凭据缺失

警告：节点“HTTP_Request”可能需要凭据

解决方案：

1. 在 n8n 界面中配置凭据
2. 将凭据分配给节点
3. 在激活前测试连接

文件结构

~/clawd/skills/n8n/
├── SKILL.md                    # 本文档
├── scripts/
│   ├── n8n_api.py             # 核心 API 客户端（扩展版）
│   ├── n8n_tester.py          # 测试与验证
│   └── n8n_optimizer.py       # 性能优化
└── references/
    └── api.md                 # n8n API 参考

API 参考

有关详细的 n8n REST API 文档，请参阅 references/api.md 或访问： https://docs.n8n.io/api/

Su

…