当前位置: 首页 > news >正文

二、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 的典型工作流程如下(以知识库中的描述为例):

  1. 定义函数

    • 开发者预先向 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"]}
      }
  2. 用户请求

    • 用户通过自然语言提出需求,例如:“明天北京的天气如何?”
  3. 模型解析与决策

    • LLM 解析用户输入,判断是否需要调用外部函数。例如,若用户询问天气,模型会识别出需要调用 get_weather 函数。
  4. 生成函数调用指令

    • 模型生成结构化指令(如 JSON),指定函数名及参数:
      {"name": "get_current_weather","arguments": {"location": "北京","unit": "celsius"}
      }
  5. 执行函数并返回结果

    • 开发者实现的代码逻辑(Agent 程序)解析指令,调用对应函数(如向天气 API 发送请求),获取结果(如温度、湿度等)。
  6. 整合结果生成响应

    • 函数返回结果后,模型结合结果和上下文生成最终回答,例如:“北京明天的天气为 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)的接口可能略有不同,需参考对应文档。

3. 用户请求解析与函数调用指令生成

  • 目标:LLM 解析用户输入,判断是否需要调用外部函数,并生成结构化调用指令。
  • 流程
    1. 用户输入:用户通过自然语言提出请求(如“明天上海的天气如何?”)。
    2. 模型解析:LLM 分析语义,识别需要调用的函数(如 get_weather)及其参数(如 city="上海")。
    3. 生成调用指令:模型输出符合函数定义的 JSON 格式指令:
      {"name": "get_weather","arguments": {"city": "上海", "unit": "celsius"}
      }

4. 执行函数调用(开发者代码层)

  • 目标:开发者代码解析 LLM 的调用指令,实际执行函数或 API 调用。
  • 关键步骤
    1. 解析指令:将 JSON 格式的调用指令转换为可执行的参数。
      import json
      tool_call = response.output[0]
      args = json.loads(tool_call.arguments)
    2. 执行函数:调用对应函数(如 get_weather("上海", "celsius"))。
    3. 处理异常:若函数调用失败(如参数错误),需返回错误信息给 LLM。

5. 返回结果并生成最终响应

  • 目标:将函数执行结果反馈给 LLM,生成最终用户可见的自然语言回答。
  • 流程
    1. 结果传递:将函数返回的结果(如天气数据)作为输入重新提交给 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
      )
    2. 生成回答:LLM 结合结果和上下文,生成最终回答(如“上海明天的温度为 28°C”)。

6. 标准化与扩展(MCP 协议)

  • 挑战:不同厂商的 Function Calling 接口不统一,导致代码复用困难。
  • 解决方案:采用 MCP(Model Context Protocol) 标准化协议:
    • 特点
      • 使用 JSON-RPC 2.0 标准格式,确保跨平台兼容性。
      • 开发者可直接调用符合 MCP 标准的工具库(如天气 API、数据库接口)。
    • 优势:通过 MCP 集合站(如 mcp.so)快速集成第三方工具,降低开发成本。

关键挑战与解决方案

  1. 参数传递准确性

    • 问题:LLM 可能生成不完整的参数或错误类型。
    • 解决方案:严格定义参数格式(如 JSON Schema),并添加参数验证逻辑。
  2. 多步骤任务处理

    • 问题:复杂任务需多次函数调用(如“预订机票并安排接送”)。
    • 解决方案:结合 Agent 架构,让 LLM 自主规划多步骤流程(参考知识库条目[4])。
  3. 安全与权限控制

    • 问题:调用敏感函数(如支付 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 的实现过程可概括为以下步骤:

  1. 定义函数:明确函数功能、参数及返回值。
  2. 集成到 LLM:通过 API 将函数信息传递给模型。
  3. 解析用户请求:LLM 生成结构化调用指令。
  4. 执行函数调用:开发者代码解析指令并调用实际函数。
  5. 反馈结果:将结果返回 LLM,生成最终回答。
  6. 标准化与扩展:通过 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
      };
步骤 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;
        }
    • JSON-RPC
      • 基于JSON格式,适合简单场景。
      • 示例请求:
        {"jsonrpc": "2.0","method": "get_weather","params": {"city": "北京", "date": "2025-04-15"},"id": 1
        }
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远程调用天气查询服务。
步骤
  1. 定义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;
    }
  2. 服务端实现(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()
  3. 客户端调用(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)实现跨网络的函数调用。其关键步骤包括:

  1. 定义接口:明确函数参数和返回类型。
  2. 序列化参数:将数据转换为网络传输格式。
  3. 选择协议:根据需求选择HTTP、gRPC等协议。
  4. 处理响应:解析结果并集成到应用逻辑中。
  5. 安全与优化:确保数据安全和高效传输。

五、支持 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?

  1. 通过 API 接口
    • Qwen:使用阿里云 API 或魔搭 GPT 工具。
    • DeepSeek/Baichuan:调用其 OpenAI 兼容接口(如 api.deepseek.com/v1)。
  2. 框架集成
    • 迅策框架:通过其 API 动态调用远程资源。
    • LangChain:结合国产模型的 API 实现复杂流程(如自然语言到 SQL 转换)。

六、扩展

6.1 多个函数参数数量和名称不一致

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)")
  2. 在提示中明确可用函数

    • 在模型输入中列出所有可用函数及其参数,引导模型选择正确的函数。例如:
      "可用工具:\n"
      "1. add:计算两个整数之和(参数:a, b)\n"
      "2. get_weather:查询指定城市的天气(参数:city, date)\n"
      "请根据用户请求选择合适的工具并返回结构化参数。"
  3. 模型输出结构化指令

    • 要求模型返回包含 函数名称 和 参数字典 的 JSON 对象。例如:
      {"function_name": "add","arguments": {"a": 1, "b": 2}
      }
      {"function_name": "get_weather","arguments": {"city": "北京", "date": "2025-04-15"}
      }

2. 参数提取:如何从 LLM 产出中获取不同函数的参数?

核心思路

通过 函数名称匹配预定义的参数结构动态解析参数键值。

实现步骤
  1. 维护函数映射表

    • 将函数名称与对应的参数结构(如 Pydantic 模型)映射,例如:
      tool_mapping = {"add": AddFunction,"get_weather": GetWeatherFunction
      }
  2. 解析模型输出

    • 根据 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)}")
  3. 动态调用函数

    • 根据函数名称和参数调用对应的后端逻辑:
      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

关键点总结

  1. 函数选择

    • 显式标识函数名称:要求模型返回 function_name 字段。
    • 映射表关联:通过 tool_mapping 将函数名称与参数模型绑定。
  2. 参数提取

    • 类型安全验证:使用 Pydantic 模型确保参数类型和必填项正确。
    • 动态解析:根据函数名称动态选择参数模型,避免硬编码。
  3. 扩展性

    • 新增函数:只需添加新的 Pydantic 模型到 tool_mapping,无需修改核心逻辑。
    • 参数兼容性:即使参数名称不同,映射表和模型会自动适配。

常见问题处理

  • 参数缺失或类型错误
    • Pydantic 会自动抛出验证错误,可捕获异常并返回用户友好的提示。
  • 多函数调用
    • 对于需要多次函数调用的场景,可将中间结果存储在上下文中,供后续调用使用。
  • 函数名称冲突
    • 可通过命名规范(如 weather.get_weather)或分组(如按模块分组)避免冲突。

6.2 基于 LangChain + FastChat 框架实现 通义千问私有化部署Function Calling 具体案例

目标

用户输入“北京现在的天气如何?”,通过 FastChat 服务调用 LangChain Agent,结合本地天气工具函数返回实时天气信息。


1. 环境准备

1.1 安装依赖库
pip install langchain fastchat transformers
1.2 部署 FastChat 服务
  1. 下载通义千问模型(如 qwen-max)并解压到本地路径(假设路径为 ./qwen-model)。
  2. 启动 FastChat 服务(本地 REST API):
    fastchat-serve-api --model-path ./qwen-model --num-gpu 1 --num_cpu_threads_per_gpu 4
    • 这会启动一个本地 API 服务,默认监听 http://localhost:8000/v1

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 工作流程
  1. 用户输入解析:Agent 根据用户指令(如“北京天气”)分析是否需要调用工具。
  2. 工具调用:Agent 调用 get_current_weather 函数获取天气数据。
  3. 结果整合:将工具返回的 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. 注意事项

  1. FastChat 配置

    • 确保模型服务地址(http://localhost:8000/v1)正确。
    • 根据显存调整 num-gpu 和 num_cpu_threads_per_gpu
  2. LangChain Agent 类型

    • 可选其他 Agent 类型(如 AgentType.OPENAI_FUNCTIONS),需适配模型能力。
  3. 工具函数安全性

    • 避免直接执行用户输入(如 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 流程如下:

  1. 启动 FastChat 服务:本地部署通义千问模型并提供 REST API。
  2. 定义工具函数:实现本地工具(如天气查询)并包装为 LangChain 工具。
  3. 构建 Agent:使用 LangChain 的 Agent 调用模型和工具。
  4. 执行任务:通过 Agent 处理用户指令,自动调用工具并生成回答。

相关文章:

  • 京东商品详情API接口调用讲解(实战案例)
  • Day56 | 99. 恢复二叉搜索树、103. 二叉树的锯齿形层序遍历、109. 有序链表转换二叉搜索树、113. 路径总和 II
  • 使用openpyxl时的一些注意点
  • 1.2 腾讯校招通关指南-面试官评分标准:技术岗/产品岗核心考核点揭秘
  • 【自相关】全局 Moran’s I 指数
  • OPPO Android 移动设备日志文件目录结构及其内容分析
  • git合并分支原理
  • 适合stm32 前端adc使用的放大器芯片
  • jetson orin nano 开发板conda 的 base 环境在 shell 启动时自动激活
  • 如何解决服务器文件丢失或损坏的问题?
  • JAVA学习-Stream
  • Spring IoC与DI详解:从Bean概念到手写实现
  • Spring Batch 专题系列(四):配置与调度 Spring Batch 作业
  • 分库分表-除了hash分片还有别的吗?
  • 算法思想之分治-快排
  • Java基础 4.15
  • PCL八叉树聚类
  • Python基础语法2
  • 游戏代码编辑
  • 凸优化第2讲:凸优化建模
  • 新消费观察 | 重点深耕,外资科技企业继续看好中国发展
  • 涉嫌在饭局后性侵一女子,湖南机场董事长邱继兴被警方刑拘
  • 自然资源部一季度新批用海项目中,涉历史遗留围填海项目56个
  • 李公明︱一周书记:时代风雨中的……叛逆者在前行中
  • 东南亚三国行第三日|中马将在人工智能、大熊猫保护、铁路等多领域深化合作
  • 泽连斯基称乌美矿产协议谈判相关法律事宜已基本敲定