大语言模型(LLM)本身是一个静态的、基于训练数据的概率模型。它能够回答通用问题、生成文本,但无法主动获取实时信息(如天气、股票价格),也无法执行业务操作(如发送邮件、创建订单)。为了让大模型“动手做事”,我们需要一种机制:模型输出结构化的指令,由外部程序执行这些指令,并将结果反馈给模型。这种可被模型调用的外部能力单元,在本文中称为 Skill。
OpenAI 的 Function Calling(函数调用)是目前最成熟、最广泛使用的Skill实现方式。它允许你在调用Chat Completions API时,传入一组可用的函数定义(JSON Schema),模型会判断何时需要调用某个函数,并以结构化JSON返回函数名和参数。你的应用程序收到这个指令后,执行对应的函数,再将结果传回模型,模型最终生成面向用户的自然语言回答。
本文将以Python为例,逐步教你如何创建一个可供大模型调用的Skill。你会学到从定义最简单的函数,到处理复杂参数、错误重试、并行调用等完整工程实践。
在开始编码前,必须理解一个核心原则:大模型不执行任何代码,它只输出“调用意图”。整个过程分为四个阶段:
- 用户输入 → 你发送给模型的请求中包含用户消息和可用的函数定义。
- 模型决策 → 模型返回一个
function_call对象(也可能返回普通文本)。 - 你执行 → 你的代码解析
function_call,调用本地或远程函数,获得结果。 - 反馈模型 → 你把函数执行结果作为新的消息(角色为
function)发回模型,模型据此生成最终回答。
这个循环保证了模型的可控性和安全性——所有实际的系统操作都由你编写的代码完成,模型只负责理解意图和生成回复。
3.1 最简单的函数
假设我们要做一个查询天气的Skill。首先,编写一个普通的Python函数:
python
def get_current_weather(city: str) -> dict: """ 模拟从天气API获取温度。 实际项目中可替换为requests调用真实服务。 """ # 这里模拟数据 weather_data = { "北京": {"temperature": 22, "unit": "celsius", "condition": "晴"}, "上海": {"temperature": 25, "unit": "celsius", "condition": "多云"}, "深圳": {"temperature": 28, "unit": "celsius", "condition": "雨"}, } return weather_data.get(city, {"temperature": "unknown", "condition": "未知"})
这个函数接受一个字符串参数city,返回一个字典。关键点是:函数的输入和输出都必须是可JSON序列化的,因为模型只理解JSON,并且你需要在网络传输中传递这些数据。
3.2 函数签名的重要性
为了让模型正确调用,函数应该做到:
- 单一职责:一个函数只做一件事,比如
get_current_weather不要同时发邮件。 - 参数类型明确:使用类型注解(
str,int,float,bool,List,Dict),这有助于生成Schema。 - 提供清晰的文档字符串:模型可能会读取函数描述(通过Schema中的
description字段),但并非直接解析docstring,不过对你的维护者很有用。
3.3 支持其他语言(Node.js示例)
如果你使用Node.js,等价函数为:
javascript
async function getCurrentWeather(city) { const weatherMap = { "北京": { temperature: 22, unit: "celsius", condition: "晴" }, "上海": { temperature: 25, unit: "celsius", condition: "多云" } }; return weatherMap[city] || { temperature: "unknown", condition: "未知" }; }
函数本身不依赖OpenAI SDK,任何语言都可以实现。
OpenAI API要求你提供每个函数的JSON Schema,它描述了函数名、参数类型、是否必填、参数描述等。模型会根据这个Schema决定如何填充参数。
4.1 手动编写Schema
对于上面的get_current_weather,Schema如下:
json
}, "required": ["city"], "additionalProperties": false } }
name:必须与你的函数名完全一致(模型返回时会用它来标识)。description:非常重要!模型据此判断何时调用该函数。写清楚函数的能力边界。parameters:遵循JSON Schema规范。type: "object"表示参数是一个对象。properties:每个参数的名称、类型、描述。支持string,number,integer,boolean,array,object。required:列出必须提供的参数名。additionalProperties: false:禁止模型传入未定义的参数,提高严谨性。
4.2 自动生成Schema
手动写Schema容易出错,尤其当函数参数复杂时。推荐使用工具从函数签名自动生成。Python中可以使用pydantic或inspect模块。OpenAI官方提供了一个openai库的辅助函数format_function_definitions,但更通用的是使用pydantic创建参数模型。
使用Pydantic(推荐):
python
from pydantic import BaseModel, Field from typing import List, Optional class WeatherParams(BaseModel): city: str = Field(..., description="城市名称,例如:北京、上海") unit: Optional[str] = Field("celsius", description="温度单位,celsius或fahrenheit") # 生成JSON Schema params_schema = WeatherParams.model_json_schema() print(params_schema) # 然后手动组装完整的function定义
使用inspect模块(无需额外依赖):
python
import inspect import json def get_function_schema(func): sig = inspect.signature(func) params = {} required = [] for name, param in sig.parameters.items(): param_type = "string" # 简化处理,实际可映射 params[name] = {"type": param_type, "description": f"参数 {name}"} if param.default == inspect.Parameter.empty: required.append(name) return { "name": func.__name__, "description": func.__doc__ or "", "parameters": { "type": "object", "properties": params, "required": required, }, }
但这种方式无法获取参数的详细描述。更好的实践是维护一个FUNCTION_DEFINITIONS列表,由开发者手工编写或通过装饰器生成。
4.3 处理复杂类型
- 数组:
"type": "array", "items": {"type": "string"}。 - 枚举:
"enum": ["low", "medium", "high"]。 - 嵌套对象:
"type": "object", "properties": {...}。 - 任意JSON:
"type": "object", "additionalProperties": true。
示例:一个发送邮件的函数,参数包含收件人列表和可选附件。
json
{ "name": "send_email", "description": "向指定收件人发送电子邮件", "parameters": { "type": "object", "properties": { "to": { "type": "array", "items": {"type": "string", "format": "email"}, "description": "收件人邮箱地址列表" }, "subject": {"type": "string", "description": "邮件主题"}, "body": {"type": "string", "description": "邮件正文"}, "attachments": { "type": "array", "items": {"type": "string"}, "description": "附件URL列表(可选)" } }, "required": ["to", "subject", "body"] } }
当你调用OpenAI Chat Completions API时,在请求体中增加functions数组,每个元素就是上面定义的Schema。还可以设置function_call参数来控制模型是否强制调用某个函数。
5.1 基本请求示例
python
import openai openai.api_key = "your-api-key" response = openai.ChatCompletion.create( model="gpt-3.5-turbo-0613", # 支持function calling的模型 messages=[ {"role": "user", "content": "北京今天天气怎么样?"} ], functions=[ }, "required": ["city"] } } ], function_call="auto" # 默认auto,模型自主决定是否调用 )
5.2 强制调用某个函数
如果你希望模型必须调用某个函数(比如在特定流程中),可以设置:
python
function_call=
这样即使对话上下文不需要,模型也会返回一个function_call。但需要确保提供的参数合理(模型会尽力填充)。
5.3 多个函数的处理
你可以传入多个函数定义,模型会根据用户输入选择最合适的函数,也可能不调用任何函数直接回复文本。
python
functions = [ get_weather_schema, send_email_schema, search_web_schema ]
模型内部会进行语义匹配。为了提升准确性,务必为每个函数写清晰、详细的description,并在参数描述中包含示例值。
模型返回的响应是一个JSON对象。你需要检查response["choices"][0]["message"]中是否包含function_call字段。
6.1 解析function_call
python
message = response["choices"][0]["message"] if message.get("function_call"): function_name = message["function_call"]["name"] arguments_str = message["function_call"]["arguments"] # 这是一个JSON字符串 arguments = json.loads(arguments_str) print(f"模型要求调用函数: {function_name}, 参数: {arguments}") else: # 模型直接回复文本 print(message["content"])
6.2 执行本地函数
你需要维护一个从函数名到实际函数的映射字典:
python
available_functions = if function_name in available_functions: func = available_functions[function_name] result = func(arguments) # 解包参数 else: result = {"error": f"未知函数: {function_name}"}
注意:模型可能生成错误的参数(例如类型不匹配、缺少必填参数)。建议在调用前使用JSON Schema校验库(如jsonschema)对arguments进行验证,若不通过则向模型返回错误信息。
6.3 将结果返回给模型
函数执行完成后,你需要构造一条function角色的消息,添加到对话历史中,然后再次调用API让模型生成最终回答。
python
# 追加函数执行结果 messages.append(message) # 原模型的function_call消息 messages.append({ "role": "function", "name": function_name, "content": json.dumps(result, ensure_ascii=False), # 必须是字符串 }) # 第二次调用,这次不再传入functions(也可传入,但通常不需要) second_response = openai.ChatCompletion.create( model="gpt-3.5-turbo-0613", messages=messages, # 不再需要functions,因为函数已经执行完毕 ) final_answer = second_response["choices"][0]["message"]["content"] print(final_answer)
最终模型会基于函数返回的数据生成自然语言,例如:“北京今天天气晴,温度22摄氏度。”
下面是一个完整的Python脚本,实现了带有Function Calling的对话循环,支持连续多轮调用。
python
import openai import json import sys openai.api_key = "your-api-key" # ----- 定义函数实现 ----- def get_current_weather(city: str) -> dict: # 模拟实现 weather_db = { "北京": {"temp": 22, "unit": "celsius", "condition": "晴"}, "上海": {"temp": 25, "unit": "celsius", "condition": "多云"}, } return weather_db.get(city, {"temp": "未知", "condition": "无数据"}) def send_email(to: str, subject: str, body: str) -> dict: # 模拟发送邮件 print(f"[模拟发送] 收件人: {to}, 主题: {subject}, 正文: {body}") return {"success": True, "message_id": "mock-12345"} # ----- 函数定义(Schema)----- functions = [ }, "required": ["city"] } }, { "name": "send_email", "description": "发送电子邮件", "parameters": { "type": "object", "properties": { "to": {"type": "string", "format": "email", "description": "收件人邮箱"}, "subject": {"type": "string", "description": "邮件主题"}, "body": {"type": "string", "description": "邮件正文"} }, "required": ["to", "subject", "body"] } } ] # ----- 函数映射表 ----- available_funcs = def run_conversation(user_input): messages = [{"role": "user", "content": user_input}] # 首次请求 response = openai.ChatCompletion.create( model="gpt-3.5-turbo-0613", messages=messages, functions=functions, function_call="auto" ) message = response["choices"][0]["message"] messages.append(message) # 循环处理可能的多次函数调用 while message.get("function_call"): func_name = message["function_call"]["name"] args = json.loads(message["function_call"]["arguments"]) if func_name in available_funcs: result = available_funcs[func_name](args) else: result = {"error": f"Function {func_name} not found"} messages.append({ "role": "function", "name": func_name, "content": json.dumps(result, ensure_ascii=False) }) # 再次请求模型,看是否还需要更多函数调用 response = openai.ChatCompletion.create( model="gpt-3.5-turbo-0613", messages=messages, functions=functions, function_call="auto" ) message = response["choices"][0]["message"] messages.append(message) return message["content"] if __name__ == "__main__": while True: query = input("你: ") if query.lower() in ["exit", "quit"]: break answer = run_conversation(query) print(f"AI: {answer}")
这个例子展示了处理连续多次函数调用的能力(比如模型先查天气,再根据结果发邮件)。注意,while循环可能无限进行,实际生产环境应设置最大迭代次数(如5次)。
8.1 异步Skill与长时间任务
有些Skill执行时间较长(如调用第三方API、生成报告),如果同步等待,会导致用户长时间无响应。解决方案:
- 异步执行:使用
asyncio或后台任务队列(Celery)。模型第一次返回function_call后,立即返回一个“任务已提交”的中间消息,稍后通过Webhook或轮询获取结果。 - 流式处理:OpenAI支持流式响应,但函数调用在流式模式下处理较复杂,建议非流式。
示例:启动一个后台线程执行函数,主线程返回占位符。
python
import threading def long_running_task(arg): # 耗时操作 time.sleep(10) return {"result": "完成"} def handle_function_call(func_name, args): if func_name == "long_task": thread = threading.Thread(target=long_running_task, args=(args,)) thread.start() return {"status": "processing", "message": "任务已启动,稍后查询"} # ...
但更好的方式是引入任务ID和状态查询函数。
8.2 处理函数调用错误
模型可能生成无效参数,或函数执行抛出异常。必须优雅处理,将错误信息返回给模型,让模型可以修正。
python
try: result = func(args) except TypeError as e: result = {"error": f"参数错误: {e}"} except Exception as e: result = {"error": f"执行失败: {str(e)}"}
然后在function消息中返回result,模型通常会道歉并尝试重新调用或询问用户。
8.3 并行函数调用
OpenAI从2023年11月起支持并行函数调用(parallel function calling)。模型可以在一次响应中返回多个function_call(放在一个数组里)。这适用于可以并发执行且相互独立的操作。
请求中需要设置parallel_tool_calls: true(默认开启)。模型返回的message中会包含tool_calls字段,每个元素是一个函数调用。
python
# 响应示例 "}}, "}} ] } }] }
你的代码需要并发执行这些函数,然后将结果以tool消息(每个结果对应一个tool_call_id)返回。
python
import concurrent.futures tool_calls = message["tool_calls"] with concurrent.futures.ThreadPoolExecutor() as executor: futures = [] for tc in tool_calls: func_name = tc["function"]["name"] args = json.loads(tc["function"]["arguments"]) futures.append(executor.submit(available_funcs[func_name], args)) results = [f.result() for f in futures] # 构造多个tool消息 for tc, res in zip(tool_calls, results): messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": json.dumps(res) })
8.4 安全与权限控制
Skill可以访问敏感数据或执行写操作,必须严格限制。
- 输入校验:对模型生成的参数进行白名单校验。例如城市名只能从列表中选,避免SQL注入(如果参数用于数据库查询)。
- 操作权限:在函数内部检查调用者的身份(如通过API Key中的user_id)。不要依赖模型来判断权限。
- 速率限制:对每个函数/每个用户设置调用频率上限。
- 审计日志:记录每次函数调用的参数、结果和调用者,便于追溯。
8.5 跨模型兼容性
OpenAI的Function Calling格式是事实标准,但其他模型(如Anthropic Claude 3的Tool Use、Google Gemini的Function Calling)格式略有差异。若需多模型兼容,可以抽象一层适配器,将各自的工具定义转换为标准格式。
- 函数粒度适中:太细(如
add_one)浪费模型调用次数;太粗(如process_order包含多个步骤)降低灵活性。推荐每个函数对应一个独立的业务操作。 - 描述要具体:模型的决策高度依赖
description。写清楚“何时使用”、“参数含义”、“返回值格式”。例如:“当用户询问某城市天气时调用此函数,不要用于查询历史天气。” - 处理边缘情况:模型可能传入不存在的城市、负数价格等。函数应返回明确的错误信息,并允许模型重新尝试。
- 控制token消耗:每个函数定义都会占用输入token。如果函数很多(>20),可考虑动态选择最相关的几个函数通过
functions参数传入,或者使用工具检索(如Semantic Kernel)。 - 测试覆盖:编写单元测试,模拟模型返回的各种
function_call,确保你的执行逻辑正确。 - 从简单开始:先实现一个只调用一次函数的流程,再逐步添加多轮、并行等高级特性。
通过Function Calling为LLM构建可调用的Skill,是当前将大模型集成到业务系统的最主流、最可靠的方法。它保持了模型与执行环境的清晰边界,既发挥了模型的意图理解优势,又保证了业务操作的确定性和安全性。
本文从定义最简单的一个天气查询函数开始,逐步深入到复杂参数、错误处理、并行调用和生产级安全考量,为你提供了一套完整的工程实践指南。现在,你可以动手为自己的业务场景创建第一个Skill了——无论是查询数据库、发送通知,还是控制物联网设备,大模型都能成为你的智能调度中枢。
记住核心四步:定义函数 → 生成Schema → 请求API → 执行并反馈。熟练之后,你将能构建出真正“会动手”的AI应用。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/255709.html