L'API Assistant prend en charge l'appel de fonction. Cette fonctionnalité permet à un agent d'invoquer automatiquement des fonctions externes pour accomplir des tâches, telles que la traduction de texte. Cette rubrique s'appuie sur un exemple simple d'« Agent de traduction » pour vous aider à comprendre rapidement les principes fondamentaux de l'appel de fonction.
ImportantL'API Assistant est en cours de dépréciation. Migrez vers l'API Responses comme alternative. L'API Responses intègre plusieurs outils natifs et assure la gestion du contexte multi-tours.
Démarrage rapide
Dans cet exemple, vous créez un agent de traduction ainsi qu'une fonction nommée translate_text que l'agent peut appeler. Vous demandez ensuite à l'agent de traduire « Hello world » en chinois.
Prérequis
Installez les bibliothèques de dépendances requises, telles que requests et dashscope, en exécutant la commande suivante :
pip install requests dashscope
Étape 1 : Créer la fonction « translate_text »
Commencez par créer une fonction de traduction simple. À des fins de démonstration, cette fonction utilise une table de traduction prédéfinie.
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)}"
Explication :
- Fonctionnalité de traduction : Simule une capacité de traduction via une table prédéfinie prenant en charge la conversion entre plusieurs langues.
- Gestion des erreurs : La fonction intègre un mécanisme basique de gestion des erreurs afin de garantir une réponse appropriée en toutes circonstances.
Utilisez ensuite l'API Assistant pour créer un agent. Celui-ci traite automatiquement les requêtes utilisateur et appelle la fonction translate_text définie précédemment pour fournir des services de traduction.
Étape 2 : Décrire la fonction « translate_text »
Décrivez la fonction translate_text à l'agent. Ce dernier s'appuie sur cette description pour invoquer correctement la fonction.
from dashscope import Assistants, Messages, Runs, Threads
import json
# 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"]
}
}
}
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"]
}
}
}
Explication :
- name : Nom de la fonction (
translate_text). L'agent utilise cet identifiant pour l'appeler. - description : Description de l'outil aidant l'agent à en comprendre l'utilité.
- parameters : Définit les paramètres de la fonction, comprenant le texte à traduire et la langue cible.
Étape 3 : Créer un agent
Créez maintenant une instance Assistant. Cette instance agit comme un agent exploitant l'outil de traduction défini.
# 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]
)
Explication :
- model : Spécifie le modèle à utiliser. Cet exemple emploie
qwen-plus, capable de comprendre le langage naturel et de traiter des tâches. - name : Nom de l'agent. Définissez la valeur sur « Translation Assistant ».
- description : Décrit l'objectif de l'agent, à savoir aider les utilisateurs à traduire du texte.
- tools : Enregistre le
translation_tooldéfini précédemment, autorisant ainsi l'agent à l'invoquer.
Étape 4 : Créer un fil de conversation et interagir avec l'agent
Créez un nouveau fil de conversation, ajoutez-y un message utilisateur, puis lancez l'agent pour traiter la requête.
# 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)
Explication :
- Threads.create() : Initialise un nouveau fil de conversation pour les messages suivants.
- Messages.create() : Ajoute un message utilisateur au fil. Ici, l'utilisateur demande la traduction de « Hello world » en chinois.
- Runs.create() : Déclenche le traitement du message utilisateur par l'agent.
- Runs.wait() : Attend la fin du traitement par l'agent.
Étape 5 : Gérer l'appel de fonction et retourner le résultat
Si l'agent doit invoquer un outil durant son traitement, la fonction translate_text est appelée et le résultat est retourné.
# 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)
Explication :
- Vérification de l'appel de fonction : Si l'agent nécessite un appel de fonction, votre code vérifie si la fonction demandée est
translate_text, puis exécute la traduction via la fonction définie plus haut. - Soumission du résultat : Transmettez le résultat de la traduction à l'agent via
Runs.submit_tool_outputs, puis attendez sa prochaine réponse.
Étape 6 : Obtenir la réponse de l'agent
Une fois le traitement terminé, récupérez la réponse de l'agent depuis le fil de conversation pour l'afficher à l'utilisateur.
# 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}")
Résumé
En suivant ces étapes, vous avez créé un agent capable de gérer les demandes de traduction des utilisateurs et d'utiliser une fonction dédiée pour convertir du texte. L'API Assistant simplifie la conception d'agents complexes orientés tâches.
Vous pouvez étendre les fonctionnalités de l'agent selon vos besoins, notamment en ajoutant d'autres outils ou en modifiant ses instructions comportementales.
Générer rapidement des descriptions pour les fonctions métier
Dans l'exemple de démarrage rapide, vous devez décrire la fonction « translate_text » à l'agent, ce qui peut s'avérer fastidieux. Nous fournissons donc une fonction de conversion simple pour décrire rapidement vos fonctions métier.
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
},
},
}
Prenons par exemple la fonction translate_text du démarrage rapide :
translation_tool = function_to_schema(translate_text)
print(json.dumps(translation_tool, indent=4, ensure_ascii=False))
La fonction translate_text est automatiquement convertie en :
{
"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"
]
}
}
}
Transmettez ensuite la description de la fonction au modèle.
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]
)
Utiliser la sortie en streaming
Lorsque vous utilisez la sortie en streaming, modifiez la logique du code présentée à l'Étape 5 : Gérer l'appel de fonction et retourner le résultat. En effet, l'objet Runs retourne désormais un flux d'événements Assistant.
Quand l'Assistant décide d'appeler une fonction, l'objet Runs émet l'événement thread.run.requires_action ainsi que les paramètres d'entrée data.required_action.submit_tool_outputs.tool_calls fournis par le grand modèle de langage (LLM). Soumettez alors la sortie de la fonction.
Notez que vous devez également activer la sortie en streaming lors de la soumission du résultat de la fonction avec 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.
Une boucle while supplémentaire englobe la boucle for traitant le flux d'événements. Le système génère en effet un nouvel objet Runs lorsque vous soumettez la sortie de la fonction. La boucle while suit automatiquement le dernier flux d'événements, permettant à l'Assistant de poursuivre la génération de sa réponse après réception du résultat de l'appel de fonction.