LobeHub Agent Signal 实战指南：构建事件驱动的 Agent 后台管线

在 LobeHub 这类"Chief Agent Operator"系统中，Agent 的记忆写入、技能合成、夜间复盘等后台工作绝不能阻塞前台聊天请求。Agent Signal 就是 LobeHub 用来实现这一点的核心机制：它把"发生了什么事"（source）、"这意味着什么"（signal）和"该做什么副作用"（action）拆成三段式管线，配合去重、scope 锁、异步 Workflow 交接与可观测性投影，让 Agent 团队可以 7×24 小时安静地自我迭代。读完本文，你将掌握 Agent Signal 的四种入口函数选择、完整实现流程、handler 编写模式，以及如何排查"信号发出但没有动作"这类典型问题。

一、运行时形态：一条固定的三段式管线

Agent Signal 只有一种一致的运行时形状（引自 SKILL.md）：

source event -> signal interpretation -> action execution -> built-in result signals

而对于"持久化自我迭代"（durable self-iteration，即记忆/技能的后台写入）这类工作，还多出一条完成分支：

memory/skill action -> execAgent enqueue -> agent.execution.completed -> selfIteration receipt projection

用更完整的视角看（引自 架构参考），整条链路是：

producer
  -> emitAgentSignalSourceEvent(...) or enqueueAgentSignalSourceEvent(...)
    -> emitSourceEvent(...)
      -> dedupe + scope lock + source normalization
        -> runtime.emitNormalized(source)
          -> source handlers
            -> signal handlers
              -> action handlers
                -> built-in result signals
                  -> observability projection + persistence

调度器（scheduler）是"队列驱动"的，而不是为某一个 policy 硬编码的：source 节点被匹配的 source handler 处理，handler 向队列分发 signals/actions，signal handler 再分发更多 signals/actions，最终 action handler 产出 ExecutorResult，由调度器转成内置结果信号 signal.action.applied | skipped | failed。这种结构保证了新增一条业务链路只需要注册新 handler，而不用改动运行时内核。

二、四个入口函数：按执行时机选对入口

服务端为 Agent Signal 提供了四个入口（引自 SKILL.md），选择逻辑取决于"谁拥有执行时机"：

入口函数	适用场景
emitAgentSignalSourceEvent(...)	服务端生产方希望立即执行管线时使用
executeAgentSignalSourceEvent(...)	worker 或受控后端路径已经拥有执行时机，并且可能需要注入 runtime guard backend 时使用
enqueueAgentSignalSourceEvent(...)	调用方希望快速返回，把事件交给 Upstash Workflow 在带外（out-of-band）处理时使用
emitAgentSignalSourceEventWithStore(...)	隔离的测试或 eval 场景，希望绕开环境级（ambient）Redis 状态时使用

从源码可以印证前两者的差异。服务端 emitter 中的 emitAgentSignalSourceEvent 会先通过 isAgentSignalEnabledForUser(db, userId) 做用户级特性门控，未启用时直接返回 undefined，随后在正常去重存储边界内执行策略。而 workflow 触发器 的 AgentSignalWorkflow.triggerRun 则按 appEnv.enableQueueAgentRuntime 分流：

• Queue 模式：通过 workflowClient.trigger(...) 把负载投递到 /api/workflows/agent-signal/run，获得 Upstash Workflow 提供的持久化执行、重试与流控；
• Local 模式：scheduleLocalAgentSignalRun(...) 用 setTimeout 延迟一次进程内执行，返回一个合成（synthetic）的 workflowRunId——它不提供持久化和重试，但通过进程级的 source-event store 与 runtime guard 保住了去重、scope 锁、时间窗与 action 幂等语义。

值得注意的实现细节：触发 workflow 时，代码会把当前活跃 trace 头通过 SDK 的 headers 选项注入（底层被改写成 Upstash-Forward-* 头），否则 Upstash Workflow/QStash 不会把用户头转发到 workflow 目的地；此外 flow control key 会先做 value.replaceAll(/[^\w.-]/g, '_') 归一化，因为 Upstash 流控 key 会拒绝 : 等 scope-key 分隔符。这些约束说明"交接给异步运行时"并非简单的一次 HTTP 调用，trace 连续性与 scope key 稳定性都是必须处理的工程问题。

入口代码主要位于：

• 服务端入口（re-export emitter）
• Workflow 触发器
• Workflow 执行器

三、核心模型：Source / Signal / Action / Policy 的严格边界

Agent Signal 用四个概念描述整条管线（SKILL.md 的定义）：

• source：一个被归一化的"已发生的事实"。生产方包括 runtime 生命周期事件、用户消息、bot 消息入口等；
• signal：从一个 source（或另一个 signal）推导出的语义解释。signal 表达的是"含义、路由或策略状态"；
• action：由一个 signal 规划出的具体副作用。action 是真正干活的地方；
• policy：可安装的中间件束（middleware bundle），负责注册 source、signal、action 三类 handler；
• procedure：不是一个独立的运行时节点。应把 "procedure" 理解为某个用例的端到端流程：入口 source、匹配的 handler、规划的 action、执行结果与可观测性。

边界纪律是这套模型的关键——什么时候加哪种节点：

场景	应新增
外部世界产生了一个新事件	新的 source
系统需要一个可复用的语义解释	新的 signal
运行时需要一个具体的副作用	新的 action
把上述部件接线组装	新增或更新 policy

如果某个 handler 既在解释语义又在做副作用，应把它拆开——这是保持链路可检查、可测试的前提。

3.1 共享语义内核：packages/agent-signal

仓库中 @lobechat/agent-signal 是"共享语义核心"包。它的约束很明确：只定义 Agent Signal 节点与 source 事件的语言，不执行 policy、不调用模型运行时、不读数据库、不触发 workflow、不导入 app server 模块。它提供：

• 基础节点类型：source、signal、action；
• 构建器：createSource、createSignal、createAction；
• 共享的 source 类型与负载目录；
• source 事件信封与 scope 辅助函数；
• 内置结果信号类型；
• 运行时结果契约，如 RuntimeProcessorResult 与 ExecutorResult。

核心代码集中在 sourceTypes.ts、sourceEvent.ts、scopeKey.ts 与 builders.ts。

3.2 服务端实现层与 OTEL 层

与纯语义包不同，apps/server/src/services/agentSignal 是"服务端拥有的实现层"，负责：source 归一化/水合/渲染、policy 专属的 signal 与 action 目录、中间件注册、运行时调度与 guard 后端、基于 Redis 的去重/waypoint/策略状态、同步与异步执行入口、以及自我迭代完成后的回执投影。

packages/observability-otel/src/modules/agent-signal/index.ts 则共享 OTEL 归属权，提供 tracer 与全套指标仪器，避免各处自造 feature-local 的 meter。

四、Source 事件目录：16 种内置事件及其负载

packages/agent-signal 定义了完整的 source 事件目录，AGENT_SIGNAL_SOURCE_TYPES 常量与按类型键控的负载 AgentSignalSourcePayloadMap 都在 sourceTypes.ts 中。主要类别：

Source 事件	常量	主要生产方	负载意图
agent.execution.completed	agentExecutionCompleted	Agent Runtime	服务端 agent 执行完成，携带 operation、step、topic 与上下文元数据
agent.execution.failed	agentExecutionFailed	Agent Runtime	服务端 agent 执行失败，携带 reason、error 等元数据
agent.user.message	agentUserMessage	Workflow bridge、Bot router	供策略分析记忆/提示词/文档/技能变化的用户反馈消息
bot.message.merged	botMessageMerged	Bot router	已合并进会话 scope 的 bot 平台消息
runtime.before_step / runtime.after_step	runtimeBeforeStep / runtimeAfterStep	Agent Runtime	服务端运行时步骤开始/结束
client.runtime.start / client.runtime.complete	clientRuntimeStart / clientRuntimeComplete	浏览器聊天流式执行器	浏览器聊天运行时启动/完成；workflow 可将 start 桥接为 agent.user.message
client.gateway.stream_start / step_complete / runtime_end / error	clientGateway*	浏览器网关事件处理器	浏览器网关流的各阶段事件

源码中还存在若干文档表格之外的 source 类型（如 agent.nightly_review.requested、agent.self_reflection.requested、agent.self_feedback_intent.declared、tool.outcome.completed/failed），从 sourceTypes.ts 可以看出它们是夜间复盘、自我反思、自我反馈意图声明与工具结果投影的入口。

两个负载细节值得注意：

1. agent.execution.completed 的负载中带 selfIteration?: unknown 字段——这是执行器为内置后台 agent 附加的"不透明完成副作用负载"（例如自我迭代的工具结果，用于回执投影），由生产层负责其形状，管线原样携带；
2. agent.user.message 的负载支持 intents（document | memory | persona | prompt | skill 数组）、memoryPayload、anchorMessageId、triggerMessageId 等字段，前者用于意图预判，后两者用于把回执/UI 挂到正确的消息节点上。

浏览器侧白名单：AGENT_SIGNAL_CLIENT_SOURCE_TYPES 只允许 client.* 六种 source 事件通过鉴权的 lambda 路由从浏览器进入（sourceTypes.ts 末尾的常量），其余 source 类型必须由服务端生产方发出。

4.1 构造 source 事件与 scope key

createSourceEvent 会自动补齐两个字段（README）：

• scopeKey：优先取 topicId，其次 bot 线程元数据，最后回退 fallback:global；
• timestamp：调用方未提供时取 Date.now()。

import {
  AGENT_SIGNAL_SOURCE_TYPES,
  createSourceEvent,
} from '@lobechat/agent-signal/source';

const event = createSourceEvent({
  payload: {
    message: 'Please remember that I prefer concise answers.',
    messageId: 'msg_1',
    topicId: 'topic_1',
  },
  sourceId: 'source_1',
  sourceType: AGENT_SIGNAL_SOURCE_TYPES.agentUserMessage,
});

scope key 也可独立构造：getSourceEventScopeKey({ topicId: 'topic_1' }) 处理 source 事件负载，AgentSignalScopeKey.forBotThread({ applicationId, platform, platformThreadId }) 处理结构化 scope 元数据（scopeKey.ts）。

scopeKey 是关联后台工作的串行化边界，它同时承担四个职责（架构参考）：source 去重时间窗、source 生成期间的 scope 锁、运行时 guard 状态、队列处理的 waypoint 持久化。因此实现规则明确要求：当同一 source 可能重复到达时，使用稳定的 id 与幂等 key，并保持 scopeKey 在重试与 workflow 交接之间稳定。

五、实现新链路：十步工作流

LobeHub 给出了一套固定的实现流程（SKILL.md），新增一条 Agent Signal 链路时应逐步对照：

1. 判断用例是同步工作还是安静的后台工作（决定用 emit 还是 enqueue）；
2. 在 packages/agent-signal/src/source/sourceTypes.ts 中定义或复用 source 类型；
3. 在 apps/server/src/services/agentSignal/policies/types.ts 中定义或复用 signal 与 action 类型；
4. 用 defineSourceHandler、defineSignalHandler 或 defineActionHandler 实现 handler；
5. 若生产方负载需要整形，在 apps/server/src/services/agentSignal/sources/** 下添加 source 归一化或水合；
6. 用 defineAgentSignalHandlers(...) 把 handler 束成 policy；
7. 在 apps/server/src/services/agentSignal/policies/index.ts 注册 policy，必要时传入运行时工厂；
8. 添加或更新发起 source 事件 emit/enqueue 的入口代码；
9. 对异步自我迭代写入，打上 Agent Signal operation marker，并从完成路径投影用户可见回执；
10. 在认为流程完成之前，补齐可观测性与测试。

其中第 2、3 步的"先复用后新增"是硬性规则：新增 source/signal/action 类型之前先复用现有类型，因为 source 类型字符串会被持久化进 trace、去重 key 与 workflow 负载，改名即破坏兼容性。

六、Handler 编写模式：注册 API 与返回契约

6.1 流式注册 API

middleware 辅助函数位于 apps/server/src/services/agentSignal/runtime/middleware.ts，提供四个函数：defineSourceHandler(...)、defineSignalHandler(...)、defineActionHandler(...)、defineAgentSignalHandlers(...)。它们的作用有二：让 handler 注册保持简洁，并且在 listen 指向具体 source/signal/action 类型时保留强类型推导。

每个 handler 接收当前运行时节点与 RuntimeProcessorContext，后者提供 scopeKey、now()、runtimeState.getGuardState(lane)、runtimeState.touchGuardState(lane, now?)（context.ts）。

6.2 五种返回契约

handler 的返回值决定了链路如何继续（handlers 参考）：

返回值	含义
void	不扇出，到此 handler 停止
{ status: 'dispatch', signals?, actions? }	向队列继续分发 signal/action
{ status: 'wait', pending? }	暂停，等待宿主后续协调
{ status: 'schedule', nextHop }	调度另一跳
{ status: 'conclude', concluded? }	以终端运行时结果停止
ExecutorResult	仅限执行了具体副作用的 action handler

6.3 三类 handler 的标准形态

Source handler——把生产方事件解释成语义信号（参考 feedbackSatisfaction.ts）：

return defineSourceHandler(
  AGENT_SIGNAL_SOURCE_TYPES.agentUserMessage,
  'agent.user.message:my-handler',
  async (source, ctx): Promise<RuntimeProcessorResult | void> => {
    // 解释 source 负载，可选使用 ctx.runtimeState
    return {
      signals: [/* 一个或多个语义信号 */],
      status: 'dispatch',
    };
  },
);

适用条件：原始消息、生命周期事件或 bot 入口需要解释，但工作仍属于"语义"而非"副作用"。

Signal handler——把一个语义状态分支成更多语义状态或规划的 action（参考 feedbackDomain.ts、feedbackAction.ts），用于路由、扇出、过滤、冲突消解：

return defineSignalHandler(
  MY_SIGNAL_TYPE,
  'signal.my-policy-router',
  async (signal): Promise<RuntimeProcessorResult | void> => {
    return {
      actions: [/* 规划出的工作 */],
      status: 'dispatch',
    };
  },
);

Action handler——执行（或入队）真正的副作用（参考 actions/userMemory.ts）：

return defineActionHandler(
  MY_ACTION_TYPE,
  'action.my-policy-executor',
  async (action, ctx): Promise<ExecutorResult> => {
    // 执行服务/工具/模型副作用；如需要，先做幂等检查
    return {
      actionId: action.actionId,
      attempt: {
        completedAt: ctx.now(),
        current: 1,
        startedAt,
        status: 'succeeded',
      },
      status: 'applied',
    };
  },
);

Action handler 的规则：幂等检查放在这里或紧邻副作用之前；返回稳定的 actionId；失败细节写进 error；让调度器把 ExecutorResult 转成内置结果信号，而不是自己发结果信号。

6.4 Policy 组装

policy 是"把通用运行时变成具体特性"的组装单元。以 analyzeIntent 为例（policies/analyzeIntent/index.ts）：

return defineAgentSignalHandlers([
  ...(options.procedure ? [createToolOutcomeSourceHandler(options.procedure)] : []),
  createFeedbackSatisfactionJudgeProcessor(...),
  createFeedbackDomainJudgeSignalHandler(...),
  createFeedbackActionPlannerSignalHandler(),
  defineSkillManagementActionHandler(...),
  createCompletionSkillSynthesisSourceHandler(...),
  defineUserMemoryActionHandler(...),
]);

该束随后经由 createDefaultAgentSignalPolicies(...) 与 createAgentSignalRuntime({ policies }) 传入运行时（policies/index.ts）。

七、参考链路剖析：analyzeIntent 政策

analyzeIntent 是官方推荐的参考实现，其主链路为（架构参考）：

agent.user.message
  -> feedback satisfaction source handler
    -> signal.feedback.satisfaction
      -> feedback domain signal handler
        -> signal.feedback.domain.*
          -> feedback action planner
            -> action.user-memory.handle | action.skill-management.handle
              -> signal.action.applied | skipped | failed

即：用户消息进入 → 满意度判断 source handler 产出 signal.feedback.satisfaction → 领域信号 handler 细分出 signal.feedback.domain.memory | prompt | skill 等语义信号 → 动作规划器规划出 action.user-memory.handle 或 action.skill-management.handle → 执行并产生内置结果信号。

该政策还包含可选的工具结果投影、技能管理、延迟的完成期技能合成（completion skill synthesis）、夜间复盘与完成扇出：

action.user-memory.handle | action.skill-management.handle
  -> enqueue execAgent with AgentSignal marker
    -> agent.execution.completed
      -> completion policy
        -> buildSelfIterationReceipts(...)
          -> memory | skill | review receipts

两条关键设计决策：

1. 入队动作只报告入队结果。action.user-memory.handle / action.skill-management.handle 的具体副作用是 enqueueSelfIterationRun(...)——把一次带 Agent Signal operation marker 的 execAgent 运行入队。立即的运行时链路仍会为入队结果发出 signal.action.applied | skipped | failed，但用户可见的记忆/技能回执必须从 agent.execution.completed 投影，不能从入队动作投影（SKILL.md 的实现规则明确禁止"从 enqueue action 投影回执"）。
2. 完成期技能合成可"先搁置、后恢复"：一条用户消息带来的技能候选可以在前台链路期间被搁置（parked），等到 agent.execution.completed 之后、完整轨迹与工具结果都可用时再恢复执行，从而避免基于不完整上下文做技能合成。

回执投影的实现位于 buildSelfIterationReceipts.ts 与 selfIterationCompletionHandler.ts，完成策略工厂 createCompletionPolicy(...) 位于 completionPolicy.ts。

八、异步交接：enqueue 路径与 AGENT_RUNTIME_MODE

当"UI 请求应立即结束、policy 在后台运行"时，应使用 enqueueAgentSignalSourceEvent(...)。该路径（架构参考）依次：

1. 归一化 source 信封；
2. 派生或复用 scopeKey；
3. 触发 AgentSignalWorkflow；
4. 之后在 runAgentSignalWorkflow（run.ts）中延迟执行。

交付方式跟随 AGENT_RUNTIME_MODE：

• Queue 模式：Upstash Workflow 提供持久化执行、重试与流控；
• Local 模式：setTimeout 延迟进程内执行，返回合成 workflowRunId，不提供持久化与重试；但通过进程级 source-event store 与 runtime guard 维持去重、scope 锁、时间窗与 action 幂等，在服务端进程生命周期内保持功能完整。

这条路径是"安静的后台工作"的首选：前台响应路径短，policy 异步跑完，且 scope 纪律保证同一 scope 内的关联工作被串行化。

九、可观测性：投影管线与调试手册

9.1 OTEL 仪器

packages/observability-otel/src/modules/agent-signal/index.ts 提供共享 tracer 与全套指标：tracer、sourceCounter、signalCounter、actionCounter、actionResultCounter、chainCounter、signalActionTransitionCounter、chainDurationHistogram、actionDurationHistogram、sourceEventCounter、sourceEventDurationHistogram、workflowRunCounter、workflowRunDurationHistogram、handlerCounter、handlerDurationHistogram、terminalResultCounter。需要共享遥测归属时用它，不要自建 feature-local 的 meter/tracer。

9.2 投影管线与检查顺序

运行时执行结束后，服务从完整链路投影出一个紧凑的可观测模型（observability/projector.ts、observability/traceEvents.ts）：

• 一个 trace 信封，含 source、signals、actions、results、edges 与 handler runs；
• 一条紧凑遥测记录，含主导路径（dominant path）、状态分布与链路元数据。

投影基于 source 节点、已发出 signals、已规划 actions 与执行器结果构建；toAgentSignalTraceEvents(...) 可把链路拍平成紧凑事件记录。检查一条链路时按固定顺序：source 类型与负载 → 发出的 signals → 规划的 actions → 执行器结果 → 投影 edges 与主导路径。

注意：回执投影与可观测投影是两回事。orchestrator 在立即运行时链路之后仍会调用 projectAgentSignalReceipts(...)，但记忆/技能自我迭代回执从 agent.execution.completed 经 buildSelfIterationReceipts(...) 投影并经回执 store 持久化。

9.3 工作流快照桥

workflow 触发的运行天然不经过前台运行时快照路径，因此 runAgentSignalWorkflow 加了一个仅开发环境的桥接，把 trace 写入 .agent-tracing/。当 source 是用 enqueueAgentSignalSourceEvent(...) 入队、又需要本地看到"安静后台工作"的 trace 时，走这条路径。

9.4 常见问题排查

可观测性参考 给出了四类典型故障的排查清单：

Source 发出了但什么都没发生：检查用户级特性门控是否开启、source 类型是否有匹配的已注册 source handler、去重或 scope 锁是否短路了生成。入口看 index.ts 与 sources/index.ts。

Signal 存在但没有 action 运行：检查该 signal 类型是否有已注册的 signal handler、handler 是否返回 status: 'dispatch'、是否真的返回了 actions。

Action 运行了两次：检查 source 去重 key 的稳定性、action 幂等策略、scope key 在重试与 workflow 交接过程中的稳定性。参考 actionIdempotency.ts 与 actions/userMemory.ts。

Action 已 applied 但记忆/技能回执缺失（自我迭代链路特有的六项检查）：action handler 是否入队了 execAgent 并打上 Agent Signal operation marker；agent.execution.completed 是否携带了运行 finalState 的 selfIteration 负载；createCompletionPolicy(...) 是否接上了 onSelfIterationCompleted；buildSelfIterationReceipts(...) 是否识别了持久化变更的 api 名；回执 store 是否接受了确定性回执 id；当 UI 定位重要时 marker 是否带了可用的 anchorMessageId 或 triggerMessageId。

9.5 完成前最小检查清单

• source 入口可测试；
• handler 注册可从 policy 工厂发现；
• action 执行器返回结构化结果；
• 投影干净地包含新路径；
• 异步自我迭代回执（如适用）在完成路径上被覆盖；
• 测试至少覆盖一条正常路径与一条 no-op 或失败路径。

十、测试与默认阅读集

测试放在被触碰代码的旁边，参考模式包括 AgentSignalRuntime.test.ts、index.integration.test.ts、analyzeIntent 政策测试 下的 __tests__/*，以及 selfIteration 完成投影测试。原则是在能证明行为的最小层级测试：单条路由规则用 handler 单测、队列扇出用运行时测试、异步记忆/技能回调用完成投影测试、服务入口与可观测持久化用集成测试。

开发 Agent Signal 功能时的默认阅读集（SKILL.md 的 "Default Reading Set"）：

• 共享语义核心：packages/agent-signal/src/index.ts、base/builders.ts、base/types.ts、source/sourceTypes.ts、source/sourceEvent.ts；
• 服务端运行时与中间件：runtime/AgentSignalRuntime.ts、runtime/AgentSignalScheduler.ts、runtime/middleware.ts、runtime/context.ts；
• 现成 policy 示例：policies/analyzeIntent/index.ts、feedbackSatisfaction.ts、feedbackDomain.ts、feedbackAction.ts、actions/userMemory.ts、actions/skillManagement.ts、completionSkillSynthesis.ts、completionPolicy.ts、buildSelfIterationReceipts.ts、selfIterationCompletionHandler.ts；
• 可观测性：observability/projector.ts、observability/traceEvents.ts、packages/observability-otel/src/modules/agent-signal/index.ts。

总结

Agent Signal 的价值在于用一套小而严格的词汇（source/signal/action/policy）把 Agent 的后台自我迭代从聊天主链路中解耦出来：入口按执行时机四选一，scopeKey 保证关联工作串行且幂等，queue/local 两种运行时模式让"安静后台工作"既能在生产上获得持久化与重试、也能在无 Redis 的本地环境保持功能完整，而 agent.execution.completed + selfIteration 回执投影机制则把"入队成功"与"工作真正完成"明确区分开。按本文的十步实现流程与四类故障排查清单，可以在不触碰运行时内核的前提下，为 LobeHub 安全地新增任意一条事件驱动的 Agent 后台管线。