Spaces:
Sleeping
Sleeping
File size: 7,335 Bytes
fd22f8e 7196ae9 a72fec7 7196ae9 |
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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
# ────────────────────────────── memo/plan/strategy.py ──────────────────────────────
"""
Strategy Planning
Handles memory strategy planning and parameter optimization.
"""
from typing import List, Dict, Any, Tuple, Optional
from enum import Enum
import os
from utils.logger import get_logger
from memo.plan.intent import QueryIntent
logger = get_logger("STRATEGY_PLANNER", __name__)
class MemoryStrategy(Enum):
"""Memory retrieval strategies"""
FOCUSED_QA = "focused_qa" # Focus on past Q&A pairs
BROAD_CONTEXT = "broad_context" # Use broad semantic context
RECENT_FOCUS = "recent_focus" # Focus on recent memories
SEMANTIC_DEEP = "semantic_deep" # Deep semantic search
MIXED_APPROACH = "mixed_approach" # Combine multiple strategies
class StrategyPlanner:
"""Handles memory strategy planning and parameter optimization"""
def __init__(self):
pass
def determine_strategy(self, intent: QueryIntent, question: str,
conversation_context: Dict[str, Any]) -> MemoryStrategy:
"""Determine the optimal memory retrieval strategy"""
try:
# Enhancement requests need focused Q&A retrieval
if intent == QueryIntent.ENHANCEMENT:
return MemoryStrategy.FOCUSED_QA
# Clarification requests need recent context and Q&A
if intent == QueryIntent.CLARIFICATION:
return MemoryStrategy.RECENT_FOCUS
# Comparison requests need broad context
if intent == QueryIntent.COMPARISON:
return MemoryStrategy.BROAD_CONTEXT
# Reference requests need focused Q&A
if intent == QueryIntent.REFERENCE:
return MemoryStrategy.FOCUSED_QA
# New topics need semantic deep search
if intent == QueryIntent.NEW_TOPIC:
return MemoryStrategy.SEMANTIC_DEEP
# Continuation requests use mixed approach
if intent == QueryIntent.CONTINUATION:
return MemoryStrategy.MIXED_APPROACH
# Default to mixed approach
return MemoryStrategy.MIXED_APPROACH
except Exception as e:
logger.warning(f"[STRATEGY_PLANNER] Strategy determination failed: {e}")
return MemoryStrategy.MIXED_APPROACH
def plan_retrieval_parameters(self, user_id: str, question: str, intent: QueryIntent,
strategy: MemoryStrategy, conversation_context: Dict[str, Any],
nvidia_rotator) -> Dict[str, Any]:
"""Plan specific retrieval parameters based on strategy"""
try:
params = {
"recent_limit": 3,
"semantic_limit": 5,
"qa_focus": False,
"enhancement_mode": False,
"priority_types": ["conversation"],
"similarity_threshold": 0.15,
"use_ai_selection": False
}
# Adjust parameters based on strategy
if strategy == MemoryStrategy.FOCUSED_QA:
params.update({
"recent_limit": 5, # More recent Q&A pairs
"semantic_limit": 10, # More semantic Q&A pairs
"qa_focus": True,
"enhancement_mode": True,
"priority_types": ["conversation", "qa"],
"similarity_threshold": 0.1, # Lower threshold for more results
"use_ai_selection": True
})
elif strategy == MemoryStrategy.RECENT_FOCUS:
params.update({
"recent_limit": 5,
"semantic_limit": 3,
"qa_focus": True,
"priority_types": ["conversation"]
})
elif strategy == MemoryStrategy.BROAD_CONTEXT:
params.update({
"recent_limit": 3,
"semantic_limit": 15,
"qa_focus": False,
"priority_types": ["conversation", "general", "knowledge"],
"similarity_threshold": 0.2
})
elif strategy == MemoryStrategy.SEMANTIC_DEEP:
params.update({
"recent_limit": 2,
"semantic_limit": 20,
"qa_focus": False,
"priority_types": ["conversation", "general", "knowledge", "qa"],
"similarity_threshold": 0.1,
"use_ai_selection": True
})
elif strategy == MemoryStrategy.MIXED_APPROACH:
params.update({
"recent_limit": 4,
"semantic_limit": 8,
"qa_focus": True,
"priority_types": ["conversation", "qa"],
"use_ai_selection": True
})
# Special handling for enhancement requests
if intent == QueryIntent.ENHANCEMENT:
params["enhancement_mode"] = True
params["qa_focus"] = True
params["use_ai_selection"] = True
params["similarity_threshold"] = 0.05 # Very low threshold for maximum recall
return params
except Exception as e:
logger.warning(f"[STRATEGY_PLANNER] Parameter planning failed: {e}")
return {
"recent_limit": 3,
"semantic_limit": 5,
"qa_focus": False,
"enhancement_mode": False,
"priority_types": ["conversation"],
"similarity_threshold": 0.15,
"use_ai_selection": False
}
def get_fallback_plan(self) -> Dict[str, Any]:
"""Get fallback plan when planning fails"""
return {
"intent": QueryIntent.CONTINUATION,
"strategy": MemoryStrategy.MIXED_APPROACH,
"retrieval_params": {
"recent_limit": 3,
"semantic_limit": 5,
"qa_focus": False,
"enhancement_mode": False,
"priority_types": ["conversation"],
"similarity_threshold": 0.15,
"use_ai_selection": False
},
"conversation_context": {},
"enhancement_focus": False,
"qa_focus": False
}
# ────────────────────────────── Global Instance ──────────────────────────────
_strategy_planner: Optional[StrategyPlanner] = None
def get_strategy_planner() -> StrategyPlanner:
"""Get the global strategy planner instance"""
global _strategy_planner
if _strategy_planner is None:
_strategy_planner = StrategyPlanner()
logger.info("[STRATEGY_PLANNER] Global strategy planner initialized")
return _strategy_planner
|