Function calling (Deprecated)

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)}"

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"]
        }
    }
}

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
            },
        },
    }

# 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.