Alibaba Cloud Model Studio:Function calling (Deprecated)

Quick start

In this example, you will create a translation agent and a function named translate_text that the agent can call. You will then ask the agent to translate "Hello world" into Chinese.

Before you begin

You can install the required dependency libraries, such as requests and dashscope, by running the following command:

pip install requests dashscope

Step 1: Create the "translate_text" function

First, you can create a simple translation function. This function uses a predefined translation table for demonstration purposes.

def translate_text(text, target_language):
    """
    Translates text into the specified target language.
    This is a simple demonstration that uses a predefined translation.

    Parameters:
        text (str): The text to translate.
        target_language (str): The target language code (for example, 'zh', 'es', or 'ja').

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

Explanation:

• Translation feature: Simulates a translation feature using a predefined translation table that supports conversion between multiple languages.

• Error handling: The function includes a basic error handling mechanism to ensure that it returns an appropriate response in all situations.

Now, you can use the Assistant API to create an agent. This agent automatically processes user queries and calls the defined `translate_text` function to provide translation services.

Step 2: Describe the "translate_text" function

You must describe the `translate_text` function to the agent. The agent uses this description to call the function correctly.

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": "Translates text into the specified target language",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "The text to translate"
                },
                "target_language": {
                    "type": "string",
                    "description": "The target language code (for example, 'zh', 'es', or 'ja')"
                }
            },
            "required": ["text", "target_language"]
        }
    }
}

Explanation:

• name: The name of the function is `translate_text`. The agent uses this name to call the function.

• description: A description of the tool that helps the agent understand its purpose.

• parameters: Defines the function's parameters, which include the text to translate and the target language.

Step 3: Create an agent

Now, you can create an Assistant instance. This instance is an agent that will use the translation tool that you defined.

# Create an Assistant
assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Agent',
    description='An agent that can translate text between different languages',
    instructions='You are a translation agent. When a user requests a translation, use the translate_text function to help them.',
    tools=[translation_tool]
)

Explanation:

• model: Specifies the model to use. This example uses `qwen-plus`, which supports language understanding and task processing.

• name: The name of the agent. Set the value to "Translation Assistant".

• description: A description of the agent's purpose, which is to help users translate text.

• tools: Registers the `translation_tool` that you defined earlier. This allows the agent to call the tool.

Step 4: Create a conversation thread and interact with the agent

Create a new conversation thread, add a user message to it, and then run the agent to process the user's query.

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

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

# Run the 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 conversation thread for subsequent messages.

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

• Runs.create(): Triggers the agent to start processing the user message.

• Runs.wait(): Waits for the agent to finish processing.

Step 5: Handle the function call and return the result

If the agent needs to call a tool during processing, the translate_text function is called and the result is returned.

# Check if a function call is required
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 the 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 for a function call: If the agent needs to call a function, your code checks whether the requested function is `translate_text` and then performs the translation using the previously defined function.

• Submit the result: Submit the translation result to the agent using `Runs.submit_tool_outputs`, and then wait for the agent's next response.

Step 6: Get the agent's response

After the agent finishes processing, you can retrieve the agent's response from the conversation thread and display it to the user.

# Get the 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

By following these steps, you have successfully created an agent that can handle user translation requests and use a translation function to perform text conversion. The Assistant API makes it simple and efficient to build complex, task-driven agents.

You can extend the agent's features as needed, such as by adding more tools or modifying the agent's behavior instructions.

Quickly generate descriptions for business functions

In the Quick Start example, you need to describe the "translate_text" function to the agent. This process can be tedious. Therefore, we provide a simple conversion function to help you quickly describe your business functions.

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 getting the signature fails, raise an error with the error message
        raise ValueError(
            f"Failed to get signature for function {func.__name__}: {str(e)}"
        )

    # Initialize a dictionary to store parameter types
    parameters = {}
    # Iterate over 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 a default value)
    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 the function description (docstring)
            "parameters": {
                "type": "object",
                "properties": parameters,  # Parameter types
                "required": required,  # List of required parameters
            },
        },
    }

For example, consider the `translate_text` function from the Quick Start:

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

The `translate_text` function is automatically converted to:

{
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": "Translates text into the specified target language.\n    This is a simple demonstration that uses a predefined translation.\n\n    Parameters:\n        text (str): The text to translate.\n        target_language (str): The target language code (for example, 'zh', 'es', or '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"
            ]
        }
    }
}

Now, you can pass the function's description to the model.

assistant = Assistants.create(
    model='qwen-plus',
    name='Translation Agent',
    description='An agent that can translate text between different languages',
    instructions='You are a translation agent. When a user requests a translation, use the translate_text function to help them.',
    tools=[translation_tool]
)

Use streaming output

When you use streaming output, you must modify the code logic in Step 5: Handle the function call and return the result. This is because the Runs object now returns an Assistant event stream.

When the Assistant decides to call a function, the Runs object returns the thread.run.requires_action event and the input parameters data.required_action.submit_tool_outputs.tool_calls that are provided by the Large Language Model (LLM). You must submit the function output at this point.

Note that you must also enable streaming output when you submit the function output with run = Runs.submit_tool_outputs.

# This code is for demonstration only. Integrate it into your project after you fully understand the logic.
# Assume that the assistant, thread, and message objects have been created.

# Define the tool function mapping
tools_map = {
    "translate_text": translate_text,  # Translation function
}

run = Runs.create(
        thread_id=thread.id,
        assistant_id=assistant.id,
        stream=True  # Enable streaming output
    )
while True:  # Add an outer loop
    for event, data in run:  # For more information about the event stream and event data, see the Assistant API streaming output documentation.
        if event == 'thread.run.requires_action':   # The Assistant has called a tool and is waiting for the function output.
            tool_outputs = []  # The method for submitting the output is similar to that in Step 5.
            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 the function output
                thread_id=thread.id,
                run_id=data.id,
                tool_outputs=tool_outputs,
                stream=True  # Streaming output must also be enabled here.
            )
            break  # Break out of the current for loop. The next loop will poll the new Runs object.
    else:
        break  # If the first for loop finishes normally without triggering a function call, break out of the while loop.

You may notice an extra `while` loop outside the `for` loop that processes the event stream. This is because the system generates a new Runs object when you submit the function output. The `while` loop helps you automatically track the latest event stream. This allows the Assistant to continue generating a response after it receives the result of the function call.