"""Quick-capture intent classifier. Classifies short capture text (note, task, event, research) for the /api/quick-capture endpoint using a dedicated prompt and the primary model. """ import json import logging import re from dataclasses import dataclass, field from datetime import date as date_type from fabledassistant.services.llm import generate_completion logger = logging.getLogger(__name__) @dataclass class IntentResult: tool_name: str | None = None # None = no tool, just chat arguments: dict = field(default_factory=dict) confidence: str = "high" # "high", "medium", or "low" ack: str | None = None # One-sentence acknowledgment to stream immediately @property def should_execute(self) -> bool: """True if a tool was identified with sufficient confidence.""" return self.tool_name is not None and self.confidence != "low" def _build_tool_summary(tools: list[dict]) -> str: """Build a compact tool description string from Ollama tool defs.""" lines: list[str] = [] for tool in tools: fn = tool.get("function", {}) name = fn.get("name", "") desc = fn.get("description", "") params = fn.get("parameters", {}).get("properties", {}) required = set(fn.get("parameters", {}).get("required", [])) param_parts: list[str] = [] for pname, pinfo in params.items(): req = " (required)" if pname in required else "" pdesc = pinfo.get("description", "") param_parts.append(f" - {pname}: {pdesc}{req}") lines.append(f"- {name}: {desc}") lines.extend(param_parts) return "\n".join(lines) def _parse_intent(raw: str, tools: list[dict]) -> IntentResult: """Parse the LLM's JSON response into an IntentResult.""" text = raw.strip() # Strip markdown code fences if present text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text) text = text.strip() # Try direct JSON parse parsed = _try_json(text) # Fallback: extract first JSON object from response if parsed is None: match = re.search(r"\{.*\}", text, re.DOTALL) if match: parsed = _try_json(match.group()) if parsed is None or not isinstance(parsed, dict): logger.warning("Could not parse intent from LLM response: %s", text[:200]) return IntentResult() tool_name = parsed.get("tool") confidence = parsed.get("confidence", "high") if confidence not in ("high", "medium", "low"): confidence = "high" if tool_name is None: return IntentResult(confidence=confidence) # Validate tool name against available tools valid_names = { t.get("function", {}).get("name") for t in tools } if tool_name not in valid_names: logger.warning("Intent returned unknown tool '%s'", tool_name) return IntentResult() arguments = parsed.get("arguments", {}) if not isinstance(arguments, dict): arguments = {} ack = parsed.get("ack") or None if ack is not None: ack = ack.strip() or None logger.info( "Intent classified: tool=%s, confidence=%s, args=%s", tool_name, confidence, json.dumps(arguments)[:200], ) return IntentResult(tool_name=tool_name, arguments=arguments, confidence=confidence, ack=ack) def _try_json(text: str) -> dict | list | None: """Try to parse JSON, return None on failure.""" try: return json.loads(text) except (json.JSONDecodeError, TypeError): return None # ── Quick-capture classifier ────────────────────────────────────────────────── # A stripped-down prompt designed for the /api/quick-capture endpoint. # Unlike the general intent prompt, this ALWAYS routes to a create tool — # null is not a valid response. _CAPTURE_SYSTEM_PROMPT = """\ You are a quick-capture classifier. The user has sent a short snippet of text \ from a mobile app or external client. Classify it as a note, task, or calendar \ event, then extract the relevant fields. Today's date is {today}. Available tools: {tool_summary} Rules: - You MUST choose one of the available tools. Never return null. - create_task: action items, todos, reminders, things to do ("buy milk", "call John", "fix the bug", "remind me to…") - create_event: appointments, meetings, scheduled occurrences with a date/time ("dentist Friday 2pm", "team meeting next Tuesday") - update_note: updating, editing, appending to an existing note or task ("add to my shopping list: eggs", "mark buy milk done", "append to my meeting notes", "update my project note") - research_topic: user wants a comprehensive research note from web sources ("research X", "look up X and make a note", "find everything about X", "compile a note on X") - create_note: everything else — ideas, observations, links, excerpts, longer text - For create_task / create_event: extract a concise title; put any extra detail in "body" - For create_note: use a short descriptive title (≤60 chars); put the FULL original text as "body" - For update_note: set "query" to the note or task title to find; set other fields as needed - For research_topic: set "topic" to the subject being researched - For dates use YYYY-MM-DD; for datetime use ISO 8601 - confidence: "high" if the type is clear; "medium" if you're guessing Respond with ONLY a JSON object: {{"tool": "tool_name", "arguments": {{...}}, "confidence": "high"|"medium"}} Do NOT wrap in markdown code fences.""" async def classify_capture_intent( text: str, tools: list[dict], model: str, ) -> IntentResult: """Classify quick-capture text and extract arguments. Uses a simplified prompt that always routes to a create tool — never null. Returns IntentResult with tool_name set. Falls back to IntentResult() only on LLM/parse failure (caller should handle that case). """ if not tools: return IntentResult() tool_summary = _build_tool_summary(tools) today = date_type.today().isoformat() messages = [ { "role": "system", "content": _CAPTURE_SYSTEM_PROMPT.format( today=today, tool_summary=tool_summary ), }, {"role": "user", "content": text}, ] try: raw = await generate_completion(messages, model, max_tokens=300, num_ctx=2048) except Exception: logger.warning("Quick-capture intent LLM call failed", exc_info=True) return IntentResult() return _parse_intent(raw, tools)