File size: 9,582 Bytes
9abace2 | 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 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 | """
Multi-Provider Adapter with MoE Routing
Part of SOVEREIGN PYTHON LLM ENGINE
Mixture of Experts routing:
- Code tasks → Nemotron 70B (best for coding)
- Creative/chat → Mistral 7B (fast, creative)
- Reasoning → Nemotron 70B (best reasoning)
- Fallback → Ollama (local Llama 3.2, Muse 1.0)
"""
import os
from typing import Any, AsyncIterator
from .openrouter import OpenRouterProvider
from .ollama import OllamaProvider
# MoE Task Classification
def classify_task(messages: list[dict[str, Any]], system: str | None = None) -> str:
"""
Classify task type from messages to route to best expert.
Returns:
"code" | "creative" | "reasoning" | "chat"
"""
# Combine all text
all_text = (system or "").lower()
for msg in messages or []:
all_text += " " + msg.get("content", "").lower()
# Code indicators
code_keywords = ["function", "class", "def ", "import", "const", "let", "var",
"python", "javascript", "typescript", "rust", "go", "code",
"bug", "error", "debug", "implement", "refactor"]
# Creative indicators
creative_keywords = ["write", "story", "poem", "creative", "imagine", "describe",
"explain like", "eli5", "metaphor", "analogy"]
# Reasoning indicators
reasoning_keywords = ["analyze", "compare", "evaluate", "reason", "logic", "proof",
"theorem", "mathematical", "calculate", "solve", "deduce"]
code_score = sum(1 for kw in code_keywords if kw in all_text)
creative_score = sum(1 for kw in creative_keywords if kw in all_text)
reasoning_score = sum(1 for kw in reasoning_keywords if kw in all_text)
if code_score >= 2:
return "code"
elif reasoning_score >= 2:
return "reasoning"
elif creative_score >= 2:
return "creative"
else:
return "chat"
class MultiProvider:
"""
Multi-provider with MoE routing and fallback.
Expert routing:
- Code → Nemotron 70B (best coding model)
- Reasoning → Nemotron 70B (best logic)
- Creative → Mistral 7B (fast, creative)
- Chat → Mistral 7B or Llama 3.2
Fallback chain:
1. OpenRouter (if API key set)
2. Ollama (local)
"""
def __init__(self, key_manager=None):
"""Initialize multi-provider with MoE routing."""
self.providers = []
self.has_openrouter = False
self.key_manager = key_manager
# Try OpenRouter first if API key available
openrouter_key = None
if key_manager and key_manager.is_valid("openrouter"):
openrouter_key = key_manager.get_key("openrouter")
else:
openrouter_key = os.getenv("OPENROUTER_API_KEY")
if openrouter_key:
try:
self.providers.append({
"name": "openrouter",
"provider": OpenRouterProvider(api_key=openrouter_key),
"models": {
"code": "nvidia/llama-3.1-nemotron-70b-instruct:free",
"reasoning": "nvidia/llama-3.1-nemotron-70b-instruct:free",
"creative": "mistralai/mistral-7b-instruct:free",
"chat": "mistralai/mistral-7b-instruct:free"
}
})
self.has_openrouter = True
print("OK: OpenRouter loaded: Nemotron 70B (code/reasoning), Mistral 7B (creative/chat)")
except Exception as e:
print(f"OpenRouter init failed: {e}")
# Always add Ollama as fallback
self.providers.append({
"name": "ollama",
"provider": OllamaProvider(),
"models": {
"code": "codellama",
"reasoning": "llama3.2",
"creative": "muse:1.0",
"chat": "llama3.2"
}
})
print("OK: Ollama loaded: CodeLlama (code), Llama 3.2 (reasoning/chat), Muse 1.0 (creative)")
async def invoke_model(
self,
model_id: str | None = None,
messages: list[dict[str, Any]] = None,
max_tokens: int = 2048,
temperature: float = 0.7,
system: str | None = None,
tools: list[dict[str, Any]] | None = None,
**kwargs
) -> dict[str, Any]:
"""
Invoke model with MoE routing and fallback.
Routes to best expert based on task type, with fallback chain.
Args:
model_id: Override model (skips MoE routing)
messages: Chat messages
max_tokens: Max response tokens
temperature: Sampling temperature
system: System prompt
tools: Tool definitions
**kwargs: Additional parameters
Returns:
Response dict with "content" field
"""
# Classify task for MoE routing (unless model specified)
if model_id is None:
task_type = classify_task(messages, system)
else:
task_type = "chat" # Default if user specified model
last_error = None
for provider_config in self.providers:
provider_name = provider_config["name"]
provider = provider_config["provider"]
model_map = provider_config["models"]
# Select expert for this task
if model_id:
effective_model = model_id
else:
effective_model = model_map.get(task_type, model_map.get("chat", "llama3.2"))
try:
print(f"→ Routing {task_type} task to {provider_name}:{effective_model}")
result = await provider.invoke_model(
model_id=effective_model,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
system=system,
tools=tools,
**kwargs
)
# Success - return result with metadata
result["_provider"] = provider_name
result["_model"] = effective_model
result["_task_type"] = task_type
return result
except Exception as e:
last_error = e
print(f"ERROR: {provider_name} failed: {e}")
print(f" Trying next provider...")
continue
# All providers failed
raise RuntimeError(f"All providers failed. Last error: {last_error}")
async def invoke_model_stream(
self,
model_id: str | None = None,
messages: list[dict[str, Any]] = None,
max_tokens: int = 2048,
temperature: float = 0.7,
system: str | None = None,
**kwargs
) -> AsyncIterator[dict[str, Any]]:
"""
Invoke model with streaming, MoE routing, and fallback.
Args:
model_id: Override model (skips MoE routing)
messages: Chat messages
max_tokens: Max response tokens
temperature: Sampling temperature
system: System prompt
**kwargs: Additional parameters
Yields:
Response chunks
"""
# Classify task for MoE routing
if model_id is None:
task_type = classify_task(messages, system)
else:
task_type = "chat"
last_error = None
for provider_config in self.providers:
provider_name = provider_config["name"]
provider = provider_config["provider"]
model_map = provider_config["models"]
# Select expert for this task
if model_id:
effective_model = model_id
else:
effective_model = model_map.get(task_type, model_map.get("chat", "llama3.2"))
try:
print(f"→ Streaming {task_type} task via {provider_name}:{effective_model}")
async for chunk in provider.invoke_model_stream(
model_id=effective_model,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
system=system,
**kwargs
):
chunk["_provider"] = provider_name
chunk["_model"] = effective_model
chunk["_task_type"] = task_type
yield chunk
# If we successfully streamed, we're done
return
except Exception as e:
last_error = e
print(f"ERROR: {provider_name} stream failed: {e}")
print(f" Trying next provider...")
continue
# All providers failed
raise RuntimeError(f"All providers failed. Last error: {last_error}")
async def list_providers(self) -> list[dict[str, Any]]:
"""
List available providers and their models.
Returns:
List of provider configs
"""
result = []
for config in self.providers:
result.append({
"name": config["name"],
"available_models": config["models"]
})
return result
|