通过API使用通义千问

1. 获取API-KEY

开通百炼服务并获得API-KEY

登录阿里云百炼大模型服务平台。
鼠标悬停于页面右上角的个人中心（人性标志）上，再在下拉菜单中单击API-KEY。
在左侧导航栏，选择全部API-KEY或我的API-KEY。（说明：仅主账号拥有查看全部API-KEY的权限。）
在待获取的API-KEY的操作列单击查看，再单击复制，将API-KEY保存在安全的地方。

API-KEY的权限范围说明:

-主账号的API-KEY可以访问所有业务空间下的服务（模型部署、模型调用、模型训练等），但访问指定业务空间时，必须传入对应业务空间的“业务空间ID”。调用指定业务空间模型的具体操作，请参见子业务空间的模型调用。

- 子账号的API-KEY仅可以访问API-KEY绑定的业务空间下已被授权的服务（模型部署、模型调用、模型训练等）。主账号授权业务空间的具体操作，请参见授权子业务空间模型调用和部署。

- 如果子账号想要访问“默认业务空间”（即主账号的业务空间）下的服务（模型部署、模型调用、模型训练等），有2种解决方案：

	-获取主账号API-KEY来进行调用。

	-主账号先给子账号授权“默认业务空间”的权限，然后子账号在“默认业务空间”下创建自己的API-KEY来访问“默认业务空间”下的服务。

2. 将API-KEY配置到环境变量

2.1 linux

当使用Linux系统（如Ubuntu、CentOS等）中的命令行添加DashScope的API-KEY为环境变量时，可以选择在当前会话添加临时性环境变量，或对当前用户添加永久性环境变量。

2.1.1 添加临时性环境变量

如果仅想在当前会话中添加并使用临时性环境变量，可以运行以下命令：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

可以在当前会话运行以下命令检查环境变量是否生效：

echo $DASHSCOPE_API_KEY

2.1.2 对当前用户添加永久性环境变量

如果想对当前用户添加永久性环境变量，使得在该用户的新会话中也可以使用该环境变量，可以把以下命令语句复制并添加到~/.bashrc文件中：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

或直接运行以下命令将上述命令语句添加到~/.bashrc中：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
echo "export DASHSCOPE_API_KEY='YOUR_DASHSCOPE_API_KEY'" >> ~/.bashrc

添加完成后，可以运行以下命令使得环境变量生效：

source ~/.bashrc

可以新建立一个会话，运行以下命令检查环境变量是否生效：

echo $DASHSCOPE_API_KEY

2.2 macOS

2.2.1 添加临时性环境变量

如果仅想在当前会话中添加并使用临时性环境变量，可以运行以下命令：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

可以在当前会话运行以下命令检查环境变量是否生效：

echo $DASHSCOPE_API_KEY

2.2.2 对当前用户添加永久性环境变量

如果想对当前用户添加永久性环境变量，使得在该用户的新会话中也可以使用该环境变量，可以把以下命令语句复制并添加到~/.zshrc或~/.bash_profile中：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

或直接运行以下命令将上述命令语句添加到~/.bashrc中：

# 用您的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
## 如果您的Shell类型是Zsh，运行以下命令
echo "export DASHSCOPE_API_KEY='YOUR_DASHSCOPE_API_KEY'" >> ~/.zshrc

## 如果您的Shell类型是Bash，运行以下命令
echo "export DASHSCOPE_API_KEY='YOUR_DASHSCOPE_API_KEY'" >> ~/.bash_profile

添加完成后，可以运行以下命令使得环境变量生效：

# 如果您的Shell类型是Zsh，运行以下命令
source ~/.zshrc

# 如果您的Shell类型是Bash，运行以下命令
source ~/.bash_profile

可以新建立一个会话，运行以下命令检查环境变量是否生效：

echo $DASHSCOPE_API_KEY

2.3 windows (CMD or PowerShell(推荐))

2.3.1 CMD

2.3.1.1 添加临时性环境变量

# 用的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
set DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY"

在当前会话运行以下命令检查环境变量是否生效：

echo %DASHSCOPE_API_KEY%

2.3.1.2 对当前用户添加永久性环境变量

# 用 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
setx DASHSCOPE_API_KEY "YOUR_DASHSCOPE_API_KEY"

重新打开一个CMD窗口，运行以下命令检查环境变量是否生效：

echo %DASHSCOPE_API_KEY%

2.3.2 PowerShell

2.3.2.1 添加临时性环境变量

# 用的 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
$env:DASHSCOPE_API_KEY = "YOUR_DASHSCOPE_API_KEY"

在当前会话运行以下命令检查环境变量是否生效：

echo $env:DASHSCOPE_API_KEY

2.3.2.2 对当前用户添加永久性环境变量

# 用 DashScope API-KEY 代替 YOUR_DASHSCOPE_API_KEY
[Environment]::SetEnvironmentVariable("DASHSCOPE_API_KEY", "YOUR_DASHSCOPE_API_KEY", [EnvironmentVariableTarget]::User)

重新打开一个CMD窗口，运行以下命令检查环境变量是否生效：

echo $env:DASHSCOPE_API_KEY

3 安装SDK

3.1 Python

已安装Python3.8及以上版本。
执行以下命令，通过pip安装SDK

pip install dashscope

如需更新SDK包，请执行以下命令。

pip install dashscope --upgrade

Java SDK install: https://help.aliyun.com/zh/model-studio/developer-reference/install-sdk?spm=a2c4g.11186623.0.0.2d6a212b7ecVpy

5. 快速开始

前提条件
1, DashScope SDK提供了Python和Java两个版本，请确保您已安装最新版SDK：安装SDK。您也可以通过HTTP接口调用，详情请见：通过API使用通义千问。

2, 已开通服务并获得API-KEY：获取API-KEY。

3,将API-KEY配置到环境变量中以降低API-KEY的泄漏风险，详情可参考通过环境变量配置API-KEY。也可以在代码中配置API-KEY，但是泄漏风险会提高。

可以通过两种方式来调用模型：通过messages调用或通过prompt调用。如果有多轮对话的需求，更推荐您通过messages调用，多轮对话的调用方式请见：通过API使用通义千问。

5.1 通过messages调用（推荐）

from http import HTTPStatus
import dashscope


def call_with_messages():
    messages = [{'role': 'system', 'content': 'You are a helpful assistant.'},
                {'role': 'user', 'content': '请介绍一下通义千问'}]

    response = dashscope.Generation.call(
        model="qwen-turbo",
        messages=messages,
        result_format='message',  # 将返回结果格式设置为 message
    )
    if response.status_code == HTTPStatus.OK:
        print(response)
    else:
        print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
            response.request_id, response.status_code,
            response.code, response.message
        ))


if __name__ == '__main__':
    call_with_messages()

调用成功后将输出以下结果。

{
    "status_code": 200,
    "request_id": "a75a1b22-e512-957d-891b-37db858ae738",
    "code": "",
    "message": "",
    "output": {
        "text": null,
        "finish_reason": null,
        "choices": [
            {
                "finish_reason": "stop",
                "message": {
                    "role": "assistant",
                    "content": "通义千问是阿里云自主研发的超大规模语言模型，能够回答问题、创作文字，还能表达观点、撰写代码。作为一个大型预训练语言模型，我能够根据您提出的指令产出相关的回复，并尽可能提供准确和有用的信息。我会不断学习和进步，不断提升自己的能力，为用户提供更好的服务。如果您有任何问题或需要帮助，请随时告诉我。"
                }
            }
        ]
    },
    "usage": {
        "input_tokens": 25,
        "output_tokens": 77,
        "total_tokens": 102
    }
}

5.2 通过prompt调用

通过prompt方式调用时，输入与输出的数据格式比messages方式简单，更适合单轮问答等简单场景；在更复杂的场景中推荐您通过messages方式调用。

from http import HTTPStatus
import dashscope


def call_with_prompt():
    response = dashscope.Generation.call(
        model="qwen-turbo",
        prompt='请介绍一下通义千问'
    )
    # 如果调用成功，则打印模型的输出
    if response.status_code == HTTPStatus.OK:
        print(response)
    # 如果调用失败，则打印出错误码与失败信息
    else:
        print(response.code)  
        print(response.message) 

if __name__ == '__main__':
    call_with_prompt()

调用成功后将输出以下结果。

{
    "status_code": 200,
    "request_id": "b9b5767a-7b94-9df1-93bd-f3cf97e2acaa",
    "code": "",
    "message": "",
    "output": {
        "text": "你好，我是阿里云自主研发的超大规模语言模型“通义千问”。作为一个AI助手，我的目标是帮助用户获得准确、有用的信息，解决他们的问题和困惑。我能够回答各种主题的问题，提供知识性解释，进行创造性的思考，并与用户进行对话。我会不断学习和进步，不断提升自己的能力，为用户提供更好的服务。如果您有任何问题，请随时向我提问。",
        "finish_reason": "stop",
        "choices": null
    },
    "usage": {
        "input_tokens": 25,
        "output_tokens": 84,
        "total_tokens": 109
    }
}

6. 通过API使用通义千问

6.1 场景示例

6.1.1 单轮对话

OpenAI兼容

from openai import OpenAI
import os

def get_response():
    client = OpenAI(
        api_key=os.getenv("DASHSCOPE_API_KEY"), # 如果您没有配置环境变量，请在此处用您的API Key进行替换
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 填写DashScope服务的base_url
    )
    completion = client.chat.completions.create(
        model="qwen-turbo",
        messages=[
            {'role': 'system', 'content': 'You are a helpful assistant.'},
            {'role': 'user', 'content': '你是谁？'}]
        )
    print(completion.model_dump_json())

if __name__ == '__main__':
    get_response()

返回结果

{
  "id": "chatcmpl-ee338a7c-b5b3-9139-a726-b7b749d6b49d",
  "choices": [
    {
      "finish_reason": "stop",
      "index": 0,
      "logprobs": null,
      "message": {
        "content": "我是阿里云开发的一款超大规模语言模型，我叫通义千问。",
        "refusal": null,
        "role": "assistant",
        "function_call": null,
        "tool_calls": null
      }
    }
  ],
  "created": 1725005215,
  "model": "qwen-turbo",
  "object": "chat.completion",
  "service_tier": null,
  "system_fingerprint": null,
  "usage": {
    "completion_tokens": 17,
    "prompt_tokens": 22,
    "total_tokens": 39
  }
}

DashScope

import random
from http import HTTPStatus
# 建议dashscope SDK 的版本 >= 1.14.0
from dashscope import Generation


def call_with_messages():
    messages = [{'role': 'system', 'content': 'You are a helpful assistant.'},
                {'role': 'user', 'content': '你是谁？'}]
    response = Generation.call(model="qwen-turbo",
                               messages=messages,
                               # 将输出设置为"message"格式
                               result_format='message')
    if response.status_code == HTTPStatus.OK:
        print(response)
    else:
        print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
            response.request_id, response.status_code,
            response.code, response.message
        ))


if __name__ == '__main__':
    call_with_messages()

返回结果

{
  "status_code": 200,
  "request_id": "902fee3b-f7f0-9a8c-96a1-6b4ea25af114",
  "code": "",
  "message": "",
  "output": {
    "text": null,
    "finish_reason": null,
    "choices": [
      {
        "finish_reason": "stop",
        "message": {
          "role": "assistant",
          "content": "我是阿里云开发的一款超大规模语言模型，我叫通义千问。"
        }
      }
    ]
  },
  "usage": {
    "input_tokens": 22,
    "output_tokens": 17,
    "total_tokens": 39
  }
}

6.1.2 多轮对话

相比于单轮对话，多轮对话可以让大模型参考历史对话信息，更符合日常交流的场景。实现多轮对话的关键在于维护一个存放历史对话信息的列表，并将更新后的列表作为大模型的输入，从而使大模型可以参考历史对话信息进行回复。可以参考以下示例代码，将每一轮的对话历史添加到messages列表中，实现多轮对话的功能。

OpenAI兼容

from openai import OpenAI
import os

def get_response(messages):
    client = OpenAI(
        # 如果您没有配置环境变量，请在此处用您的API Key进行替换
        api_key=os.getenv("DASHSCOPE_API_KEY"), 
        # 填写DashScope服务的base_url
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    completion = client.chat.completions.create(
        model="qwen-turbo",
        messages=messages
        )
    return completion

messages = [{'role': 'system', 'content': 'You are a helpful assistant.'}]
# 您可以自定义设置对话轮数，当前为3
for i in range(3):
    user_input = input("请输入：")
    # 将用户问题信息添加到messages列表中
    messages.append({'role': 'user', 'content': user_input})
    assistant_output = get_response(messages).choices[0].message.content
    # 将大模型的回复信息添加到messages列表中
    messages.append({'role': 'assistant', 'content': assistant_output})
    print(f'用户输入：{user_input}')
    print(f'模型输出：{assistant_output}')
    print('\n')

返回结果
在这里插入图片描述

DashScope

from dashscope import Generation

def get_response(messages):
    response = Generation.call(model="qwen-turbo",
                               messages=messages,
                               # 将输出设置为"message"格式
                               result_format='message')
    return response


messages = [{'role': 'system', 'content': 'You are a helpful assistant.'}]

# 您可以自定义设置对话轮数，当前为3
for i in range(3):
    user_input = input("请输入：")
    # 将用户问题信息添加到messages列表中
    messages.append({'role': 'user', 'content': user_input})
    assistant_output = get_response(messages).output.choices[0]['message']['content']
    # 将大模型的回复信息添加到messages列表中
    messages.append({'role': 'assistant', 'content': assistant_output})
    print(f'用户输入：{user_input}')
    print(f'模型输出：{assistant_output}')
    print('\n')

6.1.3 流式输出

大模型并不是一次性生成最终结果，而是逐步地生成中间结果，最终结果由中间结果拼接而成。使用非流式输出方式需要等待模型生成结束后再将生成的中间结果拼接后返回，而流式输出可以实时地将中间结果返回，可以在模型进行输出的同时进行阅读，减少等待模型回复的时间。

OpenAI兼容

from openai import OpenAI
import os

def get_response():
    client = OpenAI(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    completion = client.chat.completions.create(
        model="qwen-turbo",
        messages=[{'role': 'system', 'content': 'You are a helpful assistant.'},
                  {'role': 'user', 'content': '你是谁？'}],
        stream=True,
        # 可选，配置以后会在流式输出的最后一行展示token使用信息
        stream_options={"include_usage": True}
        )
    for chunk in completion:
        print(chunk.model_dump_json())

if __name__ == '__main__':
    get_response()

返回结果

{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"","function_call":null,"role":"assistant","tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"我是","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"通","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"义","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"千问，由阿里","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"云开发的AI助手。我被","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"设计用来回答各种问题、提供信息","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"和与用户进行对话。有什么我可以","function_call":null,"role":null,"tool_calls":null},"finish_reason":null,"index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[{"delta":{"content":"帮助你的吗？","function_call":null,"role":null,"tool_calls":null},"finish_reason":"stop","index":0,"logprobs":null}],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":null}
{"id":"chatcmpl-xxx","choices":[],"created":1721099636,"model":"qwen-turbo","object":"chat.completion.chunk","service_tier":null,"system_fingerprint":null,"usage":{"completion_tokens":36,"prompt_tokens":22,"total_tokens":58}}

DashScope

from http import HTTPStatus
from dashscope import Generation


def call_with_stream():
    messages = [
        {'role':'system','content':'you are a helpful assistant'},
        {'role': 'user','content': '你是谁？'}
        ]
    responses = Generation.call(
        model="qwen-turbo",
        messages=messages,
        # 设置输出为'message'格式
        result_format='message',
        # 设置输出方式为流式输出
        stream=True,
        # 增量式流式输出
        incremental_output=True
        )
    full_content = ""
    for response in responses:
        if response.status_code == HTTPStatus.OK:
            print(response)
            full_content += response.output.choices[0].message.content
        else:
            print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
                response.request_id, response.status_code,
                response.code, response.message
            ))
    print(f"Full content:{full_content}")

if __name__ == '__main__':
    call_with_stream()

6.1.4 Function Call（调用外部工具）

大模型在面对实时性问题、私域知识型问题或数学计算等问题时可能效果不佳。可以使用function call功能，通过调用外部工具来提升模型的输出效果。可以在调用大模型时，通过tools参数传入工具的名称、描述、入参等信息。Function Call的工作流程示意图如下所示：
在这里插入图片描述
Function Call的使用涉及到参数解析功能，因此对大模型的响应质量要求较高，推荐优先使用qwen-max模型。

OpenAI兼容

from openai import OpenAI
from datetime import datetime
import json
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"), # 如果您没有配置环境变量，请在此处用您的API Key进行替换
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 填写DashScope SDK的base_url
)

# 定义工具列表，模型在选择使用哪个工具时会参考工具的name和description
tools = [
    # 工具1 获取当前时刻的时间
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "当你想知道现在的时间时非常有用。",
            "parameters": {}  # 因为获取当前时间无需输入参数，因此parameters为空字典
        }
    },  
    # 工具2 获取指定城市的天气
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {  # 查询天气时需要提供位置，因此参数设置为location
                        "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。"
                    }
                }
            },
            "required": [
                "location"
            ]
        }
    }
]

# 模拟天气查询工具。返回结果示例：“北京今天是晴天。”
def get_current_weather(location):
    return f"{location}今天是雨天。 "

# 查询当前时间的工具。返回结果示例：“当前时间：2024-04-15 17:15:18。“
def get_current_time():
    # 获取当前日期和时间
    current_datetime = datetime.now()
    # 格式化当前日期和时间
    formatted_time = current_datetime.strftime('%Y-%m-%d %H:%M:%S')
    # 返回格式化后的当前时间
    return f"当前时间：{formatted_time}。"

# 封装模型响应函数
def get_response(messages):
    completion = client.chat.completions.create(
        model="qwen-max",
        messages=messages,
        tools=tools
        )
    return completion.model_dump()

def call_with_messages():
    print('\n')
    messages = [
            {
                "content": input('请输入：'),  # 提问示例："现在几点了？" "一个小时后几点" "北京天气如何？"
                "role": "user"
            }
    ]
    print("-"*60)
    # 模型的第一轮调用
    i = 1
    first_response = get_response(messages)
    assistant_output = first_response['choices'][0]['message']
    print(f"\n第{i}轮大模型输出信息：{first_response}\n")
    if  assistant_output['content'] is None:
        assistant_output['content'] = ""
    messages.append(assistant_output)
    # 如果不需要调用工具，则直接返回最终答案
    if assistant_output['tool_calls'] == None:  # 如果模型判断无需调用工具，则将assistant的回复直接打印出来，无需进行模型的第二轮调用
        print(f"无需调用工具，我可以直接回复：{assistant_output['content']}")
        return
    # 如果需要调用工具，则进行模型的多轮调用，直到模型判断无需调用工具
    while assistant_output['tool_calls'] != None:
        # 如果判断需要调用查询天气工具，则运行查询天气工具
        if assistant_output['tool_calls'][0]['function']['name'] == 'get_current_weather':
            tool_info = {"name": "get_current_weather", "role":"tool"}
            # 提取位置参数信息
            location = json.loads(assistant_output['tool_calls'][0]['function']['arguments'])['properties']['location']
            tool_info['content'] = get_current_weather(location)
        # 如果判断需要调用查询时间工具，则运行查询时间工具
        elif assistant_output['tool_calls'][0]['function']['name'] == 'get_current_time':
            tool_info = {"name": "get_current_time", "role":"tool"}
            tool_info['content'] = get_current_time()
        print(f"工具输出信息：{tool_info['content']}\n")
        print("-"*60)
        messages.append(tool_info)
        assistant_output = get_response(messages)['choices'][0]['message']
        if  assistant_output['content'] is None:
            assistant_output['content'] = ""
        messages.append(assistant_output)
        i += 1
        print(f"第{i}轮大模型输出信息：{assistant_output}\n")
    print(f"最终答案：{assistant_output['content']}")

if __name__ == '__main__':
    call_with_messages()

返回结果

当输入：几点了？时，程序会进行如下输出：
在这里插入图片描述

DashScope

from dashscope import Generation
from datetime import datetime
import random
import json


# 定义工具列表，模型在选择使用哪个工具时会参考工具的name和description
tools = [
    # 工具1 获取当前时刻的时间
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "当你想知道现在的时间时非常有用。",
            "parameters": {}  # 因为获取当前时间无需输入参数，因此parameters为空字典
        }
    },  
    # 工具2 获取指定城市的天气
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {  # 查询天气时需要提供位置，因此参数设置为location
                        "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。"
                    }
                }
            },
            "required": [
                "location"
            ]
        }
    }
]

# 模拟天气查询工具。返回结果示例：“北京今天是晴天。”
def get_current_weather(location):
    return f"{location}今天是晴天。 "

# 查询当前时间的工具。返回结果示例：“当前时间：2024-04-15 17:15:18。“
def get_current_time():
    # 获取当前日期和时间
    current_datetime = datetime.now()
    # 格式化当前日期和时间
    formatted_time = current_datetime.strftime('%Y-%m-%d %H:%M:%S')
    # 返回格式化后的当前时间
    return f"当前时间：{formatted_time}。"

# 封装模型响应函数
def get_response(messages):
    response = Generation.call(
        model='qwen-max',
        messages=messages,
        tools=tools,
        seed=random.randint(1, 10000),  # 设置随机数种子seed，如果没有设置，则随机数种子默认为1234
        result_format='message'  # 将输出设置为message形式
    )
    return response

def call_with_messages():
    print('\n')
    messages = [
            {
                "content": input('请输入：'),  # 提问示例："现在几点了？" "一个小时后几点" "北京天气如何？"
                "role": "user"
            }
    ]
    
    # 模型的第一轮调用
    first_response = get_response(messages)
    assistant_output = first_response.output.choices[0].message
    print(f"\n大模型第一轮输出信息：{first_response}\n")
    messages.append(assistant_output)
    if 'tool_calls' not in assistant_output:  # 如果模型判断无需调用工具，则将assistant的回复直接打印出来，无需进行模型的第二轮调用
        print(f"最终答案：{assistant_output.content}")
        return
    # 如果模型选择的工具是get_current_weather
    elif assistant_output.tool_calls[0]['function']['name'] == 'get_current_weather':
        tool_info = {"name": "get_current_weather", "role":"tool"}
        location = json.loads(assistant_output.tool_calls[0]['function']['arguments'])['properties']['location']
        tool_info['content'] = get_current_weather(location)
    # 如果模型选择的工具是get_current_time
    elif assistant_output.tool_calls[0]['function']['name'] == 'get_current_time':
        tool_info = {"name": "get_current_time", "role":"tool"}
        tool_info['content'] = get_current_time()
    print(f"工具输出信息：{tool_info['content']}\n")
    messages.append(tool_info)

    # 模型的第二轮调用，对工具的输出进行总结
    second_response = get_response(messages)
    print(f"大模型第二轮输出信息：{second_response}\n")
    print(f"最终答案：{second_response.output.choices[0].message['content']}")

if __name__ == '__main__':
    call_with_messages()

以下是发起Function Call流程（模型的第一轮调用）时模型的返回信息。当输入“杭州天气”时，模型会返回tool_calls参数；当输入“你好”时，模型判断无需调用工具，模型不会返回tool_calls参数。
输入:“杭州天气”

{
    "status_code": 200,
    "request_id": "bd803417-56a7-9597-9d3f-a998a35b0477",
    "code": "",
    "message": "",
    "output": {
        "text": null,
        "finish_reason": null,
        "choices": [
            {
                "finish_reason": "tool_calls",
                "message": {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [
                        {
                            "function": {
                                "name": "get_current_weather",
                                "arguments": "{\"properties\": {\"location\": \"杭州市\"}, \"type\": \"object\"}"
                            },
                            "id": "",
                            "type": "function"
                        }
                    ]
                }
            }
        ]
    },
    "usage": {
        "input_tokens": 222,
        "output_tokens": 27,
        "total_tokens": 249
    }
}

输入:“你好”

{
    "status_code": 200,
    "request_id": "28e9d70c-c4d7-9bfb-bd07-8cf4228dda91",
    "code": "",
    "message": "",
    "output": {
        "text": null,
        "finish_reason": null,
        "choices": [
            {
                "finish_reason": "stop",
                "message": {
                    "role": "assistant",
                    "content": "你好！有什么能帮到你的吗？如果有关于天气、时间或者其他问题，随时告诉我。"
                }
            }
        ]
    },
    "usage": {
        "input_tokens": 221,
        "output_tokens": 21,
        "total_tokens": 242
    }
}

6.1.5 Asyncio接口（实现异步调用）

OpenAI SDK

import os
import asyncio
from openai import AsyncOpenAI
import platform

# 创建异步客户端实例
client = AsyncOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# 定义异步任务列表
async def task(question):
    print(f"Sending question: {question}")
    response = await client.chat.completions.create(
        messages=[
            {"role": "user", "content": question}
        ],
        model="qwen-max",
    )
    print(f"Received answer: {response.choices[0].message.content}")

# 主异步函数
async def main():
    questions = ["你是谁？", "你会什么？", "天气怎么样？"]
    tasks = [task(q) for q in questions]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    # 设置事件循环策略
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # 运行主协程
    asyncio.run(main(), debug=False)

DashScope SDK

Dashscope Python SDK版本需要不低于 1.19.0。

import asyncio
import platform
from dashscope.aigc.generation import AioGeneration

# 定义异步任务列表
async def task(question):
    print(f"Sending question: {question}")
    response = await AioGeneration.call("qwen-turbo",
                                        prompt=question)
    print(f"Received answer: {response.output.text}")

# 主异步函数
async def main():
    questions = ["你是谁？", "你会什么？", "天气怎么样？"]
    tasks = [task(q) for q in questions]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    # 设置事件循环策略
    if platform.system() == 'Windows':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    # 运行主协程
    asyncio.run(main(), debug=False)