Shane
44e8d05e2c
- agent/task_registry.py: file-based JSONL event registry (cross-process) - agent/task_worker.py: autonomous LLM loop with weather/time/memory/web tools - agent/dispatch_mcp.py: MCP tool exposing dispatch_task to the main agent - agent/agent.py: registers dispatch toolset, polls task events → room data - web: slide-out task panel (FAB button + badge), live step streaming via data channel topic 'tasks', status dots (running/completed/failed) - Dockerfile: copies new task_*.py and dispatch_mcp.py files The dispatch MCP runs in its own process; events flow through /tmp/tasks/events.jsonl which the main agent tails every second and forwards to the browser. Tasks run up to 10 LLM iterations with tool calls.
244 lines
9.6 KiB
Python
244 lines
9.6 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)
|
|
tools = self._load_tools()
|
|
|
|
messages = [
|
|
{"role": "system", "content": self._system_prompt()},
|
|
{"role": "user", "content": self.description},
|
|
]
|
|
|
|
max_iterations = 10
|
|
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=500,
|
|
temperature=0.7,
|
|
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
|
|
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})")
|
|
|
|
def _system_prompt(self) -> str:
|
|
return (
|
|
"You are Hope's research assistant working on a background task. "
|
|
"Use your tools to gather information thoroughly. When you have "
|
|
"enough, provide a final answer in 2-4 sentences suitable for "
|
|
"speaking aloud. Plain text only, 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 using Firecrawl.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
},
|
|
},
|
|
},
|
|
]
|
|
|
|
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": 3},
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
items = data.get("data", [])
|
|
lines = [
|
|
f"{r.get('title', '')}: {r.get('description', '')}"
|
|
for r in items[:3]
|
|
]
|
|
return "\n".join(lines) or "No results found."
|
|
return f"Search failed (HTTP {resp.status_code})."
|
|
|
|
else:
|
|
return f"Unknown tool: {name}"
|
|
|
|
except Exception as e:
|
|
return f"Tool error: {e}"
|
|
|
|
def cancel(self):
|
|
self._cancelled = True
|