写一个Agent参数校验层,告别模型生成乱参

> 本文同步发布在 Agent评测站,一位在Agent坑里泡了三个月的开发者。


Agent翻车最多的地方不是模型不行,是模型生成的工具调用参数全是错的


字段名不对、类型不匹配、多了不存在的参数、少了必填字段——我见过的翻车案例里,80%跟参数传递有关。


这篇直接给一个可复用的校验层,写一次,所有Agent都受益。


---


问题的典型表现


模型调你的工具时,传的参数一般是这样的:


# 模型自以为传对了的参数
{
  "action": "query_database",
  "table": "defects",
  "limit": "10",
  "date_range": "2025年1月"
}

# 但你的工具实际上要的是:
{
  "sql": "SELECT * FROM defects WHERE date >= ? AND date <= ?",
  "params": ["2025-01-01", "2025-01-31"],
  "max_rows": 10
}

差异点:字段名不一样、limit是字符串不是数字、日期格式要转换。


直接传进去,工具崩了,模型还不知道为什么崩。


---


直接上代码:一个通用的参数校验器


# tool_validator.py
"""
Agent工具调用参数校验层
负责:类型转换、字段映射、缺失字段检测、额外字段过滤
"""
import json
from typing import Dict, Any, Optional, Callable

class ToolParamValidator:
    """
    工具参数校验器
    
    用法:
    validator = ToolParamValidator({
        "sql": {"type": str, "required": True, "desc": "SQL查询语句"},
        "params": {"type": list, "required": False, "default": [], "desc": "查询参数"},
        "max_rows": {"type": int, "required": False, "default": 50, "desc": "最大返回行数"},
    })
    
    validated = validator.validate({
        "sql": "SELECT * FROM defects",
        "max_rows": "100"  # 字符串"100"会被自动转为int
    })
    # → {"sql": "SELECT * FROM defects", "params": [], "max_rows": 100}
    """
    
    def __init__(self, schema: Dict[str, Dict]):
        """
        schema格式:
        {
            "field_name": {
                "type": str | int | float | bool | list | dict,  # 期望类型
                "required": bool,         # 是否必填
                "default": Any,           # 默认值(非必填时生效)
                "choices": [...],         # 可选值列表(如有)
                "validator": Callable,    # 自定义校验函数
                "alias": str,             # 字段别名(模型可能用不同名字)
                "desc": str               # 字段说明(仅文档用)
            }
        }
        """
        self.schema = schema
        # 构建别名映射
        self.alias_map = {}
        for field, spec in schema.items():
            alias = spec.get("alias")
            if alias:
                self.alias_map[alias] = field
    
    def validate(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        校验并清洗模型传入的参数
        返回清洗后的参数字典
        抛 ValueError 如果必填参数缺失或参数类型错误(且无法转换)
        """
        # 1. 处理别名
        resolved = {}
        for key, value in params.items():
            real_key = self.alias_map.get(key, key)
            resolved[real_key] = value
        
        # 2. 过滤不存在的字段(模型可能多传参数)
        result = {}
        extra_fields = []
        for key, value in resolved.items():
            if key in self.schema:
                result[key] = value
            else:
                extra_fields.append(key)
        
        # 3. 类型检查和转换
        for field, spec in self.schema.items():
            if field not in result:
                if spec.get("required", False):
                    raise ValueError(f"缺少必填参数: {field} ({spec.get('desc', '')})")
                else:
                    # 非必填参数用默认值
                    result[field] = spec.get("default")
                    continue
            
            value = result[field]
            expected_type = spec.get("type")
            
            if expected_type and value is not None:
                try:
                    # 如果类型不匹配,尝试转换
                    if not isinstance(value, expected_type):
                        if expected_type == int:
                            result[field] = int(value)
                        elif expected_type == float:
                            result[field] = float(value)
                        elif expected_type == str:
                            result[field] = str(value)
                        elif expected_type == list:
                            if isinstance(value, str):
                                # "a,b,c" → ["a", "b", "c"]
                                result[field] = [v.strip() for v in value.split(",")]
                            elif isinstance(value, (int, float)):
                                result[field] = [value]
                            else:
                                result[field] = list(value)
                        elif expected_type == bool:
                            if isinstance(value, str):
                                result[field] = value.lower() in ("true", "1", "yes", "是")
                            else:
                                result[field] = bool(value)
                except (ValueError, TypeError) as e:
                    raise ValueError(
                        f"参数 {field} 类型错误: 期望 {expected_type.__name__}, "
                        f"实际 {type(value).__name__}, 值={value}"
                    )
            
            # 4. 可选值检查
            choices = spec.get("choices")
            if choices and result[field] not in choices:
                raise ValueError(
                    f"参数 {field} 取值 {result[field]} 不在可选范围内: {choices}"
                )
            
            # 5. 自定义校验
            custom_validator = spec.get("validator")
            if custom_validator:
                result[field] = custom_validator(result[field])
        
        return result


# ============ 使用示例 ============

# 定义数据库查询工具的参数schema
db_query_schema = {
    "sql": {
        "type": str,
        "required": True,
        "desc": "SQL查询语句"
    },
    "params": {
        "type": list,
        "required": False,
        "default": [],
        "desc": "查询参数列表"
    },
    "max_rows": {
        "type": int,
        "required": False,
        "default": 50,
        "desc": "最大返回行数"
    },
    "table": {
        "type": str,
        "required": False,
        "default": "",
        "alias": "from",  # 模型可能写 "from": "defects"
        "desc": "查询的表名(备用字段)"
    },
}

validator = ToolParamValidator(db_query_schema)

# 示例1:模型传了正确的参数
try:
    result = validator.validate({
        "sql": "SELECT * FROM defects WHERE date = ?",
        "params": ["2026-05-26"],
        "max_rows": "20"  # 字符串会被自动转int
    })
    print(f"✅ 校验通过: {result}")
except ValueError as e:
    print(f"❌ 校验失败: {e}")

# 示例2:模型用了别名 "from" 而不是 "table"
try:
    result = validator.validate({
        "sql": "SELECT * FROM defects",
        "from": "defects",  # 别名自动映射
        "limit": 100  # 不存在的字段会被过滤
    })
    print(f"✅ 别名映射成功: {result}")
except ValueError as e:
    print(f"❌ 校验失败: {e}")

# 示例3:缺少必填参数
try:
    result = validator.validate({
        "max_rows": 10
    })
except ValueError as e:
    print(f"✅ (预期) 校验失败: {e}")

# 示例4:类型错误
try:
    result = validator.validate({
        "sql": "SELECT * FROM defects",
        "max_rows": "abc"  # 无法转int
    })
except ValueError as e:
    print(f"✅ (预期) 校验失败: {e}")

---


接入Agent的完整流程


写一个工具注册器,把校验器和工具绑定:


# tool_registry.py
from tool_validator import ToolParamValidator

class ToolRegistry:
    """Agent工具注册器——统一管理工具定义+参数校验+执行"""
    
    def __init__(self):
        self._tools = {}
    
    def register(self, name: str, handler: callable, 
                 schema: dict, description: str = ""):
        """
        注册一个工具
        name: 工具名(LLM调用时用的名字)
        handler: 实际执行函数
        schema: 参数schema
        description: 工具描述(给LLM看的)
        """
        self._tools[name] = {
            "handler": handler,
            "validator": ToolParamValidator(schema),
            "description": description,
            "schema_raw": schema,
        }
    
    def call(self, tool_name: str, raw_params: dict):
        """调用工具(带参数校验和安全检查)"""
        if tool_name not in self._tools:
            return {"error": f"未知工具: {tool_name},可用工具: {list(self._tools.keys())}"}
        
        tool = self._tools[tool_name]
        
        try:
            # 1. 参数校验
            validated = tool["validator"].validate(raw_params)
            
            # 2. 安全检查(高危操作需要确认)
            if tool_name in ("delete_file", "send_email", "execute_shell"):
                confirmed = self._human_confirm(tool_name, validated)
                if not confirmed:
                    return {"error": "操作已被拒绝", "reason": "需要人工确认"}
            
            # 3. 执行工具
            result = tool["handler"](**validated)
            
            # 4. 结果标准化
            return {
                "tool": tool_name,
                "status": "success",
                "data": result,
            }
        
        except ValueError as e:
            return {
                "tool": tool_name,
                "status": "error",
                "error": f"参数校验失败: {str(e)}",
                "raw_params": raw_params,
            }
        except Exception as e:
            return {
                "tool": tool_name,
                "status": "error",
                "error": f"执行失败: {str(e)}",
            }
    
    def _human_confirm(self, tool_name, params):
        """高危操作确认(实际项目中通过消息队列/WebSocket)"""
        print(f"\n⚠️ 高危操作: {tool_name}")
        print(f"   参数: {params}")
        # 生产环境这里发审批通知
        return True  # demo返回True
    
    def get_definitions(self):
        """获取所有工具定义(给LLM看的JSON Schema格式)"""
        definitions = []
        for name, tool in self._tools.items():
            props = {}
            required = []
            for field, spec in tool["schema_raw"].items():
                type_map = {str: "string", int: "integer", float: "number", 
                           bool: "boolean", list: "array", dict: "object"}
                field_type = type_map.get(spec.get("type"), "string")
                field_desc = spec.get("desc", "")
                props[field] = {"type": field_type, "description": field_desc}
                
                if spec.get("choices"):
                    props[field]["enum"] = spec["choices"]
                
                if spec.get("required"):
                    required.append(field)
            
            definitions.append({
                "type": "function",
                "function": {
                    "name": name,
                    "description": tool["description"],
                    "parameters": {
                        "type": "object",
                        "properties": props,
                        "required": required,
                    }
                }
            })
        return definitions


# ============ 完整使用示例 ============

import sqlite3

# 初始化
registry = ToolRegistry()

# 注册数据库查询工具
registry.register(
    name="query_database",
    handler=lambda sql, params=[], max_rows=50: 
        [dict(row) for row in sqlite3.connect(":memory:").execute(sql, params).fetchmany(max_rows)],
    schema={
        "sql": {"type": str, "required": True, "desc": "SQL查询语句"},
        "params": {"type": list, "required": False, "default": [], "desc": "查询参数"},
        "max_rows": {"type": int, "required": False, "default": 50, "desc": "最多返回行数"},
    },
    description="执行SQL数据库查询,返回结果列表"
)

# 在Agent中使用
class SafeAgent:
    def __init__(self, registry):
        self.registry = registry
        self.llm = OpenAI(api_key="sk-xxx")
    
    def process(self, query):
        # ...(其他处理逻辑)
        
        # 当LLM决定调用工具时:
        tool_call = self.llm.generate_tool_call(
            tools=self.registry.get_definitions(),
            query=query
        )
        
        # 通过校验器安全调用
        result = self.registry.call(
            tool_call["name"],
            tool_call["arguments"]
        )
        
        return result

---


进阶:自动生成Schema(不用手写)


手写schema费事?写个函数从函数签名自动生成:


import inspect

def auto_schema(func: callable) -> dict:
    """从函数签名自动生成参数schema"""
    sig = inspect.signature(func)
    schema = {}
    for name, param in sig.parameters.items():
        field = {
            "desc": f"参数 {name}",
            "required": param.default is inspect.Parameter.empty,
        }
        # 类型推断
        if param.annotation is not inspect.Parameter.empty:
            type_map = {
                str: str, int: int, float: float,
                bool: bool, list: list, dict: dict,
            }
            field["type"] = type_map.get(param.annotation, str)
        else:
            field["type"] = str
        
        if param.default is not inspect.Parameter.empty:
            field["required"] = False
            field["default"] = param.default
        
        schema[name] = field
    return schema

# 用法
def query_orders(order_id: str, max_rows: int = 10):
    """查询订单信息"""
    pass

schema = auto_schema(query_orders)
# → {"order_id": {"type": str, "required": True, "desc": "参数 order_id"},
#     "max_rows": {"type": int, "required": False, "default": 10, "desc": "参数 max_rows"}}

---


效果


上线参数校验层后,我这边Agent的工具调用成功率从73%升到了96%。剩下的4%失败主要是LLM选了错误的工具(不是参数问题),那是另一层的事了。


一句话总结:永远不要信模型生成的参数。加一层校验,省十次调试。


如果你也有翻车案例,评论区聊聊。