Alibaba Cloud Model Studio：函數調用（下線中）

Assistant API 支援函數調用（Function Calling），讓智能體能夠根據您的需求自動調用外部函數。例如，智能體可以調用函數進行文本翻譯或處理其他任務。本文介紹了一個簡單的"翻譯助手"樣本，協助您快速上手函數調用的基本方法。

快速開始

在這個樣本中，我們將建立一個翻譯助手，並建立一個函數translate_text作為智能體可以調用的工具。在這個樣本中，我們將詢問智能體將"Hello world"翻譯成中文。

在您開始前

首先，確保您已經安裝了必要的依賴庫，如 requests 和 dashscope。您可以通過以下命令安裝它們：

pip install requests dashscope

第一步：建立"翻譯文本"函數

我們首先建立一個簡單的翻譯函數，用於示範目的，它使用預定義的翻譯對照表。

def translate_text(text, target_language):
    """
    將文本翻譯成指定的目標語言。
    這是一個使用預定義翻譯的簡單示範。

    參數：
        text (str): 需要翻譯的文本
        target_language (str): 目標語言代碼（例如：'zh'、'es'、'ja'）

    返回：
        str: 翻譯後的文本或錯誤資訊
    """
    # 用於示範的翻譯字典
    mock_translations = {
        ('Hello world', 'zh'): '你好世界',
        ('Hello world', 'es'): '¡Hola Mundo!',
        ('Hello world', 'ja'): 'こんにちは世界',
        ('How are you?', 'zh'): '你好嗎？',
        ('How are you?', 'es'): '¿Cómo estás?',
        ('How are you?', 'ja'): 'お元気ですか？'
    }

    try:
        return mock_translations.get((text, target_language),
            f"未找到翻譯。在實際生產環境中，這裡會調用翻譯服務。")
    except Exception as e:
        return f"翻譯失敗：{str(e)}"

解釋：

• 翻譯功能：通過預定義的翻譯對照表來類比翻譯功能，支援多種語言之間的轉換。
• 錯誤處理：包含了基本的錯誤處理機制，確保函數在各種情況下都能返回合適的響應。

現在我們將通過 Assistant API 來建立一個智能體，該智能體將自動處理使用者查詢，並調用我們定義的 translate_text 函數來提供翻譯服務。

第二步：描述"翻譯文本"函數

您需要向智能體描述"翻譯文本"函數。智能體會根據描述資訊正確調用翻譯函數。

from dashscope import Assistants, Messages, Runs, Threads
import json
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
# 定義翻譯工具
translation_tool = {
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": "將文本翻譯成指定的目標語言",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "需要翻譯的文本"
                },
                "target_language": {
                    "type": "string",
                    "description": "目標語言代碼（例如：'zh'、'es'、'ja'）"
                }
            },
            "required": ["text", "target_language"]
        }
    }
}

解釋：

• name: 函數的名稱為 translate_text，將在智能體中被調用。
• description: 提供了該工具的描述，協助使用者理解其用途。
• parameters: 定義了函數的參數，包括要翻譯的文本和目標語言。

第三步：建立智能體

現在我們可以建立一個 Assistant 執行個體，它是一個智能體，並且將會使用我們定義的翻譯工具。

# 建立 Assistant
assistant = Assistants.create(
    model='qwen-plus',
    name='翻譯助手',
    description='一個能夠在不同語言之間進行文本翻譯的助手',
    instructions='你是一個翻譯助手。當使用者請求翻譯時，使用 translate_text 函數來協助他們。',
    tools=[translation_tool]
)

解釋：

• model: 使用的模型類型，這裡是 qwen-plus，它支援語言理解和任務處理。
• name: 給智能體命名為"Translation Assistant"。
• description: 描述了智能體的功能，它能夠協助使用者進行文本翻譯。
• tools: 註冊了我們之前定義的 translation_tool，使得智能體可以調用該工具。

第四步：建立對話線程並與智能體互動

建立一個新的對話線程，並向其中添加使用者訊息，然後運行智能體，處理使用者的問題。

# 建立一個新的線程
thread = Threads.create()

# 添加使用者訊息到線程
Messages.create(
    thread_id=thread.id,
    role="user",
    content="請將'Hello world'翻譯成中文。"
)

# 運行 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(): 向線程添加使用者訊息，這裡使用者請求將"Hello world"翻譯成中文。
• Runs.create(): 觸發智能體開始處理使用者訊息。
• Runs.wait(): 等待智能體處理完畢。

第五步：處理函數調用並返回結果

如果智能體在處理過程中需要調用工具，我們將調用 translate_text 函數進行翻譯，並將結果返回給使用者。

# 檢查是否需要調用函數
if run.required_action:
    for tool_call in run.required_action.submit_tool_outputs.tool_calls:
        if tool_call.function.name == "translate_text":
            args = json.loads(tool_call.function.arguments)
            translation = translate_text(args["text"], args["target_language"])

            # 提交工具輸出
            Runs.submit_tool_outputs(
                thread_id=thread.id,
                run_id=run.id,
                tool_outputs=[{"tool_call_id": tool_call.id, "output": translation}]
            )

            # 等待新的運行完成
            run = Runs.wait(thread_id=thread.id, run_id=run.id)

解釋：

• 檢查函數調用：如果智能體需要調用某個函數，我們檢查它調用的是否是 translate_text，並通過之前定義的函數進行翻譯。
• 提交結果：將翻譯結果通過 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[0].text.value}")

總結

通過以上步驟，您已經成功建立了一個智能體，它可以處理使用者的翻譯請求，並使用翻譯函數來完成文本轉換。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,  # 必需參數列表
            },
        },
    }

以快速開始的翻譯文本函數為例：

translation_tool = function_to_schema(translate_text)
print(json.dumps(translation_tool, indent=4, ensure_ascii=False))

翻譯文本函數將被自動轉換為：

{
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": "將文本翻譯成指定的目標語言。\n    這是一個使用預定義翻譯的簡單示範。\n\n    參數：\n        text (str): 需要翻譯的文本\n        target_language (str): 目標語言代碼（例如：'zh'、'es'、'ja'）\n\n    返回：\n        str: 翻譯後的文本或錯誤資訊",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string"
                },
                "target_language": {
                    "type": "string"
                }
            },
            "required": [
                "text",
                "target_language"
            ]
        }
    }
}

現在，我們可以將函數的描述資訊傳遞給模型。

assistant = Assistants.create(
    model='qwen-plus',
    name='翻譯助手',
    description='一個能夠在不同語言之間進行文本翻譯的助手',
    instructions='你是一個翻譯助手。當使用者請求翻譯時，使用 translate_text 函數來協助他們。',
    tools=[translation_tool]
)

使用流式輸出

在流式輸出下，您需要修改第五步：處理函數調用並返回結果的代碼邏輯，這是因為 Runs 現在返回的是 Assistant 事件流。

當 Assistant 決定調用函數時，Runs 會返回事件 thread.run.requires_action和大模型給出的入參資訊 data.required_action.submit_tool_outputs.tool_calls，您需要在此時提交函數輸出結果。

請注意，在提交函數輸出run = Runs.submit_tool_outputs時，也需啟用流式輸出。

# 此代碼僅供示範使用，請在充分理解邏輯後引入您的專案
# 假設已經建立了 assistant、thread 和 message 對象

# 定義工具函數映射
tools_map = {
    "translate_text": translate_text,  # 翻譯函數
}

run = Runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        stream=True  # 開啟流式輸出
    )
while True:  # 添加外層迴圈
    for event, data in run:  # 事件流和事件數目據的詳細資料，請參閱 Assistant API 流式輸出文檔
        if event == 'thread.run.requires_action':   # Assistant 調用了工具，正在等待函數輸出
            tool_outputs = []  # 提交輸出的方法與第五步類似
            for tool in data.required_action.submit_tool_outputs.tool_calls:
                name = tool.function.name
                args = json.loads(tool.function.arguments)
                output = tools_map[name](**args)
                tool_outputs.append({
                    "tool_call_id": tool.id,
                    "output": output,
                })
            run = Runs.submit_tool_outputs(  # 提交函數輸出
                thread_id=thread.id,
                run_id=data.id,
                tool_outputs=tool_outputs,
                stream=True  # 此處也需要開啟流式輸出
            )
            break  # 跳出當前的 for 迴圈，下一次迴圈將輪詢新的 Runs 對象。
    else:
        break  # 如果首輪 for 迴圈正常結束（沒有觸發函數調用），就直接退出 while 迴圈

您可能會注意到，這裡在處理事件流的 for 迴圈外，額外添加了一層 while 迴圈。這是因為您在提交函數輸出時，系統會產生一個新的 Runs 對象。while 迴圈將協助您自動跟蹤最新的事件流，從而使 Assistant 在擷取函數調用的結果後，繼續產生相應的回複。