diff --git a/ai_engines/openai_engine.py b/ai_engines/openai_engine.py index 3bf8723..7867aac 100644 --- a/ai_engines/openai_engine.py +++ b/ai_engines/openai_engine.py @@ -19,6 +19,11 @@ except Exception: def get_system_config(key, default=None): return default +try: + from xcli_core import model_registry +except Exception: + model_registry = None + try: from xcli_core.verbose import vprint except Exception: @@ -140,6 +145,9 @@ OpenAI 兼容格式引擎帮助信息 self.conversation_history = [] self.shared_conversation_history = None + # 应用注册表中的当前模型(覆盖扁平默认配置,向后兼容) + self._apply_registry_model() + # 最近一次请求的 token 用量(供 token 感知压缩使用) self.last_prompt_tokens = None @@ -155,6 +163,63 @@ OpenAI 兼容格式引擎帮助信息 else: vprint(f"{Fore.GREEN}OpenAI 兼容引擎已启用 (base_url: {self.base_url}){Style.RESET_ALL}") + # ── 模型注册表(多模型在线切换)── + + def _apply_registry_model(self): + """若注册表存在 current 模型,用它覆盖 __init__ 读到的扁平默认配置。""" + if not model_registry: + return + try: + m = model_registry.get_model() + if m: + self.api_key = m.get("api_key", self.api_key) + self.base_url = (m.get("base_url") or self.base_url).rstrip('/') + self.model = m.get("model", self.model) + except Exception: + pass + + def list_registry_models(self): + """返回 (models, current_name),供 CLI 列出已配置模型。""" + if not model_registry: + return [], None + return model_registry.list_models() + + def current_model_name(self): + """当前激活模型在注册表中的名称(兼容单槽位时返回 model 值)。""" + if not model_registry: + return self.model + try: + models, current = model_registry.list_models() + return current or self.model + except Exception: + return self.model + + def set_registry_model(self, name): + """切换到注册表中指定模型(含 base_url/api_key/model)并持久化 current。 + 成功返回 True。""" + if not model_registry: + return False + m = model_registry.get_model(name) + if not m: + return False + self.api_key = m.get("api_key", "") + self.base_url = (m.get("base_url") or "").rstrip('/') + self.model = m.get("model", "") + model_registry.set_current(name) + return True + + def add_model(self, entry): + """新增/更新模型并持久化。entry: {name, base_url, api_key, model}。""" + if not model_registry: + return None + return model_registry.add_model(entry) + + def remove_model(self, name): + """删除模型并持久化。成功返回 True。""" + if not model_registry: + return False + return model_registry.remove_model(name) + def _get_headers(self): """构建请求头""" headers = { diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py new file mode 100644 index 0000000..d217e9e --- /dev/null +++ b/tests/test_model_registry.py @@ -0,0 +1,188 @@ +"""模型注册表 + /model 命令的回归测试。 + +不污染真实 config.json:所有读写都指向临时文件。 +""" +import os +import json +import tempfile + +import pytest + +from xcli_core import model_registry as mr +import xcli_core.config as cfgmod +import ai_engines.openai_engine as oe_mod + + +@pytest.fixture +def tmp_config(): + """准备临时 config.json,并把注册表与 openai 引擎的读取都重定向过去。""" + d = tempfile.mkdtemp() + path = os.path.join(d, "config.json") + payload = { + "api": {"engines": { + "openai": {"api_key": "K0", "base_url": "https://old/v1", "model": "old-model"} + }}, + "system": {}, + } + with open(path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + + saved_path = mr._CONFIG_PATH + saved_proj = oe_mod.project_root + mr._set_config_path(path) + oe_mod.project_root = d + try: + yield path + finally: + mr._set_config_path(saved_path) + oe_mod.project_root = saved_proj + + +# ── 注册表单元 ── + +def test_legacy_flat_config_is_single_implicit_model(tmp_config): + models, current = mr.list_models() + assert len(models) == 1 + assert models[0]["model"] == "old-model" + assert current == "old-model" + + +def test_add_first_model_migrates_flat_and_keeps_current(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "https://api.openai.com/v1", + "api_key": "sk", "model": "gpt-4o"}) + models, current = mr.list_models() + names = [m["name"] for m in models] + assert names == ["old-model", "gpt4o"] + # current 不应跳变到新加的模型 + assert current == "old-model" + + +def test_add_duplicate_name_updates_in_place(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "u1", "api_key": "k1", "model": "gpt-4o"}) + mr.add_model({"name": "gpt4o", "base_url": "u2", "api_key": "k2", "model": "gpt-4o-mini"}) + models, _ = mr.list_models() + assert len(models) == 2 # 未重复追加 + assert models[1]["base_url"] == "u2" + + +def test_set_current_and_get_model(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "https://api.openai.com/v1", + "api_key": "sk", "model": "gpt-4o"}) + assert mr.set_current("gpt4o") is True + m = mr.get_model() + assert m["model"] == "gpt-4o" + assert mr.set_current("ghost") is False # 越权返回 False + + +def test_remove_current_falls_back_to_first(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "u", "api_key": "k", "model": "gpt-4o"}) + mr.set_current("gpt4o") + assert mr.remove_model("gpt4o") is True + _, current = mr.list_models() + assert current == "old-model" + + +def test_remove_missing_returns_false(tmp_config): + assert mr.remove_model("nope") is False + + +def test_persisted_to_disk(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "u", "api_key": "k", "model": "gpt-4o"}) + mr.set_current("gpt4o") + with open(tmp_config, "r", encoding="utf-8") as f: + saved = json.load(f) + block = saved["api"]["engines"]["openai"] + assert block["current"] == "gpt4o" + assert any(m["name"] == "gpt4o" for m in block["models"]) + + +# ── 引擎接入 ── + +def test_engine_init_applies_registry_current(tmp_config): + mr.add_model({"name": "gpt4o", "base_url": "https://api.openai.com/v1", + "api_key": "sk", "model": "gpt-4o"}) + mr.set_current("gpt4o") + eng = oe_mod.OpenaiAI() + assert eng.model == "gpt-4o" + assert "api.openai.com" in eng.base_url + assert eng.current_model_name() == "gpt4o" + + +def test_engine_set_registry_model(tmp_config): + eng = oe_mod.OpenaiAI() + eng.add_model({"name": "gpt4o", "base_url": "https://api.openai.com/v1", + "api_key": "sk", "model": "gpt-4o"}) + assert eng.set_registry_model("gpt4o") is True + assert eng.model == "gpt-4o" + assert eng.set_registry_model("ghost") is False + + +def test_engine_does_not_clobber_existing_set_model(tmp_config): + """原有轻量 set_model(只改 model 名)必须仍然可用,不被注册表方法覆盖。""" + eng = oe_mod.OpenaiAI() + eng.set_model("gpt-4o-mini") # 原有方法:只改 self.model + assert eng.model == "gpt-4o-mini" + # 注册表方法仍在 + assert hasattr(eng, "set_registry_model") + assert hasattr(eng, "list_registry_models") + + +# ── CLI 命令路由 ── + +def _make_fake_cli(eng): + """复制 cli_base 的模型命令方法到轻量对象,避免实例化整个 CliBase。""" + from xcli_core import cli_base + cls = type("FakeCli", (), {}) + cls.handle_model_command = cli_base.BaseAICLI.handle_model_command + cls._model_list = cli_base.BaseAICLI._model_list + cls._model_switch = cli_base.BaseAICLI._model_switch + cls._model_add_interactive = cli_base.BaseAICLI._model_add_interactive + return cls() + + +def test_cli_model_list_and_switch(tmp_config, monkeypatch): + from xcli_core import cli_base + eng = oe_mod.OpenaiAI() + eng.add_model({"name": "gpt4o", "base_url": "u", "api_key": "k", "model": "gpt-4o"}) + + cli = _make_fake_cli(eng) + cli.current_engine = eng + cli.current_model = None + + # 列出 + captured = {} + monkeypatch.setattr("builtins.print", lambda *a, **k: captured.setdefault("out", []).append(a)) + cli.handle_model_command("list") + assert any("gpt4o" in str(x) for x in captured["out"]) + + # 切换 + cli.handle_model_command("gpt4o") + assert cli.current_model == "gpt4o" + assert eng.model == "gpt-4o" + + +def test_cli_model_switch_missing(tmp_config, monkeypatch): + eng = oe_mod.OpenaiAI() + cli = _make_fake_cli(eng) + cli.current_engine = eng + cli.current_model = None + captured = {} + monkeypatch.setattr("builtins.print", lambda *a, **k: captured.setdefault("out", []).append(a)) + cli.handle_model_command("ghost") + assert any("未找到" in str(x) for x in captured["out"]) + assert cli.current_model is None + + +def test_cli_model_add_interactive(tmp_config, monkeypatch): + eng = oe_mod.OpenaiAI() + cli = _make_fake_cli(eng) + cli.current_engine = eng + cli.current_model = None + answers = iter(["mygpt", "https://x/v1", "sk-1", "gpt-x"]) + monkeypatch.setattr("builtins.input", lambda *a, **k: next(answers)) + captured = {} + monkeypatch.setattr("builtins.print", lambda *a, **k: captured.setdefault("out", []).append(a)) + cli.handle_model_command("add") + models, current = eng.list_registry_models() + assert any(m["name"] == "mygpt" for m in models) + assert cli.current_model == "mygpt" diff --git a/tests/test_startup_and_fc.py b/tests/test_startup_and_fc.py index 1910d84..8757494 100644 --- a/tests/test_startup_and_fc.py +++ b/tests/test_startup_and_fc.py @@ -241,7 +241,7 @@ class TestStartupSlim: out = buf.getvalue() assert "/fc" in out and "/diff" in out, "速查表要覆盖新命令" - assert 18 <= len(out.strip().splitlines()) <= 26 + assert 18 <= len(out.strip().splitlines()) <= 34 def test_diff_mode_remembered(self): src = _read("xcli_core", "cli_core.py") diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py index e974680..ba77c44 100644 --- a/xcli_core/cli_base.py +++ b/xcli_core/cli_base.py @@ -81,6 +81,13 @@ class BaseAICLI: self.current_engine = None if self.current_engine: self.current_engine.cli = self # 设置引用以便访问插件 + # 记录当前模型(openai 引擎注册表名,其它引擎取 model 值) + try: + self.current_model = self.current_engine.current_model_name() + except Exception: + self.current_model = getattr(self.current_engine, 'model', None) + else: + self.current_model = None # 初始化安全层 from .safety import get_safety @@ -306,6 +313,83 @@ class BaseAICLI: help_lines.append(f" /engine {cmd:<10} - 执行{cmd}命令") return '\n'.join(help_lines) if help_lines else " 无可用引擎管理命令" + # ── 模型管理(OpenAI 引擎多模型切换)── + + def handle_model_command(self, args): + """处理 /model 命令族。仅对 openai 引擎有效。""" + eng = self.current_engine + if getattr(eng, 'name', None) != 'openai': + cur = getattr(eng, 'name', '?') + print(f"{Fore.RED}模型管理仅适用于 openai 引擎(当前: {cur})。" + f"先 /engine switch openai{Style.RESET_ALL}") + return + + if not args or args == 'list': + self._model_list(eng) + return + + parts = args.split(' ', 1) + cmd = parts[0] + rest = parts[1].strip() if len(parts) > 1 else '' + if cmd == 'add': + self._model_add_interactive(eng) + elif cmd in ('rm', 'remove', 'del'): + if not rest: + print(f"{Fore.RED}用法: /model rm <名称>{Style.RESET_ALL}") + return + if eng.remove_model(rest): + print(f"{Fore.GREEN}已删除模型: {rest}{Style.RESET_ALL}") + else: + print(f"{Fore.RED}未找到模型: {rest}{Style.RESET_ALL}") + elif cmd in ('set', 'switch'): + self._model_switch(eng, rest) + else: + # 无子命令:/model <名称> 视为切换 + self._model_switch(eng, cmd) + + def _model_list(self, eng): + models, current = eng.list_registry_models() + if not models: + print(f"{Fore.YELLOW}尚未配置任何模型,用 /model add 添加{Style.RESET_ALL}") + return + print(f"{Fore.GREEN}已配置的 OpenAI 模型:{Style.RESET_ALL}") + for m in models: + mark = f"{Fore.CYAN} *{Style.RESET_ALL}" if m.get('name') == current else "" + print(f" {m.get('name')}{mark} ({m.get('model')} @ {m.get('base_url') or '无 base_url'})") + print(f"{Fore.CYAN}当前: {current} | /model <名称> 切换 | /model add 新增 | " + f"/model rm <名称> 删除{Style.RESET_ALL}") + + def _model_switch(self, eng, name): + if not name: + print(f"{Fore.RED}用法: /model <名称> 或 /model 查看列表{Style.RESET_ALL}") + return + if eng.set_registry_model(name): + self.current_model = name + print(f"{Fore.GREEN}已切换到模型: {name}{Style.RESET_ALL}") + else: + print(f"{Fore.RED}未找到模型: {name}(/model 查看可用){Style.RESET_ALL}") + + def _model_add_interactive(self, eng): + print(f"{Fore.CYAN}添加 OpenAI 模型(回车留空则用默认/为空){Style.RESET_ALL}") + try: + name = input(f"{Fore.CYAN}名称(用于切换, 如 gpt4o): {Style.RESET_ALL}").strip() + if not name: + print(f"{Fore.RED}名称不能为空{Style.RESET_ALL}") + return + base_url = input(f"{Fore.CYAN}base_url(如 https://api.openai.com/v1): {Style.RESET_ALL}").strip() + api_key = input(f"{Fore.CYAN}api_key: {Style.RESET_ALL}").strip() + model = input(f"{Fore.CYAN}model(如 gpt-4o): {Style.RESET_ALL}").strip() + if not model: + print(f"{Fore.RED}model 不能为空{Style.RESET_ALL}") + return + except (EOFError, KeyboardInterrupt): + print(f"{Fore.YELLOW}\n已取消添加{Style.RESET_ALL}") + return + eng.add_model({"name": name, "base_url": base_url, "api_key": api_key, "model": model}) + eng.set_registry_model(name) + self.current_model = name + print(f"{Fore.GREEN}已添加并切换到模型: {name}{Style.RESET_ALL}") + # ── 引擎切换 ── def get_available_engines(self): diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index a7e22ee..7b140fa 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -1040,7 +1040,7 @@ multi 操作支持一次修改多处: # 一行速览,完整命令清单收进 /help print(f"{Fore.CYAN}/help 全部命令 · /quit 退出 · /tui 图形界面 · " - f"@文件路径 读文件给 AI{Style.RESET_ALL}") + f"/model 换模型 · @文件路径 读文件给 AI{Style.RESET_ALL}") extras = [] loaded_engines = list(self.engines.keys()) @@ -1141,6 +1141,11 @@ multi 操作支持一次修改多处: print(f"{Fore.RED}当前引擎不支持该命令或命令执行失败{Style.RESET_ALL}") continue + # /model 命令族 — OpenAI 引擎多模型在线增删切换 + if user_input == '/model' or user_input.startswith('/model '): + self.handle_model_command(user_input[7:].strip()) + continue + # /manual 快捷命令 — 切换到 manual 引擎或执行 manual 子命令 if user_input == '/manual' or user_input.startswith('/manual '): args = user_input[8:].strip() if user_input.startswith('/manual ') else '' diff --git a/xcli_core/cli_display.py b/xcli_core/cli_display.py index 4c46a1b..7536764 100644 --- a/xcli_core/cli_display.py +++ b/xcli_core/cli_display.py @@ -384,6 +384,10 @@ class DisplayMixin: ('/tui', '切换到 TUI 图形模式'), ('/engine list', '查看可用 AI 引擎'), ('/engine switch <名>', '切换 AI 引擎'), + ('/model', '列出已配置的 OpenAI 模型'), + ('/model <名称>', '切换当前 OpenAI 模型'), + ('/model add', '交互式新增 OpenAI 模型'), + ('/model rm <名称>', '删除已配置的模型'), ('/fc [reset]', '查看/清除 FC 能力探测缓存'), ('/safe', '切换安全模式 (普通→人工→无限制)'), ('/diff', '切换 diff 显示模式 (弹窗/主终端)'), diff --git a/xcli_core/model_registry.py b/xcli_core/model_registry.py new file mode 100644 index 0000000..6123c4b --- /dev/null +++ b/xcli_core/model_registry.py @@ -0,0 +1,145 @@ +"""OpenAI 格式模型注册表。 + +把「多个 OpenAI 格式模型」从 config.json 的单槽位升级为可管理的列表, +支持在线增删、切换、持久化,免去每次手改配置文件。 + +配置结构(api.engines.openai): + { + "models": [ + {"name": "讯飞xop", "base_url": "...", "api_key": "...", "model": "xophunyuan7bmt"}, + {"name": "gpt4o", "base_url": "https://api.openai.com/v1", "api_key": "...", "model": "gpt-4o"} + ], + "current": "讯飞xop" + } + +向后兼容:若没有 models 列表,把扁平的 api_key/base_url/model 当作单条隐式模型, +其 name 取 model 值(空则 "default")。 +""" + +from xcli_core.config import load_config, save_config + +# 测试可临时覆盖配置路径 +_CONFIG_PATH = None + + +def _set_config_path(path): + """仅供测试:指向临时 config.json""" + global _CONFIG_PATH + _CONFIG_PATH = path + + +def _load(): + if _CONFIG_PATH: + try: + with open(_CONFIG_PATH, 'r', encoding='utf-8') as f: + return __import__('json').load(f) + except (FileNotFoundError, __import__('json').JSONDecodeError): + return {"api": {"engines": {}}, "system": {}} + return load_config() + + +def _persist(cfg): + if _CONFIG_PATH: + with open(_CONFIG_PATH, 'w', encoding='utf-8') as f: + __import__('json').dump(cfg, f, ensure_ascii=False, indent=2) + return + save_config(cfg) + + +def _openai_block(cfg): + return cfg.setdefault("api", {}).setdefault("engines", {}).setdefault("openai", {}) + + +def list_models(cfg=None): + """返回 (models:list[dict], current:str|None)。 + + 无 models 列表时返回单条隐式模型(由扁平 api_key/base_url/model 构成)。 + """ + cfg = cfg or _load() + block = _openai_block(cfg) + models = block.get("models") + if not models: + m = block.get("model") or "default" + implicit = { + "name": m, + "base_url": block.get("base_url", ""), + "api_key": block.get("api_key", ""), + "model": block.get("model", ""), + } + return [implicit], (block.get("current") or m) + current = block.get("current") or (models[0]["name"] if models else None) + return models, current + + +def get_model(name=None, cfg=None): + """取某模型配置;name=None 取 current。找不到返回 None。""" + models, current = list_models(cfg) + target = name or current + if not target: + return None + for m in models: + if m.get("name") == target: + return dict(m) + return None + + +def set_current(name): + """把 current 指针指向 name 并落盘。name 必须存在,否则返回 False。""" + cfg = _load() + block = _openai_block(cfg) + models = block.get("models") or [] + if not any(m.get("name") == name for m in models): + return False + block["current"] = name + _persist(cfg) + return True + + +def add_model(entry): + """新增或更新模型,落盘并返回其 name。 + + entry: {"name", "base_url", "api_key", "model"}。同名则更新。 + 首次添加会从扁平配置迁移出第一条隐式模型,且保持 current 不跳变。 + """ + name = (entry.get("name") or "").strip() + if not name: + raise ValueError("模型 name 不能为空") + cfg = _load() + block = _openai_block(cfg) + models = block.get("models") + if not models: + old_name = block.get("model") or "default" + models = [{ + "name": old_name, + "base_url": block.get("base_url", ""), + "api_key": block.get("api_key", ""), + "model": block.get("model", ""), + }] + block["models"] = models + if not block.get("current"): + block["current"] = old_name + for m in models: + if m.get("name") == name: + m.update(entry) + _persist(cfg) + return name + models.append(dict(entry)) + _persist(cfg) + return name + + +def remove_model(name): + """删除模型,返回是否成功(找不到返回 False)。 + 若删除的是 current,则 current 回退到列表第一个。 + """ + cfg = _load() + block = _openai_block(cfg) + models = block.get("models") or [] + new = [m for m in models if m.get("name") != name] + if len(new) == len(models): + return False + block["models"] = new + if block.get("current") == name: + block["current"] = new[0]["name"] if new else None + _persist(cfg) + return True