基于Qwen大模型，采用“ReAct“模式（Reason + Act）完成天气调用Function Calling

print(f"❌ 地名查询失败: {data.get('info')} (infocode: {data.get('infocode')})")f"风力：{w['windpower']}级，湿度：{w['humidity']}%，"return {"status": "error", "msg": data.get("info", "天气查询失败")}"content": f"城市：{w['ci

coding2008

799人浏览 · 2025-09-30 21:25:43

coding2008 · 2025-09-30 21:25:43 发布

基于Qwen3-max大模型实现，用户如要使用Qwen3大模型，需要注册阿里云百炼平台账户，获取DASHSCOPE_API_KEY。

本程序在Python中，使用DashScope调用Qwen3大模型的API接口。也可以使用OpenAI兼容接口。如使用DashScope，需要安装DashScope包。

从高德平台获取天气状况信息，需要注册高德平台，获取GAODE_API_KEY。

以下为实现的代码：

import os

import json

import requests

from dashscope import Generation

# ========================

# 1. 获取环境变量

# ========================

DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")

GAODE_API_KEY = os.getenv("GAODE_API_KEY")

if not DASHSCOPE_API_KEY:

raise EnvironmentError("请设置环境变量：DASHSCOPE_API_KEY")

if not GAODE_API_KEY:

raise EnvironmentError("请设置环境变量：GAODE_API_KEY")

# ========================

# 2. 利用高德adcode API实现用户输入的地名和adcode的转换，否则无法识别"广州市天河区"这种格式

# ========================

def geocode_location(location_name):

"""

根据地名（如“天河区”）查询对应的 adcode 和城市名

使用高德「输入提示 API」

"""

url = "https://restapi.amap.com/v3/assistant/inputtips"

params = {

"key": GAODE_API_KEY,

"keywords": location_name,

"city": "", # 可选：限定城市，如“广州”

"offset": 1,

"page": 1,

"output": "json"

}

try:

response = requests.get(url, params=params, timeout=10)

if response.status_code != 200:

print("❌ 网络请求失败")

return None, None

data = response.json()

if data.get("status") != "1":

print(f"❌ 地名查询失败: {data.get('info')} (infocode: {data.get('infocode')})")

return None, None

tips = data.get("tips", [])

if not tips:

print("⚠️ 未找到匹配的地点")

return None, None

# 取第一个最匹配的结果

place = tips[0]

adcode = place.get("adcode")

city_name = place.get("city") or place.get("name") # 优先取 city，否则取 name

return adcode, city_name

except requests.exceptions.RequestException as e:

print("❌ 请求异常:", e)

return None, None

# ========================

# 3. 高德天气 API 封装

# ========================

def get_real_weather(location: str) -> dict:

"""

调用高德地图天气 API 获取实时天气

文档：https://lbs.amap.com/api/webservice/guide/api/weatherinfo

"""

WEATHER_URL = "https://restapi.amap.com/v3/weather/weatherInfo"

adcode, city = geocode_location(location)

if adcode:

params = {

"key": GAODE_API_KEY,

"city": adcode, #使用高德的adcode代替location

"extensions": "base", # base: 实时天气；all: 预报

"output": "json"

}

try:

response = requests.get(WEATHER_URL, params=params, timeout=10)

data = response.json()

if data["status"] == "1" and data.get("lives"):

w = data["lives"][0]

return {

"status": "success",

"data": {

"city": w["city"],

"temperature": w["temperature"],

"weather": w["weather"],

"winddirection": w["winddirection"],

"windpower": w.get("windpower", "未知"),

"humidity": w["humidity"],

"report_time": w["reporttime"]

}

else:

return {"status": "error", "msg": data.get("info", "天气查询失败")}

except Exception as e:

return {"status": "error", "msg": str(e)}

# ========================

# 4. 主程序采用业内广泛采用的"ReAct"模式（Reason + Act），两次调用大模型

# ========================

def main():

print("🌤️ 欢迎使用通义千问 + 高德天气助手")

print("💡 输入地区名称，例如：广州市天河区、北京市海淀区")

user_input = input("\n📍 请输入要查询的地区：").strip()

if not user_input:

print("❌ 请输入有效地区名称。")

return

# 构建对话历史

messages = [

{

"role": "system",

"content": "你是一个天气助手。当用户查询天气时，如果要查询的地名为中国大陆的地名，你必须调用 get_real_weather 工具获取真实数据，禁止编造或猜测。如果要查询的地名根本不是中国大陆的地名，则直接告诉查询者没有这个地方。"

{

"role": "user",

"content": f"请查询 {user_input} 的实时天气"

}

]

# 定义工具

tools = [

{

"type": "function",

"function": {

"name": "get_real_weather",

"description": "获取指定城市的实时天气信息（温度、天气、风力等）",

"parameters": {

"type": "object",

"properties": {

"location": {

"type": "string",

"description": "城市或区域名称，如'北京市'、'广州市天河区'"

}

"required": ["location"]

}

]

try:

# 第一轮：调用模型，判断是否需要工具

response = Generation.call(

model="qwen-max",

messages=messages,

tools=tools,

tool_choice="auto", # 可改为强制调用

result_format="message", # 必须！才能正确返回 tool_calls

api_key=DASHSCOPE_API_KEY

)

if response.status_code != 200:

print(f"❌ 大模型调用失败：{response.message}")

return

output = response.output # 返回的是完整结构

# ========================

# 解析 tool_calls（嵌套在 choices 中）

# ========================

if (

isinstance(output, dict) and

'choices' in output and

len(output['choices']) > 0 and

'message' in output['choices'][0] and

'tool_calls' in output['choices'][0]['message']

#解析tool_calls，需要调用外部工具

tool_calls = output['choices'][0]['message']['tool_calls']

if tool_calls and tool_calls[0]['function']['name'] == 'get_real_weather':

tool_call = tool_calls[0]

func_name = tool_call['function']['name']

args_str = tool_call['function']['arguments']

# 解析 arguments（是 JSON 字符串）

try:

args = json.loads(args_str)

except Exception as e:

print(f"❌ 参数解析失败：{e}")

args = {}

# location = args.get('location', user_input)

location = args.get('location')

print(f"\n🔍 正在通过高德查询 {location} 现在的天气状况...")

weather_result = get_real_weather(location)

if weather_result["status"] == "success":

w = weather_result["data"]

# 构造 tool 响应

tool_response = {

"role": "tool",

"content": f"城市：{w['city']}，天气：{w['weather']}，"

f"温度：{w['temperature']}°C，风向：{w['winddirection']}，"

f"风力：{w['windpower']}级，湿度：{w['humidity']}%，"

f"更新时间：{w['report_time']}",

"tool_call_id": tool_call["id"] # 必须匹配

}

# 把模型的回复和 tool 响应都加入对话

messages.append(output['choices'][0]['message']) # 模型的 tool_calls

messages.append(tool_response)

# 第二轮：让模型生成自然语言回复

final_response = Generation.call(

model="qwen-max",

messages=messages,

result_format="message",

api_key=DASHSCOPE_API_KEY

)

if final_response.status_code == 200:

final_msg = final_response.output

content = ""

if isinstance(final_msg, dict):

content = final_response["output"]["choices"][0]["message"]["content"]

if content:

print(f"\n🌤️ 天气信息：\n{content}")

else:

print("❌ 模型未生成有效回复。")

else:

print(f"❌ 最终回复生成失败：{final_response.message}")

else:

print(f"❌ 天气查询失败：{weather_result['msg']}")

else:

print("⚠️ 未调用 get_real_weather 函数。")

else:

# 没有 tool_calls，直接输出模型文本

content = ""

if isinstance(output, dict):

if 'text' in output and output['text']:

content = output['text'].strip()

elif 'choices' in output and len(output['choices']) > 0:

msg = output['choices'][0]['message'].get('content', '')

content = msg.strip()

if content:

print(f"\n🤖 回复：\n{content}")

else:

print("❌ 模型未返回有效内容。")

except Exception as e:

print(f"❌ 程序异常：{str(e)}")

# ========================

#5. 启动入口

# ========================

if __name__ == "__main__":

main()

以上代码，在Python3.13.5版本运行正常。欢迎技术交流

北京朝阳AI社区

更多推荐

深度学习前沿技术全景解析 —— 设计思想、源码剖析与业务落地

迁移学习与预训练模型通过知识迁移，大幅提升新任务表现。生成对抗网络实现高质量数据生成，激发创新应用。强化学习智能体自适应环境，推动自动决策技术落地。大模型与多模态学习突破单一信息壁垒，赋能复杂场景理解与生成。迁移换头，冻结微调；真假对抗，交替优化；Q表迭代，奖励驱动；编码融合，智能生成。善用预训练，巧用Prompt，关注资源优化，监控训练曲线。结合主流框架、部署工具，实现模型高效落地。未来趋势：更

北京朝阳AI社区

基于全阿里RAG技术栈，实现企业内部PDF格式规章制度内容的查询

本文介绍了一个基于阿里云技术栈的企业规章制度查询系统实现方案。系统采用PyPDF2提取PDF文档内容并进行切分，使用阿里云text-embedding-v1工具生成文本向量，通过Chroma_DB向量数据库存储和检索文档。系统核心功能包括：PDF内容提取与分段、远程Embedding处理、向量数据库存储、相似内容检索，以及调用Qwen-max大模型生成回答。该方案实现了对企业规章制度的智能问答功能