Alibaba Cloud Model Studio:Tool calling

Get started

In this example, we will create a translation assistant and a function translate_text that the assistant can use. Then, we can ask the assistant to translate "Hello world" into Chinese.

Before you begin

Make sure you have installed the required dependency libraries, such as requests and dashscope. You can run the following command to install them:

pip install requests dashscope

Step 1: Create the translate_text function

We start with a basic translation function that uses a predefined dictionary for demonstration purpose.

def translate_text(text, target_language):
    """
    Translate text into the specified target language.
    This is a simple demonstration using a predefined dictionary.

    Parameters:
        text (str): The text to be translated
        target_language (str): The target language code (e.g., 'zh', 'es', 'ja')

    Returns:
        str: The translated text or an error message
    """
    # Translation dictionary for demonstration
    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"Translation not found. In a real production environment, a translation service would be called here.")
    except Exception as e:
        return f"Translation failed: {str(e)}"

Explanation:

• How to translate: Uses a predefined dictionary to simulate translation capabilities that support multiple languages.

• Error handling: Includes basic error handling to ensure the function delivers an appropriate response under various conditions.

Then, use the Assistant API to create an assistant that automatically processes user queries and uses the translate_text function to provide translation services.

Step 2: Describe the translate_text function

We need to describe the translate_text function to the assistant so that it can accurately call the function based on this description.

from dashscope import Assistants, Messages, Runs, Threads
import json
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
# Define the translation tool
translation_tool = {
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": "Translate text into the specified target language",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "The text to be translated"
                },
                "target_language": {
                    "type": "string",
                    "description": "The target language code (e.g., 'zh', 'es', 'ja')"
                }
            },
            "required": ["text", "target_language"]
        }
    }
}

Explanation:

• name: The name of the function is translate_text, which the assistant will use when making a call.

• description: The description of the function. This helps the assistant understand the use of the function.

• parameters: The parameters of the function, including the source text and the target language.

Step 3: Create an assistant

We now create an assistant, which can use the translation tool we have defined.

# Create Assistant
assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Assistant',
    description='An assistant capable of translating text between different languages',
    instructions='You are a translation assistant. When users request a translation, use the translate_text function.',
    tools=[translation_tool]
)

Explanation:

• model: The model used by the assistant. In this example, qwen-plus is used, which supports language comprehension and task execution.

• name: The name of the assistant, Translation Assistant.

• description: Describes the capabilities of the assistant, which is text translation.

• tools: Integrates the translation tool we have defined, enabling the assistant to use this tool.

Step 4: Create a thread and interact with the assistant

Initiate a new thread and add user messages. Then, run the assistant to process user queries.

# Create a new thread
thread = Threads.create()

# Add user message to the thread
Messages.create(
    thread_id=thread.id, 
    role="user", 
    content="Please translate 'Hello world' into Chinese."
)

# Run Assistant
run = Runs.create(thread_id=thread.id, assistant_id=assistant.id)

# Wait for the run to complete
run = Runs.wait(thread_id=thread.id, run_id=run.id)

Explanation:

• Threads.create(): Creates a new thread for subsequent messages.

• Messages.create(): Adds user message to the thread. In this example, the user wants to translate "Hello world" into Chinese.

• Runs.create(): Activates the assistant to begin processing the user messages.

• Runs.wait(): Waits for the completion of the assistant's processing.

Step 5: Handle function calling and return results

If the assistant needs to use the tool during processing, it can call the translate_text function and then return the results to the user.

# Check whether tool calling is needed
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"])

            # Submit tool output
            Runs.submit_tool_outputs(
                thread_id=thread.id,
                run_id=run.id,
                tool_outputs=[{"tool_call_id": tool_call.id, "output": translation}]
            )

            # Wait for the new run to complete
            run = Runs.wait(thread_id=thread.id, run_id=run.id)

Explanation:

• Check function calling: If the assistant needs to call a function, check whether the called function is translate_text.

• Submit results: Delivers the translation outcomes to the assistant with Runs.submit_tool_outputs and waits for the subsequent response from the assistant.

Step 6: Get the response from the assistant

After the assistant completes processing, retrieve the response from the thread and deliver it to the user.

# Get Assistant's response
messages = Messages.list(thread_id=thread.id)
for message in messages.data:
    if message.role == "assistant":
        print(f"Assistant: {message.content[0].text.value}")

Summary

Following these steps, you have built an assistant capable of processing translation requests and using the translation function. The Assistant API simplifies the creation of complex, task-oriented assistants.

You can enhance the capabilities of your assistant based on your requirements, by adding more tools or modifying the instructions.

Quickly generate tool descriptions

In Step 2 of the above example, we need to describe the translate_text function for the assistant, which can be a laborious job. To streamline this process, we offer an easy transformation function that can generate descriptions for your business tools rapidly.

import inspect

def function_to_schema(func) -> dict:
    # Map Python types to JSON schema types
    type_map = {
        str: "string",
        int: "integer",
        float: "number",
        bool: "boolean",
        list: "array",
        dict: "object",
        type(None): "null",
    }

    # Try to get the function's signature
    try:
        signature = inspect.signature(func)
    except ValueError as e:
        # If signature retrieval fails, raise an error with an error message
        raise ValueError(
            f"Failed to get signature for function {func.__name__}: {str(e)}"
        )

    # Initialize a dictionary to store parameter types
    parameters = {}
    # Traverse the function's parameters and map their types
    for param in signature.parameters.values():
        try:
            param_type = type_map.get(param.annotation, "string")
        except KeyError as e:
            # If the parameter's type annotation is unknown, raise an error
            raise KeyError(
                f"Unknown type annotation {param.annotation} for parameter {param.name}: {str(e)}"
            )
        parameters[param.name] = {"type": param_type}

    # Create a list of required parameters (those without default values)
    required = [
        param.name
        for param in signature.parameters.values()
        if param.default == inspect._empty
    ]

    # Return the function's schema as a dictionary
    return {
        "type": "function",
        "function": {
            "name": func.__name__,
            "description": (func.__doc__ or "").strip(),  # Get function description (docstring)
            "parameters": {
                "type": "object",
                "properties": parameters,  # Parameter types
                "required": required,  # List of required parameters
            },
        },
    }

Take the translate_text function as an example:

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

The translate_text function will be automatically converted into:

{
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": "Translate text into the specified target language.\n    This is a simple demonstration using predefined translations.\n\n    Parameters:\n        text (str): The text to be translated\n        target_language (str): The target language code (e.g., 'zh', 'es', 'ja')\n\n    Returns:\n        str: The translated text or an error message",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string"
                },
                "target_language": {
                    "type": "string"
                }
            },
            "required": [
                "text",
                "target_language"
            ]
        }
    }
}

Pass the function description to the model as:

assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Assistant',
    description='An assistant capable of translating text between different languages',
    instructions='You are a translation assistant. When users request a translation, use the translate_text function.',
    tools=[translation_tool]
)

Use streaming output

In streaming output mode, you need to modify the logic of Step 5: Handle function calling and return results, because Runs returns an event stream for the Assistant.

When Assistant decides a function to call, Runs returns an event thread.run.requires_action and the input parameters provided by the model data.required_action.submit_tool_outputs.tool_calls. You need to specify the function output at this time.

Note that you need to enable stream when submitting function output run = Runs.submit_tool_outputs.

# This code is for demonstration purposes only; please incorporate it into your project after fully understanding the logic.
# Assume that assistant, thread, and message objects have been created

run = Runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        stream=True  # Enable stream
    )
while True:  # Add other loop
    for event, data in run:  # Event stream and event data
        if event == 'thread.run.requires_action':   # The Assistant has called a tool and is waiting for tool output
            tool_outputs = []  # The method for submitting output is similar to Step Five
            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(  # Submit tool output
                thread_id=thread.id,
                run_id=data.id,
                tool_outputs=tool_outputs,
                stream=True  # Enable stream here
            )
            break  # Exit the current for loop; the next iteration will poll the new Runs object
    else:
        break  # If the first round of the for loop ends normally (without triggering a function call), exit the while loop

You may notice an additional outer while loop added outside the for loop that handles the event stream. This is because when you submit tool outputs, the system generates a new Runs object. The while loop will help you automatically track the latest event stream, allowing the Assistant to continue generating appropriate responses after obtaining the results of the tool calls.