Assistant API 支持函数调用(Function Calling),让智能体能够根据您的需求自动调用外部函数。例如,智能体可以调用函数查询天气或安排日程。本文介绍了一个简单的“天气查询智能体”示例,帮助您快速上手函数调用的基本方法。
快速开始
在这个示例中,我们将创建一个天气查询智能体,并创建一个函数get_weather
作为智能体可以调用的工具。在这个示例中,我们将询问智能体“杭州今天的天气状况”。
在您开始前
首先,确保您已经安装了必要的依赖库,如 requests 和 dashscope。您可以通过以下命令安装它们:
pip install requests dashscope
其次,您需要参考高德天气API以获取高德天气查询的API-KEY。
第一步:创建“获取天气信息”函数
我们首先使用了高德地图的API来获取城市的天气信息。
import requests
def get_weather(city):
"""
获取指定城市的实时天气信息。
这个函数使用高德地图API来获取指定城市的天气数据。首先,它通过城市名称获取城市编码,
然后使用该编码查询实时天气信息。
参数:
city (str): 要查询天气的城市名称,例如 "杭州"、"北京" 等。
返回:
str: 包含城市天气信息的字符串。如果成功,返回格式为
"{城市}的天气:温度 {温度}°C,天气 {天气状况},风向 {风向},风力 {风力}级"。
如果查询失败,返回相应的错误信息。
注意:
- 此函数依赖于有效的高德地图API密钥。
- 如果无法找到指定的城市或获取天气信息失败,会返回相应的错误消息。
"""
# 请替换为你的高德API密钥
amap_key = "YOUR_AMAP_API_KEY"
# 获取城市编码
city_url = f"https://restapi.amap.com/v3/config/district?keywords={city}&subdistrict=0&key={amap_key}"
city_response = requests.get(city_url)
city_data = city_response.json()
if city_data['count'] == '0':
return f"无法找到城市: {city}"
adcode = city_data['districts'][0]['adcode']
# 获取天气信息
weather_url = f"https://restapi.amap.com/v3/weather/weatherInfo?city={adcode}&key={amap_key}"
weather_response = requests.get(weather_url)
weather_data = weather_response.json()
if weather_data['status'] == '1' and weather_data['lives']:
live_weather = weather_data['lives'][0]
return f"{city}的天气:温度 {live_weather['temperature']}°C,天气 {live_weather['weather']},风向 {live_weather['winddirection']},风力 {live_weather['windpower']}级"
else:
return f"无法获取{city}的天气信息"
解释:
获取城市编码:通过请求高德API来查询城市的行政编码(adcode),这是获取天气信息的必要参数。
获取天气信息:使用城市的 adcode 来获取该城市的当前天气。
现在我们将通过 Assistant API 来创建一个智能体,该智能体将自动处理用户查询,并调用我们定义的 get_weather 函数来提供天气信息。
第二步:描述“获取天气信息”函数
您需要向智能体描述“获取天气信息”函数。智能体会根据描述信息正确调用天气查询函数。
from dashscope import Assistants, Messages, Runs, Threads
import json
# 定义天气查询工具
weather_tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的当前天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "要查询天气的城市名称"
}
},
"required": ["city"]
}
}
}
解释:
name: 函数的名称为 get_weather,将在智能体中被调用。
description: 提供了该工具的描述,帮助用户理解其用途。
parameters: 定义了函数的参数,这里是城市名称 city。
第三步:创建智能体
现在我们可以创建一个 Assistant 实例,它是一个智能体,并且将会使用我们定义的天气查询工具。
# 创建 Assistant
assistant = Assistants.create(
model='qwen-plus',
name='天气查询智能体',
description='一个智能体,可以查询天气信息并解答用户的问题',
instructions='你是一个智能体,你的任务是解答用户的问题,特别是关于天气的查询。当用户询问天气时,使用 get_weather 函数获取实时天气信息。',
tools=[weather_tool]
)
解释:
model: 使用的模型类型,这里是 qwen-plus,它支持语言理解和任务处理。
name: 给智能体命名为“天气查询智能体”。
description: 描述了智能体的功能,它能够帮助用户查询天气。
tools: 注册了我们之前定义的 weather_tool,使得智能体可以调用该工具。
第四步:创建对话线程并与智能体互动
创建一个新的对话线程,并向其中添加用户消息,然后运行智能体,处理用户的问题。
# 创建一个新的线程
thread = Threads.create()
# 添加用户消息到线程
Messages.create(thread_id=thread.id, role="user", content="杭州今年中秋节的天气怎么样?")
# 运行 Assistant
run = Runs.create(thread_id=thread.id, assistant_id=assistant.id)
# 等待运行完成
run = Runs.wait(thread_id=thread.id, run_id=run.id)
解释:
Threads.create(): 创建一个新的对话线程,后续的消息将在此线程中进行。
Messages.create(): 向线程添加用户消息,这里用户询问了“杭州今年中秋节的天气怎么样?”
Runs.create(): 触发智能体开始处理用户消息。
Runs.wait(): 等待智能体处理完毕。
第五步:处理函数调用并返回结果
如果智能体在处理过程中需要调用工具,我们将调用 get_weather 函数获取天气信息,并将结果返回给用户。
# 检查是否需要调用函数
if run.required_action:
for tool_call in run.required_action.submit_tool_outputs.tool_calls:
if tool_call.function.name == "get_weather":
city = json.loads(tool_call.function.arguments)["city"]
weather_info = get_weather(city)
# 提交工具输出
Runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=[{"tool_call_id": tool_call.id, "output": weather_info}]
)
# 等待新的运行完成
run = Runs.wait(thread_id=thread.id, run_id=run.id)
解释:
检查函数调用:如果智能体需要调用某个函数,我们检查它调用的是否是 get_weather,并通过之前定义的函数获取天气信息。
提交结果:将结果通过 Runs.submit_tool_outputs 返回给智能体,并继续等待智能体的下一步回复。
第六步:获取智能体的回复
智能体处理完成后,我们可以从对话线程中获取智能体的回复并展示给用户。
# 获取 Assistant 的回复
messages = Messages.list(thread_id=thread.id)
for message in messages.data:
if message.role == "assistant":
print(f"Assistant: {message.content}")
总结
通过以上步骤,您已经成功创建了一个智能体,它可以处理用户的天气查询请求,并调用高德API来获取实时天气信息。Assistant API 使得构建复杂的任务驱动型智能体变得简单且高效。
您可以根据需求进一步扩展智能体的功能,例如增加更多工具或修改智能体的行为指令。
快速生成业务函数的描述信息
在快速入门案例中,您需要向智能体描述“查询天气”函数,而这一过程比较繁琐。因此,我们提供了一个简单的转换函数,帮助您快速描述业务函数。
import inspect
def function_to_schema(func) -> dict:
# 将 Python 类型映射为 JSON schema 类型
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
type(None): "null",
}
# 尝试获取函数的签名
try:
signature = inspect.signature(func)
except ValueError as e:
# 如果签名获取失败,则抛出错误并附带错误信息
raise ValueError(
f"Failed to get signature for function {func.__name__}: {str(e)}"
)
# 初始化一个字典来存储参数类型
parameters = {}
# 遍历函数的参数,并映射它们的类型
for param in signature.parameters.values():
try:
param_type = type_map.get(param.annotation, "string")
except KeyError as e:
# 如果参数的类型注解未知,则抛出错误
raise KeyError(
f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
)
parameters[param.name] = {"type": param_type}
# 创建必需参数的列表(那些没有默认值的参数)
required = [
param.name
for param in signature.parameters.values()
if param.default == inspect._empty
]
# 返回函数的 schema 作为字典
return {
"type": "function",
"function": {
"name": func.__name__,
"description": (func.__doc__ or "").strip(), # 获取函数描述(docstring)
"parameters": {
"type": "object",
"properties": parameters, # 参数类型
"required": required, # 必需参数列表
},
},
}
以快速开始的天气查询函数为例:
weather_tool = function_to_schema(get_weather)
print(json.dumps(weather_tool, indent=2))
查询天气函数将被自动转换为:
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的实时天气信息。\n\n 这个函数使用高德地图API来获取指定城市的天气数据。首先,它通过城市名称获取城市编码,\n 然后使用该编码查询实时天气信息。\n\n Parameters:\n city (str): 要查询天气的城市名称,例如 \"杭州\"、\"北京\" 等。\n\n Returns:\n str: 包含城市天气信息的字符串。如果成功,返回格式为 \n \"{城市}的天气:温度 {温度}°C,天气 {天气状况},风向 {风向},风力 {风力}级\"。\n 如果查询失败,返回相应的错误信息。\n\n 注意:\n - 此函数依赖于有效的高德地图API密钥。\n - 如果无法找到指定的城市或获取天气信息失败,会返回相应的错误消息。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string"
}
},
"required": [
"city"
]
}
}
}
现在,我们可以将函数的描述信息传递给模型。
assistant = Assistants.create(
model='qwen-plus',
name='天气查询助手',
description='一个智能助手,可以查询天气信息并解答用户的问题',
instructions='你是一个助手,你的任务是解答用户的问题,特别是关于天气的查询。当用户询问天气时,使用 get_weather 函数获取实时天气信息。',
tools=[weather_tool]
)