From 3a86da744449409f03060016ab088a8137a0414a Mon Sep 17 00:00:00 2001 From: shuiyu67 Date: Wed, 5 Aug 2026 15:57:38 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E5=90=AF=E5=8A=A8=E8=BE=93=E5=87=BA?= =?UTF-8?q?=E7=98=A6=E8=BA=AB=20+=20FC=20=E8=83=BD=E5=8A=9B=E6=8E=A2?= =?UTF-8?q?=E6=B5=8B=E7=BC=93=E5=AD=98=EF=BC=8C=E4=BF=AE=E5=A4=8D=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E5=A5=97=E4=BB=B6=E5=A4=B1=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 自用体验（同时以人类和被接入 AI 的视角）暴露出三个问题，一并处理： 1. 启动太吵：117 行 / 16983 字符的开屏，人还没开始用就先刷屏 - 完整 ASCII 开机动画改为 --logo / --banner 按需召唤，默认两行紧凑标识 - 21 行命令说明折叠成一行「/help 全部命令 · /quit 退出 · ...」， 完整速查表挪进 /help（新增 show_quick_commands） - 插件/引擎/工具的逐条加载日志改走 vprint，由 --verbose / -v / XCLI_VERBOSE=1 召唤；工具加载汇总成一行 - 日志控制台 handler 降到 WARNING（文件仍留全量 INFO），并静音 httpx/httpcore/urllib3 等第三方心跳日志 => 默认启动 10 行 / 475 字符，--verbose 仍可看到全部 36 行细节 2. 不支持 FC 的模型每次会话都白撞一发 500 讯飞 xophunyuan7bmt 不支持 Function Calling，请求带 tools 字段必报 500 Invalid Params，虽有降级重试兜底，但每个新会话都要先错一次。 - 新增 xcli_core/fc_tools 能力缓存（按 base_url::model 落盘 .fc_support.json） - 降级重试成功后记住该模型不支持 FC，后续直接不发 tools - 降级也失败时不写缓存，避免把别的故障误判成 FC 问题 - 新增 /fc [reset] 查看与清除缓存 3. diff 弹窗模式每次启动都要重新问一遍 - 首次选择后经 config.json 持久化，新增 /diff [popup|inline] 随时切换 测试： - 新增 tests/test_startup_and_fc.py（27 项，含假 HTTP 验证降级与缓存链路） - 修复 tests/test_plugins.py 长期失效：插件入口类早已由 Plugin 改名 Liugin， 8 处 fixture 未跟进导致 65 个用例直接 error；断言也未适配 ToolResult 返回值 与「❌」错误前缀。改为 plugin_class() / as_text() / is_error() 语义判断。 - 全量：217 passed（此前 152 passed + 65 error + 27 failed） --- .gitignore | 6 + ai_engines/ollama_engine.py | 8 +- ai_engines/openai_engine.py | 37 ++++- tests/test_plugins.py | 199 ++++++++++++----------- tests/test_startup_and_fc.py | 297 +++++++++++++++++++++++++++++++++++ unified_tool_manager.py | 13 +- xcli_core/cli_base.py | 21 +-- xcli_core/cli_core.py | 150 +++++++++++------- xcli_core/cli_display.py | 34 ++++ xcli_core/config.py | 26 ++- xcli_core/fc_tools.py | 80 ++++++++++ xcli_core/verbose.py | 24 +++ 12 files changed, 728 insertions(+), 167 deletions(-) create mode 100644 tests/test_startup_and_fc.py create mode 100644 xcli_core/verbose.py diff --git a/.gitignore b/.gitignore index 0896722..9c8dfe9 100644 --- a/.gitignore +++ b/.gitignore @@ -50,3 +50,9 @@ screenshots/gen_*.png # v8.0 TUI 快照产物（运行时生成，供 AI 自验/预览） .tui_snapshots/ + +# FC 能力探测缓存（运行时按 base_url+model 生成，机器相关） +.fc_support.json + +# pytest 缓存 +.pytest_cache/ diff --git a/ai_engines/ollama_engine.py b/ai_engines/ollama_engine.py index 79d2037..676aa33 100644 --- a/ai_engines/ollama_engine.py +++ b/ai_engines/ollama_engine.py @@ -22,6 +22,12 @@ current_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(current_dir) sys.path.insert(0, project_root) +try: + from xcli_core.verbose import vprint as _vprint +except Exception: + def _vprint(*args, **kwargs): + pass + # 特殊语法解析（配置模型时可交互式询问用户） try: from xcli_core.config import resolve_input_value @@ -116,7 +122,7 @@ Ollama AI引擎插件帮助信息 self.is_service_running = self.check_ollama_service() if self.is_service_running: - print(f"{Fore.GREEN}Ollama服务检测成功,引擎已启用{Style.RESET_ALL}") + _vprint(f"{Fore.GREEN}Ollama服务检测成功,引擎已启用{Style.RESET_ALL}") else: print(f"{Fore.RED}Ollama服务未运行,引擎将不可用{Style.RESET_ALL}") diff --git a/ai_engines/openai_engine.py b/ai_engines/openai_engine.py index 04ee819..3bf8723 100644 --- a/ai_engines/openai_engine.py +++ b/ai_engines/openai_engine.py @@ -19,6 +19,12 @@ except Exception: def get_system_config(key, default=None): return default +try: + from xcli_core.verbose import vprint +except Exception: + def vprint(*args, **kwargs): + pass + try: from xcli_core.tool_result import normalize_tool_text except Exception: @@ -147,7 +153,7 @@ OpenAI 兼容格式引擎帮助信息 if not self.base_url: print(f"{Fore.YELLOW}警告: openai 引擎的 base_url 未设置{Style.RESET_ALL}") else: - print(f"{Fore.GREEN}OpenAI 兼容引擎已启用 (base_url: {self.base_url}){Style.RESET_ALL}") + vprint(f"{Fore.GREEN}OpenAI 兼容引擎已启用 (base_url: {self.base_url}){Style.RESET_ALL}") def _get_headers(self): """构建请求头""" @@ -167,6 +173,26 @@ OpenAI 兼容格式引擎帮助信息 url = url.rstrip('/') + '/chat/completions' return url + # ── FC 能力探测缓存 ── + + def _fc_supported(self): + """当前 (base_url, model) 是否支持 function calling（未探测过默认支持）""" + try: + from xcli_core.fc_tools import supports_fc + return supports_fc(self.base_url, self.model) + except Exception: + return True + + def _mark_fc_unsupported(self): + """降级成功后记住该模型不支持 FC，后续请求不再带 tools""" + try: + from xcli_core.fc_tools import mark_fc_unsupported + if mark_fc_unsupported(self.base_url, self.model): + print(f"{Fore.CYAN}已记住 {self.model} 不支持 FC，后续请求将跳过工具字段" + f"（/fc reset 可清除）{Style.RESET_ALL}") + except Exception: + pass + def generate_response(self, user_input, tool_results=None, system_prompt=None, tools=None): """ 生成 AI 响应 @@ -196,7 +222,8 @@ OpenAI 兼容格式引擎帮助信息 } # ── Function Calling: 注入工具定义（统一纯官方格式）── - if tools: + # 已探测过不支持 FC 的模型直接跳过 tools，省掉每次白撞一次 500 + if tools and self._fc_supported(): pure_tools = [] for t in tools: if not isinstance(t, dict): @@ -260,12 +287,13 @@ OpenAI 兼容格式引擎帮助信息 return f"API 端点不存在，请检查 base_url 是否正确: {api_url}" elif response.status_code == 400: # 工具定义可能不兼容，降级为无工具重试 - if tools and "tools" in str(response.text).lower(): + if body.get("tools") and "tools" in str(response.text).lower(): print(f"{Fore.YELLOW}FC 不支持，降级为普通模式{Style.RESET_ALL}") body.pop("tools", None) body.pop("tool_choice", None) retry = requests.post(url=api_url, json=body, headers=self._get_headers(), timeout=None) if retry.status_code == 200: + self._mark_fc_unsupported() rd = retry.json() if "choices" in rd and rd["choices"]: ai_response = rd["choices"][0].get("message", {}).get("content", "") @@ -275,12 +303,13 @@ OpenAI 兼容格式引擎帮助信息 # 500 等错误：部分服务（如讯飞星火经 one-api 转发）不支持 FC 工具， # 会直接报 500 Invalid Params —— 尝试去掉 tools 降级重试一次 lower = str(response.text).lower() - if tools and ("xunfei" in lower or "invalid params" in lower or "requestparamserror" in lower): + if body.get("tools") and ("xunfei" in lower or "invalid params" in lower or "requestparamserror" in lower): print(f"{Fore.YELLOW}该接口可能不支持 FC 工具，降级为普通模式重试...{Style.RESET_ALL}") body.pop("tools", None) body.pop("tool_choice", None) retry = requests.post(url=api_url, json=body, headers=self._get_headers(), timeout=None) if retry.status_code == 200: + self._mark_fc_unsupported() rd = retry.json() retry_usage = rd.get("usage", {}) if isinstance(retry_usage, dict) and retry_usage.get("prompt_tokens"): diff --git a/tests/test_plugins.py b/tests/test_plugins.py index b0b0ed6..bff2a78 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -29,6 +29,29 @@ def load_module(name, path): return mod +def plugin_class(mod): + """取插件入口类（项目里叫 Liugin，历史上叫 Plugin，两种都认）""" + cls = getattr(mod, "Liugin", None) or getattr(mod, "Plugin", None) + assert cls is not None, f"{mod.__name__}: 缺少 Plugin/Liugin 类" + return cls + + +def as_text(result): + """插件返回值统一转文本（可能是 str，也可能是 ToolResult）""" + return result if isinstance(result, str) else str(result) + + +ERROR_MARKERS = ("❌", "错误", "失败", "缺少", "不存在", "无效", "请提供", "用法") + + +def is_error(result): + """判断插件返回是否表示失败（不依赖具体文案措辞）""" + success = getattr(result, "success", None) + if success is not None: + return not success + return any(m in as_text(result) for m in ERROR_MARKERS) + + # ══════════════════════════════════════ # 插件加载测试 # ══════════════════════════════════════ @@ -89,7 +112,7 @@ class TestCodeEditor: @pytest.fixture def editor(self): mod = load_module("code_editor", os.path.join(PLUGINS_DIR, "code_editor.py")) - return mod.Plugin() + return plugin_class(mod)() @pytest.fixture def tmp_file(self, tmp_path): @@ -99,47 +122,47 @@ class TestCodeEditor: def test_read_range(self, editor, tmp_file): result = editor.handle(f"read_range {tmp_file} 1 2") - assert "hello" in result - assert "1 |" in result + assert "hello" in as_text(result) + assert "1 |" in as_text(result) def test_edit(self, editor, tmp_file): result = editor.handle(f"edit {tmp_file} print('hello') <<<>>> print('hi')") - assert "已编辑" in result + assert "已编辑" in as_text(result) content = open(tmp_file).read() assert "print('hi')" in content def test_edit_not_found(self, editor, tmp_file): result = editor.handle(f"edit {tmp_file} nonexistent_code <<<>>> new_code") - assert "错误" in result + assert is_error(result) def test_create(self, editor, tmp_path): new_file = str(tmp_path / "new.py") result = editor.handle(f"create {new_file} import os") - assert "已创建" in result + assert "已创建" in as_text(result) assert os.path.exists(new_file) def test_diff_text(self, editor): result = editor.handle("diff_text old <<<>>> new") - assert "变更预览" in result + assert "变更预览" in as_text(result) def test_ast_info(self, editor, tmp_file): result = editor.handle(f"ast_info {tmp_file}") - assert "def hello" in result - assert "def world" in result + assert "def hello" in as_text(result) + assert "def world" in as_text(result) def test_stats(self, editor, tmp_path): result = editor.handle(f"stats {tmp_path}") - assert "代码统计" in result - assert "文件" in result + assert "代码统计" in as_text(result) + assert "文件" in as_text(result) def test_structure(self, editor, tmp_path): result = editor.handle(f"structure {tmp_path} 1") - assert "" in result + assert "" in as_text(result) def test_find(self, editor, tmp_path): (tmp_path / "a.py").write_text("x = 1\ny = 2\n") result = editor.handle(f"find {tmp_path} x *.py") - assert "x = 1" in result + assert "x = 1" in as_text(result) class TestGitTools: @@ -148,11 +171,11 @@ class TestGitTools: @pytest.fixture def git(self): mod = load_module("git_tools", os.path.join(PLUGINS_DIR, "git_tools.py")) - return mod.Plugin() + return plugin_class(mod)() def test_status(self, git): result = git.handle("status") - assert "Git" in result or "干净" in result or "分支" in result + assert "Git" in as_text(result) or "干净" in as_text(result) or "分支" in as_text(result) def test_log(self, git): result = git.handle("log --oneline 3") @@ -161,7 +184,7 @@ class TestGitTools: def test_branch(self, git): result = git.handle("branch") - assert "master" in result or "main" in result + assert "master" in as_text(result) or "main" in as_text(result) class TestCmdExecutor: @@ -170,27 +193,27 @@ class TestCmdExecutor: @pytest.fixture def cmd(self): mod = load_module("cmd_executor", os.path.join(PLUGINS_DIR, "cmd_executor.py")) - return mod.Plugin() + return plugin_class(mod)() def test_echo(self, cmd): result = cmd.handle("run echo test_ok") - assert "test_ok" in result + assert "test_ok" in as_text(result) def test_python_version(self, cmd): result = cmd.handle("run python3 --version") - assert "Python" in result + assert "Python" in as_text(result) def test_blocked_command(self, cmd): result = cmd.handle("run rm -rf /") - assert "拦截" in result + assert "拦截" in as_text(result) def test_timeout(self, cmd): result = cmd.handle("run echo fast --timeout 5") - assert "fast" in result + assert "fast" in as_text(result) def test_empty_command(self, cmd): result = cmd.handle("run") - assert "错误" in result + assert is_error(result) class TestFileManager: @@ -199,53 +222,53 @@ class TestFileManager: @pytest.fixture def fm(self): mod = load_module("file_manager", os.path.join(PLUGINS_DIR, "file_manager.py")) - return mod.Plugin() + return plugin_class(mod)() def test_list(self, fm, tmp_path): (tmp_path / "a.txt").write_text("hi") result = fm.handle(f"list {tmp_path}") - assert "a.txt" in result + assert "a.txt" in as_text(result) def test_read(self, fm, tmp_path): f = tmp_path / "test.txt" f.write_text("hello world") result = fm.handle(f"read {f}") - assert "hello world" in result + assert "hello world" in as_text(result) def test_write_and_read(self, fm, tmp_path): f = tmp_path / "new.txt" result = fm.handle(f"write {f} -c test_content") - assert "已写入" in result + assert "已写入" in as_text(result) result = fm.handle(f"read {f}") - assert "test_content" in result + assert "test_content" in as_text(result) def test_copy(self, fm, tmp_path): src = tmp_path / "src.txt" src.write_text("data") dst = tmp_path / "dst.txt" result = fm.handle(f"copy {src} {dst}") - assert "已复制" in result + assert "已复制" in as_text(result) assert dst.exists() def test_delete(self, fm, tmp_path): f = tmp_path / "del.txt" f.write_text("bye") result = fm.handle(f"delete {f}") - assert "已删除" in result + assert "已删除" in as_text(result) assert not f.exists() def test_mkdir(self, fm, tmp_path): d = tmp_path / "newdir" result = fm.handle(f"mkdir {d}") - assert "已创建" in result + assert "已创建" in as_text(result) assert d.is_dir() def test_info(self, fm, tmp_path): f = tmp_path / "info.txt" f.write_text("x") result = fm.handle(f"info {f}") - assert "路径" in result - assert "大小" in result + assert "路径" in as_text(result) + assert "大小" in as_text(result) class TestAutoEngineer: @@ -254,23 +277,23 @@ class TestAutoEngineer: @pytest.fixture def eng(self): mod = load_module("auto_engineer", os.path.join(PLUGINS_DIR, "auto_engineer.py")) - return mod.Plugin() + return plugin_class(mod)() def test_metrics(self, eng, tmp_path): (tmp_path / "a.py").write_text("x = 1\n# comment\n\ny = 2\n") result = eng.handle(f"metrics {tmp_path}") - assert "代码指标" in result - assert "文件" in result + assert "代码指标" in as_text(result) + assert "文件" in as_text(result) def test_security_clean(self, eng, tmp_path): (tmp_path / "safe.py").write_text("x = 1\nprint(x)\n") result = eng.handle(f"security {tmp_path}") - assert "没有发现" in result or "风险" in result + assert "没有发现" in as_text(result) or "风险" in as_text(result) def test_security_risk(self, eng, tmp_path): (tmp_path / "risk.py").write_text('password = "secret123"\neval("1+1")\n') result = eng.handle(f"security {tmp_path}") - assert "风险" in result + assert "风险" in as_text(result) def test_complexity(self, eng, tmp_path): code = "def f():\n if True:\n if True:\n if True:\n pass\n" @@ -292,34 +315,34 @@ class TestTaskManager: def test_add(self, tm): result = tm.handle("add 测试任务") - assert "已添加" in result + assert "已添加" in as_text(result) def test_list_empty(self, tm): result = tm.handle("list") - assert "没有任务" in result + assert "没有任务" in as_text(result) def test_add_and_list(self, tm): tm.handle("add 任务A") tm.handle("add 任务B") result = tm.handle("list") - assert "任务A" in result - assert "任务B" in result + assert "任务A" in as_text(result) + assert "任务B" in as_text(result) def test_complete(self, tm): tm.handle("add 任务X") result = tm.handle("complete 1") - assert "已更新" in result + assert "已更新" in as_text(result) def test_delete(self, tm): tm.handle("add 任务Y") result = tm.handle("delete 1") - assert "已删除" in result + assert "已删除" in as_text(result) def test_clear(self, tm): tm.handle("add A") tm.handle("complete 1") result = tm.handle("clear") - assert "已清除" in result + assert "已清除" in as_text(result) # ══════════════════════════════════════ @@ -332,19 +355,19 @@ class TestBrowserAuto: @pytest.fixture def browser(self): mod = load_module("browser_auto", os.path.join(PLUGINS_DIR, "browser_auto.py")) - return mod.Plugin() + return plugin_class(mod)() def test_status_stopped(self, browser): result = browser.handle("status") - assert "未运行" in result + assert "未运行" in as_text(result) def test_invalid_operation(self, browser): result = browser.handle("nonexistent") - assert "错误" in result + assert is_error(result) def test_empty_args(self, browser): result = browser.handle("") - assert "错误" in result + assert is_error(result) def test_tool_info(self, browser): info = browser.get_tool_info() @@ -363,7 +386,7 @@ class TestGuiAuto: @pytest.fixture def gui(self): mod = load_module("gui_auto", os.path.join(PLUGINS_DIR, "gui_auto.py")) - return mod.Plugin() + return plugin_class(mod)() def test_tool_info(self, gui): info = gui.get_tool_info() @@ -380,88 +403,88 @@ class TestGuiAuto: def test_convert_mcp_args(self, gui): args = {"operation": "click", "target": "确定"} result = gui.convert_mcp_args(args) - assert "click" in result - assert "确定" in result + assert "click" in as_text(result) + assert "确定" in as_text(result) def test_convert_mcp_args_xy(self, gui): args = {"operation": "click", "x": 100, "y": 200} result = gui.convert_mcp_args(args) - assert "--xy" in result - assert "100" in result + assert "--xy" in as_text(result) + assert "100" in as_text(result) def test_convert_mcp_args_snapshot(self, gui): args = {"operation": "snapshot", "depth": 3} result = gui.convert_mcp_args(args) - assert "snapshot" in result - assert "3" in result + assert "snapshot" in as_text(result) + assert "3" in as_text(result) def test_empty_args(self, gui): result = gui.handle("") - assert "错误" in result + assert is_error(result) def test_invalid_operation(self, gui): result = gui.handle("nonexistent_op") - assert "不支持" in result + assert "不支持" in as_text(result) def test_click_no_args(self, gui): result = gui.handle("click") - assert "错误" in result + assert is_error(result) def test_type_no_args(self, gui): result = gui.handle("type") - assert "错误" in result + assert is_error(result) def test_type_missing_text(self, gui): result = gui.handle("type 按钮") - assert "错误" in result + assert is_error(result) def test_find_no_args(self, gui): result = gui.handle("find") - assert "错误" in result + assert is_error(result) def test_findall_no_args(self, gui): result = gui.handle("findall") - assert "错误" in result + assert is_error(result) def test_info_no_args(self, gui): result = gui.handle("info") - assert "错误" in result + assert is_error(result) def test_value_no_args(self, gui): result = gui.handle("value") - assert "错误" in result + assert is_error(result) def test_tree_no_args(self, gui): result = gui.handle("tree") - assert "错误" in result + assert is_error(result) def test_wait_no_args(self, gui): result = gui.handle("wait") - assert "错误" in result + assert is_error(result) def test_exists_no_args(self, gui): result = gui.handle("exists") - assert "错误" in result + assert is_error(result) def test_highlight_no_args(self, gui): result = gui.handle("highlight") - assert "错误" in result + assert is_error(result) def test_state_no_args(self, gui): result = gui.handle("state") - assert "错误" in result + assert is_error(result) def test_keys_no_args(self, gui): result = gui.handle("keys") - assert "错误" in result + assert is_error(result) def test_focus_no_args(self, gui): result = gui.handle("focus") - assert "错误" in result + assert is_error(result) def test_click_xy_format_error(self, gui): result = gui.handle("click --xy abc") - assert "错误" in result + assert is_error(result) def test_format_element_info(self, gui): el = { @@ -472,9 +495,9 @@ class TestGuiAuto: "enabled": True, "focused": False } result = gui._format_element_info(el) - assert "测试按钮" in result - assert "Button" in result - assert "btn_ok" in result + assert "测试按钮" in as_text(result) + assert "Button" in as_text(result) + assert "btn_ok" in as_text(result) def test_format_element_info_detailed(self, gui): el = { @@ -485,8 +508,8 @@ class TestGuiAuto: "right": 200, "bottom": 25} } result = gui._format_element_info(el, detailed=True) - assert "输入框" in result - assert "hello" in result + assert "输入框" in as_text(result) + assert "hello" in as_text(result) def test_all_operations_route(self, gui): """所有操作都能正确路由""" @@ -504,28 +527,28 @@ class TestSubAgent: @pytest.fixture def sa(self): mod = load_module("sub_agent", os.path.join(PLUGINS_DIR, "sub_agent.py")) - return mod.Plugin() + return plugin_class(mod)() def test_list_empty(self, sa): result = sa.handle("list") - assert "没有" in result + assert "没有" in as_text(result) def test_stats_empty(self, sa): result = sa.handle("stats") - assert "总数: 0" in result + assert "总数: 0" in as_text(result) def test_spawn(self, sa): result = sa.handle("spawn tester 测试任务") - assert "已创建" in result - assert "agent_tester" in result + assert "已创建" in as_text(result) + assert "agent_tester" in as_text(result) def test_spawn_and_list(self, sa): sa.handle("spawn a 任务A") sa.handle("spawn b 任务B") import time; time.sleep(0.1) result = sa.handle("list") - assert "agent_a" in result - assert "agent_b" in result + assert "agent_a" in as_text(result) + assert "agent_b" in as_text(result) def test_status(self, sa): sa.handle("spawn x 某任务") @@ -533,24 +556,24 @@ class TestSubAgent: # 获取 agent ID agent_id = list(sa.agents.keys())[0] result = sa.handle(f"status {agent_id}") - assert "名称" in result + assert "名称" in as_text(result) def test_kill(self, sa): sa.handle("spawn y 要终止的任务") import time; time.sleep(0.1) agent_id = list(sa.agents.keys())[0] result = sa.handle(f"kill {agent_id}") - assert "已终止" in result + assert "已终止" in as_text(result) def test_clean(self, sa): sa.handle("spawn z 清理测试") import time; time.sleep(0.1) result = sa.handle("clean") - assert "已清理" in result + assert "已清理" in as_text(result) def test_invalid_agent(self, sa): result = sa.handle("status nonexistent") - assert "未找到" in result + assert "未找到" in as_text(result) # ══════════════════════════════════════ diff --git a/tests/test_startup_and_fc.py b/tests/test_startup_and_fc.py new file mode 100644 index 0000000..1910d84 --- /dev/null +++ b/tests/test_startup_and_fc.py @@ -0,0 +1,297 @@ +"""启动输出瘦身 + FC 能力探测缓存 回归测试 + +覆盖两件事： +1. 无 FC 能力的模型（如讯飞 xophunyuan7bmt）被 500 打回后会被记住， + 之后不再每次白撞一发 500 试探。 +2. 启动输出从 117 行折叠到 10 行，详细日志改由 --verbose 召唤。 +""" +import contextlib +import io +import json +import os +import sys +import types + +import pytest + +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if PROJECT_ROOT not in sys.path: + sys.path.insert(0, PROJECT_ROOT) + +BASE_URL = "https://maas-api.cn-huabei-1.xf-yun.com/v2" +MODEL = "xophunyuan7bmt" + +TOOL = { + "type": "function", + "function": { + "name": "read_file", + "description": "读文件", + "parameters": {"type": "object", "properties": {}}, + }, +} + +OK_BODY = {"choices": [{"message": {"content": "好"}}], "usage": {"prompt_tokens": 12}} +ERR_500 = {"error": {"message": "xunfei response error: sid: xxx msg: RequestParamsError:Invalid Params"}} + + +@pytest.fixture +def fc(tmp_path): + """把 FC 缓存文件隔离到临时目录，避免污染真实运行时缓存""" + from xcli_core import fc_tools + + old_file, old_mem = fc_tools._FC_CACHE_FILE, fc_tools._fc_cache + fc_tools._FC_CACHE_FILE = str(tmp_path / ".fc_support.json") + fc_tools._fc_cache = None + try: + yield fc_tools + finally: + fc_tools._FC_CACHE_FILE, fc_tools._fc_cache = old_file, old_mem + + +class FakeResp: + def __init__(self, code, payload): + self.status_code = code + self._payload = payload + self.text = json.dumps(payload, ensure_ascii=False) + + def json(self): + return self._payload + + +def make_engine(model=MODEL): + """不走 __init__（会读配置/连网），直接摆一个可用实例""" + import ai_engines.openai_engine as oe + + eng = oe.OpenaiAI.__new__(oe.OpenaiAI) + eng.name = "openai" + eng.cli = None + eng.api_key = "k" + eng.base_url = BASE_URL + eng.model = model + eng.max_history = 10 + eng.conversation_history = [] + eng.shared_conversation_history = None + eng.last_prompt_tokens = None + eng.thinking_start_marker = "" + eng.thinking_end_marker = "" + return eng + + +@pytest.fixture +def http(monkeypatch): + """替换 openai_engine 里的 requests，记录每次请求体""" + import ai_engines.openai_engine as oe + + calls = [] + + def install(responder): + def fake_post(url=None, json=None, headers=None, timeout=None, **kw): + calls.append(dict(json)) + return responder(json) + + monkeypatch.setattr( + oe, "requests", + types.SimpleNamespace(post=fake_post, exceptions=oe.requests.exceptions), + ) + return calls + + install.calls = calls + return install + + +# ══════════════════════════════════════ +# 1. FC 能力探测缓存 +# ══════════════════════════════════════ + +class TestFCCache: + def test_default_assume_supported(self, fc): + assert fc.supports_fc(BASE_URL, MODEL) is True + + def test_mark_and_remember(self, fc): + assert fc.mark_fc_unsupported(BASE_URL, MODEL) is True + assert fc.supports_fc(BASE_URL, MODEL) is False + + def test_repeat_mark_not_reported(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + # 第二次返回 False，调用方据此不重复刷屏提示 + assert fc.mark_fc_unsupported(BASE_URL, MODEL) is False + + def test_persist_across_restart(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + assert os.path.exists(fc._FC_CACHE_FILE) + fc._fc_cache = None # 模拟进程重启 + assert fc.supports_fc(BASE_URL, MODEL) is False + + def test_other_model_unaffected(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + assert fc.supports_fc(BASE_URL, "spark-4.0-ultra") is True + + def test_trailing_slash_normalized(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + assert fc.supports_fc(BASE_URL + "/", MODEL) is False + + def test_get_cache_returns_copy(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + snap = fc.get_fc_cache() + assert isinstance(snap, dict) and snap is not fc._fc_cache + + def test_reset(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + assert fc.reset_fc_cache() == 1 + assert fc.supports_fc(BASE_URL, MODEL) is True + + def test_reset_single_entry(self, fc): + fc.mark_fc_unsupported(BASE_URL, MODEL) + fc.mark_fc_unsupported(BASE_URL, "other") + assert fc.reset_fc_cache(BASE_URL, MODEL) == 1 + assert fc.supports_fc(BASE_URL, MODEL) is True + assert fc.supports_fc(BASE_URL, "other") is False + + def test_corrupted_file_does_not_crash(self, fc): + with open(fc._FC_CACHE_FILE, "w", encoding="utf-8") as f: + f.write("{ not json") + fc._fc_cache = None + assert fc.supports_fc(BASE_URL, MODEL) is True + + +# ══════════════════════════════════════ +# 2. openai 引擎接线（假 HTTP，不动真实接口） +# ══════════════════════════════════════ + +class TestOpenAIDegrade: + def test_first_call_degrades_and_caches(self, fc, http): + calls = http(lambda body: FakeResp(500, ERR_500) if "tools" in body else FakeResp(200, OK_BODY)) + + reply = make_engine().generate_response("说'好'", tools=[TOOL]) + + assert len(calls) == 2, f"应为 带tools→500 + 降级重试，实际 {len(calls)} 次" + assert "tools" in calls[0] + assert "tools" not in calls[1] and "tool_choice" not in calls[1] + assert reply == "好" + assert fc.supports_fc(BASE_URL, MODEL) is False, "降级成功后应记住该模型不支持 FC" + + def test_second_call_skips_probe(self, fc, http): + fc.mark_fc_unsupported(BASE_URL, MODEL) + calls = http(lambda body: FakeResp(500, ERR_500) if "tools" in body else FakeResp(200, OK_BODY)) + + reply = make_engine().generate_response("再说一次", tools=[TOOL]) + + assert len(calls) == 1, f"已知不支持 FC 时不该再撞 500，实际 {len(calls)} 次" + assert "tools" not in calls[0] + assert reply == "好" + + def test_unmarked_model_still_sends_tools(self, fc, http): + calls = http(lambda body: FakeResp(200, OK_BODY)) + + make_engine(model="gpt-4o").generate_response("hi", tools=[TOOL]) + + assert len(calls) == 1 and "tools" in calls[0] + + def test_tools_payload_is_pure_official_format(self, fc, http): + calls = http(lambda body: FakeResp(200, OK_BODY)) + + make_engine(model="gpt-4o").generate_response("hi", tools=[TOOL]) + + tool = calls[0]["tools"][0] + assert tool.get("type") == "function" + assert set(tool.keys()) == {"type", "function"} + assert calls[0].get("tool_choice") == "auto" + + def test_failed_degrade_does_not_cache(self, fc, http): + http(lambda body: FakeResp(500, ERR_500)) + + reply = make_engine().generate_response("hi", tools=[TOOL]) + + assert fc.supports_fc(BASE_URL, MODEL) is True, "降级也失败说明不是 FC 的锅，不能误判" + assert "API 调用失败" in str(reply) + + +# ══════════════════════════════════════ +# 3. 启动输出瘦身 +# ══════════════════════════════════════ + +def _read(*parts): + with open(os.path.join(PROJECT_ROOT, *parts), encoding="utf-8") as f: + return f.read() + + +class TestStartupSlim: + def test_no_21_line_help_dump(self): + src = _read("xcli_core", "cli_core.py") + run_body = src.split(" def run(self):", 1)[1].split(" def _get_code_editor_plugin", 1)[0] + assert run_body.count("输入 '/") == 0, "启动不该再逐条打印命令说明" + assert "/help 全部命令" in run_body + + def test_logo_is_opt_in(self): + src = _read("xcli_core", "cli_core.py") + assert "'--logo' in sys.argv" in src + assert "--banner" in src + + def test_quick_commands_moved_into_help(self): + from xcli_core import cli_display + + names = [n for n in dir(cli_display) if n.endswith("Mixin")] + assert names, "cli_display 里应有 Mixin 类" + cls = getattr(cli_display, names[0]) + assert hasattr(cls, "show_quick_commands") + + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + cls.show_quick_commands(cls.__new__(cls)) + out = buf.getvalue() + + assert "/fc" in out and "/diff" in out, "速查表要覆盖新命令" + assert 18 <= len(out.strip().splitlines()) <= 26 + + def test_diff_mode_remembered(self): + src = _read("xcli_core", "cli_core.py") + assert "set_system_config('diff_popup_mode'" in src + assert "if saved is not None:" in src, "存过就不该再问第二次" + + def test_new_commands_registered(self): + src = _read("xcli_core", "cli_core.py") + assert "user_input.startswith('/diff')" in src + assert "user_input.startswith('/fc')" in src + + +# ══════════════════════════════════════ +# 4. 启动噪音折叠到 --verbose +# ══════════════════════════════════════ + +class TestVerboseGate: + def test_loading_chatter_is_gated(self): + base_src = _read("xcli_core", "cli_base.py") + utm_src = _read("unified_tool_manager.py") + assert 'vprint(f"{Fore.GREEN}已注册插件命令' in base_src + assert 'vprint(f"{Fore.GREEN}已加载AI引擎' in base_src + assert 'vprint(f"{Fore.GREEN}已加载 Skill' in utm_src + assert "已加载 {len(self._tools)} 个工具" in utm_src, "工具加载应汇总成一行" + + @pytest.mark.parametrize("argv,env,expected", [ + (["ai_cli.py"], None, False), + (["ai_cli.py", "--verbose"], None, True), + (["ai_cli.py", "-v"], None, True), + (["ai_cli.py"], "1", True), + (["ai_cli.py"], "0", False), + ]) + def test_is_verbose(self, monkeypatch, argv, env, expected): + from xcli_core import verbose as vb + + monkeypatch.setattr(sys, "argv", argv) + if env is None: + monkeypatch.delenv("XCLI_VERBOSE", raising=False) + else: + monkeypatch.setenv("XCLI_VERBOSE", env) + assert vb.is_verbose() is expected + + def test_vprint_silent_by_default(self, monkeypatch, capsys): + from xcli_core import verbose as vb + + monkeypatch.setattr(sys, "argv", ["ai_cli.py"]) + monkeypatch.delenv("XCLI_VERBOSE", raising=False) + vb.vprint("should not appear") + assert capsys.readouterr().out == "" + + monkeypatch.setattr(sys, "argv", ["ai_cli.py", "-v"]) + vb.vprint("should appear") + assert "should appear" in capsys.readouterr().out diff --git a/unified_tool_manager.py b/unified_tool_manager.py index 289d7af..55d94c1 100644 --- a/unified_tool_manager.py +++ b/unified_tool_manager.py @@ -10,6 +10,7 @@ import threading from typing import Dict, List, Any, Optional, Union, Callable from colorama import Fore, Style from xcli_core.tool_result import ToolResult, ErrorCode +from xcli_core.verbose import vprint logger = logging.getLogger(__name__) @@ -55,9 +56,11 @@ class UnifiedToolManager: self._load_skills(skills_dir) self._build_mcp_cache() - print(f"{Fore.GREEN}[统一工具管理器] 加载完成: {len(self._tools)} 个工具{Style.RESET_ALL}") - print(f" - Liugin 协议: {sum(1 for p in self._protocol_map.values() if p == 'liugin')}") - print(f" - Skill 协议: {sum(1 for p in self._protocol_map.values() if p == 'skill')}") + n_liugin = sum(1 for p in self._protocol_map.values() if p == 'liugin') + n_skill = sum(1 for p in self._protocol_map.values() if p == 'skill') + print(f"{Fore.GREEN}已加载 {len(self._tools)} 个工具{Style.RESET_ALL}" + f"{Fore.BLACK}{Style.BRIGHT} (Liugin {n_liugin} · Skill {n_skill}" + f"，--verbose 看明细){Style.RESET_ALL}") def shutdown(self) -> None: """关闭管理器，清理所有插件实例""" @@ -144,7 +147,7 @@ class UnifiedToolManager: self._skill_loader = SkillLoader(skills_dir) discovered = self._skill_loader.discover_skills(skills_dir) - print(f"{Fore.CYAN}发现 {len(discovered)} 个技能: {[s['name'] for s in discovered]}{Style.RESET_ALL}") + vprint(f"{Fore.CYAN}发现 {len(discovered)} 个技能: {[s['name'] for s in discovered]}{Style.RESET_ALL}") self._skill_loader.load_all_skills(skills_dir) @@ -182,7 +185,7 @@ class UnifiedToolManager: } self._protocol_map[skill_name] = 'skill' - print(f"{Fore.GREEN}已加载 Skill ({skill_format}): {skill_name}{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}已加载 Skill ({skill_format}): {skill_name}{Style.RESET_ALL}") except ImportError as e: print(f"{Fore.YELLOW}Skill 协议模块未找到: {e}{Style.RESET_ALL}") diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py index 6ad3af1..e974680 100644 --- a/xcli_core/cli_base.py +++ b/xcli_core/cli_base.py @@ -10,6 +10,7 @@ from .constants import ( UNIFIED_TOOL_MANAGER_AVAILABLE, TEXTUAL_AVAILABLE, ) from .config import get_system_config, set_system_config, logger +from .verbose import vprint from .plugin_manager import LiuginManager from .process_protection import get_protection, enable_process_protection @@ -70,11 +71,11 @@ class BaseAICLI: # 设置默认引擎，优先使用配置文件指定的引擎 if default_engine_name in self.engines: self.current_engine = self.engines[default_engine_name] - print(f"{Fore.GREEN}使用配置的默认AI引擎: {default_engine_name}{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}使用配置的默认AI引擎: {default_engine_name}{Style.RESET_ALL}") elif self.engines: first_engine_name = next(iter(self.engines)) self.current_engine = self.engines[first_engine_name] - print(f"{Fore.GREEN}使用默认AI引擎: {first_engine_name}{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}使用默认AI引擎: {first_engine_name}{Style.RESET_ALL}") else: print(f"{Fore.RED}警告: 没有可用的AI引擎，请检查ai_engines目录{Style.RESET_ALL}") self.current_engine = None @@ -153,7 +154,7 @@ class BaseAICLI: def load_ai_engines(self): """动态加载AI引擎插件""" - print(f"{Fore.GREEN}正在加载AI引擎插件...{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}正在加载AI引擎插件...{Style.RESET_ALL}") ai_engines_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ai_engines") if os.path.exists(ai_engines_dir): for filename in os.listdir(ai_engines_dir): @@ -171,7 +172,7 @@ class BaseAICLI: engine_key = getattr(engine_instance, 'name', engine_name) if engine_key not in self.engines: self.engines[engine_key] = engine_instance - print(f"{Fore.GREEN}已加载AI引擎: {engine_key} ({class_name}){Style.RESET_ALL}") + vprint(f"{Fore.GREEN}已加载AI引擎: {engine_key} ({class_name}){Style.RESET_ALL}") else: print(f"{Fore.YELLOW}AI引擎 {engine_key} 已存在，跳过加载{Style.RESET_ALL}") else: @@ -185,7 +186,7 @@ class BaseAICLI: except Exception as e: logger.error(f"加载AI引擎插件时发生未知错误 {filename}: {e}") print(f"{Fore.RED}加载AI引擎插件失败 {filename}: {e}{Style.RESET_ALL}") - print(f"{Fore.GREEN}总共加载了 {len(self.engines)} 个AI引擎{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}总共加载了 {len(self.engines)} 个AI引擎{Style.RESET_ALL}") def _find_engine_class(self, module, engine_name): """查找AI引擎类，使用动态方式避免硬编码类名""" @@ -213,7 +214,7 @@ class BaseAICLI: def load_liugins(self, liugins_dir=None, skills_dir=None): """加载liugin和技能""" - print(f"{Fore.GREEN}正在加载工具...{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}正在加载工具...{Style.RESET_ALL}") if self._use_unified_manager: self.liugin_manager.initialize(liugins_dir or "plugins", skills_dir or "skills") @@ -242,19 +243,19 @@ class BaseAICLI: tool_info = plugin_instance.get_tool_info() tool_info['handler'] = plugin_instance.handle self.liugin_manager.tools.append(tool_info) - print(f"{Fore.GREEN}已加载liugin: {liugin_name}{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}已加载liugin: {liugin_name}{Style.RESET_ALL}") except Exception as e: print(f"{Fore.RED}加载liugin失败 {filename}: {e}{Style.RESET_ALL}") - print(f"{Fore.GREEN}总共加载了 {len(self.liugin_manager.tools)} 个工具插件{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}总共加载了 {len(self.liugin_manager.tools)} 个工具插件{Style.RESET_ALL}") - print(f"{Fore.GREEN}工具加载完成!{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}工具加载完成!{Style.RESET_ALL}") # ── 命令注册 ── def register_liugin_command(self, command, handler): """注册插件命令""" self.liugin_commands[command] = handler - print(f"{Fore.GREEN}已注册插件命令: {command}{Style.RESET_ALL}") + vprint(f"{Fore.GREEN}已注册插件命令: {command}{Style.RESET_ALL}") def unregister_liugin_command(self, command): """注销插件命令""" diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index f411fd7..a7e22ee 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -10,7 +10,7 @@ from typing import Optional from colorama import Fore, Style from .constants import TEXTUAL_AVAILABLE, LOVE_FILE_PATH, DEFAULT_MAX_HISTORY, VERSION -from .config import get_system_config, logger +from .config import get_system_config, set_system_config, logger from .cli_base import BaseAICLI from .safety import get_safety, MODE_UNRESTRICTED, MODE_NORMAL, MODE_MANUAL from .cli_clawli import ClawliMixin @@ -1031,63 +1031,55 @@ multi 操作支持一次修改多处： """运行 CLI""" import uuid self.user_id = str(uuid.getnode()) - print(f"{Fore.GREEN}小狸 Pro-CLI v{VERSION} 已启动!{Style.RESET_ALL}") - print(f"{Fore.GREEN}用户ID: {self.user_id}{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/help' 查看帮助信息{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/quit' 退出程序{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/tui' 切换到 TUI 模式{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/engine list' 查看可用AI引擎{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/engine switch <引擎名>' 切换AI引擎{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/safe' 切换安全模式 (普通→人工→无限制){Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/notify' 切换任务完成通知 (开/关){Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/scheduler' 或 '/remind' 管理定时任务{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/memory' 管理记忆系统 (日记/搜索/聊天记录){Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/file.read <文件名> [行数]' 直接读取文件内容{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '@文件路径' 自动读取文件内容并发送给AI{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '@图片路径' 自动分析图片并发送描述给AI{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/image engines' 查看图像识别引擎{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/chat save <名称>' 保存当前聊天记录{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/chat list' 查看所有已保存的聊天记录{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/chat open <名称>' 加载聊天记录{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/remote' 查看远程连接帮助{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/plugin' 管理插件市场 (安装/卸载/搜索){Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/protect' 查看进程保护状态{Style.RESET_ALL}") current_engine_name = getattr(self.current_engine, 'name', '未设置') if self.current_engine else '未设置' - print(f"{Fore.GREEN}当前使用的AI引擎: {current_engine_name}{Style.RESET_ALL}") + model_name = getattr(self.current_engine, 'model', '') if self.current_engine else '' + engine_desc = f"{current_engine_name} ({model_name})" if model_name else current_engine_name + + print(f"{Fore.GREEN}小狸 Pro-CLI v{VERSION} 已启动!{Style.RESET_ALL} " + f"{Fore.WHITE}引擎: {engine_desc}{Style.RESET_ALL}") + + # 一行速览，完整命令清单收进 /help + print(f"{Fore.CYAN}/help 全部命令 · /quit 退出 · /tui 图形界面 · " + f"@文件路径 读文件给 AI{Style.RESET_ALL}") + + extras = [] loaded_engines = list(self.engines.keys()) if len(loaded_engines) > 1: - print(f"{Fore.GREEN}已加载的AI引擎: {', '.join(loaded_engines)}{Style.RESET_ALL}") - print(f"{Fore.GREEN}{'-' * 50}{Style.RESET_ALL}") - - # 显示进程保护状态 - if hasattr(self, 'protection_results') and self.protection_results: + extras.append(f"可切换引擎: {', '.join(loaded_engines)}") + # 进程保护：只在有未生效项时才提示，全绿就闭嘴 + if getattr(self, 'protection_results', None): r = self.protection_results - si_icon = '' if r.get('single_instance') else '' - prio_icon = '' if r.get('priority') else '' - ppl_icon = '' if r.get('ppl_protection') else '' - wd_icon = '' if r.get('watchdog') else '' - print(f"{Fore.CYAN} 进程保护: {si_icon}单实例 {prio_icon}优先级 {ppl_icon}PPL保护 {wd_icon}看门狗{Style.RESET_ALL}") + failed = [n for k, n in (('single_instance', '单实例'), ('priority', '优先级'), + ('ppl_protection', 'PPL保护'), ('watchdog', '看门狗')) + if not r.get(k)] + if failed: + extras.append(f"进程保护未生效: {', '.join(failed)}") + if extras: + print(f"{Fore.BLACK}{Style.BRIGHT}{' · '.join(extras)}{Style.RESET_ALL}") - # 询问 diff 弹窗模式 + print(f"{Fore.BLACK}{Style.BRIGHT}用户ID: {self.user_id}{Style.RESET_ALL}") + + # 恢复 diff 显示模式（首次运行才询问） self._ask_diff_popup_mode() self._run_cli_loop() - def _ask_diff_popup_mode(self): - """启动时询问 diff 显示模式""" - # 找到 code_editor 插件 - code_editor = None + def _get_code_editor_plugin(self): + """拿到 code_editor 插件实例（找不到返回 None）""" for tool in self.liugin_manager.tools: if tool.get('name') == 'code_editor': - code_editor = tool.get('handler') - break + return getattr(tool.get('handler'), '__self__', None) + return None - if code_editor is None: + def _ask_diff_popup_mode(self): + """设置 diff 显示模式：已保存过就静默沿用，只有首次运行才交互询问""" + plugin_instance = self._get_code_editor_plugin() + if plugin_instance is None: return - # 找到插件实例（通过 handler 的 __self__） - plugin_instance = getattr(code_editor, '__self__', None) - if plugin_instance is None: + saved = get_system_config('diff_popup_mode', None) + if saved is not None: + plugin_instance.diff_popup_mode = bool(saved) return print(f"\n{Fore.CYAN} Diff 显示模式（AI 修改文件后的对比方式）:{Style.RESET_ALL}") @@ -1095,19 +1087,23 @@ multi 操作支持一次修改多处： print(f" {Fore.WHITE}2{Style.RESET_ALL} - 主终端内显示（SSH 兼容，推荐）") try: - choice = input(f"\n{Fore.WHITE} 请选择 [1/2]（默认 2）: {Style.RESET_ALL}").strip() + choice = input(f"\n{Fore.WHITE} 请选择 [1/2]（默认 2，之后可用 /diff 改）: {Style.RESET_ALL}").strip() except (EOFError, KeyboardInterrupt): choice = '2' - if choice == '1': - plugin_instance.diff_popup_mode = True - print(f"{Fore.GREEN} ✓ 已选择: 弹窗模式{Style.RESET_ALL}") - else: - plugin_instance.diff_popup_mode = False - print(f"{Fore.GREEN} ✓ 已选择: 主终端内显示{Style.RESET_ALL}") - + self._set_diff_popup_mode(choice == '1', plugin_instance) print() + def _set_diff_popup_mode(self, popup, plugin_instance=None): + """写入并持久化 diff 显示模式""" + if plugin_instance is None: + plugin_instance = self._get_code_editor_plugin() + if plugin_instance is not None: + plugin_instance.diff_popup_mode = bool(popup) + set_system_config('diff_popup_mode', bool(popup)) + label = '弹窗模式' if popup else '主终端内显示' + print(f"{Fore.GREEN} ✓ Diff 显示模式: {label}{Style.RESET_ALL}") + def _run_cli_loop(self): """运行 CLI 主循环""" while True: @@ -1297,6 +1293,42 @@ multi 操作支持一次修改多处： print(f"{icon} 安全模式: {mode_name}") continue + if user_input.startswith('/diff'): + parts = user_input.split() + if self._get_code_editor_plugin() is None: + print(f"{Fore.YELLOW}code_editor 插件未加载，无法设置 diff 模式{Style.RESET_ALL}") + continue + if len(parts) > 1: + arg = parts[1].lower() + if arg in ('popup', '弹窗', '1'): + self._set_diff_popup_mode(True) + elif arg in ('inline', '内联', '主终端', '2'): + self._set_diff_popup_mode(False) + else: + print(f"{Fore.RED}用法: /diff [popup|inline]{Style.RESET_ALL}") + else: + cur = bool(get_system_config('diff_popup_mode', False)) + self._set_diff_popup_mode(not cur) + continue + + if user_input.startswith('/fc'): + from .fc_tools import get_fc_cache, reset_fc_cache + parts = user_input.split() + if len(parts) > 1 and parts[1].lower() in ('reset', 'clear', '清除'): + n = reset_fc_cache() + print(f"{Fore.GREEN}已清除 {n} 条 FC 探测记录，下次请求会重新试探{Style.RESET_ALL}") + else: + cache = get_fc_cache() + if not cache: + print(f"{Fore.GREEN}FC 探测缓存为空：所有模型都按支持 function calling 处理{Style.RESET_ALL}") + else: + print(f"{Fore.CYAN}已探测到不支持 function calling 的模型:{Style.RESET_ALL}") + for key, ok in cache.items(): + if ok is False: + print(f" {Fore.YELLOW}✗{Style.RESET_ALL} {key}") + print(f"{Fore.WHITE}这些请求会自动跳过 tools 字段；/fc reset 可清除重测{Style.RESET_ALL}") + continue + if user_input.startswith('/notify'): parts = user_input.split() nm = get_notification_manager() @@ -1436,12 +1468,22 @@ def main(): import codecs sys.stdout = codecs.getwriter('utf-8')(sys.stdout.buffer, 'strict') - for line in art_lines: + def _safe_print(line): try: print(line, flush=True) except UnicodeEncodeError: print(line.encode('gbk', errors='ignore').decode('gbk'), flush=True) - time.sleep(0.05) + + # 完整开机动画（57 行 + 逐行 0.05s）默认不放，用 --logo 显式召唤 + if '--logo' in sys.argv or '--banner' in sys.argv: + for line in art_lines: + _safe_print(line) + time.sleep(0.05) + else: + _safe_print(f"\n{Fore.CYAN} ▄▀▄ 小狸 Pro-CLI v{VERSION}{Style.RESET_ALL}") + _safe_print(f"{Fore.CYAN} ▀▄▀ {Style.RESET_ALL}" + f"{Fore.WHITE}AI 智能编程助手{Style.RESET_ALL}" + f" {Fore.BLACK}{Style.BRIGHT}(--logo 看完整开机动画){Style.RESET_ALL}") cli = AICLI() cli.run() diff --git a/xcli_core/cli_display.py b/xcli_core/cli_display.py index bc0a4e2..4c46a1b 100644 --- a/xcli_core/cli_display.py +++ b/xcli_core/cli_display.py @@ -376,8 +376,42 @@ class DisplayMixin: # ── 帮助系统 ── + def show_quick_commands(self): + """常用命令速查表（启动时不再刷屏，改由 /help 展示）""" + rows = [ + ('/help [插件名]', '显示帮助信息'), + ('/quit', '退出程序'), + ('/tui', '切换到 TUI 图形模式'), + ('/engine list', '查看可用 AI 引擎'), + ('/engine switch <名>', '切换 AI 引擎'), + ('/fc [reset]', '查看/清除 FC 能力探测缓存'), + ('/safe', '切换安全模式 (普通→人工→无限制)'), + ('/diff', '切换 diff 显示模式 (弹窗/主终端)'), + ('/notify', '切换任务完成通知 (开/关)'), + ('/scheduler 或 /remind', '管理定时任务'), + ('/memory', '管理记忆系统 (日记/搜索/聊天记录)'), + ('/file.read <文件> [行数]', '直接读取文件内容'), + ('@文件路径', '自动读取文件内容并发送给 AI'), + ('@图片路径', '自动分析图片并发送描述给 AI'), + ('/image engines', '查看图像识别引擎'), + ('/chat save|list|open', '保存/查看/加载聊天记录'), + ('/remote', '查看远程连接帮助'), + ('/plugin', '管理插件市场 (安装/卸载/搜索)'), + ('/protect', '查看进程保护状态'), + ] + width = max(len(cmd) for cmd, _ in rows) + print(f"{Fore.GREEN}常用命令:{Style.RESET_ALL}") + for cmd, desc in rows: + print(f" {Fore.WHITE}{cmd.ljust(width)}{Style.RESET_ALL} {desc}") + print() + print(f"{Fore.GREEN}启动参数:{Style.RESET_ALL}") + print(f" {Fore.WHITE}{'--verbose / -v'.ljust(width)}{Style.RESET_ALL} 打印引擎/插件/技能的逐条加载明细") + print(f" {Fore.WHITE}{'--logo'.ljust(width)}{Style.RESET_ALL} 播放完整开机动画") + print() + def show_help(self): """显示帮助信息""" + self.show_quick_commands() help_file_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "HELP.md") if os.path.exists(help_file_path): try: diff --git a/xcli_core/config.py b/xcli_core/config.py index ebc5a6f..f7a0590 100644 --- a/xcli_core/config.py +++ b/xcli_core/config.py @@ -76,13 +76,29 @@ def set_system_config(key, value): save_config(config) -# 配置日志 +# 配置日志：文件留全量 INFO，控制台默认只报 WARNING 以上（--verbose 时放开到 INFO） +try: + from .verbose import is_verbose as _is_verbose +except ImportError: # 被当作顶层模块单独导入时 + try: + from verbose import is_verbose as _is_verbose + except ImportError: + def _is_verbose(): + return False + +_file_handler = logging.FileHandler('xiaoli_cli.log', encoding='utf-8') +_file_handler.setLevel(logging.INFO) +_console_handler = logging.StreamHandler() +_console_handler.setLevel(logging.INFO if _is_verbose() else logging.WARNING) + logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('xiaoli_cli.log', encoding='utf-8'), - logging.StreamHandler() # 同时输出到控制台 - ] + handlers=[_file_handler, _console_handler] ) + +# 第三方库的心跳日志（httpx 每发一次请求就打一行）不上屏也不进日志文件 +for _noisy in ('httpx', 'httpcore', 'urllib3', 'requests', 'openai', 'PIL', 'markdown_it'): + logging.getLogger(_noisy).setLevel(logging.WARNING) + logger = logging.getLogger(__name__) diff --git a/xcli_core/fc_tools.py b/xcli_core/fc_tools.py index bced2c2..d639cb0 100644 --- a/xcli_core/fc_tools.py +++ b/xcli_core/fc_tools.py @@ -2,7 +2,9 @@ Function Calling 工具格式转换 将 MCP 插件定义转为 OpenAI FC 格式，收集所有可用工具 """ +import os import json +import threading from typing import List, Dict, Optional @@ -248,3 +250,81 @@ def _get_builtin_tools() -> List[Dict]: } } ] + + +# ══════════════════════════════════════════════════════════════════ +# FC 能力探测缓存 +# +# 部分渠道/模型（如讯飞星火经 one-api 转发的 xophunyuan* 系列） +# 不支持 function calling，请求体里只要带上非空 tools 字段就会直接 +# 返回 500 RequestParamsError:Invalid Params。 +# 引擎捕获到这种错误后会去掉 tools 降级重试；本缓存把"该模型不支持 FC" +# 这件事记下来，后续请求直接跳过 tools，避免每次冷启动都白撞一次 500。 +# ══════════════════════════════════════════════════════════════════ + +_FC_CACHE_FILE = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".fc_support.json" +) +_fc_cache = None +_fc_lock = threading.Lock() + + +def _fc_key(base_url, model) -> str: + """缓存键：base_url + model（同一渠道下不同模型能力可能不同）""" + return f"{(base_url or '').rstrip('/')}::{model or ''}" + + +def _load_fc_cache() -> dict: + global _fc_cache + if _fc_cache is not None: + return _fc_cache + try: + with open(_FC_CACHE_FILE, 'r', encoding='utf-8') as f: + data = json.load(f) + _fc_cache = data if isinstance(data, dict) else {} + except (FileNotFoundError, json.JSONDecodeError, OSError, ValueError): + _fc_cache = {} + return _fc_cache + + +def _save_fc_cache(cache: dict): + try: + with open(_FC_CACHE_FILE, 'w', encoding='utf-8') as f: + json.dump(cache, f, ensure_ascii=False, indent=2) + except OSError: + pass + + +def supports_fc(base_url, model) -> bool: + """该 (base_url, model) 是否支持 FC。未探测过时默认 True（乐观，先试一次）""" + return _load_fc_cache().get(_fc_key(base_url, model)) is not False + + +def mark_fc_unsupported(base_url, model) -> bool: + """标记该 (base_url, model) 不支持 FC 并持久化。首次标记返回 True""" + with _fc_lock: + cache = _load_fc_cache() + key = _fc_key(base_url, model) + if cache.get(key) is False: + return False + cache[key] = False + _save_fc_cache(cache) + return True + + +def reset_fc_cache(base_url=None, model=None) -> int: + """清除 FC 探测缓存。不传参数则全清，返回清除的条目数""" + with _fc_lock: + cache = _load_fc_cache() + if base_url is None and model is None: + count = len(cache) + cache.clear() + else: + count = 1 if cache.pop(_fc_key(base_url, model), None) is not None else 0 + _save_fc_cache(cache) + return count + + +def get_fc_cache() -> dict: + """返回 FC 探测缓存副本（供 /fc 命令展示）""" + return dict(_load_fc_cache()) diff --git a/xcli_core/verbose.py b/xcli_core/verbose.py new file mode 100644 index 0000000..5329f5d --- /dev/null +++ b/xcli_core/verbose.py @@ -0,0 +1,24 @@ +""" +启动详情开关 + +默认启动只报结论（加载了几个引擎/几个工具），不再逐条刷屏。 +需要看每个插件/技能的加载明细时，加 --verbose（或 -v）， +也可以设环境变量 XCLI_VERBOSE=1。 +""" +import os +import sys + +_TRUTHY = ('1', 'true', 'yes', 'on') + + +def is_verbose() -> bool: + """是否输出启动明细""" + if any(a in ('--verbose', '-v') for a in sys.argv[1:]): + return True + return str(os.environ.get('XCLI_VERBOSE', '')).lower() in _TRUTHY + + +def vprint(*args, **kwargs): + """仅在 --verbose 时打印的 print""" + if is_verbose(): + print(*args, **kwargs)