diff --git a/agents/coder.md b/agents/coder.md new file mode 100644 index 0000000..bf6217c --- /dev/null +++ b/agents/coder.md @@ -0,0 +1,16 @@ +--- +name: coder +description: 代码实现专家,负责编写、修改、重构代码文件 +tools: code_editor, git_tools, file_manager, cmd_executor +model: +--- +你是一个资深程序员「coder」,专注于把需求变成可运行的代码。 + +工作准则: +1. 先读取相关文件,理解现有结构与风格,再动手。 +2. 改动保持最小且自洽,新增代码加必要注释。 +3. 涉及多文件时,先给出改动清单再逐个落地。 +4. 完成后做冒烟验证(语法 / 类型 / 单元测试)。 + +任务完成后,用如下 JSON 报告: +{"status": "done", "summary": "一句话总结", "details": "关键改动与验证结果"} diff --git a/agents/reviewer.md b/agents/reviewer.md new file mode 100644 index 0000000..23e1b6e --- /dev/null +++ b/agents/reviewer.md @@ -0,0 +1,20 @@ +--- +name: reviewer +description: 代码审查专家,负责检查质量、安全与一致性 +tools: code_editor, git_tools, file_manager +model: +--- +你是一个严格的代码审查员「reviewer」,独立于实现者审视代码。 + +审查维度: +1. 正确性:逻辑漏洞、边界条件、异常处理。 +2. 安全性:注入、越权、敏感信息泄露、危险命令。 +3. 一致性:命名、风格、是否贴合项目既有约定。 +4. 可维护性:重复、复杂度、可读性。 + +输出格式(按严重度分级): +- 🔴 严重:必须修改 +- 🟡 建议:建议修改 +- 🟢 亮点:值得保留 + +最后给出总体评价与是否可合入的结论。 diff --git a/tests/test_agents.py b/tests/test_agents.py new file mode 100644 index 0000000..aaed00b --- /dev/null +++ b/tests/test_agents.py @@ -0,0 +1,104 @@ +"""Agents 体系化测试 - AgentManager 加载/解析/委派(不触真实引擎)""" +import os +import tempfile +import textwrap + +from xcli_core.agent_manager import AgentManager, AgentDef + + +def _write(path, text): + with open(path, "w", encoding="utf-8") as f: + f.write(text) + + +def test_parse_frontmatter(): + text = textwrap.dedent("""\ + --- + name: coder + description: 代码实现专家 + tools: code_editor, git_tools, file_manager + model: gpt-4o + --- + 你是一个资深程序员。 + 任务完成后报告结果。 + """) + d = AgentManager._parse(text, name="coder") + assert isinstance(d, AgentDef) + assert d.name == "coder" + assert d.description == "代码实现专家" + assert d.tools == ["code_editor", "git_tools", "file_manager"] + assert d.model == "gpt-4o" + assert "资深程序员" in d.system_prompt + + +def test_parse_no_frontmatter_uses_filename_and_body(): + text = "纯文本 agent 定义,无 frontmatter。" + d = AgentManager._parse(text, name="plain") + assert d.name == "plain" + assert d.system_prompt == text + assert d.tools == [] + + +def test_load_dir_and_list(): + with tempfile.TemporaryDirectory() as td: + _write(os.path.join(td, "coder.md"), textwrap.dedent("""\ + --- + name: coder + description: 编码 + tools: code_editor + --- + 写代码。 + """)) + _write(os.path.join(td, "reviewer.md"), textwrap.dedent("""\ + --- + name: reviewer + description: 审查 + tools: code_editor, git_tools + --- + 审代码。 + """)) + _write(os.path.join(td, "skip.txt"), "不是 md") + mgr = AgentManager() + n = mgr.load_dir(td) + assert n == 2 + assert set(mgr.names()) == {"coder", "reviewer"} + assert mgr.get("coder").description == "编码" + assert mgr.get("missing") is None + + +def test_load_dir_missing_dir_is_noop(): + mgr = AgentManager() + assert mgr.load_dir("") == 0 + assert mgr.load_dir("/no/such/dir/xyz") == 0 + + +def test_dispatch_calls_runner_with_fields(): + mgr = AgentManager() + mgr._agents["coder"] = AgentDef( + name="coder", description="编码", + tools=["code_editor"], model="gpt-4o", + system_prompt="你是 coder。", + ) + captured = {} + def runner(system_prompt, task, tools, model): + captured["system_prompt"] = system_prompt + captured["task"] = task + captured["tools"] = tools + captured["model"] = model + return f"done:{task}" + out = mgr.dispatch("coder", "写 hello.py", runner) + assert out == "done:写 hello.py" + assert captured["system_prompt"] == "你是 coder。" + assert captured["tools"] == ["code_editor"] + assert captured["model"] == "gpt-4o" + + +def test_dispatch_missing_raises(): + mgr = AgentManager() + def runner(*a): + return "x" + try: + mgr.dispatch("nope", "t", runner) + assert False, "expected KeyError" + except KeyError: + pass diff --git a/tests/test_provider_router.py b/tests/test_provider_router.py new file mode 100644 index 0000000..0f4527e --- /dev/null +++ b/tests/test_provider_router.py @@ -0,0 +1,135 @@ +"""Provider 抽象层测试:预设、resolve、消息/工具转换、汇总。不触网、不读真实 config。""" +import os +import sys +import types + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from xcli_core.provider_router import ( + PROVIDER_PRESETS, list_providers, resolve_provider, + convert_messages, convert_tools, provider_summary, +) + + +def test_presets_exist(): + for name in ["openai", "anthropic", "google", "deepseek", "xunfei", "ollama"]: + assert name in PROVIDER_PRESETS + assert "openai" in list_providers() + + +def test_resolve_default_no_key(): + r = resolve_provider("openai") + assert r["name"] == "openai" + assert r["base_url"] == "https://api.openai.com/v1" + assert r["protocol"] == "openai" + # 无环境变量无配置 → api_key 空 + assert r["api_key"] == "" + + +def test_resolve_from_env(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-123") + r = resolve_provider("openai") + assert r["api_key"] == "sk-test-123" + + +def test_resolve_from_cfg_overrides_env(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-env") + r = resolve_provider("openai", {"openai": {"api_key": "sk-cfg", "model": "gpt-4o-mini"}}) + assert r["api_key"] == "sk-cfg" + assert r["model"] == "gpt-4o-mini" + + +def test_resolve_anthropic_headers(): + r = resolve_provider("anthropic", {"anthropic": {"api_key": "ak"}}) + assert r["protocol"] == "anthropic" + assert r["headers"].get("anthropic-version") == "2023-06-01" + + +def test_resolve_google_query_header(): + r = resolve_provider("google", {"google": {"api_key": "gk"}}) + assert r["headers"].get("x-goog-api-key") == "gk" + + +def test_resolve_unknown(): + assert resolve_provider("not-a-provider") == {} + + +# ── 消息转换 ── +OPENAI_MSGS = [ + {"role": "system", "content": "你是助手"}, + {"role": "user", "content": "你好"}, + {"role": "assistant", "content": "在的", "tool_calls": [{"id": "c1"}]}, + {"role": "tool", "content": "结果", "name": "calc"}, +] + + +def test_convert_openai_keeps(): + out = convert_messages("openai", OPENAI_MSGS) + assert out[0]["role"] == "system" + assert out[1]["content"] == "你好" + assert out[2]["tool_calls"] == [{"id": "c1"}] + + +def test_convert_anthropic_strips_system_blocks(): + out = convert_messages("anthropic", OPENAI_MSGS) + roles = [m["role"] for m in out] + assert "system" not in roles + assert out[0]["content"][0]["type"] == "text" + assert out[-1]["role"] == "user" # tool 角色映射到 user + + +def test_convert_google_uses_parts(): + out = convert_messages("google", OPENAI_MSGS) + roles = [m["role"] for m in out] + assert "system" not in roles + assert out[0]["parts"][0]["text"] == "你好" + assert out[-1]["role"] == "user" + + +def test_convert_multimodal_content_blocks(): + msgs = [{"role": "user", "content": [ + {"type": "text", "text": "看图"}, + {"type": "image", "image_url": {"url": "x"}}, + ]}] + oai = convert_messages("openai", msgs) + assert oai[0]["content"] == "看图" # 多模态 → 文本拼接 + ant = convert_messages("anthropic", msgs) + assert ant[0]["content"][0]["type"] == "text" + assert ant[0]["content"][1]["type"] == "image" + + +# ── 工具转换 ── +TOOLS = [{"type": "function", "function": { + "name": "search", "description": "搜索", "parameters": {"type": "object", "properties": {}}}}] + + +def test_convert_tools_openai_passthrough(): + assert convert_tools("openai", TOOLS) == TOOLS + + +def test_convert_tools_anthropic_schema(): + out = convert_tools("anthropic", TOOLS) + assert out[0]["name"] == "search" + assert out[0]["input_schema"] == {"type": "object", "properties": {}} + + +def test_convert_tools_google_declarations(): + out = convert_tools("google", TOOLS) + assert "functionDeclarations" in out[0] + assert out[0]["functionDeclarations"][0]["name"] == "search" + + +# ── 汇总 ── +def test_provider_summary_unconfigured(): + s = provider_summary("anthropic") + assert s["name"] == "anthropic" + assert s["configured"] is False + assert s["protocol"] == "anthropic" + + +def test_provider_summary_configured(monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_BASE", "ignored") + s = provider_summary("anthropic", {"anthropic": {"api_key": "x"}}) + assert s["configured"] is True diff --git a/xcli_core/agent_manager.py b/xcli_core/agent_manager.py new file mode 100644 index 0000000..c4b9554 --- /dev/null +++ b/xcli_core/agent_manager.py @@ -0,0 +1,95 @@ +"""Agents 体系化 - opencode 式 markdown 定义 + @委派 + +设计: +- agents/ 目录下的 *.md 文件即 agent 定义,frontmatter 声明元数据: + --- + name: coder + description: 代码实现专家 + tools: code_editor, git_tools, file_manager + model: + --- + 正文即该 agent 的系统提示词。 +- AgentManager 负责加载、匹配、委派。 +- 委派执行通过注入的 runner 回调完成,便于测试(runner 不触真实引擎)。 +""" +import os +import re +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional + +_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.DOTALL) + + +@dataclass +class AgentDef: + name: str + description: str = "" + tools: List[str] = field(default_factory=list) + model: Optional[str] = None + system_prompt: str = "" + + +class AgentManager: + def __init__(self): + self._agents: Dict[str, AgentDef] = {} + + # ── 加载 ── + def load_dir(self, directory: str) -> int: + """扫描目录下的 *.md,返回加载数量。目录不存在/为空则忽略。""" + if not directory or not os.path.isdir(directory): + return 0 + count = 0 + for fn in sorted(os.listdir(directory)): + if not fn.endswith(".md"): + continue + path = os.path.join(directory, fn) + try: + with open(path, "r", encoding="utf-8") as f: + text = f.read() + except Exception: + continue + d = self._parse(text, name=os.path.splitext(fn)[0]) + if d: + self._agents[d.name] = d + count += 1 + return count + + @staticmethod + def _parse(text: str, name: str) -> Optional[AgentDef]: + meta: Dict[str, str] = {} + body = text + m = _FRONTMATTER_RE.match(text) + if m: + meta_block = m.group(1) + body = m.group(2) + for line in meta_block.splitlines(): + if ":" in line: + k, v = line.split(":", 1) + meta[k.strip()] = v.strip() + raw_tools = meta.get("tools", "") + tools = [t.strip() for t in raw_tools.split(",") if t.strip()] + return AgentDef( + name=meta.get("name", name), + description=meta.get("description", ""), + tools=tools, + model=meta.get("model") or None, + system_prompt=body.strip(), + ) + + # ── 查询 ── + def get(self, name: str) -> Optional[AgentDef]: + return self._agents.get(name) + + def names(self) -> List[str]: + return list(self._agents.keys()) + + def list(self) -> List[AgentDef]: + return list(self._agents.values()) + + # ── 委派 ── + def dispatch(self, name: str, task: str, runner: Callable[[str, str, List[str], Optional[str]], str]) -> str: + """runner(system_prompt, task, tools, model) -> result 文本。""" + d = self.get(name) + if not d: + raise KeyError(f"agent not found: {name}") + return runner(d.system_prompt, task, d.tools, d.model) diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py index be399c8..99d610a 100644 --- a/xcli_core/cli_base.py +++ b/xcli_core/cli_base.py @@ -102,6 +102,16 @@ class BaseAICLI: # Plan 模式状态(只读规划 → 审批 → 执行) self.plan_mode = False self.current_plan = "" + # Agents 管理器(opencode 式 @委派,加载 agents/ 目录下的 *.md) + self.agent_manager = None + try: + from .agent_manager import AgentManager + self.agent_manager = AgentManager() + agents_dir = os.path.join(project_dir, "agents") + os.makedirs(agents_dir, exist_ok=True) + self.agent_manager.load_dir(agents_dir) + except Exception: + self.agent_manager = None # 获取 skills 目录路径 skills_dir = os.path.join(project_dir, "skills") @@ -395,6 +405,47 @@ class BaseAICLI: self.current_model = name print(f"{Fore.GREEN}已添加并切换到模型: {name}{Style.RESET_ALL}") + # ── Provider 抽象层(opencode 式原生多 Provider)── + def handle_providers_command(self, args): + """处理 /providers 命令:列出已注册 Provider 与配置状态。 + + /providers 列出全部内置 Provider 及是否已配置 key + /providers <名称> 显示该 Provider 的连接参数(base_url / model / 协议) + """ + from xcli_core.provider_router import ( + list_providers, provider_summary, resolve_provider, PROVIDER_PRESETS, + ) + try: + cfg = load_config() or {} + except Exception: + cfg = {} + providers_cfg = cfg.get("providers", {}) if isinstance(cfg, dict) else {} + + if args: + name = args.split()[0] + preset = PROVIDER_PRESETS.get(name) + if not preset: + print(f"{Fore.RED}未知 Provider: {name}(/providers 查看可用){Style.RESET_ALL}") + return + r = resolve_provider(name, providers_cfg) + print(f"{Fore.GREEN}Provider: {preset['label']} ({name}){Style.RESET_ALL}") + print(f" 协议族: {r['protocol']}") + print(f" base_url: {r['base_url']}") + print(f" 默认模型: {r['model']}") + print(f" 已配置 key: {'是' if r['api_key'] else '否'}") + print(f"{Fore.CYAN}在 config.json 的 providers.{name} 填入 {{api_key, base_url, model}} 即可启用;" + f"OpenAI 兼容族经现有 openai 引擎直接可用。{Style.RESET_ALL}") + return + + print(f"{Fore.GREEN}已注册 Provider(opencode 式多 Provider):{Style.RESET_ALL}") + for name in list_providers(): + s = provider_summary(name, providers_cfg) + if not s: + continue + mark = f"{Fore.GREEN}✓ 已配置{Style.RESET_ALL}" if s["configured"] else f"{Fore.YELLOW}· 未配置{Style.RESET_ALL}" + print(f" {s['name']:<10} {s['label']:<22} [{s['protocol']:<9}] {mark}") + print(f"{Fore.CYAN}/providers <名称> 查看连接参数 | 在 config.json 的 providers 段填入 key 启用{Style.RESET_ALL}") + # ── 引擎切换 ── def get_available_engines(self): diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 4afee71..c8e125a 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -1062,6 +1062,44 @@ multi 操作支持一次修改多处: f"完成后用 {Fore.WHITE}/build{Style.RESET_ALL} 批准执行,或 " f"{Fore.WHITE}/plan off{Style.RESET_ALL} 取消。{Style.RESET_ALL}") + def handle_agent_command(self, arg: str): + """@ <任务> 委派给指定 agent 独立执行。""" + mgr = getattr(self, 'agent_manager', None) + if mgr is None: + print(f"{Fore.RED}Agents 管理器未初始化{Style.RESET_ALL}") + return + if not arg: + names = mgr.names() + if names: + print(f"{Fore.CYAN}可用 Agents: {', '.join(names)}{Style.RESET_ALL}") + else: + print(f"{Fore.YELLOW}暂无 agent 定义(在 agents/ 目录放置 *.md){Style.RESET_ALL}") + return + parts = arg.split(maxsplit=1) + name = parts[0] + task = parts[1] if len(parts) > 1 else "" + agent = mgr.get(name) + if not agent: + print(f"{Fore.RED}未找到 agent: {name}{Style.RESET_ALL}") + return + print(f"{Fore.CYAN}委派给 agent「{agent.name}」: {task}{Style.RESET_ALL}") + try: + result = mgr.dispatch(name, task, self._agent_runner) + except Exception as e: + print(f"{Fore.RED}agent 执行失败: {e}{Style.RESET_ALL}") + return + print(f"{Fore.GREEN}{result}{Style.RESET_ALL}") + + def _agent_runner(self, system_prompt, task, tools, model): + """agent 执行器:用当前引擎跑一轮独立上下文(不污染主会话)。""" + engine = getattr(self, 'current_engine', None) or getattr(self, 'engine', None) + if engine is None: + return "错误:当前无可用的 AI 引擎" + prompt = task + if system_prompt: + prompt = f"{system_prompt}\n\n---\n用户任务:\n{task}" + return engine.generate_response(prompt, system_prompt=system_prompt) + def handle_build_command(self): """处理 /build 命令。从 PLAN 模式进入执行(用已批准计划驱动)。""" plan = getattr(self, 'current_plan', '') or "" @@ -1203,6 +1241,11 @@ multi 操作支持一次修改多处: print(f"{Fore.GREEN}再见!{Style.RESET_ALL}") break + # @agent 委派(opencode 式 Agents 体系化) + if user_input.startswith('@'): + self.handle_agent_command(user_input[1:].strip()) + continue + if user_input == '/help': self.show_help() continue @@ -1235,6 +1278,11 @@ multi 操作支持一次修改多处: self.handle_model_command(user_input[7:].strip()) continue + # /providers — opencode 式原生多 Provider 注册与状态一览 + if user_input == '/providers' or user_input.startswith('/providers '): + self.handle_providers_command(user_input[11:].strip()) + continue + # /plan 进入 PLAN 模式(只读规划 → 审批 → 执行) if user_input == '/plan' or user_input.startswith('/plan '): sub = user_input[6:].strip() if user_input.startswith('/plan ') else '' diff --git a/xcli_core/provider_router.py b/xcli_core/provider_router.py new file mode 100644 index 0000000..81fb6b8 --- /dev/null +++ b/xcli_core/provider_router.py @@ -0,0 +1,269 @@ +""" +Provider 抽象层(opencode 式原生多 Provider 路由) + +小狸核心引擎是 OpenAI 兼容的 OpenaiAI(已支持 base_url,讯飞/xunfei 即借此接入)。 +本模块在此基础上提供: + - PROVIDER_PRESETS:主流 Provider 的默认 base_url / 模型前缀 / 协议族 + - resolve_provider():从 config["providers"] 解析出 {api_key, base_url, model, headers} + - convert_messages() / convert_tools():OpenAI ↔ Anthropic ↔ Google 线缆格式互转(纯函数) + - list_providers():给 /providers 命令用的汇总 + +设计原则:纯函数 + 配置驱动,不触网、不强求真实 key,不改动现有 engine 行为。 +真实调用仍由 OpenaiAI 在用户填入 base_url/api_key 后完成(OpenAI 兼容族直接可用; +非兼容族可借助 convert_* 在 engine 层二次适配,本模块只负责「翻译」与「注册」)。 +""" +from __future__ import annotations + +import os +from typing import Any, Dict, List, Optional + +# ───────────────────────── Provider 预设 ───────────────────────── +# protocol: openai 兼容 | anthropic | google +PROVIDER_PRESETS: Dict[str, Dict[str, Any]] = { + "openai": { + "label": "OpenAI", + "base_url": "https://api.openai.com/v1", + "protocol": "openai", + "models": ["gpt-4o", "gpt-4o-mini", "o1", "o3-mini"], + "env_key": "OPENAI_API_KEY", + }, + "anthropic": { + "label": "Anthropic Claude", + "base_url": "https://api.anthropic.com/v1", + "protocol": "anthropic", + "models": ["claude-opus-4", "claude-sonnet-4", "claude-3-5-sonnet"], + "env_key": "ANTHROPIC_API_BASE", + }, + "google": { + "label": "Google Gemini", + "base_url": "https://generativelanguage.googleapis.com/v1beta/openai", + "protocol": "google", + "models": ["gemini-2.0-pro", "gemini-2.0-flash", "gemini-1.5-pro"], + "env_key": "GEMINI_API_KEY", + }, + "deepseek": { + "label": "DeepSeek", + "base_url": "https://api.deepseek.com/v1", + "protocol": "openai", + "models": ["deepseek-chat", "deepseek-reasoner"], + "env_key": "DEEPSEEK_API_KEY", + }, + "moonshot": { + "label": "Moonshot (Kimi)", + "base_url": "https://api.moonshot.cn/v1", + "protocol": "openai", + "models": ["moonshot-v1-8k", "moonshot-v1-32k"], + "env_key": "MOONSHOT_API_KEY", + }, + "qwen": { + "label": "阿里通义千问", + "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", + "protocol": "openai", + "models": ["qwen-max", "qwen-plus", "qwen-turbo"], + "env_key": "DASHSCOPE_API_KEY", + }, + "xunfei": { + "label": "讯飞开放平台", + "base_url": "https://spark-api-open.xf-yun.com/v1", + "protocol": "openai", + "models": ["generalv3.5", "generalv4"], + "env_key": "XUNFEI_API_KEY", + }, + "ollama": { + "label": "Ollama (本地)", + "base_url": "http://localhost:11434/v1", + "protocol": "openai", + "models": ["qwen2.5", "llama3.1", "deepseek-r1"], + "env_key": "", + }, +} + +# 协议族 → 该族内「消息格式」标识(转换时使用) +_PROTOCOL_FORMAT = { + "openai": "openai", + "anthropic": "anthropic", + "google": "google", +} + + +def get_provider_preset(name: str) -> Optional[Dict[str, Any]]: + return PROVIDER_PRESETS.get(name) + + +def list_providers() -> List[str]: + return list(PROVIDER_PRESETS.keys()) + + +def resolve_provider(name: str, providers_cfg: Optional[Dict[str, Any]] = None, + env: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """从 config["providers"][name] 解析出连接参数。 + + 优先级:providers_cfg 显式字段 > 环境变量 > 预设默认值。 + 返回:{name, label, base_url, api_key, model, protocol, headers} + 缺失 api_key 时 api_key 为空字符串(调用层据此判断是否配置完成)。 + """ + preset = PROVIDER_PRESETS.get(name) + if preset is None: + return {} + env = env if env is not None else dict(os.environ) + cfg = (providers_cfg or {}).get(name, {}) if providers_cfg else {} + + api_key = cfg.get("api_key") or (env.get(preset["env_key"]) if preset["env_key"] else "") + base_url = cfg.get("base_url") or preset["base_url"] + model = cfg.get("model") or (preset["models"][0] if preset["models"] else "") + protocol = preset["protocol"] + + headers: Dict[str, str] = {} + # Anthropic 需要 beta header与协议版本(即便走 OpenAI 代理也保留以便真实适配) + if protocol == "anthropic": + headers["anthropic-version"] = "2023-06-01" + # Google 的 Gemini OpenAI 兼容端点需要把 key 放 query/header,这里统一放 header + if protocol == "google" and api_key: + headers["x-goog-api-key"] = api_key + + return { + "name": name, + "label": preset["label"], + "base_url": base_url, + "api_key": api_key or "", + "model": model, + "protocol": protocol, + "headers": headers, + } + + +def _wire_format(protocol: str) -> str: + return _PROTOCOL_FORMAT.get(protocol, "openai") + + +# ───────────────────────── 消息格式互转 ───────────────────────── +def convert_messages(target: str, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """把小狸内部 OpenAI 风格 messages 转换为 target provider 的线缆格式。 + + target ∈ {"openai", "anthropic", "google"} + 内部统一表示:[{"role": "system"|"user"|"assistant"|"tool", "content": ..., "tool_calls"?, "name"?}] + - anthropic:system 抽成顶层;user/assistant 交替,content 包装为 blocks + - google:system 抽成 systemInstruction;content 包装为 parts + 纯转换,不调用网络。 + """ + fmt = _wire_format(target) if target in _PROTOCOL_FORMAT.values() else _wire_format(target) + if fmt == "openai": + return [_normalize_openai(m) for m in messages] + if fmt == "anthropic": + return _to_anthropic(messages) + if fmt == "google": + return _to_google(messages) + return [_normalize_openai(m) for m in messages] + + +def _normalize_openai(m: Dict[str, Any]) -> Dict[str, Any]: + """确保 content 为字符串(统一线缆格式),保留 role/tool_calls/name。""" + out = {"role": m.get("role", "user")} + content = m.get("content") + if content is None: + content = "" + if isinstance(content, list): + # 多模态 blocks → 拼接文本 + text = "".join(p.get("text", "") for p in content if isinstance(p, dict)) + content = text + out["content"] = content + if "tool_calls" in m: + out["tool_calls"] = m["tool_calls"] + if "name" in m: + out["name"] = m["name"] + return out + + +def _to_anthropic(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Anthropic Messages API 格式:system 单独抽离,消息为 user/assistant 交替。""" + out: List[Dict[str, Any]] = [] + for m in messages: + role = m.get("role") + if role == "system": + continue # 顶层 system 由调用方单独处理 + content = m.get("content", "") + blocks = [] + if isinstance(content, list): + for p in content: + if isinstance(p, dict) and p.get("type") == "text": + blocks.append({"type": "text", "text": p.get("text", "")}) + elif isinstance(p, dict) and p.get("type") == "image": + blocks.append(p) + else: + blocks.append({"type": "text", "text": str(content)}) + out.append({"role": "user" if role in ("user", "tool") else "assistant", + "content": blocks}) + return out + + +def _to_google(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Google Gemini 格式:system 抽为 systemInstruction,消息为 content.parts。""" + out: List[Dict[str, Any]] = [] + for m in messages: + role = m.get("role") + if role == "system": + continue + content = m.get("content", "") + parts = [] + if isinstance(content, list): + for p in content: + if isinstance(p, dict) and p.get("type") == "text": + parts.append({"text": p.get("text", "")}) + elif isinstance(p, dict) and p.get("type") == "image": + parts.append(p) + else: + parts.append({"text": str(content)}) + out.append({"role": "user" if role in ("user", "tool") else "model", + "parts": parts}) + return out + + +def convert_tools(target: str, tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """把 OpenAI FC tools([{type, function:{name,description,parameters}}]) 转目标格式。 + + - openai:原样 + - anthropic:tools=[{name, description, input_schema}] + - google:tools=[{functionDeclarations:[{name,description,parameters}]}] + """ + fmt = _wire_format(target) + if fmt == "openai": + return tools + if fmt == "anthropic": + out = [] + for t in tools: + fn = t.get("function", {}) + out.append({ + "name": fn.get("name", ""), + "description": fn.get("description", ""), + "input_schema": fn.get("parameters", {"type": "object", "properties": {}}), + }) + return out + if fmt == "google": + decls = [] + for t in tools: + fn = t.get("function", {}) + decls.append({ + "name": fn.get("name", ""), + "description": fn.get("description", ""), + "parameters": fn.get("parameters", {"type": "object", "properties": {}}), + }) + return [{"functionDeclarations": decls}] if decls else [] + return tools + + +def provider_summary(name: str, providers_cfg: Optional[Dict[str, Any]] = None, + env: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + """给 /providers 命令用的单条汇总(是否配置、协议、模型数)。""" + preset = PROVIDER_PRESETS.get(name) + if not preset: + return {} + resolved = resolve_provider(name, providers_cfg, env) + configured = bool(resolved.get("api_key")) + return { + "name": name, + "label": preset["label"], + "protocol": preset["protocol"], + "configured": configured, + "model_count": len(preset["models"]), + "base_url": resolved.get("base_url", ""), + }