Switch default model to qwen3 and add intent routing for reliable tool calling

Mistral didn't reliably use Ollama's structured tool calling API — it wrote
tool calls as JSON text instead of invoking them. This adds an intent routing
layer that classifies user intent via a fast non-streaming LLM call before
streaming, executing detected tools directly and bypassing native tool calling.

- Change default OLLAMA_MODEL from mistral to qwen3
- Add intent.py: classify_intent() with JSON parsing and fallback regex
- Integrate intent routing into generation_task.py round 0
- Add all-day event support (iCalendar DATE values) to CalDAV service
- Add recurring event support (RRULE) to CalDAV service and tool definition
- Improve create_event tool description for descriptive titles
- Enhance system prompt with structured tool usage guidance

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-16 16:24:01 -05:00
parent d7bc3f3222
commit 75560dee4e
7 changed files with 272 additions and 31 deletions
+151
View File
@@ -0,0 +1,151 @@
"""Intent routing — classify user message before streaming.
Makes a fast non-streaming LLM call to detect tool intent and extract
parameters. When a tool call is detected the caller can execute it
directly, bypassing the model's native (and sometimes unreliable)
structured tool-calling API.
"""
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)
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)
_SYSTEM_PROMPT_TEMPLATE = """\
You are an intent classifier. Given a user message, decide whether it \
requires calling one of the available tools or is just general chat.
Today's date is {today}.
Available tools:
{tool_summary}
Respond with ONLY a JSON object, no other text:
- If a tool should be called: {{"tool": "tool_name", "arguments": {{...}}}}
- If it's general chat: {{"tool": null}}
Rules:
- For dates like "tomorrow", "next Friday", "in 3 days", resolve them to YYYY-MM-DD format.
- For datetime parameters, use ISO 8601 format (e.g. 2026-09-30T14:00:00).
- Only include arguments the user actually specified or that can be clearly inferred.
- Infer reasonable defaults: birthdays and holidays are all-day + yearly recurring; "weekly meeting" is weekly recurring.
- Use descriptive titles: "My Birthday" not just "Birthday", "Team Standup" not just "Meeting".
- Do NOT wrap the JSON in markdown code fences."""
async def classify_intent(
user_message: str,
tools: list[dict],
model: str,
) -> IntentResult:
"""Classify user intent via a fast non-streaming LLM call.
Returns an IntentResult. On any failure, returns IntentResult(tool_name=None)
so the caller falls through to the normal streaming path.
"""
if not tools:
return IntentResult()
tool_summary = _build_tool_summary(tools)
today = date_type.today().isoformat()
messages = [
{
"role": "system",
"content": _SYSTEM_PROMPT_TEMPLATE.format(
today=today, tool_summary=tool_summary
),
},
{"role": "user", "content": user_message},
]
try:
raw = await generate_completion(messages, model)
except Exception:
logger.warning("Intent classification LLM call failed", exc_info=True)
return IntentResult()
return _parse_intent(raw, tools)
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")
if tool_name is None:
return IntentResult()
# 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 = {}
logger.info("Intent classified: tool=%s, args=%s", tool_name, json.dumps(arguments)[:200])
return IntentResult(tool_name=tool_name, arguments=arguments)
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