Skip to content

状态驱动 Agent 是怎么跑起来的 ​

这是一篇写给非后端同学的文章。不需要懂数据库、消息队列或分布式系统。 读完之后,你应该能回答三个问题:

  1. 用户点了一次“确认”之后,后台到底发生了什么?
  2. 那些术语(事件、订阅、物化、因果链、幂等)分别在说什么?
  3. 为什么不能直接“调一下 Agent”就完事?

0. 先用一句话说清楚 ​

状态驱动 Agent 就是:你写下一件“已经确认的事实”,系统去查一张规则表,发现有人关心这件事,就把对应的 Agent 叫起来干活;干完之后,把结果写回同一个地方。

这里的关键不是“Agent 很聪明”,而是**“什么时候该叫醒谁”**。 Agent 只是被叫醒之后干活的人;真正让系统运转起来的,是“事实”和“规则”。


1. 一个真实场景 ​

用户在门面里问:

“蓝牙耳机,带充电盒,出口德国,HS Code 应该归哪类?”

接下来发生的事,用户能看到的只有三件:

  1. 门面追问:“请补充产品描述、是否无线、电池类型。”
  2. 用户补充并确认。
  3. 过一会儿,出现一张结果卡片:HS 8518.30,置信度 0.9,欧盟关税 0%。

用户看不到的后台, roughly 是这样:

text
用户确认
  → 写一条“需求”事实
  → 系统发现有一条规则关心“需求”
  → 叫醒 facade-hs-code 这个 Agent
  → Agent 调模型,给出归类结果
  → 结果写回“结果”位置
  → 前端把结果读出来给用户看

下面用几个生活里的比方,把这条链路拆开。


2. 用餐厅打比方 ​

假设你开了一家餐厅。

餐厅里的东西对应系统里的东西作用
点菜单事件(Event)记录“发生了什么”,一张一张叠起来,不涂改
订单看板物化状态(Materialized State)当前这一桌点了什么,方便一眼看到
后厨规则订阅(Subscription)“只要出现凉菜单,就交给凉菜师傅”
传菜单上的桌号因果链(Causation)这道菜是被哪张单、哪一桌触发的
厨师执行器 / Agent真正动手干活的人

几个细节很关键:

  • 点菜单只追加,不修改。 客人加菜,是再写一张单,而不是把原来的单子划掉。
  • 订单看板可以随时从点菜单重算。 看板坏了、丢了,照着点菜单重新抄一遍就行。
  • 后厨规则是独立的。 今天加一条“出现生日蛋糕单就送蜡烛”,不用改厨师的做法。
  • 传菜单上有桌号。 菜上错了,能顺着单号找到是哪一桌、哪一张单。
  • 同一桌同一道菜只做一次。 客人重复确认,不应该上两盘。

这套比方,几乎一一对应下面的真实系统。


3. 把比方换回系统 ​

3.1 点菜单:state_events ​

系统里每发生一件值得记住的事,就写一条事件:

text
/tasks/{taskId}/requests/{requestId}   用户确认了需求
/tasks/{taskId}/agents/{agentId}/results/{eventId}   Agent 交付了结果
/tasks/{taskId}/agents/{agentId}/errors/{eventId}    Agent 执行失败

每条事件都带:

字段大白话例子
path这件事发生在哪个地址/tasks/T/requests/R1
value具体内容{ "product_description": "蓝牙耳机…" }
actor谁做的user:123 / agent:facade-hs-code
source从哪个入口写的api / task_node / agent_callback
seq这个任务里的第几条流水1, 2, 3…
causedByEventId被哪条事件触发上一张传菜单的单号
triggerDepth这是第几层触发0 是人工发起,1 是第一跳

为什么要只追加? 因为“当前状态”可以被改错,但“历史”不应该被悄悄改掉。 审计、追责、回放,靠的都是这份流水。

3.2 订单看板:state_entries ​

每次都从流水重算太慢,所以系统维护一份“当前值”:

text
path: /tasks/T/requests/R1
value: { product_description: "蓝牙耳机…", confirmed: true }
version: 1

它叫物化状态(materialized state)。 它只是看板,不是真相;真相永远是事件流水。 系统有 verify() 用来对账:看板必须能由流水重算出来。

3.3 后厨规则:state_subscriptions ​

一条订阅长这样:

jsonc
{
  "pattern": "/tasks/*/requests/*",          // 哪类地址
  "condition": {                              // 什么条件
    "path": "$event.value.confirmed",
    "op": "eq",
    "value": true
  },
  "action": {                                 // 叫谁干什么
    "type": "execute_agent",
    "agentId": "facade-hs-code"
  },
  "maxTriggerDepth": 3
}

翻译成人话:

只要有人往“某个任务的 requests 目录”写了一条 confirmed=true 的事件, 就把 facade-hs-code 叫起来;最多允许连锁触发 3 层。

注意这三个词:

  • pattern:地址长什么样。* 代表一段,** 代表后面任意层。
  • condition:内容满不满足。只支持等于、大于、包含这类简单比较,不能写任意代码。
  • action:命中之后做什么。目前有“跑工作流”和“叫 Agent”两种。

3.4 叫醒方式:队列 ​

如果每次叫醒都当场等 Agent 跑完,用户的请求会一直卡着。 所以系统通常把“叫醒”放进队列:

text
写事件 → 命中订阅 → 投一个 job 到队列 → 立刻返回
                                      ↓
                              后台 worker 取 job
                                      ↓
                              Agent 真正开始跑

队列用的是 Redis + Bull。job 有一个去重键:

text
agent:{eventId}:{agentId}

意思是:同一条事件,同一个 Agent,只执行一次。 即使队列重复投递,也不会重复烧 token。

3.5 干活的厨师:AgentRuntime ​

AgentRuntime 可以理解成一个“临时工”:

  • 没有长期记忆;
  • 每次只做一件事;
  • 有预算(最多几步、多少 token、多长时间);
  • 干完就释放,不占着位置。

它只关心三样东西:

text
systemPrompt + input + budget → output

它不知道状态层、订阅、队列的存在。 把“什么时候叫醒”和“醒来之后怎么干”分开,是为了两边都能独立替换。

3.6 结果回执:output 里面才是业务结果 ​

Agent 跑完后,结果路径里写的不是裸的业务结果,而是一张执行回执:

json
{
  "agentId": "facade-hs-code",
  "channel": "internal",
  "output": {
    "status": "ok",
    "summary": "蓝牙耳机归入 HS 8518.30…",
    "data": { "hs_code": "851830", "confidence": 0.9 }
  },
  "contractVersion": 2,
  "at": "2026-10-23T08:23:28.654Z"
}

所以下游读的时候要看 value.output.status,不是 value.status。 这一点很容易踩坑。


4. 一次咨询的完整旅程 ​

把上面的零件串起来:

text
① 用户说“蓝牙耳机出口德国,帮我归类”
   → 门面追问缺失字段

② 用户补充并确认
   → 门面调用工具 submit_request

③ 系统写一条事件
   /tasks/T/requests/R1
   actor=user:123, source=api, depth=0

④ 系统同步做三件事(一个数据库事务)
   - 追加事件到 state_events
   - 更新看板 state_entries
   - 如果开了异步派发,写 state_outbox

⑤ 事务提交后,派发订阅
   - 找出 enabled 的订阅
   - pattern 匹配 /tasks/*/requests/*
   - condition 检查 confirmed=true
   - 命中 1 条

⑥ 叫醒 facade-hs-code
   - 它没有外部端点 ⇒ 走平台托管的 AgentRuntime
   - 投一个 job 到队列,立刻返回
   - 前端先收到 done(注意:done 不是结果,只是一个指针)

⑦ 后台 worker 取 job
   - 读取事件快照作为输入(不是执行时再读当前值)
   - 调模型(JSON 模式)
   - 解析 + 按契约 schema 校验

⑧ 写结果
   /tasks/T/agents/facade-hs-code/results/{eventId}
   actor=agent:facade-hs-code, source=agent_callback,
   depth=1, causedBy=③ 的事件 id

⑨ 前端拿到结果
   - 同一条 SSE 连接推 result,或
   - GET /api/gateway/dialogue/sessions/{id}/result 轮询

这条链在演示里真实跑通过,事件链是:

顺序地址谁写的深度被谁触发
1requests/R1user0—
2agents/facade-hs-code/memoryagent0—
3agents/facade-hs-code/results/E1agent1第 1 条

第 2 条是 Agent 的运行时记忆,不是业务结果。 第 3 条才是用户真正要的交付。


5. 术语表 ​

下面按“学习顺序”排,不按字母。每个术语都给三样东西: 大白话 / 项目里的名字 / 一个例子。

5.1 数据:东西存在哪、怎么记 ​

事件(Event) ​

  • 大白话:一张只追加、不涂改的记录单。
  • 项目里:state_events。
  • 例子:/tasks/T/requests/R1 = { product_description: "蓝牙耳机" }。

事件溯源(Event Sourcing) ​

  • 大白话:不存“最后结果”,只存“一路发生过什么”;结果随时可以重算。
  • 项目里:state_events + replay()。
  • 例子:银行只存流水,余额是算出来的。

物化状态 / 物化视图(Materialized State / View) ​

  • 大白话:为了查得快,把流水算成一个“当前值”的看板。
  • 项目里:state_entries。
  • 例子:订单看板;看板坏了可以照点菜单重抄。

读写分离(CQRS) ​

  • 大白话:写的时候记流水,读的时候看看板,两条路分开。
  • 项目里:写走 StateService.write(),读走 state_entries。
  • 例子:记账和查账是两套动作。

路径(Path)/ 命名空间(Namespace) ​

  • 大白话:这件事存在哪个“文件夹”。
  • 项目里:/tasks/{taskId}/...;前两段是 namespace。
  • 例子:一单一个文件夹,taskId 是文件夹名。

版本(Version) ​

  • 大白话:同一条记录被改了几次。
  • 项目里:state_entries.version,从 1 开始只增。
  • 例子:合同第 1 版、第 2 版。

流水号(Sequence / seq) ​

  • 大白话:同一个任务里,事件谁先谁后。
  • 项目里:state_events.seq,按 namespace 原子递增。
  • 例子:排队小票上的号码。

乐观并发(Optimistic Concurrency) ​

  • 大白话:改之前先看版本号;别人先改了,你就失败,重来。
  • 项目里:expectedVersion 不匹配 → HTTP 409。
  • 例子:两个人同时编辑文档,后保存的人会收到“版本冲突”。

重放(Replay)/ 对账(Verify)/ 历史版本(readVersion) ​

  • 大白话:照流水重算一遍;对一下看板对不对;看看当时那一版是什么。
  • 项目里:replay() / verify() / readVersion(path, n)。
  • 例子:财务月底对账;查“签合同时看到的是哪一版”。

5.2 触发:什么时候叫醒谁 ​

订阅(Subscription) ​

  • 大白话:一条“如果有人写了某类东西,就通知我”的规则。
  • 项目里:state_subscriptions。
  • 例子:关注公众号;或者 Excel 里的条件格式。

模式(Pattern) ​

  • 大白话:地址的通配写法。
  • 项目里:* 匹配一段,** 匹配后面任意层。
  • 例子:/tasks/*/requests/* = 任何任务的任何需求。

条件(Condition) ​

  • 大白话:内容要满足什么才触发。
  • 项目里:白名单比较器,支持 eq/ne/gt/gte/lt/lte/in/contains/exists。
  • 例子:只有 confirmed=true 才触发,草稿不触发。

动作(Action) ​

  • 大白话:命中之后做什么。
  • 项目里:execute_workflow(跑工作流)或 execute_agent(叫 Agent)。
  • 例子:出现凉菜单 → 交给凉菜师傅。

事件-条件-动作(ECA,Event-Condition-Action) ​

  • 大白话:上面三件事的学术名字。
  • 项目里:订阅的三段式。
  • 例子:温度 > 60°C 且检测到烟雾 → 打开喷淋。

派发器(Dispatcher) ​

  • 大白话:拿到事件后,去查规则、挑执行者的人。
  • 项目里:StateTriggerDispatcher。
  • 例子:前台收到快递,按楼层分给不同的人。

执行器(Executor) ​

  • 大白话:规则命中后,真正被调用的那一段程序。
  • 项目里:IStateTriggerExecutor,有 workflow 和 agent 两个实现。
  • 例子:后厨的不同档口。

防环(maxTriggerDepth) ​

  • 大白话:连环触发的最大层数,防止无限套娃。
  • 项目里:订阅上的 maxTriggerDepth,默认 3。
  • 例子:多米诺骨牌最多倒 3 层;再多就强制停下。

5.3 可靠性:不丢、不重、能追责 ​

幂等(Idempotency) ​

  • 大白话:同一件事做多次,结果和做一次一样。
  • 项目里:jobId = agent:{eventId}:{agentId};submit_request 读后写去重。
  • 例子:同一张发票只能报销一次。

至少一次(At-least-once) ​

  • 大白话:消息保证送到,但可能重复送;所以接收方要会去重。
  • 项目里:state_outbox + worker 重试。
  • 例子:快递可能送两次,但收件系统按单号只入账一次。

发件箱(Transactional Outbox) ​

  • 大白话:先把“要寄的信”和“业务数据”写在同一个事务里,再异步寄;这样不会出现“数据写了、信没寄”。
  • 项目里:state_outbox,worker 每 500ms 领取。
  • 例子:公司发件箱;先登记,再让快递取走。

退避重试(Exponential Backoff) ​

  • 大白话:失败后等一会儿再试,而且越等越久。
  • 项目里:1s → 2s → 4s → …,上限 60s,超过次数标记失败。
  • 例子:打电话没人接,隔一会儿再打,而不是狂拨。

因果链(Causation) ​

  • 大白话:这件事是被哪件事触发的,一路能串起来。
  • 项目里:causedByEventId + traceCausation()。
  • 例子:传菜单上的桌号;菜上错了能找回是哪一桌。

责任方(Actor)/ 来源(Source) ​

  • 大白话:谁签的字、从哪个门进来的。
  • 项目里:actor 必填;source 区分 api/workflow_node/task_node/agent_callback。
  • 例子:报销单上的签字人和提交入口。

审计 / 取证 ​

  • 大白话:出了事,能只靠记录讲清楚“谁在什么时候做了什么”。
  • 项目里:事件流水 + actor + source + traceCausation。
  • 例子:一单交付从需求到验收的完整时间线。

5.4 AI Agent:门面、运行时、契约 ​

门面 Agent(Facade Agent) ​

  • 大白话:企业的 AI 前台。负责接待、追问、登记、转交,不负责承诺结果。
  • 项目里:没有外部端点、由平台托管的 AgentProfile。
  • 例子:餐厅门口的接待员,不是后厨。

AgentRuntime ​

  • 大白话:真正执行一次任务的“临时工”。
  • 项目里:AgentRuntime.run(),无会话、有预算、跑完释放。
  • 例子:叫来的临时厨师,做完这道菜就走。

契约(Contract) ​

  • 大白话:这门面接什么单、必填什么、结果写到哪、什么条件下被叫醒。
  • 项目里:AgentProfile.stateContract,编译成订阅。
  • 例子:点菜单的格式说明。

JSON Schema ​

  • 大白话:字段规则:哪些必填、类型是什么、允不允许额外字段。
  • 项目里:Ajv 校验,validateContractValue()。
  • 例子:报销单必须有金额和发票号。

结果信封(Envelope) ​

  • 大白话:统一的结果外壳,让大家知道去哪找“结论”和“明细”。
  • 项目里:{ status, summary, data, follow_up?, confidence?, text? }。
  • 例子:快递盒外面的面单,里面才是商品。

工具调用(Tool Calling) ​

  • 大白话:模型可以调用平台给它的工具,而不是只会聊天。
  • 项目里:submit_request(提交需求)、use_skill(选择技能)。
  • 例子:前台可以按“提交”按钮,而不是只说“好的”。

结构化输出(Structured Output) ​

  • 大白话:要求模型必须返回 JSON,而不是一段自由文本。
  • 项目里:response_format: { type: 'json_object' } + 平台侧校验。
  • 例子:填表,而不是写作文。

SSE ​

  • 大白话:服务器像直播一样持续往前端推消息,不用一直问“好了吗”。
  • 项目里:start → delta* → done → [result]。
  • 例子:外卖骑手的实时位置,而不是每分钟刷新一次。

A2A / Agent Card ​

  • 大白话:让别的 Agent 能找到你、知道你能做什么、怎么调用你。
  • 项目里:GET /api/a2a/agent-card/:agentId。
  • 例子:企业的电子名片 + 服务目录。

6. 如果没有这些机制,会发生什么 ​

如果只做…会出现什么问题
只写数据库,不做订阅需求写进去了,但没人被叫醒;用户等到天荒地老
直接在一个接口里调 Agent请求会卡很久;失败没法重试;不知道是谁触发的
只用消息队列消息可能丢;查不到“这一单现在到哪了”;无法按任务取证
不做幂等用户重复确认,Agent 重复执行,重复烧 token
不做防环A 触发 B,B 又触发 A,无限循环
不记 actor出了问题不知道是谁做的、谁该负责
不做结构化输出下游拿到一段自然语言,没法按字段读取
不做版本验收之后有人改了交付,无法证明“当时是什么”

这些机制不是为了“看起来架构很复杂”,而是为了让系统在出问题时说得清、停得住、查得到。


7. 四个常见误解 ​

误解 1:写了状态,就会触发 Agent ​

不一定。 只有存在 enabled 的订阅、并且 pattern + condition 都命中,才会触发。 没有订阅,写入只是留痕。

误解 2:done 就是结果 ​

不是。 done 只是“这一轮对话结束了”,里面有一个 submission 指针,指向结果应该出现的位置。 真正的结果是后面的 result 事件,或者 GET .../result 读到的内容。

误解 3:门面上架了,就自动能被状态唤醒 ​

不是。 **上架(active)**只代表它能被看到、能对话; **发布订阅(publish)**才代表“状态写入会叫醒它”。 这两件事是分开的。

误解 4:对话原文会进状态层 ​

不会。 状态层只写结构化事实、指针和确认人。 对话原文留在会话里,避免隐私泄露和回放噪声。


8. 一张图总结 ​

text
用户确认
   │
   ▼
┌─────────────────────────────┐
│ StateService.write()        │
│  ① 追加事件 state_events     │
│  ② 更新看板 state_entries    │
│  ③ 写发件箱 state_outbox     │
└──────────────┬──────────────┘
               │ 事务提交后
               ▼
┌─────────────────────────────┐
│ StateTriggerDispatcher      │
│  pattern → condition → 命中  │
└──────────────┬──────────────┘
               │
       ┌───────┴────────┐
       ▼                ▼
 execute_workflow   execute_agent
       │                │
       │                ▼
       │        ┌───────────────┐
       │        │ 队列 / 后台 worker │
       │        └───────┬───────┘
       │                ▼
       │        ┌───────────────┐
       │        │ AgentRuntime  │
       │        │ 调模型 + 校验  │
       │        └───────┬───────┘
       └────────────────┤
                        ▼
              写结果 /tasks/.../results/{eventId}
              actor=agent, causedBy=触发事件, depth+1
                        │
                        ▼
              前端读结果 / SSE result

9. 今天真实在跑什么 ​

机制存在,不等于默认在跑。当前默认状态是:

行为默认是否发生
任务执行写 batches / steps✅ 会(留痕与恢复依据)
外部 Agent 回传写交付✅ 会
门面对话写 requests✅ 会(有 taskId + 契约时)
状态写入唤醒门面❌ 只有存在 enabled 订阅且命中时
26 个行业门面进入市场❌ 默认 draft
门面订阅自动发布❌ 必须显式 --activate --publish

两个最直接的观察点:

  • submit_request 返回的 triggered 数量;
  • 数据库里 state_subscriptions 的 enabled 数量。

如果这两个是 0,说明链路没有启动,不是“下游慢”。


10. 自己动手看 ​

bash
# 1) 看有多少启用的订阅(0 ⇒ 不会被状态唤醒)
mongosh "<mongo-uri>" --eval 'db.state_subscriptions.countDocuments({enabled:true})'

# 2) 看某个任务的事件流水
mongosh "<mongo-uri>" --eval '
  db.state_events.find({taskId:"<taskId>"},{path:1,source:1,actor:1,triggerDepth:1,causedByEventId:1}).sort({seq:1})
'

# 3) 跑状态层的真库 e2e(会自建自清)
npx jest --config ./apps/hushwork/test/jest-e2e.json --runInBand \
  --testPathPatterns="state-layer"

# 4) 跑门面对话闭环
npx jest --config ./apps/hushwork/test/jest-e2e.json --runInBand \
  --testPathPatterns="facade-dialogue"

11. 这些名词借自哪些成熟技术 ​

这套机制没有发明新概念,基本是把几个成熟模型拼在一起:

借自借了什么在本项目里
Event Sourcing只追加事件、可重放state_events、replay()
CQRS读写分离事件写入 vs state_entries 读取
物化视图当前值看板state_entries
Pub/Sub发布订阅state_subscriptions
ECA 规则事件-条件-动作pattern + condition + action
Transactional Outbox先记账再寄信state_outbox + worker
Saga / Process Manager多跳编排触发链 + causedBy + maxTriggerDepth
分布式追踪因果链traceCausation()
Ports & Adapters插座标准IStateTriggerExecutor 注册表
JSON Schema字段校验Ajv
Tool Calling模型调工具submit_request / use_skill
SSE服务端推送start/delta/done/result

没有用 Kafka、Temporal、EventStoreDB、NATS 这些中间件。 当前实现是用已有的 MongoDB + Redis + 自写代码拼出来的,重点是“够用、能查、能追责”,不是“组件最先进”。


12. 最后记住三句话 ​

  1. 只记流水,余额可以重算。
  2. 订阅就是 IFTTT:地址 + 条件 → 动作。
  3. 每次执行都有单号、有防环、有幂等键。

如果这三句话记住了,剩下的术语都只是它们的展开。

Released under the MIT License.