fix: force tool calls for drifted intents and harden tool-use prompt
Small local models (qwen2.5-vl-3b, gemma-4-e4b) narrate actions ('Sure,
I'll pause the music') without emitting the tool call once the full
system prompt and 26-tool set are in context. Measured: qwen emitted
tools in 2/8 realistic intents, gemma 5/8; single-tool tests pass 100%,
so this is context-induced drift, not plumbing.
- agent/intent_router.py: deterministic forced-call router for
high-frequency intents — media controls, now_playing/list_library,
'remember that ...' memory saves, explicit YouTube Music searches,
and personal-fact recall (memory_recall/search_notes/skill_recall,
excluded for weather/time questions). Fires only when the tool is
available; guards against how-do-I questions.
- agent.py: generalize _ForcedToolCallStream to any tool; route through
intent_router after the dispatch check. System prompt adds a hard
'call, don't narrate' rule and reconditions 'answer directly'.
- tests/test_intent_router.py: 30 unit tests pinning router behavior.
Verified against the live LLM with the full fidelity harness: combined
forced-or-emitted coverage 8/8 on qwen2.5-vl-3b (was 2/8) and 8/8 on
gemma-4-e4b (was 5/8). Full suite: 79 passed.
This commit is contained in:
1 parent
48b92585c6
commit
8a7cca456f
4 files changed
+464
-14
No files matched your search
@@ -0,0 +1,197 @@
|
||||
"""Unit tests for the deterministic intent router (host-side, pure functions).
|
||||
|
||||
The router forces tool calls for intents small local models drift on —
|
||||
these tests pin the matching behavior that the agent relies on.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
_SPEC = importlib.util.spec_from_file_location(
|
||||
"intent_router", Path(__file__).parents[1] / "agent" / "intent_router.py"
|
||||
)
|
||||
intent_router = importlib.util.module_from_spec(_SPEC)
|
||||
sys.modules["intent_router"] = intent_router
|
||||
_SPEC.loader.exec_module(intent_router)
|
||||
|
||||
detect = intent_router.detect_forced_call
|
||||
available = intent_router.tool_available
|
||||
|
||||
FULL_TOOLSET = [
|
||||
"get_weather", "get_time",
|
||||
"memory_save", "memory_recall", "search_notes", "read_note",
|
||||
"skill_recall", "skill_save",
|
||||
"dispatch_task",
|
||||
"search_music", "play_song", "offer_song_choices", "pause_music",
|
||||
"resume_music", "stop_music", "skip_song", "now_playing", "list_library",
|
||||
]
|
||||
|
||||
|
||||
class FakeTool:
|
||||
def __init__(self, name, id=None):
|
||||
self.name = name
|
||||
self.id = id
|
||||
|
||||
|
||||
def tools(names):
|
||||
return [FakeTool(n) for n in names]
|
||||
|
||||
|
||||
# ── tool availability ───────────────────────────────────────────────────────
|
||||
|
||||
def test_tool_available_exact_and_prefixed():
|
||||
assert available(tools(["pause_music"]), "pause_music")
|
||||
assert available(tools(["media__pause_music"]), "pause_music")
|
||||
assert not available(tools(["play_song"]), "pause_music")
|
||||
assert not available([], "pause_music")
|
||||
# id-only attribute (dispatch check path used by agent.py)
|
||||
assert available([FakeTool(None, id="dispatch_task")], "dispatch_task")
|
||||
|
||||
|
||||
# ── media controls ──────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.parametrize("utterance,expected", [
|
||||
("pause the music", "pause_music"),
|
||||
("pause", "pause_music"),
|
||||
("could you pause it", "pause_music"),
|
||||
("stop the music", "stop_music"),
|
||||
("stop this song", "stop_music"),
|
||||
("resume the music", "resume_music"),
|
||||
("unpause", "resume_music"),
|
||||
("play it again", "resume_music"),
|
||||
("skip this song", "skip_song"),
|
||||
("next track", "skip_song"),
|
||||
])
|
||||
def test_media_controls_forced(utterance, expected):
|
||||
assert detect(utterance, tools(FULL_TOOLSET)) == (expected, {})
|
||||
|
||||
|
||||
def test_stop_without_music_context_not_forced():
|
||||
# "stop" alone is ambiguous (could mean ending the conversation) — the
|
||||
# router leaves it to the LLM.
|
||||
assert detect("stop", tools(FULL_TOOLSET)) is None
|
||||
|
||||
|
||||
def test_media_controls_guard_against_questions():
|
||||
assert detect("how do I pause the music?", tools(FULL_TOOLSET)) is None
|
||||
assert detect("what happens if I stop the music", tools(FULL_TOOLSET)) is None
|
||||
|
||||
|
||||
def test_media_control_missing_tool_not_forced():
|
||||
no_media = [n for n in FULL_TOOLSET if not n.endswith("_music") or n == "play_song"]
|
||||
assert detect("pause the music", tools(no_media)) is None
|
||||
|
||||
|
||||
# ── memory saves ────────────────────────────────────────────────────────────
|
||||
|
||||
def test_memory_save_forced_with_extracted_body():
|
||||
result = detect("Remember that my coffee order is an oat milk latte",
|
||||
tools(FULL_TOOLSET))
|
||||
assert result is not None
|
||||
tool_name, args = result
|
||||
assert tool_name == "memory_save"
|
||||
assert args["body"] == "my coffee order is an oat milk latte"
|
||||
assert args["title"]
|
||||
assert isinstance(args["tags"], list)
|
||||
|
||||
|
||||
def test_memory_save_variants():
|
||||
for utterance in (
|
||||
"Remember my wifi password is falcon9",
|
||||
"please remember that the garage code is 1234",
|
||||
):
|
||||
result = detect(utterance, tools(FULL_TOOLSET))
|
||||
assert result and result[0] == "memory_save", utterance
|
||||
|
||||
|
||||
def test_memory_save_guard_and_short_body():
|
||||
assert detect("how do I make you remember things?", tools(FULL_TOOLSET)) is None
|
||||
assert detect("remember", tools(FULL_TOOLSET)) is None
|
||||
|
||||
|
||||
# ── music search ────────────────────────────────────────────────────────────
|
||||
|
||||
def test_search_forced_for_explicit_and_play_imperative():
|
||||
cases = [
|
||||
("find Bohemian Rhapsody on YouTube Music", "Bohemian Rhapsody"),
|
||||
("search for timeless miracle", "timeless miracle"),
|
||||
("play Timeless Miracle by Blind Guardian", "Timeless Miracle by Blind Guardian"),
|
||||
]
|
||||
for utterance, query in cases:
|
||||
result = detect(utterance, tools(FULL_TOOLSET))
|
||||
assert result and result[0] == "search_music", utterance
|
||||
assert result[1]["query"] == query
|
||||
|
||||
|
||||
@pytest.mark.parametrize("utterance", [
|
||||
"play some music",
|
||||
"play something",
|
||||
"play a game",
|
||||
"how do I find a song on YouTube Music?",
|
||||
])
|
||||
def test_search_not_forced_for_generic_or_questions(utterance):
|
||||
assert detect(utterance, tools(FULL_TOOLSET)) is None
|
||||
|
||||
|
||||
# ── question intents ────────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.parametrize("utterance,expected", [
|
||||
("what's playing right now?", "now_playing"),
|
||||
("what song is this", "now_playing"),
|
||||
("what's in my music library?", "list_library"),
|
||||
("list my songs", "list_library"),
|
||||
])
|
||||
def test_question_intents_forced_even_though_questions(utterance, expected):
|
||||
assert detect(utterance, tools(FULL_TOOLSET)) == (expected, {})
|
||||
|
||||
|
||||
# ── recall (memory / notes / skills) ────────────────────────────────────────
|
||||
|
||||
def test_memory_recall_forced_for_personal_facts():
|
||||
cases = [
|
||||
"what's my coffee order?",
|
||||
"What is my wifi password",
|
||||
"do you remember my garage code?",
|
||||
"who am I?",
|
||||
]
|
||||
for utterance in cases:
|
||||
result = detect(utterance, tools(FULL_TOOLSET))
|
||||
assert result and result[0] == "memory_recall", utterance
|
||||
assert result[1]["query"] == utterance
|
||||
|
||||
|
||||
def test_memory_recall_not_stolen_from_weather_time():
|
||||
# These must reach the weather/time tools, which work lexically.
|
||||
for utterance in (
|
||||
"what's the weather in Raleigh?",
|
||||
"what time is it?",
|
||||
"what's my weather like today?",
|
||||
"what's the date today?",
|
||||
):
|
||||
assert detect(utterance, tools(FULL_TOOLSET)) != "memory_recall"
|
||||
|
||||
|
||||
def test_search_notes_and_skill_recall_forced():
|
||||
assert detect("do I have any notes about project deadlines?",
|
||||
tools(FULL_TOOLSET))[0] == "search_notes"
|
||||
assert detect("check my notes for the garage code",
|
||||
tools(FULL_TOOLSET))[0] == "search_notes"
|
||||
assert detect("do I have a skill for brewing coffee?",
|
||||
tools(FULL_TOOLSET))[0] == "skill_recall"
|
||||
assert detect("check your skills for baking bread",
|
||||
tools(FULL_TOOLSET))[0] == "skill_recall"
|
||||
|
||||
|
||||
# ── non-intents fall through to the LLM ─────────────────────────────────────
|
||||
|
||||
def test_plain_conversation_not_forced():
|
||||
for utterance in (
|
||||
"hello",
|
||||
"what's the weather in Raleigh?",
|
||||
"tell me about the Roman empire",
|
||||
"",
|
||||
):
|
||||
assert detect(utterance, tools(FULL_TOOLSET)) is None
|
||||
Reference in new issue
Block a user