状态驱动 Agent 是怎么跑起来的
这是一篇写给非后端同学的文章。不需要懂数据库、消息队列或分布式系统。 读完之后,你应该能回答三个问题:
- 用户点了一次“确认”之后,后台到底发生了什么?
- 那些术语(事件、订阅、物化、因果链、幂等)分别在说什么?
- 为什么不能直接“调一下 Agent”就完事?
0. 先用一句话说清楚
状态驱动 Agent 就是:你写下一件“已经确认的事实”,系统去查一张规则表,发现有人关心这件事,就把对应的 Agent 叫起来干活;干完之后,把结果写回同一个地方。
这里的关键不是“Agent 很聪明”,而是**“什么时候该叫醒谁”**。 Agent 只是被叫醒之后干活的人;真正让系统运转起来的,是“事实”和“规则”。
1. 一个真实场景
用户在门面里问:
“蓝牙耳机,带充电盒,出口德国,HS Code 应该归哪类?”
接下来发生的事,用户能看到的只有三件:
- 门面追问:“请补充产品描述、是否无线、电池类型。”
- 用户补充并确认。
- 过一会儿,出现一张结果卡片:HS
8518.30,置信度0.9,欧盟关税0%。
用户看不到的后台, roughly 是这样:
用户确认
→ 写一条“需求”事实
→ 系统发现有一条规则关心“需求”
→ 叫醒 facade-hs-code 这个 Agent
→ Agent 调模型,给出归类结果
→ 结果写回“结果”位置
→ 前端把结果读出来给用户看下面用几个生活里的比方,把这条链路拆开。
2. 用餐厅打比方
假设你开了一家餐厅。
| 餐厅里的东西 | 对应系统里的东西 | 作用 |
|---|---|---|
| 点菜单 | 事件(Event) | 记录“发生了什么”,一张一张叠起来,不涂改 |
| 订单看板 | 物化状态(Materialized State) | 当前这一桌点了什么,方便一眼看到 |
| 后厨规则 | 订阅(Subscription) | “只要出现凉菜单,就交给凉菜师傅” |
| 传菜单上的桌号 | 因果链(Causation) | 这道菜是被哪张单、哪一桌触发的 |
| 厨师 | 执行器 / Agent | 真正动手干活的人 |
几个细节很关键:
- 点菜单只追加,不修改。 客人加菜,是再写一张单,而不是把原来的单子划掉。
- 订单看板可以随时从点菜单重算。 看板坏了、丢了,照着点菜单重新抄一遍就行。
- 后厨规则是独立的。 今天加一条“出现生日蛋糕单就送蜡烛”,不用改厨师的做法。
- 传菜单上有桌号。 菜上错了,能顺着单号找到是哪一桌、哪一张单。
- 同一桌同一道菜只做一次。 客人重复确认,不应该上两盘。
这套比方,几乎一一对应下面的真实系统。
3. 把比方换回系统
3.1 点菜单:state_events
系统里每发生一件值得记住的事,就写一条事件:
/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
每次都从流水重算太慢,所以系统维护一份“当前值”:
path: /tasks/T/requests/R1
value: { product_description: "蓝牙耳机…", confirmed: true }
version: 1它叫物化状态(materialized state)。 它只是看板,不是真相;真相永远是事件流水。 系统有 verify() 用来对账:看板必须能由流水重算出来。
3.3 后厨规则:state_subscriptions
一条订阅长这样:
{
"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 跑完,用户的请求会一直卡着。 所以系统通常把“叫醒”放进队列:
写事件 → 命中订阅 → 投一个 job 到队列 → 立刻返回
↓
后台 worker 取 job
↓
Agent 真正开始跑队列用的是 Redis + Bull。job 有一个去重键:
agent:{eventId}:{agentId}意思是:同一条事件,同一个 Agent,只执行一次。 即使队列重复投递,也不会重复烧 token。
3.5 干活的厨师:AgentRuntime
AgentRuntime 可以理解成一个“临时工”:
- 没有长期记忆;
- 每次只做一件事;
- 有预算(最多几步、多少 token、多长时间);
- 干完就释放,不占着位置。
它只关心三样东西:
systemPrompt + input + budget → output它不知道状态层、订阅、队列的存在。 把“什么时候叫醒”和“醒来之后怎么干”分开,是为了两边都能独立替换。
3.6 结果回执:output 里面才是业务结果
Agent 跑完后,结果路径里写的不是裸的业务结果,而是一张执行回执:
{
"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. 一次咨询的完整旅程
把上面的零件串起来:
① 用户说“蓝牙耳机出口德国,帮我归类”
→ 门面追问缺失字段
② 用户补充并确认
→ 门面调用工具 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 轮询这条链在演示里真实跑通过,事件链是:
| 顺序 | 地址 | 谁写的 | 深度 | 被谁触发 |
|---|---|---|---|---|
| 1 | requests/R1 | user | 0 | — |
| 2 | agents/facade-hs-code/memory | agent | 0 | — |
| 3 | agents/facade-hs-code/results/E1 | agent | 1 | 第 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. 一张图总结
用户确认
│
▼
┌─────────────────────────────┐
│ 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 result9. 今天真实在跑什么
机制存在,不等于默认在跑。当前默认状态是:
| 行为 | 默认是否发生 |
|---|---|
任务执行写 batches / steps | ✅ 会(留痕与恢复依据) |
| 外部 Agent 回传写交付 | ✅ 会 |
门面对话写 requests | ✅ 会(有 taskId + 契约时) |
| 状态写入唤醒门面 | ❌ 只有存在 enabled 订阅且命中时 |
| 26 个行业门面进入市场 | ❌ 默认 draft |
| 门面订阅自动发布 | ❌ 必须显式 --activate --publish |
两个最直接的观察点:
submit_request返回的triggered数量;- 数据库里
state_subscriptions的enabled数量。
如果这两个是 0,说明链路没有启动,不是“下游慢”。
10. 自己动手看
# 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. 最后记住三句话
- 只记流水,余额可以重算。
- 订阅就是 IFTTT:地址 + 条件 → 动作。
- 每次执行都有单号、有防环、有幂等键。
如果这三句话记住了,剩下的术语都只是它们的展开。