La API de Assistant admite la llamada a funciones. Esta característica permite que un agente llame automáticamente a funciones externas para realizar tareas, como traducir texto. Este tema utiliza un ejemplo simple de "Agente de Traducción" para ayudarle a comprender rápidamente los conceptos básicos de la llamada a funciones.
ImportanteLa API de Assistant está siendo descontinuada. Migre a la API de Responses como alternativa. La API de Responses incluye múltiples herramientas integradas y admite la gestión de contexto de múltiples turnos.
Inicio rápido
En este ejemplo, creará un agente de traducción y una función llamada translate_text que el agente puede invocar. Luego, pedirá al agente que traduzca "Hello world" al chino.
Antes de comenzar
Puede instalar las bibliotecas de dependencias requeridas, como requests y dashscope, ejecutando el siguiente comando:
pip install requests dashscope
Paso 1: Crear la función "translate_text"
Primero, puede crear una función de traducción simple. Esta función utiliza una tabla de traducción predefinida con fines de demostración.
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)}"
Explicación:
- Función de traducción: Simula una función de traducción utilizando una tabla de traducción predefinida que admite la conversión entre varios idiomas.
- Manejo de errores: La función incluye un mecanismo básico de manejo de errores para garantizar que devuelva una respuesta adecuada en todas las situaciones.
Ahora, puede utilizar la API de Assistant para crear un agente. Este agente procesa automáticamente las consultas de los usuarios y llama a la función translate_text definida para proporcionar servicios de traducción.
Paso 2: Describir la función "translate_text"
Debe describir la función translate_text al agente. El agente utiliza esta descripción para llamar a la función correctamente.
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"]
}
}
}
Explicación:
- name: El nombre de la función es
translate_text. El agente utiliza este nombre para llamar a la función. - description: Una descripción de la herramienta que ayuda al agente a comprender su propósito.
- parameters: Define los parámetros de la función, que incluyen el texto a traducir y el idioma de destino.
Paso 3: Crear un agente
Ahora, puede crear una instancia de Assistant. Esta instancia es un agente que utilizará la herramienta de traducción que usted definió.
# 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]
)
Explicación:
- model: Especifica el modelo a utilizar. Este ejemplo usa
qwen-plus, que admite la comprensión del lenguaje y el procesamiento de tareas. - name: El nombre del agente. Establezca el valor en "Translation Assistant".
- description: Una descripción del propósito del agente, que es ayudar a los usuarios a traducir texto.
- tools: Registra el
translation_toolque definió anteriormente. Esto permite al agente llamar a la herramienta.
Paso 4: Crear un hilo de conversación e interactuar con el agente
Cree un nuevo hilo de conversación, agregue un mensaje de usuario al mismo y luego ejecute el agente para procesar la consulta del usuario.
# 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)
Explicación:
- Threads.create(): Crea un nuevo hilo de conversación para mensajes subsiguientes.
- Messages.create(): Agrega un mensaje de usuario al hilo. En este caso, el usuario solicita traducir "Hello world" al chino.
- Runs.create(): Activa el agente para comenzar a procesar el mensaje del usuario.
- Runs.wait(): Espera a que el agente termine de procesar.
Paso 5: Manejar la llamada a la función y devolver el resultado
Si el agente necesita llamar a una herramienta durante el procesamiento, se llama a la función translate_text y se devuelve el resultado.
# 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)
Explicación:
- Verificar si hay una llamada a función: Si el agente necesita llamar a una función, su código verifica si la función solicitada es
translate_texty luego realiza la traducción utilizando la función definida previamente. - Enviar el resultado: Envíe el resultado de la traducción al agente mediante
Runs.submit_tool_outputsy luego espere la siguiente respuesta del agente.
Paso 6: Obtener la respuesta del agente
Después de que el agente termine el procesamiento, puede recuperar la respuesta del agente del hilo de conversación y mostrársela al usuario.
# 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}")
Resumen
Al seguir estos pasos, ha creado con éxito un agente que puede manejar solicitudes de traducción de usuarios y utilizar una función de traducción para realizar la conversión de texto. La API de Assistant facilita la creación de agentes complejos y orientados a tareas de manera sencilla y eficiente.
Puede ampliar las funciones del agente según sea necesario, por ejemplo, agregando más herramientas o modificando las instrucciones de comportamiento del agente.
Generar rápidamente descripciones para funciones de negocio
En el ejemplo de Inicio rápido, necesita describir la función "translate_text" al agente. Este proceso puede ser tedioso. Por lo tanto, proporcionamos una función de conversión simple para ayudarle a describir rápidamente sus funciones de negocio.
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
},
},
}
Por ejemplo, considere la función translate_text del Inicio rápido:
translation_tool = function_to_schema(translate_text)
print(json.dumps(translation_tool, indent=4, ensure_ascii=False))
La función translate_text se convierte automáticamente 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"
]
}
}
}
Ahora puede pasar la descripción de la función al modelo.
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]
)
Usar salida streaming
Cuando utilice la salida por streaming, debe modificar la lógica del código en Paso 5: Manejar la llamada a la función y devolver el resultado. Esto se debe a que el objeto Runs ahora devuelve un flujo de eventos de Assistant.
Cuando el Assistant decide llamar a una función, el objeto Runs devuelve el evento thread.run.requires_action y los parámetros de entrada data.required_action.submit_tool_outputs.tool_calls proporcionados por el Modelo de Lenguaje Grande (LLM). Debe enviar la salida de la función en este punto.
Tenga en cuenta que también debe habilitar la salida por streaming cuando envíe la salida de la función con 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.
Es posible que note un bucle while adicional fuera del bucle for que procesa el flujo de eventos. Esto se debe a que el sistema genera un nuevo objeto Runs cuando usted envía la salida de la función. El bucle while le ayuda a rastrear automáticamente el flujo de eventos más reciente. Esto permite que el Assistant continúe generando una respuesta después de recibir el resultado de la llamada a la función.