perf: 启动输出瘦身 + FC 能力探测缓存,修复测试套件失效

自用体验(同时以人类和被接入 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)
This commit is contained in:
shuiyu67 committed 2026-08-05 15:57:38 +08:00
1 parent d2ac6896ad
commit 3a86da7444
12 files changed
+727 -166

No files matched your search

+6
View File
@@ -50,3 +50,9 @@ screenshots/gen_*.png
# v8.0 TUI 快照产物(运行时生成,供 AI 自验/预览)
.tui_snapshots/
# FC 能力探测缓存(运行时按 base_url+model 生成,机器相关)
.fc_support.json
# pytest 缓存
.pytest_cache/
+7 -1
View File
@@ -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
# <input> 特殊语法解析(配置模型时可交互式询问用户)
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}")
+33 -4
View File
@@ -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"):
+111 -88
View File
@@ -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)
# ══════════════════════════════════════
+297
View File
@@ -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 = "<think>"
eng.thinking_end_marker = "</think>"
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
+8 -5
View File
@@ -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}")
+11 -10
View File
@@ -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):
"""注销插件命令"""
+95 -53
View File
@@ -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)
# 完整开机动画(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()
+34
View File
@@ -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:
+21 -5
View File
@@ -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__)
+80
View File
@@ -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())
+24
View File
@@ -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)