手写 Function Calling 引擎：从 JSON Schema 解析到工具路由与流式执行

┌─────────────────────────────────────┐ │ 应用层（用户代码） │ │ query → 工具选择 → 执行 → 返回结果 │ ├─────────────────────────────────────┤ │ 执行引擎层 │ │ 流式执行 / 并行执行 / 异常回退 │ ├─────────────────────────────────────┤ │ 路由决策器 │ │ 语义匹配 / 关键词匹配 / 自动路由 │ ├─────────────────────────────────────┤ │ Schema 解析与校验层 │ │ JSON Schema → Pydantic 模型转换 │ ├─────────────────────────────────────┤ │ 工具注册与发现层 │ │ 静态注册 / 动态扫描 / 远程发现 │ └─────────────────────────────────────┘

from typing import Any from enum import Enum from pydantic import BaseModel, Field class ToolParameter(BaseModel): """工具参数的描述，等价于 JSON Schema 的单个属性""" name: str = Field(description="参数名称") type: str = Field(default="string", description="参数类型: string/number/integer/boolean/array/object") description: str = Field(default="", description="参数描述") required: bool = Field(default=False, description="是否必填") enum: list[str] | None = Field(default=None, description="枚举值（如有）") class ToolSchema(BaseModel): """完整的工具 Schema，符合 OpenAI tools 规范""" name: str = Field(description="工具名称，唯一标识") description: str = Field(description="工具描述，供模型理解用途") parameters: dict[str, Any] = Field(default={}, description="JSON Schema 格式的参数定义") def to_openai_format(self) -> dict: """转换为 OpenAI API 的 tools 格式""" return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters, } }

import inspect from typing import Callable, Any class ToolRegister: """工具注册器，管理工具的全生命周期""" def __init__(self): self._tools: dict[str, ToolSchema] = {} self._handlers: dict[str, Callable] = {} def register( self, name: str, description: str = "", parameters: dict | None = None, ): """装饰器方式注册工具""" def decorator(func: Callable): schema = ToolSchema( name=name, description=description or func.__doc__ or "", parameters=parameters or self._infer_params(func), ) self._tools[name] = schema self._handlers[name] = func return func return decorator def _infer_params(self, func: Callable) -> dict: """从函数签名自动推导参数 Schema（简易版本）""" sig = inspect.signature(func) properties = {} required = [] for name, param in sig.parameters.items(): typ = str if param.annotation is inspect.Parameter.empty else param.annotation json_type = self._pytype_to_json(typ) properties[name] = { "type": json_type, "description": f"参数 {name}" } if param.default is inspect.Parameter.empty: required.append(name) return { "type": "object", "properties": properties, "required": required, } def _pytype_to_json(self, typ) -> str: mapping = { str: "string", int: "integer", float: "number", bool: "boolean", list: "array", dict: "object", } return mapping.get(typ, "string") def get_schema(self, name: str) -> ToolSchema | None: return self._tools.get(name) def get_handler(self, name: str) -> Callable | None: return self._handlers.get(name) def list_tools(self) -> list[ToolSchema]: return list(self._tools.values()) def to_openai_tools(self) -> list[dict]: """返回 OpenAI ChatCompletion 的 tools 参数""" return [t.to_openai_format() for t in self._tools.values()]

import json import httpx register = ToolRegister() @register.register( name="get_current_time", description="获取当前时间", parameters={ "type": "object", "properties": { "timezone": { "type": "string", "description": "时区，如 Asia/Shanghai, America/New_York", "enum": ["Asia/Shanghai", "America/New_York", "Europe/London", "Asia/Tokyo"] } }, "required": ["timezone"], } ) def get_current_time(timezone: str = "Asia/Shanghai") -> str: """获取指定时区的当前时间""" from datetime import datetime import pytz tz = pytz.timezone(timezone) now = datetime.now(tz) return now.strftime("%Y-%m-%d %H:%M:%S %Z") @register.register( name="web_search", description="搜索互联网信息，返回前5条结果摘要", parameters={ "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词", }, "count": { "type": "integer", "description": "返回结果数量，默认5", } }, "required": ["query"], } ) async def web_search(query: str, count: int = 5) -> str: """使用 httpx 执行简单的搜索（示例使用模拟实现）""" # 实际场景可集成 SerpAPI / Bing Search 等 return f"[搜索结果] 关于 '{query}' 获得 {count} 条结果" @register.register( name="calculator", description="执行数学计算，支持四则运算", parameters={ "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，如 '2 + 3 * 4'", } }, "required": ["expression"], } ) def calculator(expression: str) -> str: """安全执行数学表达式""" import ast import operator as op allowed_ops = { ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg, ast.Mod: op.mod, ast.FloorDiv: op.floordiv, } def eval_expr(node): if isinstance(node, ast.Expression): return eval_expr(node.body) if isinstance(node, ast.Constant): return node.n if isinstance(node.n, (int, float)) else node.value if isinstance(node, ast.BinOp): return allowed_ops[type(node.op)](eval_expr(node.left), eval_expr(node.right)) if isinstance(node, ast.UnaryOp): return allowed_ops[type(node.op)](eval_expr(node.operand)) raise ValueError(f"不支持的表达式类型: {type(node)}") try: tree = ast.parse(expression, mode='eval') result = eval_expr(tree) return str(result) except Exception as e: return f"计算错误: {e}"

from pydantic import BaseModel, create_model, ValidationError from typing import Any class SchemaValidator: """将 JSON Schema 转换为 Pydantic 模型并执行校验""" @staticmethod def _json_schema_to_pydantic(name: str, schema: dict) -> type[BaseModel]: """将 JSON Schema 转换为 Pydantic 模型类""" properties = schema.get("properties", {}) required = set(schema.get("required", [])) fields = {} for prop_name, prop_schema in properties.items(): field_type = SchemaValidator._resolve_type(prop_schema) description = prop_schema.get("description", "") if prop_name in required: fields[prop_name] = (field_type, Field(..., description=description)) else: default = prop_schema.get("default") fields[prop_name] = (Optional[field_type], Field(default=default, description=description)) return create_model(name, **fields) @staticmethod def _resolve_type(schema: dict) -> type: js_type = schema.get("type", "string") enum_vals = schema.get("enum") if enum_vals: from typing import Literal return Literal[tuple(enum_vals)] type_map = { "string": str, "integer": int, "number": float, "boolean": bool, "array": list, "object": dict, } return type_map.get(js_type, Any) def validate(self, name: str, schema: dict, arguments: dict | str) -> dict: """ 校验参数： 1. 如果 arguments 是 JSON 字符串，先解析 2. 转换成 Pydantic 模型校验 3. 返回校验后的合法参数 """ if isinstance(arguments, str): try: arguments = json.loads(arguments) except json.JSONDecodeError as e: raise ValueError(f"JSON 解析错误: {e}") model = self._json_schema_to_pydantic(name, schema) try: validated = model(**arguments) return validated.model_dump() except ValidationError as e: raise ValueError(f"参数校验失败: {e.errors()}")

from typing import Any import json class FunctionCallingEngine: """ Function Calling 核心执行引擎。 接收模型返回的 function_call 调用请求，查找并执行对应工具。 """ def __init__(self, register: ToolRegister): self.register = register self.validator = SchemaValidator() self.history: list[dict] = [] def execute(self, tool_name: str, arguments: dict | str) -> dict: """ 执行一个工具调用。 返回格式兼容 OpenAI 的 tool message。 """ schema = self.register.get_schema(tool_name) if not schema: raise ValueError(f"未知工具: {tool_name}") # 校验参数 validated_args = self.validator.validate( tool_name, schema.parameters, arguments, ) # 执行 handler = self.register.get_handler(tool_name) if handler is None: raise ValueError(f"工具 {tool_name} 没有注册处理器") result = handler(**validated_args) # 返回标准格式 response = { "role": "tool", "tool_call_id": f"call_{tool_name}", "name": tool_name, "content": str(result), } self.history.append(response) return response

from typing import Generator import asyncio class StreamingFunctionEngine(FunctionCallingEngine): """ 流式执行引擎： 1. 接收批量 tool_calls 2. 通过 Generator 逐步 yield 每个工具的执行结果 3. 支持并行执行 """ def stream_execute( self, tool_calls: list[dict], parallel: bool = False, ) -> Generator[dict, None, list[dict]]: """ 流式执行多个工具调用。 Args: tool_calls: [{"name": "...", "arguments": {...}}, ...] parallel: 是否启用并行执行（只对 async 工具生效） Yields: 每个工具的执行进展 Returns: 所有执行结果的列表 """ results = [] total = len(tool_calls) for i, call in enumerate(tool_calls): name = call.get("name", call.get("function", {}).get("name", "")) args = call.get("arguments", call.get("function", {}).get("arguments", {})) # Yield 进度 yield { "type": "tool_start", "tool_name": name, "index": i, "total": total, } try: result = self.execute(name, args) results.append(result) yield { "type": "tool_end", "tool_name": name, "index": i, "content": result["content"][:50] + "..." if len(result["content"]) > 50 else result["content"], } except Exception as e: err_msg = f"工具 {name} 执行失败: {e}" results.append({"role": "tool", "name": name, "content": err_msg}) yield { "type": "tool_error", "tool_name": name, "index": i, "error": str(e), } yield { "type": "tool_complete", "total": total, "success": len([r for r in results if "error" not in r.get("content", "")]), } return results

class AsyncFunctionEngine(FunctionCallingEngine): """异步执行引擎，支持并行调用多个工具""" async def execute_async(self, tool_name: str, arguments: dict | str) -> dict: """异步执行单个工具""" schema = self.register.get_schema(tool_name) if not schema: raise ValueError(f"未知工具: {tool_name}") validated_args = self.validator.validate(tool_name, schema.parameters, arguments) handler = self.register.get_handler(tool_name) if handler is None: raise ValueError(f"工具 {tool_name} 未注册 handler") # 判断 handler 是否是 async if asyncio.iscoroutinefunction(handler): result = await handler(**validated_args) else: result = handler(**validated_args) response = { "role": "tool", "tool_call_id": f"call_{tool_name}", "name": tool_name, "content": str(result), } self.history.append(response) return response async def execute_parallel(self, tool_calls: list[dict]) -> list[dict]: """并行执行多个工具（同时执行，全部完成后返回）""" tasks = [] for call in tool_calls: name = call.get("name", call.get("function", {}).get("name", "")) args = call.get("arguments", call.get("function", {}).get("arguments", {})) tasks.append(self.execute_async(name, args)) results = await asyncio.gather(*tasks, return_exceptions=True) final_results = [] for i, result in enumerate(tasks): if isinstance(result, Exception): final_results.append({ "role": "tool", "name": tool_calls[i].get("name"), "content": f"执行失败: {result}", }) else: final_results.append(result) return final_results

from dataclasses import dataclass, field @dataclass class ToolCallNode: """DAG 中的一个工具调用节点""" name: str arguments: dict = field(default_factory=dict) depends_on: list[str] = field(default_factory=list) fallback_name: str | None = None fallback_args: dict | None = None def to_dict(self) -> dict: return { "name": self.name, "arguments": self.arguments, "depends_on": self.depends_on, "fallback_name": self.fallback_name, } class DAGExecutor: """ DAG 编排执行器：按依赖关系拓扑排序， 依次执行工具，支持失败回退。 """ def __init__(self, engine: FunctionCallingEngine): self.engine = engine def execute_dag(self, nodes: list[ToolCallNode]) -> dict[str, Any]: """执行 DAG 编排的工具调用""" # 构建入度表和邻接表 indeg = {n.name: len(n.depends_on) for n in nodes} name_map = {n.name: n for n in nodes} dependents: dict[str, list[str]] = {} for n in nodes: for dep in n.depends_on: if dep not in dependents: dependents[dep] = [] dependents[dep].append(n.name) # 拓扑排序 + 执行 queue = [n.name for n in nodes if indeg[n.name] == 0] results: dict[str, Any] = {} while queue: current = queue.pop(0) node = name_map[current] # 注入依赖的结果作为参数 exec_args = dict(node.arguments) for dep in node.depends_on: if dep in results: exec_args[f"_{dep}_result"] = results[dep].get("content", "") try: result = self.engine.execute(current, exec_args) results[current] = result except Exception as e: # 尝试 fallback if node.fallback_name: try: fb_args = node.fallback_args or {} result = self.engine.execute(node.fallback_name, fb_args) results[current] = result except Exception as fb_e: results[current] = {"error": f"primary/fallback both failed: {e}, {fb_e}"} else: results[current] = {"error": str(e)} # 更新依赖关系 if current in dependents: for dep in dependents[current]: indeg[dep] -= 1 if indeg[dep] == 0: queue.append(dep) return results

# 搜索某个技术主题，然后进行数据计算 dag_nodes = [ ToolCallNode( name="web_search", arguments={"query": "2026年6月DeepSeek最新动态", "count": 3}, depends_on=[], ), ToolCallNode( name="calculator", arguments={"expression": "1024 * 768"}, depends_on=["web_search"], # 先搜索 fallback_name="web_search", fallback_args={"query": "DeepSeek 2026"}, ), ]

import time import pickle from pathlib import Path class ToolCheckpoint: """ 工具检查点：持久化工具执行状态。 支持：保存进度、恢复断点、查询状态。 """ def __init__(self, checkpoint_dir: str = "/tmp/tool_checkpoints"): self.checkpoint_dir = Path(checkpoint_dir) self.checkpoint_dir.mkdir(parents=True, exist_ok=True) def save_checkpoint(self, session_id: str, state: dict): """保存会话执行状态""" path = self.checkpoint_dir / f"{session_id}.ckpt" state["_timestamp"] = time.time() with open(path, "wb") as f: pickle.dump(state, f) def load_checkpoint(self, session_id: str) -> dict | None: """恢复会话执行状态""" path = self.checkpoint_dir / f"{session_id}.ckpt" if path.exists(): with open(path, "rb") as f: return pickle.load(f) return None def clear_checkpoint(self, session_id: str): """清理检查点（执行完成后）""" path = self.checkpoint_dir / f"{session_id}.ckpt" if path.exists(): path.unlink()

import httpx class HuaweiMaaSClient: """ 华为云 MaaS 推理服务客户端。 原生支持 Function Calling。 """ def __init__( self, endpoint: str, api_key: str, model: str = "DeepSeek-R1", ): self.endpoint = endpoint.rstrip("/") self.api_key = api_key self.model = model def chat_completion( self, messages: list[dict], tools: list[dict] | None = None, temperature: float = 0.7, max_tokens: int = 4096, ) -> dict: """ 调用华为云 MaaS 推理接口，支持 tools 参数。 """ url = f"{self.endpoint}/v1/chat/completions" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } if tools: payload["tools"] = tools with httpx.Client(timeout=60) as client: response = client.post(url, json=payload, headers=headers) response.raise_for_status() return response.json() def chat_with_tools( self, query: str, engine: FunctionCallingEngine, max_turns: int = 5, ) -> list[dict]: """ 完整的工具调用循环： 1. 用户提问 2. 模型决定是否调用工具 3. 引擎执行工具并返回结果 4. 模型基于结果生成最终回答 """ messages = [{"role": "user", "content": query}] tools_meta = engine.register.to_openai_tools() for turn in range(max_turns): # 调用 MaaS response = self.chat_completion(messages, tools=tools_meta if tools_meta else None) assistant_msg = response["choices"][0]["message"] # 检查模型是否调用了工具 if tool_calls := assistant_msg.get("tool_calls"): messages.append({ "role": "assistant", "content": assistant_msg.get("content") or "", "tool_calls": [ { "id": tc.get("id"), "type": "function", "function": tc["function"], } for tc in tool_calls ], }) # 执行每一个工具 for tc in tool_calls: tool_name = tc["function"]["name"] tool_args = tc["function"]["arguments"] result = engine.execute(tool_name, tool_args) messages.append(result) continue # 继续让模型基于工具结果回复 # 模型直接回复，不需要调用工具 final_content = assistant_msg.get("content", "") messages.append({ "role": "assistant", "content": final_content, }) return messages return messages

# 初始化引擎 engine = FunctionCallingEngine(register) # 配置华为云 MaaS 客户端 client = HuaweiMaaSClient( endpoint="https://maas-deepseek.cn-north-4.myhuaweicloud.com", api_key="your-api-key", model="DeepSeek-R1", ) # 执行一次完整对话 result = client.chat_with_tools( query="搜索一下最近Dify的更新动态，并计算DeepSeek V3与R1的参数量差异倍数", engine=engine, ) print(result[-1]["content"])

class SafeValidator(CompiledValidator): """安全增强型校验器，拦截危险参数""" DANGEROUS_PATTERNS = [ "__import__", "os.system", "subprocess", "exec(", "eval(", "open(", "rm -rf", "delet", "drop table", ] def validate(self, name: str, schema: dict, arguments: dict | str) -> dict: validated = super().validate(name, schema, arguments) # 检查所有字符串参数 for key, value in validated.items(): if isinstance(value, str): for pattern in self.DANGEROUS_PATTERNS: if pattern.lower() in value.lower(): raise ValueError(f"参数 '{key}' 包含危险内容: {pattern}") return validated

import time from collections import defaultdict class RateLimitedEngine(FunctionCallingEngine): """带频率限制的执行引擎""" def __init__(self, register: ToolRegister, max_calls: int = 10, window_seconds: int = 60): super().__init__(register) self.max_calls = max_calls self.window_seconds = window_seconds self.call_history: dict[str, list[float]] = defaultdict(list) def execute(self, tool_name: str, arguments: dict | str) -> dict: now = time.time() recent = self.call_history[tool_name] # 清理过期记录 recent[:] = [t for t in recent if now - t < self.window_seconds] if len(recent) >= self.max_calls: raise ValueError( f"工具 '{tool_name}' 调用频率超限：" f"最近 {self.window_seconds} 秒内已调用 {len(recent)} 次" ) recent.append(now) return super().execute(tool_name, arguments)

import threading class TimeoutEngine(FunctionCallingEngine): """带超时控制的执行引擎""" def __init__(self, register: ToolRegister, default_timeout: int = 30): super().__init__(register) self.default_timeout = default_timeout def execute(self, tool_name: str, arguments: dict | str, timeout: int | None = None) -> dict: timeout = timeout or self.default_timeout result_container = [] exception_container = [] def worker(): try: result_container.append(super().execute(tool_name, arguments)) except Exception as e: exception_container.append(e) thread = threading.Thread(target=worker, daemon=True) thread.start() thread.join(timeout=timeout) if thread.is_alive(): return { "role": "tool", "tool_call_id": f"call_{tool_name}", "name": tool_name, "content": f"执行超时：工具 '{tool_name}' 超过 {timeout} 秒未返回", } if exception_container: raise exception_container[0] return result_container[0]

from functools import lru_cache import hashlib class CachedEngine(FunctionCallingEngine): """带缓存的执行引擎""" def __init__(self, register: ToolRegister, cache_size: int = 128): super().__init__(register) self._cache = {} self.cache_size = cache_size def _cache_key(self, tool_name: str, arguments: dict) -> str: raw = f"{tool_name}:{json.dumps(arguments, sort_keys=True)}" return hashlib.md5(raw.encode()).hexdigest() def execute(self, tool_name: str, arguments: dict | str) -> dict: if isinstance(arguments, str): parsed_args = json.loads(arguments) if isinstance(arguments, str) else arguments else: parsed_args = arguments key = self._cache_key(tool_name, parsed_args) if key in self._cache: return self._cache[key] result = super().execute(tool_name, arguments) # 裁剪缓存大小 if len(self._cache) >= self.cache_size: self._cache.pop(next(iter(self._cache))) self._cache[key] = result return result

class CompiledValidator(SchemaValidator): """预编译 Schema 的校验器""" def __init__(self): super().__init__() self._model_cache: dict[str, type[BaseModel]] = {} def validate(self, name: str, schema: dict, arguments: dict | str) -> dict: # 缓存 Model 类 schema_key = json.dumps(schema, sort_keys=True) if schema_key not in self._model_cache: self._model_cache[schema_key] = self._json_schema_to_pydantic(name, schema) model = self._model_cache[schema_key] if isinstance(arguments, str): arguments = json.loads(arguments) try: validated = model(**arguments) return validated.model_dump() except ValidationError as e: raise ValueError(f"参数校验失败: {e.errors()}")

import asyncio # 1. 注册工具 tool_register = ToolRegister() @tool_register.register( name="get_news_headlines", description="获取指定领域的最新新闻标题列表", parameters={ "type": "object", "properties": { "topic": { "type": "string", "description": "新闻主题，如 AI / 科技 / 财经", "enum": ["AI", "科技", "财经", "医疗"], }, "date_range": { "type": "string", "description": "时间范围：today / this_week", } }, "required": ["topic"], } ) def get_news_headlines(topic: str, date_range: str = "today") -> str: """获取新闻标题（模拟实现）""" mock_news = { "AI": [ "DeepSeek R1 发布新版本，推理效率提升 40%", "OpenAI 更新 GPT-5 函数调用规范", "MCP 协议被多家云厂商采纳为标准", ], "科技": [ "华为昇腾 910C 芯片量产加速", "RISC-V 生态取得关键突破", ], } headlines = mock_news.get(topic, [f"暂无 {topic} 领域新闻"]) return "\n".join(f"- {h}" for h in headlines) @tool_register.register( name="summarize_text", description="对文本进行摘要总结", parameters={ "type": "object", "properties": { "text": {"type": "string", "description": "待总结的文本"}, "max_length": {"type": "integer", "description": "摘要最大字数"}, }, "required": ["text"], } ) def summarize_text(text: str, max_length: int = 100) -> str: """简易文本摘要（截取前 max_length 字）""" return text[:max_length] + ("..." if len(text) > max_length else "") # 2. 初始化引擎 engine = FunctionCallingEngine(tool_register) # 3. 批量执行测试 print("=== 工具列表 ===") for tool in tool_register.list_tools(): print(f" {tool.name}: {tool.description}") print("\n=== 执行测试 1: 获取 AI 新闻并摘要 ===") result1 = engine.execute( "get_news_headlines", {"topic": "AI", "date_range": "today"}, ) print(result1["content"]) print("\n=== 执行测试 2: 计算器 ===") result2 = engine.execute( "calculator", {"expression": "(2048 + 512) * 8"}, ) print(result2["content"])