一、从“对话”到“行动”：Function Calling的本质

当用户对AI说“帮我看看订单#12345的物流”时，我们希望的不只是一段描述性的回复，而是系统真正去查询订单系统并返回准确信息。这正是Function Calling要解决的核心问题——让大模型从“语言生成器”变成“行动协调器”。

Function Calling的本质，可以理解为一个结构化的调用协议：开发者将系统能力（查询订单、调价、查库存等）以工具定义的形式暴露给模型，模型根据用户意图决定是否调用工具，并生成符合约定的结构化参数。整个过程是模型表达“希望执行什么操作”，应用程序决定该操作能否执行以及如何执行。

一个完整的Function Calling过程包含五个阶段：

用户目标 → 模型理解任务 → 选择工具并生成参数 → 应用校验并执行工具 → 模型观察执行结果 → 继续调用或生成回答

模型负责判断需要调用什么能力，应用程序负责校验请求、执行工具并返回真实结果。这已经构成了一个最小的Agent执行循环。

二、工具定义：决定调用质量的关键

模型能否正确选择工具，很大程度上取决于工具定义是否清晰。一个完整的工具定义通常包括：工具名称（准确表达动作）、功能描述（说明使用边界）、参数结构（JSON Schema约束）。

2.1 工具定义的标准格式

以OpenAI的Function Calling格式为例：

# tools/definitions.py
from typing import List, Dict, Any

class ToolDefinition:
    """工具定义的标准化格式"""
    
    @staticmethod
    def get_weather_tool() -> Dict[str, Any]:
        """天气查询工具定义"""
        return {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "查询指定城市的实时天气信息。当用户询问天气时使用。",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "城市名称，如'北京'、'上海'"
                        },
                        "unit": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "温度单位，默认为摄氏"
                        }
                    },
                    "required": ["city"]
                }
            }
        }
    
    @staticmethod
    def query_order_tool() -> Dict[str, Any]:
        """跨境电商订单查询工具定义"""
        return {
            "type": "function",
            "function": {
                "name": "query_order",
                "description": "根据订单编号查询订单状态、物流信息和商品明细。本工具只读取数据，不修改订单。用户未提供订单编号时不要调用，应先要求用户补充。",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "订单编号，格式为'AMZ-XXXXX'或'WM-XXXXX'"
                        },
                        "platform": {
                            "type": "string",
                            "enum": ["amazon", "walmart", "shopify"],
                            "description": "订单来源平台"
                        }
                    },
                    "required": ["order_id"]
                }
            }
        }

2.2 工具描述的黄金法则

工具描述直接影响模型决策质量，有几个关键原则：

  • 名称用明确动词短语：search_products优于process_data，adjust_budget优于handle_ad。模型不会“读代码”，而是根据名称和描述做语义判断。

  • 描述说明使用边界：除了说明“能做什么”，还要明确“什么时候用、什么时候不能用”。例如订单查询工具在用户未提供订单号时不应调用，应先要求用户补充。

  • 参数描述尽量详细：参数描述越清晰，模型传参越准确。@ToolParam注解中的描述是模型决定如何填充参数的唯一依据。

  • JSON Schema是参数契约：Schema约束字段类型、必填属性、枚举值、数字范围。但Schema只能解决结构合法性，不能替代业务校验——退款金额符合JSON Schema，但超过订单实付金额仍需业务系统判断。

三、技术实现：从工具定义到系统执行

3.1 工具注册与执行器

在Python中，可以使用装饰器模式简化工具注册：

# tools/registry.py
from typing import Dict, Callable, Any, Optional
import inspect
import json
from functools import wraps

class ToolRegistry:
    """工具注册中心 - 管理所有可调用工具"""
    
    def __init__(self):
        self._tools: Dict[str, Dict] = {}
        self._executors: Dict[str, Callable] = {}
    
    def register(self, name: str, description: str, parameters: Dict):
        """装饰器：注册工具"""
        def decorator(func: Callable):
            self._tools[name] = {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": parameters
                }
            }
            self._executors[name] = func
            return func
        return decorator
    
    def get_tool_schemas(self) -> List[Dict]:
        """获取所有工具定义（用于发送给LLM）"""
        return list(self._tools.values())
    
    async def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
        """执行工具调用"""
        if tool_name not in self._executors:
            raise ValueError(f"未知工具: {tool_name}")
        
        # 参数校验（业务级）
        self._validate_args(tool_name, arguments)
        
        # 执行工具
        return await self._executors[tool_name](**arguments)
    
    def _validate_args(self, tool_name: str, args: Dict):
        """业务级参数校验（JSON Schema之外）"""
        # 示例：校验必填参数是否存在
        schema = self._tools[tool_name]["function"]["parameters"]
        required = schema.get("required", [])
        for field in required:
            if field not in args:
                raise ValueError(f"缺少必填参数: {field}")
        
        # 可扩展：类型校验、范围校验等

3.2 跨境电商工具注册示例

# tools/ecommerce_tools.py
from tools.registry import ToolRegistry
import httpx
from typing import Dict, List, Optional

registry = ToolRegistry()

@registry.register(
    name="search_products",
    description="按名称、品类或关键词搜索商品。当用户询问商品推荐、查找商品时使用。",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "搜索关键词，如'无线耳机'"},
            "category": {"type": "string", "description": "品类过滤（可选）"},
            "min_price": {"type": "number", "description": "最低价格"},
            "max_price": {"type": "number", "description": "最高价格"},
            "limit": {"type": "integer", "description": "返回数量，默认10"}
        },
        "required": ["query"]
    }
)
async def search_products(query: str, category: Optional[str] = None, 
                          min_price: Optional[float] = None, 
                          max_price: Optional[float] = None,
                          limit: int = 10) -> List[Dict]:
    """搜索商品 - 调用产品数据库或电商API"""
    # 实际实现中调用数据库或Amazon/Walmart API
    # 这里为示例
    return [
        {"id": "P001", "name": "Sony WH-1000XM6", "price": 129.00, "stock": 45},
        {"id": "P002", "name": "JBL Tune Beam", "price": 89.00, "stock": 120}
    ][:limit]


@registry.register(
    name="adjust_ad_bid",
    description="调整指定广告组的关键词出价。用于优化广告ROI时调用。高风险操作，需用户确认。",
    parameters={
        "type": "object",
        "properties": {
            "campaign_id": {"type": "string", "description": "广告活动ID"},
            "keyword": {"type": "string", "description": "关键词"},
            "new_bid": {"type": "number", "description": "新出价（美元）"},
            "reason": {"type": "string", "description": "调价原因"}
        },
        "required": ["campaign_id", "keyword", "new_bid"]
    }
)
async def adjust_ad_bid(campaign_id: str, keyword: str, new_bid: float, 
                         reason: str = "") -> Dict:
    """执行广告调价"""
    # 调用亚马逊广告API
    return {
        "status": "pending_review",  # 高风险操作需要审批
        "campaign_id": campaign_id,
        "keyword": keyword,
        "new_bid": new_bid,
        "message": "出价修改请求已提交审批"
    }

3.3 完整的Function Calling执行流程

# agent/function_calling_agent.py
from typing import List, Dict, Any
import json
from openai import AsyncOpenAI

class FunctionCallingAgent:
    """
    基于Function Calling的AI Agent
    实现从自然语言到系统指令的完整转换链路
    """
    
    def __init__(self, api_key: str, tool_registry):
        self.client = AsyncOpenAI(api_key=api_key)
        self.registry = tool_registry
        self.conversation_history: List[Dict] = []
    
    async def chat(self, user_message: str) -> str:
        """处理用户消息，支持多轮工具调用"""
        # 步骤1：添加用户消息到历史
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })
        
        # 步骤2：获取工具定义
        tools = self.registry.get_tool_schemas()
        
        # 步骤3：调用LLM（带工具定义）
        response = await self.client.chat.completions.create(
            model="gpt-4o",
            messages=self.conversation_history,
            tools=tools,
            tool_choice="auto"  # 由模型决定是否调用工具
        )
        
        assistant_message = response.choices[0].message
        
        # 步骤4：检查是否有工具调用请求
        if assistant_message.tool_calls:
            # 将助手的工具调用请求加入历史
            self.conversation_history.append(assistant_message.model_dump())
            
            # 步骤5：依次执行工具调用
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                arguments = json.loads(tool_call.function.arguments)
                
                # 执行工具（含业务校验）
                try:
                    result = await self.registry.execute(tool_name, arguments)
                    result_content = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    result_content = json.dumps({"error": str(e)})
                
                # 步骤6：将工具执行结果加入历史
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result_content
                })
            
            # 步骤7：再次调用LLM生成最终回答（基于工具执行结果）
            final_response = await self.client.chat.completions.create(
                model="gpt-4o",
                messages=self.conversation_history
            )
            
            final_message = final_response.choices[0].message
            self.conversation_history.append(final_message.model_dump())
            return final_message.content
        
        # 无工具调用，直接返回
        self.conversation_history.append(assistant_message.model_dump())
        return assistant_message.content

四、生产工程实践

4.1 并行调用：处理复杂业务

现代LLM支持在一次响应中返回多个工具调用，Agent可以并行执行。这在电商场景中非常实用——用户问“请对比三星和华为的最新手机”，Agent可以同时调用两次商品查询工具。

# 并行调用示例
# LLM返回的tool_calls可能包含多个调用
# 应用程序需要并发执行这些调用
import asyncio

async def execute_parallel(tool_calls):
    tasks = []
    for tc in tool_calls:
        name = tc.function.name
        args = json.loads(tc.function.arguments)
        tasks.append(registry.execute(name, args))
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

4.2 跨模型兼容：抽象是关键

不同模型厂商的Function Calling格式存在差异。采用抽象层封装差异，上层业务不感知模型切换。

# 函数调用转换器（参考OpenHands实现）
# 将OpenAI格式转换为模型原生格式
class FunctionCallConverter:
    @staticmethod
    def to_openai_format(tool_calls: List[Dict]) -> List[Dict]:
        """转换为OpenAI格式"""
        return [{
            "type": "function",
            "function": {
                "name": tc["name"],
                "description": tc.get("description", ""),
                "parameters": tc.get("parameters", {})
            }
        } for tc in tool_calls]
    
    @staticmethod
    def from_model_response(response: Dict) -> List[Dict]:
        """从模型响应解析工具调用"""
        # 不同模型的响应格式不同
        # 需实现各自解析逻辑
        pass

4.3 安全护栏：永远信任，但永远验证

即使模型生成的参数完全符合JSON Schema，应用程序也必须再次执行业务校验：

class ToolGuardrail:
    """工具调用的安全护栏"""
    
    @staticmethod
    def validate_adjust_bid(args: Dict) -> Tuple[bool, str]:
        """校验广告调价参数"""
        new_bid = args.get("new_bid", 0)
        
        # 业务规则：出价不能低于0.1美元，不能超过50美元
        if new_bid < 0.1:
            return False, "出价不能低于0.1美元"
        if new_bid > 50:
            return False, "出价不能超过50美元，超过需人工审批"
        
        # 规则：调价幅度超过50%需二次确认
        # 实际实现中需查询原出价
        return True, "参数校验通过"

五、总结

Function Calling在AI智能体中的价值，可以浓缩为一句话：让模型从“生成文本”走向“调用真实程序能力”。

从技术实现角度看，核心要把握三个关键节点：

  • 工具定义：名称、描述、参数的清晰程度，决定了模型能否选对工具、传对参数
  • 执行链路：模型发出调用请求 → 应用程序校验并执行 → 结果回传模型 → 生成最终回答，每个环节都需建立安全护栏
  • 工程化：并行调用支持复杂业务、跨模型兼容降低切换成本、可观测性保障生产稳定性

跨境电商场景中，从订单查询、库存同步到广告调价，Function Calling让AI Agent真正具备了“动手做事”的能力，而不仅仅是“开口说话”。这中间的技术桥梁，就是一套精心设计的工具定义与执行体系。

# 人工智能 # python
Logo
快递鸟一站式物流API解决方案

电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。

更多推荐

cover

启动优化案例——从 2.8 秒到 0.85 秒的电商元服务实战复盘

avatar 快递鸟社区
cover

电商多店运维：云机24小时挂机频繁掉线、账号无故风控深度原因分析及解决方案

avatar 快递鸟社区

产品种草视频生成工具项目复盘：中小电商团队批量营销素材落地实践

我们是一支6人的电商内容小组，岗位覆盖内容策划、素材制作、审核对接，主要服务中小消费类品牌，输出社媒种草以及信息流投放类视频物料。我负责创意梳理、工具选型以及素材版本管控。此前我们一直依靠策划写脚本，剪辑实拍混剪产出种草视频，整套链路消耗人力，经常卡在创意发散、开场钩子构思、脚本转画面这几个环节。

avatar 快递鸟社区