model_adapter.py
1 """ 2 Model Adapter - Base interface for AI Agent Agnosticism 3 4 All AI models implement this interface, making them interchangeable. 5 Claude, Gemini, Grok, Perplexity - all speak the same language. 6 7 Usage: 8 adapter = GeminiAdapter() 9 response = adapter.query("Your prompt here", task_type="coding") 10 """ 11 12 from abc import ABC, abstractmethod 13 from dataclasses import dataclass, field 14 from enum import Enum 15 from typing import Optional, Dict, Any, List 16 from pathlib import Path 17 import json 18 from datetime import datetime 19 20 21 class TaskType(Enum): 22 """Task types for routing optimization.""" 23 CODING = "coding" # Code generation, debugging 24 RESEARCH = "research" # Web search, fact finding 25 ANALYSIS = "analysis" # Code review, architecture 26 QUICK = "quick" # Simple questions, quick fixes 27 CONSENSUS = "consensus" # High-stakes, need multiple opinions 28 REALTIME = "realtime" # Current events, live data 29 30 31 @dataclass 32 class ModelCapabilities: 33 """What a model can do well.""" 34 coding: float = 0.5 # 0-1 coding ability 35 research: float = 0.5 # 0-1 research ability 36 speed: float = 0.5 # 0-1 response speed 37 cost_efficiency: float = 0.5 # 0-1 cost efficiency (higher = cheaper) 38 context_window: int = 100000 # Token context window 39 supports_streaming: bool = True 40 supports_tools: bool = False 41 supports_vision: bool = False 42 43 44 @dataclass 45 class ModelResponse: 46 """Standardized response from any model.""" 47 content: str 48 model_name: str 49 model_id: str 50 task_type: TaskType 51 timestamp: datetime = field(default_factory=datetime.now) 52 tokens_used: Optional[int] = None 53 latency_ms: Optional[int] = None 54 confidence: float = 1.0 55 metadata: Dict[str, Any] = field(default_factory=dict) 56 57 def to_dict(self) -> Dict[str, Any]: 58 return { 59 "content": self.content, 60 "model_name": self.model_name, 61 "model_id": self.model_id, 62 "task_type": self.task_type.value, 63 "timestamp": self.timestamp.isoformat(), 64 "tokens_used": self.tokens_used, 65 "latency_ms": self.latency_ms, 66 "confidence": self.confidence, 67 "metadata": self.metadata, 68 } 69 70 71 class ModelAdapter(ABC): 72 """ 73 Abstract base class for all AI model adapters. 74 75 Implement this interface to add a new model to Sovereign OS. 76 All adapters share: 77 - Same query interface 78 - Same context injection 79 - Same response format 80 - Same mesh integration 81 """ 82 83 def __init__(self): 84 self._sovereign_root = Path(__file__).parent.parent.parent 85 self._context_cache: Optional[str] = None 86 self._context_timestamp: Optional[datetime] = None 87 88 @property 89 @abstractmethod 90 def name(self) -> str: 91 """Human-readable model name (e.g., 'Claude Opus').""" 92 pass 93 94 @property 95 @abstractmethod 96 def model_id(self) -> str: 97 """API model identifier (e.g., 'claude-opus-4-5-20251101').""" 98 pass 99 100 @property 101 @abstractmethod 102 def capabilities(self) -> ModelCapabilities: 103 """What this model does well.""" 104 pass 105 106 @abstractmethod 107 def is_available(self) -> bool: 108 """Check if model is available (API key set, credits available, etc.).""" 109 pass 110 111 @abstractmethod 112 def _raw_query(self, prompt: str, **kwargs) -> str: 113 """ 114 Internal: Send prompt to model and get raw response. 115 Subclasses implement this with their specific API. 116 """ 117 pass 118 119 def query( 120 self, 121 prompt: str, 122 task_type: TaskType = TaskType.CODING, 123 inject_context: bool = True, 124 file_content: Optional[str] = None, 125 **kwargs 126 ) -> ModelResponse: 127 """ 128 Query the model with Sovereign OS context injection. 129 130 This is the main entry point - same for ALL models. 131 132 Args: 133 prompt: The user's question/request 134 task_type: Type of task for optimization 135 inject_context: Whether to inject Sovereign OS context 136 file_content: Optional file content for code review 137 **kwargs: Model-specific parameters 138 139 Returns: 140 ModelResponse with standardized format 141 """ 142 import time 143 start = time.time() 144 145 # Build full prompt with context 146 full_prompt = "" 147 148 if inject_context: 149 context = self._load_context() 150 if context: 151 full_prompt += f"{context}\n\n" 152 153 if file_content: 154 full_prompt += f"## File Content:\n```\n{file_content}\n```\n\n" 155 156 full_prompt += f"## User Request:\n{prompt}" 157 158 # Query the model 159 try: 160 raw_response = self._raw_query(full_prompt, **kwargs) 161 latency_ms = int((time.time() - start) * 1000) 162 163 return ModelResponse( 164 content=raw_response, 165 model_name=self.name, 166 model_id=self.model_id, 167 task_type=task_type, 168 latency_ms=latency_ms, 169 metadata={"injected_context": inject_context}, 170 ) 171 except Exception as e: 172 return ModelResponse( 173 content=f"Error: {e}", 174 model_name=self.name, 175 model_id=self.model_id, 176 task_type=task_type, 177 confidence=0.0, 178 metadata={"error": str(e)}, 179 ) 180 181 def _load_context(self) -> str: 182 """ 183 Load Sovereign OS context for injection. 184 185 All models receive the same core context: 186 1. Mesh agent protocol (how to participate) 187 2. Four Axioms (always) 188 3. Current focus (from LIVE-COMPRESSION) 189 4. Recent decisions (from mesh) 190 """ 191 # Use cached context if fresh (5 minutes) 192 if (self._context_cache and self._context_timestamp and 193 (datetime.now() - self._context_timestamp).seconds < 300): 194 return self._context_cache 195 196 context_parts = [] 197 198 # 0. Mesh agent protocol (tells agent how to participate) 199 context_parts.append(self._load_mesh_protocol()) 200 201 # 1. Core axioms section 202 context_parts.append(self._load_axioms()) 203 204 # 2. Current focus from LIVE-COMPRESSION 205 live_compression = self._sovereign_root / "sessions" / "LIVE-COMPRESSION.md" 206 if live_compression.exists(): 207 content = live_compression.read_text() 208 # Extract just the current focus section (first 500 chars) 209 if "## Current Focus" in content: 210 start = content.find("## Current Focus") 211 end = content.find("##", start + 10) 212 if end == -1: 213 end = start + 500 214 focus = content[start:end].strip() 215 context_parts.append(f"\n{focus}\n") 216 217 # 3. Recent decisions from FO-STATE 218 fo_state = self._sovereign_root / "sessions" / "FO-STATE.json" 219 if fo_state.exists(): 220 try: 221 state = json.loads(fo_state.read_text()) 222 if state.get("recent_decisions"): 223 decisions = state["recent_decisions"][-3:] # Last 3 224 context_parts.append("\n## Recent Decisions:") 225 for d in decisions: 226 context_parts.append(f"- {d}") 227 except: 228 pass 229 230 context = "\n".join(context_parts) 231 self._context_cache = context 232 self._context_timestamp = datetime.now() 233 234 return context 235 236 def _load_mesh_protocol(self) -> str: 237 """Load full agent protocol from AGENT.md.""" 238 agent_md = self._sovereign_root / "AGENT.md" 239 240 if agent_md.exists(): 241 return agent_md.read_text() 242 243 # Fallback: minimal protocol 244 return """# Sovereign OS Agent 245 246 You are part of a multi-model mesh. Core axioms: 247 - A0: Respect boundaries 248 - A1: Connect, don't isolate 249 - A2: Motion is life 250 - A3: Navigate poles dynamically 251 - A4: Prevent ruin first 252 253 Format outputs: DECISION/INSIGHT/ALERT with CONFIDENCE [0-1] 254 End with: MESH: [type] | AXIOM_CHECK: [A0-A4 status] 255 """ 256 257 def _load_axioms(self) -> str: 258 """Load axioms - now handled by AGENT.md, this is a backup.""" 259 # AGENT.md now contains axioms, so this is minimal 260 return "" 261 262 def __repr__(self) -> str: 263 status = "✓" if self.is_available() else "✗" 264 return f"<{self.name} [{status}] coding={self.capabilities.coding:.1f}>"