diff --git a/ai_engines/ollama_engine.py b/ai_engines/ollama_engine.py index 9015376..2535d02 100644 --- a/ai_engines/ollama_engine.py +++ b/ai_engines/ollama_engine.py @@ -107,74 +107,136 @@ Ollama AI引擎插件帮助信息 # 如果出现异常,说明服务未运行 return False - def generate_response(self, user_input, tool_results=None, system_prompt=None): - """生成AI响应 - 使用Ollama API""" + def generate_response(self, user_input, tool_results=None, system_prompt=None, tools=None): + """ + 生成AI响应 - 使用Ollama API + + Args: + user_input: 用户输入 + tool_results: 工具执行结果 + system_prompt: 系统提示词 + tools: OpenAI FC 工具定义列表 (可选) + """ try: # 检查Ollama服务是否运行 if not self.is_service_running: return "错误:Ollama服务未运行,请启动Ollama服务后再使用此引擎." - + # 如果有工具结果,将其作为上下文返回给AI处理 if tool_results: tool_result_text = f"工具执行结果: {tool_results.get('result', '无结果')}" - # 构建包含历史记录的消息列表 messages = self._build_messages_with_history(system_prompt, tool_result_text) - - # 使用Ollama Python库调用API - response = ollama_client.chat( - model=self.model, - messages=messages, - stream=False - ) - - # 返回AI的回复 - return response["message"]["content"] - - # 构建包含历史记录的消息列表 - messages = self._build_messages_with_history(system_prompt, user_input) - + else: + messages = self._build_messages_with_history(system_prompt, user_input) + + # 构建调用参数 + kwargs = { + "model": self.model, + "messages": messages, + "stream": False + } + + # ── Function Calling: 注入工具定义 ── + if tools: + kwargs["tools"] = tools + # 使用Ollama Python库调用API - response = ollama_client.chat( - model=self.model, - messages=messages, - stream=False - ) - - # 返回AI的回复 - content = response["message"]["content"] + response = ollama_client.chat(**kwargs) + + # ── Function Calling: 检查 tool_calls ── + message = response.get("message", {}) + tool_calls_raw = message.get("tool_calls") + if tool_calls_raw: + # Ollama 的 tool_calls 格式: [{"function": {"name": ..., "arguments": {...}}}] + fc_calls = [] + for i, tc in enumerate(tool_calls_raw): + func = tc.get("function", {}) + name = func.get("name", "") + args = func.get("arguments", {}) + call_id = f"call_{name}_{i}" + + # 统一为小狸内部格式 + if "args" in args and isinstance(args["args"], str): + final_args = args["args"] + elif "args" in args: + final_args = json.dumps(args["args"], ensure_ascii=False) + else: + parts = [] + for k, v in args.items(): + parts.append(str(v) if isinstance(v, str) else json.dumps(v, ensure_ascii=False)) + final_args = " ".join(parts) if parts else "" + + fc_calls.append({ + "id": call_id, + "name": name, + "arguments": args, + "action": "use_tool", + "tool": name, + "args": final_args, + }) + + if fc_calls: + return json.dumps({ + "_fc": True, + "tool_calls": fc_calls, + }, ensure_ascii=False) + + # 普通文本响应 + content = message.get("content", "") # 清理无效的 UTF-8 代理字符 content = content.encode('utf-8', 'replace').decode('utf-8') return content - + except Exception as e: - # 如果API调用失败,返回错误信息 return f"Ollama API调用失败: {str(e)}" def _build_messages_with_history(self, system_prompt, user_input): - """构建包含历史记录的消息列表""" + """构建包含历史记录的消息列表（支持 FC tool 消息）""" messages = [] - + # 添加系统提示词 if system_prompt: - # 清理代理字符 clean_prompt = system_prompt.encode('utf-8', 'replace').decode('utf-8') messages.append({ "role": "system", "content": clean_prompt }) - -# 添加历史对话记录（使用共享历史） + + # 添加历史对话记录（使用共享历史） if self.shared_conversation_history is not None: - # 确保历史记录格式正确 for msg in self.shared_conversation_history[-self.max_history:]: - if isinstance(msg, dict) and "role" in msg and "content" in msg: - # 清理内容中的代理字符 - clean_content = msg["content"].encode('utf-8', 'replace').decode('utf-8') - messages.append({"role": msg["role"], "content": clean_content}) - else: - # 如果格式不正确，跳过该条记录 + if not isinstance(msg, dict) or "role" not in msg: continue - + + role = msg["role"] + + # tool 角色消息（FC 工具结果） + if role == "tool": + tool_msg = {"role": "tool", "content": msg.get("content", "")} + if "tool_call_id" in msg: + tool_msg["tool_call_id"] = msg["tool_call_id"] + if "name" in msg: + tool_msg["name"] = msg["name"] + messages.append(tool_msg) + continue + + # assistant 消息可能包含 tool_calls + if role == "assistant" and "tool_calls" in msg: + assistant_msg = { + "role": "assistant", + "content": msg.get("content"), + "tool_calls": msg["tool_calls"] + } + messages.append(assistant_msg) + continue + + # 普通 user/assistant 消息 + if "content" in msg: + clean_content = msg["content"] + if isinstance(clean_content, str): + clean_content = clean_content.encode('utf-8', 'replace').decode('utf-8') + messages.append({"role": role, "content": clean_content}) + # 清理用户输入 if user_input: clean_input = user_input.encode('utf-8', 'replace').decode('utf-8') @@ -182,7 +244,7 @@ Ollama AI引擎插件帮助信息 "role": "user", "content": clean_input }) - + return messages diff --git a/ai_engines/openai_engine.py b/ai_engines/openai_engine.py index 9a0d2dd..bbc207c 100644 --- a/ai_engines/openai_engine.py +++ b/ai_engines/openai_engine.py @@ -125,8 +125,16 @@ OpenAI 兼容格式引擎帮助信息 url = url.rstrip('/') + '/chat/completions' return url - def generate_response(self, user_input, tool_results=None, system_prompt=None): - """生成 AI 响应""" + def generate_response(self, user_input, tool_results=None, system_prompt=None, tools=None): + """ + 生成 AI 响应 + + Args: + user_input: 用户输入 + tool_results: 工具执行结果 + system_prompt: 系统提示词 + tools: OpenAI FC 工具定义列表 (可选) + """ try: if not self.base_url: return "错误: openai 引擎的 base_url 未配置，请在 config.json 中设置" @@ -145,6 +153,11 @@ OpenAI 兼容格式引擎帮助信息 "stream": False } + # ── Function Calling: 注入工具定义 ── + if tools: + body["tools"] = tools + body["tool_choice"] = "auto" + # 发送请求 api_url = self._get_api_url() response = requests.post( @@ -157,18 +170,49 @@ OpenAI 兼容格式引擎帮助信息 if response.status_code == 200: response_data = response.json() if "choices" in response_data and len(response_data["choices"]) > 0: - ai_response = response_data["choices"][0]["message"]["content"] - # 清理无效的 UTF-8 代理字符 - ai_response = ai_response.encode('utf-8', 'replace').decode('utf-8') - # 处理思考内容 - ai_response = self._process_thinking_content(ai_response) - return ai_response + message = response_data["choices"][0].get("message", {}) + + # ── Function Calling: 检查 tool_calls ── + tool_calls_raw = message.get("tool_calls") + if tool_calls_raw: + # 返回 FC 结果（由调用方处理工具执行） + from xcli_core.fc_tools import parse_fc_response + fc_calls = parse_fc_response(response_data) + if fc_calls: + # 返回一个特殊的 JSON 字符串，标记为 FC 调用 + return json.dumps({ + "_fc": True, + "tool_calls": fc_calls, + "raw_message": message + }, ensure_ascii=False) + + # 普通文本响应 + ai_response = message.get("content", "") + if ai_response: + # 清理无效的 UTF-8 代理字符 + ai_response = ai_response.encode('utf-8', 'replace').decode('utf-8') + # 处理思考内容 + ai_response = self._process_thinking_content(ai_response) + return ai_response or "" else: return f"API 返回格式错误: {response_data}" elif response.status_code == 401: return "API 认证失败，请检查 api_key 是否正确" elif response.status_code == 404: return f"API 端点不存在，请检查 base_url 是否正确: {api_url}" + elif response.status_code == 400: + # 工具定义可能不兼容，降级为无工具重试 + if 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: + rd = retry.json() + if "choices" in rd and rd["choices"]: + ai_response = rd["choices"][0].get("message", {}).get("content", "") + return ai_response.encode('utf-8', 'replace').decode('utf-8') if ai_response else "" + return f"API 调用失败: {response.status_code} - {response.text}" else: return f"API 调用失败: {response.status_code} - {response.text}" @@ -178,7 +222,7 @@ OpenAI 兼容格式引擎帮助信息 return f"API 调用失败: {str(e)}" def _build_messages_with_history(self, system_prompt, user_input): - """构建包含历史记录的消息列表""" + """构建包含历史记录的消息列表（支持 FC tool 消息）""" messages = [] # 添加系统提示词 @@ -192,12 +236,38 @@ OpenAI 兼容格式引擎帮助信息 # 添加历史对话记录（使用共享历史） if self.shared_conversation_history is not None: for msg in self.shared_conversation_history[-self.max_history:]: - if isinstance(msg, dict) and "role" in msg and "content" in msg: - clean_content = msg["content"].encode('utf-8', 'replace').decode('utf-8') - messages.append({"role": msg["role"], "content": clean_content}) - else: + if not isinstance(msg, dict) or "role" not in msg: continue + role = msg["role"] + + # tool 角色消息（FC 工具结果） + if role == "tool": + tool_msg = {"role": "tool", "content": msg.get("content", "")} + if "tool_call_id" in msg: + tool_msg["tool_call_id"] = msg["tool_call_id"] + if "name" in msg: + tool_msg["name"] = msg["name"] + messages.append(tool_msg) + continue + + # assistant 消息可能包含 tool_calls + if role == "assistant" and "tool_calls" in msg: + assistant_msg = { + "role": "assistant", + "content": msg.get("content"), + "tool_calls": msg["tool_calls"] + } + messages.append(assistant_msg) + continue + + # 普通 user/assistant 消息 + if "content" in msg: + clean_content = msg["content"] + if isinstance(clean_content, str): + clean_content = clean_content.encode('utf-8', 'replace').decode('utf-8') + messages.append({"role": role, "content": clean_content}) + # 添加当前用户输入 if user_input: clean_input = user_input.encode('utf-8', 'replace').decode('utf-8') diff --git a/scheduler_tasks.json b/scheduler_tasks.json new file mode 100644 index 0000000..0637a08 --- /dev/null +++ b/scheduler_tasks.json @@ -0,0 +1 @@ +[] \ No newline at end of file diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 329f9cd..74840e4 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -6,6 +6,7 @@ import time import shutil import threading import random +from typing import Optional from colorama import Fore, Style from .constants import TEXTUAL_AVAILABLE, LOVE_FILE_PATH, DEFAULT_MAX_HISTORY @@ -473,6 +474,14 @@ multi 操作支持一次修改多处： processed_response = self.process_thinking_response(response) processed_response = self._process_code_blocks(processed_response) + # ── Function Calling 响应检测 ── + fc_handled = self._handle_fc_response(processed_response, user_input) + if fc_handled: + # FC 工具已执行，用结果继续循环 + current_input = fc_handled + loop_count += 1 + continue + try: text_content, json_data = self._parse_mixed_response(processed_response) @@ -707,6 +716,99 @@ multi 操作支持一次修改多处： summary = summary[:77] + "..." return summary + def _handle_fc_response(self, response: str, original_input: str) -> Optional[str]: + """ + 处理 Function Calling 响应 + 如果响应是 FC 格式，执行工具并返回结果（供下一轮循环使用） + 如果不是 FC 响应，返回 None + """ + if not response or '_fc' not in response: + return None + + try: + data = json.loads(response) + if not data.get("_fc"): + return None + + tool_calls = data.get("tool_calls", []) + if not tool_calls: + return None + + # 记录 assistant 的 FC 消息到历史 + fc_history_entry = { + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": tc.get("id", ""), + "type": "function", + "function": { + "name": tc.get("name", ""), + "arguments": json.dumps(tc.get("arguments", {}), ensure_ascii=False) + } + } for tc in tool_calls] + } + self.shared_conversation_history.append(fc_history_entry) + + # 执行每个工具调用 + results = [] + for tc in tool_calls: + tool_name = tc.get("tool", tc.get("name", "")) + tool_args = tc.get("args", "") + call_id = tc.get("id", "") + + print(f"\n{Fore.GREEN} FC 调用: {tool_name}({tool_args[:60]}){Style.RESET_ALL}") + + # 通过插件管理器执行 + result = self._execute_tool_by_name(tool_name, tool_args) + + # 记录工具结果到历史 + tool_result_msg = { + "role": "tool", + "tool_call_id": call_id, + "name": tool_name, + "content": result[:5000] if result else "执行完成" + } + self.shared_conversation_history.append(tool_result_msg) + + results.append(f"[{tool_name}]: {result[:200] if result else '完成'}") + + # 返回工具结果，供下一轮 AI 处理 + return "工具执行结果:\n" + "\n".join(results) + + except (ValueError, KeyError, TypeError): + return None + + def _execute_tool_by_name(self, tool_name: str, tool_args: str) -> str: + """根据名称执行工具（FC 调用路径）""" + # 安全检查 + safety = get_safety() + allowed, msg = safety.check(tool_name, tool_args) + if not allowed: + return f"操作被拒绝: {msg}" + + # 查找工具 + tool = self.liugin_manager.get_tool_by_name(tool_name) + if tool and 'handler' in tool: + try: + return tool['handler'](tool_args) + except Exception as e: + return f"工具执行失败: {e}" + + # memory 内置工具 + if tool_name == 'memory' and hasattr(self, 'memory_manager'): + return self.memory_manager.handle_tool(tool_args) + + # scheduler 内置工具 + if tool_name == 'scheduler': + for t in self.liugin_manager.tools: + if t.get('name') == 'scheduler' and 'handler' in t: + try: + return t['handler'](tool_args) + except Exception as e: + return f"工具执行失败: {e}" + + return f"未找到工具: {tool_name}" + def _generate_response_with_animation(self, current_input, liugin_prompts=None): """生成AI响应并显示等待动画 — 支持 ESC 跨平台取消""" # TUI 模式：不做动画 @@ -714,7 +816,14 @@ multi 操作支持一次修改多处： system_prompt = self._build_system_prompt(liugin_prompts) if not self.current_engine: return "错误: 当前没有可用的AI引擎，请检查ai_engines目录中的引擎插件" - response = self.current_engine.generate_response(current_input, system_prompt=system_prompt) + # TUI 模式也支持 FC + fc_tools = None + try: + from .fc_tools import mcp_to_openai_tools + fc_tools = mcp_to_openai_tools(self.liugin_manager) + except Exception: + pass + response = self.current_engine.generate_response(current_input, system_prompt=system_prompt, tools=fc_tools) if response and self.shared_conversation_history is not None: self.shared_conversation_history.append({ "role": "assistant", @@ -801,8 +910,18 @@ 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 + fc_tools = mcp_to_openai_tools(self.liugin_manager) + if fc_tools: + fc_tools = fc_tools # 已经是列表 + except Exception: + fc_tools = None + response_result[0] = self.current_engine.generate_response( - current_input, system_prompt=system_prompt + current_input, system_prompt=system_prompt, tools=fc_tools ) except Exception as e: response_result[0] = f"AI 响应生成失败: {e}" diff --git a/xcli_core/fc_tools.py b/xcli_core/fc_tools.py new file mode 100644 index 0000000..bbf5b8e --- /dev/null +++ b/xcli_core/fc_tools.py @@ -0,0 +1,250 @@ +""" +Function Calling 工具格式转换 +将 MCP 插件定义转为 OpenAI FC 格式，收集所有可用工具 +""" +import json +from typing import List, Dict, Optional + + +def mcp_to_openai_tools(plugin_manager) -> List[Dict]: + """ + 从插件管理器收集所有工具定义，转为 OpenAI function calling 格式 + + Returns: + [{"type": "function", "function": {"name": ..., "description": ..., "parameters": ...}}, ...] + """ + tools = [] + + # 从插件管理器获取工具 + if not plugin_manager: + return tools + + # 统一工具管理器 + if hasattr(plugin_manager, 'plugins'): + for plugin in plugin_manager.plugins: + fc_def = _extract_fc_def(plugin) + if fc_def: + tools.append(fc_def) + # 普通插件管理器 + elif hasattr(plugin_manager, 'tools'): + for tool_info in plugin_manager.tools: + fc_def = _tool_info_to_fc(tool_info) + if fc_def: + tools.append(fc_def) + + # 追加内置工具（memory 等） + builtin_tools = _get_builtin_tools() + tools.extend(builtin_tools) + + return tools + + +def _extract_fc_def(plugin) -> Optional[Dict]: + """从插件实例提取 FC 定义""" + if not hasattr(plugin, 'get_mcp_definition'): + return None + + try: + mcp_def = plugin.get_mcp_definition() + return mcp_to_fc(mcp_def) + except Exception: + return None + + +def _tool_info_to_fc(tool_info: dict) -> Optional[Dict]: + """从 tool_info 字典转为 FC 定义""" + name = tool_info.get('name', '') + desc = tool_info.get('description', '') + if not name: + return None + + return { + "type": "function", + "function": { + "name": name, + "description": desc, + "parameters": { + "type": "object", + "properties": { + "args": { + "type": "string", + "description": f"操作参数。格式参见 {name} 工具说明" + } + }, + "required": ["args"] + } + } + } + + +def mcp_to_fc(mcp_def: dict) -> Optional[Dict]: + """ + MCP 定义 → OpenAI FC 定义 + + MCP 格式: {"name": ..., "description": ..., "inputSchema": {...}} + OpenAI FC: {"type": "function", "function": {"name": ..., "description": ..., "parameters": {...}}} + """ + name = mcp_def.get('name', '') + desc = mcp_def.get('description', '') + schema = mcp_def.get('inputSchema', mcp_def.get('parameters', {})) + + if not name: + return None + + # 如果 schema 为空，生成默认的 args 参数 + if not schema or not schema.get('properties'): + schema = { + "type": "object", + "properties": { + "args": { + "type": "string", + "description": f"操作参数字符串，格式参见 {name} 工具说明" + } + }, + "required": ["args"] + } + + return { + "type": "function", + "function": { + "name": name, + "description": desc, + "parameters": schema + } + } + + +def fc_result_to_message(tool_call_id: str, tool_name: str, result: str) -> Dict: + """ + 工具执行结果 → OpenAI 消息格式 + + Returns: + {"role": "tool", "tool_call_id": ..., "content": ...} + """ + return { + "role": "tool", + "tool_call_id": tool_call_id, + "name": tool_name, + "content": result[:10000] if result else "" # 截断过长结果 + } + + +def parse_fc_response(response_data: dict) -> Optional[List[Dict]]: + """ + 解析 OpenAI FC 响应，提取 tool_calls + + Returns: + [{"id": "call_xxx", "name": "tool_name", "arguments": {...}}, ...] + 或 None（如果不是 FC 响应） + """ + choices = response_data.get("choices", []) + if not choices: + return None + + message = choices[0].get("message", {}) + tool_calls_raw = message.get("tool_calls") + + if not tool_calls_raw: + return None + + tool_calls = [] + for tc in tool_calls_raw: + func = tc.get("function", {}) + name = func.get("name", "") + args_str = func.get("arguments", "{}") + + # 解析 arguments JSON + try: + if isinstance(args_str, str): + args = json.loads(args_str) + else: + args = args_str + except json.JSONDecodeError: + args = {"raw": args_str} + + # 统一为小狸内部格式: {"action": "use_tool", "tool": name, "args": "..." } + # 如果 args 里有 args 字段，直接用；否则把整个 args 序列化 + if "args" in args and isinstance(args["args"], str): + final_args = args["args"] + elif "args" in args: + final_args = json.dumps(args["args"], ensure_ascii=False) + else: + # 把所有参数拼成字符串 + parts = [] + for k, v in args.items(): + if isinstance(v, str): + parts.append(f"{v}") + else: + parts.append(json.dumps(v, ensure_ascii=False)) + final_args = " ".join(parts) if parts else "" + + tool_calls.append({ + "id": tc.get("id", f"call_{name}"), + "name": name, + "arguments": args, + # 小狸内部格式 + "action": "use_tool", + "tool": name, + "args": final_args, + }) + + return tool_calls + + +def _get_builtin_tools() -> List[Dict]: + """内置工具定义（不依赖插件系统）""" + return [ + { + "type": "function", + "function": { + "name": "memory", + "description": "持久化记忆系统。写日记、记录长期记忆、搜索历史记忆和聊天记录。", + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["write", "diary", "remember", "search", "context", + "diary_list", "diary_read", "chat_list", "chat_search"], + "description": "操作类型" + }, + "content": { + "type": "string", + "description": "内容或关键词" + } + }, + "required": ["action"] + } + } + }, + { + "type": "function", + "function": { + "name": "scheduler", + "description": "定时任务工具，设定定时提醒。支持相对时间(30m/2h)、今日时间(14:30)、绝对时间、重复任务(r:前缀)。", + "parameters": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["add", "list", "delete", "clear"], + "description": "操作类型" + }, + "time": { + "type": "string", + "description": "时间: 30m / 14:30 / 2026-04-28 10:00 / r:30m" + }, + "message": { + "type": "string", + "description": "任务描述" + }, + "task_id": { + "type": "integer", + "description": "任务ID (用于 delete)" + } + }, + "required": ["action"] + } + } + } + ]