diff --git a/routes/openclaw_bot_routes.py b/routes/openclaw_bot_routes.py index 12e84a5..f8a549c 100644 --- a/routes/openclaw_bot_routes.py +++ b/routes/openclaw_bot_routes.py @@ -8143,6 +8143,58 @@ def telegram_webhook(): answer_callback(cq_id) send_typing(chat_id) + # ── Phase 11 RAG 反饋（v5.0 護欄 #1）───────────────────── + # rag_fb:{log_id}:{score} → 寫回 rag_query_log.feedback_score + # pg_ok:{episode_id} → PromotionGate 人工通過 + # pg_no:{episode_id} → PromotionGate 人工駁回 + # 早於 menu: / cmd: / await: 處理；命中即 short-circuit。 + if data.startswith('rag_fb:'): + try: + parts = data.split(':') + if len(parts) >= 3: + log_id = int(parts[1]) + score = int(parts[2]) + from services.rag_service import rag_service as _rag + ok = _rag.feedback(log_id, score) + ack_text = "已記錄 👍" if score >= 4 else "已記錄 👎" + if not ok: + ack_text = "反饋寫入失敗（已 log）" + sys_log.info( + f"[OpenClawBot] RAG feedback log_id={log_id} score={score} ok={ok}" + ) + send_message(chat_id, ack_text, None, None) + except Exception as exc: + sys_log.warning(f"[OpenClawBot] rag_fb 處理失敗: {exc}") + return jsonify({'ok': True}) + + if data.startswith('pg_ok:') or data.startswith('pg_no:'): + try: + action, _eid = data.split(':', 1) + episode_id = int(_eid) + from services.learning_pipeline import promotion_gate, hash_human_approver + approver_hash = hash_human_approver(str(cq_from_id or '')) + if action == 'pg_ok': + # 人工通過 → 直接 promote（不重跑 4 stage） + insight_id = promotion_gate.promote(episode_id) + ack = ( + f"已晉升至 ai_insights #{insight_id}" + if insight_id else "晉升失敗（已 log）" + ) + else: + promotion_gate.reject( + episode_id, 'rejected_human', + detail=f'human_approver={approver_hash}', + ) + ack = "已駁回（rejected_human）" + sys_log.info( + f"[OpenClawBot] PromotionGate {action} episode_id={episode_id} " + f"by={approver_hash}" + ) + send_message(chat_id, ack, None, None) + except Exception as exc: + sys_log.warning(f"[OpenClawBot] pg_ok/pg_no 處理失敗: {exc}") + return jsonify({'ok': True}) + if data.startswith('menu:'): # 顯示子選單或返回主選單 key = data[5:] diff --git a/services/hermes_analyst_service.py b/services/hermes_analyst_service.py index 76f5d7c..8977401 100644 --- a/services/hermes_analyst_service.py +++ b/services/hermes_analyst_service.py @@ -17,6 +17,7 @@ import json import logging import re import time +import uuid from dataclasses import dataclass from typing import Optional @@ -25,6 +26,7 @@ from sqlalchemy import text from services.mcp_context_service import build_mcp_context from services.ollama_service import resolve_ollama_host, get_host_label from services.ai_call_logger import log_ai_call # Operation Ollama-First v5.0 P1 +from services.rag_service import rag_service, is_rag_enabled # Phase 11: RAG-first 快取 logger = logging.getLogger(__name__) @@ -93,6 +95,73 @@ class HermesAnalystService: def __init__(self, engine=None): self.engine = engine # SQLAlchemy engine，可外部注入 + # ────────────────────────────────────────────── + # Phase 11: RAG-first Q&A 介面（Operation Ollama-First v5.0） + # ────────────────────────────────────────────── + def analyze(self, query: str, request_id: Optional[str] = None) -> dict: + """RAG-first Q&A 入口（feature flag RAG_ENABLED 預設 OFF）。 + + 流程： + 1. RAG_ENABLED=true 時先查 ai_insights（cosine >= 0.85 直接回） + 2. 命中 → 不呼 LLM，回 {"source": "rag", "content": ..., "rag_log_id": ...} + 3. 未命中或 flag OFF → fallback 到 _call_hermes_intent（既有 LLM 路徑） + - LLM 結果同時 enqueue learning_episodes 給 PromotionGate 蒸餾 + - 回 {"source": "llm", "content": ..., "intent": ...} + + Args: + query: 使用者問題（繁中） + request_id: 與 ai_calls.request_id 串鏈；None 自動產生 + + Returns: + dict 一律含 'source' / 'content' 兩個欄位，caller 不必自己分支判斷。 + """ + rid = request_id or f"hermes-{uuid.uuid4().hex[:8]}" + + # ── Step 1: RAG-first（feature flag 預設 OFF → 立即跳過）── + if is_rag_enabled(): + rag = rag_service.query( + text=query, + caller='hermes_analyst', + threshold=0.85, + request_id=rid, + ) + if rag.has_high_confidence: + logger.info( + "[Hermes.analyze] RAG hit request_id=%s top_score=%.3f hits=%d", + rid, rag.hits[0].get('score', 0), len(rag.hits), + ) + return { + 'source': 'rag', + 'content': rag.synthesize(), + 'rag_log_id': rag.log_id, + 'request_id': rid, + } + + # ── Step 2: LLM fallback（既有 _call_hermes_intent 路徑）── + intent_result = self._call_hermes_intent(query) or self._rule_based_intent(query) + content = intent_result.get('preliminary_answer') or '' + + # ── Step 3: enqueue learning_episodes（給 PromotionGate 蒸餾；失敗不影響主流程）── + if content: + try: + from services.learning_pipeline import learning_pipeline + learning_pipeline.enqueue( + episode_type='llm_response', + raw_content=content, + source_table=None, # Hermes 此處未取 ai_calls.id；保留 None 讓 source check 通過 + source_id=None, + ) + except Exception as exc: + logger.debug("[Hermes.analyze] learning_pipeline enqueue 跳過: %s", exc) + + return { + 'source': 'llm', + 'content': content, + 'intent': intent_result.get('intent'), + 'metadata': intent_result.get('metadata', {}), + 'request_id': rid, + } + # ────────────────────────────────────────────── # L1 意圖分析介面（給 EventRouter / Telegram NLP 使用） # ────────────────────────────────────────────── diff --git a/services/learning_pipeline.py b/services/learning_pipeline.py new file mode 100644 index 0000000..c74cf18 --- /dev/null +++ b/services/learning_pipeline.py @@ -0,0 +1,750 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +services/learning_pipeline.py +Operation Ollama-First v5.0 / Phase 11 — 自主學習管線 + +兩大核心： + 1. Distiller — LLM/MCP 結果 → learning_episodes（pending） + 2. PromotionGate — learning_episodes → ai_insights 4 階段晉升閘 + + expire_stale_reviews() 24h 自動降級 + +Owen 強調的 v5.0 護欄 #1（ADR-033）： + 反饋按鈕從「選配」升級為「強制晉升門檻」 + Stage 1: quality_score >= 0.7 + Stage 2: 規則引擎幻覺檢測 + Stage 3: 與既有 insight cosine < 0.95（去重） + Stage 4: weight >= 0.8 必經 Telegram 👍/👎 人工驗收（24h 無回應降級 0.5） + +對應: + - migrations/028_create_learning_episodes.sql + - docs/adr/ADR-033 (Promotion Gate) +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import re +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + + +# ───────────────────────────────────────────────────────────────────────────── +# 常數（避免 magic number） +# ───────────────────────────────────────────────────────────────────────────── +STAGE_1_AUTO_QUALITY = 0.7 # quality_score 下限 +STAGE_3_DEDUP_THRESHOLD = 0.95 # cosine similarity 視為重複 +STAGE_4_HUMAN_REVIEW_WEIGHT = 0.8 # weight >= 此值強制人工驗收 +HUMAN_REVIEW_TIMEOUT_HOURS = 24 # awaiting_review 過期門檻 +EXPIRED_FALLBACK_WEIGHT = 0.5 # 過期降級後的 weight + +DISTILLED_TEXT_MAX_BYTES = 16384 # 與 028 CHECK octet_length<=16384 對齊 + +# 蒸餾品質規則 +_MCP_MIN_LEN = 200 # MCP 結果最小長度 +_LLM_FREE_TEXT_MIN = 500 # LLM 自由文本最小長度 +_LLM_KEYWORDS_HINT = ['結論', '建議', '分析', '預測', '趨勢', '威脅', '機會'] + +# 幻覺規則（Stage 2） +_HALLUCINATION_HEDGE_WORDS = ['我猜', '可能是', '也許', '大概是', '應該是', '我猜測'] +# 「具體數字」啟發式：阿拉伯數字 ≥ 1 個 +_NUMBER_PATTERN = re.compile(r'\d') +# 「自相矛盾」啟發式：簡化判斷 — 同一句裡同一主詞同時被指派不同值（v5.0 不深做語意，rule-only） + +# user_feedback 蒸餾後預設 weight（高權重，必入 Stage 4 人工驗收） +_USER_FEEDBACK_DEFAULT_WEIGHT = 0.9 +_MANUAL_CURATED_DEFAULT_WEIGHT = 1.0 + + +# ───────────────────────────────────────────────────────────────────────────── +# 容器 +# ───────────────────────────────────────────────────────────────────────────── +@dataclass +class DistillResult: + """Distiller 產出的候選筆（尚未寫入 DB）。caller 拿到後可再調整或直接 enqueue。""" + + episode_type: str # mcp_result / llm_response / user_feedback / manual_curated + distilled_text: str + quality_score: float + weight: float + source_table: Optional[str] = None + source_id: Optional[int] = None + + +@dataclass +class GateDecision: + """PromotionGate 4 階段決策。""" + + can_promote: bool + reason: str # approved / awaiting_review / rejected_* + detail: Optional[str] = None # 失敗時的人類可讀原因 + similar_insight_id: Optional[int] = None # Stage 3 命中的相似 insight id + + +# ───────────────────────────────────────────────────────────────────────────── +# Distiller +# ───────────────────────────────────────────────────────────────────────────── +class Distiller: + """LLM/MCP 結果蒸餾器。 + + 純 Hermes 規則引擎（不再呼叫 LLM 避免循環燒錢）。 + 輸出 DistillResult，由 caller 決定是否 enqueue。 + """ + + def distill( + self, + episode_type: str, + raw_content: str, + source_table: Optional[str] = None, + source_id: Optional[int] = None, + user_feedback_score: Optional[int] = None, + ) -> Optional[DistillResult]: + """主入口。 + + Args: + episode_type: mcp_result / llm_response / user_feedback / manual_curated + raw_content: 原始文本（MCP 結果 / LLM response / 用戶留言） + source_table: ai_calls / mcp_calls / None + source_id: 對應 source_table.id + user_feedback_score: 1-5（episode_type=user_feedback 時帶入； + 5 → quality 1.0；1 → 0.0 負樣本不晉升） + + Returns: + DistillResult 或 None（不可蒸餾，例如空字串）。 + """ + if not raw_content or not raw_content.strip(): + return None + + cleaned = raw_content.strip() + # 截長度（與 028 CHECK octet_length<=16384 對齊） + encoded = cleaned.encode('utf-8', errors='replace') + if len(encoded) > DISTILLED_TEXT_MAX_BYTES: + cleaned = encoded[:DISTILLED_TEXT_MAX_BYTES].decode('utf-8', errors='ignore') + + et = (episode_type or '').strip() + if et not in ('mcp_result', 'llm_response', 'user_feedback', 'manual_curated'): + logger.warning("[Distiller] unknown episode_type=%s", et) + return None + + # ── 各類型品質規則 ── + if et == 'manual_curated': + quality, weight = 1.0, _MANUAL_CURATED_DEFAULT_WEIGHT + + elif et == 'user_feedback': + score = max(1, min(int(user_feedback_score or 3), 5)) + # 5 → 1.0 高品質；1 → 0.0 負樣本 + quality = round((score - 1) / 4.0, 3) + weight = _USER_FEEDBACK_DEFAULT_WEIGHT # 用戶直陳的事實，需人工驗收 + + elif et == 'mcp_result': + # MCP：>200 字 + 含 2+ 關鍵字 → 0.8；否則 0.5 + keyword_hits = sum(1 for kw in _LLM_KEYWORDS_HINT if kw in cleaned) + if len(cleaned) >= _MCP_MIN_LEN and keyword_hits >= 2: + quality = 0.8 + elif len(cleaned) >= _MCP_MIN_LEN: + quality = 0.65 + else: + quality = 0.5 + weight = 0.6 # MCP 事實但非用戶確認，中等權重 + + else: # llm_response + # LLM 結構化 JSON + status='ok' → 0.9；自由文本 >500 字 → 0.6 + quality, weight = self._distill_llm_response(cleaned) + + return DistillResult( + episode_type=et, + distilled_text=cleaned, + quality_score=round(quality, 3), + weight=round(weight, 3), + source_table=source_table, + source_id=source_id, + ) + + @staticmethod + def _distill_llm_response(text: str) -> Tuple[float, float]: + """LLM 回應蒸餾：JSON 結構化 vs 自由文本兩種路徑。""" + # 路徑 1：嘗試解析為 JSON（結構化 ⇒ 高品質） + stripped = text.strip() + if stripped.startswith('{') or stripped.startswith('['): + try: + obj = json.loads(stripped) + # status == 'ok' 或非空 dict/list → 0.9 + status_ok = ( + isinstance(obj, dict) and obj.get('status') == 'ok' + ) or ( + isinstance(obj, list) and len(obj) > 0 + ) or ( + isinstance(obj, dict) and len(obj) > 0 + ) + if status_ok: + return 0.9, 0.7 + return 0.7, 0.6 + except (json.JSONDecodeError, ValueError): + pass + + # 路徑 2：自由文本 — 長度 + 繁中數字判斷 + if len(text) >= _LLM_FREE_TEXT_MIN: + # >500 字 + 含具體數字 → 0.65；無數字 → 0.55 + if _NUMBER_PATTERN.search(text): + return 0.65, 0.55 + return 0.55, 0.5 + + # 太短的自由文本 → 0.4 不會過 Stage 1 + return 0.4, 0.5 + + +# ───────────────────────────────────────────────────────────────────────────── +# Learning Pipeline 主入口（enqueue + 整合 Distiller） +# ───────────────────────────────────────────────────────────────────────────── +class LearningPipeline: + """蒸餾 + enqueue 統一入口。 + + 使用範例: + from services.learning_pipeline import learning_pipeline + learning_pipeline.enqueue( + episode_type='llm_response', + raw_content=response_text, + source_table='ai_calls', source_id=ai_call_id, + ) + """ + + def __init__(self): + self.distiller = Distiller() + + def enqueue( + self, + episode_type: str, + raw_content: str, + source_table: Optional[str] = None, + source_id: Optional[int] = None, + user_feedback_score: Optional[int] = None, + ) -> Optional[int]: + """蒸餾後寫入 learning_episodes（pending 狀態）。 + + Returns: + learning_episodes.id 或 None（蒸餾失敗 / DB 寫入失敗）。 + """ + result = self.distiller.distill( + episode_type=episode_type, + raw_content=raw_content, + source_table=source_table, + source_id=source_id, + user_feedback_score=user_feedback_score, + ) + if not result: + return None + + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + row = session.execute( + sa_text(""" + INSERT INTO learning_episodes ( + episode_type, source_table, source_id, + distilled_text, quality_score, weight, + promotion_status + ) VALUES ( + :episode_type, :source_table, :source_id, + :distilled_text, :quality_score, :weight, + 'pending' + ) + RETURNING id + """), + { + 'episode_type': result.episode_type, + 'source_table': result.source_table, + 'source_id': result.source_id, + 'distilled_text': result.distilled_text, + 'quality_score': result.quality_score, + 'weight': result.weight, + }, + ).fetchone() + session.commit() + return int(row[0]) if row else None + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.warning("[LearningPipeline] enqueue failed: %s", exc) + return None + + +# ───────────────────────────────────────────────────────────────────────────── +# PromotionGate (v5.0 核心護欄 #1) +# ───────────────────────────────────────────────────────────────────────────── +class PromotionGate: + """learning_episodes → ai_insights 4 階段晉升閘。 + + Owen 強調：高權重必經人工驗收，避免幻覺污染 RAG。 + + 使用範例: + gate = PromotionGate() + decision = gate.can_promote(episode_id) + if decision.can_promote: + insight_id = gate.promote(episode_id) + elif decision.reason == 'awaiting_review': + # 推 Telegram 等 👍/👎 + else: + gate.reject(episode_id, decision.detail or decision.reason) + """ + + # ────────────────────────────────────────────────────────────────────── + # Stage 1: quality_score 自動門檻 + # ────────────────────────────────────────────────────────────────────── + def _stage_1_quality(self, episode: Dict[str, Any]) -> Optional[GateDecision]: + """quality_score < 0.7 → rejected_quality。""" + q = float(episode.get('quality_score') or 0) + if q < STAGE_1_AUTO_QUALITY: + return GateDecision( + can_promote=False, + reason='rejected_quality', + detail=f'quality_score={q:.3f} < {STAGE_1_AUTO_QUALITY}', + ) + return None + + # ────────────────────────────────────────────────────────────────────── + # Stage 2: 幻覺檢測（規則引擎） + # ────────────────────────────────────────────────────────────────────── + def _stage_2_hallucination(self, episode: Dict[str, Any]) -> Optional[GateDecision]: + """規則： + R1. 含 hedge words（'我猜' / '可能' / '也許'）+ 無具體數字 → suspect + R2. 文本中含「A 是 X」又含「A 是 Y」（同主詞矛盾，rule-light） → suspect + R3. （可擴充）引用 SKU 不在 product_master → suspect + + Returns: rejected_hallucination 或 None（通過）。 + """ + text = (episode.get('distilled_text') or '').strip() + if not text: + return None + + # R1: hedge + 無數字 + hedge_hits = [w for w in _HALLUCINATION_HEDGE_WORDS if w in text] + if hedge_hits and not _NUMBER_PATTERN.search(text): + return GateDecision( + can_promote=False, + reason='rejected_hallucination', + detail=f'hedge words {hedge_hits} 但缺具體數字', + ) + + # R2: 簡單矛盾偵測 — 找「X 是 A」與「X 是 B」（A != B） + contradiction = _detect_simple_contradiction(text) + if contradiction: + return GateDecision( + can_promote=False, + reason='rejected_hallucination', + detail=f'自相矛盾偵測: {contradiction}', + ) + + return None + + # ────────────────────────────────────────────────────────────────────── + # Stage 3: 去重（cosine similarity vs ai_insights） + # ────────────────────────────────────────────────────────────────────── + def _stage_3_dedup(self, episode: Dict[str, Any]) -> Optional[GateDecision]: + """與既有 ai_insights cosine similarity >= 0.95 → rejected_duplicate。 + + 若 episode.embedding 為 NULL（蒸餾時尚未 embed）：略過此 stage（warning）。 + """ + embedding = episode.get('embedding') + if not embedding: + logger.debug( + "[PromotionGate] episode_id=%s embedding 為 NULL，略過 Stage 3 去重", + episode.get('id'), + ) + return None + + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + row = session.execute( + sa_text(""" + SELECT id, content, + 1.0 - (embedding <=> CAST(:vec AS vector)) AS similarity + FROM ai_insights + WHERE embedding IS NOT NULL + AND status IN ('approved','active','executed') + ORDER BY embedding <=> CAST(:vec AS vector) ASC + LIMIT 1 + """), + {'vec': str(embedding)}, + ).fetchone() + finally: + session.close() + + if row and float(row.similarity or 0) >= STAGE_3_DEDUP_THRESHOLD: + return GateDecision( + can_promote=False, + reason='rejected_duplicate', + detail=f'similarity={row.similarity:.4f} >= {STAGE_3_DEDUP_THRESHOLD}', + similar_insight_id=int(row.id), + ) + return None + except Exception as exc: + logger.warning( + "[PromotionGate] Stage 3 dedup query failed (episode_id=%s): %s — 視為通過", + episode.get('id'), exc, + ) + return None + + # ────────────────────────────────────────────────────────────────────── + # Stage 4: 高權重強制人工驗收 + # ────────────────────────────────────────────────────────────────────── + def _stage_4_review(self, episode: Dict[str, Any]) -> GateDecision: + """weight >= 0.8 → awaiting_review；否則自動晉升 approved。""" + weight = float(episode.get('weight') or 0) + if weight >= STAGE_4_HUMAN_REVIEW_WEIGHT: + return GateDecision( + can_promote=False, + reason='awaiting_review', + detail=f'weight={weight:.3f} >= {STAGE_4_HUMAN_REVIEW_WEIGHT} 強制人工驗收', + ) + return GateDecision(can_promote=True, reason='approved') + + # ────────────────────────────────────────────────────────────────────── + # 主入口 + # ────────────────────────────────────────────────────────────────────── + def can_promote(self, episode_id: int) -> GateDecision: + """執行 4 階段檢查，回 GateDecision。 + + - approved → 可呼叫 promote() + - awaiting_review → 不晉升，caller 負責推 Telegram 等 👍/👎 + - rejected_* → 不晉升，caller 應呼叫 reject() 標狀態 + """ + episode = self._load_episode(episode_id) + if not episode: + return GateDecision( + can_promote=False, + reason='rejected_quality', + detail=f'episode_id={episode_id} not found', + ) + + for stage_fn in (self._stage_1_quality, self._stage_2_hallucination, self._stage_3_dedup): + verdict = stage_fn(episode) + if verdict: + return verdict + + return self._stage_4_review(episode) + + def promote(self, episode_id: int) -> Optional[int]: + """執行晉升：寫 ai_insights + 更新 learning_episodes.{insight_id, promotion_status}。 + + Returns: + ai_insights.id 或 None（晉升失敗）。 + + 注意：呼叫前 caller 必須先 can_promote() 確認 reason='approved'。 + 本函式不重複跑 4 stage（避免 race condition + 雙重檢查浪費 query）。 + """ + episode = self._load_episode(episode_id) + if not episode: + logger.warning("[PromotionGate] promote skipped: episode_id=%s not found", episode_id) + return None + + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + # 1. 寫入 ai_insights（content + insight_type 從 episode 推斷） + inferred_type = _infer_insight_type(episode) + row = session.execute( + sa_text(""" + INSERT INTO ai_insights ( + insight_type, content, avg_quality, status, + confidence, created_by + ) VALUES ( + :insight_type, :content, :quality, 'approved', + :confidence, 'learning_pipeline' + ) + RETURNING id + """), + { + 'insight_type': inferred_type, + 'content': episode['distilled_text'], + 'quality': float(episode.get('quality_score') or 0.5), + 'confidence': float(episode.get('weight') or 0.5), + }, + ).fetchone() + insight_id = int(row[0]) if row else None + + if not insight_id: + raise RuntimeError('ai_insights INSERT 未回 id') + + # 2. 更新 learning_episodes（approved + insight_id 回填，CHECK chk_le_approved_consistent 強制一致） + session.execute( + sa_text(""" + UPDATE learning_episodes + SET promotion_status = 'approved', + insight_id = :insight_id, + reviewed_at = NOW() + WHERE id = :id + """), + {'insight_id': insight_id, 'id': episode_id}, + ) + session.commit() + logger.info( + "[PromotionGate] episode_id=%s promoted → insight_id=%s", + episode_id, insight_id, + ) + return insight_id + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.error("[PromotionGate] promote failed (episode_id=%s): %s", episode_id, exc) + return None + + def reject(self, episode_id: int, reason: str, detail: Optional[str] = None) -> bool: + """拒絕晉升：標 promotion_status='rejected_*' + rejected_reason。 + + Args: + reason: rejected_quality / rejected_hallucination / rejected_duplicate / rejected_human + detail: 補充說明（會與 reason 拼成 rejected_reason 文本） + """ + valid_statuses = ( + 'rejected_quality', + 'rejected_hallucination', + 'rejected_duplicate', + 'rejected_human', + ) + if reason not in valid_statuses: + logger.warning("[PromotionGate] invalid reject reason=%s", reason) + return False + + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + full_reason = detail or reason + session.execute( + sa_text(""" + UPDATE learning_episodes + SET promotion_status = :status, + rejected_reason = :rej_reason, + reviewed_at = NOW() + WHERE id = :id + """), + {'status': reason, 'rej_reason': full_reason, 'id': episode_id}, + ) + session.commit() + return True + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.warning( + "[PromotionGate] reject failed (episode_id=%s, reason=%s): %s", + episode_id, reason, exc, + ) + return False + + def mark_awaiting_review(self, episode_id: int) -> bool: + """進入 awaiting_review 狀態（caller 推 Telegram 後呼叫，準備等 👍/👎）。""" + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + session.execute( + sa_text(""" + UPDATE learning_episodes + SET promotion_status = 'awaiting_review' + WHERE id = :id + AND promotion_status = 'pending' + """), + {'id': episode_id}, + ) + session.commit() + return True + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.warning( + "[PromotionGate] mark_awaiting_review failed (episode_id=%s): %s", + episode_id, exc, + ) + return False + + # ────────────────────────────────────────────────────────────────────── + # 內部 + # ────────────────────────────────────────────────────────────────────── + @staticmethod + def _load_episode(episode_id: int) -> Optional[Dict[str, Any]]: + """讀取 learning_episodes 單筆（dict 化方便 stage 函式 unit test mock）。""" + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + row = session.execute( + sa_text(""" + SELECT id, episode_type, distilled_text, + quality_score, weight, embedding, + promotion_status + FROM learning_episodes + WHERE id = :id + """), + {'id': int(episode_id)}, + ).fetchone() + if not row: + return None + return { + 'id': int(row.id), + 'episode_type': row.episode_type, + 'distilled_text': row.distilled_text, + 'quality_score': float(row.quality_score or 0), + 'weight': float(row.weight or 0), + 'embedding': row.embedding, + 'promotion_status': row.promotion_status, + } + finally: + session.close() + except Exception as exc: + logger.warning("[PromotionGate] load_episode failed (id=%s): %s", episode_id, exc) + return None + + +# ───────────────────────────────────────────────────────────────────────────── +# 24h 自動降級 +# ───────────────────────────────────────────────────────────────────────────── +def expire_stale_reviews(hours: int = HUMAN_REVIEW_TIMEOUT_HOURS) -> int: + """awaiting_review 超過 N 小時無 👍/👎 → 自動降級 weight=0.5 + status='expired'。 + + 呼叫時機：scheduler 每 4 小時跑一次（建議與 ai_calls 90 天保留同排程）。 + + Returns: + 被降級的筆數。 + """ + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + result = session.execute( + sa_text(""" + UPDATE learning_episodes + SET promotion_status = 'expired', + weight = :fallback_weight, + rejected_reason = COALESCE(rejected_reason, '') + || '24h 無人工反饋自動降級', + reviewed_at = NOW() + WHERE promotion_status = 'awaiting_review' + AND created_at < NOW() - (:hours || ' hours')::INTERVAL + """), + { + 'fallback_weight': EXPIRED_FALLBACK_WEIGHT, + 'hours': str(int(hours)), + }, + ) + session.commit() + count = result.rowcount or 0 + if count: + logger.info( + "[PromotionGate] expire_stale_reviews: %d episodes 降級 weight=%.2f", + count, EXPIRED_FALLBACK_WEIGHT, + ) + return count + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.warning("[PromotionGate] expire_stale_reviews failed: %s", exc) + return 0 + + +# ───────────────────────────────────────────────────────────────────────────── +# 工具函式 +# ───────────────────────────────────────────────────────────────────────────── +def _detect_simple_contradiction(text: str) -> Optional[str]: + """偵測「X 是 A」與「X 是 B」（A != B）的同主詞矛盾。 + + 純規則 light 偵測；不深做語意，避免誤殺合理推論。 + Returns: 矛盾描述 或 None。 + """ + pattern = re.compile(r'([一-龥A-Za-z0-9]{1,8})\s*[是為]\s*([一-龥A-Za-z0-9]{1,12})') + matches = pattern.findall(text) + if len(matches) < 2: + return None + seen: Dict[str, str] = {} + for subject, value in matches: + if subject in seen and seen[subject] != value: + return f'"{subject}" 同時被指為 "{seen[subject]}" 與 "{value}"' + seen[subject] = value + return None + + +def _infer_insight_type(episode: Dict[str, Any]) -> str: + """從 episode 推斷 ai_insights.insight_type。 + + 規則： + - episode_type=user_feedback → 'human_review' + - episode_type=manual_curated → 'manual_curated' + - episode_type=mcp_result → 'mcp_grounding' + - episode_type=llm_response → 'llm_distilled' + """ + et = episode.get('episode_type') or '' + return { + 'user_feedback': 'human_review', + 'manual_curated': 'manual_curated', + 'mcp_result': 'mcp_grounding', + 'llm_response': 'llm_distilled', + }.get(et, 'llm_distilled') + + +def hash_human_approver(username: str) -> str: + """Telegram username SHA1[:8]（與 028 schema 註解一致 — 避免 PII 落地）。""" + if not username: + return '' + return hashlib.sha1(username.encode('utf-8')).hexdigest()[:8] + + +# ───────────────────────────────────────────────────────────────────────────── +# 全域單例 +# ───────────────────────────────────────────────────────────────────────────── +distiller = Distiller() +learning_pipeline = LearningPipeline() +promotion_gate = PromotionGate() + + +__all__ = [ + 'Distiller', + 'DistillResult', + 'GateDecision', + 'LearningPipeline', + 'PromotionGate', + 'distiller', + 'learning_pipeline', + 'promotion_gate', + 'expire_stale_reviews', + 'hash_human_approver', + 'STAGE_1_AUTO_QUALITY', + 'STAGE_3_DEDUP_THRESHOLD', + 'STAGE_4_HUMAN_REVIEW_WEIGHT', + 'HUMAN_REVIEW_TIMEOUT_HOURS', +] diff --git a/services/openclaw_strategist_service.py b/services/openclaw_strategist_service.py index b77bd42..68263c6 100644 --- a/services/openclaw_strategist_service.py +++ b/services/openclaw_strategist_service.py @@ -33,6 +33,7 @@ from database.manager import get_session from sqlalchemy import bindparam, text from services.ai_call_logger import log_ai_call # Operation Ollama-First v5.0 P1 +from services.rag_service import rag_service, is_rag_enabled # Phase 11 RAG-first（Q&A 限定） logger = logging.getLogger(__name__) @@ -171,6 +172,24 @@ def generate_strategy_response(query: str, context: Optional[Dict[str, Any]] = N request_id = f"qa-{uuid.uuid4().hex[:8]}" + # ── Phase 11 RAG-first（feature flag 預設 OFF；只在 Q&A 入口接，週月年報不接）── + # 高信心 RAG 命中 → 直接回 ai_insights 內容，避免 LLM 呼叫 + # 低信心或 flag OFF → 走既有 Ollama → Gemini → NIM 路徑 + if is_rag_enabled(): + try: + rag = rag_service.query( + text=q, caller='openclaw_qa', + threshold=0.85, request_id=request_id, + ) + if rag.has_high_confidence: + logger.info( + "[OpenClaw][QA] RAG hit request_id=%s top_score=%.3f → 跳過 LLM", + request_id, rag.hits[0].get('score', 0), + ) + return rag.synthesize() + except Exception as exc: + logger.warning("[OpenClaw][QA] RAG query failed (%s), fallback LLM", exc) + # ── 灰度路徑：Ollama 優先（flag=true 才走，預設 OFF）── if _qa_ollama_first_enabled(): ollama_reply = _call_qwen3_qa(q, context, request_id) diff --git a/services/rag_service.py b/services/rag_service.py new file mode 100644 index 0000000..282f34b --- /dev/null +++ b/services/rag_service.py @@ -0,0 +1,532 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +services/rag_service.py +Operation Ollama-First v5.0 / Phase 11 — RAG 查詢服務 + +設計原則（憲法級）： + 1. 純讀 ai_insights + 寫 rag_query_log（不動 ai_insights schema） + 2. cosine similarity threshold 預設 0.85，太低不採用避免幻覺 + 3. embedding 走 bge-m3:latest（與 ai_insights 一致簽名 — Phase 11.0 護欄 #3） + 4. feature flag RAG_ENABLED 預設 OFF（避免影響戰前行為） + 5. 失敗安全：DB 掛 / embedding 失敗 / threshold 不到 → 回 RAGResult(hits=[]) + caller 自行 fallback LLM + 6. fire-and-forget log：rag_query_log INSERT async daemon thread，不阻塞主流程 + 7. PII 保護：query_text 寫入時截 4KB（CHECK constraint 也會擋） + +對應: + - migrations/027_create_rag_query_log.sql + - migrations/026_add_embedding_signature.sql + - docs/adr/ADR-029-hermes-first-twin-tower.md + - docs/phase0_audit_report_20260503.md Section 3 (BGE-M3 一致性護欄) + +主入口: + - rag_service.query(...) : 主查詢介面 + - rag_service.feedback(log_id, ...) : Telegram 👍/👎 反饋寫回 + - rag_service.invalidate_by_caller(...) : 預留快取失效鉤（v5.0 暫無 cache 層） +""" + +from __future__ import annotations + +import hashlib +import logging +import os +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +# ───────────────────────────────────────────────────────────────────────────── +# Feature flag + 預設參數 +# ───────────────────────────────────────────────────────────────────────────── +def is_rag_enabled() -> bool: + """環境變數即時讀取（允許 runtime toggle，與 ai_call_logger 一致風格）。 + + feature flag 預設 OFF — 戰前部署後行為與 v4.x 完全相同。 + """ + val = os.environ.get('RAG_ENABLED', 'false').strip().lower() + return val in ('true', '1', 'yes', 'on') + + +# 內部別名（沿用既有 _is_rag_enabled 命名相容性） +_is_rag_enabled = is_rag_enabled + + +RAG_DEFAULT_THRESHOLD = float(os.getenv('RAG_DEFAULT_THRESHOLD', '0.85')) +RAG_DEFAULT_TOP_K = int(os.getenv('RAG_DEFAULT_TOP_K', '5')) + +# bge-m3 一致性參數（與 ai_insights 簽名計算同源） +RAG_EMBED_MODEL = os.getenv('RAG_EMBED_MODEL', 'bge-m3:latest') +RAG_EMBED_DIM = int(os.getenv('RAG_EMBED_DIM', '1024')) +RAG_EMBED_NORMALIZE = os.getenv('RAG_EMBED_NORMALIZE', 'true').strip().lower() in ( + 'true', '1', 'yes', 'on', +) + +# query_text 寫入長度上限（與 027 CHECK octet_length<=4096 對齊；中文 1 字 3 byte → ~1300 字） +_QUERY_TEXT_MAX_BYTES = 4096 + +# 連續失敗門檻（與 ai_call_logger 同模式） +_MAX_CONSECUTIVE_FAILURES = 10 +_failure_counter_lock = threading.Lock() +_failure_state = {'count': 0, 'killed': False} + + +def _record_failure() -> None: + with _failure_counter_lock: + _failure_state['count'] += 1 + if _failure_state['count'] >= _MAX_CONSECUTIVE_FAILURES and not _failure_state['killed']: + _failure_state['killed'] = True + logger.error( + "[RAGService] consecutive write failures hit %d — kill-switch ON, " + "downgrading rag_query_log writes to logger.info", + _MAX_CONSECUTIVE_FAILURES, + ) + + +def _record_success() -> None: + with _failure_counter_lock: + if _failure_state['count'] > 0: + _failure_state['count'] = 0 + + +def _is_killed() -> bool: + with _failure_counter_lock: + return _failure_state['killed'] + + +def _reset_kill_switch() -> None: + """測試專用：重置 kill-switch 狀態。""" + with _failure_counter_lock: + _failure_state['count'] = 0 + _failure_state['killed'] = False + + +# ───────────────────────────────────────────────────────────────────────────── +# BGE-M3 一致性簽名（v5.0 護欄 #3） +# 與 migration 026 註解一致：SHA1({model}|{normalize}|{dim}|{ollama_digest})[:12] +# Python 端不查 ollama digest（避免每次 query 都 GET /api/show）， +# 改用 model+normalize+dim 三元組已足以擋住「升級 bge-m3 / 改 normalize」雙寫漂移。 +# ───────────────────────────────────────────────────────────────────────────── +def get_embedding_signature( + model: str = RAG_EMBED_MODEL, + dim: int = RAG_EMBED_DIM, + normalize: bool = RAG_EMBED_NORMALIZE, +) -> str: + """產生 12 碼 BGE-M3 一致性簽名。 + + 與 ai_insights.embedding_signature 比對；不一致 → log warning + 不採該筆。 + """ + raw = f"{model}|{str(normalize).lower()}|{dim}" + return hashlib.sha1(raw.encode('utf-8')).hexdigest()[:12] + + +# ───────────────────────────────────────────────────────────────────────────── +# 結果容器 +# ───────────────────────────────────────────────────────────────────────────── +@dataclass +class RAGResult: + """RAG 查詢結果。caller 透過 has_high_confidence / synthesize() 決定是否走 LLM。""" + + query: str + embedding_signature: str + hits: List[Dict[str, Any]] = field(default_factory=list) + threshold: float = RAG_DEFAULT_THRESHOLD + saved_call: bool = False # 是否成功避免 LLM 呼叫（caller 確認後設定） + duration_ms: int = 0 + log_id: Optional[int] = None # rag_query_log.id（fire-and-forget，可能為 None） + + @property + def has_high_confidence(self) -> bool: + """有至少 1 個 hit 且 top-1 score >= threshold。""" + if not self.hits: + return False + top_score = self.hits[0].get('score', 0.0) or 0.0 + return float(top_score) >= self.threshold + + def synthesize(self) -> str: + """組合前 3 筆 hits.content（用 \\n\\n---\\n\\n 分隔，與 OCLearn 既有風格一致）。 + + caller 拿到後可直接當 LLM 回覆呈現給用戶；避免再次 LLM 呼叫。 + """ + if not self.hits: + return "" + parts = [] + for h in self.hits[:3]: + content = h.get('content') or "" + if content: + parts.append(content) + return "\n\n---\n\n".join(parts) + + +# ───────────────────────────────────────────────────────────────────────────── +# 主類別 +# ───────────────────────────────────────────────────────────────────────────── +class RAGService: + """RAG 查詢主入口 — 雙寫 rag_query_log + 回傳 hits。 + + 使用範例: + from services.rag_service import rag_service + result = rag_service.query("本週業績趨勢", caller='openclaw_qa') + if result.has_high_confidence: + return result.synthesize() + # 否則走既有 LLM 路徑 + """ + + def query( + self, + text: str, + caller: str, + top_k: int = RAG_DEFAULT_TOP_K, + threshold: float = RAG_DEFAULT_THRESHOLD, + request_id: Optional[str] = None, + insight_type: Optional[str] = None, + ) -> RAGResult: + """執行 RAG 召回。 + + Args: + text: 查詢文本（用戶問題或 LLM prompt） + caller: 與 ai_calls.caller 同白名單（hermes_qa / openclaw_qa / ...） + top_k: 召回筆數（1-50） + threshold: cosine similarity 門檻（0-1，預設 0.85） + request_id: 與 ai_calls.request_id 串鏈 + insight_type: 限制 ai_insights.insight_type（None = 全類型） + + Returns: + RAGResult。失敗時 hits=[] + duration_ms 仍記錄。 + """ + signature = get_embedding_signature() + start = time.monotonic() + + # ── 路徑 1：feature flag OFF → 短路（不查 DB / 不寫 log）── + if not _is_rag_enabled(): + return RAGResult( + query=text or "", + embedding_signature=signature, + threshold=threshold, + duration_ms=0, + ) + + # ── 路徑 2：empty text → 早退（避免無謂 embedding 呼叫）── + if not text or not text.strip(): + return RAGResult( + query="", + embedding_signature=signature, + threshold=threshold, + duration_ms=int((time.monotonic() - start) * 1000), + ) + + # 護欄：top_k / threshold 範圍夾擠（與 027 CHECK 對齊） + top_k = max(1, min(int(top_k or RAG_DEFAULT_TOP_K), 50)) + threshold = max(0.0, min(float(threshold or RAG_DEFAULT_THRESHOLD), 1.0)) + + # ── 路徑 3：embedding ── + query_vec: Optional[List[float]] = None + try: + from services.ollama_service import ollama_service + query_vec = ollama_service.generate_embedding(text, model=RAG_EMBED_MODEL) + if not query_vec: + logger.warning( + "[RAGService] embedding empty (caller=%s, len=%d) — fallback LLM", + caller, len(text), + ) + except Exception as exc: + logger.warning( + "[RAGService] embedding failed (caller=%s): %s — fallback LLM", + caller, exc, + ) + + hits: List[Dict[str, Any]] = [] + + # ── 路徑 4：DB 召回（只在 embedding 成功時）── + if query_vec: + try: + hits = self._select_hits( + query_vec=query_vec, + threshold=threshold, + top_k=top_k, + insight_type=insight_type, + expected_signature=signature, + ) + except Exception as exc: + logger.warning( + "[RAGService] DB select failed (caller=%s): %s — fallback LLM", + caller, exc, + ) + hits = [] + + duration_ms = int((time.monotonic() - start) * 1000) + result = RAGResult( + query=text, + embedding_signature=signature, + hits=hits, + threshold=threshold, + saved_call=False, # caller 確認後再設（_call_after_decision API 預留） + duration_ms=duration_ms, + ) + + # ── 路徑 5：fire-and-forget rag_query_log ── + self._async_log( + caller=caller, + text=text, + query_vec=query_vec, + top_k=top_k, + threshold=threshold, + hits=hits, + request_id=request_id, + ) + + return result + + # ────────────────────────────────────────────────────────────────────── + # DB 召回 + # ────────────────────────────────────────────────────────────────────── + def _select_hits( + self, + query_vec: List[float], + threshold: float, + top_k: int, + insight_type: Optional[str], + expected_signature: str, + ) -> List[Dict[str, Any]]: + """從 ai_insights 召回 top_k 筆（cosine similarity >= threshold）。 + + embedding_signature 不一致的列：log warning + 不採該筆（v5.0 護欄 #3）。 + """ + from sqlalchemy import text as sa_text + from database.manager import get_session + + # cosine_distance = embedding <=> qvec； similarity = 1 - distance + # 多取 top_k * 2 緩衝給簽名漂移過濾，最終裁回 top_k + fetch_limit = max(top_k * 2, top_k) + filters = [ + "embedding IS NOT NULL", + "status IN ('approved', 'active', 'executed')", + ] + params: Dict[str, Any] = { + 'qvec': str(query_vec), + 'lim': fetch_limit, + 'max_distance': 1.0 - threshold, + } + if insight_type: + filters.append("insight_type = :insight_type") + params['insight_type'] = insight_type + + sql = sa_text(f""" + SELECT id, insight_type, period, content, + embedding_signature, + embedding <=> CAST(:qvec AS vector) AS distance + FROM ai_insights + WHERE {' AND '.join(filters)} + AND (embedding <=> CAST(:qvec AS vector)) <= :max_distance + ORDER BY distance ASC + LIMIT :lim + """) + + session = get_session() + try: + rows = session.execute(sql, params).fetchall() + finally: + session.close() + + hits: List[Dict[str, Any]] = [] + signature_mismatch = 0 + for row in rows: + if len(hits) >= top_k: + break + row_signature = getattr(row, 'embedding_signature', None) + # v5.0 護欄 #3：簽名漂移檢查（NULL = 既有未回填資料，暫時放行避免戰前資料完全失效） + if row_signature and row_signature != expected_signature: + signature_mismatch += 1 + continue + distance = float(row.distance or 1.0) + similarity = 1.0 - distance + hits.append({ + 'id': int(row.id), + 'insight_type': row.insight_type, + 'period': row.period, + 'content': row.content or '', + 'score': round(similarity, 4), + 'distance': round(distance, 4), + 'embedding_signature': row_signature, + }) + + if signature_mismatch: + logger.warning( + "[RAGService] %d hits skipped due to embedding_signature mismatch " + "(expected=%s); 建議跑批次回填腳本", + signature_mismatch, expected_signature, + ) + + return hits + + # ────────────────────────────────────────────────────────────────────── + # 反饋（Telegram 👍/👎） + # ────────────────────────────────────────────────────────────────────── + def feedback(self, rag_query_log_id: int, score: int) -> bool: + """寫回 rag_query_log.feedback_score。 + + Args: + rag_query_log_id: rag_query_log.id + score: 1-5（1=很沒用，5=非常有用；常用：5=👍，1=👎） + + Returns: + True 寫入成功；False 寫入失敗（不 raise，靜默 log warning）。 + """ + if not rag_query_log_id or not isinstance(rag_query_log_id, int): + return False + score = max(1, min(int(score or 0), 5)) + + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + session = get_session() + try: + session.execute( + sa_text(""" + UPDATE rag_query_log + SET feedback_score = :score + WHERE id = :id + """), + {'score': score, 'id': rag_query_log_id}, + ) + session.commit() + return True + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + logger.warning( + "[RAGService] feedback write failed (id=%s, score=%s): %s", + rag_query_log_id, score, exc, + ) + return False + + # ────────────────────────────────────────────────────────────────────── + # 預留：caller 級失效（v5.0 暫無 in-memory cache 層） + # ────────────────────────────────────────────────────────────────────── + def invalidate_by_caller(self, caller: str) -> None: + """預留鉤：caller 的 prompt 模板更新時呼叫。 + + v5.0 RAG 主要靠 ai_insights 寫入時的 embedding 自動更新， + 無 in-memory cache 層 → 此函式為 no-op，留 API 一致性給後續 cache layer 啟用。 + """ + if caller: + logger.debug("[RAGService] invalidate_by_caller(%s) — no-op (no cache layer yet)", caller) + + # ────────────────────────────────────────────────────────────────────── + # fire-and-forget log + # ────────────────────────────────────────────────────────────────────── + def _async_log( + self, + caller: str, + text: str, + query_vec: Optional[List[float]], + top_k: int, + threshold: float, + hits: List[Dict[str, Any]], + request_id: Optional[str], + ) -> None: + """放到 daemon thread 寫入 rag_query_log，主流程不阻塞。 + + kill-switch 觸發 → 退化為 logger.info。 + """ + if _is_killed(): + logger.info( + "[RAGQuery|killed] caller=%s hits=%d threshold=%.3f request_id=%s", + caller, len(hits), threshold, request_id, + ) + return + + threading.Thread( + target=self._write_log, + args=(caller, text, query_vec, top_k, threshold, hits, request_id), + name=f"rag-query-log-{caller}", + daemon=True, + ).start() + + def _write_log( + self, + caller: str, + text: str, + query_vec: Optional[List[float]], + top_k: int, + threshold: float, + hits: List[Dict[str, Any]], + request_id: Optional[str], + ) -> None: + """try/except 全包；DB 掛了只 log warning 不爆炸。""" + try: + from sqlalchemy import text as sa_text + from database.manager import get_session + + # PII 保護：query_text 截 4KB（與 027 CHECK 對齊） + safe_text = (text or '') + encoded = safe_text.encode('utf-8', errors='replace') + if len(encoded) > _QUERY_TEXT_MAX_BYTES: + # 截到 byte 邊界後 decode 容錯（errors='ignore' 避免 UTF-8 multi-byte 截斷） + safe_text = encoded[:_QUERY_TEXT_MAX_BYTES].decode('utf-8', errors='ignore') + + used_results = [int(h['id']) for h in hits if h.get('id')] + embedding_str = str(query_vec) if query_vec else None + + session = get_session() + try: + session.execute( + sa_text(""" + INSERT INTO rag_query_log ( + caller, query_text, query_embedding, + top_k, threshold, + hit_count, used_results, + saved_call, request_id + ) VALUES ( + :caller, :query_text, + CAST(:embedding AS vector), + :top_k, :threshold, + :hit_count, CAST(:used_results AS BIGINT[]), + :saved_call, :request_id + ) + """), + { + 'caller': (caller or 'unknown')[:64], + 'query_text': safe_text, + 'embedding': embedding_str, + 'top_k': int(top_k), + 'threshold': round(float(threshold), 3), + 'hit_count': len(hits), + 'used_results': used_results if used_results else None, + 'saved_call': False, # caller 確認後另寫；INSERT 階段固定 False + 'request_id': (request_id or None), + }, + ) + session.commit() + _record_success() + except Exception: + session.rollback() + raise + finally: + session.close() + except Exception as exc: + _record_failure() + logger.warning( + "[RAGService] rag_query_log write failed (caller=%s): %s", + caller, exc, + ) + + +# 全域單例（與 ollama_service / ai_call_logger 同模式） +rag_service = RAGService() + + +__all__ = [ + 'RAGService', + 'RAGResult', + 'rag_service', + 'get_embedding_signature', + 'is_rag_enabled', +] diff --git a/services/telegram_templates.py b/services/telegram_templates.py index 173fd45..a292b12 100644 --- a/services/telegram_templates.py +++ b/services/telegram_templates.py @@ -369,6 +369,47 @@ def batch_decision_msg(items: List[Dict], batch_id: str) -> tuple: return "\n".join(lines), keyboard +# ══════════════════════════════════════════════════════════════════════════════ +# 🧠 Phase 11 RAG 反饋（v5.0 護欄 #1：強制晉升門檻 — Stage 4 人工驗收） +# ══════════════════════════════════════════════════════════════════════════════ + +def rag_feedback_keyboard(rag_query_log_id: int) -> dict: + """產生 RAG 命中後的 👍/👎 反饋鍵盤（callback prefix 'rag_fb:'）。 + + callback_data 設計（與 ADR-012 短 prefix 規範一致，<= 64 byte）: + rag_fb:{log_id}:5 → 👍（feedback_score=5） + rag_fb:{log_id}:1 → 👎（feedback_score=1） + + 使用範例（caller 拿 RAGResult 後）: + from services.telegram_templates import rag_feedback_keyboard + send_message(chat_id, rag.synthesize(), + keyboard=rag_feedback_keyboard(rag.log_id)) + """ + _id = int(rag_query_log_id) if rag_query_log_id else 0 + return { + 'inline_keyboard': [[ + {'text': '👍 有用', 'callback_data': f'rag_fb:{_id}:5'}, + {'text': '👎 沒用', 'callback_data': f'rag_fb:{_id}:1'}, + ]], + } + + +def promotion_review_keyboard(episode_id: int) -> dict: + """蒸餾池高權重晉升人工驗收鍵盤（PromotionGate Stage 4）。 + + callback_data: + pg_ok:{episode_id} → 通過 → 寫 ai_insights + pg_no:{episode_id} → 駁回 → rejected_human + """ + _id = int(episode_id) if episode_id else 0 + return { + 'inline_keyboard': [[ + {'text': '✅ 通過晉升', 'callback_data': f'pg_ok:{_id}'}, + {'text': '🚫 駁回', 'callback_data': f'pg_no:{_id}'}, + ]], + } + + # ══════════════════════════════════════════════════════════════════════════════ # 🤖 系統類模板 # ══════════════════════════════════════════════════════════════════════════════ diff --git a/tests/test_learning_pipeline.py b/tests/test_learning_pipeline.py new file mode 100644 index 0000000..d3af0e0 --- /dev/null +++ b/tests/test_learning_pipeline.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +tests/test_learning_pipeline.py +Operation Ollama-First v5.0 / Phase 11 — Distiller + LearningPipeline 單元測試 + +涵蓋： + - Distiller 各 quality_score 規則（mcp / llm_response / user_feedback / manual_curated） + - LearningPipeline.enqueue() DB 寫入路徑 + - expire_stale_reviews() 24h 自動降級 + - hash_human_approver() PII 保護 +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest + + +# ───────────────────────────────────────────────────────────────────────────── +# Distiller 各規則 +# ───────────────────────────────────────────────────────────────────────────── +class TestDistillerMcpResult: + def test_long_with_keywords_high_quality(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "本週業績分析顯示，建議聚焦保濕品類。" + "詳細說明 " * 80 # > 200 字 + result = d.distill(episode_type='mcp_result', raw_content=text) + assert result is not None + assert result.quality_score == 0.8 + assert result.episode_type == 'mcp_result' + + def test_long_no_keywords_medium_quality(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "啦啦啦" * 100 # > 200 字但無關鍵字 + result = d.distill(episode_type='mcp_result', raw_content=text) + assert result.quality_score == 0.65 + + def test_short_low_quality(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "短內容" + result = d.distill(episode_type='mcp_result', raw_content=text) + assert result.quality_score == 0.5 + + def test_empty_returns_none(self): + from services.learning_pipeline import Distiller + d = Distiller() + assert d.distill(episode_type='mcp_result', raw_content='') is None + assert d.distill(episode_type='mcp_result', raw_content=' ') is None + + +class TestDistillerLlmResponse: + def test_json_structured_high_quality(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = json.dumps({"status": "ok", "summary": "本週重點"}) + result = d.distill(episode_type='llm_response', raw_content=text) + assert result.quality_score == 0.9 + + def test_json_array_non_empty_high(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = json.dumps([{"sku": "A001", "risk": "HIGH"}]) + result = d.distill(episode_type='llm_response', raw_content=text) + assert result.quality_score == 0.9 + + def test_json_dict_no_status_lower(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = json.dumps({"some_field": "value"}) + result = d.distill(episode_type='llm_response', raw_content=text) + # dict 非空 → 0.9 (status_ok 條件含 "len(obj)>0") + assert result.quality_score == 0.9 + + def test_free_text_long_with_numbers(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "本週業績漲了 15.3%。" + "詳細說明 " * 100 # > 500 字 + 數字 + result = d.distill(episode_type='llm_response', raw_content=text) + assert result.quality_score == 0.65 + + def test_free_text_long_no_numbers(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "本週業績趨勢上升。" + "詳細說明 " * 100 # > 500 字無數字 + result = d.distill(episode_type='llm_response', raw_content=text) + assert result.quality_score == 0.55 + + def test_free_text_short_below_quality_gate(self): + from services.learning_pipeline import Distiller + d = Distiller() + text = "本週業績有變化" # 短文本 + result = d.distill(episode_type='llm_response', raw_content=text) + # 0.4 → Stage 1 會 reject + assert result.quality_score == 0.4 + + +class TestDistillerUserFeedback: + def test_score_5_high_quality(self): + from services.learning_pipeline import Distiller + d = Distiller() + result = d.distill( + episode_type='user_feedback', + raw_content='這個建議幫我增加了 8% 銷量', + user_feedback_score=5, + ) + assert result.quality_score == 1.0 + assert result.weight == 0.9 # 高權重 → Stage 4 人工驗收 + + def test_score_1_negative_sample(self): + from services.learning_pipeline import Distiller + d = Distiller() + result = d.distill( + episode_type='user_feedback', + raw_content='完全沒幫助', + user_feedback_score=1, + ) + assert result.quality_score == 0.0 # Stage 1 reject + + def test_default_score_3_mid(self): + from services.learning_pipeline import Distiller + d = Distiller() + result = d.distill( + episode_type='user_feedback', + raw_content='普通', + user_feedback_score=None, + ) + # 預設 3 → (3-1)/4 = 0.5 + assert result.quality_score == 0.5 + + +class TestDistillerManualCurated: + def test_max_quality_and_weight(self): + from services.learning_pipeline import Distiller + d = Distiller() + result = d.distill(episode_type='manual_curated', raw_content='手動入庫') + assert result.quality_score == 1.0 + assert result.weight == 1.0 + + +class TestDistillerInvalidType: + def test_unknown_type_returns_none(self): + from services.learning_pipeline import Distiller + d = Distiller() + result = d.distill(episode_type='garbage', raw_content='whatever') + assert result is None + + +class TestDistillerLengthGuard: + def test_distilled_text_truncated_to_16kb(self): + from services.learning_pipeline import Distiller, DISTILLED_TEXT_MAX_BYTES + d = Distiller() + text = '建議分析 ' * 5000 # 遠超 16KB + result = d.distill(episode_type='mcp_result', raw_content=text) + encoded = result.distilled_text.encode('utf-8') + assert len(encoded) <= DISTILLED_TEXT_MAX_BYTES + + +# ───────────────────────────────────────────────────────────────────────────── +# LearningPipeline.enqueue +# ───────────────────────────────────────────────────────────────────────────── +class TestLearningPipelineEnqueue: + def test_enqueue_returns_id_on_success(self, monkeypatch): + from services.learning_pipeline import learning_pipeline + + fake_session = MagicMock() + fake_row = MagicMock() + fake_row.__getitem__.return_value = 42 + fake_session.execute.return_value.fetchone.return_value = fake_row + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + new_id = learning_pipeline.enqueue( + episode_type='manual_curated', + raw_content='手動入庫測試內容', + ) + assert new_id == 42 + fake_session.commit.assert_called_once() + + def test_enqueue_returns_none_when_distill_fails(self): + from services.learning_pipeline import learning_pipeline + # 空內容 → distill 回 None → enqueue 回 None + result = learning_pipeline.enqueue( + episode_type='mcp_result', + raw_content='', + ) + assert result is None + + def test_enqueue_db_failure_returns_none(self, monkeypatch): + from services.learning_pipeline import learning_pipeline + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db down") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + result = learning_pipeline.enqueue( + episode_type='manual_curated', + raw_content='測試內容', + ) + assert result is None + + +# ───────────────────────────────────────────────────────────────────────────── +# expire_stale_reviews +# ───────────────────────────────────────────────────────────────────────────── +class TestExpireStaleReviews: + def test_expire_uses_correct_sql(self, monkeypatch): + from services.learning_pipeline import expire_stale_reviews + + fake_session = MagicMock() + fake_result = MagicMock() + fake_result.rowcount = 3 + fake_session.execute.return_value = fake_result + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + count = expire_stale_reviews(hours=24) + assert count == 3 + # 確認 commit 跑了 + fake_session.commit.assert_called_once() + + def test_expire_db_failure_returns_zero(self, monkeypatch): + from services.learning_pipeline import expire_stale_reviews + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db down") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + count = expire_stale_reviews(hours=24) + assert count == 0 + + +# ───────────────────────────────────────────────────────────────────────────── +# hash_human_approver +# ───────────────────────────────────────────────────────────────────────────── +class TestHashHumanApprover: + def test_returns_8_char_hex(self): + from services.learning_pipeline import hash_human_approver + h = hash_human_approver('owen.tsai') + assert len(h) == 8 + assert all(c in '0123456789abcdef' for c in h) + + def test_empty_returns_empty(self): + from services.learning_pipeline import hash_human_approver + assert hash_human_approver('') == '' + assert hash_human_approver(None) == '' # type: ignore + + def test_deterministic(self): + from services.learning_pipeline import hash_human_approver + a = hash_human_approver('alice') + b = hash_human_approver('alice') + c = hash_human_approver('bob') + assert a == b + assert a != c + + +# ───────────────────────────────────────────────────────────────────────────── +# 工具函式：_detect_simple_contradiction +# ───────────────────────────────────────────────────────────────────────────── +class TestContradictionDetector: + def test_no_contradiction_returns_none(self): + from services.learning_pipeline import _detect_simple_contradiction + text = "業績是上升。市場是競爭。" + # subject=業績→上升, subject=市場→競爭，沒矛盾 + assert _detect_simple_contradiction(text) is None + + def test_contradiction_detected(self): + from services.learning_pipeline import _detect_simple_contradiction + text = "A是黑色。A是白色。" + result = _detect_simple_contradiction(text) + assert result is not None + assert 'A' in result diff --git a/tests/test_promotion_gate.py b/tests/test_promotion_gate.py new file mode 100644 index 0000000..9c64aab --- /dev/null +++ b/tests/test_promotion_gate.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +tests/test_promotion_gate.py +Operation Ollama-First v5.0 / Phase 11 — PromotionGate 4 階段晉升閘單元測試 + +涵蓋： + Stage 1: quality_score < 0.7 → rejected_quality + Stage 2: 規則引擎幻覺檢測（hedge words / 矛盾） + Stage 3: cosine similarity >= 0.95 → rejected_duplicate + Stage 4: weight >= 0.8 強制 awaiting_review（不能跳） + promote() / reject() / mark_awaiting_review() DB 操作 +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + + +# ───────────────────────────────────────────────────────────────────────────── +# 共用工具 +# ───────────────────────────────────────────────────────────────────────────── +def _fake_episode( + id_=1, episode_type='llm_response', + distilled_text='內容', quality_score=0.8, weight=0.5, + embedding=None, status='pending', +): + return { + 'id': id_, + 'episode_type': episode_type, + 'distilled_text': distilled_text, + 'quality_score': quality_score, + 'weight': weight, + 'embedding': embedding, + 'promotion_status': status, + } + + +def _patch_load_episode(monkeypatch, episode): + """讓 PromotionGate._load_episode 直接回 episode（不走 DB）。""" + from services.learning_pipeline import PromotionGate + monkeypatch.setattr( + PromotionGate, '_load_episode', + staticmethod(lambda episode_id: episode if episode else None), + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# Stage 1: quality_score +# ───────────────────────────────────────────────────────────────────────────── +class TestStage1Quality: + def test_low_quality_rejected(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode(quality_score=0.5) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is False + assert decision.reason == 'rejected_quality' + assert '0.500' in (decision.detail or '') + + def test_quality_at_threshold_passes(self, monkeypatch): + from services.learning_pipeline import PromotionGate, STAGE_1_AUTO_QUALITY + ep = _fake_episode(quality_score=STAGE_1_AUTO_QUALITY, weight=0.5) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + # 過 Stage 1 + 2 + 3 + 4 自動晉升 + assert decision.can_promote is True + assert decision.reason == 'approved' + + def test_episode_not_found(self, monkeypatch): + from services.learning_pipeline import PromotionGate + _patch_load_episode(monkeypatch, None) + + gate = PromotionGate() + decision = gate.can_promote(99999) + assert decision.can_promote is False + assert decision.reason == 'rejected_quality' + assert 'not found' in (decision.detail or '') + + +# ───────────────────────────────────────────────────────────────────────────── +# Stage 2: 幻覺檢測 +# ───────────────────────────────────────────────────────────────────────────── +class TestStage2Hallucination: + def test_hedge_words_without_numbers_rejected(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, + distilled_text='我猜本週業績可能會有點成長吧，也許不錯。', + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is False + assert decision.reason == 'rejected_hallucination' + + def test_hedge_words_with_numbers_passes(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, weight=0.5, + distilled_text='我猜本週業績會漲 5.2%，根據過去 30 天平均。', + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + # 通過 Stage 2（有具體數字）→ 進到 Stage 4 → approved + assert decision.can_promote is True + + def test_contradiction_rejected(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, + distilled_text='A是黑色。A是白色。', + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is False + assert decision.reason == 'rejected_hallucination' + assert '自相矛盾' in (decision.detail or '') + + def test_clean_text_passes(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, weight=0.5, + distilled_text='本週業績漲 5.2%，建議聚焦保濕品類。', + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is True + assert decision.reason == 'approved' + + +# ───────────────────────────────────────────────────────────────────────────── +# Stage 3: 去重 +# ───────────────────────────────────────────────────────────────────────────── +class TestStage3Dedup: + def test_high_similarity_rejected(self, monkeypatch): + from services.learning_pipeline import PromotionGate, STAGE_3_DEDUP_THRESHOLD + ep = _fake_episode( + quality_score=0.8, weight=0.5, + distilled_text='本週業績漲 5%。', + embedding=[0.1] * 1024, # 模擬非空 embedding + ) + _patch_load_episode(monkeypatch, ep) + + # 模擬 DB 回 similarity=0.96 + fake_row = MagicMock() + fake_row.id = 999 + fake_row.similarity = 0.96 + fake_session = MagicMock() + fake_session.execute.return_value.fetchone.return_value = fake_row + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is False + assert decision.reason == 'rejected_duplicate' + assert decision.similar_insight_id == 999 + + def test_low_similarity_passes(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, weight=0.5, + distilled_text='全新內容', embedding=[0.1] * 1024, + ) + _patch_load_episode(monkeypatch, ep) + + fake_row = MagicMock() + fake_row.id = 999 + fake_row.similarity = 0.5 + fake_session = MagicMock() + fake_session.execute.return_value.fetchone.return_value = fake_row + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is True + + def test_null_embedding_skips_dedup(self, monkeypatch): + """蒸餾時尚未 embed → 略過 Stage 3，不阻擋晉升。""" + from services.learning_pipeline import PromotionGate + ep = _fake_episode(quality_score=0.8, weight=0.5, embedding=None) + _patch_load_episode(monkeypatch, ep) + + # DB 不應被呼叫 + called = {'count': 0} + + def _spy_session(): + called['count'] += 1 + return MagicMock() + + monkeypatch.setattr('database.manager.get_session', _spy_session) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is True + assert called['count'] == 0 + + def test_dedup_query_failure_passes(self, monkeypatch): + """DB 查詢失敗 → 視為通過（避免 DB 故障阻塞晉升）。""" + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + quality_score=0.8, weight=0.5, + embedding=[0.1] * 1024, + ) + _patch_load_episode(monkeypatch, ep) + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db down") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is True + + +# ───────────────────────────────────────────────────────────────────────────── +# Stage 4: 強制人工驗收（v5.0 護欄 #1 核心） +# ───────────────────────────────────────────────────────────────────────────── +class TestStage4HumanReview: + def test_high_weight_forces_awaiting_review(self, monkeypatch): + """weight=0.85 (>=0.8) 必經人工驗收，不能跳 Stage 4。""" + from services.learning_pipeline import PromotionGate, STAGE_4_HUMAN_REVIEW_WEIGHT + ep = _fake_episode(quality_score=0.9, weight=0.85) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is False + assert decision.reason == 'awaiting_review' + assert str(STAGE_4_HUMAN_REVIEW_WEIGHT) in (decision.detail or '') + + def test_high_weight_at_threshold_forces_review(self, monkeypatch): + """weight 剛好 0.8 也要進人工驗收（>= not >）。""" + from services.learning_pipeline import PromotionGate + ep = _fake_episode(quality_score=0.9, weight=0.8) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.reason == 'awaiting_review' + + def test_low_weight_auto_promoted(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode(quality_score=0.9, weight=0.79) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.can_promote is True + assert decision.reason == 'approved' + + def test_high_weight_user_feedback_forces_review(self, monkeypatch): + """user_feedback episode_type 預設 weight=0.9 → 必經人工。""" + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + episode_type='user_feedback', + quality_score=1.0, weight=0.9, + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + assert decision.reason == 'awaiting_review' + + +# ───────────────────────────────────────────────────────────────────────────── +# promote() DB 操作 +# ───────────────────────────────────────────────────────────────────────────── +class TestPromote: + def test_promote_inserts_ai_insights_and_updates_episode(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + id_=1, episode_type='llm_response', + quality_score=0.9, weight=0.5, + distilled_text='本週業績漲 5%', + ) + _patch_load_episode(monkeypatch, ep) + + # 模擬 INSERT RETURNING id = 555 + fake_row = MagicMock() + fake_row.__getitem__.return_value = 555 + fake_session = MagicMock() + fake_session.execute.return_value.fetchone.return_value = fake_row + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + insight_id = gate.promote(1) + assert insight_id == 555 + # 檢查 INSERT + UPDATE 各跑一次（execute 至少 2 次） + assert fake_session.execute.call_count >= 2 + fake_session.commit.assert_called_once() + + def test_promote_episode_not_found_returns_none(self, monkeypatch): + from services.learning_pipeline import PromotionGate + _patch_load_episode(monkeypatch, None) + gate = PromotionGate() + assert gate.promote(99999) is None + + def test_promote_db_failure_returns_none(self, monkeypatch): + from services.learning_pipeline import PromotionGate + ep = _fake_episode(quality_score=0.9, weight=0.5) + _patch_load_episode(monkeypatch, ep) + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db down") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + assert gate.promote(1) is None + + +# ───────────────────────────────────────────────────────────────────────────── +# reject() / mark_awaiting_review() +# ───────────────────────────────────────────────────────────────────────────── +class TestRejectAndMark: + def test_reject_valid_reason(self, monkeypatch): + from services.learning_pipeline import PromotionGate + + fake_session = MagicMock() + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + ok = gate.reject(1, 'rejected_quality', detail='quality 0.3 < 0.7') + assert ok is True + fake_session.commit.assert_called_once() + + def test_reject_invalid_reason_returns_false(self): + from services.learning_pipeline import PromotionGate + gate = PromotionGate() + assert gate.reject(1, 'invalid_reason') is False + + def test_reject_db_failure_returns_false(self, monkeypatch): + from services.learning_pipeline import PromotionGate + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db down") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + assert gate.reject(1, 'rejected_quality') is False + + def test_mark_awaiting_review_runs_update(self, monkeypatch): + from services.learning_pipeline import PromotionGate + + fake_session = MagicMock() + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + gate = PromotionGate() + ok = gate.mark_awaiting_review(1) + assert ok is True + fake_session.execute.assert_called_once() + fake_session.commit.assert_called_once() + + +# ───────────────────────────────────────────────────────────────────────────── +# 完整 4 階段流程串接（高 weight 必經人工） +# ───────────────────────────────────────────────────────────────────────────── +class TestEndToEndFlow: + def test_user_feedback_high_quality_must_await_review(self, monkeypatch): + """v5.0 護欄 #1 核心案例：user_feedback weight=0.9 強制 awaiting_review， + 即使 quality=1.0 + 無幻覺 + 無重複也不能直接晉升。 + """ + from services.learning_pipeline import PromotionGate + ep = _fake_episode( + episode_type='user_feedback', + quality_score=1.0, # Stage 1 過 + distilled_text='2026-04-29 業績漲 12%，廣告 ROI 4.2 倍。', # 有數字 → Stage 2 過 + embedding=None, # Stage 3 略過（無 embedding） + weight=0.9, # >= 0.8 → 強制 Stage 4 + ) + _patch_load_episode(monkeypatch, ep) + + gate = PromotionGate() + decision = gate.can_promote(1) + # 鐵律：高權重必經人工 + assert decision.can_promote is False + assert decision.reason == 'awaiting_review' diff --git a/tests/test_rag_service.py b/tests/test_rag_service.py new file mode 100644 index 0000000..416c2f4 --- /dev/null +++ b/tests/test_rag_service.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +tests/test_rag_service.py +Operation Ollama-First v5.0 / Phase 11 — RAGService 單元測試 + +涵蓋： + - feature flag RAG_ENABLED 預設 OFF → skip 不查 DB / 不寫 log + - RAG_ENABLED=true 時的 hit / miss 分支 + - embedding signature 不一致 → log warning + 不採該筆 + - fire-and-forget rag_query_log 失敗不影響主流程 + - feedback() / invalidate_by_caller() / get_embedding_signature() + +Mock 策略： + - ollama_service.generate_embedding 用 monkeypatch + - database.manager.get_session 用 MagicMock 回 session（不真的查 DB） +""" + +from __future__ import annotations + +import os +import time +from typing import Any, Dict, List +from unittest.mock import MagicMock, patch + +import pytest + + +# ───────────────────────────────────────────────────────────────────────────── +# 共用 fixture +# ───────────────────────────────────────────────────────────────────────────── +@pytest.fixture(autouse=True) +def _reset_rag_kill_switch(): + """每個測試重置 kill-switch，避免互相污染。""" + from services.rag_service import _reset_kill_switch + _reset_kill_switch() + yield + _reset_kill_switch() + + +@pytest.fixture +def rag_disabled(monkeypatch): + monkeypatch.setenv('RAG_ENABLED', 'false') + + +@pytest.fixture +def rag_enabled(monkeypatch): + monkeypatch.setenv('RAG_ENABLED', 'true') + + +def _fake_embedding(dim: int = 1024) -> List[float]: + """產生穩定的假 embedding（避免每次 random）。""" + return [0.01 * (i % 100) for i in range(dim)] + + +def _make_row_obj(**fields): + """模擬 SQLAlchemy row（支援屬性與 row.id 風格）。""" + obj = MagicMock() + for k, v in fields.items(): + setattr(obj, k, v) + return obj + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 1: feature flag 預設 OFF → 短路 +# ───────────────────────────────────────────────────────────────────────────── +class TestFeatureFlagOff: + def test_query_skips_when_disabled(self, rag_disabled): + """RAG_ENABLED=false → 不查 DB / 不 embed / 回空 hits。""" + from services.rag_service import rag_service + + with patch('services.ollama_service.ollama_service.generate_embedding') as mock_embed: + result = rag_service.query("本週業績", caller='test_caller') + + assert result.hits == [] + assert result.has_high_confidence is False + mock_embed.assert_not_called() + + def test_synthesize_returns_empty_when_no_hits(self, rag_disabled): + from services.rag_service import rag_service + + result = rag_service.query("本週業績", caller='test_caller') + assert result.synthesize() == "" + + def test_default_flag_is_off(self, monkeypatch): + """RAG_ENABLED 未設 → 預設 OFF。""" + monkeypatch.delenv('RAG_ENABLED', raising=False) + from services.rag_service import is_rag_enabled + assert is_rag_enabled() is False + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 2: RAG_ENABLED=true 時的 hit / miss +# ───────────────────────────────────────────────────────────────────────────── +class TestRagEnabledHits: + def test_high_confidence_hit_returns_synthesized(self, rag_enabled, monkeypatch): + """top-1 distance=0.05 → similarity=0.95 >= threshold 0.85 → has_high_confidence。""" + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + + sig = rs.get_embedding_signature() + fake_rows = [ + _make_row_obj( + id=101, insight_type='weekly_strategy', period='2026-W17', + content='本週業績漲 5%，建議聚焦保濕品類。', + embedding_signature=sig, distance=0.05, + ), + _make_row_obj( + id=102, insight_type='weekly_strategy', period='2026-W17', + content='競品 PChome 同類降價 8%。', + embedding_signature=sig, distance=0.10, + ), + ] + fake_session = MagicMock() + fake_session.execute.return_value.fetchall.return_value = fake_rows + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + result = rs.rag_service.query("本週業績趨勢", caller='openclaw_qa') + + assert len(result.hits) == 2 + assert result.hits[0]['score'] == pytest.approx(0.95, abs=1e-3) + assert result.has_high_confidence is True + assert '本週業績漲' in result.synthesize() + + def test_low_confidence_no_hit(self, rag_enabled, monkeypatch): + """所有結果 distance>0.15 → similarity<0.85 → SQL 已過濾，回空 → has_high_confidence False。""" + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + fake_session = MagicMock() + fake_session.execute.return_value.fetchall.return_value = [] + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + result = rs.rag_service.query("非常冷僻的問題", caller='openclaw_qa') + + assert result.hits == [] + assert result.has_high_confidence is False + assert result.synthesize() == "" + + def test_empty_query_short_circuits(self, rag_enabled, monkeypatch): + from services import rag_service as rs + + # generate_embedding 不應被呼叫 + embed_called = {'count': 0} + + def _spy(*a, **kw): + embed_called['count'] += 1 + return _fake_embedding() + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', _spy, + ) + + result = rs.rag_service.query("", caller='openclaw_qa') + assert result.hits == [] + assert embed_called['count'] == 0 + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 3: embedding signature 不一致 → 不採用 +# ───────────────────────────────────────────────────────────────────────────── +class TestEmbeddingSignature: + def test_signature_format(self): + from services.rag_service import get_embedding_signature + sig = get_embedding_signature() + # 12 碼 hex + assert len(sig) == 12 + assert all(c in '0123456789abcdef' for c in sig) + + def test_signature_changes_with_model(self): + from services.rag_service import get_embedding_signature + a = get_embedding_signature(model='bge-m3:latest', dim=1024, normalize=True) + b = get_embedding_signature(model='bge-m3:v2', dim=1024, normalize=True) + c = get_embedding_signature(model='bge-m3:latest', dim=512, normalize=True) + d = get_embedding_signature(model='bge-m3:latest', dim=1024, normalize=False) + assert a != b and a != c and a != d + + def test_mismatched_signature_rows_skipped(self, rag_enabled, monkeypatch, caplog): + """row.embedding_signature != expected → 不採用 + log warning。""" + from services import rag_service as rs + import logging + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + sig = rs.get_embedding_signature() + fake_rows = [ + _make_row_obj( + id=201, insight_type='x', period=None, content='舊簽名資料', + embedding_signature='deadbeef0001', # 不同 + distance=0.05, + ), + _make_row_obj( + id=202, insight_type='x', period=None, content='新簽名資料', + embedding_signature=sig, + distance=0.06, + ), + ] + fake_session = MagicMock() + fake_session.execute.return_value.fetchall.return_value = fake_rows + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + with caplog.at_level(logging.WARNING, logger='services.rag_service'): + result = rs.rag_service.query("query", caller='openclaw_qa') + + # 只剩 1 筆（簽名一致的） + assert len(result.hits) == 1 + assert result.hits[0]['id'] == 202 + # warning 有寫 + assert any('signature mismatch' in r.message for r in caplog.records) + + def test_null_signature_passes_through(self, rag_enabled, monkeypatch): + """既有未回填的 row（signature=NULL）暫時放行避免戰前資料完全失效。""" + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + fake_rows = [ + _make_row_obj( + id=301, insight_type='x', period=None, content='legacy data', + embedding_signature=None, + distance=0.1, + ), + ] + fake_session = MagicMock() + fake_session.execute.return_value.fetchall.return_value = fake_rows + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + result = rs.rag_service.query("query", caller='openclaw_qa') + assert len(result.hits) == 1 + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 4: fire-and-forget log 失敗不影響主流程 +# ───────────────────────────────────────────────────────────────────────────── +class TestFireAndForgetLog: + def test_log_write_failure_does_not_crash(self, rag_enabled, monkeypatch): + """rag_query_log INSERT 失敗 → main 仍回 RAGResult。""" + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + # SELECT session 正常；INSERT session 故意 raise + select_session = MagicMock() + select_session.execute.return_value.fetchall.return_value = [] + insert_session = MagicMock() + insert_session.execute.side_effect = RuntimeError("DB unavailable") + + sessions_iter = iter([select_session, insert_session]) + monkeypatch.setattr( + 'database.manager.get_session', lambda: next(sessions_iter), + ) + + # 應不 raise（fire-and-forget thread 內部包 try/except） + result = rs.rag_service.query("query", caller='openclaw_qa') + assert result.hits == [] + # 等 daemon thread 跑完 + time.sleep(0.2) + + def test_embedding_failure_falls_back_to_empty(self, rag_enabled, monkeypatch): + """embedding 回 [] → 不查 DB → 回空 hits 給 caller fallback LLM。""" + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": [], + ) + # DB 不應被呼叫 + called = {'count': 0} + + def _spy_session(): + called['count'] += 1 + return MagicMock() + + monkeypatch.setattr('database.manager.get_session', _spy_session) + + result = rs.rag_service.query("query", caller='openclaw_qa') + assert result.hits == [] + # SELECT 不應發生（log 寫入仍會用 session） + # rag_query_log 寫入也許會跑（embedding=None 仍 log），所以 called 可能是 0 或 1 + time.sleep(0.2) + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 5: feedback() / invalidate_by_caller() +# ───────────────────────────────────────────────────────────────────────────── +class TestFeedback: + def test_feedback_writes_score(self, monkeypatch): + from services.rag_service import rag_service + + fake_session = MagicMock() + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + ok = rag_service.feedback(123, 5) + assert ok is True + fake_session.execute.assert_called_once() + fake_session.commit.assert_called_once() + + def test_feedback_clamps_score(self, monkeypatch): + """score=99 應被夾擠到 5；score=0 應被夾擠到 1。""" + from services.rag_service import rag_service + + captured = {} + fake_session = MagicMock() + + def _exec(stmt, params): + captured.update(params) + return MagicMock() + + fake_session.execute.side_effect = _exec + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + rag_service.feedback(1, 99) + assert captured['score'] == 5 + + captured.clear() + rag_service.feedback(1, 0) + assert captured['score'] == 1 + + def test_feedback_invalid_id_returns_false(self): + from services.rag_service import rag_service + assert rag_service.feedback(0, 5) is False + assert rag_service.feedback(None, 5) is False # type: ignore + + def test_feedback_db_failure_returns_false(self, monkeypatch): + from services.rag_service import rag_service + + fake_session = MagicMock() + fake_session.execute.side_effect = RuntimeError("db fail") + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + ok = rag_service.feedback(123, 5) + assert ok is False + + def test_invalidate_by_caller_is_noop(self): + """v5.0 暫無 cache 層 → no-op，不 raise。""" + from services.rag_service import rag_service + rag_service.invalidate_by_caller('openclaw_qa') # 不應 raise + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 6: RAGResult 屬性與 synthesize +# ───────────────────────────────────────────────────────────────────────────── +class TestRAGResultStructure: + def test_synthesize_takes_top_3(self): + from services.rag_service import RAGResult + result = RAGResult( + query='q', embedding_signature='abc', + hits=[{'content': f'第 {i} 筆內容'} for i in range(5)], + ) + out = result.synthesize() + assert '第 0 筆' in out + assert '第 1 筆' in out + assert '第 2 筆' in out + assert '第 3 筆' not in out + assert '第 4 筆' not in out + + def test_has_high_confidence_threshold(self): + from services.rag_service import RAGResult + # top-1 score=0.86 >= 0.85 → True + r1 = RAGResult( + query='q', embedding_signature='abc', + hits=[{'score': 0.86}], threshold=0.85, + ) + assert r1.has_high_confidence is True + + # top-1 score=0.84 < 0.85 → False + r2 = RAGResult( + query='q', embedding_signature='abc', + hits=[{'score': 0.84}], threshold=0.85, + ) + assert r2.has_high_confidence is False + + # 無 hits → False + r3 = RAGResult(query='q', embedding_signature='abc', hits=[]) + assert r3.has_high_confidence is False + + +# ───────────────────────────────────────────────────────────────────────────── +# Test 7: top_k / threshold 護欄夾擠 +# ───────────────────────────────────────────────────────────────────────────── +class TestParamGuards: + def test_top_k_clamped_to_50(self, rag_enabled, monkeypatch): + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + fake_session = MagicMock() + fake_session.execute.return_value.fetchall.return_value = [] + captured = {} + + def _exec(stmt, params): + captured.update(params) + return MagicMock(fetchall=lambda: []) + + fake_session.execute.side_effect = _exec + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + rs.rag_service.query("query", caller='openclaw_qa', top_k=999) + assert captured.get('lim', 0) <= 100 # top_k=50, fetch_limit = 100 + + def test_threshold_clamped_to_unit_interval(self, rag_enabled, monkeypatch): + from services import rag_service as rs + + monkeypatch.setattr( + 'services.ollama_service.ollama_service.generate_embedding', + lambda text, model="bge-m3:latest": _fake_embedding(), + ) + fake_session = MagicMock() + captured = {} + + def _exec(stmt, params): + captured.update(params) + return MagicMock(fetchall=lambda: []) + + fake_session.execute.side_effect = _exec + monkeypatch.setattr('database.manager.get_session', lambda: fake_session) + + rs.rag_service.query("query", caller='openclaw_qa', threshold=2.0) + # 2.0 → clamp 1.0；max_distance = 1.0 - 1.0 = 0.0 + assert captured.get('max_distance', 1.0) == pytest.approx(0.0)