diff --git a/app/search/tools.py b/app/search/tools.py new file mode 100644 index 0000000..f9d8f7a --- /dev/null +++ b/app/search/tools.py @@ -0,0 +1,321 @@ +"""搜索 Agent 工具调度器 — 工具描述与 dispatch 分发。 + +实现 ``core/agent/protocols.ToolDispatcher`` Protocol。 +连接 TreeEnvironment(数据)、summarizer(LLM 摘要)、 +vision(VLM 观察)和 skills(策略加载)。 + +与 TRM4 ``core/tree/tools.py`` 的差异: +- 自由函数 ``dispatch()`` → ``SearchToolDispatcher`` 类(依赖注入); +- 同步 → 全异步; +- view_node / search_similar 内部拆分为 env 数据读取 + summarizer LLM 摘要。 +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from app.search.summarizer import summarize_children, summarize_node, summarize_nodes_batch +from app.search.vision import observe_frame +from app.tree.environment import _LEVEL_LABEL, TreeEnvironment, _node_level + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + import numpy as np + + from app.ports import OCRProvider + from app.search.skills import SkillRegistry + from core.protocols import LLMProvider, VLMProvider + +# ── 工具描述文本(与 TRM4 core/tree/tools.py 完全一致) ───────────────── + +_BASE_DESCRIPTIONS = """\ +## 可用工具 + +在 action 中指定 tool 和 args 来调用工具。 + +### view_node +查看节点信息,获取与问题相关的内容摘要和子节点概览。 +- args: {"node_id": "节点 ID", "question": "当前关注的具体问题"} + +### search_similar +语义检索最相关的节点,返回与问题相关的内容摘要。 +- args: {"query": "搜索关键词(2-4 词)", "question": "当前关注的具体问题", "k": 返回数量(可选,默认 5)} + +### observe_frame +调用视觉模型查看关键帧图像,回答针对性的视觉问题。 +- args: {"node_ids": ["L3 节点 ID 列表(1-4 个),或单个 L2 节点 ID"], "question": "针对帧内容的具体视觉问题"} + +### submit_answer +提交最终答案。 +- args: {"answer": "选项字母 A/B/C/D", "evidence": "关键证据摘要", "reasoning": "每个选项的判断理由"}""" + +_SKILL_DESCRIPTION = """ + +### read_skill +加载指定题型技能的详细搜索策略。 +- args: {"name": "技能名称"}""" + + +def get_tool_descriptions(include_read_skill: bool = False) -> str: + """返回工具描述文本,用于写入 system prompt。 + + 参数: + include_read_skill: 是否包含 read_skill 工具(manual 模式用)。 + + 返回: + Markdown 格式的工具描述文本。 + """ + text = _BASE_DESCRIPTIONS + if include_read_skill: + text += _SKILL_DESCRIPTION + return text + + +# ── SearchToolDispatcher ────────────────────────────────────────────── + + +class SearchToolDispatcher: + """搜索 Agent 工具调度器,实现 ToolDispatcher Protocol。 + + 按工具名路由到对应私有处理方法。未知工具抛 ValueError + (AgentLoop 捕获后不计步数);节点不存在等运行时错误 + 捕获后返回错误文本。 + + 参数: + env: 视频树运行时环境(纯数据访问)。 + tool_llm: 摘要用 LLM 端口。 + vlm: 视觉模型端口。 + ocr: 帧文字转录端口(None 不启用)。 + prompts_dir: prompt 文件目录。 + skills: 技能注册表(None 不启用 read_skill)。 + embed_fn: 文本嵌入函数(search_similar 用)。 + verify_vision: observe_frame 是否执行验证轮。 + anchor: view_node 是否启用行号锚模式。 + assemble_mode: 锚模式装配形态("ids"/"ids_expand"/"expand_only")。 + stats_sink: 统计回调(None 不收集)。 + """ + + def __init__( + self, + env: TreeEnvironment, + tool_llm: LLMProvider, + vlm: VLMProvider, + ocr: OCRProvider | None, + prompts_dir: Path, + skills: SkillRegistry | None, + *, + embed_fn: Callable[[str | list[str]], np.ndarray], + verify_vision: bool, + anchor: bool, + assemble_mode: str, + stats_sink: Callable[[dict[str, Any]], None] | None = None, + ) -> None: + self._env = env + self._tool_llm = tool_llm + self._vlm = vlm + self._ocr = ocr + self._prompts_dir = prompts_dir + self._skills = skills + self._embed_fn = embed_fn + self._verify_vision = verify_vision + self._anchor = anchor + self._assemble_mode = assemble_mode + self._stats_sink = stats_sink + + # ── ToolDispatcher Protocol 实现 ────────────────────────────────── + + async def dispatch( + self, tool_name: str, args: dict[str, Any], *, context: dict[str, Any] + ) -> str: + """按工具名分发到对应处理方法。 + + 参数: + tool_name: 工具名称。 + args: 工具参数字典。 + context: 调用上下文(含 session_id、parent_call_id 等遥测字段)。 + + 返回: + 工具执行结果文本。 + + 异常: + ValueError: 未知工具名——上抛给 AgentLoop,不计步数。 + """ + try: + if tool_name == "view_node": + return await self._handle_view_node(args, context) + if tool_name == "search_similar": + return await self._handle_search_similar(args, context) + if tool_name == "observe_frame": + return await self._handle_observe_frame(args, context) + if tool_name == "submit_answer": + return f"[ok] 答案已提交: {args['answer']}" + if tool_name == "read_skill": + return self._handle_read_skill(args) + except (KeyError, FileNotFoundError) as e: + return f"工具执行错误: {e}" + + raise ValueError(f"未知工具: {tool_name}") + + # ── 私有处理方法 ────────────────────────────────────────────────── + + async def _handle_view_node(self, args: dict[str, Any], context: dict[str, Any]) -> str: + """view_node:节点摘要 + 子节点概览。 + + 参数: + args: {"node_id": str, "question": str}。 + context: 遥测上下文。 + + 返回: + "[节点] {id} | {level} | {time}\\n\\n{summary}\\n\\n[子节点概览] ..." + """ + node_id: str = args["node_id"] + question: str = args["question"] + session_id = context.get("session_id") + parent_call_id = context.get("parent_call_id") + + # Phase 1: 节点元数据(头部格式化) + node = self._env._id_to_node[node_id] + level = _node_level(node) + level_label = _LEVEL_LABEL[level] + time_str = TreeEnvironment._format_time_range(node) + + # Phase 2: 节点内容摘要 + raw_text, anchor_map = self._env.get_node_text(node_id, anchor=self._anchor) + summary = await summarize_node( + self._tool_llm, + raw_text, + question, + self._prompts_dir, + anchor_map=anchor_map, + assemble_mode=self._assemble_mode, + stats_sink=self._stats_sink, + session_id=session_id, + parent_call_id=parent_call_id, + ) + + parts: list[str] = [ + f"[节点] {node_id} | {level_label} | {time_str}", + "", + summary, + ] + + # Phase 3: 子节点概览 + children_info = self._env.get_children_info(node_id) + if children_info: + children_text = await summarize_children( + self._tool_llm, + children_info, + question, + self._prompts_dir, + session_id=session_id, + parent_call_id=parent_call_id, + ) + parts.append(f"\n[子节点概览] {len(children_info)} 个子节点\n{children_text}") + + return "\n".join(parts) + + async def _handle_search_similar(self, args: dict[str, Any], context: dict[str, Any]) -> str: + """search_similar:语义检索 + 批量摘要。 + + 参数: + args: {"query": str, "question": str, "k": int (可选)}。 + context: 遥测上下文。 + + 返回: + "[搜索结果] 查询 \\"{query}\\" → N 个相关节点\\n\\n1. ..." + """ + query: str = args["query"] + question: str = args["question"] + top_k: int = args.get("k", 5) + session_id = context.get("session_id") + parent_call_id = context.get("parent_call_id") + + # Phase 1: 语义检索 + results = self._env.search_similar(query, top_k=top_k, embed_fn=self._embed_fn) + + if not results: + return f'[搜索结果] 查询 "{query}" → 0 个相关节点' + + # Phase 2: 构建摘要输入 + items: list[tuple[str, str, str]] = [] + for nid, score in results: + node = self._env._id_to_node[nid] + raw_text, _ = self._env.get_node_text(nid) + level = _node_level(node) + time_str = TreeEnvironment._format_time_range(node) + extra = f"{level} score={score:.4f} [{time_str}]" + items.append((nid, raw_text, extra)) + + # Phase 3: 并发批量摘要 + summaries = await summarize_nodes_batch( + self._tool_llm, + items, + question, + self._prompts_dir, + session_id=session_id, + parent_call_id=parent_call_id, + ) + + # Phase 4: 格式化输出 + lines: list[str] = [] + for i, (nid, summary_text) in enumerate(summaries): + _, _, extra = items[i] + lines.append(f"{i + 1}. {nid} | {extra}\n {summary_text}") + + header = f'[搜索结果] 查询 "{query}" → {len(results)} 个相关节点' + return header + "\n\n" + "\n\n".join(lines) + + async def _handle_observe_frame(self, args: dict[str, Any], context: dict[str, Any]) -> str: + """observe_frame:VLM 帧观察 + 字幕前置。 + + 参数: + args: {"node_ids": list[str], "question": str}。 + context: 遥测上下文。 + + 返回: + "[字幕上下文] ...\\n[视觉观察] ..." 或 "[视觉观察] ..." + """ + node_ids: list[str] = args["node_ids"] + question: str = args.get("question", "") + session_id = context.get("session_id") + parent_call_id = context.get("parent_call_id") + + if not question.strip(): + return "工具执行错误: question 不能为空" + + # Phase 1: 解析帧路径和字幕 + frame_paths = self._env.resolve_frame_paths(node_ids) + subtitle = self._env.get_subtitle(node_ids[0]) + + # Phase 2: VLM 调用 + result = await observe_frame( + self._vlm, + frame_paths, + question, + self._prompts_dir, + ocr=self._ocr, + verify=self._verify_vision, + stats_sink=self._stats_sink, + session_id=session_id, + parent_call_id=parent_call_id, + ) + + # Phase 3: 字幕前置拼接 + if subtitle: + return f"[字幕上下文] {subtitle}\n{result}" + return result + + def _handle_read_skill(self, args: dict[str, Any]) -> str: + """read_skill:加载指定技能的搜索策略正文。 + + 参数: + args: {"name": str}。 + + 返回: + 技能正文或错误提示。 + """ + if self._skills is None: + return "错误: skills 未启用" + return self._skills.read(args["name"]) diff --git a/app/tree/environment.py b/app/tree/environment.py index ca9d7bf..cf7d1e3 100644 --- a/app/tree/environment.py +++ b/app/tree/environment.py @@ -339,6 +339,7 @@ class TreeEnvironment: 返回: 子节点信息列表,每项包含 {"id", "time_range", "summary"}。 + time_range 为 (start, end) 数值元组(L3 节点退化为 (ts, ts))。 L3 叶子节点返回空列表。 异常: @@ -357,7 +358,7 @@ class TreeEnvironment: result.append( { "id": child.id, - "time_range": self._format_time_range(child), + "time_range": self._node_time_range_raw(child), "summary": desc, } ) @@ -504,6 +505,25 @@ class TreeEnvironment: return f"{node.timestamp:.1f}s" return "N/A" + @staticmethod + def _node_time_range_raw(node: AnyNode) -> tuple[float, float]: + """提取节点时间范围的原始数值元组。 + + L1/L2 返回 time_range 元组;L3 退化为 (timestamp, timestamp); + 全部为 None 时兜底 (0.0, 0.0)。 + + 参数: + node: 树节点。 + + 返回: + (start, end) 秒级数值元组。 + """ + if isinstance(node, (L1Node, L2Node)) and node.time_range: + return node.time_range + if isinstance(node, L3Node) and node.timestamp is not None: + return (node.timestamp, node.timestamp) + return (0.0, 0.0) + @staticmethod def _get_children(node: AnyNode) -> list[AnyNode]: """获取节点的直接子节点列表。 diff --git a/tests/unit/test_search_tools.py b/tests/unit/test_search_tools.py new file mode 100644 index 0000000..12cce60 --- /dev/null +++ b/tests/unit/test_search_tools.py @@ -0,0 +1,446 @@ +"""SearchToolDispatcher 与 get_tool_descriptions 单元测试。 + +验证工具描述生成和五种工具的 dispatch 路由: +view_node、search_similar、observe_frame、submit_answer、read_skill, +以及未知工具 ValueError 和节点不存在错误文本。 +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pathlib import Path + +import numpy as np +import pytest + +from app.search.skills import SkillRegistry +from app.search.tools import SearchToolDispatcher, get_tool_descriptions +from app.tree.environment import TreeEnvironment +from app.tree.index import ( + IndexMeta, + L1Card, + L1Node, + L2Card, + L2Node, + L3Card, + L3Node, + TreeIndex, +) +from core.types import LLMResponse + +# ── 假实现 ──────────────────────────────────────────────────────────── + + +def _make_llm_response(content: str = "fake summary") -> LLMResponse: + """构造固定的 LLMResponse 实例。""" + return LLMResponse( + content=content, + thinking="", + model="fake-model", + provider="fake", + prompt_tokens=10, + completion_tokens=5, + latency_ms=50, + ttft_ms=None, + max_inter_token_ms=None, + cache_hit=False, + call_id="fake-call-id", + ) + + +class FakeLLM: + """最小 LLMProvider 假实现。""" + + async def chat( + self, + messages: list[dict[str, Any]], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + """返回固定摘要内容。""" + return _make_llm_response("fake summary") + + +class FakeVLM: + """最小 VLMProvider 假实现。""" + + async def chat_with_images( + self, + messages: list[dict[str, Any]], + images: list[str | Path], + *, + session_id: str | None = None, + parent_call_id: str | None = None, + ) -> LLMResponse: + """返回固定视觉观察内容。""" + return _make_llm_response("fake visual observation") + + +class FakeOCR: + """最小 OCRProvider 假实现。""" + + async def transcribe_frames(self, frame_paths: list[Path]) -> str: + """返回固定 OCR 文本。""" + return "OCR text" + + +def _fake_embed_fn(texts: str | list[str]) -> np.ndarray: + """返回固定维度的 L2 归一化嵌入向量。""" + if isinstance(texts, str): + vec = np.ones((1, 4), dtype=np.float32) + else: + vec = np.ones((len(texts), 4), dtype=np.float32) + norms = np.linalg.norm(vec, axis=1, keepdims=True) + return vec / norms + + +# ── Fixtures ────────────────────────────────────────────────────────── + + +def _make_test_tree() -> TreeIndex: + """构建包含 L1→L2→L3 的最小测试树。""" + l3 = L3Node( + id="vid_L1_000_L2_000_L3_000", + card=L3Card( + frame_summary="test frame summary", + visible_entities=["person"], + ongoing_actions=["walking"], + visible_text=[], + spatial_layout="center", + visual_attributes={}, + ), + timestamp=10.0, + frame_path="frames/L1_000_L2_000_L3_000.jpg", + subtitle="test subtitle text", + ) + l2 = L2Node( + id="vid_L1_000_L2_000", + card=L2Card( + event_description="test event description", + entities=["person"], + actions=["walking"], + action_subjects=["person"], + visible_text=[], + spatial_relations="none", + state_changes=None, + ), + time_range=(5.0, 15.0), + children=[l3], + ) + l1 = L1Node( + id="vid_L1_000", + card=L1Card( + scene_summary="test scene summary", + main_setting="outdoor", + key_entities=["person"], + main_actions=["walking"], + topic_keywords=["outdoor"], + visible_text=[], + temporal_flow="linear", + ), + time_range=(0.0, 30.0), + children=[l2], + ) + return TreeIndex( + metadata=IndexMeta(source_path="test.mp4", modality="video"), + roots=[l1], + ) + + +@pytest.fixture() +def env() -> TreeEnvironment: + """带最小树的 TreeEnvironment。""" + return TreeEnvironment(_make_test_tree()) + + +@pytest.fixture() +def prompts_dir(tmp_path: Path) -> Path: + """在 tmp 目录中创建必需的 prompt 文件。""" + prompt_files = [ + "view_node_extract.md", + "view_node_verify.md", + "view_node_children_extract.md", + "view_node_children_verify.md", + "search_similar_extract.md", + "search_similar_verify.md", + "observe_frame_extract.md", + "observe_frame_verify.md", + ] + for name in prompt_files: + (tmp_path / name).write_text(f"fake prompt for {name}", encoding="utf-8") + return tmp_path + + +@pytest.fixture() +def skills_registry(tmp_path: Path) -> SkillRegistry: + """带一个预注册技能的 SkillRegistry。""" + skill_path = tmp_path / "test_skill.md" + skill_path.write_text( + "---\nname: test_skill\ndescription: test\n---\nskill body content", + encoding="utf-8", + ) + registry = SkillRegistry() + registry.set_paths({"test_skill": skill_path}) + return registry + + +@pytest.fixture() +def dispatcher( + env: TreeEnvironment, + prompts_dir: Path, + skills_registry: SkillRegistry, +) -> SearchToolDispatcher: + """标准配置的 SearchToolDispatcher 实例。""" + return SearchToolDispatcher( + env=env, + tool_llm=FakeLLM(), + vlm=FakeVLM(), + ocr=FakeOCR(), + prompts_dir=prompts_dir, + skills=skills_registry, + embed_fn=_fake_embed_fn, + verify_vision=False, + anchor=False, + assemble_mode="ids", + ) + + +@pytest.fixture() +def dispatcher_no_skills( + env: TreeEnvironment, + prompts_dir: Path, +) -> SearchToolDispatcher: + """skills=None 的 SearchToolDispatcher 实例。""" + return SearchToolDispatcher( + env=env, + tool_llm=FakeLLM(), + vlm=FakeVLM(), + ocr=None, + prompts_dir=prompts_dir, + skills=None, + embed_fn=_fake_embed_fn, + verify_vision=False, + anchor=False, + assemble_mode="ids", + ) + + +# ── get_tool_descriptions 测试 ─────────────────────────────────────── + + +class TestGetToolDescriptions: + """get_tool_descriptions 工具描述生成测试。""" + + def test_without_read_skill(self) -> None: + """不含 read_skill 时应包含四个基础工具。""" + text = get_tool_descriptions(include_read_skill=False) + assert "view_node" in text + assert "search_similar" in text + assert "observe_frame" in text + assert "submit_answer" in text + assert "read_skill" not in text + + def test_with_read_skill(self) -> None: + """含 read_skill 时应额外包含 read_skill 工具描述。""" + text = get_tool_descriptions(include_read_skill=True) + assert "view_node" in text + assert "read_skill" in text + assert "加载指定题型技能" in text + + +# ── dispatch 路由测试 ───────────────────────────────────────────────── + + +class TestDispatchViewNode: + """dispatch view_node 工具测试。""" + + @pytest.mark.asyncio() + async def test_view_node_returns_header_and_summary( + self, dispatcher: SearchToolDispatcher + ) -> None: + """view_node 应返回含节点头部、摘要和子节点概览的文本。""" + result = await dispatcher.dispatch( + "view_node", + {"node_id": "vid_L1_000", "question": "what happens?"}, + context={}, + ) + # 头部格式 + assert "[节点] vid_L1_000 | 场景层 |" in result + assert "0.0-30.0s" in result + # 摘要内容(来自 FakeLLM) + assert "fake summary" in result + # 子节点概览(L1 有 L2 子节点) + assert "[子节点概览]" in result + assert "1 个子节点" in result + + @pytest.mark.asyncio() + async def test_view_node_l3_no_children(self, dispatcher: SearchToolDispatcher) -> None: + """L3 叶子节点应无子节点概览段。""" + result = await dispatcher.dispatch( + "view_node", + {"node_id": "vid_L1_000_L2_000_L3_000", "question": "test"}, + context={}, + ) + assert "[节点] vid_L1_000_L2_000_L3_000 | 关键帧层 |" in result + assert "[子节点概览]" not in result + + +class TestDispatchSearchSimilar: + """dispatch search_similar 工具测试。""" + + @pytest.mark.asyncio() + async def test_search_similar_returns_results(self, dispatcher: SearchToolDispatcher) -> None: + """search_similar 应返回搜索头部和编号结果列表。""" + result = await dispatcher.dispatch( + "search_similar", + {"query": "walking", "question": "what is the person doing?"}, + context={}, + ) + assert '[搜索结果] 查询 "walking"' in result + assert "个相关节点" in result + # 至少有一个编号结果 + assert "1." in result + # 包含分数信息 + assert "score=" in result + + @pytest.mark.asyncio() + async def test_search_similar_custom_k(self, dispatcher: SearchToolDispatcher) -> None: + """search_similar 的 k 参数应限制返回数量。""" + result = await dispatcher.dispatch( + "search_similar", + {"query": "test", "question": "test", "k": 1}, + context={}, + ) + assert "1 个相关节点" in result + + +class TestDispatchObserveFrame: + """dispatch observe_frame 工具测试。""" + + @pytest.mark.asyncio() + async def test_observe_frame_with_subtitle( + self, dispatcher: SearchToolDispatcher, tmp_path: Path + ) -> None: + """有字幕的 L3 节点应在输出前添加字幕上下文。""" + # 创建帧文件使路径存在检查通过 + frame_file = tmp_path / "L1_000_L2_000_L3_000.jpg" + frame_file.write_bytes(b"\xff\xd8\xff\xe0") + + # 重建 dispatcher 指定 frames_dir + tree = _make_test_tree() + env_with_frames = TreeEnvironment(tree, frames_dir=tmp_path) + + d = SearchToolDispatcher( + env=env_with_frames, + tool_llm=FakeLLM(), + vlm=FakeVLM(), + ocr=FakeOCR(), + prompts_dir=dispatcher._prompts_dir, + skills=None, + embed_fn=_fake_embed_fn, + verify_vision=False, + anchor=False, + assemble_mode="ids", + ) + + result = await d.dispatch( + "observe_frame", + { + "node_ids": ["vid_L1_000_L2_000_L3_000"], + "question": "what is visible?", + }, + context={}, + ) + assert "[字幕上下文] test subtitle text" in result + assert "fake visual observation" in result + + @pytest.mark.asyncio() + async def test_observe_frame_empty_question(self, dispatcher: SearchToolDispatcher) -> None: + """空 question 应返回错误文本。""" + result = await dispatcher.dispatch( + "observe_frame", + {"node_ids": ["vid_L1_000_L2_000_L3_000"], "question": " "}, + context={}, + ) + assert "question 不能为空" in result + + +class TestDispatchSubmitAnswer: + """dispatch submit_answer 工具测试。""" + + @pytest.mark.asyncio() + async def test_submit_answer_returns_confirmation( + self, dispatcher: SearchToolDispatcher + ) -> None: + """submit_answer 应返回确认文本。""" + result = await dispatcher.dispatch( + "submit_answer", + {"answer": "B", "evidence": "seen in frame", "reasoning": "clear visual"}, + context={}, + ) + assert result == "[ok] 答案已提交: B" + + +class TestDispatchReadSkill: + """dispatch read_skill 工具测试。""" + + @pytest.mark.asyncio() + async def test_read_skill_returns_body(self, dispatcher: SearchToolDispatcher) -> None: + """read_skill 应返回去除 frontmatter 后的技能正文。""" + result = await dispatcher.dispatch( + "read_skill", + {"name": "test_skill"}, + context={}, + ) + assert "skill body content" in result + + @pytest.mark.asyncio() + async def test_read_skill_disabled(self, dispatcher_no_skills: SearchToolDispatcher) -> None: + """skills=None 时 read_skill 应返回未启用提示。""" + result = await dispatcher_no_skills.dispatch( + "read_skill", + {"name": "anything"}, + context={}, + ) + assert result == "错误: skills 未启用" + + +# ── 错误处理测试 ────────────────────────────────────────────────────── + + +class TestDispatchErrors: + """dispatch 错误处理测试。""" + + @pytest.mark.asyncio() + async def test_unknown_tool_raises_value_error(self, dispatcher: SearchToolDispatcher) -> None: + """未知工具应抛出 ValueError。""" + with pytest.raises(ValueError, match="未知工具: nonexistent_tool"): + await dispatcher.dispatch("nonexistent_tool", {}, context={}) + + @pytest.mark.asyncio() + async def test_node_not_found_returns_error_text( + self, dispatcher: SearchToolDispatcher + ) -> None: + """节点不存在时应返回错误文本(非异常)。""" + result = await dispatcher.dispatch( + "view_node", + {"node_id": "nonexistent_node", "question": "test"}, + context={}, + ) + assert "工具执行错误" in result + assert "nonexistent_node" in result + + @pytest.mark.asyncio() + async def test_read_skill_not_found_returns_error_text( + self, dispatcher: SearchToolDispatcher + ) -> None: + """未注册的技能名应返回错误文本。""" + result = await dispatcher.dispatch( + "read_skill", + {"name": "nonexistent_skill"}, + context={}, + ) + assert "工具执行错误" in result