""" config_lint.py — 配置静态检查工具 ================================== 为 Critic 和 Optimizer agent 提供配置验证能力。 复用 lambdagent 的 schema 验证，并添加 builder 专用检查。 """ from __future__ import annotations import sys from pathlib import Path from typing import Dict, List, Tuple import yaml PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent sys.path.insert(0, str(PROJECT_ROOT)) # ── 常量 ────────────────────────────────────────────────── VALID_TYPES = {"react", "chain", "router", "parallel"} VALID_PROVIDERS = { "anthropic", "openai", "dashscope", "deepseek", "zhipu", "moonshot", "ollama", } PROVIDER_MODELS = { "anthropic": {"claude-sonnet-4-20250514", "claude-opus-4-20250514"}, "openai": {"gpt-4o", "gpt-4o-mini"}, "dashscope": {"qwen3-max-2026-01-23", "qwen-max", "qwen-plus"}, "deepseek": {"deepseek-chat", "deepseek-reasoner"}, "zhipu": {"glm-4-plus", "glm-4-flash"}, "moonshot": {"moonshot-v1-128k"}, "ollama": {"llama3", "qwen2"}, } VALID_LOCAL_TOOLS = { "terminate", "shell", "file", "browser", "app", "system", "screenshot", "done", } REQUIRED_FIELDS = ["agentId", "name", "type", "systemPrompt", "model"] # ── 检查函数 ────────────────────────────────────────────── def lint_config(config: Dict | str) -> Dict: """ 对 agent 配置进行静态检查。 Args: config: 配置字典或 YAML 字符串 Returns: { "valid": bool, "errors": [{"level": "error", "rule": "R001", "message": "..."}], "warnings": [{"level": "warning", "rule": "W001", "message": "..."}], "score": int # 0-100 质量分 } """ if isinstance(config, str): try: config = yaml.safe_load(config) except yaml.YAMLError as e: return { "valid": False, "errors": [{"level": "error", "rule": "R000", "message": f"YAML 解析失败: {e}"}], "warnings": [], "score": 0, } if not isinstance(config, dict): return { "valid": False, "errors": [{"level": "error", "rule": "R000", "message": "配置不是有效的字典"}], "warnings": [], "score": 0, } errors: List[Dict] = [] warnings: List[Dict] = [] # R001: 必填字段检查 for field in REQUIRED_FIELDS: if field not in config: errors.append({ "level": "error", "rule": "R001", "message": f"缺少必填字段: {field}", }) # R002: type 合法性 agent_type = config.get("type") if agent_type and agent_type not in VALID_TYPES: errors.append({ "level": "error", "rule": "R002", "message": f"无效的 type: {agent_type}，合法值: {VALID_TYPES}", }) # R003: model 配置 model = config.get("model", {}) if isinstance(model, dict): provider = model.get("provider") model_name = model.get("name") if provider and provider not in VALID_PROVIDERS: errors.append({ "level": "error", "rule": "R003", "message": f"无效的 provider: {provider}，合法值: {VALID_PROVIDERS}", }) if provider and model_name: valid_models = PROVIDER_MODELS.get(provider, set()) if valid_models and model_name not in valid_models: warnings.append({ "level": "warning", "rule": "W003", "message": f"model {model_name} 不在 {provider} 的已知模型列表中: {valid_models}", }) temp = model.get("temperature") if temp is not None and (temp < 0.0 or temp > 2.0): errors.append({ "level": "error", "rule": "R003", "message": f"temperature {temp} 超出范围 [0.0, 2.0]", }) # R004: type 与子配置匹配 if agent_type == "react" and "react" not in config: warnings.append({ "level": "warning", "rule": "W004", "message": "type=react 但缺少 react 配置块（将使用默认值）", }) if agent_type == "chain" and "chain" not in config: errors.append({ "level": "error", "rule": "R004", "message": "type=chain 必须有 chain 配置块", }) if agent_type == "router" and "router" not in config: errors.append({ "level": "error", "rule": "R004", "message": "type=router 必须有 router 配置块", }) # R005: systemPrompt 质量 prompt = config.get("systemPrompt", "") if isinstance(prompt, str): if len(prompt.strip()) < 20: errors.append({ "level": "error", "rule": "R005", "message": f"systemPrompt 过短（{len(prompt.strip())} 字符），至少 20 字符", }) elif len(prompt.strip()) < 50: warnings.append({ "level": "warning", "rule": "W005", "message": f"systemPrompt 较短（{len(prompt.strip())} 字符），建议 100+ 字符", }) # R006: mcp.localTools 必须包含 terminate mcp = config.get("mcp", {}) if isinstance(mcp, dict): local_tools = mcp.get("localTools", []) if isinstance(local_tools, list) and "terminate" not in local_tools: warnings.append({ "level": "warning", "rule": "W006", "message": "mcp.localTools 缺少 terminate（agent 可能无法正常结束）", }) # 检查工具名合法性 for tool in local_tools: if tool not in VALID_LOCAL_TOOLS and not tool.startswith("call_"): warnings.append({ "level": "warning", "rule": "W006", "message": f"未知的 localTool: {tool}", }) # R007: onlineTool 必须有对应的 app.mcp.custom.nodes online_tools = mcp.get("onlineTool", {}) if isinstance(mcp, dict) else {} if online_tools: app_nodes = _nested_get(config, "app", "mcp", "custom", "nodes") or {} for server_name in online_tools: if server_name not in app_nodes: errors.append({ "level": "error", "rule": "R007", "message": f"onlineTool 引用 {server_name} 但 app.mcp.custom.nodes 中缺少其配置", }) # W008: react.maxSteps 合理性 react_config = config.get("react", {}) if isinstance(react_config, dict): max_steps = react_config.get("maxSteps", 10) if max_steps > 30: warnings.append({ "level": "warning", "rule": "W008", "message": f"react.maxSteps={max_steps} 过大，可能导致 token 浪费", }) if max_steps < 2: warnings.append({ "level": "warning", "rule": "W008", "message": f"react.maxSteps={max_steps} 过小，agent 可能无法完成任务", }) # W009: guard 缺失 if "guard" not in config: warnings.append({ "level": "warning", "rule": "W009", "message": "缺少 guard 配置（建议至少设置 maxOutputLength）", }) # ── 计算总分 ── score = 100 score -= len(errors) * 15 score -= len(warnings) * 5 score = max(0, min(100, score)) return { "valid": len(errors) == 0, "errors": errors, "warnings": warnings, "score": score, } def _nested_get(d: dict, *keys): for k in keys: if isinstance(d, dict): d = d.get(k) else: return None return d