external-workflow-api-spec.md 80 KB

工作流外部调用接口规范(External Workflow API Specification)

版本:v1.0 适用范围:第三方系统通过 HTTP 调用平台工作流,提交自定义输入、订阅节点级思考过程流、获取最终输出变量与工作空间产物。 文档基于现有 /api/workflows/{id}/run(SSE 同步流)与 /api/workflows/{id}/runs/{runId}/download 能力抽象出对外发布接口规范。


1. 设计目标

目标 说明
异步分离 提交运行与订阅流分离,客户端可延迟订阅、断线重连、批量并发
输入灵活 支持纯 JSON 变量、二进制文件、目录结构(保留层级路径)三种输入
全过程可观测 节点状态、思考流(thinking)、工具调用、错误事件全部以 SSE 事件推送
产物可获取 最终输出变量、工作目录 zip 均可通过稳定 URL 拉取
可认证 通过 X-API-Key 头部识别调用方,可绑定工作流白名单

2. 通用约定

2.1 基础路径

http(s)://<host>:<port>/api/v1
  • 默认端口:2438(与平台后端一致,见 application.yml: server.port)
  • 所有路径以 /api/v1 开头,与平台内部 /api/workflows、/api/run-history 隔离,便于独立鉴权

2.2 内容类型

场景 Content-Type
创建运行(仅 JSON 变量) application/json
创建运行(含文件/目录上传) multipart/form-data
订阅运行流(SSE) 响应 text/event-stream
查询运行结果 响应 application/json
下载工作空间 响应 application/zip

2.3 认证

所有 /api/v1/** 接口必须携带 API Key:

X-API-Key: <预共享密钥>
  • API Key 在 application.yml 中通过 external.api-keys 列表配置,可配置多组
  • 每条 Key 可选绑定 workflowIds 白名单(不填表示可调用所有工作流)
  • 缺失或不匹配 → HTTP 401 Unauthorized,响应体:

    { "code": "UNAUTHORIZED", "message": "API Key 缺失或无效" }
    

2.4 统一响应包络

非 SSE 接口统一返回:

{
  "code": "OK",
  "message": "",
  "data": { ... }
}

错误时:

{
  "code": "<ERROR_CODE>",
  "message": "<可读错误信息>",
  "data": null
}

SSE 接口(/stream)的认证与业务错误通过事件 payload(见 §5)传递,不使用 HTTP 状态码包络。

2.5 错误码

code HTTP 含义
OK 200 成功
UNAUTHORIZED 401 API Key 缺失或无权调用该工作流
VALIDATION_FAILED 400 入参不合法(变量缺失、文件越界等)
WORKFLOW_NOT_FOUND 404 工作流 ID 不存在
RUN_NOT_FOUND 404 运行 ID 不存在或已过期
GRAPH_INVALID 422 工作流图结构未通过校验(无输入/输出节点、不连通、有环)
RUN_ALREADY_STARTED 409 该运行已启动,不可重复 start
INTERNAL_ERROR 500 服务器内部错误(堆栈已脱敏)

2.6 标识符

名称 类型 说明
workflowId number 工作流主键(路径参数)
runId string 每次运行的唯一标识,8 位十六进制(如 a3f9b2c1),由服务端生成,贯穿工作目录、SSE 事件、下载 URL

3. 接口总览

# 方法 路径 用途
3.1 POST /api/v1/workflows/{workflowId}/runs 创建运行(提交输入与文件)
3.2 POST /api/v1/workflows/{workflowId}/runs/{runId}/start 启动执行(与 3.1 配合的两阶段模式)
3.3 GET /api/v1/workflows/{workflowId}/runs/{runId}/stream 订阅 SSE 流(思考过程、节点状态、完成事件)
3.4 GET /api/v1/workflows/{workflowId}/runs/{runId} 查询运行结果(最终变量、状态)
3.5 GET /api/v1/workflows/{workflowId}/runs/{runId}/workspace 下载工作空间 zip
3.6 GET /api/v1/workflows 列出当前 API Key 可调用的工作流

设计说明: 现有 /api/workflows/{id}/run 是「SSE 同步阻塞」模式(一次请求边执行边返回)。对外 API 改为「提交 → 启动 → 订阅」三段式,以满足「思考过程通过另一个接口获取」的诉求,并支持客户端重连。


4. 接口详细规范

3.1 创建运行

请求

POST /api/v1/workflows/{workflowId}/runs
X-API-Key: <key>
Content-Type: application/json  |  multipart/form-data

两种调用形态

形态 A:仅 JSON 变量(无文件)

POST /api/v1/workflows/12/runs
X-API-Key: demo-key
Content-Type: application/json

{
  "variables": {
    "topic": "大语言模型在企业中的应用",
    "maxTokens": 2048,
    "options": { "lang": "zh", "verbose": false }
  },
  "async": true
}

形态 B:JSON 变量 + 文件 / 目录

POST /api/v1/workflows/12/runs
X-API-Key: demo-key
Content-Type: multipart/form-data; boundary=----boundary

------boundary
Content-Disposition: form-data; name="payload"
Content-Type: application/json

{
  "variables": { "topic": "汇总报告" },
  "async": true
}
------boundary
Content-Disposition: form-data; name="files"; filename="report.md"
Content-Type: text/markdown

<文件二进制内容>
------boundary
Content-Disposition: form-data; name="files"; filename="data.csv"
Content-Type: text/csv
X-Relative-Path: inputs/2024/data.csv

<文件二进制内容>
------boundary
Content-Disposition: form-data; name="files"; filename="config.yaml"
Content-Type: application/yaml
X-Relative-Path: config/config.yaml

<文件二进制内容>
------boundary--

关键字段说明

字段 位置 类型 必填 说明
variables JSON object 否 自定义输入变量,会合并到工作流上下文(WorkflowContext.variables)
async JSON boolean 否 true(默认):返回 runId 但不自动启动,需调用 /start;false:自动启动并直接进入 SSE 流(等价于现有 /run 行为)
failStrategy JSON string 否 整体默认失败策略,覆盖节点级配置:abort(默认)| skip
ttlHours JSON number 否 工作空间保留时间,默认 720(30 天);超期清理
files multipart binary[] 否 二进制文件,可重复
X-Relative-Path part header string 否 该文件在工作目录中的相对路径;缺失时使用 filename;含 / 时自动创建目录结构(目录上传场景)

目录上传约定: 客户端在前端使用 <input webkitdirectory> 时浏览器原生提供 webkitRelativePath,通过该 header 透传到后端即可保留目录结构。后端会做 ZIP slip 防护(dest.normalize().startsWith(runDir) 校验)。

响应(async=true)

{
  "code": "OK",
  "data": {
    "runId": "a3f9b2c1",
    "workflowId": 12,
    "status": "CREATED",
    "createdAt": "2026-07-01T08:30:00Z",
    "links": {
      "start":  "/api/v1/workflows/12/runs/a3f9b2c1/start",
      "stream": "/api/v1/workflows/12/runs/a3f9b2c1/stream",
      "result": "/api/v1/workflows/12/runs/a3f9b2c1",
      "workspace": "/api/v1/workflows/12/runs/a3f9b2c1/workspace"
    }
  }
}

响应(async=false,立即启动)

直接以 SSE 流响应(Content-Type 切换为 text/event-stream),等价于 §3.3 的输出。


3.2 启动运行

仅当 async=true 创建的运行需要显式启动。

请求

POST /api/v1/workflows/12/runs/a3f9b2c1/start
X-API-Key: demo-key

无请求体。

响应

{
  "code": "OK",
  "data": {
    "runId": "a3f9b2c1",
    "status": "RUNNING",
    "streamUrl": "/api/v1/workflows/12/runs/a3f9b2c1/stream"
  }
}

启动后客户端应立即打开 /stream(§3.3)订阅事件;服务端会缓存最近 N 条事件用于断线重连补发。


3.3 订阅运行流(SSE)

GET /api/v1/workflows/{workflowId}/runs/{runId}/stream
X-API-Key: <key>
Accept: text/event-stream

响应:Content-Type: text/event-stream,按 SSE 协议推送事件,事件名与 payload 详见 §5。

断线重连

支持通过 Last-Event-ID 头部续传:

GET .../stream
Last-Event-ID: 42

服务端为每个事件分配单调递增的 id:,客户端重连时携带最后收到的事件 id,服务端补发其后所有事件。事件缓存窗口为每个运行 5000 条(超出丢弃最旧),运行结束后保留 25 小时(SseEventBus.MAX_EVENTS_PER_RUN=5000、TTL_AFTER_COMPLETION_MS=25h)。

关闭时机

  • 工作流成功完成:推送 workflow_complete 后关闭
  • 工作流出错:推送 workflow_error 后关闭
  • SSE 连接 24 小时超时(SseEventBus.SSE_TIMEOUT_MS=86_400_000L 兜底值;真正的长等待由 30 秒心跳 + 客户端重连机制保证)
  • 客户端主动断开:服务端停止推送但不影响后台执行

3.4 查询运行结果

GET /api/v1/workflows/{workflowId}/runs/{runId}
X-API-Key: <key>

响应

{
  "code": "OK",
  "data": {
    "runId": "a3f9b2c1",
    "workflowId": 12,
    "status": "SUCCESS",
    "startedAt": "2026-07-01T08:30:05Z",
    "completedAt": "2026-07-01T08:31:12Z",
    "outputs": {
      "status": 200,
      "message": "工作流执行成功",
      "data": {
        "summary": "本报告探讨了..."
      }
    },
    "nodes": [
      {
        "nodeId": "node-1",
        "nodeType": "userInput",
        "label": "用户输入",
        "status": "SUCCESS",
        "startedAt": "2026-07-01T08:30:05Z",
        "completedAt": "2026-07-01T08:30:05Z",
        "output": { "status": 200, "message": "处理成功", "data": { "topic": "..." } },
        "logsCount": 0,
        "logsUrl": "/api/v1/workflows/12/runs/a3f9b2c1/nodes/node-1/logs"
      },
      {
        "nodeId": "node-2",
        "nodeType": "agent",
        "label": "研究助手",
        "status": "SUCCESS",
        "output": { "status": 200, "message": "调用成功", "data": { "research": "..." } },
        "logsCount": 124,
        "logsUrl": "/api/v1/workflows/12/runs/a3f9b2c1/nodes/node-2/logs"
      }
    ],
    "workspace": {
      "fileCount": 12,
      "totalBytes": 456789,
      "downloadUrl": "/api/v1/workflows/12/runs/a3f9b2c1/workspace"
    }
  }
}

envelope 说明

outputs 与 nodes[].output 均为 envelope 结构({status, message, data}),服务端不解包:

  • status 是数字:200 成功 / 400 失败(NodeOutputEnvelope.STATUS_SUCCESS/STATUS_FAILED)
  • 业务字段在 data 内,前端渲染前需解包(判定规则:同时含 status 与 data 两键即视为 envelope)
  • 失败时优先展示 message 作为失败原因
  • 工作目录文件列表在 workspace 对象(fileCount/totalBytes/downloadUrl),不在 outputs 内;旧版注入 envelope.data 的 _workingDirFiles/_runId 已废弃

status 取值

  • CREATED:已创建未启动
  • RUNNING:执行中
  • SUCCESS:成功完成
  • FAILED:执行失败(error 字段填充)

节点级思考过程日志查询

logsUrl 指向:

GET /api/v1/workflows/{workflowId}/runs/{runId}/nodes/{nodeId}/logs

返回该节点执行期间产生的所有日志(thinking / tool_call / tool_result / info / error),用于回放或归档:

{
  "code": "OK",
  "data": {
    "nodeId": "node-2",
    "logs": [
      { "type": "THINKING", "timestamp": "...", "message": "首先我需要..." },
      { "type": "TOOL_CALL", "timestamp": "...", "message": "search_web", "detail": "{\"q\":\"...\"}" },
      { "type": "TOOL_RESULT", "timestamp": "...", "message": "search_web", "detail": "返回 3 条结果..." }
    ]
  }
}

3.5 下载工作空间

GET /api/v1/workflows/{workflowId}/runs/{runId}/workspace
X-API-Key: <key>

响应

HTTP/1.1 200 OK
Content-Type: application/zip
Content-Disposition: attachment; filename*=UTF-8''workflow-12-run-a3f9b2c1.zip

<zip binary>
  • 应用 .agentignore 规则过滤(默认排除 .hermes/、*.log、*.tmp、.DS_Store、Thumbs.db)
  • 404 时返回标准错误包络,code = RUN_NOT_FOUND

3.6 列出可调用工作流

GET /api/v1/workflows?page=1&size=20&search=
X-API-Key: <key>

响应

{
  "code": "OK",
  "data": {
    "total": 1,
    "items": [
      {
        "id": 12,
        "name": "研究报告生成",
        "description": "根据主题生成结构化研究报告",
        "inputs": [
          { "name": "topic", "type": "string", "required": true, "description": "研究主题" },
          { "name": "maxTokens", "type": "number", "required": false }
        ],
        "outputs": [
          { "name": "summary", "type": "string" }
        ]
      }
    ]
  }
}

inputs 与 outputs 由服务端扫描 graphData 中 userInput 与 output 节点的字段定义自动推导,便于第三方系统展示表单。


5. SSE 事件规范

⚠️ 本节为 v1.0 草案,与实际实现存在多处结构性偏差:run_started 事件实际不存在(见 §12 取消条款);node_stream.kind 实际 8 种而非 3 种;事件 payload 无 timestamp 字段;node_status 的 nodeType/nodeLabel 仅 RUNNING 时携带且字段名为 nodeLabel(v1.2,非本节的 label);新增 visit_recorded、planner_decision 事件;缓存窗口与 SSE 超时已调整(见 §3.3 修正后数值)。完整准确的事件规范以 §12(v1.1,覆盖本节)为准,本节仅保留作历史对照。

所有事件格式:

event: <事件名>
id: <单调递增整数>
data: <JSON 对象>

5.1 事件类型

事件名 触发时机 是否终止流
run_started 执行引擎启动 否
node_status 节点状态变更 否
node_stream 节点流式增量(thinking / tool_call / tool_result) 否
workflow_complete 工作流成功完成 是
workflow_error 工作流出错 是

5.2 run_started

{
  "runId": "a3f9b2c1",
  "workflowId": 12,
  "timestamp": "2026-07-01T08:30:05Z"
}

5.3 node_status

{
  "runId": "a3f9b2c1",
  "type": "node_status",
  "nodeId": "node-2",
  "nodeType": "agent",
  "label": "研究助手",
  "status": "RUNNING",
  "timestamp": "2026-07-01T08:30:06Z"
}

状态变为 SUCCESS 时额外携带:

{
  "...": "...",
  "status": "SUCCESS",
  "output": { "research": "..." },
  "completedAt": "2026-07-01T08:30:45Z"
}

状态变为 FAILED 时携带 error 字段。

5.4 node_stream(思考过程)

{
  "runId": "a3f9b2c1",
  "type": "node_stream",
  "nodeId": "node-2",
  "kind": "thinking",
  "content": "首先我需要搜索相关资料...",
  "toolName": null,
  "timestamp": "2026-07-01T08:30:07Z"
}
kind 含义 toolName
thinking LLM / Agent 的思考片段 null
tool_call 工具调用开始(参数摘要) 工具名
tool_result 工具调用返回(结果摘要) 工具名

5.5 workflow_complete

{
  "runId": "a3f9b2c1",
  "type": "workflow_complete",
  "success": true,
  "output": {
    "status": 200,
    "message": "工作流执行成功",
    "data": {
      "summary": "..."
    }
  },
  "workingDirFiles": ["report.md", "charts/fig1.png"]
}

5.6 workflow_error

{
  "runId": "a3f9b2c1",
  "type": "workflow_error",
  "success": false,
  "error": "节点 node-3 执行失败:LLM 调用超时",
  "failedNodeId": "node-3",
  "completedAt": "2026-07-01T08:31:00Z"
}

6. 调用时序示例

6.1 完整异步流程

客户端                                              服务端
  │                                                   │
  │── POST /runs (variables + files) ─────────────────▶│  创建运行 + 工作目录
  │◀────────── 200 { runId, links } ──────────────────│
  │                                                   │
  │── POST /runs/{runId}/start ───────────────────────▶│  启动执行
  │◀────────── 200 { status: RUNNING } ───────────────│
  │                                                   │
  │── GET /runs/{runId}/stream ───────────────────────▶│  订阅 SSE
  │◀── event: node_status (node-1 RUNNING) ───────────│
  │◀── event: node_status (node-1 SUCCESS) ───────────│
  │◀── event: node_status (node-2 RUNNING) ───────────│
  │◀── event: node_stream (node-2 thinking) ──────────│
  │◀── event: node_stream (node-2 thinking) ──────────│
  │◀── event: node_stream (node-2 tool_call) ─────────│
  │◀── event: node_stream (node-2 tool_result) ───────│
  │◀── event: node_status (node-2 SUCCESS) ───────────│
  │◀── event: node_status (node-3 SUCCESS) ───────────│
  │◀── event: workflow_complete ──────────────────────│  流结束
  │                                                   │
  │── GET /runs/{runId} ──────────────────────────────▶│  查询结果
  │◀────────── 200 { outputs, nodes[] } ──────────────│
  │                                                   │
  │── GET /runs/{runId}/workspace ────────────────────▶│  下载 zip
  │◀────────── 200 application/zip ───────────────────│

6.2 断线重连

客户端在事件 id=42 处断开,重连:
  │── GET /runs/{runId}/stream ───────────────────────▶│
  │   Last-Event-ID: 42                                 │
  │◀── event: node_status (重发 id=43..) ──────────────│

6.3 同步阻塞模式(async=false)

适合简单集成,一次请求直接返回 SSE 流:

  │── POST /runs (async=false) ───────────────────────▶│
  │◀── event: node_status ... ────────────────────────│
  │◀── event: node_stream ... ────────────────────────│
  │◀── event: workflow_complete ──────────────────────│

7. 限制与约束

项目 限制 说明
单次执行超时 无固定引擎超时 v1.0 设想的 MAX_EXECUTION_SECONDS=300 未实施;AskFor 等待不超时失败,Hermes 节点 readTimeout 24 小时兜底
SSE 连接超时 24 小时 SseEventBus.SSE_TIMEOUT_MS=86_400_000L;真正长等待由 30 秒心跳 + 客户端重连保证
单文件大小 100 MB 超过建议改用对象存储预签名 URL
单次上传文件数 50 multipart part 数上限
总请求体大小 500 MB Spring spring.servlet.multipart.max-request-size
工作目录保留 默认 30 天 通过创建运行时 ttlHours 调整
并发运行 受 CPU 核数约束 引擎线程池 fixedThreadPool(processors)
工作流图校验 必须有 userInput + output 节点且通过中间节点连通 详见 §8

8. 工作流图结构校验

服务端在 /runs 创建时或 /start 启动时会校验 graphData:

  1. 至少 1 个 userInput 类型节点
  2. 至少 1 个 output 类型节点
  3. 从任一 userInput 到任一 output 必须存在有向路径,且路径上至少包含 1 个中间节点(直接 input → output 边不满足)
  4. DAG 无环

不满足时返回 GRAPH_INVALID(HTTP 422),错误信息指明具体原因。


9. curl 调用示例

9.1 纯 JSON 输入

curl -X POST https://host/api/v1/workflows/12/runs \
  -H "X-API-Key: demo-key" \
  -H "Content-Type: application/json" \
  -d '{
    "variables": { "topic": "AI 在医疗领域的应用" },
    "async": true
  }'

9.2 JSON + 文件 + 目录

curl -X POST https://host/api/v1/workflows/12/runs \
  -H "X-API-Key: demo-key" \
  -F 'payload={"variables":{"topic":"汇总"},"async":true};type=application/json' \
  -F 'files=@report.md' \
  -F 'files=@data.csv;type=text/csv' \
  -H 'X-Relative-Path: inputs/2024/data.csv' \
  -F 'files=@config/config.yaml;type=application/yaml'

注:curl 的 X-Relative-Path 是自定义 part header,后端在解析 multipart 时需读取该头部;如果客户端无法发送自定义 part header,可改为在 payload.inputs.uploadedFiles 中显式提供路径列表,并使用默认的 filename。

9.3 启动 + 订阅

# 启动
curl -X POST https://host/api/v1/workflows/12/runs/a3f9b2c1/start \
  -H "X-API-Key: demo-key"

# 订阅流
curl -N -H "X-API-Key: demo-key" \
  https://host/api/v1/workflows/12/runs/a3f9b2c1/stream

9.4 查询结果 + 下载

curl -H "X-API-Key: demo-key" \
  https://host/api/v1/workflows/12/runs/a3f9b2c1

curl -H "X-API-Key: demo-key" \
  https://host/api/v1/workflows/12/runs/a3f9b2c1/workspace \
  -o workspace.zip

10. 与现有内部接口的映射关系

外部接口 内部接口 备注
POST /api/v1/.../runs (新增) 复用 WorkflowRunDirManager.createRunDir + 文件写入逻辑(现有 WorkflowController:165-192)
POST /api/v1/.../runs/{runId}/start WorkflowEngine.executeAsync 异步触发
GET /api/v1/.../runs/{runId}/stream SseEmitter + WorkflowRunEvent 复用现有事件工厂
GET /api/v1/.../runs/{runId} RunHistoryController:62 扩展为对外字段
GET /api/v1/.../runs/{runId}/workspace WorkflowController:205 复用 zip 打包 + .agentignore 过滤
节点日志 (新增) 从 WorkflowRunNode.logs 反序列化

实现建议:新增 ExternalApiController(路径 /api/v1/**)与独立 ExternalApiAuthFilter(解析 X-API-Key),不污染现有 Controller,便于后续按版本演进。


二、v1.1 集成方式补充(基于实施后的最新状态)

版本:v1.1(补充章节,2026-08-12) 适用范围:在 v1.0 接口契约基础上,补充实施后发现的差异、完整运行结果展示规范、AskFor 交互、断线重连策略与 JS 参考实现,便于业务方完整复刻平台自有的运行体验(实时状态、思考过程、待办清单、工具调用、执行日志、最终结果,以及交互式 AskFor 与断线重连)。 兼容性:v1.0 的 §1-§10 仍然有效;v1.1 仅补充,不破坏 v1.0 接口契约。


11. v1.0 与 v1.1 实施状态差异

11.1 主要差异表

维度 v1.0 描述 v1.1 实施状态 影响 / 处理建议
路径参数 {workflowId}(数字 id) {workflowName}(kebab-case 名字,兼容数字 id 兜底) 业务方 URL 用工作流名更直观,避免主键暴露
事件缓存窗口 1000 条 / 1 小时 5000 条 / 25 小时(SseEventBus.MAX_EVENTS_PER_RUN=5000、TTL_AFTER_COMPLETION_MS=25h) 长任务断线重连窗口显著加大
SSE 超时(外部 /api/v1) 30 分钟 24 小时(外部与内部端点共用 SseEventBus.SSE_TIMEOUT_MS=86_400_000L;ExternalApiController 内 30 分钟常量为废弃死代码,未被引用) 长任务基本不被切流,但业务方仍应实现重连
SSE 超时(内部 /api/workflows、/api/chat) 30 分钟 24 小时 三套端点一致
run_started 事件 §5.1 列出 代码中不存在,第一个事件通常是 node_status(input, RUNNING) 业务方不应依赖此事件
node_stream.kind 3 种(thinking/tool_call/tool_result) 实际 8 种(新增 status/step/tool_progress/todo_update/clarify_request) 业务方需处理全部 8 种才能完整展示
新增事件 未提及 visit_recorded、planner_decision(AGENTIC 必发) 业务方需识别并消费
AskFor(用户确认) 未提及 三套 API 均已实现:外部 API POST /api/v1/runs/{runId}/resume + GET /api/v1/runs/{runId}/state 已开放(v1.2,2026-08-14) 含 AskFor 的工作流可直接走外部 API,不再依赖内部 API 过渡方案
心跳帧 未提及 30 秒 :keep-alive SSE 注释帧(不触发 onmessage,仅保活) 业务方 SSE 客户端无需特殊处理
工作目录元数据 注入 envelope 的 _workingDirFiles/_runId 两键已废弃移除;文件列表改为 workflow_complete 事件顶层字段 workingDirFiles,查询接口在 workspace 对象 业务方从事件顶层 / workspace 取,而非 envelope.data

11.2 接口清单更新(v1.1)

用途 方法 内部 API 外部 API(/api/v1) Chat API(/api/chat)
创建运行 POST /api/workflows/{name}/run(同步 SSE) /api/v1/workflows/{name}/runs(异步,两阶段) /api/chat/sessions/{sid}/runs(对话场景)
启动运行 POST —(创建即启动) /api/v1/workflows/{name}/runs/{runId}/start —(创建即启动)
SSE 订阅(重连续接) GET /api/workflows/runs/{runId}/stream /api/v1/workflows/{name}/runs/{runId}/stream /api/chat/runs/{runId}/stream
查询运行状态 + 当前暂停点 GET /api/workflows/runs/{runId}/state /api/v1/runs/{runId}/state /api/chat/runs/{runId}/state
提交 AskFor 答案 POST /api/workflows/runs/{runId}/resume /api/v1/runs/{runId}/resume /api/chat/runs/{runId}/resume
查询运行详情 GET /api/run-history/runs/{id}(数据库主键 Long,非 runId 字符串) /api/v1/workflows/{name}/runs/{runId} —
下载工作空间 GET /api/workflows/{name}/runs/{runId}/download /api/v1/workflows/{name}/runs/{runId}/workspace —

三套 SSE 端点背后共用同一个 SseEventBus,事件结构完全一致,业务方按场景选择 API 前缀即可。 注意:外部 API 的 /state、/resume 路径不带 workflows/{name} 段(/api/v1/runs/{runId}/...)——runId 全局唯一即可定位,且避免页面刷新后 workflowName 丢失导致无法提交。

11.3 业务方接入路径建议

  • 简单集成(仅查看结果、不需要交互式 clarify):使用外部 API(/api/v1),按 v1.0 §6.1 三段式 创建→启动→订阅 即可
  • 完整集成(含交互式 AskFor、长任务、断线重连):外部 API(/api/v1)已具备 /state + /resume 与 24h SSE 超时,含 AskFor 的工作流可直接走外部 API;运行历史回放、会话管理等平台自有能力仍需内部 API(/api/workflows 或 /api/chat)
  • 跨域嵌入:CORS 已对 /api/** 全开放(http://*:*、https://*:*),业务方前端可直接 fetch;若需要 cookie 共享,建议同源代理

12. 完整 SSE 事件规范(v1.1 覆盖 §5)

12.1 事件类型清单(6 种 type)

type 触发时机 终止流 关键携带字段
node_status 节点开始 / 结束 否 nodeId、visitUuid、status、output、error、logs[]、contextSnapshot;RUNNING 时另携带 nodeType、nodeLabel(v1.2,外部集成方无画布数据时据此直接建卡片识别节点类型/显示名,终态事件不携带)
node_stream 节点流式增量(按 kind 进一步细分) 否 nodeId、visitUuid、kind、content、toolName
visit_recorded Visit 持久化(DAG / AGENTIC 都会推) 否 visit: WorkflowRunVisit
planner_decision AGENTIC 模式 AI 决策 否 decision: PlannerDecision
workflow_complete 工作流成功 是 output、runId、workingDirFiles、success: true
workflow_error 工作流出错 是 error、success: false

取消条款:v1.0 §5.1 列出的 run_started 事件实际不存在,业务方不应依赖。

12.2 node_stream.kind 完整清单(8 种)

kind 含义 content toolName UI 路由
thinking LLM / Agent 思考片段(增量累加) 思考文本 null 思考过程面板(追加到 streamingThinking)
status 实时状态文案(覆盖式) 状态文本 null 实时状态面板(覆盖 currentStatus)
step LLM 调用轮次进度 {iteration: N} JSON null 实时状态面板(显示"第 N 轮 LLM 调用")
tool_progress 工具细粒度进度 {name, event_type, ...} JSON event_type 实时状态面板(event_type=tool.started 时显示"正在调用 X")
tool_call 工具调用开始 参数摘要 工具名 工具调用面板(新建 currentTool)
tool_result 工具调用结果 结果文本 工具名 工具调用面板(配对填入 result)
todo_update 待办清单更新(全量替换) 完整 todos JSON "todo" 待办清单面板(覆盖 todos 数组)
clarify_request 请求用户确认 {clarify_id, question, choices} JSON null AskforPanel(挂起全局 pendingAsk)

12.3 通用字段结构

所有事件共享以下字段(按需出现):

{
  "runId": "a3f9b2c1",
  "type": "node_status | node_stream | visit_recorded | planner_decision | workflow_complete | workflow_error",
  "nodeId": "node-2",
  "visitUuid": "uuid-abc-123",
  "nodeType": "agent",
  "nodeLabel": "研究助手",
  "status": "RUNNING | SUCCESS | FAILED | SKIPPED",
  "kind": "thinking | status | step | tool_progress | tool_call | tool_result | todo_update | clarify_request",
  "content": "...",
  "toolName": "search_web",
  "output": { "status": 200, "message": "...", "data": {} },
  "error": null,
  "success": true,
  "logs": [],
  "contextSnapshot": {},
  "workingDirFiles": ["report.md"],
  "visit": {},
  "decision": {}
}

12.4 visitUuid 路由语义(关键)

  • AGENTIC 模式:同一节点可被访问多次(上限 50 次),每次访问生成独立 visitUuid
  • DAG 模式:每个节点访问一次,visitUuid 仍存在,与 nodeId 一一对应
  • 路由规则:所有 node_stream 与 node_status 事件必须按 visitUuid 路由到对应 Visit 卡片,不能按 nodeId 合并(否则同节点多次访问的思考过程会混在一起)
  • visitUuid 缺失时(旧版兼容):fallback 到 nodeId 取最近一条 Visit

12.5 visit_recorded 事件(AGENTIC 必发、DAG 也会推)

每次节点访问完成后,后端推一条 visit_recorded 事件,携带完整的 WorkflowRunVisit 对象(含 visitUuid、iterSeq、fromVisitUuid、fromNodeId、variablesJson、outputJson、logsJson、decisionJson 等)。业务方应将此事件作为权威元数据,覆盖前端临时创建的 Visit 字段:

{
  "type": "visit_recorded",
  "visit": {
    "visitUuid": "uuid-abc",
    "nodeId": "node-2",
    "nodeType": "agent",
    "nodeLabel": "研究助手",
    "iterSeq": 1,
    "fromVisitUuid": null,
    "fromNodeId": null,
    "variablesJson": "{...}",
    "outputJson": "{...}",
    "logsJson": "[...]",
    "decisionJson": "{...}",
    "status": "SUCCESS",
    "errorMsg": null,
    "sortOrder": 2,
    "offCanvas": false
  }
}

12.6 planner_decision 事件(AGENTIC 专属)

每次 Planner LLM 决策后会推一条事件,业务方可据此展示"AI 决策链":

{
  "type": "planner_decision",
  "decision": {
    "next": "node-3",
    "reason": "用户已确认研究方向,下一步进入搜索节点",
    "isFinal": false
  }
}

12.7 心跳帧

后端每 30 秒推送 SSE 注释帧 :keep-alive(不是 event),仅用于保活,不会触发浏览器 onmessage。业务方的 SSE 客户端无需特殊处理。

12.8 事件终止条件

仅以下两种事件会触发服务端关闭 SSE 流:

  • workflow_complete:成功完成
  • workflow_error:执行出错

其他事件后流仍保持开启,直到 24 小时超时(外部与内部 API 一致,SseEventBus.SSE_TIMEOUT_MS=86_400_000L)。


13. 运行结果展示规范(前端交互逻辑)

业务方需要复刻的运行结果交互由 5 个面板 + 1 个最终输出区组成。所有面板均按 visitUuid 维度独立组织数据,每次节点访问(AGENTIC 模式下同一节点可被多次访问)都拥有独立面板组。

13.1 Visit 数据模型

每个面板组对应一次节点访问,最小数据集如下:

字段 类型 说明
visitUuid string 访问唯一 ID(主键,来自事件的 visitUuid 字段)
runId string 运行 ID
nodeId string 节点 ID
nodeType string 节点类型(start/llm/condition/...)
nodeName string 节点名称
status enum running / completed / failed / skipped
startedAt number 开始时间戳(ms)
endedAt number 结束时间戳(ms)
liveStatus string 实时状态文本(来自 status kind)
thinking string 累积的思考过程(追加来自 thinking kind)
todoList array 待办清单(覆盖式更新,来自 todo_update kind)
toolCalls array 工具调用记录数组(见下)
logs array 综合日志(step/tool_progress 等合并)
pendingAsk object|null AskFor 待确认信息(见 §14)
output object|null 最终输出(envelope 结构)
error object|null 错误信息

toolCalls 数组单元素结构(SSE 的 tool_call/tool_result 事件不携带 callId,前端按同 Visit 内到达顺序配对:tool_result 关闭最早未关闭的同名调用;Hermes 工具串行执行,顺序配对安全):

{
  "toolName": "web_search",
  "arguments": { "query": "..." },
  "status": "running",
  "result": null
}

13.2 五大面板数据来源与更新规则

面板 数据来源(事件/kind) 更新规则 渲染方式 节点完成后
① 实时状态 node_stream.kind=status 覆盖式(每次替换) 单行文本(loading 动画) 隐藏(不留痕迹)
② 思考过程 node_stream.kind=thinking 追加式(按行 append) markdown 渲染,支持流式光标 折叠(默认收起)
③ 待办清单 node_stream.kind=todo_update 覆盖式(整体替换) 列表(✓/○ 状态图标) 折叠
④ 工具调用 kind=tool_call + tool_progress + tool_result 按到达顺序配对(事件无 callId,tool_result 关闭最早未关闭的同名调用) 卡片式(参数+结果) 折叠
⑤ 执行日志 kind=step + tool_progress(部分) 追加式 时间戳 + 文本行 折叠
最终输出 node_status.status=SUCCESS(终态) + output 字段 一次性写入 markdown 渲染(代码高亮) 始终展开

13.3 envelope 输出解包

节点 output 字段使用统一的 envelope 结构:

{
  "status": 200,
  "message": "人类可读消息",
  "data": { /* 节点实际产出 */ }
}

status 是数字:200 成功 / 400 失败(NodeOutputEnvelope.STATUS_SUCCESS/STATUS_FAILED)。工作流最终输出(workflow_complete.output 与查询接口 outputs)同样带此壳。

业务方需要实现的解包工具函数:

// 判断是否为 envelope 结构(必须同时含 status 与 data 两键)
function isEnvelope(obj) {
  return obj && typeof obj === 'object' && !Array.isArray(obj)
    && 'status' in obj && 'data' in obj
}

// 解包:失败时返回 message,成功时返回 data(无 data 则 message)
function unwrapNodeOutput(output) {
  if (!output) return ''
  if (typeof output !== 'object') return String(output)
  if (!isEnvelope(output)) return output  // 兼容直接返回数据的老节点
  if (Number(output.status) === 400) return output.message || '执行失败'
  return output.data != null ? output.data : (output.message || '')
}

// 提取状态(用于染色:失败红、成功绿)
function nodeOutputStatus(output) {
  if (!isEnvelope(output)) return 'completed'
  return Number(output.status) === 400 ? 'failed' : 'completed'
}

// 提取顶部消息(不论成败)
function nodeOutputMessage(output) {
  if (!isEnvelope(output)) return ''
  return output.message || ''
}

13.4 面板折叠策略

为避免运行结束后页面被大量历史信息淹没,统一采用「完成后自动折叠」策略:

时机 行为
节点 running 中 实时状态展开;思考/工具/日志展开
节点 completed/failed 实时状态 隐藏;其余面板 折叠;最终输出 展开
用户点击折叠头 切换展开/折叠,并记住状态(同 run 内不重置)

折叠头应显示节点名称 + 节点类型 + 状态徽章 + 耗时,例如:

▼ [LLM] 文档搜索节点  ✓ 完成 · 3.2s

13.5 多访问(multi-visit)渲染

AGENTIC 模式下同一节点可能被访问多次(如循环、回访),渲染规则:

  1. 每次 visit 独立一个面板组,纵向堆叠(不合并、不覆盖)
  2. 折叠头追加访问序号:▼ [LLM] 文档搜索节点(第 2 次)
  3. 同一节点不同 visit 的颜色/图标保持一致,便于关联
  4. visit 顺序由 visit_recorded 事件的 visit.sortOrder 决定;收到 sortOrder 前按事件到达顺序临时排列,收到后重排

13.6 markdown 渲染要求

思考过程与最终输出使用 markdown 格式,业务方需配置:

  • GFM:表格、删除线、任务列表
  • 代码高亮:highlight.js(推荐 highlight.js/lib/common 子集,30+ 语言)
  • 数学公式:KaTeX(如业务场景无公式可省略)
  • XSS 防护:DOMPurify sanitize
  • 行内代码与代码块风格一致,代码块使用深色背景 + 圆角 + 横向滚动

参考实现见本仓库 frontend/src/utils/markdown.js(约 80 行)。


14. AskFor 交互规范(暂停-恢复)

AskFor 是工作流运行中向用户提问、等待选择或输入的暂停-恢复机制。该机制是「实时确认」面板的核心数据源。

14.1 触发条件

当节点(通常是 LLM 节点)需要用户输入时,工作流会:

  1. 通过 node_stream.kind=clarify_request 推送问题内容
  2. 整条工作流线程阻塞(Hermes Bridge 在 clarifyId 上 wait()),不再推进任何节点
  3. SSE 流保持开启,业务方前端展示问题面板
  4. 用户作答后业务方调用 /resume 提交答案
  5. 后端 notify() 唤醒阻塞线程,工作流继续推进

14.2 clarify_request 事件结构

{
  "type": "node_stream",
  "runId": "run-...",
  "nodeId": "node-2",
  "visitUuid": "visit-...",
  "kind": "clarify_request",
  "data": {
    "clarifyId": "clarify-abc123",
    "question": "请确认要使用的数据集版本",
    "options": [
      { "label": "v1.0(生产)", "value": "v1.0" },
      { "label": "v2.0(测试)", "value": "v2.0" }
    ],
    "allowCustom": true,
    "placeholder": "或输入其他版本号",
    "multiple": false
  }
}

字段说明:

字段 必填 说明
clarifyId ✓ 问题的唯一 ID,提交答案时必须携带
question ✓ 问题文本(支持 markdown)
options ✗ 预设选项数组(无则纯文本输入)
options[].label ✓ 选项展示文本
options[].value ✓ 选项值
allowCustom ✗ 是否允许自定义输入(默认 false)
placeholder ✗ 自定义输入框 placeholder
multiple ✗ 是否允许多选(默认 false)

14.3 pendingAsk 前端数据结构

业务方收到 clarify_request 后,应将其挂到对应 visit 的 pendingAsk 字段:

// 在 SSE 事件路由中
if (kind === 'clarify_request') {
  visit.pendingAsk = {
    clarifyId: data.clarifyId,
    question: data.question,
    options: data.options || [],
    allowCustom: !!data.allowCustom,
    placeholder: data.placeholder || '',
    multiple: !!data.multiple,
    submittedAt: null,         // 用户作答时间(提交后写入)
    submittedValue: null       // 用户作答内容(提交后写入)
  }
  // 触发响应式更新,UI 显示问题面板
}

14.4 POST /api/v1/runs/{runId}/resume — 提交答案

已实施(v1.2,2026-08-14)。路径不带 workflowName 段:runId 全局唯一即可定位,且避免页面刷新后 workflowName 丢失导致无法提交。响应包络与错误码遵循 §14.6。

请求:

POST /api/v1/runs/{runId}/resume
X-API-Key: <key>
Content-Type: application/json

{
  "clarifyId": "clarify-abc123",
  "answer": "v2.0"
}
  • answer 字段(string):
    • 选择选项:该选项的 value(平台 AskFor 面板为单选,无多选场景)
    • 自定义输入:用户输入的文本

响应:

{
  "code": "OK",
  "message": "ok",
  "data": {
    "accepted": true,
    "runId": "run-...",
    "clarifyId": "clarify-abc123"
  }
}
  • accepted=false 仅表示答案已落库但 Bridge 唤醒失败(或 hermes.enabled=false 时 Bridge 未启用);clarifyId 不存在、已被回答等错误场景直接返回对应 HTTP 错误码(详见 §14.6)

关键约束:

  • 同一 clarifyId 只能提交一次,重复提交返回 409
  • 提交后 pendingAsk 应在 UI 上立即置为「已提交」状态(不可重复点击)
  • 无超时:后端不限制思考时间,业务方也不应在 UI 上设置倒计时
  • 如果运行已结束(complete/error),/resume 返回 410 Gone

14.5 GET /api/v1/runs/{runId}/state — 状态恢复

已实施(v1.2,2026-08-14)。轻量设计:只返回运行状态 + 当前未答暂停点,不回放 visits 明细——节点过程数据由 SSE 事件缓存(5000 条 / 25 小时)通过 Last-Event-ID 续传补齐,不重复建设重放接口。

刷新页面或断线重连时,业务方应优先调用 /state 拉取状态与暂停点,恢复 AskFor 面板后再接 SSE 续传:

响应:

{
  "code": "OK",
  "message": "",
  "data": {
    "runId": "abcd1234",
    "status": "RUNNING",
    "currentPause": {
      "clarifyId": "clar-5678",
      "nodeId": "node-2",
      "question": "请确认要使用的数据集版本",
      "choices": ["v1.0", "v2.0"],
      "createdAt": "2026-08-14T10:00:00"
    }
  }
}
  • status 取值:CREATED(已注册未启动)/ RUNNING / SUCCESS / FAILED(大写枚举)
  • currentPause:当前 UNANSWERED 状态的最新暂停点;无暂停或运行未启动时为 null
  • choices 已解析为字符串数组(区别于内部 API 返回 JSON 字符串需前端自行解析;解析失败兜底空数组)
  • createdAt 为 ISO_LOCAL_DATE_TIME 格式(如 2026-08-14T10:00:00)

/state 配合 SSE Last-Event-ID 续传构成断线重连的关键恢复路径,详见 §15。

14.6 AskFor 错误码

HTTP code 场景 前端处理
400 VALIDATION_FAILED runId / clarifyId 格式非法,或缺请求体、缺 clarifyId 提示参数错误
404 RUN_NOT_FOUND runId 不存在,或 API Key 无该 run 所属工作流的访问权限(统一 404,不向未授权方泄露 runId 存在性) 提示"运行不存在",跳转列表
410 RUN_ALREADY_FINISHED 运行已 SUCCESS / FAILED 提示"运行已结束",转查看模式
404 CLARIFY_NOT_FOUND clarifyId 不属于该 run、不存在,或运行尚在 CREATED 阶段未启动 提示"问题已过期",刷新 /state
409 CLARIFY_ALREADY_ANSWERED 同一 clarifyId 已提交过 忽略(视为成功),刷新 /state
400 CLARIFY_ANSWER_INVALID 预留:answer 类型不匹配。当前 answer 恒为 string,此码暂未启用 提示校验错误

14.7 端到端时序图

后端                          业务方前端                    用户
 │                              │                            │
 │   ──node_stream(clarify)──▶  │                            │
 │                              │   ──显示问题面板──────────▶ │
 │   (线程阻塞,SSE 保活)        │                            │
 │                              │                            │
 │                              │   ◀──用户选择/输入───────── │
 │                              │                            │
 │   ◀──POST /resume(answer)──  │                            │
 │   ──200 OK─────────────────▶ │                            │
 │   (唤醒线程,继续执行)        │                            │
 │                              │                            │
 │   ──node_stream(status)───▶  │                            │
 │   ──node_stream(thinking)─▶  │                            │
 │   ──node_status(completed)▶  │                            │

15. 断线重连规范

15.1 总体策略

业务方应采用「SSE 优先 + REST 兜底」的迭代式重连策略:

  1. SSE 断开后立即重连(指数退避,最多 5 次)
  2. 5 次仍失败则降级为 /state 轮询(30 秒/次)
  3. 任意时刻 SSE 重连成功 → 取消轮询,回到 SSE 模式
  4. 运行结束(complete/error)后停止一切重连

15.2 重连期间 UI 表现

状态 UI 表现
SSE 正常 实时面板正常更新
SSE 断开,重连中(< 5 次) 顶部显示"重新连接中..."横幅,已有面板保持显示
重连失败 5 次,进入轮询 顶部显示"网络异常,正在降级同步",每 30s 静默刷新
任意时刻 SSE 恢复 横幅消失,回到实时模式
运行已结束 横幅消失,停止重连/轮询

15.3 SSE 重连 URL 格式

服务端 SSE 端点支持 Last-Event-ID 头部自动续传:

GET /api/v1/runs/{runId}/stream
Accept: text/event-stream
Last-Event-ID: <最后收到的 event id>

服务端会从缓存(默认 5000 条/25 小时)中重放缺失事件,业务方无需自己补齐。

提示:浏览器原生 EventSource 会自动携带 Last-Event-ID,无需手动处理;使用 fetch + ReadableStream 实现 SSE 时需手动添加。

15.4 /state 兜底窗口

  • 何时调用:5 次 SSE 重连失败后,立即调用一次 /state,之后每 30 秒轮询一次
  • 数据合并规则:
    • visits 数组整体覆盖前端状态(不合并、不 diff)
    • pendingAskGlobal 直接覆盖到全局状态
    • recentEvents 用于增量补充 SSE 期间错过的事件(前端按时间戳去重)
  • 何时停止:
    • status === 'completed' || 'failed' → 停止轮询,进入终态
    • SSE 重连成功 → 停止轮询

15.5 重连算法伪代码

function startReconnectLoop(runId) {
  let sseAttempts = 0
  let polling = false

  function trySSE() {
    if (runFinished()) return
    sseAttempts++
    openSSE(runId, {
      onOpen: () => { sseAttempts = 0; stopPolling(); hideBanner() },
      onEvent: handleSSEEvent,
      onError: () => {
        if (sseAttempts < 5) {
          setTimeout(trySSE, backoff(sseAttempts))  // 1s, 2s, 4s, 8s, 16s
        } else {
          startPolling()
        }
      }
    })
  }

  function startPolling() {
    if (polling) return
    polling = true
    showBanner('网络异常,正在降级同步')
    pollOnce()  // 立即执行一次
  }

  async function pollOnce() {
    if (!polling || runFinished()) return
    const state = await fetchState(runId)
    applyState(state)
    setTimeout(pollOnce, 30000)
  }

  function stopPolling() {
    polling = false
  }

  trySSE()
}

15.6 重连期间的事件去重

由于 /state 返回 recentEvents 与 SSE 重放可能产生重叠,业务方应:

  • 为每个事件维护 eventId(SSE 头部 id: 字段,全局唯一递增)
  • 维护一个长度为 500 的 LRU 已处理 eventId 集合
  • 事件到达时先查集合,已存在则丢弃

16. JavaScript 参考实现

⚠️ 本节为 v1.0 时期的参考代码,与当前实现存在多处偏差,不再建议直接复制使用,主要包括:

  • 成功判断应为字符串 resp.code === 'OK'(而非数字 0)
  • 浏览器直连必须用 fetch + ReadableStream 手动解析 SSE(原生 EventSource 无法携带 X-API-Key 请求头)
  • 断线续传应为 Last-Event-ID 请求头(而非 query 参数);SSE 事件 payload 为扁平结构(增量在 content 字符串字段,status 为大写枚举)
  • 外部 API 未开放 AskFor 的 /resume、/state(仅内部 API 有)

最新可运行的参考实现见技能目录 skills/agent-management-external-workflow/references/js-implementation.md(配套接口契约 external-workflow-api.md 与展示规范 run-history-ui.md),本节仅保留作历史对照。

以下实现使用浏览器原生 EventSource + fetch,无任何依赖,可直接放入业务方项目。框架无关(Vue/React/原生皆可)。

16.1 配置与基础工具

// workflow-client-config.js
export const API_BASE = 'https://your-host/api/v1'
export const API_KEY = 'YOUR_X_API_KEY'

// 统一处理返回 envelope {code, message, data}
export async function apiRequest(method, path, body) {
  const res = await fetch(`${API_BASE}${path}`, {
    method,
    headers: {
      'X-API-Key': API_KEY,
      'Content-Type': 'application/json'
    },
    body: body ? JSON.stringify(body) : undefined
  })
  const text = await res.text()
  let json
  try { json = JSON.parse(text) } catch { throw new Error(`非 JSON 响应: ${text}`) }
  if (!res.ok || json.code !== 0) {
    const err = new Error(json.message || `HTTP ${res.status}`)
    err.code = json.code
    err.httpStatus = res.status
    err.payload = json
    throw err
  }
  return json.data
}

// 指数退避
function backoff(attempt) {
  return Math.min(1000 * Math.pow(2, attempt - 1), 16000)
}

// LRU eventId 去重
class EventIdDedup {
  constructor(size = 500) {
    this.size = size
    this.set = new Set()
    this.queue = []
  }
  seen(id) {
    if (!id) return false
    if (this.set.has(id)) return true
    this.set.add(id)
    this.queue.push(id)
    if (this.queue.length > this.size) {
      const old = this.queue.shift()
      this.set.delete(old)
    }
    return false
  }
}

16.2 启动运行

// 启动外部工作流
export async function startWorkflowRun(workflowKey, inputs = {}, options = {}) {
  return apiRequest('POST', '/workflows/' + encodeURIComponent(workflowKey) + '/runs', {
    inputs,
    callbackUrl: options.callbackUrl || null
  })
}
// 返回: { runId, status: 'running', streamUrl: '/api/v1/runs/{runId}/stream' }

16.3 SSE 客户端封装

// sse-client.js
export class WorkflowSSEClient {
  constructor({ runId, onEvent, onOpen, onError, onClose, getLastEventId }) {
    this.runId = runId
    this.url = `${API_BASE}/runs/${runId}/stream`
    this.onEvent = onEvent
    this.onOpen = onOpen
    this.onError = onError
    this.onClose = onClose
    this.getLastEventId = getLastEventId || (() => null)
    this.es = null
    this.closed = false
  }

  open() {
    if (this.closed) return
    // EventSource 不支持自定义 header,但原生支持 Last-Event-ID
    // 浏览器会自动在重连时携带 Last-Event-ID,无需手动加
    const lastId = this.getLastEventId()
    const url = lastId ? `${this.url}&lastEventId=${lastId}` : this.url
    // 注:服务端 SSE 端点也应支持 lastEventId query 参数(容错)
    this.es = new EventSource(url, { withCredentials: false })

    this.es.onopen = () => this.onOpen && this.onOpen()
    this.es.onerror = (e) => {
      // EventSource 自带重连,但这里我们需要业务层感知
      this.onError && this.onError(e)
    }
    this.es.onmessage = (e) => this._dispatch(e)
  }

  _dispatch(e) {
    // EventSource 默认 event 是 'message';服务端用 data: 推送 JSON
    const dataStr = e.data
    if (!dataStr) return
    let payload
    try { payload = JSON.parse(dataStr) } catch { return }
    // 记录 lastEventId
    if (e.lastEventId) this._lastEventId = e.lastEventId
    this.onEvent && this.onEvent(payload, e.lastEventId)
  }

  close() {
    this.closed = true
    if (this.es) {
      this.es.close()
      this.es = null
    }
    this.onClose && this.onClose()
  }

  getLastEventIdSnapshot() {
    return this._lastEventId || null
  }
}

16.4 VisitStore(运行状态容器)

// visit-store.js
export class VisitStore {
  constructor() {
    this.runStatus = 'idle'        // idle/running/completed/failed
    this.runStartedAt = null
    this.runEndedAt = null
    this.visits = new Map()        // visitUuid -> Visit
    this.visitOrder = []           // 按到达顺序的 visitUuid 数组
    this.pendingAskGlobal = null   // 当前活跃的 AskFor(指向某 visit 的 pendingAsk)
    this.error = null
    this.dedup = new EventIdDedup(500)
  }

  // 获取或创建 visit
  _getOrCreate(visitUuid, init = {}) {
    if (!this.visits.has(visitUuid)) {
      this.visits.set(visitUuid, {
        visitUuid,
        runId: null,
        nodeId: null,
        nodeType: null,
        nodeName: null,
        status: 'running',
        startedAt: Date.now(),
        endedAt: null,
        liveStatus: '',
        thinking: '',
        todoList: [],
        toolCalls: [],
        logs: [],
        pendingAsk: null,
        output: null,
        error: null,
        collapsed: false,
        ...init
      })
      this.visitOrder.push(visitUuid)
    }
    return this.visits.get(visitUuid)
  }

  // 统一事件入口
  applyEvent(event, eventId) {
    if (eventId && this.dedup.seen(eventId)) return  // 去重

    switch (event.type) {
      case 'node_status':       return this._onNodeStatus(event)
      case 'node_stream':       return this._onNodeStream(event)
      case 'visit_recorded':    return this._onVisitRecorded(event)
      case 'planner_decision':  return this._onPlannerDecision(event)
      case 'workflow_complete': return this._onWorkflowComplete(event)
      case 'workflow_error':    return this._onWorkflowError(event)
    }
  }

  _onNodeStatus(event) {
    const v = this._getOrCreate(event.visitUuid, {
      runId: event.runId,
      nodeId: event.nodeId,
      nodeType: event.nodeType,
      nodeName: event.nodeName
    })
    v.status = event.status
    if (event.status === 'completed' || event.status === 'failed' || event.status === 'skipped') {
      v.endedAt = event.timestamp || Date.now()
      v.collapsed = true   // 完成后默认折叠
      if (event.output) v.output = event.output
      if (event.error) v.error = event.error
      if (event.status === 'failed') v.error = v.error || { message: '执行失败' }
    }
    if (event.nodeName) v.nodeName = event.nodeName
  }

  _onNodeStream(event) {
    const v = this._getOrCreate(event.visitUuid)
    const kind = event.kind
    const data = event.data || {}

    switch (kind) {
      case 'status':
        v.liveStatus = data.text || data.message || ''
        break
      case 'thinking':
        // 追加(不换行可由后端控制;前端按需 \n)
        v.thinking += (data.text || data.chunk || '')
        break
      case 'todo_update':
        // 覆盖式
        v.todoList = Array.isArray(data.items) ? data.items : (data.todos || [])
        break
      case 'tool_call': {
        const call = {
          callId: data.callId || data.id,
          toolName: data.toolName || data.name,
          arguments: data.arguments || data.args || {},
          status: 'running',
          result: null,
          startedAt: Date.now(),
          endedAt: null
        }
        v.toolCalls.push(call)
        v.logs.push({ ts: Date.now(), text: `调用工具: ${call.toolName}` })
        break
      }
      case 'tool_progress': {
        const call = v.toolCalls.find(c => c.callId === (data.callId || data.id))
        if (call) call.status = 'running'
        if (data.text) v.logs.push({ ts: Date.now(), text: data.text })
        break
      }
      case 'tool_result': {
        const call = v.toolCalls.find(c => c.callId === (data.callId || data.id))
        if (call) {
          call.status = data.error ? 'failed' : 'completed'
          call.result = data.result || data.output
          call.endedAt = Date.now()
        }
        break
      }
      case 'step':
        v.logs.push({ ts: Date.now(), text: data.text || data.message || '' })
        break
      case 'clarify_request':
        v.pendingAsk = {
          clarifyId: data.clarifyId,
          question: data.question,
          options: data.options || [],
          allowCustom: !!data.allowCustom,
          placeholder: data.placeholder || '',
          multiple: !!data.multiple,
          submittedAt: null,
          submittedValue: null
        }
        this.pendingAskGlobal = v.pendingAsk
        break
    }
  }

  _onVisitRecorded(event) {
    // 仅作为访问落盘通知,无需特殊处理(Visit 已在 node_status 时创建)
    // 可用于触发"已固化"标识
  }

  _onPlannerDecision(event) {
    // 业务方可选择展示 AI 决策链;这里记录到 run 级元信息
    if (!this.plannerDecisions) this.plannerDecisions = []
    this.plannerDecisions.push(event.decision)
  }

  _onWorkflowComplete(event) {
    this.runStatus = 'completed'
    this.runEndedAt = Date.now()
    if (event.output) this.finalOutput = event.output
  }

  _onWorkflowError(event) {
    this.runStatus = 'failed'
    this.runEndedAt = Date.now()
    this.error = event.error || { message: event.message || '工作流执行失败' }
  }

  // /state 恢复
  applyStateSnapshot(state) {
    this.runStatus = state.status
    this.runStartedAt = state.startedAt
    this.runEndedAt = state.endedAt
    this.pendingAskGlobal = state.pendingAskGlobal || null
    this.visitOrder = []
    this.visits.clear()
    for (const v of (state.visits || [])) {
      this.visits.set(v.visitUuid, { ...v, collapsed: v.status !== 'running' })
      this.visitOrder.push(v.visitUuid)
    }
  }

  // 提交 AskFor
  markAskSubmitted(clarifyId, answer) {
    for (const v of this.visits.values()) {
      if (v.pendingAsk && v.pendingAsk.clarifyId === clarifyId) {
        v.pendingAsk.submittedAt = Date.now()
        v.pendingAsk.submittedValue = answer
        break
      }
    }
    if (this.pendingAskGlobal && this.pendingAskGlobal.clarifyId === clarifyId) {
      this.pendingAskGlobal = null
    }
  }

  // 序列化为数组(给 UI 渲染)
  toArray() {
    return this.visitOrder.map(uuid => this.visits.get(uuid))
  }
}

16.5 RunController(运行总控)

// run-controller.js
import { WorkflowSSEClient } from './sse-client.js'
import { VisitStore } from './visit-store.js'
import { apiRequest } from './workflow-client-config.js'

export class RunController {
  constructor(runId, { onStateChange, onBannerChange }) {
    this.runId = runId
    this.store = new VisitStore()
    this.onStateChange = onStateChange
    this.onBannerChange = onBannerChange
    this.sseAttempts = 0
    this.pollingTimer = null
    this.sseClient = null
    this.terminated = false
  }

  // 启动:先拉一次 state,再开 SSE
  async start() {
    // 若运行已存在(如刷新页面),先恢复
    try {
      const state = await apiRequest('GET', `/runs/${this.runId}/state`)
      this.store.applyStateSnapshot(state)
      this._emitChange()
      if (state.status === 'completed' || state.status === 'failed') {
        this.terminated = true
        return
      }
    } catch (e) {
      // 新运行(尚未落盘),忽略 404
    }
    this._openSSE()
  }

  _openSSE() {
    if (this.terminated) return
    this.sseClient = new WorkflowSSEClient({
      runId: this.runId,
      getLastEventId: () => this._lastEventId,
      onOpen: () => {
        this.sseAttempts = 0
        this._stopPolling()
        this._setBanner(null)
      },
      onEvent: (event, eventId) => {
        if (eventId) this._lastEventId = eventId
        this.store.applyEvent(event, eventId)
        this._emitChange()
        if (event.type === 'workflow_complete' || event.type === 'workflow_error') {
          this.terminated = true
          this.sseClient.close()
          this._stopPolling()
        }
      },
      onError: () => {
        // EventSource 自带重连,但若 readyState 已 CLOSED,说明彻底失败
        if (this.sseClient.es.readyState === EventSource.CLOSED) {
          this.sseAttempts++
          if (this.sseAttempts <= 5) {
            setTimeout(() => this._openSSE(), backoff(this.sseAttempts))
          } else if (!this.pollingTimer) {
            this._setBanner('网络异常,正在降级同步')
            this._startPolling()
          }
        }
      }
    })
    this.sseClient.open()
  }

  _startPolling() {
    const poll = async () => {
      if (this.terminated) return
      try {
        const state = await apiRequest('GET', `/runs/${this.runId}/state`)
        this.store.applyStateSnapshot(state)
        this._emitChange()
        if (state.status === 'completed' || state.status === 'failed') {
          this.terminated = true
          this._stopPolling()
          this._setBanner(null)
        }
      } catch (e) {
        // 忽略,下次再试
      }
      if (!this.terminated) {
        this.pollingTimer = setTimeout(poll, 30000)
      }
    }
    poll()
  }

  _stopPolling() {
    if (this.pollingTimer) {
      clearTimeout(this.pollingTimer)
      this.pollingTimer = null
    }
  }

  _setBanner(text) {
    this.onBannerChange && this.onBannerChange(text)
  }

  _emitChange() {
    this.onStateChange && this.onStateChange(this.store)
  }

  // 提交 AskFor 答案
  async submitAsk(clarifyId, answer) {
    try {
      await apiRequest('POST', `/runs/${this.runId}/resume`, { clarifyId, answer })
      this.store.markAskSubmitted(clarifyId, answer)
      this._emitChange()
    } catch (e) {
      if (e.code === 'CLARIFY_ALREADY_ANSWERED') {
        // 视为成功,刷新 state
        const state = await apiRequest('GET', `/runs/${this.runId}/state`)
        this.store.applyStateSnapshot(state)
        this._emitChange()
      } else {
        throw e
      }
    }
  }

  destroy() {
    this.terminated = true
    if (this.sseClient) this.sseClient.close()
    this._stopPolling()
  }
}

function backoff(attempt) {
  return Math.min(1000 * Math.pow(2, attempt - 1), 16000)
}

16.6 Vue 3 组件示例(业务方接入参考)

<!-- WorkflowRunPanel.vue -->
<template>
  <div class="run-panel">
    <div v-if="banner" class="banner">{{ banner }}</div>

    <div v-for="visit in visits" :key="visit.visitUuid" class="visit-block">
      <div class="visit-header" @click="toggle(visit)">
        <span class="badge" :class="visit.status">
          {{ visit.status === 'running' ? '●' : visit.status === 'completed' ? '✓' : visit.status === 'failed' ? '✗' : '−' }}
        </span>
        <span class="type">[{{ visit.nodeType }}]</span>
        <span class="name">{{ visit.nodeName }}</span>
        <span v-if="visitDuration(visit)" class="duration">· {{ visitDuration(visit) }}</span>
      </div>

      <div v-show="!visit.collapsed" class="visit-body">
        <!-- ① 实时状态(running 中显示) -->
        <div v-if="visit.status === 'running' && visit.liveStatus" class="live-status">
          {{ visit.liveStatus }}
        </div>

        <!-- ② 思考过程 -->
        <details v-if="visit.thinking" open>
          <summary>思考过程</summary>
          <div class="md-body" v-html="renderMarkdown(visit.thinking)" />
        </details>

        <!-- ③ 待办清单 -->
        <details v-if="visit.todoList.length">
          <summary>待办清单</summary>
          <ul class="todo">
            <li v-for="(t, i) in visit.todoList" :key="i" :class="{ done: t.done }">
              <span class="todo-icon">{{ t.done ? '✓' : '○' }}</span>{{ t.text }}
            </li>
          </ul>
        </details>

        <!-- ④ 工具调用 -->
        <details v-if="visit.toolCalls.length">
          <summary>工具调用 ({{ visit.toolCalls.length }})</summary>
          <div v-for="call in visit.toolCalls" :key="call.callId" class="tool-call">
            <div class="tool-head">
              <span>{{ call.toolName }}</span>
              <span class="tool-status" :class="call.status">{{ call.status }}</span>
            </div>
            <pre class="tool-args">{{ JSON.stringify(call.arguments, null, 2) }}</pre>
            <pre v-if="call.result" class="tool-result">{{ formatResult(call.result) }}</pre>
          </div>
        </details>

        <!-- ⑤ 执行日志 -->
        <details v-if="visit.logs.length">
          <summary>执行日志</summary>
          <ul class="logs">
            <li v-for="(log, i) in visit.logs" :key="i">
              <span class="log-ts">{{ formatTs(log.ts) }}</span>{{ log.text }}
            </li>
          </ul>
        </details>

        <!-- AskFor 面板 -->
        <AskForPanel
          v-if="visit.pendingAsk && !visit.pendingAsk.submittedAt"
          :ask="visit.pendingAsk"
          @submit="onAskSubmit"
        />
        <div v-else-if="visit.pendingAsk && visit.pendingAsk.submittedAt" class="ask-submitted">
          已提交:{{ formatAnswer(visit.pendingAsk.submittedValue) }}
        </div>
      </div>

      <!-- 最终输出(始终展开) -->
      <div v-if="visit.output" class="final-output" :class="outputStatus(visit.output)">
        <div class="output-head">{{ outputMessage(visit.output) }}</div>
        <div class="md-body" v-html="renderMarkdown(outputBody(visit.output))" />
      </div>
    </div>
  </div>
</template>

<script setup>
import { ref, onMounted, onUnmounted, computed } from 'vue'
import { RunController } from '../composables/run-controller'
import { renderMarkdown } from '../utils/markdown'
import {
  isEnvelope, unwrapNodeOutput, nodeOutputStatus, nodeOutputMessage
} from '../utils/envelope'
import AskForPanel from './AskForPanel.vue'

const props = defineProps({ runId: { type: String, required: true } })

const store = ref({ visits: [], runStatus: 'idle' })
const banner = ref(null)

let controller = null
onMounted(() => {
  controller = new RunController(props.runId, {
    onStateChange: s => { store.value = s },
    onBannerChange: text => { banner.value = text }
  })
  controller.start()
})
onUnmounted(() => controller && controller.destroy())

const visits = computed(() => store.value.toArray ? store.value.toArray() : [])

function toggle(visit) { visit.collapsed = !visit.collapsed }
function visitDuration(v) {
  if (!v.endedAt) return ''
  return ((v.endedAt - v.startedAt) / 1000).toFixed(1) + 's'
}
function outputStatus(o) { return nodeOutputStatus(o) }
function outputMessage(o) { return nodeOutputMessage(o) }
function outputBody(o) {
  const u = unwrapNodeOutput(o)
  return typeof u === 'string' ? u : '```json\n' + JSON.stringify(u, null, 2) + '\n```'
}
function formatResult(r) { return typeof r === 'string' ? r : JSON.stringify(r, null, 2) }
function formatTs(ts) { return new Date(ts).toLocaleTimeString() }
function formatAnswer(v) { return Array.isArray(v) ? v.join(', ') : String(v) }
function onAskSubmit({ clarifyId, answer }) {
  controller.submitAsk(clarifyId, answer)
}
</script>

16.7 AskForPanel 组件示例

<!-- AskForPanel.vue -->
<template>
  <div class="ask-panel">
    <div class="question" v-html="renderMarkdown(ask.question)" />
    <div v-if="ask.options.length" class="options">
      <button
        v-for="opt in ask.options"
        :key="opt.value"
        :class="{ active: isSelected(opt.value) }"
        @click="toggleSelect(opt.value)"
      >{{ opt.label }}</button>
    </div>
    <div v-if="ask.allowCustom" class="custom">
      <input
        v-model="customText"
        :placeholder="ask.placeholder || '请输入'"
        @keyup.enter="submitCustom"
      />
      <button @click="submitCustom">提交</button>
    </div>
    <div class="actions" v-if="ask.options.length">
      <button :disabled="!canSubmit" @click="submitSelection">确认</button>
    </div>
  </div>
</template>

<script setup>
import { ref, computed } from 'vue'
import { renderMarkdown } from '../utils/markdown'

const props = defineProps({ ask: { type: Object, required: true } })
const emit = defineEmits(['submit'])

const selected = ref([])
const customText = ref('')

function isSelected(v) {
  return props.ask.multiple ? selected.value.includes(v) : selected.value[0] === v
}
function toggleSelect(v) {
  if (props.ask.multiple) {
    const i = selected.value.indexOf(v)
    if (i >= 0) selected.value.splice(i, 1)
    else selected.value.push(v)
  } else {
    selected.value = [v]
  }
}
const canSubmit = computed(() => selected.value.length > 0 || customText.value.trim())

function submitSelection() {
  if (props.ask.multiple) {
    emit('submit', { clarifyId: props.ask.clarifyId, answer: selected.value })
  } else {
    emit('submit', { clarifyId: props.ask.clarifyId, answer: selected.value[0] })
  }
}
function submitCustom() {
  const text = customText.value.trim()
  if (!text) return
  emit('submit', { clarifyId: props.ask.clarifyId, answer: text })
}
</script>

16.8 关键集成要点

  1. 状态容器统一:所有事件统一过 VisitStore.applyEvent(),UI 只读 store.toArray()。框架响应式(Vue 的 ref / React 的 useState)订阅 store 变化即可。
  2. 事件去重必须:/state + SSE 重放会产生重复事件,EventIdDedup 是必须的(500 容量足够)。
  3. AskFor 提交后立即禁用按钮:避免重复点击;同时记录 submittedAt,UI 上展示"已提交"。
  4. 运行结束不停止 SSE:服务端会主动关闭流,前端收到 workflow_complete/error 后置 terminated=true 即可。
  5. markdown 渲染必须 sanitize:DOMPurify 是必需品(用户输入与 LLM 输出都可能含恶意脚本)。
  6. 代码块横向滚动:长代码必须能横向滚动,不要换行;推荐使用 pre::-webkit-scrollbar 美化(见本仓库 frontend/src/styles/hljs-theme.css)。

17. 业务方集成模式

业务方根据自身前端架构、定制需求、上线节奏,可从以下三种模式中选择:

17.1 模式 A:iframe 嵌入(最快接入,1-2 天)

适用场景:业务方有现成的运行结果页 URL(即本系统的工作流运行记录页),仅需在自有系统中嵌入展示,不需要深度定制 UI。

实现:

<iframe
  src="https://agent-host/workflows/runs/{runId}?embed=1&apiKey=xxx"
  width="100%" height="800"
  frameborder="0"
  allowfullscreen
></iframe>

优点:

  • 接入成本最低(一个 iframe 标签)
  • 自动跟随本系统升级(UI 与功能始终保持最新)
  • 不需要业务方实现 SSE / 状态管理 / markdown 渲染

缺点:

  • 视觉风格固定(深蓝主题,业务方无法覆盖)
  • 跨域 cookie / 认证需配合(推荐用一次性 token)
  • iframe 内的 AskFor 提交对父页面不可见(业务方无法记录用户行为)

约束:

  • 本系统需开放 embed=1 模式(隐藏导航栏与侧边栏,仅保留运行结果区)
  • 需要在本系统后台为业务方申请 API Key 与白名单域名
  • iframe 不可跨域访问父页面 DOM(postMessage 通信可解决,但本系统当前未实现)

17.2 模式 B:原生复刻(推荐,5-10 天)

适用场景:业务方有自身设计系统,需要深度融入自家产品;交互逻辑与本系统一致即可。

实现步骤:

  1. 复刻组件结构:参照 §13 的 5 面板 + 最终输出布局
  2. 集成 JS 参考实现:直接使用 §16 的 RunController + VisitStore(约 400 行代码)
  3. 替换样式与文案:节点状态徽章、面板折叠头、按钮风格按业务方设计系统
  4. 保留交互逻辑:折叠策略(完成后折叠)、AskFor 暂停-恢复、断线重连

优点:

  • 视觉与业务方系统完全融合
  • 用户感知不到「嵌入第三方组件」
  • 可以扩展(如增加业务方自定义面板)

缺点:

  • 需要业务方投入前端资源
  • 本系统升级时业务方需要跟随(建议订阅 release note)

17.3 模式 C:跨域代理 + 组件复用(折中,3-5 天)

适用场景:业务方使用 Vue 3(或愿意引入),希望直接复用本系统已开源的 Vue 组件,但前端跨域。

实现:

  1. 业务方后端反代:将 /agent-api/* 反代到 https://agent-host/api/v1/*,注入 X-API-Key
  2. 业务方前端引入组件:从本仓库 npm 包或 git submodule 拉取以下文件:
    • frontend/src/utils/markdown.js
    • frontend/src/utils/envelope.js(封装 §13.3 的解包函数)
    • frontend/src/composables/useWorkflowRunner.js(VisitStore + SSE 路由)
    • frontend/src/components/agent-thinking/*(5 大面板组件)
  3. 样式按需调整:覆盖 .agent-thinking-* 类名的前缀色(CSS 变量)

优点:

  • 复用经过验证的组件,bug 更少
  • 升级成本中等(git submodule + 合并冲突即可)
  • 业务方前端无需实现复杂 SSE 状态机

缺点:

  • 依赖本系统的 Vue 3 + 依赖项(marked / highlight.js / dompurify / katex)
  • 组件 API 可能随版本演进(建议钉版本)

17.4 决策树

是否需要 UI 与本系统一致?
├─ 是 ── 是否能接受 iframe?
│       ├─ 是 ── 模式 A(iframe 嵌入)
│       └─ 否 ── 是否使用 Vue 3?
│               ├─ 是 ── 模式 C(跨域代理 + 组件复用)
│               └─ 否 ── 模式 B(原生复刻)
└─ 否 ── 模式 B(原生复刻,UI 完全自定义)

17.5 安全与跨域配置

无论哪种模式,业务方需要:

  1. 后端允许业务方域名跨域访问 SSE:
    • Access-Control-Allow-Origin: https://business-host
    • Access-Control-Allow-Headers: X-API-Key, Content-Type, Last-Event-ID
    • SSE 流需支持 text/event-stream MIME
  2. API Key 不落地前端:模式 A/C iframe 或反代时,由业务方后端注入;模式 B 可由业务方后端换取临时 token 下发到前端
  3. Callback URL 加签:若使用 callbackUrl,业务方后端需校验请求签名(HMAC-SHA256 + timestamp),避免伪造

18. 集成实施检查清单

业务方按以下清单逐项确认,确保接入完整。

18.1 后端接口

  • 已申请 X-API-Key 并配置到业务方后端
  • 业务方后端可访问本系统 /api/v1/workflows/{workflowName}/runs(POST 创建运行)
  • 业务方后端可访问本系统 /api/v1/workflows/{workflowName}/runs/{runId}/start(POST 启动)
  • 业务方后端可访问本系统 /api/v1/workflows/{workflowName}/runs/{runId}/stream(GET SSE,连接可持续 24 小时,注意超时设置)
  • (可选)查询/工作空间端点可达:GET .../runs/{runId}、GET .../runs/{runId}/workspace
  • (含 AskFor 的场景)GET /api/v1/runs/{runId}/state + POST /api/v1/runs/{runId}/resume 可达(v1.2 已开放,路径不带 workflowName 段,见 §14.4/§14.5)
  • HTTP 客户端超时设置:SSE 不设读超时(依赖 30 秒心跳 + 重连),REST 接口 ≤ 30 秒
  • CORS 配置允许业务方域名

18.2 SSE 事件处理

  • 实现 6 种事件类型分发:node_status / node_stream / visit_recorded / planner_decision / workflow_complete / workflow_error
  • 实现 8 种 node_stream.kind:status / thinking / todo_update / tool_call / tool_progress / tool_result / step / clarify_request
  • 事件去重:基于 eventId / LRU 500 容量
  • 心跳帧 :keep-alive 不触发业务逻辑
  • 收到 workflow_complete 或 workflow_error 后停止重连

18.3 状态展示

  • 实时状态(status):覆盖式更新,节点完成后隐藏
  • 思考过程(thinking):追加式更新,markdown 渲染
  • 待办清单(todo_update):覆盖式更新,✓/○ 状态图标
  • 工具调用(tool_call/result):按到达顺序配对(事件无 callId),卡片式
  • 执行日志(step):追加式,时间戳
  • 最终输出:envelope 解包,markdown 渲染,状态染色

18.4 交互逻辑

  • AskFor 面板:选项按钮 + 自定义输入框 + 提交按钮
  • AskFor 提交后立即禁用按钮,展示"已提交"
  • AskFor 错误处理:CLARIFY_ALREADY_ANSWERED 视为成功
  • 面板折叠:完成后自动折叠,点击头部展开/收起
  • 多 visit 渲染:纵向堆叠,不合并

18.5 断线重连

  • SSE 断开后指数退避重连(1s/2s/4s/8s/16s,5 次)
  • 5 次失败后降级轮询 GET /api/v1/workflows/{workflowName}/runs/{runId}(30s/次,依据返回 status 判断终态)
  • SSE 恢复后立即停止轮询
  • 重连期间顶部横幅提示
  • 运行结束后停止一切重连

18.6 markdown 渲染

  • GFM(表格/删除线/任务列表)
  • 代码高亮(highlight.js /lib/common 子集)
  • DOMPurify sanitize(XSS 防护)
  • 代码块横向滚动 + 滚动条美化
  • KaTeX(如业务场景含公式)

18.7 兼容性

  • 浏览器兼容:Chrome 90+ / Edge 90+ / Firefox 88+ / Safari 14+
  • EventSource polyfill(如需支持 IE / 老版 Safari)
  • 移动端响应式(折叠头触摸友好,按钮 ≥ 44px)
  • 屏幕阅读器:面板折叠头有 aria-expanded、状态徽章有 aria-label

19. 参考资源

19.1 本仓库源码位置

主题 文件路径
markdown 渲染 frontend/src/utils/markdown.js
代码高亮主题 + 滚动条 frontend/src/styles/hljs-theme.css
SSE 事件路由(工作流) frontend/src/composables/useWorkflowRunner.js
运行池(chat 侧) frontend/src/composables/useGlobalRuns.js
5 大面板组件 frontend/src/components/agent-thinking/*
AskFor 面板 frontend/src/components/agent-thinking/AskforPanel.vue
Chat 消息列表(assistant 渲染) frontend/src/components/chat/ChatMessageList.vue
工作流编辑器运行结果区 frontend/src/views/workflow/WorkflowEditor.vue
运行历史页 frontend/src/views/workflow/RunHistory.vue
SSE 事件总线(后端) backend/src/main/java/com/agent/management/external/SseEventBus.java
心跳(后端) backend/src/main/java/com/agent/management/external/SseHeartbeats.java
外部 API 控制器 backend/src/main/java/com/agent/management/controller/ExternalApiController.java
AskFor 阻塞-唤醒 backend/src/main/java/com/agent/management/engine/hermes/ClarifyHooks.java

19.2 相关规范与文档

  • 本系统 v1.0 API 规范:docs/specs/external-workflow-api-spec.md §1-§10(基础接口契约)
  • Agent 通信与思考过程:docs/design/agent-chat-page-design.md
  • Hermes Bridge 集成:docs/reference/hermes-bridge-reference.md(如有)
  • AskFor 暂停-恢复机制:docs/design/agentic-workflow-plan.md

19.3 第三方依赖版本建议

依赖 推荐版本 用途
marked ^12.0.0 markdown 解析
marked-highlight ^2.2.0 marked 与 hljs 集成
marked-katex-extension ^4.0.0 KaTeX 公式扩展(可选)
highlight.js ^11.12.0 代码高亮(使用 /lib/common 子集)
dompurify ^3.2.0 XSS sanitize
katex ^0.16.0 数学公式渲染(可选)

19.4 联系与反馈

  • 接入过程中遇到问题,请通过本系统 Issue 跟踪系统反馈
  • API 行为与本规范不一致时,以实际后端实现为准,规范会随后补齐
  • 业务方特殊需求(如自定义事件、私有节点类型)需双方评审后纳入规划

文档版本:v1.1(基于实际实现,2026-08-12 更新) 适用版本:agent-management backend ≥ 1.5.0 下次评审:v1.2 计划增加流式节点输出(LLM token 级流)、批量启动接口