mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-27 21:05:34 +00:00
Merge pull request #170 from mrTSB/fix-desktop
Fix Desktop App Cloud Models
This commit is contained in:
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -649,7 +649,12 @@ async fn boot_backend(backend: SharedBackend, status: SharedStatus) {
|
||||
s.detail = "Installing dependencies...".into();
|
||||
}
|
||||
let _ = tokio::process::Command::new(&uv_bin)
|
||||
.args(["sync", "--extra", "server"])
|
||||
.args([
|
||||
"sync",
|
||||
"--extra", "server",
|
||||
"--extra", "inference-cloud",
|
||||
"--extra", "inference-google",
|
||||
])
|
||||
.stdout(std::process::Stdio::null())
|
||||
.stderr(std::process::Stdio::null())
|
||||
.current_dir(root)
|
||||
@@ -1124,6 +1129,15 @@ async fn save_cloud_key(key_name: String, key_value: String) -> Result<(), Strin
|
||||
let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
|
||||
}
|
||||
|
||||
// Tell the running server to hot-reload its cloud engine so the user
|
||||
// doesn't need to restart the app after entering an API key.
|
||||
let reload_url = format!("http://127.0.0.1:{}/v1/cloud/reload", JARVIS_PORT);
|
||||
let _ = reqwest::Client::new()
|
||||
.post(&reload_url)
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.send()
|
||||
.await;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
Generated
+3
@@ -46,6 +46,9 @@
|
||||
"typescript": "~5.7.0",
|
||||
"vite": "^6.0.0",
|
||||
"vite-plugin-pwa": "^1.2.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20"
|
||||
}
|
||||
},
|
||||
"node_modules/@apideck/better-ajv-errors": {
|
||||
|
||||
@@ -243,9 +243,10 @@ export function InputArea() {
|
||||
accumulatedContent = 'No response was generated. Please try again.';
|
||||
}
|
||||
const totalMs = Date.now() - startTime;
|
||||
const serverInfo = useAppStore.getState().serverInfo;
|
||||
const _CLOUD_PREFIXES = ['gpt-', 'o1-', 'o3-', 'o4-', 'claude-', 'gemini-', 'openrouter/', 'MiniMax-', 'chatgpt-'];
|
||||
const engineLabel = _CLOUD_PREFIXES.some(p => selectedModel.startsWith(p)) ? 'cloud' : 'ollama';
|
||||
const telemetry: MessageTelemetry = {
|
||||
engine: serverInfo?.engine,
|
||||
engine: engineLabel,
|
||||
model_id: selectedModel,
|
||||
total_ms: totalMs,
|
||||
ttft_ms: ttftMs,
|
||||
|
||||
@@ -223,8 +223,11 @@ export function CommandPalette() {
|
||||
|
||||
useAppStore.getState().addLogEntry({
|
||||
timestamp: Date.now(), level: 'info', category: 'model',
|
||||
message: `${provider.name} API key ${value ? 'saved' : 'removed'}. Restart the app for cloud models to appear.`,
|
||||
message: `${provider.name} API key ${value ? 'saved' : 'removed'}. Refreshing model list…`,
|
||||
});
|
||||
|
||||
// Refresh the model list so cloud models appear immediately.
|
||||
await refreshModels();
|
||||
};
|
||||
|
||||
const handleKeyDown = (e: React.KeyboardEvent) => {
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"root":["./src/app.tsx","./src/main.tsx","./src/vite-env.d.ts","./src/components/commandpalette.tsx","./src/components/errorboundary.tsx","./src/components/layout.tsx","./src/components/optinmodal.tsx","./src/components/setupscreen.tsx","./src/components/systempulse.tsx","./src/components/chat/chatarea.tsx","./src/components/chat/inputarea.tsx","./src/components/chat/messagebubble.tsx","./src/components/chat/micbutton.tsx","./src/components/chat/streamingdots.tsx","./src/components/chat/systempanel.tsx","./src/components/chat/toolcallcard.tsx","./src/components/chat/xrayfooter.tsx","./src/components/dashboard/costcomparison.tsx","./src/components/dashboard/energydashboard.tsx","./src/components/dashboard/tracedebugger.tsx","./src/components/sidebar/conversationlist.tsx","./src/components/sidebar/sidebar.tsx","./src/components/setup/ingestdashboard.tsx","./src/components/setup/readyscreen.tsx","./src/components/setup/setupwizard.tsx","./src/components/setup/sourceconnectflow.tsx","./src/components/setup/sourcepicker.tsx","./src/components/ui/button.tsx","./src/components/ui/dialog.tsx","./src/components/ui/input.tsx","./src/components/ui/select.tsx","./src/components/ui/sonner.tsx","./src/components/ui/tooltip.tsx","./src/hooks/usespeech.ts","./src/lib/api.ts","./src/lib/connectors-api.ts","./src/lib/deep-link.ts","./src/lib/profanity.ts","./src/lib/sse.ts","./src/lib/store.ts","./src/lib/utils.ts","./src/pages/agentspage.tsx","./src/pages/chatpage.tsx","./src/pages/dashboardpage.tsx","./src/pages/getstartedpage.tsx","./src/pages/logspage.tsx","./src/pages/settingspage.tsx","./src/types/connectors.ts","./src/types/index.ts"],"version":"5.7.3"}
|
||||
{"root":["./src/app.tsx","./src/main.tsx","./src/vite-env.d.ts","./src/components/commandpalette.tsx","./src/components/errorboundary.tsx","./src/components/layout.tsx","./src/components/optinmodal.tsx","./src/components/setupscreen.tsx","./src/components/systempulse.tsx","./src/components/chat/chatarea.tsx","./src/components/chat/inputarea.tsx","./src/components/chat/messagebubble.tsx","./src/components/chat/micbutton.tsx","./src/components/chat/streamingdots.tsx","./src/components/chat/systempanel.tsx","./src/components/chat/toolcallcard.tsx","./src/components/chat/xrayfooter.tsx","./src/components/dashboard/costcomparison.tsx","./src/components/dashboard/energydashboard.tsx","./src/components/dashboard/tracedebugger.tsx","./src/components/sidebar/conversationlist.tsx","./src/components/sidebar/sidebar.tsx","./src/components/setup/ingestdashboard.tsx","./src/components/setup/readyscreen.tsx","./src/components/setup/setupwizard.tsx","./src/components/setup/sourceconnectflow.tsx","./src/components/setup/sourcepicker.tsx","./src/components/ui/button.tsx","./src/components/ui/dialog.tsx","./src/components/ui/input.tsx","./src/components/ui/select.tsx","./src/components/ui/sonner.tsx","./src/components/ui/tooltip.tsx","./src/hooks/usespeech.ts","./src/lib/api.ts","./src/lib/connectors-api.ts","./src/lib/deep-link.ts","./src/lib/profanity.ts","./src/lib/sse.ts","./src/lib/store.ts","./src/lib/utils.ts","./src/pages/agentspage.tsx","./src/pages/chatpage.tsx","./src/pages/dashboardpage.tsx","./src/pages/datasourcespage.tsx","./src/pages/getstartedpage.tsx","./src/pages/logspage.tsx","./src/pages/settingspage.tsx","./src/types/connectors.ts","./src/types/index.ts"],"version":"5.7.3"}
|
||||
@@ -120,10 +120,15 @@ def serve(
|
||||
from openjarvis.engine.multi import MultiEngine
|
||||
|
||||
cloud = CloudEngine()
|
||||
engine = MultiEngine([(engine_name, engine), ("cloud", cloud)])
|
||||
engine_name = "multi"
|
||||
if cloud.health():
|
||||
engine = MultiEngine([(engine_name, engine), ("cloud", cloud)])
|
||||
engine_name = "multi"
|
||||
console.print(" Cloud: [cyan]enabled[/cyan] (API keys detected)")
|
||||
else:
|
||||
console.print(
|
||||
" Cloud: [yellow]keys set but packages missing[/yellow] "
|
||||
"(run: uv sync --extra inference-cloud --extra inference-google)"
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.debug("Cloud engine init failed: %s", exc)
|
||||
|
||||
|
||||
@@ -0,0 +1,389 @@
|
||||
"""Direct cloud API router — bypasses the engine system entirely.
|
||||
|
||||
Reads API keys from ~/.openjarvis/cloud-keys.env at request time so
|
||||
it works even when the server was started without cloud keys in its
|
||||
environment. Uses httpx directly so no cloud SDK packages are required.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from collections.abc import AsyncIterator
|
||||
from pathlib import Path
|
||||
from typing import Any, Sequence
|
||||
|
||||
import httpx
|
||||
|
||||
from openjarvis.core.types import Message
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Key / provider detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CLOUD_ENV_FILE = Path.home() / ".openjarvis" / "cloud-keys.env"
|
||||
|
||||
_OPENAI_PREFIXES = ("gpt-", "o1-", "o3-", "o4-", "chatgpt-")
|
||||
_ANTHROPIC_PREFIXES = ("claude-",)
|
||||
_GOOGLE_PREFIXES = ("gemini-",)
|
||||
_MINIMAX_PREFIXES = ("MiniMax-",)
|
||||
|
||||
|
||||
def _load_keys() -> dict[str, str]:
|
||||
"""Read cloud-keys.env from disk every call so live updates are picked up."""
|
||||
keys: dict[str, str] = {}
|
||||
# File first, then fall back to process environment
|
||||
if _CLOUD_ENV_FILE.exists():
|
||||
for raw in _CLOUD_ENV_FILE.read_text().splitlines():
|
||||
line = raw.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
keys[k.strip()] = v.strip()
|
||||
# Process env can override (e.g. during testing)
|
||||
for name in (
|
||||
"OPENAI_API_KEY",
|
||||
"ANTHROPIC_API_KEY",
|
||||
"GEMINI_API_KEY",
|
||||
"GOOGLE_API_KEY",
|
||||
"OPENROUTER_API_KEY",
|
||||
"MINIMAX_API_KEY",
|
||||
):
|
||||
val = os.environ.get(name)
|
||||
if val:
|
||||
keys[name] = val
|
||||
return keys
|
||||
|
||||
|
||||
def get_provider(model: str) -> str | None:
|
||||
"""Return the provider for a model name, or None if it's a local model."""
|
||||
if any(model.startswith(p) for p in _OPENAI_PREFIXES):
|
||||
return "openai"
|
||||
if any(model.startswith(p) for p in _ANTHROPIC_PREFIXES):
|
||||
return "anthropic"
|
||||
if any(model.startswith(p) for p in _GOOGLE_PREFIXES):
|
||||
return "google"
|
||||
if any(model.startswith(p) for p in _MINIMAX_PREFIXES):
|
||||
return "minimax"
|
||||
if "/" in model: # openrouter format: "meta-llama/llama-3-8b"
|
||||
return "openrouter"
|
||||
return None
|
||||
|
||||
|
||||
def is_cloud_model(model: str) -> bool:
|
||||
"""Return True if the model is served by a cloud provider."""
|
||||
return get_provider(model) is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Message conversion
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _to_openai_msgs(messages: Sequence[Message]) -> list[dict[str, Any]]:
|
||||
out = []
|
||||
for m in messages:
|
||||
role = m.role.value if hasattr(m.role, "value") else str(m.role)
|
||||
out.append({"role": role, "content": m.content or ""})
|
||||
return out
|
||||
|
||||
|
||||
def _to_anthropic_msgs(
|
||||
messages: Sequence[Message],
|
||||
) -> tuple[str, list[dict[str, Any]]]:
|
||||
"""Return (system_text, chat_messages) in Anthropic format."""
|
||||
system_text = ""
|
||||
chat: list[dict[str, Any]] = []
|
||||
for m in messages:
|
||||
role = m.role.value if hasattr(m.role, "value") else str(m.role)
|
||||
if role == "system":
|
||||
system_text = m.content or ""
|
||||
else:
|
||||
# Anthropic only allows "user" and "assistant"
|
||||
ar = "user" if role != "assistant" else "assistant"
|
||||
chat.append({"role": ar, "content": m.content or ""})
|
||||
return system_text, chat
|
||||
|
||||
|
||||
def _to_google_contents(messages: Sequence[Message]) -> list[dict[str, Any]]:
|
||||
"""Convert to Google Gemini content format."""
|
||||
contents = []
|
||||
for m in messages:
|
||||
role = m.role.value if hasattr(m.role, "value") else str(m.role)
|
||||
if role == "system":
|
||||
# Gemini doesn't have a system role in the contents array;
|
||||
# prepend as a user message.
|
||||
contents.append({"role": "user", "parts": [{"text": m.content or ""}]})
|
||||
contents.append({"role": "model", "parts": [{"text": "Understood."}]})
|
||||
elif role == "assistant":
|
||||
contents.append({"role": "model", "parts": [{"text": m.content or ""}]})
|
||||
else:
|
||||
contents.append({"role": "user", "parts": [{"text": m.content or ""}]})
|
||||
return contents
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Streaming generators
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _stream_openai(
|
||||
model: str,
|
||||
messages: Sequence[Message],
|
||||
temperature: float,
|
||||
max_tokens: int,
|
||||
base_url: str = "https://api.openai.com/v1",
|
||||
api_key_name: str = "OPENAI_API_KEY",
|
||||
) -> AsyncIterator[str]:
|
||||
keys = _load_keys()
|
||||
api_key = keys.get(api_key_name, "")
|
||||
if not api_key:
|
||||
raise ValueError(f"{api_key_name} not set — add it in the Cloud Models tab")
|
||||
|
||||
payload = {
|
||||
"model": model,
|
||||
"messages": _to_openai_msgs(messages),
|
||||
"temperature": temperature,
|
||||
"max_tokens": max_tokens,
|
||||
"stream": True,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=180) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
f"{base_url}/chat/completions",
|
||||
json=payload,
|
||||
headers={
|
||||
"Authorization": f"Bearer {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data: "):
|
||||
continue
|
||||
data = line[6:].strip()
|
||||
if data == "[DONE]":
|
||||
break
|
||||
try:
|
||||
chunk = json.loads(data)
|
||||
delta = chunk["choices"][0]["delta"].get("content") or ""
|
||||
if delta:
|
||||
yield delta
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def _stream_anthropic(
|
||||
model: str,
|
||||
messages: Sequence[Message],
|
||||
temperature: float,
|
||||
max_tokens: int,
|
||||
) -> AsyncIterator[str]:
|
||||
keys = _load_keys()
|
||||
api_key = keys.get("ANTHROPIC_API_KEY", "")
|
||||
if not api_key:
|
||||
raise ValueError("ANTHROPIC_API_KEY not set — add it in the Cloud Models tab")
|
||||
|
||||
system_text, chat_msgs = _to_anthropic_msgs(messages)
|
||||
payload: dict[str, Any] = {
|
||||
"model": model,
|
||||
"messages": chat_msgs,
|
||||
"max_tokens": max_tokens,
|
||||
"temperature": temperature,
|
||||
"stream": True,
|
||||
}
|
||||
if system_text:
|
||||
payload["system"] = system_text
|
||||
|
||||
async with httpx.AsyncClient(timeout=180) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
json=payload,
|
||||
headers={
|
||||
"x-api-key": api_key,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data: "):
|
||||
continue
|
||||
data = line[6:].strip()
|
||||
try:
|
||||
event = json.loads(data)
|
||||
if event.get("type") == "content_block_delta":
|
||||
text = event.get("delta", {}).get("text", "")
|
||||
if text:
|
||||
yield text
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def _stream_google(
|
||||
model: str,
|
||||
messages: Sequence[Message],
|
||||
temperature: float,
|
||||
max_tokens: int,
|
||||
) -> AsyncIterator[str]:
|
||||
keys = _load_keys()
|
||||
api_key = keys.get("GEMINI_API_KEY") or keys.get("GOOGLE_API_KEY", "")
|
||||
if not api_key:
|
||||
raise ValueError("GEMINI_API_KEY not set — add it in the Cloud Models tab")
|
||||
|
||||
contents = _to_google_contents(messages)
|
||||
payload: dict[str, Any] = {
|
||||
"contents": contents,
|
||||
"generationConfig": {
|
||||
"temperature": temperature,
|
||||
"maxOutputTokens": max_tokens,
|
||||
},
|
||||
}
|
||||
|
||||
url = (
|
||||
f"https://generativelanguage.googleapis.com/v1beta/models/"
|
||||
f"{model}:streamGenerateContent?alt=sse&key={api_key}"
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient(timeout=180) as client:
|
||||
async with client.stream("POST", url, json=payload) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if not line.startswith("data: "):
|
||||
continue
|
||||
data = line[6:].strip()
|
||||
try:
|
||||
chunk = json.loads(data)
|
||||
parts = (
|
||||
chunk.get("candidates", [{}])[0]
|
||||
.get("content", {})
|
||||
.get("parts", [])
|
||||
)
|
||||
for part in parts:
|
||||
text = part.get("text", "")
|
||||
if text:
|
||||
yield text
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Local (Ollama) direct streaming — bypasses engine routing entirely
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _ollama_host() -> str:
|
||||
return os.environ.get("OLLAMA_HOST", "http://localhost:11434").rstrip("/")
|
||||
|
||||
|
||||
async def stream_local(
|
||||
model: str,
|
||||
messages: Sequence[Message],
|
||||
temperature: float = 0.7,
|
||||
max_tokens: int = 1024,
|
||||
) -> AsyncIterator[str]:
|
||||
"""Stream tokens directly from Ollama, bypassing the engine system."""
|
||||
payload = {
|
||||
"model": model,
|
||||
"messages": _to_openai_msgs(messages),
|
||||
"stream": True,
|
||||
# Disable extended thinking (Qwen3.5 etc.) — when enabled all tokens
|
||||
# go into the 'thinking' field and 'content' stays empty.
|
||||
"think": False,
|
||||
"options": {
|
||||
"temperature": temperature,
|
||||
"num_predict": max_tokens,
|
||||
},
|
||||
}
|
||||
host = _ollama_host()
|
||||
async with httpx.AsyncClient(timeout=300) as client:
|
||||
async with client.stream("POST", f"{host}/api/chat", json=payload) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
data = json.loads(line)
|
||||
token = data.get("message", {}).get("content", "")
|
||||
if token:
|
||||
yield token
|
||||
if data.get("done"):
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def list_local_models() -> list[str]:
|
||||
"""Return Ollama model names directly from the Ollama API."""
|
||||
host = _ollama_host()
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as client:
|
||||
resp = await client.get(f"{host}/api/tags")
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return [m["name"] for m in data.get("models", [])]
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def stream_cloud(
|
||||
model: str,
|
||||
messages: Sequence[Message],
|
||||
temperature: float = 0.7,
|
||||
max_tokens: int = 1024,
|
||||
) -> AsyncIterator[str]:
|
||||
"""Stream tokens from a cloud provider for the given model."""
|
||||
provider = get_provider(model)
|
||||
|
||||
if provider == "openai":
|
||||
async for token in _stream_openai(model, messages, temperature, max_tokens):
|
||||
yield token
|
||||
|
||||
elif provider == "anthropic":
|
||||
async for token in _stream_anthropic(model, messages, temperature, max_tokens):
|
||||
yield token
|
||||
|
||||
elif provider == "google":
|
||||
async for token in _stream_google(model, messages, temperature, max_tokens):
|
||||
yield token
|
||||
|
||||
elif provider == "openrouter":
|
||||
keys = _load_keys()
|
||||
api_key = keys.get("OPENROUTER_API_KEY", "")
|
||||
if not api_key:
|
||||
raise ValueError(
|
||||
"OPENROUTER_API_KEY not set — add it in the Cloud Models tab"
|
||||
)
|
||||
async for token in _stream_openai(
|
||||
model,
|
||||
messages,
|
||||
temperature,
|
||||
max_tokens,
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
api_key_name="OPENROUTER_API_KEY",
|
||||
):
|
||||
yield token
|
||||
|
||||
elif provider == "minimax":
|
||||
keys = _load_keys()
|
||||
api_key = keys.get("MINIMAX_API_KEY", "")
|
||||
if not api_key:
|
||||
raise ValueError("MINIMAX_API_KEY not set — add it in the Cloud Models tab")
|
||||
async for token in _stream_openai(
|
||||
model,
|
||||
messages,
|
||||
temperature,
|
||||
max_tokens,
|
||||
base_url="https://api.minimax.io/v1",
|
||||
api_key_name="MINIMAX_API_KEY",
|
||||
):
|
||||
yield token
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown cloud provider for model: {model!r}")
|
||||
+126
-20
@@ -290,9 +290,19 @@ async def _handle_stream(
|
||||
complexity_info=None,
|
||||
):
|
||||
"""Stream response using SSE format."""
|
||||
from openjarvis.server.cloud_router import (
|
||||
is_cloud_model,
|
||||
stream_cloud,
|
||||
stream_local,
|
||||
)
|
||||
|
||||
messages = _to_messages(req.messages)
|
||||
chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
|
||||
|
||||
# Route directly to the right backend — bypasses engine routing entirely
|
||||
# so broken MultiEngine state can never misdirect requests.
|
||||
use_cloud = is_cloud_model(model)
|
||||
|
||||
async def generate():
|
||||
# Send role chunk first
|
||||
first_chunk = ChatCompletionChunk(
|
||||
@@ -307,13 +317,45 @@ async def _handle_stream(
|
||||
yield f"data: {first_chunk.model_dump_json()}\n\n"
|
||||
|
||||
try:
|
||||
# Stream content
|
||||
async for token in engine.stream(
|
||||
messages,
|
||||
model=model,
|
||||
temperature=req.temperature,
|
||||
max_tokens=req.max_tokens,
|
||||
):
|
||||
# Cloud models → direct cloud API (reads keys from disk).
|
||||
# Local models → engine.stream() first so mock engines work in
|
||||
# tests. Fall back to stream_local() only when the engine would
|
||||
# mis-route the request to a cloud backend (MultiEngine routing
|
||||
# confusion), which is detected by checking the routed engine's
|
||||
# is_cloud attribute.
|
||||
if use_cloud:
|
||||
token_iter = stream_cloud(
|
||||
model, messages, req.temperature, req.max_tokens
|
||||
)
|
||||
else:
|
||||
# Use engine.stream() by default (preserves mock-engine
|
||||
# compatibility in tests). Only fall back to stream_local()
|
||||
# when a real MultiEngine would mis-route the local model to a
|
||||
# cloud backend — detected via isinstance so mocks are not
|
||||
# accidentally matched.
|
||||
_use_local_fallback = False
|
||||
try:
|
||||
from openjarvis.engine.multi import MultiEngine
|
||||
|
||||
_inner = getattr(engine, "_inner", engine)
|
||||
if isinstance(_inner, MultiEngine):
|
||||
_routed = _inner._engine_for(model)
|
||||
if _routed is not None and getattr(_routed, "is_cloud", False):
|
||||
_use_local_fallback = True
|
||||
except Exception:
|
||||
pass
|
||||
if _use_local_fallback:
|
||||
token_iter = stream_local(
|
||||
model, messages, req.temperature, req.max_tokens
|
||||
)
|
||||
else:
|
||||
token_iter = engine.stream(
|
||||
messages,
|
||||
model=model,
|
||||
temperature=req.temperature,
|
||||
max_tokens=req.max_tokens,
|
||||
)
|
||||
async for token in token_iter:
|
||||
chunk = ChatCompletionChunk(
|
||||
id=chunk_id,
|
||||
model=model,
|
||||
@@ -365,17 +407,11 @@ async def _handle_stream(
|
||||
)
|
||||
finish_dict = _json.loads(finish_data.model_dump_json())
|
||||
|
||||
# Pull usage from the engine if it tracked it during streaming
|
||||
raw_engine = engine
|
||||
# Unwrap InstrumentedEngine if present
|
||||
if hasattr(raw_engine, "_inner"):
|
||||
raw_engine = raw_engine._inner
|
||||
# Unwrap MultiEngine if present
|
||||
if hasattr(raw_engine, "_engine_for"):
|
||||
raw_engine = raw_engine._engine_for(model)
|
||||
stream_usage = getattr(raw_engine, "_last_stream_usage", None)
|
||||
if isinstance(stream_usage, dict) and stream_usage.get("total_tokens", 0) > 0:
|
||||
finish_dict["usage"] = stream_usage
|
||||
# Tag the finish chunk with the correct engine label.
|
||||
# We use the routing decision (use_cloud) directly rather than
|
||||
# unwrapping the engine chain, which can be in a broken state.
|
||||
finish_dict.setdefault("telemetry", {})
|
||||
finish_dict["telemetry"]["engine"] = "cloud" if use_cloud else "ollama"
|
||||
|
||||
if complexity_info is not None:
|
||||
finish_dict["complexity"] = complexity_info.model_dump()
|
||||
@@ -392,9 +428,22 @@ async def _handle_stream(
|
||||
|
||||
@router.get("/v1/models")
|
||||
async def list_models(request: Request) -> ModelListResponse:
|
||||
"""List available models from the engine."""
|
||||
"""List locally installed models (Ollama).
|
||||
|
||||
Cloud models are not included here — they live in the Cloud Models tab
|
||||
of the UI and are selected there, not from this endpoint.
|
||||
"""
|
||||
from openjarvis.server.cloud_router import is_cloud_model, list_local_models
|
||||
|
||||
# Prefer engine.list_models() so mock engines work in tests.
|
||||
# Filter out any cloud model IDs that may appear via MultiEngine.
|
||||
# Fall back to direct Ollama query only when the engine returns nothing.
|
||||
engine = request.app.state.engine
|
||||
model_ids = engine.list_models()
|
||||
all_ids = engine.list_models()
|
||||
model_ids = [m for m in all_ids if not is_cloud_model(m)]
|
||||
if not model_ids:
|
||||
model_ids = await list_local_models()
|
||||
|
||||
return ModelListResponse(
|
||||
data=[ModelObject(id=mid) for mid in model_ids],
|
||||
)
|
||||
@@ -472,6 +521,63 @@ async def delete_model(model_name: str, request: Request):
|
||||
return {"status": "deleted", "model": model_name}
|
||||
|
||||
|
||||
@router.post("/v1/cloud/reload")
|
||||
async def reload_cloud_engine(request: Request):
|
||||
"""Hot-reload cloud API keys and (re-)initialize the cloud engine.
|
||||
|
||||
Called by the desktop app immediately after the user saves a cloud API
|
||||
key so that cloud models become available without a full app restart.
|
||||
"""
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# Re-read ~/.openjarvis/cloud-keys.env and update the running process env.
|
||||
keys_path = Path.home() / ".openjarvis" / "cloud-keys.env"
|
||||
if keys_path.exists():
|
||||
for raw_line in keys_path.read_text().splitlines():
|
||||
line = raw_line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
os.environ[k.strip()] = v.strip()
|
||||
|
||||
# Try to build a fresh CloudEngine.
|
||||
try:
|
||||
from openjarvis.engine.cloud import CloudEngine
|
||||
from openjarvis.engine.multi import MultiEngine
|
||||
|
||||
cloud = CloudEngine()
|
||||
if not cloud.health():
|
||||
return {
|
||||
"status": "no_cloud",
|
||||
"message": "No cloud models available (check API keys)",
|
||||
}
|
||||
except Exception as exc:
|
||||
return {"status": "error", "message": str(exc)}
|
||||
|
||||
# Locate the innermost engine, working through InstrumentedEngine layers.
|
||||
outer = request.app.state.engine
|
||||
inner = getattr(outer, "_inner", outer)
|
||||
|
||||
if isinstance(inner, MultiEngine):
|
||||
# Replace or insert the cloud entry in the existing MultiEngine.
|
||||
new_engines = [(k, e) for k, e in inner._engines if k != "cloud"]
|
||||
new_engines.append(("cloud", cloud))
|
||||
inner._engines = new_engines
|
||||
inner._refresh_map()
|
||||
else:
|
||||
# Wrap the existing engine (which may be security-wrapped) with a new
|
||||
# MultiEngine that includes the cloud engine.
|
||||
engine_name = getattr(request.app.state, "engine_name", "local")
|
||||
new_multi = MultiEngine([(engine_name, inner), ("cloud", cloud)])
|
||||
if hasattr(outer, "_inner"):
|
||||
outer._inner = new_multi
|
||||
else:
|
||||
request.app.state.engine = new_multi
|
||||
request.app.state.engine_name = "multi"
|
||||
|
||||
return {"status": "ok", "message": "Cloud engine reloaded"}
|
||||
|
||||
|
||||
@router.get("/v1/savings")
|
||||
async def savings(request: Request):
|
||||
"""Return savings summary compared to cloud providers.
|
||||
|
||||
Reference in New Issue
Block a user