344 lines
14 KiB
Python
344 lines
14 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 Gemma 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
|
|
from datetime import datetime, timezone
|
|
|
|
logger = logging.getLogger("voice-agent.tasks")
|
|
|
|
|
|
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("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")
|
|
|
|
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):
|
|
reg.add_step(
|
|
self.task_id,
|
|
"thinking",
|
|
"Draft deferred the work; 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": (
|
|
"Do not ask me 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 _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 headline "
|
|
"requests, return concrete named events rather than links to news homepages, and "
|
|
"respect any requested item count exactly. 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 _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:
|
|
data = resp.json()
|
|
items = data.get("data", [])
|
|
lines = []
|
|
for item in items[:5]:
|
|
title = (item.get("title") or "Untitled").strip()
|
|
description = (item.get("description") or "").strip()
|
|
url = (item.get("url") or "").strip()
|
|
lines.append(f"{title}: {description}")
|
|
if url:
|
|
lines.append(f"URL: {url}")
|
|
return "\n".join(lines) or "No results found."
|
|
return f"Search failed (HTTP {resp.status_code})."
|
|
|
|
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."
|
|
async with httpx.AsyncClient(timeout=90) as client:
|
|
resp = await client.post(
|
|
f"{firecrawl_base}/v1/scrape",
|
|
json={"url": url, "formats": ["markdown"], "timeout": 60000},
|
|
)
|
|
if resp.status_code != 200:
|
|
return f"Scrape failed (HTTP {resp.status_code})."
|
|
data = resp.json()
|
|
if not data.get("success"):
|
|
return f"Scrape failed: {data.get('error', 'unknown error')}"
|
|
result = data.get("data", {})
|
|
markdown = (result.get("markdown") or "").strip()
|
|
if not markdown:
|
|
return "No page content was extracted."
|
|
title = (result.get("metadata", {}).get("title") or "").strip()
|
|
return f"Page: {title}\n\n{markdown[:8000]}"
|
|
|
|
else:
|
|
return f"Unknown tool: {name}"
|
|
|
|
except Exception as e:
|
|
return f"Tool error: {e}"
|
|
|
|
def cancel(self):
|
|
self._cancelled = True
|