实战：本地大模型+function Calling，获取北京天气

什么是function Calling？

Function Calling‌ 是大型语言模型（LLM）连接外部工具和系统的核心技术，通过结构化参数生成实现模型与真实世界交互的能力。‌

为什么有function calling?

‌弥补大模型时效性与实时性不足‌：大模型训练数据存在时间滞后性（如训练时间是2023年，但是2023年后的数据没有），通过Function Calling可动态调用外部API（如天气、股价接口）获取最新信息‌

本地大模型

ollama运行qwen3:8b-q4_K_M模型，具体可参考macOS Ollama本地部署deepseek-r1:1.5b-CSDN博客

整体代码

import requests
import json
from http import HTTPStatus# 高德天气 API 的 天气工具定义（JSON 格式）
weather_tool = {"type": "function","function": {"name": "get_current_weather","description": "Get the current weather in a given location","parameters": {"type": "object","properties": {"location": {"type": "string","description": "The city name, e.g. 北京",},"adcode": {"type": "string","description": "The city code, e.g. 110000 (北京)",}},"required": ["location"],},},
}def get_weather_from_gaode(location: str, adcode: str = None):"""调用高德地图API查询天气"""gaode_api_key = "gaodeAPIkey"base_url = "https://restapi.amap.com/v3/weather/weatherInfo"params = {"key": gaode_api_key,"city": adcode if adcode else location,"extensions": "base",  # 可改为 "all" 获取预报}response = requests.get(base_url, params=params)if response.status_code == 200:return response.json()else:return {"error": f"Failed to fetch weather: {response.status_code}"}def call_local_model(messages, tools=None):"""调用本地模型"""# 本地模型的 API 地址（根据您的部署调整）local_api_url = "http://localhost:11434/v1/chat/completions"# 构建请求数据payload = {"model": "qwen3:8b-q4_K_M",  # 根据您部署的模型名称调整"messages": messages,"temperature": 0.1,"stream": False}# 如果有工具定义，添加到请求中if tools:payload["tools"] = toolspayload["tool_choice"] = "auto"try:response = requests.post(local_api_url,headers={"Content-Type": "application/json"},json=payload,timeout=120)if response.status_code == 200:return response.json()else:print(f"本地模型调用失败: {response.status_code} - {response.text}")return Noneexcept Exception as e:print(f"调用本地模型时出错: {str(e)}")return Nonedef run_weather_query():"""使用本地模型 + 查询天气，并让大模型输出最终结果"""messages = [{"role": "system", "content": "你是一个智能助手，可以查询天气信息。"},{"role": "user", "content": "北京现在天气怎么样？"}]print("第一次调用本地模型...")response = call_local_model(messages, tools=[weather_tool])if response and "choices" in response:tool_map = {"get_current_weather": get_weather_from_gaode,# 如有更多工具，在此添加}# 从响应中获取消息assistant_message = response["choices"][0]["message"]print(f"模型响应: {assistant_message}")# 检查是否需要调用工具if "tool_calls" in assistant_message and assistant_message["tool_calls"]:print("检测到工具调用...")# 生成工具调用回复消息tool_response_messages = []for tool_call in assistant_message["tool_calls"]:print(f"处理工具调用: {tool_call['function']['name']}, ID: {tool_call['id']}")func_name = tool_call["function"]["name"]try:func_args = json.loads(tool_call["function"]["arguments"])except json.JSONDecodeError as e:print(f"解析工具参数失败: {e}")continueif func_name in tool_map:# 调用工具函数try:result = tool_map[func_name](**func_args)print(f"工具调用结果: {result}")# 创建工具回复消息tool_response = {"role": "tool","tool_call_id": tool_call["id"],"name": func_name,"content": json.dumps(result, ensure_ascii=False)}tool_response_messages.append(tool_response)except Exception as e:print(f"工具调用出错: {str(e)}")tool_response = {"role": "tool","tool_call_id": tool_call["id"],"name": func_name,"content": json.dumps({"error": str(e)}, ensure_ascii=False)}tool_response_messages.append(tool_response)else:print(f"未知的工具: {func_name}")# 组装完整消息列表updated_messages = messages + [assistant_message] + tool_response_messagesprint(f"更新后的消息列表: {updated_messages}")print("第二次调用本地模型...")response2 = call_local_model(updated_messages, tools=[weather_tool])if response2 and "choices" in response2:final_response = response2["choices"][0]["message"]["content"]print("最终回复:", final_response)else:print("第二次调用本地模型失败")else:# 如果没有调用工具，直接输出模型回复content = assistant_message.get("content", "")if content:print("无工具调用，直接输出回复:", content)else:print("模型没有返回有效内容")else:print("第一次调用本地模型失败")if __name__ == "__main__":run_weather_query()

结果