二、The Power of LLM Function Calling
一、Function Calling 的诞生背景
1. 传统LLM的局限性
- 静态文本生成的不足:早期的LLM(如早期版本的ChatGPT)主要依赖预训练的知识库生成文本,但无法直接与外部系统或API交互。这意味着它们只能基于历史数据回答问题,无法获取实时信息或执行具体操作(如查询天气、下单支付等)。
- 缺乏执行能力:用户需要的许多任务(如预订机票、查询股票价格)需要动态数据或外部系统支持,而传统LLM无法直接完成这些操作,导致其应用范围受限。
2. 实际应用场景的需求驱动
- 实时性与动态数据需求:许多实际应用(如客服系统、智能助手)需要实时数据支持(如当前天气、最新股价、物流状态等),而静态文本生成无法满足这一需求。
- 复杂任务的执行需求:用户希望AI不仅能回答问题,还能直接执行操作(如调用API完成支付、生成报告、控制设备等)。例如:
- 医疗领域:根据患者症状调用医疗数据库获取诊断建议。
- 电商领域:根据用户需求调用库存系统完成订单处理。
3. 技术发展的推动
- API与外部工具的普及:随着互联网和云计算的发展,大量API(如天气API、支付API)和工具(如数据库、第三方服务)已标准化,为LLM调用外部功能提供了基础设施。
- 函数式编程与模块化设计:函数式编程(如Haskell、Lisp)强调将功能封装为可复用的函数,这一思想被借鉴到LLM中,使其能够通过调用预定义函数扩展能力。
4. 开发者效率的提升需求
- 手动集成的低效性:在Function Calling出现前,开发者需手动编写代码将LLM与外部API结合(如使用LangChain或Semantic Kernel框架),流程复杂且易出错(如参数传递错误)。
- 标准化与自动化:Function Calling机制通过标准化接口(如OpenAI的API规范)简化了开发流程,允许模型自动识别并调用函数,减少人工干预。
5. 行业应用的迫切需求
- 企业级应用落地:2023年后,大模型开始从实验室走向实际业务场景(如客服、金融、工业设计),但需与企业内部系统(如ERP、CRM)无缝对接。Function Calling为此提供了技术桥梁。
- Agent(智能体)技术的兴起:通过Function Calling,AI Agent能够自主规划任务流程(如AgentFoundry平台),进一步推动了自动化决策和复杂任务的实现。
关键时间节点与里程碑
- 2023年6月:OpenAI在GPT-4中引入Function Calling机制,允许模型直接调用外部函数和API,标志着LLM从“文本生成”向“功能执行”跨越。
- 2024年:Function Calling成为大模型应用的元年,推动了标准化接口的普及和行业协同效率的提升。
二、如何理解 Function Calling
Function Calling(函数调用) 是大型语言模型(LLM)通过调用外部工具、API或自定义函数来扩展自身能力的核心机制。它允许模型突破预训练知识的限制,直接与外部系统交互,完成动态任务(如实时数据查询、执行操作等)。以下是对其核心概念、工作原理、优势及应用场景的详细解析:
1. 核心概念
-
定义:
- Function Calling 允许 LLM 根据用户的自然语言请求,主动调用外部定义的函数或 API,获取实时数据或执行具体操作(如查询天气、下单支付、控制设备等)。
- 它是 LLM 从“文本生成工具”升级为“智能执行者”的关键能力,弥补了传统 LLM 仅依赖静态知识库的不足。
-
与传统 LLM 的区别:
- 传统 LLM:仅能基于预训练数据生成文本,无法直接调用外部工具,无法处理实时数据或执行操作。
- Function Calling:通过调用外部函数,LLM 可以动态获取最新信息、执行复杂任务,例如:
- 查询实时天气(调用天气 API)。
- 预定机票(调用第三方订票系统)。
- 分析用户文档(调用文件处理工具)。
2. 工作原理
Function Calling 的典型工作流程如下(以知识库中的描述为例):
-
定义函数:
- 开发者预先向 LLM 注册一组外部函数,包括函数名、参数描述、功能说明等(如
get_weather(location, unit)
)。 - 函数通常以结构化格式(如 JSON)定义,例如:
{"name": "get_current_weather","description": "获取指定位置的当前天气","parameters": {"type": "object","properties": {"location": {"type": "string", "description": "城市名称"},"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}},"required": ["location", "unit"]} }
- 开发者预先向 LLM 注册一组外部函数,包括函数名、参数描述、功能说明等(如
-
用户请求:
- 用户通过自然语言提出需求,例如:“明天北京的天气如何?”
-
模型解析与决策:
- LLM 解析用户输入,判断是否需要调用外部函数。例如,若用户询问天气,模型会识别出需要调用
get_weather
函数。
- LLM 解析用户输入,判断是否需要调用外部函数。例如,若用户询问天气,模型会识别出需要调用
-
生成函数调用指令:
- 模型生成结构化指令(如 JSON),指定函数名及参数:
{"name": "get_current_weather","arguments": {"location": "北京","unit": "celsius"} }
- 模型生成结构化指令(如 JSON),指定函数名及参数:
-
执行函数并返回结果:
- 开发者实现的代码逻辑(Agent 程序)解析指令,调用对应函数(如向天气 API 发送请求),获取结果(如温度、湿度等)。
-
整合结果生成响应:
- 函数返回结果后,模型结合结果和上下文生成最终回答,例如:“北京明天的天气为 22°C,晴。”
3. 核心优势
- 突破静态知识限制:
- LLM 可获取实时数据(如股票价格、新闻事件),而非仅依赖预训练知识。
- 执行复杂任务:
- 直接操作外部系统(如下单、生成报告、控制设备),实现“端到端”自动化。
- 模块化扩展能力:
- 开发者可通过添加新函数快速扩展 LLM 的功能,例如集成 CRM 系统或数据库。
4. 典型应用场景
- 实时数据查询:
- 天气预报、股票行情、物流追踪等需要动态数据的场景。
- 自动化操作:
- 预订机票、创建日程、发送邮件、生成代码等。
- 企业级应用:
- 客户服务(自动处理退款、查询订单)、数据分析(调用 BI 工具生成报表)、IT 自动化(关闭服务器、部署代码)。
5. Function Calling 的挑战与解决方案
-
挑战:
- 平台依赖性:不同 LLM(如 OpenAI、阿里云、Anthropic)的 Function Calling 接口不统一,切换模型需重写代码。
- 复杂任务支持不足:多步骤任务(如“预订机票并安排接送”)可能需要多次函数调用,逻辑复杂度高。
- 安全与控制:调用外部 API 需处理敏感数据权限(如用户隐私)。
-
解决方案:
- MCP 协议:通过标准化协议(如 Model Context Protocol,MCP)统一接口,降低平台依赖性(参考知识库中 MCP 的描述)。
- Agent 架构:结合 Agent(智能体)技术,让 LLM 自主规划多步骤任务流程(如调用多个函数并协调结果)。
- 安全机制:通过权限控制、数据加密和审计日志确保调用安全。
6. 与相关技术的对比
-
Function Calling vs. MCP(模型上下文协议):
- Function Calling 是 LLM 调用外部函数的具体机制,依赖特定平台的 API。
- MCP 是一种通用协议,旨在标准化 LLM 与外部工具的交互(如统一接口、跨平台兼容性),解决 Function Calling 的碎片化问题。
-
Function Calling vs. Agent(智能体):
- Function Calling 是 Agent 的核心能力之一,负责执行具体操作。
- Agent 是更高级的架构,包含感知环境、规划任务、调用函数的完整流程(如自主决策“先查天气,再建议行程”)。
三、Function Calling 的实现过程
1. 定义外部函数(Function Definition)
- 目标:向 LLM 注册可调用的函数,描述其功能、参数和返回值。
- 具体步骤:
- 函数描述:定义函数的名称、参数类型、参数描述和功能说明。例如:
{"name": "get_weather","description": "获取指定城市的天气信息","parameters": {"type": "object","properties": {"city": {"type": "string", "description": "城市名称"},"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}},"required": ["city"]} }
- 函数实现:开发者需编写实际执行函数的代码(如调用天气 API),例如:
def get_weather(city, unit="celsius"):# 调用第三方天气 API 获取数据return {"temperature": 25, "unit": unit}
- 函数描述:定义函数的名称、参数类型、参数描述和功能说明。例如:
2. 集成函数到 LLM 环境
- 目标:将定义好的函数信息传递给 LLM,使其能够识别和调用这些函数。
- 实现方式:
- 通过 API 参数:在调用 LLM 的接口时,将函数列表作为参数传递。例如 OpenAI 的
tools
参数:client = OpenAI() tools = [ # 定义函数列表{"type": "function","name": "get_weather","description": "获取指定城市的天气信息","parameters": {...}} ] response = client.responses.create(model="gpt-4o",input=[{"role": "user", "content": "北京的天气如何?"}],tools=tools )
- 平台差异:不同厂商(如 OpenAI、Deepseek、Anthropic)的接口可能略有不同,需参考对应文档。
- 通过 API 参数:在调用 LLM 的接口时,将函数列表作为参数传递。例如 OpenAI 的
3. 用户请求解析与函数调用指令生成
- 目标:LLM 解析用户输入,判断是否需要调用外部函数,并生成结构化调用指令。
- 流程:
- 用户输入:用户通过自然语言提出请求(如“明天上海的天气如何?”)。
- 模型解析:LLM 分析语义,识别需要调用的函数(如
get_weather
)及其参数(如city="上海"
)。 - 生成调用指令:模型输出符合函数定义的 JSON 格式指令:
{"name": "get_weather","arguments": {"city": "上海", "unit": "celsius"} }
4. 执行函数调用(开发者代码层)
- 目标:开发者代码解析 LLM 的调用指令,实际执行函数或 API 调用。
- 关键步骤:
- 解析指令:将 JSON 格式的调用指令转换为可执行的参数。
import json tool_call = response.output[0] args = json.loads(tool_call.arguments)
- 执行函数:调用对应函数(如
get_weather("上海", "celsius")
)。 - 处理异常:若函数调用失败(如参数错误),需返回错误信息给 LLM。
- 解析指令:将 JSON 格式的调用指令转换为可执行的参数。
5. 返回结果并生成最终响应
- 目标:将函数执行结果反馈给 LLM,生成最终用户可见的自然语言回答。
- 流程:
- 结果传递:将函数返回的结果(如天气数据)作为输入重新提交给 LLM:
input_messages.append({"type": "function_call_output","call_id": tool_call.call_id,"output": {"temperature": 28, "unit": "celsius"} }) response_final = client.responses.create(model="gpt-4o",input=input_messages,tools=tools )
- 生成回答:LLM 结合结果和上下文,生成最终回答(如“上海明天的温度为 28°C”)。
- 结果传递:将函数返回的结果(如天气数据)作为输入重新提交给 LLM:
6. 标准化与扩展(MCP 协议)
- 挑战:不同厂商的 Function Calling 接口不统一,导致代码复用困难。
- 解决方案:采用 MCP(Model Context Protocol) 标准化协议:
- 特点:
- 使用 JSON-RPC 2.0 标准格式,确保跨平台兼容性。
- 开发者可直接调用符合 MCP 标准的工具库(如天气 API、数据库接口)。
- 优势:通过 MCP 集合站(如 mcp.so)快速集成第三方工具,降低开发成本。
- 特点:
关键挑战与解决方案
-
参数传递准确性:
- 问题:LLM 可能生成不完整的参数或错误类型。
- 解决方案:严格定义参数格式(如 JSON Schema),并添加参数验证逻辑。
-
多步骤任务处理:
- 问题:复杂任务需多次函数调用(如“预订机票并安排接送”)。
- 解决方案:结合 Agent 架构,让 LLM 自主规划多步骤流程(参考知识库条目[4])。
-
安全与权限控制:
- 问题:调用敏感函数(如支付 API)需权限管理。
- 解决方案:通过 OAuth 2.0 或 API Key 验证,限制函数调用范围。
示例代码(基于 OpenAI)
以下是一个完整的 Function Calling 流程示例(参考知识库条目[1]):
from openai import OpenAI
import json# 定义函数描述
tools = [{"type": "function","name": "get_traffic_status","description": "获取指定城市的实时交通状况","parameters": {"type": "object","properties": {"city": {"type": "string", "description": "城市名称"}},"required": ["city"]}}
]# 用户输入
input_messages = [{"role": "user", "content": "现在旧金山的交通状况如何?"}]# 调用模型生成函数调用指令
client = OpenAI()
response = client.responses.create(model="gpt-4o",input=input_messages,tools=tools
)# 执行函数
tool_call = response.output[0]
args = json.loads(tool_call.arguments)
traffic_status = get_traffic_status(args["city"]) # 实际调用函数# 返回结果并生成最终回答
input_messages.append(tool_call)
input_messages.append({"type": "function_call_output","call_id": tool_call.call_id,"output": traffic_status
})
final_response = client.responses.create(model="gpt-4o",input=input_messages,tools=tools
)
print(final_response.output_text)
总结
Function Calling 的实现过程可概括为以下步骤:
- 定义函数:明确函数功能、参数及返回值。
- 集成到 LLM:通过 API 将函数信息传递给模型。
- 解析用户请求:LLM 生成结构化调用指令。
- 执行函数调用:开发者代码解析指令并调用实际函数。
- 反馈结果:将结果返回 LLM,生成最终回答。
- 标准化与扩展:通过 MCP 协议提升跨平台兼容性。
四、远程 Function Calling 的调用
1. 远程 Function Calling 的核心概念
定义:
远程 Function Calling 是指通过网络协议(如HTTP、gRPC、JSON-RPC等)调用部署在远程服务器上的函数或服务。例如,LLM需要调用天气API获取实时数据,或调用数据库服务查询信息。
关键组件:
- 客户端:发起函数调用的实体(如LLM、前端应用)。
- 远程服务端:提供函数或API的服务器,负责执行实际操作。
- 通信协议:定义如何序列化参数、传输数据、处理响应(如RPC、REST API)。
- 函数描述:远程函数的元数据(名称、参数、返回类型等)。
2. 远程 Function Calling 的实现步骤
步骤 1:定义远程函数接口
- 接口描述:使用标准化格式(如JSON Schema、Protocol Buffers)定义远程函数的参数、返回类型和功能。
// 示例:天气查询函数的接口定义 {"name": "get_weather","parameters": {"city": {"type": "string"},"date": {"type": "string", "format": "date"}},"returns": {"type": "object", "properties": {"temperature": "number"}} }
步骤 2:序列化参数与构建请求
- 参数序列化:将模型生成的参数(如JSON对象)转换为网络传输格式(如JSON、Protobuf)。
# 示例:模型输出的参数 arguments = {"city": "北京", "date": "2025-04-15"} serialized_data = json.dumps(arguments) # 转换为JSON字符串
步骤 3:选择通信协议并发送请求
- 协议选择:
- HTTP/REST:简单易用,适合无状态服务。
import requests response = requests.post("http://weather-service/api/get_weather",json=arguments )
- gRPC:高性能、二进制传输,适合分布式系统。
// 定义proto文件 service WeatherService {rpc GetWeather (WeatherRequest) returns (WeatherResponse); } message WeatherRequest {string city = 1;string date = 2; }
- JSON-RPC:轻量级,支持异步调用。
const request = {"jsonrpc": "2.0","method": "get_weather","params": arguments,"id": 1 };
- HTTP/REST:简单易用,适合无状态服务。
步骤 4:远程服务端处理请求
- 反序列化参数:将网络数据转换为本地对象。
# 示例:服务端接收请求 @app.route('/get_weather', methods=['POST']) def handle_weather():data = request.get_json()city = data['city']date = data['date']# 调用本地函数处理return get_weather(city, date)
- 执行函数:调用实际业务逻辑(如查询数据库或外部API)。
步骤 5:返回结果与错误处理
- 结果序列化:将结果转换为网络传输格式。
result = {"temperature": 25} return jsonify(result)
- 错误处理:捕获异常并返回标准化错误信息(如HTTP状态码、RPC错误码)。
except Exception as e:return jsonify({"error": str(e)}), 500
步骤 6:客户端处理响应
- 解析结果:将返回的数据反序列化并集成到应用逻辑中。
if response.status_code == 200:temperature = response.json()["temperature"]# 生成最终回答 else:# 处理错误(如提示用户重试)
3. 关键技术与协议
1. RPC(远程过程调用)
- 原理:允许客户端像调用本地函数一样调用远程服务,隐藏底层网络细节。
- 实现方式:
- gRPC(推荐):
- 使用 Protocol Buffers 定义接口(
.proto
文件)。 - 支持高性能二进制传输和流式处理。
- 示例(proto文件):
service WeatherService {rpc GetWeather(WeatherRequest) returns (WeatherResponse); } message WeatherRequest {string city = 1;string date = 2; } message WeatherResponse {float temperature = 1; }
- 使用 Protocol Buffers 定义接口(
- JSON-RPC:
- 基于JSON格式,适合简单场景。
- 示例请求:
{"jsonrpc": "2.0","method": "get_weather","params": {"city": "北京", "date": "2025-04-15"},"id": 1 }
- gRPC(推荐):
2. REST API
- 特点:
- 使用HTTP动词(GET/POST/PUT/DELETE)操作资源。
- 参数通过URL、查询参数或请求体传递。
- 示例:
POST /api/weather HTTP/1.1 Content-Type: application/json{"city": "北京","date": "2025-04-15" }
3. MCP(Model Context Protocol)
- 作用:标准化AI与远程服务的交互,支持跨平台调用。
- 优势:
- 统一函数调用格式(如JSON-RPC 2.0)。
- 内置错误处理和上下文管理。
- 示例:
{"method": "call_function","params": {"function_name": "get_weather","arguments": {"city": "北京"}} }
4. 安全与优化
安全措施
- 身份验证:使用API Key、OAuth 2.0或JWT。
- 加密传输:通过HTTPS或gRPC的TLS加密数据。
- 参数校验:验证参数格式和类型(如使用JSON Schema)。
from pydantic import BaseModel, Field class WeatherRequest(BaseModel):city: str = Field(..., max_length=50)date: str = Field(..., regex=r"^\d{4}-\d{2}-\d{2}$")
性能优化
- 批量调用:将多个请求合并为一个批次(如Excel的批量函数调用)。
- 缓存:对频繁调用的函数结果进行缓存(如Redis)。
- 异步处理:使用异步框架(如asyncio)提升响应速度。
5. 完整代码示例(gRPC实现)
场景:通过gRPC远程调用天气查询服务。
步骤:
-
定义proto文件(weather.proto):
syntax = "proto3"; package weather;service WeatherService {rpc GetWeather (WeatherRequest) returns (WeatherResponse); }message WeatherRequest {string city = 1;string date = 2; }message WeatherResponse {float temperature = 1; }
-
服务端实现(Python):
import grpc from concurrent import futures import weather_pb2 import weather_pb2_grpcclass WeatherService(weather_pb2_grpc.WeatherServiceServicer):def GetWeather(self, request, context):# 模拟调用天气APIreturn weather_pb2.WeatherResponse(temperature=25.0)def serve():server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))weather_pb2_grpc.add_WeatherServiceServicer_to_server(WeatherService(), server)server.add_insecure_port('[::]:50051')server.start()server.wait_for_termination()if __name__ == '__main__':serve()
-
客户端调用(Python):
import grpc import weather_pb2 import weather_pb2_grpcchannel = grpc.insecure_channel('localhost:50051') stub = weather_pb2_grpc.WeatherServiceStub(channel)request = weather_pb2.WeatherRequest(city="北京", date="2025-04-15") response = stub.GetWeather(request) print(f"温度:{response.temperature}°C")
6. 总结
远程 Function Calling 的核心是通过标准化协议(如gRPC、REST、JSON-RPC)实现跨网络的函数调用。其关键步骤包括:
- 定义接口:明确函数参数和返回类型。
- 序列化参数:将数据转换为网络传输格式。
- 选择协议:根据需求选择HTTP、gRPC等协议。
- 处理响应:解析结果并集成到应用逻辑中。
- 安全与优化:确保数据安全和高效传输。
五、支持 Function Calling 的国产模型
1. Qwen(通义千问系列)
- 支持情况:明确支持 Function Calling。
- 功能特点:
- 可通过阿里云 API 或魔搭 GPT 工具调用外部工具、数据库或模型。
- 支持与 OpenAI API 兼容的接口(如魔搭 GPT 的调用框架)。
- 应用场景:
- 智能家居控制(如语音指令调用设备)。
- 调用其他模型或 API(如通过魔搭 GPT 整合多模型协作)。
- 技术实现:
- 支持通过
api.aliyun.com/qwen
接口直接调用函数,实现复杂任务流程。
- 支持通过
2. 深度求索(DeepSeek)
- 支持情况:兼容 OpenAI API,支持类似 Function Calling 的功能。
- 功能特点:
- 通过
api.deepseek.com/v1
接口调用模型,支持自定义函数调用逻辑。 - 可与外部工具、数据库或第三方服务集成。
- 通过
- 应用场景:
- 企业级应用中的数据分析、文档生成等复杂任务。
3. 清华智谱AI(GLM 系列)
- 支持情况:兼容 OpenAI API,支持 Function Calling。
- 功能特点:
- 通过
open.bigmodel.cn/api/paas/v4
接口调用模型。 - 支持多模态任务(文本、视觉、语音)的函数调用。
- 通过
- 技术特点:
- 可通过 OpenAI 兼容接口实现与外部工具的交互。
4. 百川智能(Baichuan)
- 支持情况:兼容 OpenAI API,支持 Function Calling。
- 功能特点:
- 通过
api.baichuan-ai.com/v1
接口调用模型。 - 适用于企业级场景中的多任务处理(如数据分析、代码生成)。
- 通过
- 应用场景:
- 通过函数调用整合外部 API 或数据库。
5. 魔搭 GPT(ModelScopeGPT)
- 支持情况:阿里云提供的 大模型调用工具,支持 Function Calling。
- 功能特点:
- 可调用 Qwen、GLM 等国产模型,并整合为复杂任务流程。
- 支持多模型串联(如文本生成、语音合成、视频生成的联合调用)。
- 技术优势:
- 集成 180 万开发者和 900+ 模型,提供丰富的生态支持。
不支持或需验证的模型
讯飞星火(Xinghuo)
- 支持情况:不兼容 OpenAI API,Function Calling 能力有限。
- 说明:
- 需通过讯飞官方文档确认是否支持自定义函数调用接口。
- 部分模型提供免费额度,但功能扩展性可能受限。
如何使用国产模型的 Function Calling?
- 通过 API 接口:
- Qwen:使用阿里云 API 或魔搭 GPT 工具。
- DeepSeek/Baichuan:调用其 OpenAI 兼容接口(如
api.deepseek.com/v1
)。
- 框架集成:
- 迅策框架:通过其 API 动态调用远程资源。
- LangChain:结合国产模型的 API 实现复杂流程(如自然语言到 SQL 转换)。
六、扩展
6.1 多个函数参数数量和名称不一致
1. 函数选择:如何确定调用哪个函数?
核心思路:
通过 函数名称显式标识 和 参数上下文匹配,让模型明确返回目标函数的名称和参数。
实现步骤:
-
定义函数架构(Tool Schema):
- 使用 Pydantic 模型 或 字典结构 明确每个函数的参数名称、类型和描述。例如:
from pydantic import BaseModel, Fieldclass AddFunction(BaseModel):a: int = Field(..., description="第一个整数")b: int = Field(..., description="第二个整数")class GetWeatherFunction(BaseModel):city: str = Field(..., description="城市名称")date: str = Field(..., description="日期(格式:YYYY-MM-DD)")
- 使用 Pydantic 模型 或 字典结构 明确每个函数的参数名称、类型和描述。例如:
-
在提示中明确可用函数:
- 在模型输入中列出所有可用函数及其参数,引导模型选择正确的函数。例如:
"可用工具:\n" "1. add:计算两个整数之和(参数:a, b)\n" "2. get_weather:查询指定城市的天气(参数:city, date)\n" "请根据用户请求选择合适的工具并返回结构化参数。"
- 在模型输入中列出所有可用函数及其参数,引导模型选择正确的函数。例如:
-
模型输出结构化指令:
- 要求模型返回包含 函数名称 和 参数字典 的 JSON 对象。例如:
或{"function_name": "add","arguments": {"a": 1, "b": 2} }
{"function_name": "get_weather","arguments": {"city": "北京", "date": "2025-04-15"} }
- 要求模型返回包含 函数名称 和 参数字典 的 JSON 对象。例如:
2. 参数提取:如何从 LLM 产出中获取不同函数的参数?
核心思路:
通过 函数名称匹配预定义的参数结构,动态解析参数键值。
实现步骤:
-
维护函数映射表:
- 将函数名称与对应的参数结构(如 Pydantic 模型)映射,例如:
tool_mapping = {"add": AddFunction,"get_weather": GetWeatherFunction }
- 将函数名称与对应的参数结构(如 Pydantic 模型)映射,例如:
-
解析模型输出:
- 根据
function_name
从映射表中获取对应的参数模型,然后验证和提取参数:def parse_tool_call(response_json):function_name = response_json["function_name"]arguments = response_json["arguments"]# 获取对应的参数模型tool_model = tool_mapping.get(function_name)if not tool_model:raise ValueError(f"无效的函数名称:{function_name}")# 验证参数并提取有效值try:validated_args = tool_model(**arguments).dict()return function_name, validated_argsexcept Exception as e:raise ValueError(f"参数验证失败:{str(e)}")
- 根据
-
动态调用函数:
- 根据函数名称和参数调用对应的后端逻辑:
def execute_tool(function_name, arguments):if function_name == "add":return add_function(**arguments)elif function_name == "get_weather":return get_weather(**arguments)else:raise NotImplementedError("未实现的函数")
- 根据函数名称和参数调用对应的后端逻辑:
3. 完整代码示例
场景:用户输入“计算1+2”或“北京明天的天气”。
代码实现:
from pydantic import BaseModel, Field
import json# 定义工具的参数模型
class AddFunction(BaseModel):a: int = Field(..., description="第一个整数")b: int = Field(..., description="第二个整数")class GetWeatherFunction(BaseModel):city: str = Field(..., description="城市名称")date: str = Field(..., description="日期(格式:YYYY-MM-DD)")# 函数映射表
tool_mapping = {"add": AddFunction,"get_weather": GetWeatherFunction
}# 模拟后端函数
def add_function(a, b):return a + bdef get_weather(city, date):# 模拟调用天气APIreturn f"{city} {date} 的天气:晴,20°C"# 处理模型输出的函数
def process_response(response_json):try:function_name = response_json["function_name"]arguments = response_json["arguments"]# 验证参数tool_model = tool_mapping.get(function_name)if not tool_model:return f"错误:不支持的函数 {function_name}"validated_args = tool_model(**arguments).dict()# 执行函数if function_name == "add":result = add_function(**validated_args)elif function_name == "get_weather":result = get_weather(**validated_args)else:return "未实现该函数"return f"结果:{result}"except Exception as e:return f"参数错误:{str(e)}"# 示例输入(模拟LLM返回的JSON)
llm_response_add = {"function_name": "add","arguments": {"a": 1, "b": 2}
}llm_response_weather = {"function_name": "get_weather","arguments": {"city": "北京", "date": "2025-04-16"}
}# 测试
print(process_response(llm_response_add)) # 输出:结果:3
print(process_response(llm_response_weather)) # 输出:结果:北京 2025-04-16 的天气:晴,20°C
关键点总结
-
函数选择:
- 显式标识函数名称:要求模型返回
function_name
字段。 - 映射表关联:通过
tool_mapping
将函数名称与参数模型绑定。
- 显式标识函数名称:要求模型返回
-
参数提取:
- 类型安全验证:使用 Pydantic 模型确保参数类型和必填项正确。
- 动态解析:根据函数名称动态选择参数模型,避免硬编码。
-
扩展性:
- 新增函数:只需添加新的 Pydantic 模型到
tool_mapping
,无需修改核心逻辑。 - 参数兼容性:即使参数名称不同,映射表和模型会自动适配。
- 新增函数:只需添加新的 Pydantic 模型到
常见问题处理
- 参数缺失或类型错误:
- Pydantic 会自动抛出验证错误,可捕获异常并返回用户友好的提示。
- 多函数调用:
- 对于需要多次函数调用的场景,可将中间结果存储在上下文中,供后续调用使用。
- 函数名称冲突:
- 可通过命名规范(如
weather.get_weather
)或分组(如按模块分组)避免冲突。
- 可通过命名规范(如
6.2 基于 LangChain + FastChat 框架实现 通义千问私有化部署 的 Function Calling 具体案例
目标
用户输入“北京现在的天气如何?”,通过 FastChat 服务调用 LangChain Agent,结合本地天气工具函数返回实时天气信息。
1. 环境准备
1.1 安装依赖库
pip install langchain fastchat transformers
1.2 部署 FastChat 服务
- 下载通义千问模型(如
qwen-max
)并解压到本地路径(假设路径为./qwen-model
)。 - 启动 FastChat 服务(本地 REST API):
fastchat-serve-api --model-path ./qwen-model --num-gpu 1 --num_cpu_threads_per_gpu 4
- 这会启动一个本地 API 服务,默认监听
http://localhost:8000/v1
。
- 这会启动一个本地 API 服务,默认监听
2. 定义工具函数
2.1 实现天气查询工具
import json
import requestsdef get_current_weather(location):"""通过本地工具获取实时天气(模拟或真实API)"""# 示例:使用模拟数据(如无外部API权限)if location == "北京":return json.dumps({"location": location,"weather": "晴","temperature": "22°C"})else:return json.dumps({"error": "未找到该城市的天气数据"})
3. 使用 LangChain 构建 Agent
3.1 配置 FastChat 模型
from langchain.llms import HuggingFacePipeline
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline# 配置 FastChat 模型(假设已通过 FastChat 启动服务)
model_url = "http://localhost:8000/v1" # FastChat API 地址# 或者直接加载本地模型(替代 FastChat 服务)
# tokenizer = AutoTokenizer.from_pretrained("qwen-model")
# model = AutoModelForCausalLM.from_pretrained("qwen-model")
# pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)
# llm = HuggingFacePipeline(pipeline=pipe)# 使用 FastChat 的 REST API(需要自定义适配器)
class FastChatLLM:def __init__(self, endpoint):self.endpoint = endpointdef generate(self, prompt):response = requests.post(f"{self.endpoint}/generate",json={"prompt": prompt, "max_tokens": 200, "temperature": 0.1})return response.json()["text"]llm = FastChatLLM(endpoint=model_url)
3.2 定义工具描述
from langchain.agents import Tool# 将天气工具包装为 LangChain 工具
weather_tool = Tool(name="get_current_weather",func=get_current_weather,description="获取指定城市的实时天气信息,参数:location(城市名称)",
)
3.3 创建 Agent
from langchain.agents import initialize_agent
from langchain.agents import AgentType# 初始化 Agent,使用工具和模型
agent = initialize_agent(tools=[weather_tool],llm=llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, # 选择 Agent 类型verbose=True,max_iterations=3,
)
4. 完整流程代码
def main():user_input = "北京现在的天气如何?"# 调用 Agent 执行任务response = agent.run(user_input)print(f"最终答案:{response}")if __name__ == "__main__":main()
5. 运行结果
> Entering new Agent execution
> Action: get_current_weather
> Action Input: {"location": "北京"}
> Observation: {"location": "北京", "weather": "晴", "temperature": "22°C"}
> Thought: 我需要将天气信息整合到回答中。
> Final Answer: 北京当前天气晴,气温22°C。
关键步骤解析
5.1 FastChat 服务启动
- 通过 FastChat 启动本地模型服务,提供 REST API 接口(如
http://localhost:8000/v1
)。 - 可通过
curl
测试服务:curl -X POST "http://localhost:8000/v1/generate" -H "Content-Type: application/json" -d '{"prompt": "你好"}'
5.2 LangChain Agent 工作流程
- 用户输入解析:Agent 根据用户指令(如“北京天气”)分析是否需要调用工具。
- 工具调用:Agent 调用
get_current_weather
函数获取天气数据。 - 结果整合:将工具返回的 JSON 数据转化为自然语言回答。
6. 扩展应用
6.1 添加更多工具
例如,添加时间查询工具:
import datetimedef get_current_time():now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")return json.dumps({"current_time": now})time_tool = Tool(name="get_current_time",func=get_current_time,description="获取当前时间"
)# 重新初始化 Agent 包含新工具
agent = initialize_agent(tools=[weather_tool, time_tool],llm=llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True
)
6.2 支持多轮对话
# 使用 ConversationChain 或记忆组件
from langchain.memory import ConversationBufferMemorymemory = ConversationBufferMemory()
agent_with_memory = initialize_agent(tools=[weather_tool],llm=llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,memory=memory,verbose=True
)
7. 注意事项
-
FastChat 配置:
- 确保模型服务地址(
http://localhost:8000/v1
)正确。 - 根据显存调整
num-gpu
和num_cpu_threads_per_gpu
。
- 确保模型服务地址(
-
LangChain Agent 类型:
- 可选其他 Agent 类型(如
AgentType.OPENAI_FUNCTIONS
),需适配模型能力。
- 可选其他 Agent 类型(如
-
工具函数安全性:
- 避免直接执行用户输入(如
eval()
),确保参数校验。
- 避免直接执行用户输入(如
完整代码整合
import json
from langchain.agents import Tool, initialize_agent, AgentType# 1. 定义工具函数
def get_current_weather(location):if location == "北京":return json.dumps({"location": location,"weather": "晴","temperature": "22°C"})else:return json.dumps({"error": "未找到该城市的天气数据"})# 2. 配置 FastChat 模型(假设已启动服务)
class FastChatLLM:def __init__(self, endpoint):self.endpoint = endpointdef generate(self, prompt):response = requests.post(f"{self.endpoint}/generate",json={"prompt": prompt, "max_tokens": 200, "temperature": 0.1})return response.json()["text"]model_url = "http://localhost:8000/v1"
llm = FastChatLLM(endpoint=model_url)# 3. 创建工具和 Agent
weather_tool = Tool(name="get_current_weather",func=get_current_weather,description="获取指定城市的实时天气信息,参数:location(城市名称)",
)agent = initialize_agent(tools=[weather_tool],llm=llm,agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,verbose=True,
)# 4. 执行查询
def main():user_input = "北京现在的天气如何?"response = agent.run(user_input)print(f"最终答案:{response}")if __name__ == "__main__":main()
总结
通过 LangChain + FastChat 实现私有化部署的通义千问 Function Calling 流程如下:
- 启动 FastChat 服务:本地部署通义千问模型并提供 REST API。
- 定义工具函数:实现本地工具(如天气查询)并包装为 LangChain 工具。
- 构建 Agent:使用 LangChain 的 Agent 调用模型和工具。
- 执行任务:通过 Agent 处理用户指令,自动调用工具并生成回答。