feat: switch LLM to OpenWebUI with live model selection
Rebuild & Deploy / rebuild (push) Successful in 3m18s

- Point the agent at OpenWebUI's OpenAI-compatible API (LLM_BASE_URL,
  LLM_API_KEY, LLM_MODEL) with GEMMA_* env fallbacks
- Add list_available_models() fetching {base_url}/models, filtering
  non-chat ids (embedding/reranker)
- Rename GemmaLLM -> HopeLLM with update_options(model=...) for runtime
  model switching (mutates _opts.model) and instance-based compaction
- Data channel: agent publishes {type: models} after connect and on
  get_models; set_model switches live and acks model_changed
- Web UI: Brain section in the settings sheet with a model picker that
  persists the choice in localStorage
- Update tests, CI workflow env, and docs for the new endpoint
This commit is contained in:
snowspeeder committed 2026-08-29 12:09:03 -04:00
1 parent 024c42ffb1
commit 360e3276b0
14 files changed
+334 -76

No files matched your search

+140 -24
View File
@@ -1,7 +1,7 @@
"""
Voice Agent — real-time voice assistant.
Pipeline: Azure STT → Gemma LLM (xNAS, OpenAI-compatible) → Azure TTS
Pipeline: Azure STT → LLM (OpenWebUI, OpenAI-compatible) → Azure TTS
Runs inside the single Docker container alongside LiveKit server and web frontend.
"""
@@ -37,9 +37,23 @@ AZURE_REGION = os.environ.get("AZURE_SPEECH_REGION", "eastus")
DEFAULT_VOICE = os.environ.get(
"AZURE_TTS_VOICE", "en-US-Ava:DragonHDOmniLatestNeural"
)
GEMMA_BASE_URL = os.environ.get("GEMMA_BASE_URL", "http://192.168.86.2:8023/v1")
GEMMA_MODEL = os.environ.get("GEMMA_MODEL", "gemma-4-e4b")
GEMMA_API_KEY = os.environ.get("GEMMA_API_KEY", "not-needed")
LLM_BASE_URL = (
os.environ.get("LLM_BASE_URL")
or os.environ.get("GEMMA_BASE_URL")
or "http://192.168.86.2:3005/api"
)
LLM_MODEL = (
os.environ.get("LLM_MODEL")
or os.environ.get("GEMMA_MODEL")
or "gemma-4-e4b"
)
LLM_API_KEY = (
os.environ.get("LLM_API_KEY")
or os.environ.get("GEMMA_API_KEY")
or "not-needed"
)
# Model ids that cannot hold a conversation (OpenWebUI hosts these too).
NON_CHAT_MODEL_ID = re.compile(r"embed|rerank|whisper|tts-|stable-?diffusion|flux", re.I)
WEB_MCP_ENABLED = os.environ.get("WEB_MCP_ENABLED", "true").lower() in ("1", "true", "yes")
FIRECRAWL_BASE = os.environ.get("FIRECRAWL_BASE", "http://192.168.86.2:3002")
SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills")
@@ -67,6 +81,39 @@ def user_id_from_room(room_name: str) -> str:
return match.group(1)
async def list_available_models() -> list[str]:
"""Fetch chat-capable model ids from the OpenAI-compatible endpoint.
Works with OpenWebUI (GET /api/models) and plain OpenAI-compatible
servers (GET /v1/models) — both live under {base_url}/models. The
configured default model is always first so it survives fetch failures.
"""
ids: list[str] = []
try:
import httpx
headers = {}
if LLM_API_KEY and LLM_API_KEY != "not-needed":
headers["Authorization"] = f"Bearer {LLM_API_KEY}"
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(f"{LLM_BASE_URL}/models", headers=headers)
resp.raise_for_status()
data = resp.json()
ids = [
model_id
for model in data.get("data", [])
if isinstance(model, dict) and (model_id := model.get("id", ""))
]
except Exception as e: # noqa: BLE001
logger.warning("Could not list models from %s: %s", LLM_BASE_URL, e)
chat_models = [m for m in ids if not NON_CHAT_MODEL_ID.search(m)]
if LLM_MODEL in chat_models:
chat_models.remove(LLM_MODEL)
ordered = [LLM_MODEL] + sorted(chat_models)
return ordered or [LLM_MODEL]
SYSTEM_PROMPT = textwrap.dedent("""\
You are Hope, a warm, conversational voice assistant speaking with
{user_display_name}. You are talking TO someone, not writing for them to
@@ -167,14 +214,15 @@ except ImportError:
from task_router import should_force_background_task
# ── Gemma LLM with reasoning_content fallback ───────────────────────────────
# ── OpenAI-compatible LLM with reasoning_content fallback ───────────────────
class _ReasoningFallbackWrapper:
"""Wraps an LLMStream to fall back to ``reasoning_content`` when the model
finishes a turn with empty visible content (Gemma spends its whole budget
on hidden reasoning). Delegates all iteration to the underlying stream and
injects a final content chunk if no real content was produced."""
finishes a turn with empty visible content (reasoning models such as Gemma
spend their whole budget on hidden reasoning). Delegates all iteration to
the underlying stream and injects a final content chunk if no real content
was produced."""
def __init__(self, inner: lk_llm.LLMStream) -> None:
self._inner = inner
@@ -257,25 +305,42 @@ class _ForcedToolCallStream(lk_llm.LLMStream):
)
class GemmaLLM(openai.LLM):
"""OpenAI-compatible LLM (llama.cpp Gemma 4) with thinking disabled."""
class HopeLLM(openai.LLM):
"""OpenAI-compatible LLM (OpenWebUI) with runtime model switching."""
# Context compaction: when the conversation grows past this many items,
# ask Gemma to summarize the older messages into a compact recap, then
# ask the LLM to summarize the older messages into a compact recap, then
# replace them with that summary. Keeps recent turns verbatim.
COMPACTION_THRESHOLD = 24 # ~12 turns of user/agent exchange
KEEP_RECENT = 10 # last N items kept verbatim after compaction
IDLE_COMPACTION_SECONDS = 300 # also compact after 5 min of silence
def __init__(self, *args, **kwargs):
def __init__(self, *, model: str, **kwargs):
# llama.cpp's Gemma tool template ignores named tool_choice when OpenAI
# strict schemas are enabled. Non-strict schemas make forced dispatch
# (and ordinary MCP calls) work reliably.
kwargs.setdefault("_strict_tool_schema", False)
super().__init__(*args, **kwargs)
self._model = model
super().__init__(model=model, **kwargs)
import time
self._last_call_time: float = time.monotonic()
@property
def model(self) -> str:
"""Currently selected model id (openai.LLM exposes the same property)."""
return self._model
def update_options(self, *, model: str | None = None) -> None:
"""Switch the backing model for all subsequent chat calls.
The base class re-reads ``self._opts`` on every request, so mutating
the opts dataclass is enough — no client rebuild needed.
"""
if model and model != self._model:
logger.info("LLM model change: %s -> %s", self._model, model)
self._model = model
self._opts.model = model
def chat(self, *, chat_ctx, tools=None, conn_options=None, **kwargs):
if conn_options is None:
from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS
@@ -303,7 +368,8 @@ class GemmaLLM(openai.LLM):
chat_ctx = self._compact_context(chat_ctx, conn_options)
# Force dispatch for clear background-work requests. Without this,
# Gemma often says "I'll get on that" but emits no tool call at all.
# small local models often say "I'll get on that" but emit no tool
# call at all.
items = list(chat_ctx.items)
latest = items[-1] if items else None
latest_role = getattr(latest, "role", None)
@@ -333,7 +399,7 @@ class GemmaLLM(openai.LLM):
"""Summarize older messages via a side LLM call, keep recent ones.
Splits the context into [system | old | recent]. The old portion is
summarized by Gemma into a short recap message that replaces it.
summarized by the LLM into a short recap message that replaces it.
Falls back to hard truncation if the summary call fails.
"""
from livekit.agents import llm as lk_llm
@@ -361,7 +427,7 @@ class GemmaLLM(openai.LLM):
if not transcript.strip():
return chat_ctx # nothing meaningful to summarize
# Side call to Gemma for the summary (no tools, no streaming needed)
# Side call to the LLM for the summary (no tools, no streaming needed)
try:
import httpx
@@ -371,11 +437,14 @@ class GemmaLLM(openai.LLM):
"Be concise — this replaces the full history as context.\n\n"
+ transcript
)
headers = {"Content-Type": "application/json"}
if LLM_API_KEY and LLM_API_KEY != "not-needed":
headers["Authorization"] = f"Bearer {LLM_API_KEY}"
resp = httpx.post(
f"{GEMMA_BASE_URL}/chat/completions",
headers={"Content-Type": "application/json"},
f"{LLM_BASE_URL}/chat/completions",
headers=headers,
json={
"model": GEMMA_MODEL,
"model": self._model,
"messages": [
{"role": "system", "content": "You summarize conversations concisely."},
{"role": "user", "content": summary_prompt},
@@ -554,10 +623,10 @@ class VoiceAssistant(Agent):
display_name = user_profile.get("display_name") or user_profile["user_id"]
instructions = SYSTEM_PROMPT.replace("{user_display_name}", display_name)
super().__init__(
llm=GemmaLLM(
model=GEMMA_MODEL,
base_url=GEMMA_BASE_URL,
api_key=GEMMA_API_KEY,
llm=HopeLLM(
model=LLM_MODEL,
base_url=LLM_BASE_URL,
api_key=LLM_API_KEY,
max_completion_tokens=1000,
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
),
@@ -613,6 +682,27 @@ async def handle_job(ctx: JobContext) -> None:
),
)
# The assistant owns the LLM instance; data-channel handlers below reach
# into it for live model switching.
voice_assistant = VoiceAssistant(user_profile, task_dir)
# Publish the chat-capable model list so the web UI can offer a picker.
async def _publish_models() -> None:
models = await list_available_models()
payload = json.dumps(
{
"type": "models",
"models": models,
"selected": voice_assistant.llm.model,
}
)
try:
await ctx.room.local_participant.publish_data(
payload, reliable=True, topic="voice-control"
)
except Exception as e: # noqa: BLE001
logger.warning("failed to publish model list: %s", e)
# Listen for conversation controls (voice selection and voice/text mode).
# Defined here so it captures this job's session via closure.
def _on_room_data(packet) -> None:
@@ -650,6 +740,29 @@ async def handle_job(ctx: JobContext) -> None:
except Exception as e: # noqa: BLE001
logger.warning("Failed to update voice: %s", e)
if msg.get("type") == "get_models":
asyncio.create_task(_publish_models())
return
if msg.get("type") == "set_model":
model = msg.get("model", "")
if not model:
return
try:
llm = voice_assistant.llm
if hasattr(llm, "update_options"):
llm.update_options(model=model)
ack = json.dumps(
{"type": "model_changed", "model": llm.model}
)
asyncio.create_task(
ctx.room.local_participant.publish_data(
ack, reliable=True, topic="voice-control"
)
)
except Exception as e: # noqa: BLE001
logger.warning("Failed to update model: %s", e)
ctx.room.on("data_received", _on_room_data)
# Publish user/agent transcripts to the room so the web UI can render them.
@@ -743,7 +856,7 @@ async def handle_job(ctx: JobContext) -> None:
ctx.add_shutdown_callback(_stop_task_poller)
await session.start(
agent=VoiceAssistant(user_profile, task_dir),
agent=voice_assistant,
room=ctx.room,
room_options=room_io.RoomOptions(
# Keep the agent in the room when a participant leaves; it must
@@ -769,6 +882,9 @@ async def handle_job(ctx: JobContext) -> None:
)
except Exception as exc: # noqa: BLE001
logger.warning("failed to publish agent readiness: %s", exc)
# Announce which models the UI can switch between (selected one included).
await _publish_models()
logger.info("Agent connected to room %s", ctx.room.name)
+16 -4
View File
@@ -1,7 +1,7 @@
"""Background task worker — runs an autonomous LLM loop with tools.
Each dispatched task gets its own TaskWorker instance running in a separate
asyncio task. The worker calls Gemma iteratively, executing tool calls and
asyncio task. The worker calls the LLM iteratively, executing tool calls and
logging each step to the registry so the UI can display live progress.
"""
from __future__ import annotations
@@ -31,9 +31,21 @@ class TaskWorker:
except ImportError:
from task_registry import registry as reg
base_url = os.environ.get("GEMMA_BASE_URL", "http://192.168.86.2:8023/v1")
model = os.environ.get("GEMMA_MODEL", "gemma-4-e4b")
api_key = os.environ.get("GEMMA_API_KEY", "not-needed")
base_url = (
os.environ.get("LLM_BASE_URL")
or os.environ.get("GEMMA_BASE_URL")
or "http://192.168.86.2:3005/api"
)
model = (
os.environ.get("LLM_MODEL")
or os.environ.get("GEMMA_MODEL")
or "gemma-4-e4b"
)
api_key = (
os.environ.get("LLM_API_KEY")
or os.environ.get("GEMMA_API_KEY")
or "not-needed"
)
reg.add_step(self.task_id, "thinking", f"Starting: {self.description}")