Skip to content

事件与工作流设计.md

Latest commit

 

History

History
150 lines (98 loc) · 13.7 KB

事件与工作流设计.md

File metadata and controls

150 lines (98 loc) · 13.7 KB
Raw

BubblePilot 事件与工作流设计

核心合同

工作流是经过校验的版本化有向图，不是任意脚本。每个节点必须定义类型、版本、配置 Schema、输入输出、成功或失败出口和错误分类；会阻塞的叶子 I/O 必须在所属领域定义单次超时，编排器只遍历图和维护状态，不再叠加工作流或节点总时间预算。

Message Trigger → Load Context → AI Chat → Reply → End
                         └──────→ failure branch ──────┘

当前生产合同由 contracts/workflow.schema.json 定义：最多 64 个节点、默认最多 64 步。发布前会校验节点配置、唯一 ID、边引用、不可达节点、环路、模板变量、正则、资源上限和 AI 路由。旧定义中的 maxExecutionMs 和 AI 节点 timeoutMs 会在读取时忽略，新版本不再写入。

节点

节点 作用 主要边界
message-trigger@1 按消息条件启动工作流 每个图固定一个起点；必须显式启用
condition@1 对消息字段做条件分支 只读，不产生外部副作用
render-text@1 使用 Context 内容渲染文本 模板最长 12,000 字符，渲染结果最长 32,000 字符
load-context@1 读取当前聊天最近消息和成员身份 1-50 条、100-20,000 字符，可排除自身消息；映射只含窗口内成员和当前发言者
ai-chat@1 通过版本化 Provider Route 调用 AI 有超时、输出、Retry、联网和安全上限
reply@1 向当前聊天发送文本 最多 4,000 字符，发送幂等且只有限重试
log@1 写入脱敏执行摘要 不记录正文、Secret 或完整 Prompt
end@1 以成功或跳过结束分支 无外部副作用

新增节点必须实现合同、注册到节点注册表，覆盖成功、超时、重试、重复执行和敏感数据测试，并同步 Schema 与本文。已发布节点版本的行为不能在生产环境直接替换。

set-variable@1 仅为已发布工作流保留运行兼容，不再出现在动作目录中；未进入生产合同的旧“文本模板”“JSON 解析”和“JSON 字段提取”目录项已经移除。

触发匹配

message-trigger 支持：

  • 聊天范围、发送者范围和消息类型；
  • 文本包含关键词、前缀或正则；
  • 供应商范围；
  • 是否包含自己发送的消息。

不同条件之间按 AND 组合，同一条件中的列表按 OR 组合。生产运行固定拒绝 isFromMe 消息；只有未来增加可持久化循环预算后才允许自触发，不能仅依靠画布开关绕过。

只有启用的触发器和启用工作流的已发布版本处理生产消息。同一消息命中多个触发器时，每个触发器创建独立逻辑执行；配置端会提示可能重叠，但该静态提示不阻止启用，也不替代运行时匹配。

触发预览只能使用虚构样本做无副作用判断，不调用 AI、不发送回复、不创建执行。历史 API 仍保留 IANA 时区时间窗口合同：起点包含、终点不包含，跨午夜时星期表示窗口开始日；运行时以消息的 sentAt 判断，不使用服务器时区。

版本和执行

  • 新工作流先保存为已校验候选版本；启用时发布候选并将新执行切到该版本。
  • 活动工作流保存新版本时，会发布新版本；已开始执行继续引用原版本。
  • 节点默认串行执行。并行分支只有在节点和存储合同支持后才能开放。
  • 节点尝试单独记录状态、耗时、错误码、是否可重试和下一次重试时间。
  • Context、历史消息、输出字符、模型 Token、节点步数和总执行时间都有硬上限。
  • AI 节点不能修改消息归档、权限或系统配置。

消息归档与调度之间使用持久化 evaluation-pending。队列满、等待超时或最终判定写入前进程中断时，同一 Webhook 可以重投并恢复调度；执行 Claim 和回复幂等键阻止重复副作用。已经形成最终判定的事件不会按当前配置补跑。

Context 模板与 Prompt

render-text@1 没有数据输入端口。模板编辑器可以插入经过白名单限制的 Context 内容，并通过 text 端口把渲染结果交给 AI、回复或其他文本消费者。例如：

当前发送者：{{context.event.message.senderId}}
当前消息：{{context.event.message.text}}
聊天名称：{{context.event.chat.displayName}}
AI 草稿：{{context.outputs.ask-ai.text}}

当前允许读取消息 Provider、消息正文、发送者标识、消息 ID、时间、内容类型、附件、Chat ID、聊天类型、聊天名称、已加载历史、context.history.participants 成员映射，以及已声明的上游动作输出。数组和对象以 JSON 文本渲染，null 或尚不存在的值渲染为空字符串。发布时拒绝未知 Context 路径、未知输出端口和不完整模板语法；执行摘要只记录渲染字符数，不记录正文。

AI、回复和历史兼容节点原有的简写模板仍允许引用运行时已知键，例如：

{{message.text}}
{{message.senderId}}
{{chat.providerChatId}}
{{variables.aiReply}}

AI 请求按稳定性组织：长期角色和规则放在 system，稳定任务说明放在 <task_instructions>，动态聊天历史放在 <chat_history>，直接来自当前事件的消息放在带发送者标签的 <current_input>，上游动作输出放在带来源动作 ID 的 <upstream_input>。不要把时间戳、当前消息或完整历史拼入稳定 System Prompt。

对话成员身份

“消息”页从单个聊天已经归档的非本人消息聚合 sender ID。管理员完成二次验证后，可以按聊天分别填写本名和昵称；保存使用独立版本号防止并发覆盖，只接受该聊天历史中已经出现的 ID。映射不跨聊天共享，本名和昵称均为空表示删除映射。

load-context@1 读取消息后，只解析历史窗口内出现的 sender ID 和当前消息 sender ID，并通过 participants 输出提供这部分映射。AI 节点不再重复注入成员映射名单，而是在每条 <chat_history> 消息及直接的 <current_input> 上把本名、昵称和原始 ID 格式化为唯一发送者标签；System 只保留不含身份数据的简短使用规则。本名与昵称被明确视为同一个人的不同称呼；未出现在本次窗口或当前消息中的成员映射不会进入 AI 请求，也不得被模型提及或推断。

多个 AI 串联时，每个节点独立决定是否包含已加载上下文。负责理解原始对话的首个 AI 通常开启 includeLoadedContext，因此获得成员标签、历史和当前消息；后续润色、结构化或审核 AI 通常关闭该配置，并把前一节点的 text 或 json 作为 <upstream_input>，不会再次携带成员映射或历史，也不会把 AI 输出误标成原始用户发言。只有后续节点确实需要重新阅读原对话时才继续开启，这会有意再次消耗历史 Token。

对“加载上下文 → AI → 回复”的流程，历史只加载一次，通常只保留一个生成最终回复的 AI 节点，并把输出上限设为场景所需的最小值。缓存命中只能通过 Provider 返回的 cachedPromptTokens、cacheWritePromptTokens 和 cacheMissPromptTokens 观测，不能把它当作应用保证。

AI Provider 路由

AI 节点引用版本化 providerRouteId，不在工作流中保存 Base URL 或 Secret。节点启动时锁定路由快照，包括候选顺序、Provider 版本、单请求超时、Retry 和降级策略，但不复制 Secret 明文。

术语保持统一：

  • Provider Attempt：对一个 Provider 的一次真实调用。
  • Fallback：同一 Node Retry 轮次内切换到下一个候选。
  • Node Retry：本轮候选耗尽后，按退避开始下一轮。
  • Degrade：连续故障达到阈值后，Provider 在冷却期退出常规候选。

执行规则：

  1. 运行时排除停用、删除、未配置 Secret 或仍在冷却的 Provider；管理员固定顺序不会被健康状态改写。
  2. 每个 Provider 每轮最多调用一次；得到有效输出后立即停止 Fallback。
  3. 超时、连接失败、限流、可恢复服务端错误、局部鉴权或模型错误、空输出和协议无效可以按策略切换候选。
  4. 请求取消、执行中断、工作流配置错误和明确安全拒绝不得盲目 Fallback。
  5. 全部候选失败后，只有存在可重试错误且仍有预算时才进入 Node Retry。
  6. 全部候选降级时，每轮最多选择一个已到恢复时间的候选进行半开探测。

fallbackEnabled=false 只禁止切换其他候选，不禁止当前 Provider 的配置轮次。成功 HTTP 响应但没有可见输出使用 AI_PROVIDER_EMPTY_OUTPUT；响应结构不兼容使用 AI_PROVIDER_INVALID_RESPONSE。

每次 Attempt 保存 Provider、模型、候选顺序、轮次、耗时、稳定错误类别和脱敏诊断。可排障字段包括客户端/供应商请求 ID、HTTP 状态、请求响应哈希、消息与字符计数、响应字节数、finish_reason 和 Token/缓存计数；不得保存完整 Prompt、历史正文、AI 输出、API Key 或原始响应体。

联网搜索

ai-chat 的联网策略：

策略 行为
disabled 不提供搜索工具
auto 由模型判断当前问题是否需要实时信息
required 至少取得一条可用结果，否则节点失败

实例总开关 ENABLE_WEB_SEARCH=false 时，auto 和 required 均不可使用。通过能力探测的 Provider 托管搜索优先；否则 Provider 必须支持 Function Calling，由进程内 AgentRunner 调用 SearXNG。

单个节点最多 4 个模型轮次和 3 次真实工具调用。一次 web_search 可以在同一工具调用内对 SearXNG 进行有限 HTTP 重试；内部重试不增加 Agent 工具调用计数。每个查询策略默认最多 2 次、每次独立使用 8 秒单请求超时，基础退避 300 毫秒；只重试超时、连接失败、HTTP 429/502/503/504。有效零结果、其他 4xx 和无效响应不重试。带 site: 的精确查询无结果时，放宽查询拥有自己的有限尝试次数，不受另一层工具总预算截断。

这些工具参数不属于工作流节点，也不通过 Docker Env 日常调整。管理员在 AI 页面的“联网搜索全局配置”统一设置 HTTP 尝试次数、单次请求超时、退避、最大结果数和失败策略；配置保存到 ai_web_search_settings，新的 Agent 执行每次从数据库读取，因此无需重启容器。搜索请求超时、结果数量、字段长度和正文长度仍有硬上限；外部网页一律是不可信材料，不能覆盖系统指令。达到工具上限时，AgentRunner 会把未执行原因返回模型，并保留一次使用现有结果完成回答的机会。

全局 failurePolicy=mode-default|fail|continue 控制本地搜索在内部重试耗尽后的行为。mode-default 下，auto 会把失败作为工具结果回填，关闭后续工具并要求模型明确说明无法联网核实；required 仍让节点失败。continue 允许 required 也降级回答，fail 则让 auto 也严格失败。Provider 托管搜索故障仍由 Provider Retry/Fallback 处理，不能被本地工具策略掩盖。

webSearchSources=full|compact|hidden 只控制聊天回复中的来源展示，不删除管理端工具轨迹。工具轨迹保存实际查询、语言与引擎参数、每次 HTTP 尝试的状态/耗时/HTTP 状态/错误码、规范化后的有限结果、结果数量和不可用引擎，不保存原始聊天正文或 SearXNG 未裁剪响应。

HTTP 200 零结果记录为 succeeded 且 outcome=no_results，不等于搜索后端宕机；required 仍会因没有可用结果失败。site: 查询没有同域结果时可以在同一预算内放宽一次，但最终仍只保留目标域名及其子域。

回复幂等与失败恢复

reply 使用 executionId + nodeId 作为内部幂等键，并把稳定 providerTempGuid 交给 BlueBubbles。明确的 429 或 5xx 可有限重试；超时和网络异常无法证明是否已发送，必须记录 unknown 并停止自动重发。

允许人工恢复的源状态为 failed、dead-lettered，或超过 STALE_RETRY_SECONDS 的 retrying。恢复会：

  1. 原子确认源状态和出站记录；
  2. 继续使用原消息、Trigger 和 Workflow Version；
  3. 创建新的 correlationId、节点尝试和出站幂等键；
  4. 通过 retryOfExecutionId 与递增 recoveryAttempt 保留链路。

仍在计划等待期的 Retry 返回冲突，不与原执行并发。存在 sending、unknown 或 confirmed 出站记录时禁止人工重试；前两者必须先确认外部事实，后者已经产生副作用。人工关闭只改变状态，不删除错误、节点、Provider Attempt、工具或发送历史。

当前 WorkflowExecutionDispatcher 是进程内实现，并由并发数、等待队列和等待时间限制容量。未来独立 Worker 必须实现同一语义和持久化原子 Claim，不能只迁移内存队列。