"""双 Claw 适配器（#873 · P0）。 单引擎原则: 能力只在 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 的差异必须、且只能收敛在本文件的两个 Adapter 里。 """ from __future__ import annotations import asyncio import json import logging import re from typing import Any, Dict, List, Optional from app.core.engine_event import ( CLAW_IM, CLAW_WEB, DELTA, DONE, ERROR, REASONING, STEP, Diagnosis, EngineEvent, from_wire, make_error, ) logger = logging.getLogger(__name__) __all__ = ["BaseClawAdapter", "SseAdapter", "BufferAdapter"] # 单条回复的软上限：防止异常循环把内存/微信消息撑爆 _MAX_ASSEMBLED_CHARS = 20000 _MEDIA_RE = re.compile(r'<(img|video)[^>]*\ssrc="([^"]+)"', re.IGNORECASE) class BaseClawAdapter: """Adapter 基类。实例可直接当作 `on_event` 回调传给 EngineCore。""" claw: str = "" def __init__(self, *, trace_id: str = "", tenant_id: str = ""): self.trace_id = trace_id or "" self.tenant_id = tenant_id or "" self._seq = 0 self.counts: Dict[str, int] = {} # -- 供 EngineCore 直接 await adapter(event) ----------------------- async def __call__(self, event: Any) -> None: await self.emit(event) async def emit(self, event: Any) -> None: ev = self._normalize(event) if ev is None: return self.counts[ev.type] = self.counts.get(ev.type, 0) + 1 try: await self._handle(ev) except Exception as exc: # Adapter 永不把异常反噬引擎 logger.warning("[claw:%s] emit failed: %s", self.claw, exc, exc_info=True) # -- 子类实现 ------------------------------------------------------- async def _handle(self, ev: EngineEvent) -> None: raise NotImplementedError # -- 归一化 --------------------------------------------------------- def _normalize(self, event: Any) -> Optional[EngineEvent]: ev = from_wire(event) if ev is None: logger.warning("[claw:%s] drop malformed event: %r", self.claw, event) return None self._seq += 1 if not ev.seq: ev.seq = self._seq if not ev.trace_id: ev.trace_id = self.trace_id if not ev.tenant_id: ev.tenant_id = self.tenant_id if not ev.claw: ev.claw = self.claw return ev # ===================================================================== # Web Claw —— SSE 逐帧 # ===================================================================== class SseAdapter(BaseClawAdapter): """把引擎事件序列化为 SSE 帧并入队，由路由的 event_gen 逐帧 yield。 行为与改造前的 chat.py 内联实现逐字节一致： `data: ` + json.dumps(wire, ensure_ascii=False) + `\\n\\n` """ claw = CLAW_WEB _SENTINEL = {"type": "__end__"} # 仅用于内部可观测(#876)，默认不上线 —— 见 _thin() _META_KEYS = ("seq", "trace_id", "tenant_id", "claw") def __init__(self, queue: "Optional[asyncio.Queue]" = None, *, wire_meta: bool = False, **kw): super().__init__(**kw) self.queue: asyncio.Queue = queue if queue is not None else asyncio.Queue() self.wire_meta = wire_meta self._closed = False def _thin(self, ev: EngineEvent, wire: Dict[str, Any]) -> Dict[str, Any]: """裁剪线上元数据。 why: delta 是最热的帧，一次长回复有上千个。实测带全量元数据会让 单帧从 46B 膨胀到 117B(2.5x)，2000 帧即多传 ~140KB 纯噪音， 而前端根本不消费这些键。故默认剥离，只在终帧留 trace_id —— 单帧成本可忽略，却是线上排障唯一的抓手。 """ if self.wire_meta: return wire for k in self._META_KEYS: wire.pop(k, None) if ev.is_terminal() and ev.trace_id: wire["trace_id"] = ev.trace_id return wire async def _handle(self, ev: EngineEvent) -> None: await self.queue.put(self._thin(ev, ev.to_wire())) async def emit_error(self, message: str, *, diagnosis: Any = None) -> None: await self.emit(make_error(message, diagnosis=diagnosis)) async def close(self) -> None: """通知 frames() 结束。幂等。""" if self._closed: return self._closed = True await self.queue.put(dict(self._SENTINEL)) @staticmethod def frame(wire: Dict[str, Any]) -> str: return "data: " + json.dumps(wire, ensure_ascii=False) + "\n\n" async def frames(self): """异步生成 SSE 帧文本，直到收到哨兵。""" while True: wire = await self.queue.get() if not isinstance(wire, dict): continue if wire.get("type") == "__end__": break yield self.frame(wire) # ===================================================================== # IM Bot Claw —— 缓冲装配 # ===================================================================== class BufferAdapter(BaseClawAdapter): """累积引擎事件，最后一次性装配为可直接推送给 IM 的回复。 IM 通道（微信/iLink）没有流式能力，也没有卡片 UI： * delta 累积成正文； * reasoning 默认丢弃（思维链不该刷屏微信），可用 keep_reasoning 打开； * step 里的 error 级 diagnosis 降级成文本尾注，用户至少知道「哪一步炸了」； * done 的 content 优先于累积的 delta（引擎已做最终裁剪/落库）。 """ claw = CLAW_IM def __init__(self, *, keep_reasoning: bool = False, keep_diagnosis: bool = True, **kw): super().__init__(**kw) self.keep_reasoning = keep_reasoning self.keep_diagnosis = keep_diagnosis self._delta_parts: List[str] = [] self._reasoning_parts: List[str] = [] self.steps: List[Dict[str, Any]] = [] self.done: Optional[Dict[str, Any]] = None self.error: Optional[str] = None self.error_diagnosis: Optional[Diagnosis] = None self.used_skills: List[Any] = [] self.used_mcp: List[Any] = [] self.generating: bool = False async def _handle(self, ev: EngineEvent) -> None: if ev.type == DELTA: self._delta_parts.append(ev.content) elif ev.type == REASONING: if self.keep_reasoning: self._reasoning_parts.append(ev.content) elif ev.type == STEP: self._record_step(ev) elif ev.type == DONE: wire = ev.to_wire() self.done = wire self.used_skills = wire.get("used_skills") or [] self.used_mcp = wire.get("used_mcp") or [] self.generating = bool(wire.get("generating")) if wire.get("process_steps"): self.steps = list(wire["process_steps"]) elif ev.type == ERROR: self.error = str(ev.extra.get("message") or ev.content or "未知错误") self.error_diagnosis = ev.diagnosis def _record_step(self, ev: EngineEvent) -> None: step = ev.extra.get("step") if not isinstance(step, dict): return idx = ev.extra.get("index") if isinstance(idx, int) and 0 <= idx < len(self.steps): self.steps[idx] = step else: self.steps.append(step) # -- 装配 ----------------------------------------------------------- @property def text(self) -> str: """当前累积正文（done 优先）。""" if self.done and isinstance(self.done.get("content"), str) and self.done["content"]: return self.done["content"] return "".join(self._delta_parts) def assemble(self) -> "str | dict": """产出最终回复：媒体则给结构化 dict，否则给纯文本。 返回的 dict 契约必须与 im_llm 媒体快路径一致： {"__media__": True, "media_type": "image"|"video", "url": "..."} —— im_bot_user.py 依赖它做微信内联发图/发视频。 """ body = self.text media = self._extract_media(body) if media is not None: return media if not body: if self.error: return self._render_failure() if self.generating: return "（内容正在生成中，请稍后在平台「AI 聊天」查看结果）" return "（未生成有效回复，请换个说法再试一次）" notes = self._diagnosis_notes() if notes: body = body.rstrip() + "\n\n" + "\n".join(notes) if self.error: body = body.rstrip() + "\n\n" + self._render_failure() if self.generating: body = body.rstrip() + "\n\n（媒体正在后台生成，稍后可在平台「AI 聊天」查看）" if len(body) > _MAX_ASSEMBLED_CHARS: body = body[:_MAX_ASSEMBLED_CHARS] + "\n…（内容过长已截断）" return body # -- 内部 ----------------------------------------------------------- @staticmethod def _extract_media(body: str) -> Optional[Dict[str, Any]]: s = (body or "").strip() if not (s.startswith(" List[str]: """把失败步骤的结构化诊断降级成文本尾注（IM 无卡片 UI）。""" if not self.keep_diagnosis: return [] notes: List[str] = [] for step in self.steps: if not isinstance(step, dict): continue diag = Diagnosis.from_dict(step.get("diagnosis")) if diag is not None and diag.severity == "error": notes.append(diag.to_text()) # 去重保序 return list(dict.fromkeys(notes)) def _render_failure(self) -> str: if self.error_diagnosis is not None: return self.error_diagnosis.to_text() return f"⚠️ 生成中断：{self.error}"