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 要写触发条件不是描述函数做什么，而是何时用“当用户询问商品价格时使用”
枚举优于自由文本能枚举就不要用 stringcategory 用 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论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。