refactor(quick-capture): replace intent router with native tool-calling
Removes the custom classify_capture_intent + _process_note two-pass approach. The LLM now picks the right tool directly via Ollama's native tool_calls API (same path as the main chat pipeline). _should_think decides whether extended reasoning is needed based on input length/ complexity. intent.py deleted — no longer needed. Android app and response format unchanged. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,77 +1,39 @@
|
||||
"""Quick-capture endpoint for mobile/external clients.
|
||||
|
||||
POST /api/quick-capture — classifies natural-language text and creates the
|
||||
appropriate item (note, task, calendar event, todo) in a single synchronous
|
||||
request. No SSE, no conversation ID, no streaming.
|
||||
POST /api/quick-capture — sends text through the main LLM tool-calling pipeline
|
||||
and returns a single synchronous JSON response. No SSE, no conversation ID.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import date
|
||||
|
||||
from quart import Blueprint, jsonify, request
|
||||
|
||||
from fabledassistant.auth import get_current_user_id, login_required
|
||||
from fabledassistant.config import Config
|
||||
from fabledassistant.services.intent import classify_capture_intent
|
||||
from fabledassistant.services.llm import generate_completion
|
||||
from fabledassistant.services.generation_task import _should_think
|
||||
from fabledassistant.services.llm import stream_chat_with_tools
|
||||
from fabledassistant.services.tools import execute_tool, get_tools_for_user
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
quick_capture_bp = Blueprint("quick_capture", __name__, url_prefix="/api/quick-capture")
|
||||
|
||||
# Tools offered to the quick-capture classifier. Excludes destructive ops
|
||||
# (delete_*) and read-only queries — worst-case fallback is a plain note.
|
||||
# Tools offered to the quick-capture endpoint. Excludes destructive ops,
|
||||
# read-only queries, and conversational-only tools.
|
||||
_CAPTURE_TOOL_NAMES = {"create_note", "create_task", "create_event", "update_note", "research_topic"}
|
||||
|
||||
_NOTE_PROCESS_PROMPT = """\
|
||||
You are a note-taking assistant. The user has sent a quick-capture snippet. \
|
||||
Transform it into a well-formed note.
|
||||
|
||||
Respond with ONLY a JSON object — no other text, no code fences:
|
||||
{{"title": "short descriptive title", "body": "note content in markdown"}}
|
||||
|
||||
Rules:
|
||||
- title: 3–8 words, a genuine summary — do NOT copy the input verbatim
|
||||
- body: process the input thoughtfully:
|
||||
- Lists of items → formatted bullet list
|
||||
- A stream-of-thought or observation → clean prose, lightly organised
|
||||
- Raw notes or fragments → organised paragraphs with a brief intro line
|
||||
- URLs → include the URL and a one-sentence description of what it points to
|
||||
- Preserve ALL information from the original; do not invent new facts
|
||||
- Use markdown formatting (##, -, **, etc.) where it aids readability
|
||||
- Keep it concise — do not pad with filler"""
|
||||
|
||||
|
||||
async def _process_note(text: str, model: str) -> tuple[str, str]:
|
||||
"""Use the main model to transform raw capture text into a title + body.
|
||||
|
||||
Returns (title, body). Falls back to (truncated text, full text) on any failure.
|
||||
"""
|
||||
messages = [
|
||||
{"role": "system", "content": _NOTE_PROCESS_PROMPT},
|
||||
{"role": "user", "content": text},
|
||||
]
|
||||
try:
|
||||
raw = await generate_completion(messages, model, max_tokens=1024, num_ctx=4096)
|
||||
raw = raw.strip()
|
||||
raw = re.sub(r"^```(?:json)?\s*", "", raw)
|
||||
raw = re.sub(r"\s*```$", "", raw).strip()
|
||||
parsed = json.loads(raw)
|
||||
title = str(parsed.get("title", "")).strip() or text[:60]
|
||||
body = str(parsed.get("body", "")).strip() or text
|
||||
return title, body
|
||||
except Exception:
|
||||
logger.warning("Note processing LLM call failed, using raw text", exc_info=True)
|
||||
fallback_title = text if len(text) <= 80 else text[:77] + "..."
|
||||
return fallback_title, text
|
||||
_SYSTEM_PROMPT = """\
|
||||
Today is {today}. You are a quick-capture assistant. The user has sent a short \
|
||||
snippet from their mobile device. Create the appropriate item — note, task, or \
|
||||
calendar event — using the available tools. Always call a tool; never reply \
|
||||
conversationally."""
|
||||
|
||||
|
||||
@quick_capture_bp.route("", methods=["POST"])
|
||||
@login_required
|
||||
async def quick_capture_route():
|
||||
"""Classify text and create the appropriate item, returning a single JSON response."""
|
||||
"""Classify text via native tool-calling and create the appropriate item."""
|
||||
uid = get_current_user_id()
|
||||
data = await request.get_json(silent=True) or {}
|
||||
text = data.get("text", "").strip()
|
||||
@@ -81,26 +43,37 @@ async def quick_capture_route():
|
||||
from fabledassistant.services.settings import get_setting
|
||||
model = await get_setting(uid, "default_model", Config.OLLAMA_MODEL)
|
||||
|
||||
# Build tool list for this user, then restrict to capture-only operations.
|
||||
all_tools = await get_tools_for_user(uid)
|
||||
capture_tools = [
|
||||
t for t in all_tools if t.get("function", {}).get("name") in _CAPTURE_TOOL_NAMES
|
||||
]
|
||||
|
||||
intent = await classify_capture_intent(text, capture_tools, model)
|
||||
messages = [
|
||||
{"role": "system", "content": _SYSTEM_PROMPT.format(today=date.today().isoformat())},
|
||||
{"role": "user", "content": text},
|
||||
]
|
||||
|
||||
if intent.should_execute:
|
||||
# research_topic bypasses execute_tool — run the pipeline directly
|
||||
if intent.tool_name == "research_topic" and Config.searxng_enabled():
|
||||
think = _should_think(text, think_requested=True)
|
||||
|
||||
tool_calls: list[dict] = []
|
||||
try:
|
||||
async for chunk in stream_chat_with_tools(messages, model, tools=capture_tools, think=think, num_ctx=4096):
|
||||
if chunk.type == "tool_calls" and chunk.tool_calls:
|
||||
tool_calls = chunk.tool_calls
|
||||
except Exception:
|
||||
logger.warning("Quick-capture LLM call failed for uid=%d", uid, exc_info=True)
|
||||
|
||||
if tool_calls:
|
||||
tc = tool_calls[0]
|
||||
tool_name = tc.get("function", {}).get("name", "")
|
||||
arguments = tc.get("function", {}).get("arguments", {})
|
||||
|
||||
if tool_name == "research_topic" and Config.searxng_enabled():
|
||||
from fabledassistant.services.research import run_research_pipeline
|
||||
|
||||
topic = intent.arguments.get("topic", text)
|
||||
topic = arguments.get("topic", text)
|
||||
try:
|
||||
note = await run_research_pipeline(topic, uid, model)
|
||||
logger.info(
|
||||
"Quick-capture uid=%d: research note id=%d '%s'",
|
||||
uid, note.id, note.title,
|
||||
)
|
||||
logger.info("Quick-capture uid=%d: research note id=%d '%s'", uid, note.id, note.title)
|
||||
return jsonify({
|
||||
"success": True,
|
||||
"type": "note",
|
||||
@@ -108,48 +81,27 @@ async def quick_capture_route():
|
||||
"data": {"id": note.id, "title": note.title},
|
||||
})
|
||||
except Exception as exc:
|
||||
logger.exception("Quick-capture research failed for topic: %s", topic)
|
||||
logger.exception("Quick-capture research failed: %s", topic)
|
||||
return jsonify({"error": f"Research failed: {exc}"}), 500
|
||||
|
||||
# For notes, run a second LLM pass to generate a proper title and
|
||||
# well-formed body rather than using the raw capture text verbatim.
|
||||
if intent.tool_name == "create_note":
|
||||
title, body = await _process_note(text, model)
|
||||
intent.arguments["title"] = title
|
||||
intent.arguments["body"] = body
|
||||
|
||||
result = await execute_tool(uid, intent.tool_name, intent.arguments)
|
||||
result = await execute_tool(uid, tool_name, arguments)
|
||||
if result.get("success"):
|
||||
item_type = result.get("type", "note")
|
||||
title = (result.get("data") or {}).get("title", "")
|
||||
logger.info(
|
||||
"Quick-capture uid=%d: %s '%s' via intent '%s'",
|
||||
uid, item_type, title, intent.tool_name,
|
||||
)
|
||||
logger.info("Quick-capture uid=%d: %s '%s'", uid, item_type, title)
|
||||
return jsonify({
|
||||
"success": True,
|
||||
"type": item_type,
|
||||
"message": f"{item_type.capitalize()}: {title}",
|
||||
"data": result.get("data"),
|
||||
})
|
||||
logger.warning(
|
||||
"Quick-capture uid=%d: tool '%s' returned failure: %s",
|
||||
uid, intent.tool_name, result.get("error"),
|
||||
)
|
||||
# Fall through to plain-note fallback
|
||||
logger.warning("Quick-capture uid=%d: tool '%s' failed: %s", uid, tool_name, result.get("error"))
|
||||
|
||||
# Fallback: classify_capture_intent returned no-tool (e.g. LLM parse failure).
|
||||
# Still process the text through the note enrichment pass.
|
||||
fallback_title, fallback_body = await _process_note(text, model)
|
||||
|
||||
result = await execute_tool(
|
||||
uid, "create_note", {"title": fallback_title, "body": fallback_body}
|
||||
)
|
||||
# Fallback: create a plain note with the raw text
|
||||
result = await execute_tool(uid, "create_note", {"title": text[:80], "body": text})
|
||||
if result.get("success"):
|
||||
title = (result.get("data") or {}).get("title", "")
|
||||
logger.info(
|
||||
"Quick-capture uid=%d: fallback note created '%s'", uid, title
|
||||
)
|
||||
logger.info("Quick-capture uid=%d: fallback note '%s'", uid, title)
|
||||
return jsonify({
|
||||
"success": True,
|
||||
"type": "note",
|
||||
|
||||
@@ -1,182 +0,0 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user