Spaces:
Sleeping
Sleeping
File size: 2,515 Bytes
aa9003d a72fec7 aa9003d 6d2a17c aa9003d 8f17704 aa9003d 7196ae9 aa9003d 3855268 aa9003d 3855268 aa9003d 3855268 aa9003d 3855268 aa9003d 3855268 aa9003d |
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 |
# ────────────────────────────── memo/context.py ──────────────────────────────
"""
Context Management
Functions for retrieving and managing conversation context.
"""
import numpy as np
from typing import List, Dict, Any, Tuple, Optional
import os
from utils.logger import get_logger
from utils.rag.embeddings import EmbeddingClient
logger = get_logger("CONTEXT_MANAGER", __name__)
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
"""Calculate cosine similarity between two vectors"""
denom = (np.linalg.norm(a) * np.linalg.norm(b)) or 1.0
return float(np.dot(a, b) / denom)
async def semantic_context(question: str, memories: List[str], embedder: EmbeddingClient, topk: int = 3) -> str:
"""
Get semantic context from memories using cosine similarity.
"""
if not memories:
return ""
try:
qv = np.array(embedder.embed([question])[0], dtype="float32")
mats = embedder.embed([s.strip() for s in memories])
sims = [(cosine_similarity(qv, np.array(v, dtype="float32")), s) for v, s in zip(mats, memories)]
sims.sort(key=lambda x: x[0], reverse=True)
top = [s for (sc, s) in sims[:topk] if sc > 0.15] # small threshold
return "\n\n".join(top) if top else ""
except Exception as e:
logger.error(f"[CONTEXT_MANAGER] Semantic context failed: {e}")
return ""
# get_conversation_context function removed - use memory_system.get_conversation_context() instead
async def get_legacy_context(user_id: str, question: str, memory_system,
embedder: EmbeddingClient, topk_sem: int) -> 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)
# Use semantic similarity to select most relevant recent memories
recent_text = ""
if recent3:
try:
recent_text = await semantic_context(question, recent3, embedder, 2)
except Exception as e:
logger.warning(f"[CONTEXT_MANAGER] Recent context selection failed: {e}")
# Get semantic context from remaining memories
sem_text = ""
if rest17:
sem_text = await semantic_context(question, rest17, embedder, topk_sem)
return recent_text, sem_text
|