diff --git a/agent/agent.py b/agent/agent.py index 4ee50a7..26a71e8 100644 --- a/agent/agent.py +++ b/agent/agent.py @@ -79,19 +79,18 @@ SYSTEM_PROMPT = textwrap.dedent("""\ Use get_time when the user asks what time or day it is. If they mention a city, pass it as the location argument. - # Memory - You have persistent memory across conversations, stored as notes you can - search and add to. - - Use memory_recall at the start of a conversation, or whenever the user - references past information ("what did I tell you about...", "remember - when..."). Weave what you find in naturally. - - Use memory_save when the user shares personal information, preferences, - or important facts worth remembering: names, birthdays, preferences, - projects, anything they'd expect you to know later. Pick a short topic - name for each thing you save. - - Be natural about it. Never announce "I'm saving that to memory" — just - remember it and move on. If a recall comes up empty, don't mention the - search; just answer as if you'd never heard it before. + # Memory (CRITICAL — always use these tools) + You MUST call memory_save whenever the user tells you something to remember, + shares a preference, or says "remember that...". Do NOT just say "okay I'll + remember that" without actually calling the tool. Always call it. + - User says "remember my coffee order is oat milk latte" → CALL memory_save + with topic="coffee order", content="oat milk latte" + - User says "what did I tell you about my birthday?" → CALL memory_recall + with query="birthday" + - User says "list everything you remember about me" → CALL memory_list + After calling memory_save, confirm briefly ("Got it, I'll remember that") + but do NOT say "I saved it to my memory file" or similar. + If a recall returns nothing, just say you don't have that info yet. # Skills You have a skill library of learned procedures. Use skill_recall when you need diff --git a/agent/task_registry.py b/agent/task_registry.py new file mode 100644 index 0000000..d722367 --- /dev/null +++ b/agent/task_registry.py @@ -0,0 +1,126 @@ +"""In-memory task registry for background dispatch tasks. + +One global instance per agent process. Workers add steps as they progress; +the agent subscribes to push updates to the room (UI) and to speak results. +""" +from __future__ import annotations + +import asyncio +import time +import uuid +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable + + +class TaskStatus(str, Enum): + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + + +@dataclass +class TaskStep: + role: str # "thinking", "tool_call", "tool_result", "text" + content: str + timestamp: float = field(default_factory=time.time) + + +@dataclass +class Task: + id: str + description: str + status: TaskStatus = TaskStatus.PENDING + steps: list[TaskStep] = field(default_factory=list) + result: str | None = None + error: str | None = None + created_at: float = field(default_factory=time.time) + completed_at: float | None = None + + +class TaskRegistry: + """Thread-safe registry of background tasks with pub/sub notifications.""" + + def __init__(self): + self._tasks: dict[str, Task] = {} + self._listeners: list[Callable[[str, dict], Any]] = [] + self._lock = asyncio.Lock() + + async def create(self, description: str) -> Task: + task = Task(id=uuid.uuid4().hex[:8], description=description) + async with self._lock: + self._tasks[task.id] = task + await self._notify(task.id, {"event": "created", "task": self._task_dict(task)}) + return task + + async def add_step(self, task_id: str, role: str, content: str): + async with self._lock: + task = self._tasks.get(task_id) + if not task: + return + task.steps.append(TaskStep(role=role, content=content)) + if task.status == TaskStatus.PENDING: + task.status = TaskStatus.RUNNING + await self._notify(task_id, {"event": "step", "role": role, "content": content}) + + async def complete(self, task_id: str, result: str): + async with self._lock: + task = self._tasks.get(task_id) + if not task: + return + task.status = TaskStatus.COMPLETED + task.result = result + task.completed_at = time.time() + await self._notify(task_id, {"event": "completed", "result": result}) + + async def fail(self, task_id: str, error: str): + async with self._lock: + task = self._tasks.get(task_id) + if not task: + return + task.status = TaskStatus.FAILED + task.error = error + task.completed_at = time.time() + await self._notify(task_id, {"event": "failed", "error": error}) + + async def get(self, task_id: str) -> dict | None: + async with self._lock: + t = self._tasks.get(task_id) + return self._task_dict(t) if t else None + + async def list_all(self) -> list[dict]: + async with self._lock: + return [self._task_dict(t) for t in self._tasks.values()] + + def subscribe(self, callback: Callable[[str, dict], Any]): + self._listeners.append(callback) + + async def _notify(self, task_id: str, event: dict): + for cb in self._listeners: + try: + result = cb(task_id, event) + if asyncio.iscoroutine(result): + await result + except Exception: + pass + + @staticmethod + def _task_dict(task: Task) -> dict: + return { + "id": task.id, + "description": task.description, + "status": task.status.value, + "steps": [ + {"role": s.role, "content": s.content, "ts": s.timestamp} + for s in task.steps + ], + "result": task.result, + "error": task.error, + "created_at": task.created_at, + "completed_at": task.completed_at, + } + + +# Global singleton — one per agent process +registry = TaskRegistry() diff --git a/agent/task_worker.py b/agent/task_worker.py new file mode 100644 index 0000000..4dd45b4 --- /dev/null +++ b/agent/task_worker.py @@ -0,0 +1,240 @@ +"""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 + from agent.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") + + await 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: + await reg.fail(self.task_id, "Cancelled") + return + + await 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: + await 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 = {} + await reg.add_step( + self.task_id, + "tool_call", + f"{tool_name}({json.dumps(tool_args)})", + ) + result = await self._execute_tool(tool_name, tool_args) + await 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: + await reg.fail(self.task_id, "LLM returned empty response") + return + await reg.add_step(self.task_id, "text", text) + await reg.complete(self.task_id, text) + return + + await 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