Demo 与生产的差距 Function Calling 的 demo 很简单：定义一个函数，LLM 返回参数，调用函数，返回结果。但在生产中，你需要处理：
LLM 返回了不存在的函数 参数类型不对、必填字段缺失 函数调用超时 并行调用之间的依赖冲突 用户通过参数注入恶意命令 调用链过深导致上下文爆炸 Schema 设计：函数定义是契约 好的 Schema 长什么样 # ❌ 糟糕的 Schema: 模糊、无约束 BAD_SCHEMA = { "name": "search", "description": "搜索东西", "parameters": { "type": "object", "properties": { "query": {"type": "string"} } } } # ✅ 好的 Schema: 精确、有约束、有枚举 GOOD_SCHEMA = { "name": "search_product", "description": "在商品库中搜索商品。当用户询问有没有某种商品、价格、库存时使用此函数。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，从用户意图中提取，去除无意义词" }, "category": { "type": "string", "enum": ["electronics", "clothing", "food", "books", "other"], "description": "商品类别" }, "max_price": { "type": "number", "minimum": 0, "description": "用户能接受的最高价格(元)" }, "sort_by": { "type": "string", "enum": ["price_asc", "price_desc", "relevance", "sales"], "default": "relevance" } }, "required": ["query"], "additionalProperties": False } } Schema 设计原则 原则 说明 示例 名字即文档 函数名要自解释 search_product > search description 要写触发条件 不是描述函数做什么，而是何时用 “当用户询问商品价格时使用” 枚举优于自由文本 能枚举就不要用 string category 用 enum 设默认值 减少LLM猜测 sort_by 默认 relevance 禁止额外属性 additionalProperties: false 防止 LLM 编造参数 范围约束 用 min/max 限制数值 max_price minimum: 0 参数验证：不要信任 LLM 的输出 import json from pydantic import BaseModel, Field, ValidationError from typing import Optional, Literal # 用 Pydantic 做二次验证 class SearchProductParams(BaseModel): query: str = Field(..., min_length=1, max_length=100) category: Optional[Literal["electronics", "clothing", "food", "books", "other"]] = None max_price: Optional[float] = Field(None, ge=0, le=1000000) sort_by: Literal["price_asc", "price_desc", "relevance", "sales"] = "relevance" class FunctionCallValidator: def __init__(self): self.schemas = { "search_product": SearchProductParams, } def validate(self, function_name: str, arguments: dict) -> tuple[bool, any]: schema_cls = self.schemas.get(function_name) if not schema_cls: return False, {"error": f"Unknown function: {function_name}"} try: validated = schema_cls(**arguments) return True, validated except ValidationError as e: return False, {"error": e.errors()} 处理 LLM 返回的常见问题 class RobustFunctionExecutor: async def execute(self, llm_response: dict) -> dict: tool_calls = llm_response.get("tool_calls", []) results = [] for call in tool_calls: name = call["function"]["name"] # 1. 函数是否存在? if name not in self.registry: results.append(self._error_result(call, f"未知函数: {name}")) continue # 2. 参数解析 try: args = json.loads(call["function"]["arguments"]) except json.JSONDecodeError: # LLM 返回了非法 JSON，尝试修复 args = self._repair_json(call["function"]["arguments"]) if args is None: results.append(self._error_result(call, "参数 JSON 解析失败")) continue # 3. 参数验证 ok, validated = self.validator.validate(name, args) if not ok: results.append(self._error_result(call, str(validated))) continue # 4. 执行(带超时) try: result = await asyncio.wait_for( self.registry[name](**validated.dict()), timeout=10 ) results.append({"tool_call_id": call["id"], "result": result}) except asyncio.TimeoutError: results.append(self._error_result(call, "函数执行超时")) except Exception as e: results.append(self._error_result(call, f"执行错误: {str(e)}")) return results def _repair_json(self, broken: str) -> dict | None: """尝试修复 LLM 输出的破损 JSON""" # 去除尾部逗号 fixed = broken.rstrip().rstrip(",") # 补全括号 open_braces = fixed.count("{") - fixed.count("}") open_brackets = fixed.count("[") - fixed.count("]") fixed += "}" * open_braces + "]" * open_brackets try: return json.loads(fixed) except: return None 错误恢复：让 LLM 从错误中学习 async def function_call_loop(messages: list, max_rounds: int = 5): """多轮函数调用循环，带错误恢复""" for round_num in range(max_rounds): response = await client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=tool_definitions, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content # LLM 认为完成了 # 执行所有工具调用 for tool_call in msg.tool_calls: success, result = await executor.execute_single(tool_call) # 把结果（包括错误）反馈给 LLM messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) if not success: # 在错误信息中给出修复建议 messages.append({ "role": "system", "content": f"函数 {tool_call.function.name} 调用失败: {result['error']}。请修正参数后重试，或换一种方式回答。" }) return "达到最大调用轮数，请简化请求。" 并行调用与依赖管理 import asyncio from typing import Dict, List, Set class ParallelCallOrchestrator: def __init__(self): # 声明函数间的依赖关系 self.dependencies = { "get_order_detail": [], # 无依赖 "check_inventory": [], # 无依赖 "calculate_shipping": ["check_inventory"], # 依赖库存检查 "create_order": ["get_order_detail", "check_inventory"], } async def execute_parallel(self, tool_calls: list) -> list: # 构建依赖图 independent = [] dependent = {} for call in tool_calls: name = call["function"]["name"] deps = self.dependencies.get(name, []) if not deps: independent.append(call) else: dependent[call["id"]] = {"call": call, "deps": set(deps)} # 先执行无依赖的 results = {} tasks = [self._execute_and_store(call, results) for call in independent] await asyncio.gather(*tasks, return_exceptions=True) # 再执行有依赖的（拓扑排序） while dependent: ready = [ cid for cid, info in dependent.items() if info["deps"].issubset(set(results.keys())) ] if not ready: # 有循环依赖，强制执行剩余的 ready = list(dependent.keys()) tasks = [] for cid in ready: tasks.append(self._execute_and_store(dependent[cid]["call"], results)) del dependent[cid] await asyncio.gather(*tasks, return_exceptions=True) return list(results.values()) 安全沙箱：永远不要直接执行 LLM 生成的代码 import subprocess import tempfile import os class CodeExecutionSandbox: """安全执行 LLM 生成的代码""" def __init__(self): self.allowed_modules = {"math", "statistics", "json", "re"} self.timeout = 5 self.memory_limit = "256m" async def execute_python(self, code: str) -> dict: # 1. 静态检查 violations = self._static_check(code) if violations: return {"error": "安全检查失败", "violations": violations} # 2. 在 Docker 容器中执行 with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: f.write(code) f.flush() try: result = subprocess.run( ["docker", "run", "--rm", "--memory", self.memory_limit, "--cpus", "0.5", "--network", "none", # 无网络 "--read-only", # 只读文件系统 "--tmpfs", "/tmp:size=10m", # 临时目录 "-v", f"{f.name}:/app/code.py:ro", "python:3.12-slim", "python", "/app/code.py"], capture_output=True, timeout=self.timeout, text=True ) return { "stdout": result.stdout[:5000], # 截断 "stderr": result.stderr[:2000], "exit_code": result.returncode } except subprocess.TimeoutExpired: return {"error": "执行超时"} finally: os.unlink(f.name) def _static_check(self, code: str) -> list: violations = [] # 检查 import for line in code.split("\n"): if "import" in line: module = line.split("import")[-1].strip().split(".")[0] if module not in self.allowed_modules: violations.append(f"禁止导入模块: {module}") # 检查危险函数 dangerous = ["open(", "exec(", "eval(", "os.system", "subprocess", "__import__"] for d in dangerous: if d in code: violations.append(f"禁止使用: {d}") return violations 审计日志：记录每次调用 import logging from datetime import datetime logger = logging.getLogger("function_caller") class AuditLogger: def log_call(self, function_name: str, arguments: dict, result: dict, duration_ms: float, success: bool): logger.info(json.dumps({ "timestamp": datetime.utcnow().isoformat(), "function": function_name, "arguments": self._sanitize(arguments), # 脱敏 "result_size": len(str(result)), "duration_ms": duration_ms, "success": success, "trace_id": self._get_trace_id(), }, ensure_ascii=False)) def _sanitize(self, args: dict) -> dict: """脱敏处理""" sensitive_keys = {"password", "token", "api_key", "credit_card"} sanitized = {} for k, v in args.items(): if k.lower() in sensitive_keys: sanitized[k] = "***REDACTED***" else: sanitized[k] = v return sanitized 性能优化 函数定义缓存 # 工具定义不要每次请求都重新构建 from functools import lru_cache @lru_cache(maxsize=1) def get_tool_definitions(): return [ {"type": "function", "function": schema} for schema in load_all_schemas() ] 函数调用结果缓存 import hashlib from datetime import timedelta class ResultCache: def __init__(self, redis_client): self.redis = redis_client self.ttl = timedelta(minutes=10) async def get_or_execute(self, func_name: str, args: dict, executor): # 对幂等函数做缓存 cache_key = self._make_key(func_name, args) cached = await self.redis.get(cache_key) if cached: return json.loads(cached) result = await executor(func_name, args) await self.redis.setex( cache_key, int(self.ttl.total_seconds()), json.dumps(result, ensure_ascii=False) ) return result def _make_key(self, name: str, args: dict) -> str: arg_hash = hashlib.md5(json.dumps(args, sort_keys=True).encode()).hexdigest() return f"fc:{name}:{arg_hash}" 总结 Function Calling 在生产中可靠运行的关键：严格的 Schema 设计是契约，二次验证是防线，错误恢复是韧性，并行管理是效率，安全沙箱是底线，审计日志是追溯。把每个 LLM 返回的函数调用都当作不可信输入来处理，就能避免大部分生产事故。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
