agentflow

# AgentFlow Workflow Orchestration 你负责把用户的目标落成一个可验证、可恢复、可观测的 AgentFlow workflow。先和用户确认流程边界，再生成或修改 workflow JSON；不要猜测 provider、workspace、session 或审批规则。 ## 1. 先向用户确认 至少确认这些信息： 1. **目标**：流程最终要产出什么，成功标准是什么。 2. **输入**：运行时需要哪些值，哪些是必填；敏感值不要写入 workflow 文件。 3. **节点**：哪些步骤由 shell 执行，哪些步骤由 agent 执行，哪些步骤必须人工审批。 4. **依赖**：节点之间的先后关系；没有依赖的节点可以并行执行，但当前 MVP 的 Runner 会按定义顺序推进可执行节点。 5. **工作目录**：每个 agent 使用哪个绝对或相对 workspace。 6. **会话**：是启动新 session，还是继续已有 session；已有 session 的 ID 从哪里来。 7. **失败策略**：超时、重试次数、失败后是否跳过下游节点。 8. **输出**：每个节点要写入哪些结构化 JSON 字段，哪些文件需要作为 artifact 保留。 如果用户还没有提供 session ID，应明确让该值成为 workflow input，或先使用 provider 的新 session command；不要把“当前目录最近的会话”当成确定绑定。 ## 2. 先查 CLI 契约 在编排前查询当前版本的命令和 workflow schema： ```bash agentflow --help agentflow run --help agentflow schema --all --format json agentflow schema workflow --format json ``` 优先使用长命令和 JSON 输出。通过管道调用时 CLI 默认返回 JSON；stdout 是数据，stderr 是诊断。 ## 3. Workflow 最小结构 ```json { "version": "1", "name": "release-review", "inputs": { "workspace": { "type": "path", "required": true }, "session_id": { "type": "string", "required": false } }, "nodes": [ { "id": "inspect", "type": "agent", "provider": "claude-code", "workspace": "{{ inputs.workspace }}", "session": "{{ inputs.session_id }}", "prompt": "检查当前变更，并把结论写入 AGENTFLOW_RESULT_FILE。" }, { "id": "approve", "type": "human", "needs": ["inspect"], "prompt": "请审批检查结论。" } ] } ``` ### 节点约定 - `shell`：使用 `command`，适合确定性的脚本和校验。 - `agent`：使用 `provider`、`prompt`、可选的 `workspace` 和 `session`。 - `human`：使用 `prompt`，运行后状态为 `waiting_human`，必须通过 CLI 决策才能继续。 - `needs`：声明依赖节点 ID。 - `when`：根据 `inputs` 或上游 `result` 做条件跳过。 - `retry.maxAttempts`：包含首次执行在内的最大尝试次数。 - `timeoutMs`：单次命令超时。 ## 4. Provider、workspace 和 session Provider 配置在 `.agentflow/providers.json` 或 `~/.agentflow/providers.json`，不放进 workflow： ```json { "providers": { "claude-code": { "kind": "claude-code", "binary": "claude" }, "codex": { "kind": "codex", "binary": "codex", "model": "gpt-5.6-sol", "sandbox": "workspace-write" } } } ``` 规则： - `claude-code` 和 `codex` 是内置 provider，provider 负责拼接各自 CLI 的启动和 resume 参数。 - `binary`、`model`、`sandbox` 等运行参数放在 provider 配置，不要写进 workflow。 - 自定义 agent 使用 `kind: "command"`，并提供 `command` 与 `resumeCommand` 模板。 - 有 `session` 时，内置 provider 会调用对应 CLI 的 resume 子命令；自定义 command provider 必须提供 `resumeCommand`。 - 有 session 但无法 resume 时，节点应失败并修复 provider 配置；不要静默启动新 session。 - Runner 使用节点的 workspace 作为进程工作目录，并设置 `AGENTFLOW_WORKSPACE`、`AGENTFLOW_SESSION_ID`。 - provider command 可以使用 `{{ prompt }}`、`{{ workspace }}`、`{{ session_id }}`、`{{ result_file }}`、`{{ run_id }}` 和 `{{ node_id }}`。 - Agent 也可以在结构化结果中返回 `session_id`，后续节点用 `{{ nodes.<id>.result.session_id }}` 继续该会话。 ## 5. 结构化结果契约 Agent 成功退出前必须向 `AGENTFLOW_RESULT_FILE` 写入一个 JSON object： ```json { "decision": "pass", "summary": "测试通过", "session_id": "optional-session-id" } ``` 不要把 stdout 或 stderr 当作节点输出。Runner 会把它们保存成日志 artifact，并只把结果文件解析为 `node.result`。下游节点通过 `{{ nodes.inspect.result.summary }}` 读取结果。 节点可以把上下游边界写成显式契约： ```json { "input": { "summary": "{{ nodes.inspect.result.summary }}" }, "inputSchema": { "type": "object", "required": ["summary"], "properties": { "summary": { "type": "string" } }, "additionalProperties": false }, "outputSchema": { "type": "object", "required": ["decision"], "properties": { "decision": { "type": "string" } } } } ``` `input` 在节点启动前渲染并校验，`outputSchema` 在结果文件解析后校验。契约使用 JSON Schema draft 2020-12；不要只依赖 prompt 描述字段。精确占位符会保留源值类型；不要通过 stdout/stderr 传递业务数据。 对于字段较多的流程，使用 Multigent 风格的字段声明： ```json { "id": "verify", "type": "agent", "needs": ["inspect"], "inputFields": [ { "name": "summary", "type": "string", "description": "检查摘要", "required": true } ], "outputFields": [ { "name": "verified", "type": "boolean", "description": "是否通过", "required": true } ], "inputMapping": { "summary": "$node.inspect.output.summary" } } ``` `outputFields` 在没有 `outputSchema` 时会生成严格 object schema，缺少必填字段或返回未知字段都会失败。`inputMapping` 支持 `$node.<id>.output.<field>`、`$output.<field>` 和 `$input.<field>`；没有显式 mapping 时，直接依赖节点之间同名字段会自动传递，显式 `input` 模板优先级最高。每个节点的渲染后输入会保存为 `.agentflow/runs/<run-id>/artifacts/<node-id>.input.json`，便于审计和重放。 ## 6. 校验和执行流程 ### 6.1 校验 ```bash agentflow workflow validate workflow.json --format json ``` 校验失败时，根据 JSON 错误里的 `code`、`message`、`suggestion` 修复后再试。不要跳过校验直接执行。 ### 6.2 预演 ```bash agentflow run start workflow.json \ --input workspace=/absolute/repo \ --input session_id=<session-id> \ --dry-run \ --format json ``` 退出码 `10` 表示 dry-run 成功；它不会创建或执行 run。检查预演中的节点顺序、workspace、session 和输入。 ### 6.3 正式执行 为可重试的启动使用稳定的幂等 key： ```bash agentflow run start workflow.json \ --input workspace=/absolute/repo \ --input session_id=<session-id> \ --idempotency-key <stable-request-key> \ --format json ``` 同一 workflow、输入和幂等 key 会复用已有 run；同 key 对应不同请求时退出码为 `5`。 ### 6.4 人工审批和恢复 当 run 返回 `waiting_human`： ```bash agentflow run get <run-id> --format json agentflow run approve <run-id> <node-id> --message "approved" --format json # 或 agentflow run reject <run-id> <node-id> --message "需要修改" --format json ``` 审批命令由 Runner 推进下游节点。不要让 agent 直接修改 `.agentflow/runs/<run-id>/state.json`，也不要让 agent 自己启动下一个节点。 如果进程被中断，先查看状态，再恢复： ```bash agentflow run resume <run-id> --dry-run --format json agentflow run resume <run-id> --format json ``` ## 7. 观测单次执行和历史效率 ### 单次 run ```bash agentflow run metrics <run-id> --format json ``` 返回： - run 总耗时和状态； - 每个节点的 `duration_ms`、`waiting_ms`、`attempts`； - `input_bytes`：发送给节点的渲染后 prompt 加结构化 input 字节数； - `output_bytes`：结构化 result 文件字节数； - `stdout_bytes` / `stderr_bytes`：日志字节数； - agent 的 workspace 和 session ID。 事件时间线和原始日志： ```bash agentflow run logs <run-id> --format ndjson agentflow run get <run-id> --format json ``` 运行数据默认保存在： ```text .agentflow/runs/<run-id>/ ├── workflow.json ├── state.json ├── events.jsonl └── artifacts/ ├── <node>.result.json └── logs/<node>.stdout.log └── logs/<node>.stderr.log ``` ### 历史聚合 ```bash agentflow run stats --format json agentflow run stats release-review --format json agentflow run stats release-review --by-version --format json agentflow run stats release-review --digest <workflow_digest> --format json ``` 聚合结果包括成功/失败/等待审批/取消数量，run 和节点耗时的平均值、P50、P95、最大值，以及输入、输出和日志字节总量。run 级别的 `duration_ms` 只统计已结束的 run，并拆成 `active_ms`（执行）和 `waiting_ms`（等人）两部分，所以还在等审批的 run 不会拉高平均值。 每个 run 都记录 `workflow_digest`：workflow 定义内容的哈希（只改格式不变，改了节点、契约或输入就会变）。`--by-version` 按版本分组、按首次运行时间排序，用来比较改流程前后的效果。 查询某个 workflow 的最新结果时用稳定命令，不要读取 `.agentflow/runs/*/state.json`： ```bash agentflow run latest release-review --status succeeded,failed --format json # 没有匹配时退出码 3 agentflow run list --workflow release-review --status waiting_human --format json ```用它回答“哪个节点最慢”“审批等待占多少时间”“输出是否异常膨胀”“失败是否集中在某个节点”。 ## 8. Web 可视化 需要和用户一起检查流程结构时，打开 workflow 图： ```bash agentflow web workflow.json ``` 需要检查一次实际执行时，打开 run 图： ```bash agentflow web --run <run-id> ``` Web 页面会显示节点依赖、节点类型、状态、耗时、attempts 和输出大小；run 页面会每两秒刷新一次状态。CLI 会打印本地 URL。可以用 `--port 0` 选择可用端口，用 `--host 0.0.0.0` 暴露到局域网时必须确认网络访问范围。 ## 9. 谁来启动 workflow workflow 本身是完整的，谁启动它只是入口。以下方式等价，按用户已有的东西选： - Claude Code 或 Codex 自己的定时任务唤醒 agent，agent 调 `agentflow run start <workflow.json> --idempotency-key <窗口标识>`，再用 `agentflow run latest <name>` 看结果； - 系统 cron / launchd / CI 调 `agentflow run start` 或 `agentflow trigger fire`； - `agentflow trigger serve` 常驻处理 cron 和 webhook（下文）； - 人直接在命令行启动。 幂等 key 用“这次唤醒对应的业务窗口”，例如 `release-2026-10-23-beta`，这样重复唤醒不会重复执行。 ### Cron 和 webhook trigger Trigger 配置放在 `.agentflow/triggers.json`，不是 workflow JSON。先检查再运行服务： ```bash agentflow trigger validate --format json agentflow trigger list --format json agentflow trigger serve --host 127.0.0.1 --port 7071 ``` `trigger serve` 内置 cron scheduler、webhook HTTP server 和持久化 run worker。每个事件先写入 `trigger-deliveries/`，再写入 `dispatch/<run-id>.json`，随后异步执行目标 workflow。进程重启会恢复仍处于 queued 的 dispatch。 如果使用系统 cron/launchd/Kubernetes CronJob，只调用一次 dispatcher： ```bash agentflow trigger fire nightly-review \ --scheduled-at 2026-10-13T02:00:00+08:00 \ --delivery-id nightly-review-20261013 \ --format json ``` Webhook 默认是 `POST` 到 trigger 的 `source.path`。使用 `source.auth` 配置 HMAC SHA-256 或 bearer token；签名 secret 只能来自环境变量。`delivery-id` 是幂等边界，重复投递会复用原 run。`policy.concurrency: "forbid"` 会在同一 trigger 有活动 run 时记录 skipped。 ## 10. 效率优化闭环 1. 先看 `run stats <workflow-name>`，定位 P95 高或失败率高的节点。 2. 再看对应 run 的 `run metrics`，区分 agent 执行时间、人工等待时间和重试时间。 3. 用 `run logs --format ndjson` 对照 `node.started`、`node.retrying`、`node.failed` 事件。 4. 根据字节数判断 prompt/result/log 是否过大；不要用 stdout 文本猜结构化输出。 5. 调整 workflow：拆分过大的 prompt、增加明确的 result schema、设置合理 timeout/retry、减少不必要的人审节点。 6. 重新 `workflow validate`、`run start --dry-run`，再用新的幂等 key 执行。 7. 用 `run stats <workflow-name> --by-version` 比较修改前后两个版本的 P95、`active_ms`、`waiting_ms`、失败数、attempts 和字节量是否改善。 ## 10. 与用户协作的交付格式 在提交 workflow 前，向用户说明： - workflow 的目标、输入和最终输出； - 每个节点做什么以及依赖关系； - 哪些节点会调用真实 agent、使用哪个 workspace/session； - 哪些节点需要人工审批； - 失败、重试、超时和取消如何处理； - 如何执行、如何审批、如何查看 metrics/stats； - 内置 provider 使用哪个 binary、model、sandbox 或 permission 设置；自定义 command provider 的命令模板是否需要按本机 CLI 版本调整。 执行后报告 run ID、最终状态、失败节点、耗时、attempts、输入/输出字节数，以及需要用户处理的审批节点。