diff --git a/tests/test_prune_tools.py b/tests/test_prune_tools.py new file mode 100644 index 0000000..1f77c69 --- /dev/null +++ b/tests/test_prune_tools.py @@ -0,0 +1,115 @@ +""" +测试:FC 工具去重 + 按场景动态裁剪(方案 A+B)。 + +不依赖真实插件/引擎加载,用构造的 fake 数据验证: +- mcp_to_openai_tools 按 name 去重(消除 cache + builtin 叠加重复) +- prune_tools 分层裁剪:纯聊天极简、代码信号追加 lsp__*、外部 mcp 默认保留 +""" +import pytest + +from xcli_core.fc_tools import mcp_to_openai_tools, prune_tools + + +def _fc(name, desc="d"): + return { + "type": "function", + "function": { + "name": name, + "description": desc, + "parameters": {"type": "object", "properties": {}, "required": []}, + }, + } + + +def _mcp(name, desc="d"): + """MCP 格式(get_mcp_tools 返回的就是这种)""" + return { + "name": name, + "description": desc, + "inputSchema": {"type": "object", "properties": {}, "required": []}, + } + + +class _FakePM: + """最小 plugin_manager:get_mcp_tools 返回预置列表。""" + + def __init__(self, tools): + self._tools = tools + + def get_mcp_tools(self): + return self._tools + + +# ───────────── 去重 ───────────── + +def test_mcp_to_openai_tools_dedup(): + # 模拟真实场景:cache 里已有 memory/scheduler,_get_builtin_tools 又追加一遍 + pm = _FakePM([ + _mcp("a"), _mcp("a"), # 同名重复 + _mcp("b"), + _mcp("memory"), _mcp("scheduler"), # cache 已有 + ]) + out = mcp_to_openai_tools(pm) + names = [t["function"]["name"] for t in out] + assert names.count("a") == 1, "同名工具应去重为 1" + assert names.count("memory") == 1, "builtin 与 cache 重复应去重" + assert names.count("scheduler") == 1 + assert "b" in names + # builtin 追加的 memory/scheduler 不应再产生第二份 + assert names.count("memory") == 1 + + +# ───────────── 分层裁剪 ───────────── + +_CORE = ["tool_search", "memory", "scheduler", "code_editor", "code_search", + "file_manager", "lsp__diagnostics", "lsp__hover", "lsp__definition", + "lsp__references", "lsp__completion", "browser_auto", "network_tools", + "ai_search", "git_tools", "cmd_executor", "sub_agent", "auto_engineer"] + + +def _full_set(): + return [_fc(n) for n in _CORE] + + +def test_prune_pure_chat_keeps_only_always(): + out = prune_tools(_full_set(), "你好啊,今天天气真不错") + names = {t["function"]["name"] for t in out} + assert names == {"tool_search", "memory", "scheduler"}, \ + f"纯聊天应只留常驻工具,实际: {names}" + + +def test_prune_code_signal_adds_lsp(): + out = prune_tools(_full_set(), "帮我写一个 python 脚本读取文件") + names = {t["function"]["name"] for t in out} + assert "code_editor" in names + assert "file_manager" in names + assert "lsp__diagnostics" in names, "写代码应追加 pylsp 的 lsp__* 工具" + assert "lsp__completion" in names + assert "tool_search" in names + + +def test_prune_git_signal(): + out = prune_tools(_full_set(), "提交代码并查看 git 状态") + names = {t["function"]["name"] for t in out} + assert "git_tools" in names + assert "tool_search" in names + + +def test_prune_external_mcp_kept_even_in_chat(): + tools = [_fc("my_mcp__search"), _fc("code_editor"), _fc("tool_search")] + out = prune_tools(tools, "你好") + names = {t["function"]["name"] for t in out} + assert "my_mcp__search" in names, "外部桥接工具(mcp__)应默认保留" + assert "code_editor" not in names, "未命中信号的低频工具应被裁剪" + assert "tool_search" in names + + +def test_prune_empty_input_no_crash(): + out = prune_tools(_full_set(), "") + names = {t["function"]["name"] for t in out} + assert names == {"tool_search", "memory", "scheduler"} + + +def test_prune_none_tools_returns_none(): + assert prune_tools(None, "任意输入") is None + assert prune_tools([], "任意输入") == [] diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index c8e125a..120c1dc 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -310,11 +310,14 @@ multi 操作支持一次修改多处: plugins.append(tool) if plugins: - prompts.append("## 内置工具(使用 tool_search 查询详细用法):") - for tool in plugins: - tool_name = tool.get('name', '') - tool_desc = tool.get('description', '') - prompts.append(f"- {tool_name}: {tool_desc}") + # 工具名清单(与 FC tools 字段互补;不逐条重复 description,避免双重曝光)。 + # 无 function-calling 能力的模型只能靠此清单知道有哪些工具,故保留名称。 + names = ", ".join(t.get('name', '') for t in plugins) + prompts.append( + "## 内置工具(已通过 function calling 的 tools 字段提供,调用前请先用 " + "tool_search 查询其详细用法与示例)\n" + f"工具名: {names}" + ) if skills: prompts.append("\n## 扩展技能(已加载详细指令):") @@ -883,8 +886,8 @@ multi 操作支持一次修改多处: # TUI 模式也支持 FC fc_tools = None try: - from .fc_tools import mcp_to_openai_tools - fc_tools = mcp_to_openai_tools(self.liugin_manager) + from .fc_tools import mcp_to_openai_tools, prune_tools + fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input) except Exception: pass response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=fc_tools) @@ -969,13 +972,11 @@ multi 操作支持一次修改多处: if not self.current_engine: response_result[0] = "错误: 当前没有可用的AI引擎,请检查ai_engines目录中的引擎插件" else: - # ── 收集 FC 工具定义 ── + # ── 收集 FC 工具定义(按当前输入动态裁剪)── fc_tools = None try: - from .fc_tools import mcp_to_openai_tools - fc_tools = mcp_to_openai_tools(self.liugin_manager) - if fc_tools: - fc_tools = fc_tools # 已经是列表 + from .fc_tools import mcp_to_openai_tools, prune_tools + fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input) except Exception: fc_tools = None diff --git a/xcli_core/fc_tools.py b/xcli_core/fc_tools.py index 1382e78..eebcc34 100644 --- a/xcli_core/fc_tools.py +++ b/xcli_core/fc_tools.py @@ -47,7 +47,117 @@ def mcp_to_openai_tools(plugin_manager) -> List[Dict]: builtin_tools = _get_builtin_tools() tools.extend(builtin_tools) - return tools + # ── 去重:同名工具只保留首个 ── + # 来源:get_mcp_tools() 的缓存里已含 memory/scheduler/lsp 等, + # 下方 _get_builtin_tools() 又追加了一遍,导致同名重复(约 17% 的 FC token 是垃圾)。 + # 按 function.name 去重,消除双重曝光,对真实工具无影响。 + seen = set() + deduped = [] + for t in tools: + fname = (t.get("function") or {}).get("name", "") + if fname in seen: + continue + seen.add(fname) + deduped.append(t) + return deduped + + +# ══════════════════════════════════════════════════════════════════ +# 动态工具裁剪(方案 B:按场景分层) +# +# 把 40+ 的 FC 工具全量发给模型,每轮要重发数千 token schema,且 +# 模型要在大列表里选对工具,决策面大、易选错、成本高。 +# 这里按「当前用户输入」的信号,只下发本轮相关的工具子集: +# - ALWAYS_KEEP:始终保留(含强制 tool_search 规则依赖、记忆、定时) +# - 外部桥接工具(含 "mcp__")默认保留(用户显式配置,误杀代价高) +# - 命中关键词信号才追加对应工具(代码/文件/git/shell/浏览器/搜索/子agent/工程化) +# 纯聊天时模型只背 ~10 个核心工具而非 40+,FC token 可降约 70%+。 +# ══════════════════════════════════════════════════════════════════ + +# 始终保留的工具(与系统提示里的强制规则/记忆机制强绑定) +ALWAYS_KEEP = { + "tool_search", # 系统提示的「强制工具查询规则」依赖它 + "memory", # 持久化记忆 + "scheduler", # 定时任务 +} + +# 信号:关键词集合 → 追加「以此前缀开头」的工具 +_SIGNAL_RULES = [ + # 代码 / 文件操作:Python 等源码信号命中时,连 pylsp 的 lsp__* 一起给 + ( + {"写", "创建", "新建", "修改", "编辑", "代码", "脚本", "函数", "bug", "修复", + "调试", "重构", "文件", "报错", "语法", "实现", "module", "def ", "import ", + "class ", "print(", ".py", ".js", ".ts", ".java", ".go", ".cpp", ".c", ".rs"}, + ["code_editor", "code_search", "file_manager", "lsp__", "syntax_check", "check"], + ), + # 版本控制 + ( + {"git", "提交", "commit", "分支", "branch", "版本", "diff", "stash", "merge", "rebase"}, + ["git_tools"], + ), + # Shell 命令 + ( + {"命令", "shell", "终端", "执行", "运行", "cmd", "bash", "powershell", "terminal"}, + ["cmd_executor"], + ), + # 浏览器 + ( + {"浏览器", "网页", "点击", "browser", "打开网页", "截图", "scroll", "页面"}, + ["browser_auto"], + ), + # 联网 / 搜索 / HTTP + ( + {"搜索", "联网", "http", "请求", "url", "curl", "api", "查一下", "上网", "fetch"}, + ["network_tools", "ai_search"], + ), + # 子 Agent 并行 + ( + {"子agent", "并行", "subagent", "sub_agent", "多智能体"}, + ["sub_agent"], + ), + # 工程化自动化 + ( + {"测试", "test", "lint", "build", "自动化", "工程化", "ci", "部署", "deploy"}, + ["auto_engineer"], + ), +] + + +def prune_tools(fc_tools: List[Dict], user_input: str) -> List[Dict]: + """ + 按当前用户输入动态裁剪 FC 工具子集。 + + 返回保留下来的工具列表。无输入或空列表时原样返回(不裁剪)。 + 规则见模块顶部 _SIGNAL_RULES / ALWAYS_KEEP 注释。 + """ + if not fc_tools: + return fc_tools + + text = (user_input or "").lower() + + keep_prefixes = set() + for keywords, prefixes in _SIGNAL_RULES: + if any(kw in text for kw in keywords): + keep_prefixes.update(prefixes) + + out = [] + for t in fc_tools: + name = (t.get("function") or {}).get("name", "") + # 1) 常驻工具 + if name in ALWAYS_KEEP: + out.append(t) + continue + # 2) 外部桥接工具(用户显式配置)默认保留,避免误杀 + if "mcp__" in name: + out.append(t) + continue + # 3) 命中当前轮信号 → 追加(前缀匹配,如 lsp__diagnostics 命中 "lsp__") + if any(name.startswith(p) for p in keep_prefixes): + out.append(t) + continue + # 4) 其余(未命中信号的低频工具)本轮裁剪掉 + + return out def _extract_fc_def(plugin) -> Optional[Dict]: