snowspeeder
b01ee6478d
TaskWorker resolved its model from LLM_MODEL env (frozen at MCP spawn), so after a live set_model switch in the UI, dispatched background tasks still asked for the old model. On local-vl-models (llama-swap --models-max 1) that forces a swap against the loaded conversation model; a force-kill mid-load surfaces as 400 'model failed to load'. The agent now writes the selected model to <task_dir>/model.txt at session start and on every switch; TaskWorker reads it with the env value as fallback.
487 lines
20 KiB
Python
487 lines
20 KiB
Python
"""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 the LLM iteratively, executing tool calls and
|
|
logging each step to the registry so the UI can display live progress.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
from datetime import datetime, timezone
|
|
|
|
logger = logging.getLogger("voice-agent.tasks")
|
|
|
|
|
|
def _html_to_text(html: str) -> str:
|
|
"""Strip a raw HTML page to readable plain text (no external deps)."""
|
|
if not html:
|
|
return ""
|
|
# Drop non-content blocks first.
|
|
html = re.sub(
|
|
r"(?is)<(script|style|noscript|head|title|svg|nav|footer|form|iframe)[^>]*>.*?</\1>",
|
|
" ",
|
|
html,
|
|
)
|
|
# Turn block-level tags into paragraph breaks.
|
|
html = re.sub(
|
|
r"(?i)</?(?:p|div|li|ul|ol|h[1-6]|br|tr|td|th|article|section|"
|
|
r"blockquote|header|figure|table|hr)[^>]*>",
|
|
"\n",
|
|
html,
|
|
)
|
|
text = re.sub(r"<[^>]+>", " ", html)
|
|
for ent, ch in (
|
|
(" ", " "), ("&", "&"), ("<", "<"), (">", ">"),
|
|
(""", '"'), ("'", "'"), ("'", "'"), ("'", "'"),
|
|
("…", "..."), ("—", "-"), ("–", "-"),
|
|
):
|
|
text = text.replace(ent, ch)
|
|
text = re.sub(r"[ \t]+", " ", text)
|
|
text = re.sub(r"\n\s*\n+", "\n", text)
|
|
return text.strip()
|
|
|
|
|
|
def _live_model(default: str) -> str:
|
|
"""Follow the conversation's live model selection when it is fresher.
|
|
|
|
The main agent writes the UI-selected model to the session's task dir
|
|
(shared with this MCP subprocess) on every set_model. Env is the
|
|
fallback for tasks dispatched before any switch happened.
|
|
"""
|
|
env = os.environ.get("TASKS_DIR", "/tmp/tasks")
|
|
try:
|
|
with open(os.path.join(env, "model.txt")) as f:
|
|
chosen = f.read().strip()
|
|
except OSError:
|
|
return default
|
|
return chosen or default
|
|
|
|
|
|
class TaskWorker:
|
|
"""Runs a background research/lookup task using the LLM + tools."""
|
|
|
|
def __init__(self, task_id: str, description: str):
|
|
self.task_id = task_id
|
|
self.description = description
|
|
self._cancelled = False
|
|
|
|
async def run(self):
|
|
from openai import AsyncOpenAI
|
|
try:
|
|
from agent.task_registry import registry as reg
|
|
except ImportError:
|
|
from task_registry import registry as reg
|
|
|
|
base_url = (
|
|
os.environ.get("LLM_BASE_URL")
|
|
or os.environ.get("GEMMA_BASE_URL")
|
|
or "http://192.168.86.2:3005/api"
|
|
)
|
|
model = _live_model(
|
|
os.environ.get("LLM_MODEL")
|
|
or os.environ.get("GEMMA_MODEL")
|
|
or "qwen2.5-vl-3b"
|
|
)
|
|
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}")
|
|
|
|
client = AsyncOpenAI(base_url=base_url, api_key=api_key, timeout=60.0)
|
|
tools = self._load_tools()
|
|
|
|
messages = [
|
|
{"role": "system", "content": self._system_prompt()},
|
|
{"role": "user", "content": self.description},
|
|
]
|
|
|
|
max_iterations = 10
|
|
try:
|
|
await self._run_loop(client, model, tools, messages, max_iterations, reg)
|
|
finally:
|
|
await client.close()
|
|
|
|
async def _run_loop(self, client, model, tools, messages, max_iterations, reg):
|
|
for iteration in range(max_iterations):
|
|
if self._cancelled:
|
|
reg.fail(self.task_id, "Cancelled")
|
|
return
|
|
|
|
reg.add_step(
|
|
self.task_id, "thinking", f"Step {iteration + 1}: analyzing..."
|
|
)
|
|
|
|
try:
|
|
response = await client.chat.completions.create(
|
|
model=model,
|
|
messages=messages,
|
|
tools=tools or None,
|
|
max_tokens=700,
|
|
temperature=0.3,
|
|
extra_body={"chat_template_kwargs": {"enable_thinking": False}},
|
|
)
|
|
except Exception as e:
|
|
reg.fail(self.task_id, f"LLM error: {e}")
|
|
return
|
|
|
|
msg = response.choices[0].message
|
|
|
|
if msg.tool_calls:
|
|
messages.append(msg)
|
|
for tc in msg.tool_calls:
|
|
tool_name = tc.function.name
|
|
try:
|
|
tool_args = json.loads(tc.function.arguments)
|
|
except json.JSONDecodeError:
|
|
tool_args = {}
|
|
reg.add_step(
|
|
self.task_id,
|
|
"tool_call",
|
|
f"{tool_name}({json.dumps(tool_args)})",
|
|
)
|
|
result = await self._execute_tool(tool_name, tool_args)
|
|
reg.add_step(self.task_id, "tool_result", str(result)[:500])
|
|
messages.append(
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": tc.id,
|
|
"content": str(result),
|
|
}
|
|
)
|
|
else:
|
|
text = (msg.content or "").strip()
|
|
if not text:
|
|
reg.fail(self.task_id, "LLM returned empty response")
|
|
return
|
|
if (
|
|
self._is_deferred_answer(text)
|
|
or self._is_underfilled_news_answer(text)
|
|
or self._is_underfilled_recipe_answer(text)
|
|
):
|
|
reg.add_step(
|
|
self.task_id,
|
|
"thinking",
|
|
"Draft was incomplete; continuing automatically.",
|
|
)
|
|
if iteration == max_iterations - 1:
|
|
reg.fail(self.task_id, "Could not produce a complete answer")
|
|
return
|
|
messages.extend(
|
|
[
|
|
{"role": "assistant", "content": text},
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"The previous draft did not fully answer the original request. "
|
|
"Do not ask a follow-up or offer to continue later. Continue "
|
|
"the task now, run any more targeted tool calls "
|
|
"you need, make reasonable assumptions, and only finish "
|
|
"when you have directly answered the original request."
|
|
),
|
|
},
|
|
]
|
|
)
|
|
continue
|
|
reg.add_step(self.task_id, "text", text)
|
|
reg.complete(self.task_id, text)
|
|
return
|
|
|
|
reg.fail(self.task_id, f"Reached max iterations ({max_iterations})")
|
|
|
|
@staticmethod
|
|
def _is_deferred_answer(text: str) -> bool:
|
|
"""Reject drafts that ask permission or defer the requested work."""
|
|
lowered = text.lower()
|
|
deferral_phrases = (
|
|
"would you like me",
|
|
"do you want me",
|
|
"should i ",
|
|
"shall i ",
|
|
"if you'd like",
|
|
"if you would like",
|
|
"let me know if",
|
|
"i need to perform",
|
|
"i need more information",
|
|
"i can refine",
|
|
"i can search",
|
|
"i can provide",
|
|
"need a more targeted search",
|
|
)
|
|
return "?" in text or any(phrase in lowered for phrase in deferral_phrases)
|
|
|
|
def _is_underfilled_news_answer(self, text: str) -> bool:
|
|
"""Reject generic news summaries that do not name actual events."""
|
|
description = self.description.lower()
|
|
if "news" not in description and "headline" not in description:
|
|
return False
|
|
lowered = text.lower()
|
|
generic_phrases = (
|
|
"reports also cover global developments",
|
|
"other sources indicate",
|
|
"various international perspectives",
|
|
"ongoing coverage of world events",
|
|
"recommend checking major",
|
|
"unable to retrieve",
|
|
"could not retrieve",
|
|
)
|
|
return any(phrase in lowered for phrase in generic_phrases)
|
|
|
|
def _is_underfilled_recipe_answer(self, text: str) -> bool:
|
|
"""A recipe is not usable without quantities and a method."""
|
|
if "recipe" not in self.description.lower():
|
|
return False
|
|
lowered = text.lower()
|
|
has_quantity = bool(
|
|
re.search(
|
|
r"\b(?:\d+(?:/\d+)?|one|two|three|half|quarter)\s*"
|
|
r"(?:cups?|tablespoons?|tbsp|teaspoons?|tsp|sticks?|ounces?|oz|grams?|g)\b",
|
|
lowered,
|
|
)
|
|
)
|
|
has_method = any(
|
|
verb in lowered
|
|
for verb in ("beat ", "mix ", "whip ", "stir ", "blend ", "combine ")
|
|
)
|
|
return not (has_quantity and has_method)
|
|
|
|
def _system_prompt(self) -> str:
|
|
today = datetime.now(timezone.utc).strftime("%B %d, %Y")
|
|
return (
|
|
"You are Hope's autonomous research assistant working on a background task. "
|
|
f"Today is {today}. Use your tools until you have directly completed the request. "
|
|
"Never ask the user a follow-up question, ask permission to search again, or offer "
|
|
"to finish later; make reasonable assumptions and continue autonomously. If a web "
|
|
"search is broad or inconclusive, immediately run a more targeted query and scrape "
|
|
"the most promising result pages, including today's date and specific reputable "
|
|
"sources when freshness matters. For plural news or headline requests, provide three "
|
|
"distinct concrete events unless the user requests a different count. For three "
|
|
"headlines, write exactly three sentences with one event per sentence and no extras. Never pad the "
|
|
"answer with generic phrases about broad coverage or international perspectives, and "
|
|
"respect any requested item count exactly. For recipes, include specific ingredient "
|
|
"quantities and a concise preparation method so the result is actually usable. Only "
|
|
"produce a final answer once it fulfills the original request. The final answer must "
|
|
"be 1-3 concise sentences suitable for speaking aloud, with plain text only and no "
|
|
"markdown or formatting."
|
|
)
|
|
|
|
def _load_tools(self) -> list[dict]:
|
|
return [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_weather",
|
|
"description": "Get current weather and today's forecast for a location in Fahrenheit.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"location": {"type": "string"}},
|
|
"required": ["location"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "get_time",
|
|
"description": "Get the current date and time, optionally for a specific location.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"location": {"type": "string"}},
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "memory_recall",
|
|
"description": "Search saved memories for relevant information.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"description": "Search the web for current information. Returns titles, descriptions, and URLs.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
},
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_scrape",
|
|
"description": "Read the full content of a promising URL returned by web_search.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"url": {"type": "string"}},
|
|
"required": ["url"],
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
async def _fetch_page_text(self, url: str, timeout: float = 20.0) -> str:
|
|
"""Fetch a page directly from the container and reduce it to plain text.
|
|
|
|
Used as a fallback when Firecrawl's scrape times out or returns nothing.
|
|
The container has outbound internet (the self-hosted Firecrawl does not
|
|
reliably, and its 30s default scrape limit is too short for news sites).
|
|
"""
|
|
if not url:
|
|
return "No URL provided."
|
|
import httpx
|
|
|
|
try:
|
|
async with httpx.AsyncClient(
|
|
timeout=timeout,
|
|
follow_redirects=True,
|
|
headers={
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
|
|
)
|
|
},
|
|
) as client:
|
|
resp = await client.get(url)
|
|
if resp.status_code >= 400:
|
|
return f"Fetch failed (HTTP {resp.status_code})."
|
|
text = _html_to_text(resp.text)
|
|
return text[:8000]
|
|
except httpx.HTTPError as exc:
|
|
return f"Fetch failed: {exc}"
|
|
|
|
async def _execute_tool(self, name: str, args: dict) -> str:
|
|
import httpx
|
|
|
|
try:
|
|
if name == "get_weather":
|
|
location = args.get("location", "")
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.get(
|
|
f"https://wttr.in/{location}?format=j1",
|
|
headers={"User-Agent": "curl/8.0"},
|
|
)
|
|
data = resp.json()
|
|
current = data.get("current_condition", [{}])[0] or {}
|
|
temp_c = current.get("temp_C")
|
|
if temp_c is None:
|
|
return f"Could not get weather for {location}."
|
|
temp_f = round(int(temp_c) * 9 / 5 + 32)
|
|
desc = (current.get("weatherDesc", [{}])[0].get("value") or "unknown").lower()
|
|
feels_c = current.get("FeelsLikeC")
|
|
feels_f = round(int(feels_c) * 9 / 5 + 32) if feels_c else None
|
|
today = (data.get("weather") or [{}])[0] or {}
|
|
high_f = round(int(today["maxtempC"]) * 9 / 5 + 32) if today.get("maxtempC") else None
|
|
low_f = round(int(today["mintempC"]) * 9 / 5 + 32) if today.get("mintempC") else None
|
|
parts = [f"It's {temp_f} degrees and {desc} in {location}"]
|
|
if feels_f and feels_f != temp_f:
|
|
parts[0] += f", feels like {feels_f}"
|
|
if high_f and low_f:
|
|
parts.append(f"High {high_f}, low {low_f}.")
|
|
return " ".join(parts)
|
|
|
|
elif name == "get_time":
|
|
now = datetime.now(timezone.utc)
|
|
loc = args.get("location", "")
|
|
suffix = f" in {loc}" if loc else ", UTC"
|
|
return f"It's {now.strftime('%I:%M %p')} on {now.strftime('%A, %B %d')}{suffix}."
|
|
|
|
elif name == "memory_recall":
|
|
query = args.get("query", "").lower()
|
|
memory_dir = os.environ.get("MEMORY_DIR", "/memory")
|
|
results = []
|
|
try:
|
|
for fname in os.listdir(memory_dir):
|
|
if not fname.endswith(".md"):
|
|
continue
|
|
with open(os.path.join(memory_dir, fname)) as f:
|
|
content = f.read()
|
|
if any(w in content.lower() for w in query.split()):
|
|
results.append(f"[{fname}] {content[:300]}")
|
|
except OSError:
|
|
pass
|
|
return "\n".join(results) or "No memories found."
|
|
|
|
elif name == "web_search":
|
|
query = args.get("query", "")
|
|
firecrawl_base = os.environ.get("FIRECRAWL_BASE", "")
|
|
if not firecrawl_base:
|
|
return "Web search is not configured."
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
resp = await client.post(
|
|
f"{firecrawl_base}/v1/search",
|
|
json={"query": query, "limit": 5},
|
|
)
|
|
if resp.status_code != 200:
|
|
return f"Search failed (HTTP {resp.status_code})."
|
|
data = resp.json()
|
|
items = data.get("data", [])
|
|
if not items:
|
|
return "No results found for that query."
|
|
# Search results only carry titles/descriptions. Fetch the
|
|
# top pages ourselves (container has outbound internet) so the
|
|
# model gets real content instead of just snippets — small
|
|
# models won't reliably chain a second scrape call.
|
|
blocks = []
|
|
for item in items[:3]:
|
|
title = (item.get("title") or "Untitled").strip()
|
|
url = (item.get("url") or "").strip()
|
|
desc = (item.get("description") or "").strip()
|
|
body = await self._fetch_page_text(url) if url else (desc or "No URL.")
|
|
blocks.append(f"{title}\n{url}\n{body[:1800]}")
|
|
return "\n\n---\n\n".join(blocks) or "No results found."
|
|
|
|
elif name == "web_scrape":
|
|
url = args.get("url", "")
|
|
if not url:
|
|
return "A URL is required."
|
|
firecrawl_base = os.environ.get("FIRECRAWL_BASE", "")
|
|
if not firecrawl_base:
|
|
return "Web scraping is not configured."
|
|
# Firecrawl renders JS-heavy pages with a real browser, but its
|
|
# default scrape timeout (~30s) is too short for big news sites.
|
|
# Ask for 90s and give the HTTP client the same headroom; if it
|
|
# returns nothing, fall back to a direct fetch from the container.
|
|
try:
|
|
async with httpx.AsyncClient(timeout=120) as client:
|
|
resp = await client.post(
|
|
f"{firecrawl_base}/v1/scrape",
|
|
json={"url": url, "formats": ["markdown"], "timeout": 90000},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
if data.get("success"):
|
|
result = data.get("data", {})
|
|
markdown = (result.get("markdown") or "").strip()
|
|
if markdown:
|
|
title = (result.get("metadata", {}).get("title") or url).strip()
|
|
return f"Page: {title}\n\n{markdown[:8000]}"
|
|
# fall through to direct fetch on empty/failed scrape
|
|
except (httpx.HTTPError, Exception):
|
|
pass
|
|
body = await self._fetch_page_text(url)
|
|
return body or f"Could not extract content from {url}."
|
|
|
|
else:
|
|
return f"Unknown tool: {name}"
|
|
|
|
except Exception as e:
|
|
return f"Tool error: {e}"
|
|
|
|
def cancel(self):
|
|
self._cancelled = True
|