feat: 废弃 FC,改用纯 JSON action 模式调用工具
- engine.generate_response 不再下发 tools schema(TUI/CLI 分支均 tools=None) - 系统提示改为纯 JSON 调用格式说明,保留 tool_search 自查用法 - 新增 _loads_json 容错解析,容忍模型写出的 Windows 反斜杠路径(如 C:/Users) - 新增 tests/test_pure_json.py 覆盖解析容错/引擎无 schema 下发/端到端按名分发 Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
This commit is contained in:
1 parent
b577b0ff21
commit
c87a07706e
2 files changed
+179
-33
No files matched your search
@@ -0,0 +1,145 @@
|
||||
"""
|
||||
纯 JSON 工具调用模式 测试
|
||||
|
||||
取代 FC:引擎不再收到 tools schema,模型按系统提示里的
|
||||
{"action":"use_tool","tool":名,"args":参数} 文本输出,由 _parse_mixed_response
|
||||
解析、按名分发。重点验证:
|
||||
1. _loads_json / _parse_mixed_response 对 Windows 反斜杠路径健壮
|
||||
2. 引擎在 tools=None 时返回纯文本(不产 _fc / tool_calls)
|
||||
3. 端到端:真实 AICLI + FakeEngine,use_tool 被按名分发执行,且无 FC schema 下发
|
||||
"""
|
||||
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)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════
|
||||
# 1. 解析器:Windows 反斜杠路径容错
|
||||
# ════════════════════════════════════════════════
|
||||
|
||||
@pytest.fixture
|
||||
def cli():
|
||||
from xcli_core.cli_core import AICLI
|
||||
return AICLI()
|
||||
|
||||
|
||||
class TestParseRobustness:
|
||||
def test_loads_json_tolerates_backslash_path(self, cli):
|
||||
# 模型常把 Windows 路径写成单反斜杠 → 非法 JSON 转义
|
||||
raw = r'{"action": "use_tool", "tool": "code_editor", "args": "create C:\Users\me\demo.py print(1)"}'
|
||||
data = cli._loads_json(raw)
|
||||
assert data["action"] == "use_tool"
|
||||
assert data["tool"] == "code_editor"
|
||||
# 修复后应是合法的单反斜杠路径
|
||||
assert data["args"] == r"create C:\Users\me\demo.py print(1)"
|
||||
|
||||
def test_parse_mixed_extracts_use_tool_with_backslash(self, cli):
|
||||
resp = (
|
||||
'好的,我来创建文件。\n'
|
||||
r'{"action": "use_tool", "tool": "code_editor", "args": "create C:\Users\me\demo.py print(1)"}'
|
||||
)
|
||||
text, data = cli._parse_mixed_response(resp)
|
||||
assert isinstance(data, dict)
|
||||
assert data["action"] == "use_tool"
|
||||
assert data["tool"] == "code_editor"
|
||||
assert r"C:\Users\me\demo.py" in data["args"]
|
||||
|
||||
def test_parse_mixed_plain_text_no_action(self, cli):
|
||||
text, data = cli._parse_mixed_response("今天天气不错")
|
||||
assert data is None
|
||||
assert "今天天气不错" in text
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════
|
||||
# 2. 引擎:tools=None 返回纯文本,不产 tool_calls
|
||||
# ════════════════════════════════════════════════
|
||||
|
||||
class TestEngineToolsNone:
|
||||
def test_openai_engine_no_tools_sent(self, monkeypatch):
|
||||
import types
|
||||
import ai_engines.openai_engine as oe
|
||||
|
||||
calls = []
|
||||
|
||||
def fake_post(url=None, json=None, headers=None, timeout=None, **kw):
|
||||
calls.append(dict(json))
|
||||
return types.SimpleNamespace(
|
||||
status_code=200,
|
||||
json=lambda: {"choices": [{"message": {"content": "好"}}],
|
||||
"usage": {"prompt_tokens": 5}},
|
||||
text="{}",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(oe, "requests",
|
||||
types.SimpleNamespace(post=fake_post, exceptions=oe.requests.exceptions))
|
||||
|
||||
eng = oe.OpenaiAI.__new__(oe.OpenaiAI)
|
||||
eng.name = "openai"
|
||||
eng.api_key = "k"
|
||||
eng.base_url = "https://example.com/v1"
|
||||
eng.model = "gpt-4o"
|
||||
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>"
|
||||
|
||||
reply = eng.generate_response("hi", tools=None)
|
||||
assert reply == "好"
|
||||
assert "tools" not in calls[0], "tools=None 时不应下发 FC schema"
|
||||
assert "tool_choice" not in calls[0]
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════
|
||||
# 3. 端到端:真实 AICLI + FakeEngine,use_tool 按名分发
|
||||
# ════════════════════════════════════════════════
|
||||
|
||||
class FakeEngine:
|
||||
def __init__(self):
|
||||
self.calls = 0
|
||||
self.tools_seen = []
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return "fake"
|
||||
|
||||
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"纯 JSON 模式不应下发 FC tools,实际 {tools!r}"
|
||||
if self.calls == 1:
|
||||
return ('好的,我来创建文件。\n'
|
||||
'{"action": "use_tool", "tool": "code_editor", '
|
||||
'"args": "create %s print(1)"}' % self._fp)
|
||||
return "已完成。"
|
||||
|
||||
|
||||
class TestEndToEndDispatch:
|
||||
def test_use_tool_dispatched_without_fc(self):
|
||||
from xcli_core.cli_core import AICLI
|
||||
from xcli_core.safety import get_safety, MODE_UNRESTRICTED
|
||||
|
||||
cli = AICLI()
|
||||
engine = FakeEngine()
|
||||
cli.current_engine = engine
|
||||
cli.tui_output_callback = lambda *a, **k: None # 同步 TUI 分支,免动画线程
|
||||
get_safety().set_mode(MODE_UNRESTRICTED) # 跳过交互确认,验证分发本身
|
||||
|
||||
d = tempfile.mkdtemp(prefix="pj_")
|
||||
FakeEngine._fp = os.path.join(d, "demo.py") # 原始反斜杠路径
|
||||
fp = FakeEngine._fp
|
||||
|
||||
cli.process_conversation("帮我创建一个 python 文件")
|
||||
|
||||
assert all(t is None for t in engine.tools_seen), "引擎不应收到任何 FC tools"
|
||||
assert engine.calls >= 2, "至少应有:首轮JSON + 工具结果续轮"
|
||||
assert os.path.exists(fp), "code_editor 应被按名分发执行并创建文件"
|
||||
assert "print(1)" in open(fp, encoding="utf-8").read()
|
||||
+34
-33
@@ -42,7 +42,7 @@ class AICLI(BaseAICLI, ClawliMixin, ToolMixin, CodeExecMixin, DisplayMixin, Hist
|
||||
def process_thinking_response(self, response):
|
||||
"""处理深度思考AI返回的内容, 分离思考和回复内容"""
|
||||
try:
|
||||
response_data = json.loads(response)
|
||||
response_data = self._loads_json(response)
|
||||
if isinstance(response_data, dict) and response_data.get('action') == 'use_tool':
|
||||
return response
|
||||
if isinstance(response_data, dict):
|
||||
@@ -310,13 +310,14 @@ multi 操作支持一次修改多处:
|
||||
plugins.append(tool)
|
||||
|
||||
if plugins:
|
||||
# 工具名清单(与 FC tools 字段互补;不逐条重复 description,避免双重曝光)。
|
||||
# 无 function-calling 能力的模型只能靠此清单知道有哪些工具,故保留名称。
|
||||
# 工具名清单(纯 JSON 模式:不发送 FC schema,模型按 action JSON 调用)。
|
||||
# 只列名字 + 由 tool_search 提供用法,避免重复曝光、省 token。
|
||||
names = ", ".join(t.get('name', '') for t in plugins)
|
||||
prompts.append(
|
||||
"## 内置工具(已通过 function calling 的 tools 字段提供,调用前请先用 "
|
||||
"tool_search 查询其详细用法与示例)\n"
|
||||
f"工具名: {names}"
|
||||
"## 内置工具(纯 JSON 模式:不发送 function calling 的 tools schema,"
|
||||
"模型按下方 JSON 格式直接调用,调用前请先用 tool_search 查询其详细用法与示例)\n"
|
||||
f"工具名: {names}\n"
|
||||
'调用格式: {"action": "use_tool", "tool": "工具名", "args": "参数"}'
|
||||
)
|
||||
|
||||
if skills:
|
||||
@@ -356,6 +357,18 @@ multi 操作支持一次修改多处:
|
||||
|
||||
return "\n".join(prompts)
|
||||
|
||||
def _loads_json(self, s):
|
||||
"""json.loads,但容忍未成对出现的反斜杠(Windows 路径 C:\\Users 常见)。
|
||||
|
||||
纯 JSON 模式下模型常把路径写成单反斜杠,直接 json.loads 会报非法转义。
|
||||
这里先按标准解析;失败时把「未成对的反斜杠」转义后再试一次。
|
||||
"""
|
||||
try:
|
||||
return json.loads(s)
|
||||
except json.JSONDecodeError:
|
||||
fixed = re.sub(r'(?<!\\)\\(?!\\\\)', r'\\\\', s)
|
||||
return json.loads(fixed)
|
||||
|
||||
def _parse_mixed_response(self, response):
|
||||
"""
|
||||
解析混合响应,处理同时包含文本和JSON指令的内容
|
||||
@@ -368,7 +381,7 @@ multi 操作支持一次修改多处:
|
||||
json_str = response[11:]
|
||||
json_str = "{" + json_str
|
||||
try:
|
||||
parsed = json.loads(json_str)
|
||||
parsed = self._loads_json(json_str)
|
||||
if isinstance(parsed, dict):
|
||||
return "", parsed
|
||||
except json.JSONDecodeError:
|
||||
@@ -376,7 +389,7 @@ multi 操作支持一次修改多处:
|
||||
|
||||
# 方法1: 直接解析为纯JSON
|
||||
try:
|
||||
parsed = json.loads(response)
|
||||
parsed = self._loads_json(response)
|
||||
if isinstance(parsed, (list, dict)):
|
||||
return "", parsed
|
||||
except json.JSONDecodeError:
|
||||
@@ -393,7 +406,7 @@ multi 操作支持一次修改多处:
|
||||
content = content.replace('"', '\\"')
|
||||
return f'"{content}"'
|
||||
fixed_response = re.sub(r'"([^"]*(?:\n[^"]*)*)"', fix_json_string, response, flags=re.DOTALL)
|
||||
parsed = json.loads(fixed_response)
|
||||
parsed = self._loads_json(fixed_response)
|
||||
if isinstance(parsed, (list, dict)):
|
||||
return "", parsed
|
||||
except:
|
||||
@@ -404,7 +417,7 @@ multi 操作支持一次修改多处:
|
||||
matches = re.findall(json_pattern, response, re.DOTALL)
|
||||
if matches:
|
||||
try:
|
||||
json_data = json.loads(matches[-1])
|
||||
json_data = self._loads_json(matches[-1])
|
||||
if isinstance(json_data, (dict, list)):
|
||||
text_content = re.sub(json_pattern, "", response, flags=re.DOTALL).strip()
|
||||
text_content = re.sub(r"\n\s*\n", "\n\n", text_content).strip()
|
||||
@@ -420,7 +433,7 @@ multi 操作支持一次修改多处:
|
||||
line = line.strip()
|
||||
if line.startswith("{") and line.endswith("}"):
|
||||
try:
|
||||
parsed = json.loads(line)
|
||||
parsed = self._loads_json(line)
|
||||
if isinstance(parsed, dict):
|
||||
json_objects.append(parsed)
|
||||
continue
|
||||
@@ -458,7 +471,7 @@ multi 操作支持一次修改多处:
|
||||
if in_json and brace_count == 0:
|
||||
json_str = response[json_start:i+1]
|
||||
try:
|
||||
json_data = json.loads(json_str)
|
||||
json_data = self._loads_json(json_str)
|
||||
if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
|
||||
text_content = response[:json_start].strip()
|
||||
return text_content, json_data
|
||||
@@ -470,7 +483,7 @@ multi 操作支持一次修改多处:
|
||||
potential_jsons = re.findall(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', response, re.DOTALL)
|
||||
for json_str in potential_jsons:
|
||||
try:
|
||||
json_data = json.loads(json_str)
|
||||
json_data = self._loads_json(json_str)
|
||||
if isinstance(json_data, dict) and json_data.get('action') in ['use_tool', 'continue']:
|
||||
json_start = response.find(json_str)
|
||||
text_content = response[:json_start].strip()
|
||||
@@ -488,12 +501,12 @@ multi 操作支持一次修改多处:
|
||||
if last_brace_pos != -1:
|
||||
json_part = cleaned[last_brace_pos:]
|
||||
if json_part.startswith('['):
|
||||
parsed = json.loads(json_part)
|
||||
parsed = self._loads_json(json_part)
|
||||
if isinstance(parsed, list):
|
||||
text_content = cleaned[:last_brace_pos].strip()
|
||||
return text_content, parsed
|
||||
else:
|
||||
parsed = json.loads(json_part)
|
||||
parsed = self._loads_json(json_part)
|
||||
if isinstance(parsed, dict):
|
||||
text_content = cleaned[:last_brace_pos].strip()
|
||||
return text_content, parsed
|
||||
@@ -788,7 +801,7 @@ multi 操作支持一次修改多处:
|
||||
return None
|
||||
|
||||
try:
|
||||
data = json.loads(response)
|
||||
data = self._loads_json(response)
|
||||
if not data.get("_fc"):
|
||||
return None
|
||||
|
||||
@@ -883,14 +896,9 @@ multi 操作支持一次修改多处:
|
||||
system_prompt = self._build_system_prompt(liugin_prompts)
|
||||
if not self.current_engine:
|
||||
return "错误: 当前没有可用的AI引擎,请检查ai_engines目录中的引擎插件"
|
||||
# TUI 模式也支持 FC
|
||||
fc_tools = None
|
||||
try:
|
||||
from .fc_tools import mcp_to_openai_tools, prune_tools
|
||||
fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input)
|
||||
except Exception:
|
||||
pass
|
||||
response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=fc_tools)
|
||||
# 纯 JSON 模式:不发送 FC tools schema,模型按 action JSON 调用,
|
||||
# 由 _parse_mixed_response 解析、按名分发(任意工具,无 schema 门槛)。
|
||||
response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=None)
|
||||
return response
|
||||
|
||||
# CLI 模式:带动画和跨平台 ESC 取消
|
||||
@@ -972,16 +980,9 @@ multi 操作支持一次修改多处:
|
||||
if not self.current_engine:
|
||||
response_result[0] = "错误: 当前没有可用的AI引擎,请检查ai_engines目录中的引擎插件"
|
||||
else:
|
||||
# ── 收集 FC 工具定义(按当前输入动态裁剪)──
|
||||
fc_tools = None
|
||||
try:
|
||||
from .fc_tools import mcp_to_openai_tools, prune_tools
|
||||
fc_tools = prune_tools(mcp_to_openai_tools(self.liugin_manager), current_input)
|
||||
except Exception:
|
||||
fc_tools = None
|
||||
|
||||
# 纯 JSON 模式:不发送 FC tools schema,模型按 action JSON 调用。
|
||||
response_result[0] = self.current_engine.generate_response(
|
||||
current_input, system_prompt=system_prompt, tools=fc_tools
|
||||
current_input, system_prompt=system_prompt, tools=None
|
||||
)
|
||||
except Exception as e:
|
||||
response_result[0] = f"AI 响应生成失败: {e}"
|
||||
|
||||
Reference in new issue
Block a user