fix(ai): add assistant tool guardrails (#1066)

Author: cosven

feeluown/ai/copilot.py

@@ -9,9 +9,14 @@
 from feeluown.app import App
 from feeluown.ai.llm import create_chat_model_with_config
 from feeluown.ai.matcher import SongSuggestionMatcher
+from feeluown.ai.model_cache import ModelCache
 from feeluown.ai.models import SongSuggestion
 from feeluown.ai.tools import copilot_tools
-from feeluown.library import BriefSongModel, SimpleSearchResult
+from feeluown.library import (
+    BaseModel,
+    BriefSongModel,
+    SimpleSearchResult,
+)
 from feeluown.utils.dispatch import Signal
 
 
@@ -99,26 +104,32 @@ class CopilotContext:
 
 通用规则：
 - 当你向用户推荐或整理一组歌曲时，优先调用 create_song_suggestions_artifact 工具创建可交互歌曲建议列表。
+- SongSuggestion 是尚未匹配成 SongModel 的歌曲建议；不要把一组 SongSuggestion 一次性转换或播放。
+- create_song_suggestions_artifact 会清洗并校验歌曲建议；单个 artifact 最多包含 20 首。
+- play_song_suggestion 只用于“最新用户消息明确要求播放某一首建议歌曲”的场景。
+- 对 SongSuggestion artifact 中的歌曲，如果用户要求播放，先用 library_search 找到 SongModel。
+- 播放真实歌曲资源时，调用 play_song_by_uri，传入歌曲 uri。
 - 上一首、下一首、暂停、继续、停止、音量调整等基础播放控制，应通过 playback_ 开头的工具完成。
 - 当用户要求搜索在线音乐资源时，优先使用 library_search 工具，并用 timeout 控制最长等待时间。
 - library_search 返回的 data.results 中的 uri 是真实资源 URI，可以在 Markdown 链接里使用。
 - library_search 会创建搜索结果 artifact，并在 data.artifact_id 返回编号。
-- 当用户要求播放搜索结果中的某首歌时，使用歌曲的 artifact_song_position 调用 play_artifact_song。
 
 AI 电台：
 - AI 电台开关、状态和偏好应优先通过 ai_radio_ 开头的工具完成，不要要求用户去其它界面操作。
 - AI 电台只是激活 FM 模式的一种方式；FeelUOwn 也可以通过歌曲电台等其它方式进入 FM 模式。
-- 当用户要求开启、启动、进入 AI 电台时，调用 ai_radio_activate。
+- 当用户明确要求开启、启动、进入 AI 电台时，才调用 ai_radio_activate。
 - 当用户要求关闭、停止、退出 AI 电台时，调用 ai_radio_deactivate。
 - 当用户反馈会影响后续 AI 电台推荐偏好时，调用 ai_radio_update_preferences。
+- 在执行依赖 AI 电台已开启的操作前，先调用 ai_radio_get_state；如果返回 inactive，不要自动开启，除非最新用户消息明确要求开启 AI 电台。
 
 FM 候选列表：
 - FM 候选歌曲指播放列表中当前播放歌曲后面的真实歌曲。
 - FM 候选歌曲不是 SongSuggestion，也不是正文中的 fuo://song-suggestion 链接。
 - FM 候选列表和 AI 电台是否开启无直接关系。
 - 查看 FM 候选列表时调用 fm_candidates_get_state。
 - 修改 FM 候选列表时只使用 fm_candidates_remove 和 fm_candidates_append。
-- fm_candidates_append 接收真实 provider 歌曲。
+- fm_candidates_append 接收真实歌曲 URI 列表。
+- fm_candidates_append 一次最多追加 3 首真实歌曲；更多歌曲需要分批处理。
 - 如果只有文字描述，先调用 library_search 找到真实歌曲资源。
 - 清空候选列表时，先调用 fm_candidates_get_state，再用 fm_candidates_remove 删除全部候选位置。
 - 替换候选列表时，先 remove 不需要的候选，再 append 新候选。
@@ -170,6 +181,7 @@ def __init__(self, app: App):
         self._agent_context = CopilotContext(copilot=self, app=app)
         self._agent_stream_callback = AgentStreamCallback(self)
         self._artifacts = ArtifactsManager()
+        self._model_cache = ModelCache(getattr(app, "library", None))
         self.artifact_added = self._artifacts.added
         self._current_thread_id = 1
         # Agent is working or not
@@ -190,6 +202,7 @@ def is_working(self, working: bool):
     def new_thread(self):
         self._current_thread_id += 1
         self._artifacts.clear()
+        self._model_cache = ModelCache(getattr(self._app, "library", None))
 
     async def match_song_suggestion(
         self, suggestion: SongSuggestion
@@ -214,7 +227,22 @@ def add_songs_artifact(
     def add_search_result_artifact(
         self, results: List[SimpleSearchResult], title: str = ""
     ) -> CopilotArtifact:
-        return self._artifacts.add_search_result(results, title=title)
+        artifact = self._artifacts.add_search_result(results, title=title)
+        for song in artifact.songs:
+            if not isinstance(song, SongSuggestion):
+                self.cache_model(song)
+        return artifact
+
+    def cache_model(self, model: BaseModel):
+        self._model_cache.set_model(model)
+
+    def get_model_by_uri(self, uri: str) -> BaseModel:
+        return self._model_cache.get(uri)
+
+    def get_song_by_uri(self, uri: str) -> BriefSongModel:
+        model = self.get_model_by_uri(uri)
+        assert isinstance(model, BriefSongModel)
+        return model
 
     def get_artifacts(self) -> List[CopilotArtifact]:
         return self._artifacts.list()

feeluown/ai/model_cache.py

@@ -0,0 +1,65 @@
+from collections import OrderedDict
+from threading import Lock
+
+from feeluown.library import BaseModel, ModelType, parse_line, reverse
+
+
+def parse_model_uri(uri: str) -> BaseModel:
+    uri = uri.strip()
+    if not uri:
+        raise ValueError("model URI is required")
+    try:
+        model, path = parse_line(uri)
+    except Exception as e:
+        raise ValueError("invalid model URI") from e
+    if path:
+        raise ValueError("model URI must not include path")
+    return model
+
+
+class ModelCache:
+    """Session scoped cache for resolving model URI to model objects."""
+
+    def __init__(self, library, maxsize: int = 256):
+        if maxsize < 0:
+            raise ValueError("maxsize must not be negative")
+        self._library = library
+        self._maxsize = maxsize
+        self._models: OrderedDict[str, BaseModel] = OrderedDict()
+        self._lock = Lock()
+
+    def set_model(self, model: BaseModel):
+        with self._lock:
+            self._set(reverse(model), model)
+
+    def get(self, uri: str) -> BaseModel:
+        model = parse_model_uri(uri)
+        cache_key = reverse(model)
+
+        with self._lock:
+            cached_model = self._models.get(cache_key)
+            if cached_model is not None:
+                self._models.move_to_end(cache_key)
+                return cached_model
+
+        if self._library is None:
+            raise RuntimeError("library is required on cache miss")
+        fetched_model = self._library.model_get(
+            model.source,
+            ModelType(model.meta.model_type),
+            model.identifier,
+        )
+
+        with self._lock:
+            cached_model = self._models.get(cache_key)
+            if cached_model is not None:
+                self._models.move_to_end(cache_key)
+                return cached_model
+            self._set(cache_key, fetched_model)
+        return fetched_model
+
+    def _set(self, uri: str, model: BaseModel):
+        self._models[uri] = model
+        self._models.move_to_end(uri)
+        while len(self._models) > self._maxsize:
+            self._models.popitem(last=False)

feeluown/ai/models.py

@@ -5,7 +5,7 @@
 
 @dataclass
 class SongSuggestion:
-    """A song suggested by the AI before it is matched to a provider song.
+    """A song suggested by the AI before it is matched to a SongModel.
 
     :param description: Recommendation reason or song description.
     """

feeluown/ai/tools/__init__.py

@@ -1,5 +1,5 @@
 from feeluown.ai.tools.library import library_search, library_tools
-from feeluown.ai.tools.artifacts import play_artifact_song, artifact_tools
+from feeluown.ai.tools.songs import play_song_by_uri, song_tools
 from feeluown.ai.tools.playback import (
     playback_adjust_volume,
     playback_get_state,
@@ -35,7 +35,7 @@
 copilot_tools = [
     *suggestion_tools,
     *library_tools,
-    *artifact_tools,
+    *song_tools,
     *playback_tools,
     *ai_radio_tools,
     *fm_candidates_tools,
@@ -56,9 +56,9 @@
     "fm_candidates_tools",
     "library_search",
     "library_tools",
-    "artifact_tools",
+    "song_tools",
     "play_song_suggestion",
-    "play_artifact_song",
+    "play_song_by_uri",
     "playback_adjust_volume",
     "playback_get_state",
     "playback_next_track",

feeluown/ai/tools/ai_radio.py

@@ -29,7 +29,10 @@ def _ai_radio_unavailable_result():
 
 
 @tool
-def ai_radio_activate(runtime: ToolRuntime, reset: bool = True) -> dict:
+def ai_radio_activate(
+    runtime: ToolRuntime,
+    reset: bool = True,
+) -> dict:
     """Activate AI Radio.
 
     This switches the playlist into FM mode and lets the current AI radio

feeluown/ai/tools/artifacts.py

@@ -1,11 +1,14 @@
 from langchain.tools import tool, ToolRuntime
 
-from feeluown.ai.tools.result import tool_bool_result, tool_success
+from feeluown.ai.tools.result import tool_bool_result, tool_error, tool_success
 from feeluown.library import BriefSongModel
 from feeluown.player.playlist import PlaylistMode
 from feeluown.serializers import serialize
 
 
+MAX_APPEND_SONGS = 3
+
+
 def _get_fm_candidates(runtime: ToolRuntime):
     return runtime.context.app.fm.candidates
 
@@ -67,15 +70,60 @@ def fm_candidates_remove(positions: list[int], runtime: ToolRuntime) -> dict:
 
 @tool
 def fm_candidates_append(
-    songs: list[BriefSongModel], runtime: ToolRuntime
+    song_uris: list[str], runtime: ToolRuntime
 ) -> dict:
-    """Append real songs to the FM candidate list.
+    """Append songs to the FM candidate list by SongModel URI.
+
+    FM candidates are SongModel items. Use library_search first when you need
+    to discover SongModel URIs from text.
 
-    FM candidates are real provider songs. Use library_search first when you
-    need to discover real provider songs from text.
+    Append at most 3 songs in one call. When adding more songs, split the work
+    into smaller batches so matching/searching remains observable and bounded.
 
-    :param songs: Real provider songs to append.
+    :param song_uris: SongModel URI list to append.
     """
+    if len(song_uris) > MAX_APPEND_SONGS:
+        return tool_error(
+            "fm_candidates_append",
+            "TOO_MANY_SONGS",
+            "Append at most 3 songs in one fm_candidates_append call.",
+            data={
+                "success": False,
+                "max_song_count": MAX_APPEND_SONGS,
+                "song_count": len(song_uris),
+                "active": _is_fm_active(runtime),
+            },
+        )
+    if not _is_fm_active(runtime):
+        return _fm_candidate_result("fm_candidates_append", False, runtime)
+
+    songs = []
+    for uri in song_uris:
+        try:
+            song = runtime.context.copilot.get_song_by_uri(uri)
+        except ValueError:
+            return tool_error(
+                "fm_candidates_append",
+                "INVALID_SONG_URI",
+                "A valid SongModel URI is required.",
+                data={
+                    "success": False,
+                    "song_uri": uri,
+                    "active": _is_fm_active(runtime),
+                },
+            )
+        except Exception:  # noqa
+            return tool_error(
+                "fm_candidates_append",
+                "SONG_MODEL_NOT_FOUND",
+                "SongModel was not found for the given URI.",
+                data={
+                    "success": False,
+                    "song_uri": uri,
+                    "active": _is_fm_active(runtime),
+                },
+            )
+        songs.append(song)
     fm_candidates = _get_fm_candidates(runtime)
     return _fm_candidate_result(
         "fm_candidates_append",