深入解析Function Call Skill的实现原理与最佳实践

1次阅读

共计 2548 个字符，预计需要花费 7 分钟才能阅读完成。

Function Call Skill（函数调用技能）本质上是一种封装特定功能逻辑的代码单元，它允许开发者通过标准化接口调用复杂操作。在现代分布式架构中，它常作为微服务间的通信基础组件，或 AI 技能调用的核心机制（如 ChatGPT 插件系统）。

定义：通过预定义接口触发远程或本地功能执行的标准化协议
典型场景 ：
微服务间 API 调用
服务器 less 函数触发
智能对话系统的技能调度

+-------------+       +-----------------+       +---------------+
|   Caller    | ----> |  Parameter      | ----> |   Function    |
| (Client)    |       |  Marshalling    |       |  Execution    |
+-------------+       +-----------------+       +---------------+
                          ^    |                      |
                          |    v                      v
                    +-----+------+             +---------------+
                    | Transport  | <-----------| Result        |
                    | (HTTP/RPC) |             | Serialization |
                    +------------+             +---------------+

参数传递 ：采用 Protocol Buffers/JSON 等序列化协议，关键字段包括：
函数标识符
输入参数列表
调用元数据（超时设置、重试策略等）
返回值处理 ：
结构化错误码（如 gRPC status codes）
多级返回结果（主数据 + 元数据）

import asyncio
from typing import Any, Dict

class FunctionCaller:
    def __init__(self, max_retries=3):
        self.retry_policy = max_retries

    async def call_skill(self, func_name: str, params: Dict[str, Any]) -> Any:
        """
        :param func_name: 注册的函数标识符
        :param params: 参数字典
        :raises: RuntimeError 当重试次数耗尽时
        """
        last_error = None
        for attempt in range(self.retry_policy):
            try:
                # 模拟远程调用
                result = await self._remote_execute(func_name, params)
                return result
            except Exception as e:
                last_error = e
                await asyncio.sleep(2 ** attempt)  # 指数退避
        raise RuntimeError(f"Call failed after {self.retry_policy} attempts") from last_error

    async def _remote_execute(self, func_name: str, params: Dict[str, Any]) -> Any:
        # 实际生产环境替换为 gRPC/HTTP 调用
        await asyncio.sleep(0.1)
        return {"result": f"processed {func_name} with {params}"}

class FunctionCaller {constructor(timeout = 5000) {this.timeoutMs = timeout;}

  async callSkill(funcName, params) {const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), this.timeoutMs);

    try {
      const response = await fetch('/api/function-call', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({funcName, params}),
        signal: controller.signal
      });

      if (!response.ok) {throw new Error(`HTTP error! status: ${response.status}`);
      }

      return await response.json();} catch (error) {if (error.name === 'AbortError') {console.error(`Call timed out after ${this.timeoutMs}ms`);
      }
      throw error;
    } finally {clearTimeout(timeoutId);
    }
  }
}