feat: FC 工具去重 + 按场景动态裁剪,降低 AI 每轮负担
方案 A(去重): - mcp_to_openai_tools 末尾按 function.name 去重,消除 cache+builtin 叠加 造成的同名重复(memory/scheduler/lsp__* 等),FC 工具 42→36,去掉 约 14% 的垃圾 schema token - get_liugin_usage_prompts 的 plugins 段由逐条 name:desc 改为工具名清单, 消除与 FC tools 字段的双重曝光,保留无 FC 模型的回退清单 方案 B(分层裁剪): - 新增 prune_tools(fc_tools, user_input): 常驻 ALWAYS(tool_search/memory/ scheduler) + 外部桥接(mcp__)默认保留 + 按关键词信号追加对应工具 (代码/文件->code_editor+file_manager+lsp__*; git/shell/浏览器/搜索/子agent/ 工程化各按信号) - cli_core 两处 FC 构建点(TUI/CLI)接入 prune_tools(current_input) 实测(AICLI): 纯聊天 FC token ~4310->285(降93%),写代码 ~973(降77%) 测试: 新增 tests/test_prune_tools.py (7 passed),全量 324 passed 无回归
This commit is contained in:
1 parent
1054932921
commit
b577b0ff21
3 files changed
+239
-13
No files matched your search
@@ -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([], "任意输入") == []
|
||||||
+13
-12
@@ -310,11 +310,14 @@ multi 操作支持一次修改多处:
|
|||||||
plugins.append(tool)
|
plugins.append(tool)
|
||||||
|
|
||||||
if plugins:
|
if plugins:
|
||||||
prompts.append("## 内置工具(使用 tool_search 查询详细用法):")
|
# 工具名清单(与 FC tools 字段互补;不逐条重复 description,避免双重曝光)。
|
||||||
for tool in plugins:
|
# 无 function-calling 能力的模型只能靠此清单知道有哪些工具,故保留名称。
|
||||||
tool_name = tool.get('name', '')
|
names = ", ".join(t.get('name', '') for t in plugins)
|
||||||
tool_desc = tool.get('description', '')
|
prompts.append(
|
||||||
prompts.append(f"- {tool_name}: {tool_desc}")
|
"## 内置工具(已通过 function calling 的 tools 字段提供,调用前请先用 "
|
||||||
|
"tool_search 查询其详细用法与示例)\n"
|
||||||
|
f"工具名: {names}"
|
||||||
|
)
|
||||||
|
|
||||||
if skills:
|
if skills:
|
||||||
prompts.append("\n## 扩展技能(已加载详细指令):")
|
prompts.append("\n## 扩展技能(已加载详细指令):")
|
||||||
@@ -883,8 +886,8 @@ multi 操作支持一次修改多处:
|
|||||||
# TUI 模式也支持 FC
|
# TUI 模式也支持 FC
|
||||||
fc_tools = None
|
fc_tools = None
|
||||||
try:
|
try:
|
||||||
from .fc_tools import mcp_to_openai_tools
|
from .fc_tools import mcp_to_openai_tools, prune_tools
|
||||||
fc_tools = mcp_to_openai_tools(self.liugin_manager)
|
fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input)
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=fc_tools)
|
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:
|
if not self.current_engine:
|
||||||
response_result[0] = "错误: 当前没有可用的AI引擎,请检查ai_engines目录中的引擎插件"
|
response_result[0] = "错误: 当前没有可用的AI引擎,请检查ai_engines目录中的引擎插件"
|
||||||
else:
|
else:
|
||||||
# ── 收集 FC 工具定义 ──
|
# ── 收集 FC 工具定义(按当前输入动态裁剪)──
|
||||||
fc_tools = None
|
fc_tools = None
|
||||||
try:
|
try:
|
||||||
from .fc_tools import mcp_to_openai_tools
|
from .fc_tools import mcp_to_openai_tools, prune_tools
|
||||||
fc_tools = mcp_to_openai_tools(self.liugin_manager)
|
fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input)
|
||||||
if fc_tools:
|
|
||||||
fc_tools = fc_tools # 已经是列表
|
|
||||||
except Exception:
|
except Exception:
|
||||||
fc_tools = None
|
fc_tools = None
|
||||||
|
|
||||||
|
|||||||
+111
-1
@@ -47,7 +47,117 @@ def mcp_to_openai_tools(plugin_manager) -> List[Dict]:
|
|||||||
builtin_tools = _get_builtin_tools()
|
builtin_tools = _get_builtin_tools()
|
||||||
tools.extend(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]:
|
def _extract_fc_def(plugin) -> Optional[Dict]:
|
||||||
|
|||||||
Reference in new issue
Block a user