- 新增 tests/test_engine_event.py (31 用例: EngineEvent 线格式/to_wire/from_wire, SseAdapter 裁剪+帧格式+frames, BufferAdapter 累积/媒体/诊断/截断, Diagnosis, test_dual_claw_wiring_source_guard 校验 chat.py/im_llm.py 接线) - 新增 docs/dual-claw-event-contract.md (#873 契约文档 7 节) - 全量 pytest 64 用例全绿 (含 test_dual_claw/test_node_registry 无回归) - 仅文档/测试,非运行时代码,不 push
8.6 KiB
8.6 KiB
双 Claw 引擎事件契约(#873 · P0)
任务归属:加速带 #873(事件契约结构化:diagnosis + IM Bot 缓冲) 关联:#874 NodeRegistry(节点身份)、#876 OTel(双 Claw 标签)、#877 落库集成 真相源:
app/core/engine_event.py+app/core/claw_adapters.py(容器qihu-aiflow,镜像with-deps=df7adefc7333) 单测哨兵:tests/test_engine_event.py(EngineEvent/Adapter 层) +tests/test_dual_claw.py(R-DUAL 双端一致性)
1. Why:为什么需要收敛事件契约
此前引擎事件是「散落在 chat_service 里的裸 dict」:
- 构造点分散,字段全靠记忆,漏一个键前端就静默不渲染;
- 没有
trace_id/tenant_id/claw标签,可观测(#876)无处落脚; - 失败只有一句人话字符串,前端靠
result.includes('失败')猜错误 —— 极脆。
#873 把事件收敛为「唯一构造入口」:所有事件必须经 make_* 工厂产出,再由 to_wire() 统一序列化。新增字段一律「非空才输出」,保证既有前端(ChatPage.tsx)零改动。
2. 单引擎原则(最高约束)
能力只在 EngineCore(`_send_message_core`) 实现一次;
Adapter 只做「引擎事件 → 通道」的搬运,不含任何业务判断。
┌──────────────┐ on_event(EngineEvent) ┌──────────────┐
│ EngineCore │ ────────────────────────► │ ClawAdapter │
└──────────────┘ └──────┬───────┘
│
┌───────────────────────────────┴────────────────────────┐
▼ ▼
SseAdapter (Web Claw) BufferAdapter (IM Bot Claw)
逐帧 `data: {...}\n\n` 缓冲装配 → 整条 str | 媒体 dict
历史教训(#866):IM 侧曾出现「第二份实现」,一次全量部署就把它冲掉、能力静默回退。因此两个 Claw 的差异必须、且只能收敛在本文件的 SseAdapter / BufferAdapter 里。任何把引擎事件序列化逻辑写回 chat.py / im_llm.py 的提交,都会被 test_engine_event.py 与 test_dual_claw.py 的接线哨兵捕获。
3. 线格式契约(authoritative —— 前端已消费,禁止破坏)
| 事件 | 线格式(dict) |
|---|---|
delta |
{"type":"delta","content": str} |
reasoning |
{"type":"reasoning","content": str} |
step |
{"type":"step","step": {...}, "index"?: int} |
done |
{"type":"done","message_id","conversation_id","content","process_steps","used_skills","used_mcp","generating"} |
error |
{"type":"error","message": str} |
step 对象契约(AiProcessBlock 消费):
| kind | 结构 |
|---|---|
thinking |
{"type":"thinking","text": str} |
skill |
{"type":"skill","name": str,"args": dict,"result": str} |
mcp |
{"type":"mcp","server": str,"name": str,"args": dict,"result": str} |
agent/kb/media |
同构,按 type 选渲染分支 |
附加字段(新增,前端可选消费,非空才输出):
diagnosis:{code, severity, summary, detail, hint, retryable}—— 结构化诊断,取代「把原因塞进 result 字符串」;phase:"start" | "ok" | "fail"—— 步骤执行阶段;node:{"id","type","label"}—— #874 图节点身份;seq/trace_id/tenant_id/claw—— 可观测维度(#876)。
4. EngineEvent API(engine_event.py)
工厂(唯一构造入口)
make_delta(content, *, seq=0, trace_id="", tenant_id="", claw="")→ 正文增量make_reasoning(content, **)→ 思维链增量make_step(kind, *, name="", text="", server="", args=None, result="", phase="", index=None, diagnosis=None, node=None, **)→ 过程步骤index:同一步骤「先phase=start占位,执行完同index发phase=ok|fail覆盖」→ 前端实时进度;node(#874):承载图节点身份,非空才输出,老前端忽略;diagnosis为error且result为空时,自动用diagnosis.to_text()兜底result(否则前端判定「运行中」)。
make_done(*, message_id, conversation_id, content="", process_steps=None, used_skills=None, used_mcp=None, generating=False, **)→ 终帧(字段集与遗留 done 帧完全一致)make_error(message, *, diagnosis=None, **)→ 错误终帧
序列化
EngineEvent.to_wire() -> dict:兼容铁律 —— 既有键原样输出;新增键一律「非空才输出」(不撑大帧体积、不触发未知分支)。EngineEvent.to_text() -> str:无结构 UI(IM Bot)降级文本;返回空串表示「本事件无需展示」。
诊断
Diagnosis(code, severity, summary, detail, hint, retryable):to_dict()/from_dict(d)双向;to_text():IM 纯文本降级(ℹ️/⚠️/❌图标 + summary + detail + 建议)。
迁移兼容
from_wire(payload):把遗留裸 dict 提升为EngineEvent(Adapter 收到老代码on_event({...})后一律归一化)。type缺失 / 非 dict → 返回None(Adapter 丢弃并告警)。
5. ClawAdapter API(claw_adapters.py)
基类 BaseClawAdapter:
- 实例可直接当
on_event回调传给 EngineCore(async def __call__(event)→emit); emit(event):先from_wire归一化 → 计数 →_handle(异常被吞掉,Adapter 永不反噬引擎);_normalize(event):补全seq/trace_id/tenant_id/claw(Adapter 自身默认值兜底)。
SseAdapter(Web Claw,逐帧)
claw = "web";构造SseAdapter(queue=None, *, wire_meta=False);_thin(ev, wire):默认剥离seq/trace_id/tenant_id/claw(实测 delta 单帧 46B→117B 膨胀 2.5x,2000 帧多传 ~140KB 噪音,前端不消费);终帧保留trace_id(排障唯一抓手);frame(wire)→"data: " + json.dumps(wire, ensure_ascii=False) + "\n\n";frames()异步生成器:从 queue 取帧,遇__end__哨兵结束;close():幂等,投送哨兵帧。
BufferAdapter(IM Bot Claw,缓冲装配)
claw = "im";构造BufferAdapter(*, keep_reasoning=False, keep_diagnosis=True);- 累积规则:
delta累积正文;reasoning默认丢弃(思维链不该刷屏微信),keep_reasoning=True打开;step的error级diagnosis降级为文本尾注(用户至少知道哪步炸了);done.content优先于累积 delta(引擎已做最终裁剪/落库);
assemble() -> str | dict:- 媒体(以
<img/<video开头)→ 返回结构化{"__media__": True, "media_type": "image"|"video", "url": "..."}(IM 内联发图/发视频依赖,禁止退化成纯文本); - 空 body + error →
_render_failure();空 body + generating → 「内容正在生成」;否则正文 + 诊断尾注 + 生成提示; - 超
_MAX_ASSEMBLED_CHARS=20000截断并标注「内容过长已截断」。
- 媒体(以
6. Web / IM 双 Claw 接线方式
| Claw | 入口 | Adapter | 线格式 |
|---|---|---|---|
| Web | chat.py /messages/stream |
SseAdapter(委托,线格式逐字节一致) |
SSE data: 逐帧 |
| IM Bot | im_llm.py(灰阶门控) |
BufferAdapter |
整条 str / __media__ dict |
im_llm.py经BufferAdapter复用EngineCore(_call_engine_core),媒体走快路径;- IM_USE_ENGINE_CORE 灰阶门(默认 OFF):生产零改动,开启后 IM 走统一引擎,单引擎红利即时生效;
- 两个 Adapter 必须分别被
chat.py/im_llm.py引用(接线哨兵test_engine_event.py::test_dual_claw_wiring_source_guard锁死)。
7. 测试与可观测
tests/test_engine_event.py:EngineEvent 工厂 /to_wire兼容 /from_wire/ SseAdapter 裁剪与 SSE 格式 / BufferAdapter 累积与媒体 / 双 Claw 接线契约(确定性、无 DB、无网络);tests/test_dual_claw.py(R-DUAL-*):Web/IM 能力一致、命名同源、__media__保留、OTeltenant_id+claw_type标签;- 指标(#876):
aiflow_span_count/aiflow_span_duration_ms/aiflow_llm_route_total,维度含claw_type ∈ {web, im_wechat}、tenant_id,Grafana 看板qihuaiflow-dual-claw-876。