GLM-4工具调用开发：Function Call对接企业API实战-程序员充电站

GLM-4工具调用开发：Function Call对接企业API实战

你是不是也遇到过这样的场景？老板让你把大模型的能力集成到公司的CRM系统里，让销售团队能直接用自然语言查询客户信息。你兴冲冲地接下了任务，结果发现要对接企业内部的API，各种权限验证、错误处理、数据格式转换……一堆问题摆在面前。

别担心，今天我就来分享一个实战案例：如何用GLM-4的Function Call功能，轻松对接企业API，实现智能化的客户信息查询。整个过程比你想象的要简单得多。

1. 为什么选择GLM-4的Function Call？

先说说为什么我推荐用GLM-4来做这件事。GLM-4-9B-Chat这个模型有个很实用的功能——自定义工具调用，也就是Function Call。简单来说，就是让大模型能够理解你的指令，然后调用你定义好的函数去执行具体任务。

这功能听起来可能有点抽象，我举个例子你就明白了。比如你对模型说：“帮我查一下客户张三最近的订单情况。”模型不会直接去数据库里查，而是会识别出你需要调用“查询客户订单”这个函数，然后把“张三”作为参数传进去。剩下的就是你的代码去执行实际的查询了。

GLM-4在这方面做得特别友好，它支持OpenAI兼容的Function Calling接口，这意味着你可以用很标准的方式定义工具，模型也能很好地理解你的意图。而且GLM-4-9B-Chat-1M版本还支持128K的超长上下文，处理复杂的对话场景完全没问题。

2. 企业API对接的核心挑战

在真正开始写代码之前，我们先来看看对接企业API通常会遇到哪些问题。我总结了一下，主要有这么几个：

权限验证问题：企业API往往需要各种认证，比如API Key、OAuth Token、或者更复杂的签名验证。这些认证逻辑怎么集成到Function Call里？

数据格式转换：企业API返回的数据格式五花八门，有的是XML，有的是JSON，有的甚至是自定义格式。怎么把这些数据转换成大模型能理解的格式？

错误处理机制：网络超时、服务器错误、权限不足……各种异常情况怎么处理？总不能用户一问就报错吧。

安全性考虑：哪些API能暴露给大模型调用？参数怎么校验？会不会有SQL注入之类的安全问题？

性能优化：API调用通常比较慢，怎么设计缓存和重试机制，让用户体验更好？

这些问题听起来挺头疼的，但别急，我们一步步来解决。

3. 实战：CRM客户查询系统

下面我就用一个实际的CRM客户查询系统为例，带你走完整个开发流程。我们会从最简单的场景开始，逐步完善功能。

3.1 环境准备与模型部署

首先，你需要一个运行起来的GLM-4模型。如果你已经有现成的服务，可以直接用。如果没有，我推荐用vLLM来部署，这样性能会好很多。

# 安装必要的库 pip install vllm transformers # 启动GLM-4服务 python -m vllm.entrypoints.openai.api_server \ --model THUDM/glm-4-9b-chat-1m \ --served-model-name glm-4 \ --port 8000 \ --trust-remote-code

启动成功后，你会得到一个兼容OpenAI API的服务端点，地址是http://localhost:8000/v1。后面我们就用这个地址来调用模型。

3.2 定义你的第一个工具函数

工具函数的定义其实很简单，就是用JSON Schema来描述这个函数能做什么、需要什么参数。我们从一个最简单的客户查询开始：

# 先定义工具的描述 customer_query_tool = { "type": "function", "function": { "name": "query_customer_info", "description": "根据客户姓名或ID查询客户基本信息", "parameters": { "type": "object", "properties": { "customer_name": { "type": "string", "description": "客户的姓名，支持模糊匹配" }, "customer_id": { "type": "string", "description": "客户的唯一标识符" } }, "required": [], "anyOf": [ {"required": ["customer_name"]}, {"required": ["customer_id"]} ] } } }

这里有几个关键点需要注意：

description要写得清晰明确，模型会根据这个描述来决定什么时候调用这个函数
parameters里定义了函数需要的参数，每个参数都要有类型和描述
required和anyOf用来定义参数约束，这里的意思是至少要提供姓名或ID中的一个

3.3 实现实际的API调用

工具定义好了，接下来就是实现具体的函数逻辑。这里我假设你们公司有个CRM系统的API，地址是https://api.your-company.com/crm/v1/customers。

import requests import json from typing import Optional, Dict, Any class CRMClient: def __init__(self, api_key: str, base_url: str = "https://api.your-company.com/crm/v1"): self.api_key = api_key self.base_url = base_url self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }) def query_customer_info(self, customer_name: Optional[str] = None, customer_id: Optional[str] = None) -> Dict[str, Any]: """ 查询客户信息 """ # 参数校验 if not customer_name and not customer_id: return {"error": "必须提供客户姓名或ID"} # 构建查询参数 params = {} if customer_name: params["name"] = customer_name if customer_id: params["id"] = customer_id try: # 调用CRM API response = self.session.get( f"{self.base_url}/customers", params=params, timeout=10 # 10秒超时 ) # 处理响应 if response.status_code == 200: data = response.json() return { "success": True, "data": data.get("customers", []), "count": len(data.get("customers", [])) } elif response.status_code == 404: return {"error": "未找到匹配的客户"} elif response.status_code == 401: return {"error": "权限验证失败，请检查API Key"} else: return {"error": f"API调用失败: {response.status_code}"} except requests.exceptions.Timeout: return {"error": "请求超时，请稍后重试"} except requests.exceptions.RequestException as e: return {"error": f"网络错误: {str(e)}"} except json.JSONDecodeError: return {"error": "API返回的数据格式错误"}

这个类封装了CRM API的调用逻辑，包括权限验证、错误处理、超时控制等。实际使用中，你可能还需要根据你们公司的具体API文档调整参数和响应处理。

3.4 让GLM-4调用你的工具

现在工具和实现都有了，怎么让GLM-4来调用呢？看下面的代码：

import openai from typing import List, Dict, Any class GLM4FunctionCaller: def __init__(self, api_base: str = "http://localhost:8000/v1", api_key: str = "dummy"): self.client = openai.OpenAI( api_key=api_key, base_url=api_base ) self.crm_client = CRMClient(api_key="your-crm-api-key") self.tools = [customer_query_tool] # 可以添加更多工具 def process_query(self, user_query: str) -> str: """ 处理用户查询，自动调用合适的工具 """ # 第一步：让模型决定是否需要调用工具 response = self.client.chat.completions.create( model="glm-4", messages=[{"role": "user", "content": user_query}], tools=self.tools, tool_choice="auto" # 让模型自动决定 ) message = response.choices[0].message # 第二步：如果模型决定调用工具，执行对应的函数 if message.tool_calls: tool_call = message.tool_calls[0] function_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) # 根据函数名调用对应的实现 if function_name == "query_customer_info": result = self.crm_client.query_customer_info( customer_name=arguments.get("customer_name"), customer_id=arguments.get("customer_id") ) # 第三步：把结果返回给模型，让模型生成最终回复 second_response = self.client.chat.completions.create( model="glm-4", messages=[ {"role": "user", "content": user_query}, {"role": "assistant", "content": None, "tool_calls": [tool_call]}, {"role": "tool", "content": json.dumps(result), "tool_call_id": tool_call.id} ] ) return second_response.choices[0].message.content else: return "抱歉，暂时不支持这个功能" else: # 如果不需要调用工具，直接返回模型的回复 return message.content def run_conversation(self): """ 运行一个简单的对话示例 """ print("CRM智能助手已启动，输入'退出'结束对话") print("-" * 50) while True: user_input = input("\n你: ").strip() if user_input.lower() in ["退出", "exit", "quit"]: print("再见！") break try: response = self.process_query(user_input) print(f"助手: {response}") except Exception as e: print(f"助手: 出错了: {str(e)}")

这个类的核心逻辑分三步：

把用户的问题和工具定义一起发给模型，让模型决定要不要调用工具
如果模型说要调用，就解析出要调用的函数和参数，执行真正的业务逻辑
把执行结果再发给模型，让模型生成最终的自然语言回复

3.5 试试效果

我们来运行一下看看效果：

if __name__ == "__main__": caller = GLM4FunctionCaller() # 测试几个查询 test_queries = [ "帮我查一下客户张三的信息", "ID是CUST001的客户最近有下单吗？", "我想了解李四这个客户的情况", "今天天气怎么样" # 这个不需要调用工具 ] for query in test_queries: print(f"\n用户: {query}") response = caller.process_query(query) print(f"助手: {response}")

运行后你会看到，对于前三个查询，模型会自动调用query_customer_info函数，然后根据API返回的结果生成回复。对于最后一个天气查询，模型发现没有合适的工具，就会直接用自己的知识来回答。

4. 处理复杂场景：OpenAPI规范转换

刚才的例子比较简单，实际工作中你可能需要对接很多不同的API。如果每个API都要手动写工具定义和调用代码，工作量就太大了。这时候可以用OpenAPI规范来简化工作。

很多企业的API都有OpenAPI/Swagger文档，我们可以利用这些文档自动生成工具定义。下面是一个简单的转换器：

import yaml import json from typing import List, Dict, Any class OpenAPIToTools: def __init__(self, openapi_spec: Dict[str, Any]): self.spec = openapi_spec def convert_endpoint_to_tool(self, path: str, method: str, operation: Dict[str, Any]) -> Dict[str, Any]: """ 将OpenAPI的一个端点转换为工具定义 """ # 提取基本信息 operation_id = operation.get("operationId", f"{method}_{path.replace('/', '_')}") description = operation.get("summary", "") + "\n" + operation.get("description", "") # 构建参数定义 parameters = {} required_params = [] # 处理路径参数 for param in operation.get("parameters", []): if param.get("in") == "path": param_name = param["name"] parameters[param_name] = { "type": param.get("schema", {}).get("type", "string"), "description": param.get("description", "") } if param.get("required", False): required_params.append(param_name) # 处理查询参数 for param in operation.get("parameters", []): if param.get("in") == "query": param_name = param["name"] parameters[param_name] = { "type": param.get("schema", {}).get("type", "string"), "description": param.get("description", "") } if param.get("required", False): required_params.append(param_name) # 处理请求体 if "requestBody" in operation: content = operation["requestBody"].get("content", {}) if "application/json" in content: schema = content["application/json"].get("schema", {}) # 这里可以递归处理复杂的JSON Schema # 为了简化，我们只处理简单属性 if "properties" in schema: for prop_name, prop_schema in schema["properties"].items(): parameters[prop_name] = { "type": prop_schema.get("type", "string"), "description": prop_schema.get("description", "") } required_params.extend(schema.get("required", [])) # 构建工具定义 tool = { "type": "function", "function": { "name": operation_id, "description": description.strip(), "parameters": { "type": "object", "properties": parameters, "required": required_params } } } return tool def convert_all_endpoints(self) -> List[Dict[str, Any]]: """ 转换所有端点 """ tools = [] for path, methods in self.spec.get("paths", {}).items(): for method, operation in methods.items(): if method.lower() in ["get", "post", "put", "delete"]: tool = self.convert_endpoint_to_tool(path, method, operation) tools.append(tool) return tools # 使用示例 def load_openapi_and_convert(openapi_file: str): with open(openapi_file, 'r', encoding='utf-8') as f: if openapi_file.endswith('.yaml') or openapi_file.endswith('.yml'): spec = yaml.safe_load(f) else: spec = json.load(f) converter = OpenAPIToTools(spec) tools = converter.convert_all_endpoints() print(f"从OpenAPI规范中生成了{len(tools)}个工具") for tool in tools[:3]: # 只显示前3个 print(json.dumps(tool, indent=2, ensure_ascii=False)) return tools

这个转换器能自动从OpenAPI文档中提取API信息，生成对应的工具定义。虽然它处理不了特别复杂的Schema，但对于大多数RESTful API来说已经够用了。

5. 错误处理与重试机制

企业环境下的API调用，错误处理特别重要。下面我设计了一个带重试和降级的调用器：

import time from functools import wraps from typing import Callable, Optional, Any import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class APICallWithRetry: def __init__(self, max_retries: int = 3, retry_delay: float = 1.0): self.max_retries = max_retries self.retry_delay = retry_delay def __call__(self, func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(self.max_retries): try: result = func(*args, **kwargs) # 检查结果是否包含错误 if isinstance(result, dict) and "error" in result: error_msg = result["error"] # 如果是暂时性错误，重试 if any(keyword in error_msg.lower() for keyword in ["timeout", "busy", "temporarily"]): if attempt < self.max_retries - 1: wait_time = self.retry_delay * (2 ** attempt) # 指数退避 logger.warning(f"API调用暂时失败: {error_msg}, 第{attempt+1}次重试，等待{wait_time}秒") time.sleep(wait_time) continue # 如果是权限错误，不重试 if "permission" in error_msg.lower() or "auth" in error_msg.lower(): return {"error": "权限验证失败，请检查API配置"} return result except requests.exceptions.Timeout as e: last_exception = e if attempt < self.max_retries - 1: wait_time = self.retry_delay * (2 ** attempt) logger.warning(f"请求超时，第{attempt+1}次重试，等待{wait_time}秒") time.sleep(wait_time) continue except requests.exceptions.ConnectionError as e: last_exception = e if attempt < self.max_retries - 1: wait_time = self.retry_delay * (2 ** attempt) logger.warning(f"连接错误，第{attempt+1}次重试，等待{wait_time}秒") time.sleep(wait_time) continue except Exception as e: last_exception = e logger.error(f"API调用异常: {str(e)}") break # 所有重试都失败了 if last_exception: return {"error": f"API调用失败: {str(last_exception)}"} else: return {"error": "API调用失败，未知错误"} return wrapper # 使用装饰器增强CRM客户端 class EnhancedCRMClient(CRMClient): @APICallWithRetry(max_retries=3, retry_delay=1.0) def query_customer_info(self, customer_name: Optional[str] = None, customer_id: Optional[str] = None) -> Dict[str, Any]: # 原有的实现... return super().query_customer_info(customer_name, customer_id) @APICallWithRetry(max_retries=2, retry_delay=2.0) def get_customer_orders(self, customer_id: str, limit: int = 10) -> Dict[str, Any]: """ 获取客户订单（新增功能） """ try: response = self.session.get( f"{self.base_url}/customers/{customer_id}/orders", params={"limit": limit}, timeout=15 ) if response.status_code == 200: data = response.json() return { "success": True, "orders": data.get("orders", []), "total": data.get("total", 0) } else: return {"error": f"获取订单失败: {response.status_code}"} except Exception as e: return {"error": f"获取订单异常: {str(e)}"}

这个重试机制有几个特点：

指数退避：每次重试等待时间加倍，避免给服务器太大压力
智能重试：只对暂时性错误（如超时、繁忙）重试，权限错误等立即返回
错误分类：根据错误类型采取不同策略
日志记录：详细记录每次重试的情况，方便排查问题

6. 完整案例：智能CRM助手

现在我们把所有功能整合起来，做一个完整的智能CRM助手：

class SmartCRMAssistant: def __init__(self, glm4_api_base: str, crm_api_key: str): self.glm4_caller = GLM4FunctionCaller(api_base=glm4_api_base) self.crm_client = EnhancedCRMClient(api_key=crm_api_key) # 定义完整的工具集 self.tools = [ { "type": "function", "function": { "name": "query_customer_info", "description": "查询客户基本信息，包括联系方式、公司、职位等", "parameters": { "type": "object", "properties": { "customer_name": {"type": "string", "description": "客户姓名"}, "customer_id": {"type": "string", "description": "客户ID"}, "company": {"type": "string", "description": "客户公司名称"} } } } }, { "type": "function", "function": { "name": "get_customer_orders", "description": "获取客户的订单历史", "parameters": { "type": "object", "properties": { "customer_id": {"type": "string", "description": "客户ID", "required": True}, "start_date": {"type": "string", "description": "开始日期，格式YYYY-MM-DD"}, "end_date": {"type": "string", "description": "结束日期，格式YYYY-MM-DD"}, "limit": {"type": "integer", "description": "返回的最大订单数", "default": 10} }, "required": ["customer_id"] } } }, { "type": "function", "function": { "name": "search_customers_by_criteria", "description": "根据多种条件搜索客户", "parameters": { "type": "object", "properties": { "industry": {"type": "string", "description": "行业"}, "region": {"type": "string", "description": "地区"}, "status": {"type": "string", "description": "客户状态，如活跃、潜在、流失等"}, "min_value": {"type": "number", "description": "最小交易金额"}, "max_value": {"type": "number", "description": "最大交易金额"} } } } } ] def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """ 执行具体的工具函数 """ if tool_name == "query_customer_info": return self.crm_client.query_customer_info( customer_name=arguments.get("customer_name"), customer_id=arguments.get("customer_id"), company=arguments.get("company") ) elif tool_name == "get_customer_orders": return self.crm_client.get_customer_orders( customer_id=arguments["customer_id"], start_date=arguments.get("start_date"), end_date=arguments.get("end_date"), limit=arguments.get("limit", 10) ) elif tool_name == "search_customers_by_criteria": # 这里实现搜索逻辑 return {"success": True, "message": "搜索功能待实现"} else: return {"error": f"未知工具: {tool_name}"} def chat(self, user_input: str, conversation_history: list = None) -> str: """ 处理用户输入，支持多轮对话 """ if conversation_history is None: conversation_history = [] # 添加用户输入到历史 conversation_history.append({"role": "user", "content": user_input}) # 调用GLM-4 response = self.glm4_caller.client.chat.completions.create( model="glm-4", messages=conversation_history, tools=self.tools, tool_choice="auto" ) message = response.choices[0].message conversation_history.append(message.to_dict()) # 处理工具调用 if message.tool_calls: all_results = [] for tool_call in message.tool_calls: tool_name = tool_call.function.name arguments = json.loads(tool_call.function.arguments) # 执行工具 result = self.execute_tool(tool_name, arguments) all_results.append(result) # 添加工具响应到历史 conversation_history.append({ "role": "tool", "content": json.dumps(result), "tool_call_id": tool_call.id }) # 获取模型的最终回复 final_response = self.glm4_caller.client.chat.completions.create( model="glm-4", messages=conversation_history ) final_message = final_response.choices[0].message conversation_history.append(final_message.to_dict()) return final_message.content else: return message.content def interactive_chat(self): """ 交互式聊天界面 """ print("=" * 60) print("智能CRM助手 v1.0") print("支持功能：查询客户信息、查看订单历史、条件搜索") print("输入'退出'结束对话") print("=" * 60) history = [] while True: try: user_input = input("\n你: ").strip() if user_input.lower() in ["退出", "exit", "quit", "bye"]: print("助手: 再见！期待下次为您服务。") break if not user_input: continue # 处理查询 response = self.chat(user_input, history) print(f"助手: {response}") except KeyboardInterrupt: print("\n\n对话已中断") break except Exception as e: print(f"助手: 抱歉，出了点问题: {str(e)}") # 可以选择清空历史或保留 # history = [] # 使用示例 if __name__ == "__main__": # 配置你的API信息 GLM4_API_BASE = "http://localhost:8000/v1" CRM_API_KEY = "your-actual-crm-api-key" assistant = SmartCRMAssistant(GLM4_API_BASE, CRM_API_KEY) assistant.interactive_chat()

这个完整的助手支持：

多工具调用：可以同时查询客户信息和订单历史
多轮对话：保持对话上下文，理解指代关系
错误处理：内置重试和降级机制
友好交互：简单的命令行界面

7. 实际使用中的注意事项

在实际部署和使用过程中，有几个点需要特别注意：

权限控制：不是所有用户都能调用所有API。你需要在工具调用前检查用户权限，比如：

def check_permission(user_id: str, tool_name: str) -> bool: # 这里实现你的权限检查逻辑 # 可以基于角色、部门、数据范围等 pass

输入验证：模型生成的参数需要验证，防止注入攻击：

def validate_customer_id(customer_id: str) -> bool: # 验证ID格式 import re return bool(re.match(r'^CUST\d{6}$', customer_id))

性能优化：频繁查询的数据可以加缓存：

from functools import lru_cache import time class CachedCRMClient(EnhancedCRMClient): @lru_cache(maxsize=100) def query_customer_info(self, customer_name=None, customer_id=None, company=None): # 原有的查询逻辑 pass def get_customer_orders(self, customer_id, start_date=None, end_date=None, limit=10): # 订单数据变化频繁，缓存时间短一些 cache_key = f"orders_{customer_id}_{start_date}_{end_date}_{limit}" if cache_key in self.order_cache: cached_data, timestamp = self.order_cache[cache_key] if time.time() - timestamp < 300: # 5分钟缓存 return cached_data result = super().get_customer_orders(customer_id, start_date, end_date, limit) self.order_cache[cache_key] = (result, time.time()) return result

监控和日志：记录每次工具调用的详细信息，方便排查问题：

import logging from datetime import datetime class MonitoredGLM4FunctionCaller(GLMFunctionCaller): def process_query(self, user_query: str) -> str: start_time = datetime.now() user_id = get_current_user_id() # 获取当前用户 logger.info(f"[{start_time}] 用户{user_id}查询: {user_query}") try: result = super().process_query(user_query) end_time = datetime.now() duration = (end_time - start_time).total_seconds() logger.info(f"[{end_time}] 查询完成，耗时{duration:.2f}秒") return result except Exception as e: logger.error(f"查询失败: {str(e)}", exc_info=True) return "抱歉，系统暂时无法处理您的请求，请稍后再试。"