lambdagent/agentexample/agentbuilder67/tools/config_lint.py

"""
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