Function Calling / Tool Use

import jsonfrom typing import Dict, Any, Listfrom dataclasses import dataclassfrom enum import Enum class ProtocolType(Enum):    OPENAI = "openai"    OLLAMA = "ollama" @dataclassclass ToolDefinition:    name: str    description: str    parameters: Dict[str, Any] class SchemaValidator:    """JSON Schema 参数校验器"""    def __init__(self, schema: Dict[str, Any]):        self.schema = schema        self.required = schema.get("required", [])        self.properties = schema.get("properties", {})     def validate(self, params: Dict[str, Any]) -> tuple:        for req in self.required:            if req not in params:                return False, f"缺少必填参数: {req}"        for key, value in params.items():            if key in self.properties:                expected_type = self.properties[key].get("type")                if expected_type == "number" and not isinstance(value, (int, float)):                    return False, f"参数 {key} 类型错误: 期望number"                if expected_type == "string" and not isinstance(value, str):                    return False, f"参数 {key} 类型错误: 期望string"                enum_vals = self.properties[key].get("enum")                if enum_vals and value not in enum_vals:                    return False, f"参数 {key} 值 {value} 不在允许范围"                max_val = self.properties[key].get("maximum")                if max_val and isinstance(value, (int, float)) and value > max_val:                    return False, f"参数 {key} 值 {value} 超过上限 {max_val}"        return True, "校验通过" class SecurityGateway:    """工具调用安全网关：Schema校验 + 参数白名单 + 审计日志"""    def __init__(self):        self.audit_log: List[Dict] = []        self.param_whitelist: Dict[str, List[str]] = {}     def register_whitelist(self, tool_name: str, allowed_params: List[str]):        self.param_whitelist[tool_name] = allowed_params     def check_call(self, tool_name, params, validator, user_role):        valid, msg = validator.validate(params)        if not valid:            self._log(tool_name, params, "rejected", msg)            return False, msg        if tool_name in self.param_whitelist:            for key in params:                if key not in self.param_whitelist[tool_name]:                    self._log(tool_name, params, "rejected", f"非白名单参数: {key}")                    return False, f"非白名单参数: {key}"        if user_role != "admin" and tool_name in ["delete_user", "transfer_money"]:            self._log(tool_name, params, "rejected", "权限不足")            return False, f"角色 {user_role} 无权调用 {tool_name}"        self._log(tool_name, params, "approved", "校验通过")        return True, "调用批准"     def _log(self, tool_name, params, status, reason):        self.audit_log.append({"tool": tool_name, "params": params, "status": status, "reason": reason}) class OpenAIAdapter:    """OpenAI协议适配器"""    def build_tools_field(self, tools):        return [{"type": "function", "function": {"name": t.name, "description": t.description, "parameters": t.parameters}} for t in tools]     def parse_response(self, response):        return response.get("choices", [{}])[0].get("message", {}).get("tool_calls", []) class OllamaAdapter:    """Ollama协议适配器：兼容OpenAI格式并扩展本地能力"""    def build_tools_field(self, tools):        return [{"type": "function", "function": {"name": t.name, "description": t.description, "parameters": t.parameters}} for t in tools]     def parse_response(self, response):        tool_calls = response.get("message", {}).get("tool_calls", [])        if not tool_calls:            tool_calls = response.get("tool_calls", [])        return tool_calls transfer_schema = {"type": "object", "properties": {"amount": {"type": "number", "maximum": 1000}, "recipient": {"type": "string", "enum": ["alice", "bob", "charlie"]}}, "required": ["amount", "recipient"]}transfer_tool = ToolDefinition("transfer_money", "转账工具", transfer_schema)validator = SchemaValidator(transfer_schema)gateway = SecurityGateway()gateway.register_whitelist("transfer_money", ["amount", "recipient"])openai_adapter = OpenAIAdapter()ollama_adapter = OllamaAdapter()print("OpenAI tools字段:", json.dumps(openai_adapter.build_tools_field([transfer_tool]), ensure_ascii=False)[:80])print("Ollama tools字段:", json.dumps(ollama_adapter.build_tools_field([transfer_tool]), ensure_ascii=False)[:80])valid, msg = gateway.check_call("transfer_money", {"amount": 100, "recipient": "alice"}, validator, "user")print(f"正常调用: {valid}, {msg}")valid, msg = gateway.check_call("transfer_money", {"amount": 10000, "recipient": "alice"}, validator, "user")print(f"超额调用: {valid}, {msg}")valid, msg = gateway.check_call("transfer_money", {"amount": 100, "recipient": "hacker"}, validator, "user")print(f"非法收款人: {valid}, {msg}")valid, msg = gateway.check_call("delete_user", {"user_id": 123}, SchemaValidator({"type": "object", "properties": {"user_id": {"type": "number"}}, "required": ["user_id"]}), "user")print(f"越权调用: {valid}, {msg}")print(f"审计日志: {len(gateway.audit_log)}条记录")