一段屏幕录制可以展示整个任务,却无法解释它。有人打开收件箱、检查发件人、将一个值复制到客户记录中、与电子表格对比、发送摘要,然后继续。
这项工作是可见的,但自动化仍需通过更严格的测试:系统能否在不遗漏任何步骤、不接错操作或导入看似正确实则日后失效的内容的情况下重建任务?
我围绕这个问题构建了屏幕分析项目的 n8n 部分。生成器不把最终文件视为一袋文本。它将发现的自动化转化为 N8NWorkflow、N8NNode 和连接对象,然后输出 n8n 兼容的 JavaScript Object Notation (JSON)。这增加了代码仪式,却能捕获字符串拼接容易引入的一类错误。
1. 保持平台词汇表狭窄
每个发出的节点类型都来自 n8n_workflow_generator.py 中的 NodeType。该枚举涵盖触发器、语言模型节点以及生成器知道如何创建的应用程序集成。当项目需要另一个 n8n 节点时,我会在生成器可以使用它之前先添加它。
这会降低编辑速度。一个临时的单一节点无法通过在提示中拼写新标识符而混入。好处是更清晰的失败:不受支持的平台名称会在 Python 中失败,而不是隐藏在可导入的文件中。
同样的思路也适用于 n8n_agent_templates.py 中的代理配置。AgentTemplate 命名可用模式;AgentConfig 携带提示、工具、集成、触发器偏好、模型选择、temperature 和迭代限制。提示只是一个字段,而不是其他所有内容的容器。
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import List
class AgentTemplate(Enum):
EMAIL_TRIAGE = "email_triage"
CRM_DATA_SYNC = "crm_data_sync"
CALENDAR_ASSISTANT = "calendar_assistant"
DOCUMENT_PROCESSOR = "document_processor"
COMMUNICATION_ROUTER = "communication_router"
REPORT_GENERATOR = "report_generator"
LEAD_QUALIFIER = "lead_qualifier"
TASK_MANAGER = "task_manager"
VOICE_ASSISTANT = "voice_assistant"
MULTI_AGENT_ORCHESTRATOR = "multi_agent_orchestrator"
@dataclass
class AgentConfig:
name: str
description: str
template: AgentTemplate
system_prompt: str
tools: List[str] = field(default_factory=list)
integrations: List[str] = field(default_factory=list)
triggers: List[str] = field(default_factory=list)
llm_model: str = "gemini-2.5-flash"
temperature: float = 0.7
max_iterations: int = 10
Enter fullscreen mode Exit fullscreen mode
该模式也是一个约束。如果新自动化需要 AgentConfig 无法表达的概念,我会先扩展模型。这会减慢实验速度,但能保持导出路径的诚实。
2. 先构建对象,再生成 JSON
生成器的结构化形式就是 n8n 图本身:节点加命名连接。它不维护第二种私有图格式。N8NWorkflow、N8NNode 和连接记录是分析结果与保存的 JSON 文件之间的表示形式。
flowchart TD
analysis[Discovered automation] --> workflow[N8NWorkflow]
workflow --> nodes[N8NNode objects]
workflow --> connections[Connection records]
nodes --> json[n8n JSON export]
connections --> json
json --> importer[REST importer]
importer --> status[Import status]```
这种区分很重要。只有当生成器有足够上下文来选择触发器、模型、集成和连接顺序时,“对这封邮件进行分类”这样的检测步骤才会被映射到具体的 n8n 节点。定位是与身份分开计算的,因此画布保持可读性,而不会将布局绑定到节点 ID。
权衡在于灵活性。确定性布局无法与手排画布相匹配,类型化构建也比直接编辑 JSON 文件更重。对于生成的自动化,我更喜欢可预测的检查而非完美的视觉布局。
核心对象形状很简单:
```python
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List
@dataclass
class N8NNode:
id: str
name: str
type: str
position: List[int]
parameters: Dict[str, Any] = field(default_factory=dict)
credentials: Dict[str, Any] = field(default_factory=dict)
type_version: float = 1.0
def to_dict(self) -> Dict[str, Any]:
node_dict = {
"id": self.id,
"name": self.name,
"type": self.type,
"position": self.position,
"parameters": self.parameters,
"typeVersion": self.type_version,
}
if self.credentials:
node_dict["credentials"] = self.credentials
return node_dict
Enter fullscreen mode Exit fullscreen mode
正是这一部分让 JSON 感觉像生成式代码。该对象在序列化发生前就拥有身份、类型、参数、凭证、版本和位置。等到文件存在时,重要决策已经通过可检查的 Python 结构。
3. 将导入视为部署状态
生成以文件结束;当该文件通过 Representational State Transfer (REST) API 到达 n8n 时,操作才开始。在 n8n_importer.py 中,导入失败有命名的异常,导入进度有明确的状态。
from enum import Enum
class N8NError(Exception):
"""Custom exception for n8n API errors."""
class ImportStatus(Enum):
PENDING = "pending"
IMPORTING = "importing"
SUCCESS = "success"
FAILED = "failed"
REQUIRES_CREDENTIALS = "requires_credentials"
Enter fullscreen mode Exit fullscreen mode
凭证问题和导入失败需要不同的恢复路径,因此它们获得不同的标签。部署流程可以生成自动化、创建支持的代理文件、导入它们,并在处理凭证时保持激活关闭。
这种分离移除了便利性。一个能生成、导入、处理凭证并激活的单一按钮对演示会更快。在生产环境中,拆分这些操作使部分失败可恢复。
4. 我测试所依据的表
| Layer | Question it answers |
|---|---|
AgentTemplate |
正在构建哪种自动化模式? |
AgentConfig |
哪种提示、工具、集成、触发器和模型设置描述了它? |
NodeType |
哪些 n8n 标识符可以被发出? |
N8NWorkflow / N8NNode
|
哪个图会变成 JSON? |
ImportStatus / N8NError
|
当制品到达 n8n 时发生了什么? |
这比字符串插值成本更高:额外的枚举、数据类、对象构建、保存步骤和导入器报告。它也使模式变更明确。我愿意付出这个代价,因为从屏幕录制推断的自动化本身就带有不确定性;导出路径应该降低这种不确定性。
当工作流 JSON 能够移动数据、调用模型并路由工作时,它就是代码。将它视为生成式代码,是我防止已发现流程变成导入事故的方法。
🎧 收听有声书 — Spotify · Google Play · 所有平台
🎬 在 YouTube 上观看视觉概述
📖 阅读完整的 13 部分系列
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.