Spaces:
Sleeping
Sleeping
File size: 5,326 Bytes
aa9003d 729a1f7 aa9003d 8f17704 6d2a17c 729a1f7 aa9003d 729a1f7 aa9003d 729a1f7 aa9003d 729a1f7 aa9003d 8f17704 aa9003d 8f17704 aa9003d 3855268 8f17704 3855268 8f17704 3855268 8f17704 3855268 729a1f7 aa9003d 729a1f7 f80e242 aa9003d 6f12b05 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
# ────────────────────────────── memo/history.py ──────────────────────────────
"""
History Management
Functions for managing conversation history and context.
"""
from typing import List, Dict, Any, Tuple, Optional
from utils.logger import get_logger
from memo.nvidia import summarize_qa, files_relevance, related_recent_context
from memo.context import semantic_context, get_legacy_context
from utils.rag.embeddings import EmbeddingClient
logger = get_logger("HISTORY_MANAGER", __name__)
class HistoryManager:
"""
Enhanced history manager that provides both legacy and enhanced functionality.
Automatically uses enhanced features when available.
"""
def __init__(self, memory_system=None):
self.memory_system = memory_system
async def summarize_qa_with_nvidia(self, question: str, answer: str, nvidia_rotator) -> str:
"""Summarize Q&A using NVIDIA model (enhanced version)"""
return await summarize_qa(question, answer, nvidia_rotator)
async def files_relevance(self, question: str, file_summaries: List[Dict[str, str]], nvidia_rotator) -> Dict[str, bool]:
"""Determine file relevance using NVIDIA model (enhanced version)"""
return await files_relevance(question, file_summaries, nvidia_rotator)
async def related_recent_and_semantic_context(self, user_id: str, question: str,
embedder: EmbeddingClient,
topk_sem: int = 3, nvidia_rotator=None) -> Tuple[str, str]:
"""Get related recent and semantic context (enhanced version)"""
try:
if self.memory_system and self.memory_system.is_enhanced_available():
# Use enhanced context retrieval
recent_context, semantic_context = await self.memory_system.get_conversation_context(
user_id, question
)
return recent_context, semantic_context
else:
# Fallback to original implementation with NVIDIA support
return await self._get_legacy_context(user_id, question, self.memory_system, embedder, topk_sem, nvidia_rotator)
except Exception as e:
logger.error(f"[HISTORY_MANAGER] Context retrieval failed: {e}")
return "", ""
async def _get_legacy_context(self, user_id: str, question: str, memory_system,
embedder: EmbeddingClient, topk_sem: int, nvidia_rotator=None) -> Tuple[str, str]:
"""Get context using legacy method with enhanced semantic selection"""
if not memory_system:
return "", ""
recent3 = memory_system.recent(user_id, 3)
rest17 = memory_system.rest(user_id, 3)
recent_text = ""
if recent3:
# Use NVIDIA to select most relevant recent memories (enhanced)
if nvidia_rotator:
try:
recent_text = await related_recent_context(question, recent3, nvidia_rotator)
except Exception as e:
logger.warning(f"[HISTORY_MANAGER] NVIDIA recent context selection failed: {e}")
# Fallback to semantic similarity
try:
recent_text = await semantic_context(question, recent3, embedder, 2)
except Exception as e2:
logger.warning(f"[HISTORY_MANAGER] Semantic fallback failed: {e2}")
else:
# Use semantic similarity directly if no NVIDIA rotator
try:
recent_text = await semantic_context(question, recent3, embedder, 2)
except Exception as e:
logger.warning(f"[HISTORY_MANAGER] Semantic recent context failed: {e}")
sem_text = ""
if rest17:
sem_text = await semantic_context(question, rest17, embedder, topk_sem)
return recent_text, sem_text
# ────────────────────────────── Legacy Functions (Backward Compatibility) ──────────────────────────────
# These legacy functions have been removed as they are no longer needed with the new memory planning system.
# Use the appropriate methods from the core memory system or memory planner instead.
# ────────────────────────────── Global Instance ──────────────────────────────
_history_manager: Optional[HistoryManager] = None
def get_history_manager(memory_system=None) -> HistoryManager:
"""Get the global history manager instance"""
global _history_manager
if _history_manager is None:
_history_manager = HistoryManager(memory_system)
logger.info("[HISTORY_MANAGER] Global history manager initialized")
return _history_manager
# def reset_history_manager():
# """Reset the global history manager (for testing)"""
# global _history_manager
# _history_manager = None |