QihuAI Design
037c965c26
- 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)
295 lines
11 KiB
Python
295 lines
11 KiB
Python
"""双 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("<img") or s.startswith("<video")):
|
||
return None
|
||
m = _MEDIA_RE.search(s)
|
||
if not m:
|
||
return None
|
||
return {
|
||
"__media__": True,
|
||
"media_type": "image" if m.group(1).lower() == "img" else "video",
|
||
"url": m.group(2),
|
||
}
|
||
|
||
def _diagnosis_notes(self) -> 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}"
|