"""Reasoning normalization and aggregation for XAI app-chat streams.""" from __future__ import annotations import re from dataclasses import dataclass from typing import Any _GENERIC_HEADERS = { "", "thinking about your request", } _PROGRESSIVE_HINTS = ( "正在", "准备", "计划", "查找", "搜索", "浏览", "确认", "核对", "整合", "挖掘", "比对", "checking", "browsing", "verifying", "integrating", "digging", "cross-checking", "searching", "planning", ) _FINDING_HINTS = ( "尚未", "已经", "已", "确认", "表明", "说明", "显示", "主要", "通常", "支持", "出现", "启动", "持续", "提升", "更新", "灰度", "发布", "上线", "多模态", "视觉", "专家", "context", "token", "参数", "每天", "大潮", "小潮", "半日潮", "引力", "周期", "模式", "confirmed", "launched", "released", "rollout", "testing", "native multimodal", "widely believed", "latest", ) _LOW_VALUE_PREFIXES = ( "用户", "user", "i can", "我可以", "我收集", "建议", "need", "需要", "应该", "since instructions", "proposed", "mermaid", "可以用", "我建议", ) _TRACK_RULES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("latest_updates", ("最新", "latest", "today", "recent", "最近", "update", "news", "本周", "4月", "april")), ("release_status", ("release date", "released", "release", "launch", "上线", "发布", "正式发布", "current status")), ("gray_rollout", ("灰度", "grayscale", "gray release", "灰度测试", "内测", "rollout")), ("official_confirmation", ("official", "官网", "official site", "site:", "platform.deepseek.com", "deepseek.ai")), ("ui_modes", ("vision", "视觉", "expert", "专家模式", "fast", "default", "ui", "界面", "mode")), ("v4_lite", ("v4 lite", "sealion", "sealion-lite", "海狮")), ("specs_architecture", ("specs", "parameters", "architecture", "engram", "mhc", "moe", "context", "benchmarks", "规格", "参数", "架构", "万亿")), ("definition_basics", ("定义", "解释", "什么是", "what is", "phenomenon", "现象")), ("causes_mechanism", ("成因", "原因", "cause", "causes", "gravity", "引力", "机制")), ("categories_types", ("春潮", "小潮", "半日潮", "全日潮", "类型", "分类")), ("impacts_applications", ("影响", "应用", "发电", "航运", "生活", "生态")), ) _ZH_LABELS = { "understanding": "理解问题", "scope": "检索范围", "evidence": "核验与证据", "finding": "关键发现", "latest_updates": "最新动态", "release_status": "发布状态与上线节奏", "gray_rollout": "灰度进展", "official_confirmation": "官方渠道确认", "ui_modes": "Expert / Vision 模式关联", "v4_lite": "V4 Lite 与 Sealion 线索", "specs_architecture": "规格、架构与上下文能力", "definition_basics": "定义与基础解释", "causes_mechanism": "成因与机制", "categories_types": "分类与相关类型", "impacts_applications": "影响与应用", } _EN_LABELS = { "understanding": "Understanding", "scope": "Research Scope", "evidence": "Verification", "finding": "Key Findings", "latest_updates": "latest updates", "release_status": "release status and rollout timing", "gray_rollout": "gray rollout progress", "official_confirmation": "official confirmation", "ui_modes": "Expert / Vision mode signals", "v4_lite": "V4 Lite and Sealion clues", "specs_architecture": "specs, architecture, and context capability", "definition_basics": "definition and basic explanation", "causes_mechanism": "causes and mechanism", "categories_types": "categories and related types", "impacts_applications": "impacts and applications", } @dataclass(slots=True) class ReasoningEvent: section: str text: str track: str = "" evidence_level: int = 0 dedupe_key: str = "" class ReasoningAggregator: """Normalize raw stream fragments into enterprise-style reasoning output.""" __slots__ = ( "_language", "_en_votes", "_zh_votes", "_agent_search_started", "_emitted_keys", "_seen_tracks", "_seen_findings", "_pending_events", "_section_started", "_track_best_level", "_track_emit_counts", ) def __init__(self) -> None: self._language: str | None = None self._en_votes = 0 self._zh_votes = 0 self._agent_search_started = False self._emitted_keys: set[str] = set() self._seen_tracks: set[str] = set() self._seen_findings: set[str] = set() self._pending_events: list[ReasoningEvent] = [] self._section_started: set[str] = set() self._track_best_level: dict[tuple[str, str], int] = {} self._track_emit_counts: dict[tuple[str, str], int] = {} def on_thinking( self, token: str, *, tag: str | None, rollout: str | None, step_id: int | None, ) -> list[str]: self._observe_language(token) tag_name = str(tag or "").strip() text = str(token or "").strip() if not text: return [] if tag_name == "header": event = self._normalize_header(text, step_id=step_id) return self._dispatch(event) if event else [] if tag_name == "summary": event = self._normalize_summary(text, step_id=step_id) return self._dispatch(event) if event else [] event = self._normalize_summary(text, step_id=step_id) return self._dispatch(event) if event else [] def on_tool_usage( self, tool_name: str, args: dict[str, Any], *, rollout: str | None, step_id: int | None, ) -> list[str]: lines: list[str] = [] self._observe_language(str(args.get("query") or args.get("message") or args.get("instructions") or "")) if tool_name == "web_search": query = str(args.get("query") or args.get("q") or "").strip() if not query: return [] if str(rollout or "").startswith("Agent") and not self._agent_search_started: self._agent_search_started = True lines.extend(self._dispatch(ReasoningEvent( "scope", self._localized_line("agents_started"), dedupe_key="scope:agents_started", ))) track = self._infer_track(query) if not track: return lines lines.extend(self._dispatch(ReasoningEvent( "scope", self._localized_track_line(track), track=track, evidence_level=1, dedupe_key=f"scope:web:{track}", ))) return lines if tool_name in {"x_search", "x_keyword_search", "x_semantic_search"}: query = str(args.get("query") or "").strip() track = self._infer_track(query) if not track: return [] return self._dispatch(ReasoningEvent( "evidence", self._localized_social_line(track), track=track, evidence_level=2, dedupe_key=f"evidence:social:{track}", )) if tool_name == "browse_page": url = str(args.get("url") or "").strip() source_kind, track = self._classify_page_source(url, args) if not source_kind: return [] return self._dispatch(ReasoningEvent( "evidence", self._localized_browse_line(source_kind, track), track=track or source_kind, evidence_level=4 if source_kind in {"official", "product"} else 3, dedupe_key=f"evidence:browse:{source_kind}:{track or ''}", )) if tool_name in {"search_images", "image_search"}: description = str(args.get("image_description") or args.get("imageDescription") or "").strip() if not description: return [] topic = self._classify_image_topic(description) if not topic: return [] return self._dispatch(ReasoningEvent( "scope", self._localized_image_line(topic), track="visual_assets", evidence_level=1, dedupe_key=f"scope:image:{topic}", )) if tool_name == "chatroom_send": message = str(args.get("message") or "").strip() if not message: return [] lines = [] for section, text, track, level in self._extract_report_events(message): lines.extend(self._dispatch(ReasoningEvent( section, text, track=track, evidence_level=level, dedupe_key=f"{section}:report:{track}:{self._normalize_key(text)}", ))) return lines if tool_name == "code_execution": return self._dispatch(ReasoningEvent( "evidence", self._localized_line("code_execution"), dedupe_key="evidence:code_execution", )) return [] def finalize(self) -> list[str]: if not self._pending_events: return [] if self._language is None: self._language = "en" if self._en_votes > 0 and self._zh_votes == 0 else "zh" return self._flush_pending() def _normalize_header(self, text: str, *, step_id: int | None) -> ReasoningEvent | None: stripped = text.strip() if stripped.lower() in _GENERIC_HEADERS: return None section = "understanding" if not self._looks_like_verification(stripped) and (step_id or 0) <= 1 else "evidence" return ReasoningEvent(section, self._to_bullet_text(stripped), dedupe_key=f"{section}:header:{self._normalize_key(stripped)}") def _normalize_summary(self, text: str, *, step_id: int | None) -> ReasoningEvent | None: summary = text.lstrip("- ").strip() if not summary: return None if summary.startswith(("建议搜索", "正在调用工具搜索")): return None track = self._infer_track(summary) if self._looks_like_progress(summary): section = "evidence" if self._looks_like_verification(summary) else "scope" return ReasoningEvent(section, self._to_bullet_text(summary), track=track, evidence_level=2 if section == "evidence" else 1, dedupe_key=f"{section}:summary:{self._normalize_key(summary)}") if self._looks_like_finding(summary): if self._is_unconfirmed_signal(summary): return ReasoningEvent("evidence", self._to_bullet_text(summary), track=track, evidence_level=2, dedupe_key=f"evidence:summary:{self._normalize_key(summary)}") if not self._agent_search_started and (step_id or 0) <= 1: return ReasoningEvent("understanding", self._to_bullet_text(summary), track=track, evidence_level=2, dedupe_key=f"understanding:summary:{self._normalize_key(summary)}") return ReasoningEvent("finding", self._to_bullet_text(summary), track=track, evidence_level=3, dedupe_key=f"finding:summary:{self._normalize_key(summary)}") section = "understanding" if (step_id or 0) <= 1 else "scope" return ReasoningEvent(section, self._to_bullet_text(summary), track=track, evidence_level=1, dedupe_key=f"{section}:summary:{self._normalize_key(summary)}") def _extract_report_events(self, message: str) -> list[tuple[str, str, str, int]]: parts = re.split(r"(?:\n+|[。！？!?；;]+|\s+-\s+)", message.replace("\\n", "\n")) candidates: list[tuple[int, str]] = [] for raw_part in parts: clause = self._clean_report_clause(raw_part) if not clause: continue if self._language == "zh" and not re.search(r"[\u4e00-\u9fff]", clause): continue if self._language == "en" and re.search(r"[\u4e00-\u9fff]", clause): continue score = self._score_report_clause(clause) if score <= 0: continue candidates.append((score, clause)) candidates.sort(key=lambda item: (-item[0], len(item[1]))) results: list[tuple[str, str, str, int]] = [] seen_local: set[str] = set() seen_track_counts: dict[tuple[str, str], int] = {} for _, clause in candidates: key = self._normalize_key(clause) if key in seen_local: continue seen_local.add(key) track = self._infer_track(clause) section = "finding" if self._looks_like_finding(clause) else "evidence" if self._is_unconfirmed_signal(clause): section = "evidence" track_key = (section, track or "_") current_track_count = seen_track_counts.get(track_key, 0) max_track_count = 2 if section == "finding" else 1 if current_track_count >= max_track_count: continue seen_track_counts[track_key] = current_track_count + 1 level = self._infer_evidence_level(clause, default=3 if section == "finding" else 2) results.append((section, self._to_bullet_text(clause), track, level)) if len(results) >= 6: break results.sort(key=lambda item: (0 if item[0] == "evidence" else 1, item[2], -item[3])) return results def _dispatch(self, event: ReasoningEvent) -> list[str]: if self._language is None: self._pending_events.append(event) if self._zh_votes > 0: self._language = "zh" elif self._en_votes >= 3: self._language = "en" elif len(self._pending_events) < 4: return [] else: self._language = "en" return self._flush_pending() lines: list[str] = [] if self._pending_events: lines.extend(self._flush_pending()) lines.extend(self._emit(event)) return lines def _flush_pending(self) -> list[str]: lines: list[str] = [] pending = self._pending_events self._pending_events = [] for event in pending: lines.extend(self._emit(event)) return lines def _emit(self, event: ReasoningEvent) -> list[str]: text = event.text.strip() if not text: return [] if event.section == "scope" and ("evidence" in self._section_started or "finding" in self._section_started): return [] if event.section == "evidence" and "finding" in self._section_started: if event.evidence_level >= 4 or event.track in { "latest_updates", "release_status", "official_confirmation", "specs_architecture", "v4_lite", }: promoted_key = event.dedupe_key or f"evidence:{self._normalize_key(text)}" event = ReasoningEvent( "finding", text, track=event.track, evidence_level=event.evidence_level, dedupe_key=f"finding:promoted:{promoted_key}", ) else: return [] dedupe_key = event.dedupe_key or f"{event.section}:{self._normalize_key(text)}" if dedupe_key in self._emitted_keys: return [] if event.track: count_key = (event.section, event.track) emitted_count = self._track_emit_counts.get(count_key, 0) max_per_track = 1 if event.section in {"scope", "evidence"} else 2 if emitted_count >= max_per_track and not dedupe_key.endswith("agents_started"): return [] best_key = (event.section, event.track) best_level = self._track_best_level.get(best_key, -1) if best_level > event.evidence_level: return [] self._track_best_level[best_key] = max(best_level, event.evidence_level) self._track_emit_counts[count_key] = emitted_count + 1 self._emitted_keys.add(dedupe_key) lines: list[str] = [] if event.section not in self._section_started: self._section_started.add(event.section) lines.append(self._section_title(event.section) + "\n") lines.append(text + "\n") return lines def _observe_language(self, text: str) -> None: if not text: return cjk_count = len(re.findall(r"[\u4e00-\u9fff]", text)) en_count = len(re.findall(r"[A-Za-z]", text)) if cjk_count >= 4 or cjk_count > max(2, en_count // 2): self._zh_votes += 1 if self._language is None: self._language = "zh" return if en_count >= 4: self._en_votes += 1 def _section_title(self, section: str) -> str: labels = _ZH_LABELS if self._language != "en" else _EN_LABELS return labels.get(section, section) def _localized_line(self, key: str) -> str: zh_map = { "agents_started": "- 已启动并行代理进行交叉检索与核验。", "code_execution": "- 正在执行代码或生成可运行内容。", } en_map = { "agents_started": "- Parallel agents have started cross-checking the topic.", "code_execution": "- Executing code or generating runnable content.", } mapping = zh_map if self._language != "en" else en_map return mapping[key] def _localized_track_line(self, track: str) -> str: label = self._track_label(track) if self._language == "en": return f"- Parallel research: {label}." return f"- 并行检索：{label}。" def _localized_social_line(self, track: str) -> str: label = self._track_label(track) if self._language == "en": return f"- Social cross-check: {label}." return f"- 社媒交叉核验：{label}。" def _localized_browse_line(self, source_kind: str, track: str) -> str: track_label = self._track_label(track) if track else "" if self._language == "en": mapping = { "official": "Page verification: official site and official pages", "product": "Page verification: product page and live UI", "community": "Page verification: public reports and community write-ups", } else: mapping = { "official": "页面核对：官网与官方页面", "product": "页面核对：产品页面与实际界面", "community": "页面核对：公开报道与社区文章", } base = mapping[source_kind] if track_label: connector = ", focusing on " if self._language == "en" else "，重点核对" return f"- {base}{connector}{track_label}{'.' if self._language == 'en' else '。'}" return f"- {base}{'.' if self._language == 'en' else '。'}" def _localized_image_line(self, topic: str) -> str: if self._language == "en": mapping = { "diagram": "- Visual asset search: diagrams and explanatory graphics.", "photo": "- Visual asset search: real-world comparison photos.", "generic": "- Visual asset search: supporting image references.", } else: mapping = { "diagram": "- 视觉素材检索：示意图与结构说明素材。", "photo": "- 视觉素材检索：实景对比图片。", "generic": "- 视觉素材检索：补充说明图片。", } return mapping[topic] def _track_label(self, track: str) -> str: labels = _ZH_LABELS if self._language != "en" else _EN_LABELS return labels.get(track, track) def _infer_track(self, text: str) -> str: lowered = self._compact_query(text).lower() if not lowered: return "" for track, keywords in _TRACK_RULES: if any(keyword in lowered for keyword in keywords): return track return "" def _classify_page_source(self, url: str, args: dict[str, Any]) -> tuple[str, str]: lowered = url.lower() instructions = str(args.get("instructions") or "") track = self._pick_browse_track(f"{url} {instructions}") if any(domain in lowered for domain in ("deepseek.ai", "deepseek.com")): if "chat.deepseek.com" in lowered or "platform.deepseek.com" in lowered: return "product", track or "ui_modes" return "official", track or "official_confirmation" if url: return "community", track return "", track def _pick_browse_track(self, text: str) -> str: lowered = self._compact_query(text).lower() priority = ( ("ui_modes", ("expert", "vision", "mode", "界面", "ui")), ("release_status", ("release", "released", "launch", "发布", "上线", "status")), ("specs_architecture", ("spec", "parameter", "architecture", "context", "engram", "moe", "规格", "参数", "架构", "上下文")), ("v4_lite", ("v4 lite", "sealion", "sealion-lite", "海狮")), ("official_confirmation", ("official", "官网", "current models", "offering")), ) for track, keywords in priority: if any(keyword in lowered for keyword in keywords): return track return self._infer_track(text) def _classify_image_topic(self, text: str) -> str: lowered = text.lower() if any(token in lowered for token in ("diagram", "示意图", "bulge")): return "diagram" if any(token in lowered for token in ("photo", "照片", "real", "high tide", "low tide", "高潮", "低潮")): return "photo" return "generic" def _looks_like_progress(self, text: str) -> bool: lowered = text.lower() return any(hint in lowered for hint in _PROGRESSIVE_HINTS) def _looks_like_verification(self, text: str) -> bool: lowered = text.lower() return any(token in lowered for token in ("确认", "核对", "浏览", "整合", "比对", "check", "verify", "browse", "integrat")) def _looks_like_finding(self, text: str) -> bool: lowered = text.lower() if self._looks_like_progress(text): return False return any(hint in lowered for hint in _FINDING_HINTS) def _clean_report_clause(self, raw_part: str) -> str: clause = re.sub(r"\s+", " ", raw_part).strip(" -•\t") if not clause: return "" delimiter = "：" if "：" in clause else ":" if ":" in clause else "" if delimiter: head, tail = clause.split(delimiter, 1) head_lower = head.strip().lower() if len(head.strip()) <= 18 or any(token in head_lower for token in ("总结", "最新", "关键", "补充", "latest", "summary", "note")): clause = tail.strip() clause = clause.strip(" -•\t") clause = re.sub(r"^(?:我知道|我收集了可靠信息|我收集到的?信息|从搜索结果总结|详细解释要点(?:（[^）]+）)?|补充)\s*", "", clause) clause = re.sub(r"^(?:that|it shows|it seems)\s+", "", clause, flags=re.IGNORECASE) if len(clause) < 8: return "" lowered = clause.lower() if any(lowered.startswith(prefix) for prefix in _LOW_VALUE_PREFIXES): return "" if "?" in clause or "？" in clause: return "" return self._compact_text(clause, limit=120) def _score_report_clause(self, clause: str) -> int: lowered = clause.lower() score = 0 if any(hint in lowered for hint in _FINDING_HINTS): score += 3 if re.search(r"\b\d+(?:\.\d+)?\b", clause): score += 2 if any(token in clause for token in ("月", "日", "年", "小时", "分钟")): score += 1 if any(token in clause for token in ("重要", "航运", "渔业", "发电", "生态", "模式", "视觉")): score += 1 if any(token in lowered for token in ("可能", "rumor", "传闻", "widely believed", "believed")): score -= 1 if any(token in lowered for token in ("可以", "suggest", "建议", "should", "friendly", "reply")): score -= 2 if len(clause) > 150: score -= 1 return score def _infer_evidence_level(self, clause: str, *, default: int) -> int: lowered = clause.lower() if any(token in lowered for token in ("官网", "official", "chat ui", "界面更新", "页面")): return 4 if any(token in lowered for token in ("x平台", "x posts", "社区", "widely believed", "传闻", "rumor")): return max(2, default - 1) return default def _is_unconfirmed_signal(self, clause: str) -> bool: lowered = clause.lower() return any( token in lowered for token in ( "x平台", "x posts", "社区", "community", "widely believed", "believed", "传闻", "rumor", "曝光", "泄露", ) ) def _to_bullet_text(self, text: str) -> str: stripped = text.strip() if stripped.startswith("- "): stripped = stripped[2:].strip() stripped = self._ensure_terminal_punctuation(stripped) return f"- {stripped}" def _ensure_terminal_punctuation(self, text: str) -> str: stripped = text.strip() if not stripped: return "" if stripped.endswith(("。", "！", "？", ".", "!", "?")): return stripped if re.search(r"[\u4e00-\u9fff]", stripped): return stripped + "。" return stripped + "." def _compact_query(self, text: str) -> str: cleaned = re.sub(r"\b(?:or|and|site:[^\s]+|since:\S+|from:\S+|date:\S+)\b", " ", text, flags=re.IGNORECASE) cleaned = re.sub(r"[()\"']", " ", cleaned) cleaned = re.sub(r"\s+", " ", cleaned).strip() return cleaned def _compact_text(self, text: str, *, limit: int) -> str: compact = re.sub(r"\s+", " ", text).strip() if len(compact) <= limit: return compact return compact[: limit - 3].rstrip() + "..." def _normalize_key(self, text: str) -> str: lowered = text.lower() lowered = re.sub(r"https?://\S+", "", lowered) lowered = re.sub(r"[^\w\u4e00-\u9fff]+", "", lowered) return lowered __all__ = ["ReasoningAggregator", "ReasoningEvent"]