workflow-builder

# Workflow Builder 创建模式按以下顺序构建工作流：**场景确认与流程设计** → **确认 workflow.md 文档语言** → **workflow.md 草稿与确认循环** → **workflow.json 生成与验证** → **逐节点测试** → **集成测试** → **性能评估与优化**。 > **产物**：`workflow.json` 中所有可执行节点均为 `type="python"` 并自带 `code`。最终交付物固定为：`workflow.md`、`workflow.json`。 > > **顺序强制**：创建工作流时，`workflow.md` 是唯一的人类意图源。必须先询问用户需要中文还是英文流程说明文档，再按所选语言创建并确认 `workflow.md`，最后基于已确认的 `workflow.md` 生成 `workflow.json`。在 `workflow.md` 写入并确认前，严禁写入或覆盖 `workflow.json`。 ## 参考资料（按需读取） | 文件 | 内容 | 何时读取 | |------|------|---------| | [references/reference.md](references/reference.md) | 节点类型详解、出边选择行为、分支/循环/Join 规则、Edge Mapping 指南、Tool vs LLM 决策、文件输出规则、报告生成模板、`workflow.json` 骨架模板 | **生成 `workflow.json` 前建议读取** | | [references/composition.md](references/composition.md) | 嵌套工作流（subworkflow）组合格式与展开规则 | 仅在用户需要嵌套工作流时读取 | | [references/workflow_zh.md](references/workflow_zh.md) | 中文 `workflow.md` 结构模板 | 用户选择中文流程说明文档时读取 | | [references/workflow_en.md](references/workflow_en.md) | English `workflow.md` structure template | 用户选择英文流程说明文档时读取 | | [references/workflow_template/](references/workflow_template/) | 工作流创建参考包，包含标准 `workflow.md`、`workflow.json`、`config.json`、`guide.md` 和 `meta.json` 模板 | **创建工作流、生成配置模板或补齐 guide.md 前按需读取** | | `~/.flocks/plugins/workflows/stream_alert_denoise/workflow.md` | 已成型业务工作流示例，展示“功能、流程、输入输出、模块逻辑、发布配置、编辑指南”的写法 | 文件存在且需要参考真实工作流表达时读取 | --- ## 0. 开始前 ### Todo List（每次创建必须生成） **在开始任何工作前，必须先用 Todo 工具列出完整任务清单**，并在整个过程中实时更新状态（pending → in_progress → completed）。标准 Todo 清单如下，根据实际工作流复杂度增减： ``` [ ] 0. 核验可用工具列表（读取 registry.py） [ ] 1. 场景深度确认：与用户对话，明确业务场景与核心目标 [ ] 1. 输出思考维度分析 + Mermaid 流程简图，与用户沟通对齐 [ ] 1. 获取样例数据（用户上传或自动构造后确认） [ ] 2. 用 Question 工具确认 workflow.md 使用中文还是英文 [ ] 2. 读取对应语言模板 workflow_zh.md 或 workflow_en.md，以及可用业务示例 [ ] 2. 生成单份 workflow.md 草稿（人读描述，包含功能、流程、节点、输入输出、处理逻辑） [ ] 2. 写入 workflow.md 文件，供页面编辑器展示 [ ] 2. 向用户展示流程摘要并收集修改建议（循环直至满意） [ ] 2. 确认 workflow.md 已是最新意图源 [ ] 3. 读取 reference.md [ ] 3. 基于已确认 workflow.md 生成完整 workflow.json（含代码） [ ] 3. 写入 workflow.json 文件 [ ] 4. 验证 JSON 格式 + Python 语法 [ ] 4. 保存样例数据到 /api/workflow/{id}/sample-inputs [ ] 5. 逐节点测试：节点 1 - <node_id> [ ] 5. 逐节点测试：节点 2 - <node_id> [ ] 5. 逐节点测试：节点 N - <node_id>（按拓扑顺序补充） [ ] 6. 集成测试：全量运行验证 [ ] 6. 记录各节点及总运行时间 [ ] 7. 性能评估：识别瓶颈节点 [ ] 7. 优化慢节点（并发/缓存/精简 prompt 等） [ ] 7. 通知用户工作流已就绪 ``` > 每完成一项立即标记为 completed；每进入一项立即标记为 in_progress。**严禁在 Todo 全部完成前宣布任务结束。** --- ### 实时工具核验（重要） 生成 workflow 前必须核验可用工具列表与参数签名： - **强制读取** `flocks/flocks/tool/registry.py`（`ToolInfo.parameters` 是参数 schema 的权威来源）。 - 工具名必须一致，参数名必须严格对齐，禁止调用 `run_workflow`（在 `WORKFLOW_TOOL_BLOCKLIST` 中）。 --- ## 1. 第一阶段：场景确认与流程设计 > 目标：通过对话真正理解用户需求，输出完整的思考维度与流程简图，让用户在花时间构建工作流之前就能确认方向。 ### 1.1 深度场景对话 使用 `Question` 工具与用户确认以下维度（根据场景选取相关项，不必逐条询问，尽量合并为 1-2 轮对话）： **业务背景** - 这个工作流解决什么安全/业务问题？触发条件是什么？ - 谁会使用这个工作流？是定时自动触发还是手动触发？ - 有没有现有的 SOP 或人工处理流程可以参考？ **数据与工具** - 输入数据是什么？（告警字段、IP/域名/哈希、日志条目等） - 需要调用哪些外部工具或服务？（威胁情报、SIEM、资产库等） - 输出结果是什么？（报告、工单、通知、打标签等） **流程要求** - 有没有需要特殊处理的条件分支？（如高危 vs 低危、内部 IP vs 外部 IP） - 对运行时间有要求吗？（实时响应 < 30s？批量处理可接受数分钟？） - 有哪些已知的"陷阱"或边界情况需要注意？ ### 1.2 输出思考维度与流程简图 对话完成后，在消息中输出以下内容供用户确认： **思考维度总结**（结构化列举，涵盖：数据流、工具调用链、分支逻辑、异常处理、性能关键点、可扩展性） **流程简图**（使用 Mermaid flowchart 语法，清晰展示节点与边关系） 示例格式： ``` ## 思考维度 **数据流**：输入告警 → 资产丰富 → 情报查询 → LLM 分析 → 报告输出 **分支逻辑**：IP 类型判断（内网 / 外网）→ 不同查询策略 **性能关键点**：情报查询可并发；LLM 调用是主要耗时节点 **异常处理**：工具调用失败时降级到日志记录，不中断流程 **可扩展性**：后续可插入 SOAR 工单节点 ## 流程简图 \`\`\`mermaid flowchart TD A[接收告警] --> B[提取 IP/域名] B --> C{IP 类型?} C -->|内网| D[查询资产库] C -->|外网| E[查询威胁情报] D --> F[汇总上下文] E --> F F --> G[LLM 分析] G --> H[生成报告] \`\`\` ``` ### 1.3 获取样例数据 在流程简图得到用户认可后，请用户上传一条完整的样例输入数据（JSON 格式）： - 若用户能提供：直接使用 - 若用户无法提供：根据场景自动构造一条最小可用样例 JSON，并请用户确认字段和数值是否合理 > 样例数据将用于后续每个节点的逐步测试，是测试阶段的核心依据。获得样例后，待工作流 ID 确定时调用 `POST /api/workflow/{id}/sample-inputs` 保存（body: `{ "sampleInputs": <样例 JSON 对象> }`）。 --- ## 2. 第二阶段：生成并确认 workflow.md（人读意图源） > 目标：先把工作流的业务意图、节点结构、输入输出和处理逻辑写成可读、可编辑的 `workflow.md`。页面左侧编辑器以 `workflow.md` 表达工作流，用户应先在这里确认意图；只有确认后才能生成 `workflow.json`。 ### 2.0 文档语言选择（必须） 创建 `workflow.md` 前，必须用 `Question` 工具询问用户需要哪种流程说明文档： - 中文流程说明文档：读取 [references/workflow_zh.md](references/workflow_zh.md)，并可参考 [references/workflow_template/workflow.md](references/workflow_template/workflow.md) 的章节完整性，生成中文 `workflow.md`。 - English workflow specification：读取 [references/workflow_en.md](references/workflow_en.md)，并可参考 [references/workflow_template/workflow.md](references/workflow_template/workflow.md) 的章节完整性，生成英文 `workflow.md`。 规则： - 工作流目录里最终只写一份 `workflow.md`。 - 不要在工作流目录里创建 `workflow_zh.md`、`workflow_en.md`、`workflow.en.md` 或其它语言副本。 - `workflow_zh.md` / `workflow_en.md` 只是本 skill 内部的结构模板。 - `references/workflow_template/` 只是本 skill 内部的创建参考包，严禁复制成可扫描的 `workflow_template` 工作流目录；需要模板内容时，只读取其中的文件并改造成当前真实工作流。 - 不要根据用户当前会话语言自动猜测文档语言；创建 `workflow.md` 前必须明确询问并得到选择。 ### 2.1 核心要求 `workflow.md` 必须让人读得懂，也必须足够结构化，便于后续稳定生成 `workflow.json`。每个步骤必须包含： - **功能概述**：用人能理解的话说明这个工作流解决什么问题、不解决什么问题。 - **总体流程**：用箭头、表格或 Mermaid 描述节点顺序和职责。 - **输入/输出**：数据来源、格式、用途。 - **模块逻辑**：每个节点的职责、处理步骤、判定条件、循环方式、异常处理。 - **工具/LLM 标注**：明确该步是 Tool-driven 还是 LLM-driven（详细决策指南见 [reference.md § Tool vs LLM](references/reference.md#5-tool-vs-llm-决策指南)）。 - **推荐组合**：`tool.run_safe(...)` 获取数据 → `llm.ask(...)` 分析 → `tool.run('write', ...)` 落盘。 - **默认使用 `tool.run_safe()`**，返回 `{"success", "text", "obj", "error"}` 统一包络。 - **文件落盘**：节点有任何文件输出时，统一写入 `~/.flocks/workspace/outputs/<YYYY-MM-DD>/` 目录下，详见 [reference.md § 文件输出规则](references/reference.md#6-文件输出规则)。 - **决策分支**：写清条件、各分支处理、跳转规则。 - **发布和配置**：写清 API、Syslog、Kafka、Webhook、Schedule 等入口是否支持，运行态配置由 `config.json` 模板和 Storage/SQL 管理。 - **编辑指南**：告诉用户修改输入、节点逻辑、输出、发布方式时应该优先改哪里。 - **报告结构**（若涉及）：除非用户要求简化，需包含摘要、分析、发现、建议、来源（模板见 [reference.md § 报告生成](references/reference.md#7-报告生成最佳实践)）。 ### 2.2 写入 workflow.md 1. 先按用户选择的语言模板生成内容，再用 `write` 工具将单份 `workflow.md` **写入文件**（路径与第 9 节一致，例如 `.../plugins/workflows/<id>/workflow.md`）。 - **⚠️ 路径必须使用绝对路径**：全局目录可用 `python3 -c "import os; print(os.path.expanduser('~/.flocks/plugins/workflows/<id>'))"`；项目目录可先解析 workspace（从 cwd 向上第一个含 `.flocks` 的目录）再拼接 `/.flocks/plugins/workflows/<id>`。 - **严禁**使用未展开的相对路径（如 `.flocks/plugins/workflows/<id>/` 相对仓库根随手写入错误位置），否则 WebUI 可能无法从实际扫描目录读到文件。 - **严禁**同时写入 `workflow.en.md` 或语言副本；UI 和生成流程只认当前工作流目录下的 `workflow.md`。 2. 写入成功后，在消息中说明：「已创建 `workflow.md`，请在左侧编辑器查看并确认。需要调整节点、输入输出或处理逻辑时，请先改 `workflow.md`。」 3. 需要用户确认是否进入 `workflow.json` 生成时，必须使用 `Question` 工具或等待页面 diff 的接受/拒绝结果；不要用普通文本提问替代确认。 ### 2.3 用户反馈循环（循环直至满意） 收集用户对 `workflow.md` 的修改建议，按照以下循环执行，**直到用户确认满意**： ``` 接收用户反馈 ↓ 分析修改需求（功能描述、节点职责、输入输出、处理逻辑、分支关系） ↓ 更新 workflow.md ↓ 重新写入文件 ↓ 向用户展示更新摘要，并用 Question 工具或页面 diff 请用户确认 ↓ [满意] → 进入第三阶段，基于已确认 workflow.md 生成 workflow.json [还有修改] → 继续循环 ``` > **禁止事项**：不要为了提前展示流程图而先写一个简化 `workflow.json`。当前创建流程必须让 `workflow.md` 先落盘并完成确认，`workflow.json` 只能作为已确认 `workflow.md` 的机器执行产物。 --- ## 3. 第三阶段：生成完整 workflow.json（机器执行） 根据已确认的 `workflow.md` 生成严格可执行的 `workflow.json`。**生成前必须读取最新磁盘上的 `workflow.md`，并建议读取 [references/reference.md](references/reference.md)**。 ### 3.0 节点生成策略 - **主路径**：每个可执行步骤 → `type="python"` 节点，必须同时包含 `code`（执行逻辑）+ `description`（文档说明）。 - **兜底**：`logic` 节点仅在用户明确要求"不写代码"或快速原型时使用，运行时由 codegen 兜底。 ### 3.1 运行时硬约束 **顶层字段：** - `start` 必须等于某个 `nodes[i].id` - `nodes[].id` 必须唯一 - `name`/`description`（可选）用于工作流级别说明 - `version` 会被运行时忽略，不需要生成 **Node 约束**（对应 `flocks/workflow/models.py`）： - `python`：`code` 必须非空 - `logic`：`description` 必须非空 - **出边选择行为**（关键）：`python` → 所有出边触发；`logic`/`branch`/`loop` → 通过 `select_key` 取值做 label 匹配选边 - `join=true`：等待所有入边到齐再执行一次 **代码约束：** - 同步 `exec()` 模型，**严禁** `await`/`async def`/`async for`/`async with`。确保节点的代码要可以独立运行。 **Edge 约束：** - JSON 中用 `"from"` 而非 `"from_"`；`from`/`to` 引用存在的 node id；`order` ≥ 0。 - **新建 workflow 的每条 edge 必须包含非空 `mapping` 对象**。不要生成无 `mapping` 的 edge；不要只写 `const` 而省略 `mapping`。新建 workflow 默认启用 strict edge mapping，无 `mapping` 会在创建或运行时失败。 ### 3.2 映射规则 - `workflow.md` 每步对应一个节点，`id` 用 snake_case。 - md 中写的输出字段，必须在 `outputs[...]` 中体现。 - md 中 `Tool: xxx` 标记 → 对应节点 `description` 保留。 - 所有 edge 都必须写 `edge.mapping`，只映射下游节点实际需要的字段，避免全量 payload 传递。 - 下游节点如需 `tool.run(..., **inputs)`，用 `edge.mapping`/`edge.const` 规整输入到匹配工具参数形状；其中 `edge.mapping` 仍然必须非空。 - 若某条边只是控制流、下游不需要业务字段，也必须映射一个确定存在的小字段（如 `case_id`、`has_results`、`status`）；如果没有合适字段，让上游节点写出 `outputs["_edge_context"] = True`，并在该边映射 `{ "_edge_context": "_edge_context" }`。 - `branch`/`loop` 出边同样必须写 `mapping`。映射源可以来自该分支节点收到的输入 payload（例如 `search_text`、`case_id`、`has_results`），不要求来自 branch 节点自身输出。 - 详细 Mapping 指南见 [reference.md § Edge Mapping](references/reference.md#4-edge-mapping-详细指南)。 ### 3.3 分支/循环与 Join - **branch/loop 选边**：`bool` 值 label 用 `"true"`/`"false"`；`str` 值精确匹配；无命中回退到空 label 默认边。上游必须把 `select_key` 所需字段写入 payload。 - **分支汇合（强制）**： - 多入边且非互斥 → **必须** `join=true` - 判断互斥：所有入边来自同一 branch/loop 的不同 label 出边 - **昂贵节点保护**：含 `llm.ask()` 或 `tool.run('write', ...)` 的节点，禁止被两条非互斥路径直达，必须先经 join 节点 - 推荐模式：join 节点（python, `join=true`）归一化多分支输出 → 再传给后续步骤 - **嵌套工作流**：见 [references/composition.md](references/composition.md)。 ### 3.4 代码实现 **辅助函数：** | 函数 | 说明 | |------|------| | `tool.run(name, **inputs)` | 返回 `ToolResult.output`（类型 `Any`，**通常是字符串**），失败抛异常 | | `tool.run_safe(name, **inputs)` | **推荐**，返回 `{"success": bool, "text": str, "obj": Any, "error": str\|None}`，永不抛异常 | | `llm.ask(prompt)` | 调用 LLM，返回字符串 | | `get_path(path)` | payload 深层取值 | **⚠️ 返回值类型警告（常见 Bug 源）：** - **`tool.run()` 返回的是 `Any` 类型**，大多数工具返回的是**字符串**（格式化文本），**不是字典**。**严禁**直接对返回值调用 `.get()` 等字典方法。 - **`tool.run_safe()["obj"]` 也是 `Any` 类型**，可能是 `str`、`dict`、`list` 或 `None`。使用前**必须检查类型**。 - 如果工具返回的是 JSON 格式的字符串，需要用 `json.loads()` 解析后再作为字典操作。 ```python # ❌ 错误：直接在 tool.run() 返回值上调用 .get() result = tool.run('some_tool', ip=ip) value = result.get("key") # AttributeError: 'str' object has no attribute 'get' # ❌ 错误：假设 obj 一定是 dict result = tool.run_safe('some_tool', ip=ip) value = result["obj"].get("key") # obj 可能是 str，同样报错 # ✅ 正确：使用 text 做字符串操作 result = tool.run_safe('some_tool', ip=ip) outputs["text"] = result["text"] # text 永远是 str，安全 # ✅ 正确：需要结构化数据时，先检查 obj 类型 result = tool.run_safe('some_tool', ip=ip) obj = result["obj"] if isinstance(obj, dict): value = obj.get("key") elif isinstance(obj, str): import json try: parsed = json.loads(obj) value = parsed.get("key") if isinstance(parsed, dict) else None except json.JSONDecodeError: value = None ``` **`tool.run_safe()` 使用指南：** - 字符串拼接 / LLM prompt 插值 → `result["text"]`（永远是 `str`，最安全）