二、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）定义远程函数的参数、返回类型和功能。
  Json

  深色版本

  // 示例：天气查询函数的接口定义
  {"name": "get_weather","parameters": {"city": {"type": "string"},"date": {"type": "string", "format": "date"}},"returns": {"type": "object", "properties": {"temperature": "number"}}
  }

步骤 2：序列化参数与构建请求

• 参数序列化：将模型生成的参数（如JSON对象）转换为网络传输格式（如JSON、Protobuf）。
  Python

  深色版本

  # 示例：模型输出的参数
  arguments = {"city": "北京", "date": "2025-04-15"}
  serialized_data = json.dumps(arguments)  # 转换为JSON字符串

步骤 3：选择通信协议并发送请求

• 协议选择：
  • HTTP/REST：简单易用，适合无状态服务。
    Python

    深色版本

    import requests
    response = requests.post("http://weather-service/api/get_weather",json=arguments
    )

  • gRPC：高性能、二进制传输，适合分布式系统。
    Proto

    深色版本

    // 定义proto文件
    service WeatherService {rpc GetWeather (WeatherRequest) returns (WeatherResponse);
    }
    message WeatherRequest {string city = 1;string date = 2;
    }

  • JSON-RPC：轻量级，支持异步调用。
    Javascript

    深色版本

    const request = {"jsonrpc": "2.0","method": "get_weather","params": arguments,"id": 1
    };

步骤 4：远程服务端处理请求

• 反序列化参数：将网络数据转换为本地对象。
  Python

  深色版本

  # 示例：服务端接收请求
  @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：返回结果与错误处理

• 结果序列化：将结果转换为网络传输格式。
  Python

  深色版本

  result = {"temperature": 25}
  return jsonify(result)

• 错误处理：捕获异常并返回标准化错误信息（如HTTP状态码、RPC错误码）。
  Python

  深色版本

  except Exception as e:return jsonify({"error": str(e)}), 500

步骤 6：客户端处理响应

• 解析结果：将返回的数据反序列化并集成到应用逻辑中。
  Python

  深色版本

  if response.status_code == 200:temperature = response.json()["temperature"]# 生成最终回答
  else:# 处理错误（如提示用户重试）

---

3. 关键技术与协议

1. RPC（远程过程调用）

• 原理：允许客户端像调用本地函数一样调用远程服务，隐藏底层网络细节。
• 实现方式：
  • gRPC（推荐）：
    • 使用 Protocol Buffers 定义接口（.proto 文件）。
    • 支持高性能二进制传输和流式处理。
    • 示例（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格式，适合简单场景。
    • 示例请求：
      Json

      深色版本

      {"jsonrpc": "2.0","method": "get_weather","params": {"city": "北京", "date": "2025-04-15"},"id": 1
      }

2. REST API

• 特点：
  • 使用HTTP动词（GET/POST/PUT/DELETE）操作资源。
  • 参数通过URL、查询参数或请求体传递。
• 示例：
  Http

  深色版本

  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）。
  • 内置错误处理和上下文管理。
• 示例：
  Json

  深色版本

  {"method": "call_function","params": {"function_name": "get_weather","arguments": {"city": "北京"}}
  }

---

4. 安全与优化

安全措施

• 身份验证：使用API Key、OAuth 2.0或JWT。
• 加密传输：通过HTTPS或gRPC的TLS加密数据。
• 参数校验：验证参数格式和类型（如使用JSON Schema）。
  Python

  深色版本

  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）：

   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）：

   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）：

   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 处理用户指令，自动调用工具并生成回答。