LLM Tool Calling 的工程化实践：Function Calling 与 Schema 校验

一、为什么需要工程化？

当你用 OpenAI 官方示例跑通第一个 Function Calling 时，可能觉得一切都很美好——模型听话地返回了结构化的工具调用请求，你的代码完美执行了天气查询。

但当你把 Agent 部署到生产环境，面对真实用户和复杂业务时，问题接踵而至：

• 模型返回的 arguments 字段是字符串，不是 JSON 对象，你忘了 json.loads() 直接报错
• 模型“编造”了一个你根本没注册的工具名称
• 参数类型对不上——要求 integer，模型给你传了 "42"
• 多步推理场景下，模型在中间步骤返回了最终结果，你的解析器直接懵了

这些问题归结为一点：你太过信任模型的输出了。

工程化的核心，就是用结构化的 Schema 校验，把模型的“自由输出”关进笼子里。本文从 Schema 生成、格式校验、多步骤工作流校验三个维度，手把手搭建一套生产级的校验体系。

二、Schema 生成：从 Python 函数到 JSON Schema

2.1 手动定义的问题

最朴素的做法是手写 JSON Schema：

weather_tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "City and state"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
}

手写有三个致命问题：

2.2 用 Pydantic 自动生成 Schema

使用 Pydantic 的 model_json_schema()，可以自动生成符合 JSON Schema 规范的定义：

from typing import Literal, Optional
from pydantic import BaseModel, Field

class GetWeatherParams(BaseModel):
"""Get the current weather for a location"""
location: str = Field(description="The city and state, e.g. San Francisco, CA")
unit: Optional[Literal["celsius", "fahrenheit"]] = Field(
default="celsius",
description="Temperature unit"
)

# 自动生成 Schema
schema = GetWeatherParams.model_json_schema()
print(schema)

输出（精简后）：

{
"type": "object",
"properties": {
"location": {"type": "string", "description": "The city and state…"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}

把 Pydantic 模型包装成 OpenAI Function Calling 格式：

def model_to_openai_tool(model_cls: type[BaseModel]) –> dict:
"""Convert a Pydantic BaseModel to OpenAI function schema"""
schema = model_cls.model_json_schema()
return {
"type": "function",
"function": {
"name": model_cls.__name__.replace("Params", "").lower(),
"description": model_cls.__doc__ or "",
"parameters": schema
}
}

# 使用
tool_def = model_to_openai_tool(GetWeatherParams)

2.3 从普通函数推断 Schema

如果你的函数已经写好了，用 Python 的 inspect 模块做自动转换：

import inspect
from typing import get_type_hints

def func_to_schema(func: callable) –> dict:
"""从函数签名生成 JSON Schema"""
sig = inspect.signature(func)
hints = get_type_hints(func)

properties = {}
required = []

for name, param in sig.parameters.items():
param_type = hints.get(name, str)
# 类型映射（简化版）
type_map = {
int: "integer",
float: "number",
str: "string",
bool: "boolean",
list: "array",
dict: "object"
}
properties[name] = {
"type": type_map.get(param_type, "string"),
"description": f"参数: {name}"
}
if param.default == inspect.Parameter.empty:
required.append(name)

return {
"type": "object",
"properties": properties,
"required": required
}

三、核心校验：让模型输出“老实”一点

3.1 最基础的校验：类型与必填

模型返回的 tool_calls 数组中，每个元素包含 function.name 和 function.arguments（注意是字符串）。你需要做的校验：

import json
from typing import Dict, Any, List, Optional

class SchemaValidator:
"""工具调用 Schema 校验器"""

def __init__(self, tool_schemas: List[Dict[str, Any]]):
# 建立 name -> schema 的索引
self.schema_index = {
t["function"]["name"]: t["function"]["parameters"]
for t in tool_schemas
}

def validate_tool_call(self, tool_call: Dict[str, Any]) –> Dict[str, Any]:
"""
校验单个工具调用
返回: {"valid": bool, "errors": List[str], "parsed_args": dict}
"""
function = tool_call.get("function", {})
name = function.get("name", "")
args_str = function.get("arguments", "{}")

errors = []

# 1. 检查工具是否存在
if name not in self.schema_index:
return {"valid": False, "errors": [f"未知工具: {name}"], "parsed_args": {}}

# 2. 解析 JSON
try:
args = json.loads(args_str) if args_str else {}
except json.JSONDecodeError as e:
return {"valid": False, "errors": [f"arguments 不是合法 JSON: {e}"], "parsed_args": {}}

schema = self.schema_index[name]

# 3. 检查必填字段
required = schema.get("required", [])
for field in required:
if field not in args:
errors.append(f"缺少必填字段: {field}")

# 4. 检查字段类型（基础版）
properties = schema.get("properties", {})
for field, value in args.items():
if field in properties:
expected_type = properties[field].get("type")
if expected_type == "integer" and not isinstance(value, int):
errors.append(f"字段 '{field}' 应为 integer, 实际为 {type(value).__name__}")
elif expected_type == "string" and not isinstance(value, str):
errors.append(f"字段 '{field}' 应为 string, 实际为 {type(value).__name__}")

return {
"valid": len(errors) == 0,
"errors": errors,
"parsed_args": args
}

3.2 进阶校验：enum 与 additionalProperties

生产环境中，你还需要校验更多规则：

def deep_validate_schema(schema: dict, path: str = "") –> List[str]:
"""
深度校验 Schema 本身的规范性
检查: required 字段必须在 properties 中存在,
enum 值必须匹配声明的类型,
additionalProperties 必须为 false 或合法的类型 schema
"""
errors = []
properties = schema.get("properties", {})
required = schema.get("required", [])

# 规则1: required 中的字段必须在 properties 中存在
for req in required:
if req not in properties:
errors.append(f"{path}: required key '{req}' not found in properties")

# 规则2: 每个 property 必须有 type
for key, prop in properties.items():
prop_path = f"{path}.{key}"
if "type" not in prop:
errors.append(f"{prop_path}: 缺少 'type' 字段")
continue

prop_type = prop.get("type")

# 规则3: enum 值必须匹配类型
if "enum" in prop:
enum_vals = prop.get("enum", [])
for val in enum_vals:
if prop_type == "integer" and not isinstance(val, int):
errors.append(f"{prop_path}: enum 值 {val} 不是 integer")
elif prop_type == "string" and not isinstance(val, str):
errors.append(f"{prop_path}: enum 值 {val} 不是 string")

# 规则4: additionalProperties 若存在，必须是 false 或 type schema
if "additionalProperties" in prop:
additional = prop["additionalProperties"]
if additional is not False and not isinstance(additional, dict):
errors.append(f"{prop_path}: additionalProperties 应为 false 或 type schema")

# 递归校验嵌套 object
if prop_type == "object" and "properties" in prop:
errors.extend(deep_validate_schema(prop, prop_path))

return errors

3.3 多步骤工作流的校验陷阱

在多步推理场景中，模型可能先调用工具，再返回最终结果。一个常见的陷阱是：模型调用了校验工具，但校验失败了，模型仍然在最终响应中声称“成功”。

解决方案：使用 ValidationTracker 追踪校验工具的执行结果。

class ValidationTracker:
"""
追踪校验工具的调用结果
关键洞察：不仅要追踪"是否被调用"，还要追踪"是否通过"
"""

def __init__(self, required_validations: List[str]):
self.tracking = {
name: {"called": False, "passed": False, "errors": []}
for name in required_validations
}

def record_call(self, validation_name: str, result: dict):
"""记录校验工具的执行结果"""
if validation_name not in self.tracking:
return

self.tracking[validation_name]["called"] = True
# 校验通过的条件：返回了 valid: True
is_valid = isinstance(result, dict) and result.get("valid") is True
self.tracking[validation_name]["passed"] = is_valid

if not is_valid and isinstance(result, dict):
if "errors" in result:
self.tracking[validation_name]["errors"].extend(result["errors"])

def all_passed(self) –> bool:
"""所有必需的校验都通过了"""
return all(
info["called"] and info["passed"]
for info in self.tracking.values()
)

def get_final_response(self, llm_response: dict) –> dict:
"""
在最终响应中检查校验是否全部通过
如果校验未通过，改写响应为错误状态
"""
if not self.all_passed():
return {
"success": False,
"error": "Validation checks failed: " + json.dumps(self.tracking)
}
return llm_response

使用示例（集成到 Agent 循环中）：

# 在 Agent 循环中
tracker = ValidationTracker(required_validations=["validate_sql", "validate_params"])

# … 模型返回 tool_calls
for tc in tool_calls:
if tc["function"]["name"] == "validate_sql":
result = execute_validation(tc)
tracker.record_call("validate_sql", result)

# … 模型返回最终响应
if tracker.all_passed():
return final_response
else:
return {"success": False, "error": "Validation failed"}

四、完整运行示例

from pydantic import BaseModel, Field
from typing import Literal, Optional
import json

# 1. 定义工具参数模型
class GetWeatherParams(BaseModel):
location: str = Field(description="City and state")
unit: Optional[Literal["celsius", "fahrenheit"]] = "celsius"

class CalculateParams(BaseModel):
expression: str = Field(description="Math expression to evaluate")

# 2. 生成工具定义
def build_tool_definitions():
return [
{
"type": "function",
"function": {
"name": "get_weather",
"description": GetWeatherParams.__doc__,
"parameters": GetWeatherParams.model_json_schema()
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": CalculateParams.__doc__,
"parameters": CalculateParams.model_json_schema()
}
}
]

# 3. 工具注册与执行
TOOL_FUNCTIONS = {
"get_weather": lambda location, unit="celsius": f"Weather in {location}: 22°{unit}",
"calculate": lambda expression: f"{expression} = {eval(expression)}"
}

def execute_tool(tool_call: dict) –> str:
func_name = tool_call["function"]["name"]
args = json.loads(tool_call["function"]["arguments"])
return TOOL_FUNCTIONS[func_name](**args)

# 4. 校验器
validator = SchemaValidator(build_tool_definitions())

# 5. 模拟模型返回
mock_tool_call = {
"function": {
"name": "get_weather",
"arguments": '{"location": "San Francisco", "unit": "celsius"}'
}
}

result = validator.validate_tool_call(mock_tool_call)
print(f"Valid: {result['valid']}")
print(f"Errors: {result['errors']}")
print(f"Parsed: {result['parsed_args']}")

# 执行
if result["valid"]:
output = execute_tool(mock_tool_call)
print(f"Result: {output}")

五、总结

工程化的核心要点：

一个关键认知： 约束 LLM 的输出格式只是“语法”层面的工作，它不会让模型突然变聪明。Schema 校验解决的是“输出是否合法”，但“输出是否合理”仍然依赖于正确的 Prompt 设计和 Few-shot 示例。校验的价值在于：当模型胡言乱语时，你能快速定位问题出在哪一层，而不是面对一个黑盒发呆。

有了这套校验体系，你的 Agent 从“玩具”到“工程”才迈出了关键一步。