From 46376aa23e5ed379bc6588f96ff0f98e25c384cc Mon Sep 17 00:00:00 2001 From: shuiyu67 Date: Thu, 6 Aug 2026 23:30:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=8D=E6=96=B0=E6=89=93=E5=BC=80=20?= =?UTF-8?q?FC=20=E5=BC=80=E5=85=B3=EF=BC=8CAI=20=E5=8F=AF=E5=9C=A8?= =?UTF-8?q?=E7=BA=AF=20JSON=20=E4=B8=8E=20FC=20=E4=B9=8B=E9=97=B4=E8=87=AA?= =?UTF-8?q?=E9=80=89=EF=BC=88=E6=B7=B7=E5=90=88=E6=A8=A1=E5=BC=8F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ai_engines/ollama_engine.py | 33 ++++++++-- tests/test_fc_hybrid.py | 121 ++++++++++++++++++++++++++++++++++++ xcli_core/cli_core.py | 46 +++++++++++--- 3 files changed, 189 insertions(+), 11 deletions(-) create mode 100644 tests/test_fc_hybrid.py diff --git a/ai_engines/ollama_engine.py b/ai_engines/ollama_engine.py index 55d449e..f4b4859 100644 --- a/ai_engines/ollama_engine.py +++ b/ai_engines/ollama_engine.py @@ -203,12 +203,21 @@ Ollama AI引擎插件帮助信息 "options": {"num_ctx": self.num_ctx} } - # ── Function Calling: 注入工具定义 ── - if tools: + # ── Function Calling: 注入工具定义（仅引擎支持 FC 时下发）── + fc_enabled = bool(tools) and self._fc_supported() + if fc_enabled: kwargs["tools"] = tools - # 使用Ollama Python库调用API - response = ollama_client.chat(**kwargs) + # 使用Ollama Python库调用API（FC 不支持时降级为无工具重试一次） + try: + response = ollama_client.chat(**kwargs) + except Exception as _fc_err: + if fc_enabled and "tool" in str(_fc_err).lower(): + self._mark_fc_unsupported() + kwargs.pop("tools", None) + response = ollama_client.chat(**kwargs) + else: + raise # ── Function Calling: 检查 tool_calls ── message = response.get("message", {}) @@ -266,6 +275,22 @@ Ollama AI引擎插件帮助信息 except Exception as e: return f"Ollama API调用失败: {str(e)}" + def _fc_supported(self): + """该 (base_url, model) 是否支持 FC。未探测过时默认 True（乐观，先试一次）""" + try: + from xcli_core.fc_tools import supports_fc + return supports_fc(self.base_url, self.model) + except Exception: + return False + + def _mark_fc_unsupported(self): + """标记该 (base_url, model) 不支持 FC 并持久化""" + try: + from xcli_core.fc_tools import mark_fc_unsupported + mark_fc_unsupported(self.base_url, self.model) + except Exception: + pass + def _build_messages_with_history(self, system_prompt, user_input): """构建包含历史记录的消息列表（支持 FC tool 消息）""" messages = [] diff --git a/tests/test_fc_hybrid.py b/tests/test_fc_hybrid.py new file mode 100644 index 0000000..95e87a8 --- /dev/null +++ b/tests/test_fc_hybrid.py @@ -0,0 +1,121 @@ +""" +FC 混合模式（AI 自选）集成测试 + +验证「让 AI 自己选纯 JSON 或 FC」的接线： +1. 支持 FC 的引擎会收到下发的 tools schema（tool_choice=auto，模型自选） +2. 引擎返回 _fc JSON 时，code_editor 被按名分发执行（走 _handle_fc_response 路径） +3. 不支持 FC 的引擎（无 _fc_supported）则不收 tools，纯 JSON 照常工作 +""" +import os +import sys +import json +import tempfile + +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) + + +class FakeFcEngine: + """模拟一个支持 FC 的引擎：首轮返回 _fc 调用 code_editor，次轮收尾。""" + def __init__(self): + self.calls = 0 + self.tools_seen = [] + + @property + def name(self): + return "fakefc" + + def _fc_supported(self): + return True + + def generate_response(self, user_input, tool_results=None, system_prompt=None, tools=None): + self.calls += 1 + self.tools_seen.append(tools) + if self.calls == 1: + return json.dumps({ + "_fc": True, + "tool_calls": [{ + "id": "c1", + "name": "code_editor", + "tool": "code_editor", + "args": "create %s print(1)" % self._fp, + }], + }, ensure_ascii=False) + return "已完成。" + + +class FakeNoFcEngine: + """模拟一个不支持 FC 的引擎（无 _fc_supported）：应降级为纯 JSON。""" + def __init__(self): + self.calls = 0 + self.tools_seen = [] + + @property + def name(self): + return "fakenofc" + + def generate_response(self, user_input, tool_results=None, system_prompt=None, tools=None): + self.calls += 1 + self.tools_seen.append(tools) + assert tools is None, f"不支持 FC 的引擎不应收到 tools，实际 {tools!r}" + if self.calls == 1: + return ('好的，我来创建文件。\n' + '{"action": "use_tool", "tool": "code_editor", ' + '"args": "create %s print(1)"}' % self._fp) + return "已完成。" + + +def _run(engine): + from xcli_core.cli_core import AICLI + from xcli_core.safety import get_safety, MODE_UNRESTRICTED + + cli = AICLI() + cli.current_engine = engine + cli.tui_output_callback = lambda *a, **k: None # 同步 TUI 分支，免动画线程 + get_safety().set_mode(MODE_UNRESTRICTED) + + d = tempfile.mkdtemp(prefix="pj_") + fp = os.path.join(d, "demo.py") + engine._fp = fp + cli.process_conversation("帮我创建一个 python 文件") + return engine, fp + + +def test_fc_engine_receives_tools_and_dispatches(): + engine, fp = _run(FakeFcEngine()) + # 首轮应向支持 FC 的引擎下发 tools schema + assert engine.tools_seen[0] is not None, "FC 引擎应收到 tools schema" + assert isinstance(engine.tools_seen[0], list) and len(engine.tools_seen[0]) > 0 + assert any( + t.get("function", {}).get("name") == "code_editor" for t in engine.tools_seen[0] + ), "code_editor 应在下发的 FC 工具集中" + # _fc 调用应被按名分发执行并真实创建文件 + assert os.path.exists(fp), "FC 路径应执行 code_editor 并创建文件" + assert "print(1)" in open(fp, encoding="utf-8").read() + assert engine.calls >= 2 + + +def test_no_fc_engine_stays_pure_json(): + engine, fp = _run(FakeNoFcEngine()) + # 不支持 FC 的引擎不应收到 tools，纯 JSON 照常分发 + assert all(t is None for t in engine.tools_seen), "不支持 FC 的引擎不应收到 tools" + assert os.path.exists(fp), "纯 JSON 路径仍应创建文件" + + +def test_build_fc_tools_returns_none_for_unsupported_engine(): + from xcli_core.cli_core import AICLI + cli = AICLI() + cli.current_engine = FakeNoFcEngine() + assert cli._build_fc_tools("帮我创建文件") is None + + +def test_build_fc_tools_returns_list_for_supported_engine(): + from xcli_core.cli_core import AICLI + cli = AICLI() + cli.current_engine = FakeFcEngine() + tools = cli._build_fc_tools("帮我创建 python 文件") + assert isinstance(tools, list) and len(tools) > 0 + assert any(t.get("function", {}).get("name") == "code_editor" for t in tools) diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 49ec1a6..b41bf1b 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -907,6 +907,38 @@ multi 操作支持一次修改多处： return f"未找到工具: {tool_name}" + def _build_fc_tools(self, current_input): + """构建本轮下发给引擎的 FC 工具子集，让引擎自行决定用 FC 结构化调用还是回退纯 JSON。 + + 引擎不支持 FC（探测缓存未通过）时返回 None —— 引擎自动降级为纯 JSON 模式。 + 仅对带 `_fc_supported()` 的引擎下发 tools（Manual 等不支持的引擎不传 tools， + 避免关键字参数错误）。 + """ + engine = self.current_engine + if not engine or not hasattr(engine, '_fc_supported'): + return None + try: + if not engine._fc_supported(): + return None + except Exception: + return None + try: + from .fc_tools import mcp_to_openai_tools, prune_tools + all_tools = mcp_to_openai_tools(self.liugin_manager) + if not all_tools: + return None + return prune_tools(all_tools, current_input) + except Exception: + return None + + def _call_engine_with_tools(self, current_input, system_prompt, fc_tools): + """调用当前引擎；fc_tools 非空时下发 FC 工具定义（tool_choice=auto，模型自选）。""" + if fc_tools: + return self.current_engine.generate_response( + current_input, system_prompt=system_prompt, tools=fc_tools) + return self.current_engine.generate_response( + current_input, system_prompt=system_prompt) + def _generate_response_with_animation(self, current_input, liugin_prompts=None): """生成AI响应并显示等待动画 — 支持 ESC 跨平台取消""" # TUI 模式：不做动画 @@ -914,9 +946,10 @@ multi 操作支持一次修改多处： system_prompt = self._build_system_prompt(liugin_prompts) if not self.current_engine: return "错误: 当前没有可用的AI引擎，请检查ai_engines目录中的引擎插件" - # 纯 JSON 模式：不发送 FC tools schema，模型按 action JSON 调用， - # 由 _parse_mixed_response 解析、按名分发（任意工具，无 schema 门槛）。 - response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=None) + # 混合模式：下发 FC 工具定义，tool_choice=auto —— 模型可自行选择 + # 用 FC 结构化调用，或回退到纯 JSON action（两种都会被正确处理）。 + fc_tools = self._build_fc_tools(current_input) + response = self._call_engine_with_tools(current_input, system_prompt, fc_tools) return response # CLI 模式：带动画和跨平台 ESC 取消 @@ -998,10 +1031,9 @@ multi 操作支持一次修改多处： if not self.current_engine: response_result[0] = "错误: 当前没有可用的AI引擎，请检查ai_engines目录中的引擎插件" else: - # 纯 JSON 模式：不发送 FC tools schema，模型按 action JSON 调用。 - response_result[0] = self.current_engine.generate_response( - current_input, system_prompt=system_prompt, tools=None - ) + # 混合模式：下发 FC 工具（tool_choice=auto），模型自选 FC 或纯 JSON + fc_tools = self._build_fc_tools(current_input) + response_result[0] = self._call_engine_with_tools(current_input, system_prompt, fc_tools) except Exception as e: response_result[0] = f"AI 响应生成失败: {e}" finally: