函数调用

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]
)