From 037c965c26b635a97b36266e6646e0c13289e806 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Tue, 4 Aug 2026 12:25:12 +0800 Subject: [PATCH] feat(engine): #873 dual-Claw event contract + SseAdapter/BufferAdapter + chat.py SSE delegate + im_llm EngineCore thin-shell (gray-gated) - app/core/engine_event.py (#880): EngineEvent dataclass + make_* factories, single construction entry, byte-compatible wire with legacy ChatPage.tsx - app/core/claw_adapters.py (#884): SseAdapter (hot frames zero-metadata, terminal keeps trace_id) + BufferAdapter (assemble() preserves __media__) - app/api/chat.py (#881): /messages/stream now delegates to SseAdapter; wire bytes unchanged (46B delta / done+message_id) - app/api/im_llm.py (#883): IM_USE_ENGINE_CORE gray gate; _call_engine_core reuses EngineCore via BufferAdapter (default OFF, preserves media fast-path) - app/api/chat_service.py: capture 3 container-ahead fixes (stray import asyncio removed, logger.info format, LiteLLM master key) --- backend/python/app/api/chat.py | 34 +-- backend/python/app/api/chat_service.py | 10 +- backend/python/app/api/im_llm.py | 86 ++++++ backend/python/app/core/claw_adapters.py | 294 +++++++++++++++++++ backend/python/app/core/engine_event.py | 349 +++++++++++++++++++++++ 5 files changed, 752 insertions(+), 21 deletions(-) create mode 100644 backend/python/app/core/claw_adapters.py create mode 100644 backend/python/app/core/engine_event.py diff --git a/backend/python/app/api/chat.py b/backend/python/app/api/chat.py index bf52e25..b4be8cd 100644 --- a/backend/python/app/api/chat.py +++ b/backend/python/app/api/chat.py @@ -84,6 +84,7 @@ from ..auth import get_current_user from ..platform_knowledge import get_platform_knowledge from ..mcp_client import list_mcp_tools, call_mcp_tool +from ..core.claw_adapters import SseAdapter from ..middleware.log_service import LogService @@ -1570,38 +1571,37 @@ async def send_message_stream( req: SendMessageRequest, current_user: User = Depends(get_current_user), ): - """SSE 流式聊天端点(Phase 0)。 - 与普通 /messages 共用 _send_message_core，仅通过 stream=True + on_event - 回调实时吐出增量(delta/reasoning)，末尾推 done 事件(含 message_id)。 + """SSE 流式聊天端点(#873)。 + + 与 /messages 共用 EngineCore(_send_message_core)，仅通过 stream=True + + on_event=SseAdapter 实时吐出增量(delta/reasoning/step)，末尾推 done 终帧 + (含 message_id 等)。SseAdapter 保证线格式与改造前逐字节一致。 """ db = SessionLocal() try: - queue = asyncio.Queue() - - async def on_event(e): - await queue.put(e) + adapter = SseAdapter( + trace_id=f"sse-{conv_id}", + tenant_id=str(getattr(current_user, "tenant_id", "") or ""), + ) async def runner(): try: await _send_message_core( conv_id, req, db, current_user, - stream=True, on_event=on_event, + stream=True, on_event=adapter, ) except Exception as exc: logger.warning(f"[sse] chat stream error: {exc}", exc_info=True) - await queue.put({"type": "error", "message": str(exc)}) + await adapter.emit_error(str(exc)) finally: - await queue.put({"type": "__end__"}) + await adapter.close() task = asyncio.create_task(runner()) async def event_gen(): try: - while True: - e = await queue.get() - if e.get("type") == "__end__": - break - yield f"data: {json.dumps(e, ensure_ascii=False)}\n\n" + async for frame in adapter.frames(): + yield frame finally: await task db.close() @@ -1611,10 +1611,6 @@ async def send_message_stream( db.close() raise - - - - @router.post("/conversations/{conv_id}/clear") diff --git a/backend/python/app/api/chat_service.py b/backend/python/app/api/chat_service.py index fbf04d5..5d22214 100644 --- a/backend/python/app/api/chat_service.py +++ b/backend/python/app/api/chat_service.py @@ -1619,7 +1619,6 @@ async def _send_message_core( _ct = completion_tokens if "completion_tokens" in dir() else 0 _uid = user_id if "user_id" in dir() else "unknown" if _pt or _ct: - import asyncio asyncio.create_task(_bill_tokens( _uid, _pt, _ct, ai_model if "ai_model" in dir() else "unknown", @@ -2899,7 +2898,7 @@ async def _send_message_core( # Defensive guard: 即使 model_id 指向 image/video 模型，若用户输入无媒体意图词， # 强制回退到 text 默认模型。修复 bug：普通多轮对话被误判走图片生成。 if ai_model and ("image" in caps or "video" in caps) and not detect_media_intent(req.content): - logger.info("[chat] Override media-cap model {} -> text (no media intent in: {:.50r})".format(ai_model.name, req.content)) + logger.info("[chat] Override media-cap model %s -> text (no media intent in: %.50s)", ai_model.name, req.content) ai_model = resolve_auto_model("text", db) caps = [c.lower() for c in (ai_model.capabilities or [])] if ai_model else [] if ai_model and ("image" in caps or "video" in caps): @@ -3081,6 +3080,13 @@ async def _send_message_core( api_url = ai_model.endpoint + # 2026-07-29 fix: 调用 LiteLLM 端点时，必须用 master key (sk-litellm-...) 而非模型自身 key (tp-c...) + # LiteLLM 内置每个 model 的真实 key 在 litellm_config.yaml，chat 不应直接传模型的 key + if api_url and "qihu-aiflow-litellm" in api_url: + api_key = os.environ.get("LITELLM_MASTER_KEY") or "sk-litellm-master-key-aiflow-2026" + auth_header = None # 用默认 Authorization: Bearer + logger.info("[chat] LiteLLM endpoint detected, using master key instead of model key") + diff --git a/backend/python/app/api/im_llm.py b/backend/python/app/api/im_llm.py index 4027133..d67207c 100644 --- a/backend/python/app/api/im_llm.py +++ b/backend/python/app/api/im_llm.py @@ -181,6 +181,16 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im return s return "（媒体生成未能返回有效内容，请重试）" + # ===== 1.5) EngineCore 薄壳（#873 · 灰阶门，默认关闭）===== + # IM_USE_ENGINE_CORE=1 时，文本回复改走与 Web Claw 同一份 EngineCore + # (_send_message_core)，经 BufferAdapter 缓冲装配为整条回复；上方媒体快路径 + # 已提前返回，不受此门影响。默认 0/未设置 = 走下方既有自包含逻辑（零改动）。 + if os.environ.get("IM_USE_ENGINE_CORE") == "1": + return _run_async( + _call_engine_core(db, user_id, prompt, platform, user), + timeout=300, + ) + # ===== 2) 文本回复：模型 failover + 技能自动发现 + 平台知识库注入 ===== from ..platform_knowledge import get_platform_knowledge @@ -413,3 +423,79 @@ def _call_real_llm(db: Session, user=None, prompt: str = "", platform: str = "im return f"（已尝试 {len(_try_models)} 个模型均调用失败。最后错误: {_summary}。请稍后重试。）" except Exception as e: return f"（AI 回复生成失败：{e}）" + + +async def _call_engine_core(db, user_id, prompt: str, platform: str, user=None) -> "str | dict": + """（#873 薄壳）复用 Web Claw 同一份 EngineCore 生成 IM 回复。 + + 单引擎原则：IM Bot 不再维护第二份「模型 failover + 技能/MCP 执行」实现， + 而是把已发现的 skill_ids/mcp_ids 交给 _send_message_core，经 BufferAdapter + 缓冲装配成整条回复（媒体结构 __media__ 由 BufferAdapter 原样保留）。 + 默认不启用（见 _call_real_llm 中的 IM_USE_ENGINE_CORE 灰阶门）。 + """ + from app.api.chat_service import _send_message_core + from app.api.chat_utils import SendMessageRequest + from app.core.claw_adapters import BufferAdapter + from ..models import Conversation, User + + # 解析用户：IM 调用方常传 user=None，仅给 user_id + real_user = user + if real_user is None and user_id: + real_user = db.query(User).filter(User.id == user_id).first() + if real_user is None: + return "（EngineCore 模式：无法解析当前用户，请检查 IM 账号绑定）" + + # 能力发现（与 IM 自包含逻辑同一套，claw="im"） + from app.core.capability_discovery import ( + discover_mcp_servers, + discover_skills, + ) + _query_hint = prompt + _marker = "用户当前问题：" + if _marker in prompt: + _query_hint = prompt[prompt.rfind(_marker) + len(_marker):].strip() + skills = [] + try: + skills = discover_skills(db, user_id, query=_query_hint, claw="im") or [] + except Exception as _e: + logger.warning(f"[im_llm.engine] 技能发现失败(降级为空): {_e}") + mcp_servers = [] + try: + mcp_servers = discover_mcp_servers(db, None, claw="im") or [] + except Exception as _e: + logger.warning(f"[im_llm.engine] MCP 发现失败(降级为空): {_e}") + + skill_ids = [ + (s.get("id") if isinstance(s, dict) else getattr(s, "id", None)) + for s in skills + ] + skill_ids = [s for s in skill_ids if s] + mcp_ids = [ + (m.get("id") if isinstance(m, dict) else getattr(m, "id", None)) + for m in mcp_servers + ] + mcp_ids = [m for m in mcp_ids if m] + + # 构造本回合专用对话：EngineCore 要求已存在且归属当前用户的 Conversation + conv = Conversation( + id=gen_uuid(), + user_id=real_user.id, + title=f"IM对话-{platform}", + source="im_bot", + ) + db.add(conv) + db.commit() + + req = SendMessageRequest( + content=prompt, + skill_ids=skill_ids or None, + mcp_ids=mcp_ids or None, + use_tools=bool(skill_ids or mcp_ids), + agent_ids=None, + knowledge_base_ids=None, + attachments=None, + ) + + adapter = BufferAdapter(tenant_id=str(getattr(real_user, "tenant_id", "") or "")) + await _send_message_core(conv.id, req, db, real_user, stream=True, on_event=adapter) + return adapter.assemble() diff --git a/backend/python/app/core/claw_adapters.py b/backend/python/app/core/claw_adapters.py new file mode 100644 index 0000000..d82bb47 --- /dev/null +++ b/backend/python/app/core/claw_adapters.py @@ -0,0 +1,294 @@ +"""双 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}" diff --git a/backend/python/app/core/engine_event.py b/backend/python/app/core/engine_event.py new file mode 100644 index 0000000..cd89fb6 --- /dev/null +++ b/backend/python/app/core/engine_event.py @@ -0,0 +1,349 @@ +"""双 Claw 引擎事件契约（#873 · P0）。 + +why: + 平台有且只有两个 Claw 出口 —— Web Claw(SSE 逐帧) 与 IM Bot Claw(整条推送)。 + 在此之前，引擎事件是「散落在 chat_service 里的裸 dict」： + + * 构造点分散，字段全靠记忆，漏一个键前端就静默不渲染； + * 没有 trace_id / tenant_id / claw 标签，可观测(#876)无处落脚； + * 失败只有一句人话字符串，前端靠 result.includes('失败') 猜错误 —— 极脆。 + + 本模块把事件收敛为「唯一构造入口」：所有事件必须经 make_* 工厂产出， + 再由 to_wire() 统一序列化。新增字段一律「非空才输出」， + 保证既有前端(ChatPage.tsx)零改动。 + +线格式契约（authoritative —— 前端已消费，禁止破坏）: + + 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 消费）: + + 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} + + 附加(新增，前端可选消费): "diagnosis": {...}, "phase": "start|ok|fail" +""" +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any, Dict, Optional + +__all__ = [ + "SCHEMA_VERSION", + "DELTA", "REASONING", "STEP", "DONE", "ERROR", "EVENT_TYPES", + "STEP_THINKING", "STEP_SKILL", "STEP_MCP", "STEP_AGENT", "STEP_KB", "STEP_MEDIA", + "PHASE_START", "PHASE_OK", "PHASE_FAIL", + "CLAW_WEB", "CLAW_IM", + "SEV_INFO", "SEV_WARN", "SEV_ERROR", + "Diagnosis", "EngineEvent", + "make_delta", "make_reasoning", "make_step", "make_done", "make_error", + "from_wire", +] + +SCHEMA_VERSION = 1 + +# ---- 事件类型（线格式 type 值，前端已硬编码，禁止改名）---- +DELTA = "delta" +REASONING = "reasoning" +STEP = "step" +DONE = "done" +ERROR = "error" +EVENT_TYPES = frozenset({DELTA, REASONING, STEP, DONE, ERROR}) + +# ---- step 子类（前端按此选图标 / 渲染分支）---- +STEP_THINKING = "thinking" +STEP_SKILL = "skill" +STEP_MCP = "mcp" +STEP_AGENT = "agent" # 新增：外部 Agent 调用 +STEP_KB = "kb" # 新增：知识库检索 +STEP_MEDIA = "media" # 新增：媒体生成 + +# ---- 执行阶段 ---- +PHASE_START = "start" +PHASE_OK = "ok" +PHASE_FAIL = "fail" + +# ---- Claw 通道标签（可观测 #876 的核心维度）---- +CLAW_WEB = "web" +CLAW_IM = "im" + +# ---- 诊断严重级 ---- +SEV_INFO = "info" +SEV_WARN = "warn" +SEV_ERROR = "error" + +_SEV_ICON = {SEV_INFO: "ℹ️", SEV_WARN: "⚠️", SEV_ERROR: "❌"} + + +@dataclass +class Diagnosis: + """结构化诊断。 + + 取代「把失败原因塞进 result 字符串、前端再 includes('失败') 猜」的旧做法： + * Web Claw 可据此渲染诊断卡片（图标/配色/重试按钮）； + * IM Bot Claw 无卡片 UI，用 to_text() 降级为纯文本行。 + """ + + code: str = "" # 机器可读，如 "MCP_TIMEOUT" / "SKILL_NOT_FOUND" + severity: str = SEV_INFO # info | warn | error + summary: str = "" # 一句话结论（面向用户） + detail: str = "" # 细节（可含技术信息，前端折叠展示） + hint: str = "" # 可执行建议 + retryable: bool = False # 是否值得重试 + + def to_dict(self) -> Dict[str, Any]: + return { + "code": self.code, + "severity": self.severity, + "summary": self.summary, + "detail": self.detail, + "hint": self.hint, + "retryable": self.retryable, + } + + def to_text(self) -> str: + """IM Bot Claw 的纯文本降级渲染。""" + icon = _SEV_ICON.get(self.severity, "•") + lines = [f"{icon} {self.summary or self.code or '执行提示'}"] + if self.detail: + lines.append(self.detail) + if self.hint: + lines.append(f"建议：{self.hint}") + return "\n".join(lines) + + @classmethod + def from_dict(cls, d: Any) -> "Optional[Diagnosis]": + if isinstance(d, Diagnosis): + return d + if not isinstance(d, dict): + return None + return cls( + code=str(d.get("code") or ""), + severity=str(d.get("severity") or SEV_INFO), + summary=str(d.get("summary") or ""), + detail=str(d.get("detail") or ""), + hint=str(d.get("hint") or ""), + retryable=bool(d.get("retryable")), + ) + + +@dataclass +class EngineEvent: + """引擎事件的唯一表示。禁止在业务代码里手搓等价 dict —— 用 make_* 工厂。""" + + type: str + content: str = "" + seq: int = 0 + ts: float = field(default_factory=time.time) + trace_id: str = "" + tenant_id: str = "" + claw: str = "" + diagnosis: Optional[Diagnosis] = None + # extra 承载各事件类型的专属键(message_id / step / process_steps ...)， + # 序列化时平铺到顶层，以保持与既有线格式逐字节兼容。 + extra: Dict[str, Any] = field(default_factory=dict) + + # -- 谓词 --------------------------------------------------------- + def is_terminal(self) -> bool: + return self.type in (DONE, ERROR) + + def is_text_chunk(self) -> bool: + return self.type in (DELTA, REASONING) + + # -- 序列化 ------------------------------------------------------- + def to_wire(self) -> Dict[str, Any]: + """序列化为通道线格式。 + + 兼容铁律：既有键原样输出；新增键(seq/ts/claw/trace_id/tenant_id/diagnosis) + 一律「非空才输出」，既不撑大帧体积，也不触发前端未知分支。 + """ + wire: Dict[str, Any] = {"type": self.type} + + if self.type in (DELTA, REASONING): + wire["content"] = self.content + elif self.content: + wire["content"] = self.content + + for key, val in self.extra.items(): + wire[key] = val + + if self.seq: + wire["seq"] = self.seq + if self.claw: + wire["claw"] = self.claw + if self.trace_id: + wire["trace_id"] = self.trace_id + if self.tenant_id: + wire["tenant_id"] = self.tenant_id + if self.diagnosis is not None: + wire["diagnosis"] = self.diagnosis.to_dict() + return wire + + def to_text(self) -> str: + """无结构 UI 时(IM Bot)的降级文本；返回空串表示「本事件无需展示」。""" + if self.type in (DELTA, DONE): + return self.content + if self.type == ERROR: + if self.diagnosis is not None: + return self.diagnosis.to_text() + return f"❌ {self.extra.get('message') or self.content or '执行失败'}" + if self.type == STEP and self.diagnosis is not None: + return self.diagnosis.to_text() + return "" + + +# ===================================================================== +# 工厂 —— 唯一构造入口 +# ===================================================================== +def _ctx(kw: Dict[str, Any]) -> Dict[str, Any]: + """抽出通用上下文字段，避免每个工厂重复列参数。""" + return { + "seq": int(kw.pop("seq", 0) or 0), + "trace_id": str(kw.pop("trace_id", "") or ""), + "tenant_id": str(kw.pop("tenant_id", "") or ""), + "claw": str(kw.pop("claw", "") or ""), + } + + +def make_delta(content: str, **kw) -> EngineEvent: + """正文增量。""" + return EngineEvent(type=DELTA, content=content or "", **_ctx(kw)) + + +def make_reasoning(content: str, **kw) -> EngineEvent: + """思维链增量。""" + return EngineEvent(type=REASONING, content=content or "", **_ctx(kw)) + + +def make_step( + kind: str, + *, + name: str = "", + text: str = "", + server: str = "", + args: Optional[Dict[str, Any]] = None, + result: str = "", + phase: str = "", + index: Optional[int] = None, + diagnosis: Any = None, + **kw, +) -> EngineEvent: + """过程步骤（思考 / 技能 / MCP / Agent / KB / 媒体）。 + + index 用于「原地更新同一步骤」：先发 phase=start 占位， + 执行完再用同一 index 发 phase=ok|fail 覆盖，前端即得到实时进度。 + """ + step: Dict[str, Any] = {"type": kind} + if kind == STEP_THINKING: + step["text"] = text or name or "" + else: + if name: + step["name"] = name + if server: + step["server"] = server + if text: + step["text"] = text + step["args"] = args or {} + step["result"] = result or "" + if phase: + step["phase"] = phase + + diag = Diagnosis.from_dict(diagnosis) if diagnosis is not None else None + if diag is not None: + step["diagnosis"] = diag.to_dict() + # result 为空时前端会判定为「运行中」，故失败必须给出可读 result 兜底 + if not step["result"] and diag.severity == SEV_ERROR: + step["result"] = diag.to_text() + + extra: Dict[str, Any] = {"step": step} + if isinstance(index, int) and index >= 0: + extra["index"] = index + + return EngineEvent(type=STEP, diagnosis=diag, extra=extra, **_ctx(kw)) + + +def make_done( + *, + message_id: Any = None, + conversation_id: Any = None, + content: str = "", + process_steps: Optional[list] = None, + used_skills: Optional[list] = None, + used_mcp: Optional[list] = None, + generating: bool = False, + **kw, +) -> EngineEvent: + """终帧。字段集必须与既有 done 帧完全一致。""" + extra = { + "message_id": message_id, + "conversation_id": conversation_id, + "content": content or "", + "process_steps": process_steps or [], + "used_skills": used_skills or [], + "used_mcp": used_mcp or [], + "generating": bool(generating), + } + return EngineEvent(type=DONE, content=content or "", extra=extra, **_ctx(kw)) + + +def make_error(message: str, *, diagnosis: Any = None, **kw) -> EngineEvent: + """错误终帧。""" + diag = Diagnosis.from_dict(diagnosis) if diagnosis is not None else None + return EngineEvent( + type=ERROR, + diagnosis=diag, + extra={"message": message or ""}, + **_ctx(kw), + ) + + +def from_wire(payload: Any) -> Optional[EngineEvent]: + """把遗留裸 dict 提升为 EngineEvent（迁移期双向兼容的关键）。 + + chat_service.py 里仍有直接 await on_event({...}) 的老代码， + Adapter 收到后一律经此函数归一化，之后全链路只面对 EngineEvent。 + """ + if isinstance(payload, EngineEvent): + return payload + if not isinstance(payload, dict): + return None + + d = dict(payload) + etype = str(d.pop("type", "") or "") + if not etype: + return None + + content = d.pop("content", "") + if not isinstance(content, str): + # done 帧的 content 理论上是 str；异常类型保留原值到 extra，避免丢数据 + d["content"] = content + content = "" + + diag = Diagnosis.from_dict(d.pop("diagnosis", None)) + seq = d.pop("seq", 0) + trace_id = d.pop("trace_id", "") + tenant_id = d.pop("tenant_id", "") + claw = d.pop("claw", "") + d.pop("ts", None) + + # done 帧的 content 需回填到 extra，才能在 to_wire() 时原样吐回 + if etype == DONE: + d["content"] = content + + return EngineEvent( + type=etype, + content=content, + seq=int(seq or 0), + trace_id=str(trace_id or ""), + tenant_id=str(tenant_id or ""), + claw=str(claw or ""), + diagnosis=diag, + extra=d, + )