nexus-os-space / app.py
specimba's picture
v5.0 Provider Control Center: provider manager, arena comparison, experiment log, pinecone chat, model registry
76da8c7 verified
"""
NEXUS OS v5.0 β€” Provider Control Center
A multi-provider LLM management dashboard.
Tabs:
1. Provider Manager β€” enter API keys, check health, see available models
2. Side-by-Side Arena β€” compare 2 providers on same prompt
3. Experiment Log β€” track runs, sort by latency/cost
4. Pinecone Chat β€” talk to pineosman2 assistant
5. Model Registry β€” browse 37+ models
Self-contained. Only dependency: gradio.
"""
import os
import sys
import json
import time
import urllib.request
import urllib.error
from typing import Optional, Dict, Any, List, Tuple
from dataclasses import dataclass, field
from enum import Enum
try:
import gradio as gr
GRADIO_AVAILABLE = True
except ImportError:
GRADIO_AVAILABLE = False
# ═══════════════════════════════════════════════════════════════
# PROVIDERS
# ═══════════════════════════════════════════════════════════════
class Provider:
def __init__(self, name, display_name, domain, key_env):
self.name = name
self.display_name = display_name
self.domain = domain
self.key_env = key_env
PROVIDERS = [
Provider("hf_router", "HF Inference Providers", "router.huggingface.co", "HF_TOKEN"),
Provider("groq", "Groq", "api.groq.com", "GROQ_API_KEY"),
Provider("deepseek", "DeepSeek", "api.deepseek.com", "DEEPSEEK_API_KEY"),
Provider("openrouter", "OpenRouter", "openrouter.ai", "OPENROUTER_API_KEY"),
Provider("together", "Together AI", "api.together.xyz", "TOGETHER_API_KEY"),
Provider("kilocode", "Kilocode", "kilocode.ai", "KILOCODE_API_KEY"),
Provider("nvidia", "NVIDIA NIM", "integrate.api.nvidia.com", "NVIDIA_API_KEY"),
]
PROVIDER_MAP = {p.name: p for p in PROVIDERS}
ENDPOINTS = {
"hf_router": "https://router.huggingface.co/v1/chat/completions",
"groq": "https://api.groq.com/openai/v1/chat/completions",
"deepseek": "https://api.deepseek.com/v1/chat/completions",
"openrouter": "https://openrouter.ai/api/v1/chat/completions",
"together": "https://api.together.xyz/v1/chat/completions",
}
FREE_MODELS = {
"hf_router": ["HuggingFaceTB/SmolLM2-1.7B-Instruct", "meta-llama/Llama-3.2-1B-Instruct", "Qwen/Qwen2.5-0.5B-Instruct"],
"groq": ["llama-3.2-1b-preview", "llama-3.2-3b-preview", "mixtral-8x7b-32768"],
"deepseek": ["deepseek-chat", "deepseek-reasoner"],
"openrouter": ["meta-llama/llama-3.2-1b-instruct:free"],
"together": ["meta-llama/Llama-3.3-70B-Instruct-Turbo-Free"],
}
# ═══════════════════════════════════════════════════════════════
# API HELPERS
# ═══════════════════════════════════════════════════════════════
def _api_call(endpoint: str, api_key: str, payload: Dict[str, Any], timeout: int = 120):
body = json.dumps(payload).encode("utf-8")
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
if "openrouter" in endpoint:
headers["HTTP-Referer"] = "https://huggingface.co/spaces/specimba/nexus-os-space"
headers["X-Title"] = "NEXUS OS"
req = urllib.request.Request(endpoint, data=body, headers=headers, method="POST")
t0 = time.time()
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return True, json.loads(resp.read().decode("utf-8")), (time.time() - t0) * 1000, ""
except urllib.error.HTTPError as e:
err = e.read().decode("utf-8", errors="replace")[:300]
return False, {}, (time.time() - t0) * 1000, f"HTTP {e.code}: {err}"
except Exception as e:
return False, {}, (time.time() - t0) * 1000, str(e)[:200]
def check_provider_health(provider_name: str, api_key: str) -> Dict[str, Any]:
"""Check provider health. Returns dict with status, latency, models, error."""
provider = PROVIDER_MAP.get(provider_name)
if not provider:
return {"provider": provider_name, "status": "unknown", "latency_ms": 0, "error": "Unknown provider", "models": []}
if not api_key:
return {"provider": provider.display_name, "status": "no_key", "latency_ms": 0,
"error": "", "models": FREE_MODELS.get(provider_name, [])}
endpoint = ENDPOINTS.get(provider_name)
if not endpoint:
return {"provider": provider.display_name, "status": "no_endpoint", "latency_ms": 0,
"error": "No endpoint configured", "models": []}
models = FREE_MODELS.get(provider_name, [])
model = models[0] if models else ""
if not model:
return {"provider": provider.display_name, "status": "no_models", "latency_ms": 0,
"error": "No free models configured", "models": []}
payload = {"model": model, "messages": [{"role": "user", "content": "Hi"}], "max_tokens": 5, "temperature": 0.1}
success, data, latency, error = _api_call(endpoint, api_key, payload, timeout=15)
if success and data.get("choices"):
return {"provider": provider.display_name, "status": "online", "latency_ms": round(latency, 1),
"error": "", "models": models}
elif "429" in error or "rate limit" in error.lower():
return {"provider": provider.display_name, "status": "rate_limited", "latency_ms": round(latency, 1),
"error": error[:100], "models": models}
else:
return {"provider": provider.display_name, "status": "offline", "latency_ms": round(latency, 1),
"error": error[:100], "models": models}
def generate_with_provider(provider_name: str, api_key: str, model: str, prompt: str,
system: Optional[str] = None, max_tokens: int = 512, temperature: float = 0.7) -> Dict[str, Any]:
"""Generate text with a provider. Returns dict with text, latency, tokens, error."""
endpoint = ENDPOINTS.get(provider_name)
if not endpoint:
return {"text": "", "latency_ms": 0, "tokens_input": 0, "tokens_output": 0, "error": "No endpoint"}
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
payload = {"model": model, "messages": messages, "max_tokens": max_tokens, "temperature": temperature}
success, data, latency, error = _api_call(endpoint, api_key, payload)
if not success:
return {"text": "", "latency_ms": round(latency, 1), "tokens_input": 0, "tokens_output": 0, "error": error}
choice = data.get("choices", [{}])[0]
message = choice.get("message", {})
usage = data.get("usage", {})
return {
"text": message.get("content", ""),
"latency_ms": round(latency, 1),
"tokens_input": usage.get("prompt_tokens", 0),
"tokens_output": usage.get("completion_tokens", 0),
"error": "",
}
# ═══════════════════════════════════════════════════════════════
# MODEL REGISTRY
# ═══════════════════════════════════════════════════════════════
REGISTRY = {
"deepseek-r1-8b": {"name": "DeepSeek-R1 8B", "family": "deepseek", "tier": "16GB", "size_gb": 5.2, "params_b": 8.0, "caps": "reasoning, coding, long_context", "max_context": 128000, "temp": 0.6},
"qwen2.5-coder-7b": {"name": "Qwen 2.5 Coder 7B", "family": "qwen", "tier": "16GB", "size_gb": 4.7, "params_b": 7.0, "caps": "coding, fast", "max_context": 32768, "temp": 0.3},
"l3.1-dark-reasoning-8b": {"name": "L3.1 Dark Reasoning 8B", "family": "llama", "tier": "16GB", "size_gb": 5.7, "params_b": 8.0, "caps": "reasoning, coding", "max_context": 32768, "temp": 0.7},
"omega-evolution-9b": {"name": "Omega Evolution 9B", "family": "omega", "tier": "16GB", "size_gb": 6.6, "params_b": 9.0, "caps": "reasoning, coding, vision", "max_context": 32768, "temp": 0.7},
"darwin-9b-opus": {"name": "Darwin 9B Opus", "family": "darwin", "tier": "16GB", "size_gb": 6.3, "params_b": 9.0, "caps": "reasoning, coding, long_context", "max_context": 65536, "temp": 0.7},
"qwopus-3.5-9b": {"name": "Qwopus 3.5 9B", "family": "qwopus", "tier": "16GB", "size_gb": 5.6, "params_b": 9.0, "caps": "reasoning, coding", "max_context": 32768, "temp": 0.7},
"carnice-9b": {"name": "Carnice 9B", "family": "carnice", "tier": "16GB", "size_gb": 5.6, "params_b": 9.0, "caps": "reasoning, coding, vision", "max_context": 32768, "temp": 0.7},
"open-search-vl-8b": {"name": "OpenSearch VL 8B", "family": "opensearch", "tier": "16GB", "size_gb": 6.6, "params_b": 8.0, "caps": "vision, reasoning, long_context", "max_context": 65536, "temp": 0.7},
"granite-4.1-8b-abliterated": {"name": "Granite 4.1 8B Abliterated", "family": "granite", "tier": "16GB", "size_gb": 5.1, "params_b": 8.0, "caps": "reasoning, coding, long_context", "max_context": 128000, "temp": 0.7},
"jaahas-qwen3.5-9b": {"name": "Jaahas Qwen 3.5 9B", "family": "qwen", "tier": "16GB", "size_gb": 7.4, "params_b": 9.0, "caps": "reasoning, coding, multilingual", "max_context": 32768, "temp": 0.7},
"lfm2-12b-deckard": {"name": "LFM2 12B Deckard", "family": "lfm", "tier": "24GB", "size_gb": 5.8, "params_b": 12.0, "caps": "reasoning, coding, long_context, fast", "max_context": 128000, "temp": 0.7},
"gemma4-e2b-opus": {"name": "Gemma4 E2B Opus", "family": "gemma", "tier": "24GB", "size_gb": 5.5, "params_b": 4.0, "caps": "reasoning, coding, long_context", "max_context": 128000, "temp": 0.7},
"gemma4-uncensored": {"name": "Gemma 4 Uncensored", "family": "gemma", "tier": "24GB", "size_gb": 4.9, "params_b": 4.0, "caps": "reasoning, coding, vision", "max_context": 32768, "temp": 0.7},
"gemma4-obliterated": {"name": "Gemma 4 OBLITERATED", "family": "gemma", "tier": "24GB", "size_gb": 6.3, "params_b": 4.0, "caps": "reasoning, coding, vision", "max_context": 32768, "temp": 0.7},
"qwen3.6-27b-dflash": {"name": "Qwen 3.6 27B DFlash", "family": "qwen", "tier": "24GB", "size_gb": 1.0, "params_b": 27.0, "caps": "reasoning, coding, long_context, fast", "max_context": 128000, "temp": 0.7},
"gemma4-31b-cloud": {"name": "Gemma4 31B Cloud", "family": "gemma", "tier": "48GB", "size_gb": 18.0, "params_b": 31.0, "caps": "reasoning, coding, vision, long_context, multilingual", "max_context": 128000, "temp": 0.7},
"nemotron-3-nano-omni-30b": {"name": "Nemotron-3 Nano-Omni 30B", "family": "nemotron", "tier": "48GB", "size_gb": 18.0, "params_b": 30.0, "caps": "reasoning, coding, vision, long_context, safety, tool_use", "max_context": 256000, "temp": 0.6},
"opensonnet-lite-max": {"name": "OpenSonnet-Lite-MAX", "family": "qwen", "tier": "8GB", "size_gb": 2.5, "params_b": 4.0, "caps": "reasoning, coding, fast, long_context", "max_context": 262144, "temp": 0.6},
"deepseek-v4-pro": {"name": "DeepSeek V4 Pro", "family": "deepseek", "tier": "cloud", "size_gb": 0.0, "params_b": 671.0, "caps": "reasoning, coding, long_context, multilingual, tool_use", "max_context": 64000, "temp": 0.6},
"qwen3-coder-next": {"name": "Qwen 3 Coder Next", "family": "qwen", "tier": "cloud", "size_gb": 0.0, "params_b": 32.0, "caps": "coding, reasoning, fast, long_context, tool_use", "max_context": 128000, "temp": 0.3},
"kimi-k2.6": {"name": "Kimi K2.6", "family": "kimi", "tier": "cloud", "size_gb": 0.0, "params_b": 32.0, "caps": "reasoning, coding, long_context, multilingual, vision", "max_context": 200000, "temp": 0.7},
"glm-5.1": {"name": "GLM 5.1", "family": "glm", "tier": "cloud", "size_gb": 0.0, "params_b": 32.0, "caps": "reasoning, coding, multilingual, tool_use, vision", "max_context": 128000, "temp": 0.7},
}
# ═══════════════════════════════════════════════════════════════
# EXPERIMENT LOG (session state)
# ═══════════════════════════════════════════════════════════════
experiment_log: List[Dict[str, Any]] = []
# ═══════════════════════════════════════════════════════════════
# GRADIO INTERFACE
# ═══════════════════════════════════════════════════════════════
def build_control_center():
with gr.Blocks(title="NEXUS OS β€” Provider Control Center") as demo:
gr.Markdown("""
# πŸ”₯ NEXUS OS β€” Provider Control Center
**Manage API providers, compare models, log experiments, chat with your knowledge base.**
""")
with gr.Tabs():
# ═══════════════════════════════════════════════════
# TAB 1: Provider Manager
# ═══════════════════════════════════════════════════
with gr.TabItem("πŸ”Œ Provider Manager"):
gr.Markdown("""
### Enter API keys and check provider health
Keys are stored **in this session only** (not saved).
""")
# Provider key inputs
key_inputs = {}
for provider in PROVIDERS:
key_inputs[provider.name] = gr.Textbox(
label=f"{provider.display_name} API Key",
placeholder=f"Paste {provider.key_env} here...",
type="password",
value=os.environ.get(provider.key_env, ""),
)
check_btn = gr.Button("πŸ” Check All Providers", variant="primary")
health_table = gr.DataFrame(
label="Provider Health Dashboard",
headers=["Provider", "Status", "Latency (ms)", "Free Models", "Error"],
interactive=False,
)
def check_all(*keys):
results = []
for provider, key in zip(PROVIDERS, keys):
h = check_provider_health(provider.name, key)
emoji = {"online": "🟒", "rate_limited": "🟑", "offline": "πŸ”΄",
"no_key": "βšͺ", "no_endpoint": "⚫", "no_models": "⚫"}.get(h["status"], "βšͺ")
models = ", ".join(h["models"][:3]) if h["models"] else "N/A"
results.append({
"Provider": f"{emoji} {h['provider']}",
"Status": h["status"],
"Latency (ms)": str(h["latency_ms"]) if h["latency_ms"] > 0 else "N/A",
"Free Models": models,
"Error": h["error"],
})
return results
check_btn.click(
fn=check_all,
inputs=list(key_inputs.values()),
outputs=[health_table],
)
# ═══════════════════════════════════════════════════
# TAB 2: Side-by-Side Arena
# ═══════════════════════════════════════════════════
with gr.TabItem("βš”οΈ Side-by-Side Arena"):
gr.Markdown("""
### Send the same prompt to 2 providers and compare
""")
with gr.Row():
arena_prompt = gr.Textbox(
label="Prompt",
placeholder="Write a Python function to reverse a linked list...",
lines=4,
scale=2,
)
arena_system = gr.Textbox(
label="System Prompt (optional)",
placeholder="You are a helpful coding assistant...",
lines=2,
scale=1,
)
with gr.Row():
left_provider = gr.Dropdown(
label="Left Provider",
choices=[(p.display_name, p.name) for p in PROVIDERS if p.name in ENDPOINTS],
value="hf_router",
)
right_provider = gr.Dropdown(
label="Right Provider",
choices=[(p.display_name, p.name) for p in PROVIDERS if p.name in ENDPOINTS],
value="groq",
)
with gr.Row():
left_model = gr.Dropdown(label="Left Model", choices=[], value="")
right_model = gr.Dropdown(label="Right Model", choices=[], value="")
with gr.Row():
arena_max_tokens = gr.Slider(minimum=64, maximum=2048, value=512, step=64, label="Max Tokens")
arena_temperature = gr.Slider(minimum=0.0, maximum=2.0, value=0.7, step=0.1, label="Temperature")
arena_go = gr.Button("πŸš€ Run Arena", variant="primary")
with gr.Row():
with gr.Column():
left_output = gr.Textbox(label="Left Output", lines=12, interactive=False)
left_metrics = gr.Textbox(label="Left Metrics", interactive=False)
with gr.Column():
right_output = gr.Textbox(label="Right Output", lines=12, interactive=False)
right_metrics = gr.Textbox(label="Right Metrics", interactive=False)
# Update model lists when provider changes
def update_models(provider_name):
models = FREE_MODELS.get(provider_name, [])
choices = [(m, m) for m in models]
default = models[0] if models else ""
return gr.Dropdown(choices=choices, value=default)
left_provider.change(fn=update_models, inputs=[left_provider], outputs=[left_model])
right_provider.change(fn=update_models, inputs=[right_provider], outputs=[right_model])
# Run arena
def run_arena(prompt, system, left_prov, right_prov, left_mod, right_mod,
max_tokens, temperature, *all_keys):
if not prompt.strip():
return "Enter a prompt", "", "Enter a prompt", ""
key_map = {p.name: k for p, k in zip(PROVIDERS, all_keys)}
# Left
left_key = key_map.get(left_prov, "")
left_result = generate_with_provider(left_prov, left_key, left_mod, prompt, system, max_tokens, temperature) if left_key else {"text": "❌ No API key", "latency_ms": 0, "tokens_output": 0, "error": "No key"}
left_text = left_result["text"] if not left_result["error"] else f"❌ {left_result['error']}"
left_met = f"⏱️ {left_result['latency_ms']}ms | πŸ“ {left_result['tokens_output']} tokens | 🎲 {left_mod}"
# Right
right_key = key_map.get(right_prov, "")
right_result = generate_with_provider(right_prov, right_key, right_mod, prompt, system, max_tokens, temperature) if right_key else {"text": "❌ No API key", "latency_ms": 0, "tokens_output": 0, "error": "No key"}
right_text = right_result["text"] if not right_result["error"] else f"❌ {right_result['error']}"
right_met = f"⏱️ {right_result['latency_ms']}ms | πŸ“ {right_result['tokens_output']} tokens | 🎲 {right_mod}"
# Log to experiment log
global experiment_log
import datetime
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
experiment_log.append({
"Time": now,
"Provider": f"{left_prov} vs {right_prov}",
"Model": f"{left_mod} vs {right_mod}",
"Prompt (first 50 chars)": prompt[:50],
"Latency Left (ms)": left_result["latency_ms"],
"Latency Right (ms)": right_result["latency_ms"],
"Tokens Left": left_result["tokens_output"],
"Tokens Right": right_result["tokens_output"],
"Status": "complete" if not left_result["error"] and not right_result["error"] else "error",
})
return left_text, left_met, right_text, right_met
arena_go.click(
fn=run_arena,
inputs=[arena_prompt, arena_system, left_provider, right_provider,
left_model, right_model, arena_max_tokens, arena_temperature] + list(key_inputs.values()),
outputs=[left_output, left_metrics, right_output, right_metrics],
)
# ═══════════════════════════════════════════════════
# TAB 3: Experiment Log
# ═══════════════════════════════════════════════════
with gr.TabItem("πŸ“Š Experiment Log"):
gr.Markdown("""
### Track and compare your arena runs
""")
log_table = gr.DataFrame(
label="Experiment History",
headers=["Time", "Provider", "Model", "Prompt (first 50 chars)",
"Latency Left (ms)", "Latency Right (ms)", "Tokens Left", "Tokens Right", "Status"],
interactive=False,
)
refresh_log_btn = gr.Button("πŸ”„ Refresh Log")
clear_log_btn = gr.Button("πŸ—‘οΈ Clear Log")
def refresh_log():
return experiment_log
def clear_log():
global experiment_log
experiment_log = []
return []
refresh_log_btn.click(fn=refresh_log, outputs=[log_table])
clear_log_btn.click(fn=clear_log, outputs=[log_table])
# ═══════════════════════════════════════════════════
# TAB 4: Pinecone Chat
# ═══════════════════════════════════════════════════
with gr.TabItem("🌲 Pinecone Chat"):
gr.Markdown("""
### Chat with your Pinecone Assistant `pineosman2`
Uses Pinecone's conversational retrieval over your uploaded documents.
""")
pinecone_key = gr.Textbox(
label="Pinecone API Key",
placeholder="pcsk_...",
type="password",
value=os.environ.get("PINECONE_API_KEY", ""),
)
pinecone_chat = gr.Chatbot(label="Conversation with pineosman2", height=400)
pinecone_msg = gr.Textbox(label="Your message", placeholder="Ask about your documents...")
pinecone_send = gr.Button("Send", variant="primary")
pinecone_status = gr.Textbox(label="Status", interactive=False)
def pinecone_chat_fn(message, history, api_key):
if not api_key:
return history + [(message, "❌ Please enter your Pinecone API key")], "No key"
if not message.strip():
return history, "Empty message"
# Use Pinecone REST API
try:
payload = json.dumps({
"messages": [{"role": "user", "content": message}],
}).encode("utf-8")
req = urllib.request.Request(
"https://api.pinecone.io/assistant/chat/pineosman2",
data=payload,
headers={
"Content-Type": "application/json",
"Api-Key": api_key,
},
method="POST",
)
with urllib.request.urlopen(req, timeout=60) as resp:
data = json.loads(resp.read().decode("utf-8"))
reply = data.get("message", {}).get("content", "No response")
return history + [(message, reply)], f"βœ“ Response received ({len(reply)} chars)"
except Exception as e:
return history + [(message, f"❌ Error: {str(e)[:200]}")], f"Error: {str(e)[:100]}"
pinecone_send.click(
fn=pinecone_chat_fn,
inputs=[pinecone_msg, pinecone_chat, pinecone_key],
outputs=[pinecone_chat, pinecone_status],
).then(lambda: "", outputs=[pinecone_msg])
# ═══════════════════════════════════════════════════
# TAB 5: Model Registry
# ═══════════════════════════════════════════════════
with gr.TabItem("πŸ“‹ Model Registry"):
gr.Markdown("""
### Browse all models in the NEXUS OS registry
""")
registry_table = gr.DataFrame(
label="Registered Models",
headers=["ID", "Name", "Family", "Tier", "Size (GB)", "Params (B)",
"Capabilities", "Context", "Temp"],
interactive=False,
)
def load_registry():
return [{
"ID": k,
"Name": v["name"],
"Family": v["family"],
"Tier": v["tier"],
"Size (GB)": v["size_gb"],
"Params (B)": v["params_b"],
"Capabilities": v["caps"],
"Context": v["max_context"],
"Temp": v["temp"],
} for k, v in REGISTRY.items()]
demo.load(fn=load_registry, outputs=[registry_table])
return demo
if __name__ == "__main__":
if not GRADIO_AVAILABLE:
print("ERROR: Gradio is required.")
sys.exit(1)
demo = build_control_center()
demo.launch(server_name="0.0.0.0", server_port=7860, share=False, show_error=True)