core/memory_system.py

import json
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime
import asyncio
import random

# 統一日誌配置
from core.logging import get_logger
logger = get_logger("MemorySystem")

# 統一 OpenAI 客戶端
from core.ai_client import get_openai_client
from core.config import settings
from core.responses_runtime import ResponsesAgentRuntime

def _get_memory_client():
    """取得記憶系統用的 OpenAI 客戶端"""
    return get_openai_client()


TRANSIENT_MEMORY_ERROR_MARKERS = (
    "502",
    "503",
    "504",
    "bad gateway",
    "upstream",
    "timeout",
    "timed out",
    "connection",
)

TRANSIENT_QUERY_MARKERS = (
    "今天",
    "現在",
    "目前",
    "最新",
    "即時",
    "收盤",
    "開盤",
    "股價",
    "股票",
    "匯率",
    "天氣",
    "新聞",
    "多少",
    "查詢",
    "search",
    "latest",
    "today",
    "now",
    "price",
    "stock",
    "weather",
    "news",
)

MEMORY_ANALYSIS_SCHEMA = {
    "type": "object",
    "properties": {
        "memories": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "type": {
                        "type": "string",
                        "enum": ["personal_info", "preferences", "goals"],
                    },
                    "content": {"type": "string"},
                    "importance": {
                        "type": "number",
                        "minimum": 0,
                        "maximum": 1,
                    },
                },
                "required": ["type", "content", "importance"],
                "additionalProperties": False,
            },
        }
    },
    "required": ["memories"],
    "additionalProperties": False,
}


def _is_transient_memory_error(exc: Exception) -> bool:
    if isinstance(exc, TimeoutError) or isinstance(exc, asyncio.TimeoutError):
        return True
    error_text = str(exc).lower()
    return any(marker in error_text for marker in TRANSIENT_MEMORY_ERROR_MARKERS)


def _should_run_ai_memory_analysis(user_message: str, assistant_response: str = "") -> bool:
    text = f"{user_message}\n{assistant_response}".strip().lower()
    if not text:
        return False

    durable_markers = (
        "我叫",
        "我的名字",
        "我喜歡",
        "我不喜歡",
        "我討厭",
        "我的偏好",
        "我住在",
        "我的工作",
        "我的目標",
        "我想達成",
        "我希望",
        "請記得",
        "記住",
        "下次",
        "my name",
        "i like",
        "i dislike",
        "remember",
        "my goal",
    )
    if any(marker in text for marker in durable_markers):
        return True

    user_text = (user_message or "").lower()
    if any(marker in user_text for marker in TRANSIENT_QUERY_MARKERS):
        return False

    return len(user_message.strip()) >= 80



# 導入數據庫函數
try:
    from .database import save_memory, get_user_memories, search_memories
    db_available = True
except ImportError:
    db_available = False
    logger.warning("無法導入記憶數據庫函數")


class MemoryExtractor:
    """記憶提取器：從對話中提取重要信息"""

    def __init__(self):
        self.memory_types = {
            "personal_info": {
                "keywords": ["我叫", "我的名字", "我今年", "我的年齡", "我是", "我從事", "我的工作", "我住在", "我的地址", "我的電話", "我的email", "我的興趣"],
                "description": "個人基本信息",
                "importance_base": 0.9
            },
            "preferences": {
                "keywords": ["我喜歡", "我不喜歡", "我的偏好", "我討厭", "我想要", "我需要", "我習慣", "我通常"],
                "description": "個人偏好和習慣",
                "importance_base": 0.8
            },
            "events": {
                "keywords": ["我有個約", "我約了", "我預計", "我計劃", "我會", "我要", "記得", "提醒我", "下次", "明天", "後天", "下週"],
                "description": "重要事件和約定",
                "importance_base": 0.8
            },
            "knowledge": {
                "keywords": ["我知道", "我學到", "我發現", "我了解", "我學會", "經驗", "教訓", "總結"],
                "description": "學習到的知識和經驗",
                "importance_base": 0.7
            },
            "goals": {
                "keywords": ["我的目標", "我想達成", "我希望", "我的夢想", "我的計劃", "長期目標", "短期目標"],
                "description": "長期和短期目標",
                "importance_base": 0.8
            }
        }

    def extract_memories(self, user_message: str, assistant_response: str = "") -> List[Dict[str, Any]]:
        """從用戶消息和助手回應中提取記憶"""
        memories = []

        # 合併用戶消息和助手回應進行分析
        full_text = f"用戶: {user_message}\n助手: {assistant_response}"

        for memory_type, config in self.memory_types.items():
            # 檢查關鍵字匹配
            matched_keywords = []
            for keyword in config["keywords"]:
                if keyword in user_message.lower():
                    matched_keywords.append(keyword)

            if matched_keywords:
                # 提取相關內容
                extracted_content = self._extract_content(user_message, matched_keywords)
                if extracted_content:
                    importance = self._calculate_importance(extracted_content, config["importance_base"])

                    memory = {
                        "type": memory_type,
                        "content": extracted_content,
                        "importance": importance,
                        "trigger_keywords": matched_keywords,
                        "source": "keyword_extraction",
                        "metadata": {
                            "extracted_at": datetime.now().isoformat(),
                            "confidence": len(matched_keywords) / len(config["keywords"])
                        }
                    }
                    memories.append(memory)

        return memories

    def _extract_content(self, text: str, keywords: List[str]) -> Optional[str]:
        """從文本中提取記憶內容"""
        # 簡單的內容提取邏輯
        sentences = text.split('。')
        relevant_sentences = []

        for sentence in sentences:
            if any(keyword in sentence for keyword in keywords):
                relevant_sentences.append(sentence.strip())

        if relevant_sentences:
            return '。'.join(relevant_sentences) + '。'

        return None

    def _calculate_importance(self, content: str, base_importance: float) -> float:
        """計算記憶的重要性分數"""
        # 基於內容長度和關鍵字密度調整重要性
        content_length = len(content)
        if content_length < 10:
            return base_importance * 0.5
        elif content_length > 100:
            return min(base_importance * 1.2, 1.0)
        else:
            return base_importance


class MemoryAnalyzer:
    """記憶分析器：使用AI分析對話內容"""

    def __init__(self):
        self.responses_runtime = ResponsesAgentRuntime()

    @staticmethod
    def _memory_model() -> str:
        return settings.OPENAI_MODEL or settings.GPT_INTENT_MODEL

    @staticmethod
    def _memory_timeout() -> float:
        return min(float(getattr(settings, "OPENAI_TIMEOUT", 30)), 20.0)

    @staticmethod
    async def _transient_backoff(attempt: int) -> None:
        delay = min(0.5 * (2 ** attempt), 3.0) + random.uniform(0, 0.2)
        await asyncio.sleep(delay)

    def _create_analysis_response(self, client: Any, messages: List[Dict[str, str]], max_tokens_value: int) -> Any:
        model = self._memory_model()
        if settings.OPENAI_USE_RESPONSES and model.startswith("gpt-5"):
            payload = self.responses_runtime.build_payload_from_messages(
                messages=messages,
                model=model,
                max_output_tokens=max_tokens_value,
                text_format={
                    "type": "json_schema",
                    "name": "memory_analysis",
                    "strict": True,
                    "schema": MEMORY_ANALYSIS_SCHEMA,
                },
            )
            payload["store"] = False
            return client.responses.create(**payload)

        return client.chat.completions.create(
            model=model,
            messages=messages,
            max_completion_tokens=max_tokens_value,
            response_format={"type": "json_object"},
        )

    async def analyze_conversation(self, user_message: str, assistant_response: str = "",
                                  conversation_history: List[Dict] = None) -> List[Dict[str, Any]]:
        """使用AI分析對話內容，提取重要記憶"""
        client = _get_memory_client()
        if not client:
            logger.warning("OpenAI客戶端不可用，跳過AI記憶分析")
            return []

        try:
            # 構建簡潔的分析提示
            system_prompt = """你是記憶分析助手。從用戶對話中提取重要資訊。

規則：
1. 只提取重要且持久的資訊
2. 避免記住無關緊要的內容
3. 區分類型：personal_info（個人信息）、preferences（偏好）、goals（目標）

返回JSON格式：
{
  "memories": [
    {
      "type": "personal_info|preferences|goals",
      "content": "具體內容",
      "importance": 0.8
    }
  ]
}

如果沒有重要資訊，返回空列表。"""

            # 準備對話歷史（最近幾條）
            recent_history = ""
            if conversation_history:
                recent_messages = conversation_history[-6:]  # 最近3輪對話
                for msg in recent_messages:
                    role = "用戶" if msg.get("role") == "user" else "助手"
                    recent_history += f"{role}: {msg.get('content', '')}\n"

            user_prompt = f"""分析對話，提取重要記憶：

用戶: {user_message}
助手: {assistant_response}

提取任何重要資訊。"""

            messages = [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ]

            max_attempts = 3
            for attempt in range(max_attempts):
                try:
                    max_tokens_value = 500

                    response = await asyncio.wait_for(
                        asyncio.to_thread(self._create_analysis_response, client, messages, max_tokens_value),
                        timeout=self._memory_timeout(),
                    )
                    break  # 成功後跳出重試循環

                except Exception as api_error:
                    error_str = str(api_error).lower()
                    if _is_transient_memory_error(api_error):
                        if attempt < max_attempts - 1:
                            logger.warning(
                                "AI記憶分析遇到暫時性上游錯誤，準備重試 (%s/%s): %s",
                                attempt + 1,
                                max_attempts,
                                api_error,
                            )
                            await self._transient_backoff(attempt)
                            continue
                        logger.error("AI記憶分析連續暫時性上游錯誤，已放棄本輪背景分析: %s", api_error)
                        return []
                    if "max_tokens" in error_str or "token limit" in error_str:
                        logger.error(f"AI分析遇到token限制錯誤: {api_error}")
                        return []  # 返回空列表，回退到關鍵字提取
                    else:
                        # 其他類型的錯誤，直接拋出
                        raise api_error

            if hasattr(response, "choices"):
                result_text = response.choices[0].message.content.strip()
            else:
                result_text = self.responses_runtime.extract_output_text(response)

            # 解析JSON結果 - 嘗試多種解析方式
            try:
                # 首先嘗試直接解析
                result = json.loads(result_text)
                memories = result.get("memories", [])
            except json.JSONDecodeError:
                # 如果直接解析失敗，嘗試提取JSON部分
                import re
                json_match = re.search(r'\{.*\}', result_text, re.DOTALL)
                if json_match:
                    try:
                        result = json.loads(json_match.group())
                        memories = result.get("memories", [])
                    except json.JSONDecodeError:
                        logger.warning(f"提取的JSON仍然無效: {json_match.group()[:200]}...")
                        memories = []
                else:
                    logger.warning(f"無法在AI響應中找到JSON: {result_text[:200]}...")
                    memories = []

            # 添加元數據
            for memory in memories:
                if isinstance(memory, dict):
                    memory["source"] = "ai_analysis"
                    memory["metadata"] = {
                        "analyzed_at": datetime.now().isoformat(),
                        "conversation_context": user_message[:100] + "..." if len(user_message) > 100 else user_message
                    }

            logger.info(f"AI分析提取到 {len(memories)} 條記憶")
            return memories

        except Exception as e:
            if _is_transient_memory_error(e):
                logger.info("AI記憶分析遇到暫時性上游錯誤，跳過本輪背景分析: %s", e)
                return []
            logger.error(f"AI記憶分析時發生錯誤: {e}")
            return []


class MemoryManager:
    """記憶管理器：統籌記憶的提取、存儲和檢索"""

    def __init__(self):
        self.extractor = MemoryExtractor()
        self.analyzer = MemoryAnalyzer()

    async def process_conversation(self, user_id: str, user_message: str,
                                 assistant_response: str = "",
                                 conversation_history: List[Dict] = None) -> Dict[str, Any]:
        """處理對話，提取並存儲記憶"""
        result = {
            "extracted_memories": 0,
            "saved_memories": 0,
            "errors": []
        }

        try:
            # 1. 使用關鍵字提取記憶
            keyword_memories = self.extractor.extract_memories(user_message, assistant_response)

            # 2. 使用AI分析提取記憶（如果可用）
            ai_memories = []
            if _get_memory_client() and _should_run_ai_memory_analysis(user_message, assistant_response):
                ai_memories = await self.analyzer.analyze_conversation(
                    user_message, assistant_response, conversation_history
                )
            else:
                logger.debug("跳過AI記憶分析：本輪內容不具備長期記憶價值或客戶端不可用")

            # 3. 合併記憶（去重）
            all_memories = self._merge_memories(keyword_memories, ai_memories)
            result["extracted_memories"] = len(all_memories)

            # 4. 存儲記憶
            if db_available and all_memories:
                saved_count = 0
                for memory in all_memories:
                    try:
                        save_result = await save_memory(
                            user_id=user_id,
                            memory_type=memory["type"],
                            content=memory["content"],
                            importance=memory["importance"],
                            metadata=memory.get("metadata", {})
                        )
                        if save_result["success"]:
                            saved_count += 1
                    except Exception as e:
                        logger.error(f"保存記憶失敗: {e}")
                        result["errors"].append(str(e))

                result["saved_memories"] = saved_count

            logger.info(f"處理用戶 {user_id} 的對話，提取 {len(all_memories)} 條記憶，保存 {result['saved_memories']} 條")

        except Exception as e:
            logger.error(f"處理對話記憶時發生錯誤: {e}")
            result["errors"].append(str(e))

        return result

    def _merge_memories(self, keyword_memories: List[Dict], ai_memories: List[Dict]) -> List[Dict]:
        """合併關鍵字和AI提取的記憶，去除重複"""
        merged = []

        # 優先使用AI記憶（更準確），只有在沒有AI記憶時才使用關鍵字記憶
        if ai_memories:
            # 如果有AI記憶，使用AI記憶
            merged.extend(ai_memories)
            logger.debug(f"使用AI分析記憶，共 {len(ai_memories)} 條")
        elif keyword_memories:
            # 如果沒有AI記憶但有關鍵字記憶，使用關鍵字記憶
            merged.extend(keyword_memories)
            logger.debug(f"使用關鍵字提取記憶，共 {len(keyword_memories)} 條")

        return merged

    def _is_similar_memory(self, memory1: Dict, memory2: Dict) -> bool:
        """檢查兩個記憶是否相似"""
        # 簡單的相似度檢查：類型相同且內容相似
        if memory1["type"] != memory2["type"]:
            return False

        content1 = memory1["content"].lower()
        content2 = memory2["content"].lower()

        # 計算相似度（簡單的Jaccard相似度）
        words1 = set(content1.split())
        words2 = set(content2.split())

        if not words1 or not words2:
            return False

        intersection = len(words1.intersection(words2))
        union = len(words1.union(words2))

        similarity = intersection / union if union > 0 else 0

        return similarity > 0.6  # 相似度大於60%視為重複

    async def get_relevant_memories(
        self,
        user_id: str,
        current_message: str,
        max_memories: int = 5,
        context_tags: Optional[List[str]] = None,
    ) -> List[Dict[str, Any]]:
        """獲取與當前消息相關的記憶"""
        if not db_available:
            return []

        try:
            collected: List[Dict[str, Any]] = []
            seen_ids: set[str] = set()

            async def _consume_query(query: str) -> None:
                if not query:
                    return
                result = await search_memories(user_id, query, limit=max_memories)
                if result.get("success"):
                    for mem in result.get("memories", []):
                        mem_id = mem.get("memory_id") or mem.get("id")
                        if mem_id and mem_id not in seen_ids:
                            collected.append(mem)
                            seen_ids.add(mem_id)
                        if len(collected) >= max_memories:
                            return

            await _consume_query(current_message)
            if len(collected) < max_memories and context_tags:
                for tag in context_tags:
                    await _consume_query(tag)
                    if len(collected) >= max_memories:
                        break

            if collected:
                return collected[:max_memories]

            general_result = await get_user_memories(
                user_id=user_id,
                limit=max_memories,
                min_importance=0.6,
            )

            if general_result["success"]:
                return general_result["memories"]

        except Exception as e:
            logger.error(f"獲取相關記憶時發生錯誤: {e}")

        return []

    def format_memories_for_context(self, memories: List[Dict[str, Any]]) -> str:
        """將記憶格式化為上下文字符串"""
        if not memories:
            return ""

        context_parts = []
        for memory in memories:
            memory_type = memory.get("type", "general")
            content = memory.get("content", "")
            importance = memory.get("importance", 0.5)

            # 只包含重要性較高的記憶
            if importance >= 0.6:
                type_labels = {
                    "personal_info": "個人信息",
                    "preferences": "偏好",
                    "events": "事件",
                    "knowledge": "知識",
                    "goals": "目標"
                }

                type_label = type_labels.get(memory_type, memory_type)
                context_parts.append(f"[{type_label}] {content}")

        if context_parts:
            return "\n".join(context_parts)
        else:
            return ""


# 全局記憶管理器實例
memory_manager = MemoryManager()

# 向後兼容別名
memory_system = memory_manager