Cree agentes de IA autónomos con llamadas a funciones | de Julian Yip | abril de 2024

Transforme su chatbot en un agente que pueda interactuar con API externas

Las llamadas a funciones no son algo nuevo. En julio de 2023, OpenAI introdujo Function Calling para sus modelos GPT, una característica que ahora están adoptando los competidores. La API Gemini de Google lo admitió recientemente y Anthropic lo está integrando en Claude. Las llamadas a funciones se están volviendo esenciales para los modelos de lenguajes grandes (LLM), mejorando sus capacidades. ¡Aún más útil aprender esta técnica!

Con esto en mente, mi objetivo es escribir un tutorial completo que cubra las llamadas a funciones más allá de las introducciones básicas (ya hay muchos tutoriales para ello). La atención se centrará en la implementación práctica, creando un agente de IA totalmente autónomo e integrándolo con Streamlit para una interfaz similar a ChatGPT. Aunque OpenAI se utiliza para la demostración, este tutorial se puede adaptar fácilmente a otros LLM que admitan llamadas a funciones, como Gemini.

Las llamadas a funciones permiten a los desarrolladores describir funciones (también conocidas como herramientas, puede considerar esto como acciones que debe realizar el modelo, como realizar cálculos o realizar un pedido) y hacer que el modelo elija de manera inteligente generar un objeto JSON que contenga argumentos para llamar a esas funciones. . En términos más simples, permite:

• Toma de decisiones autónoma: Los modelos pueden elegir inteligentemente herramientas para responder a las preguntas.
• Análisis confiable: Las respuestas están en formato JSON, en lugar de la típica respuesta tipo diálogo. Puede que no parezca mucho a primera vista, pero esto es lo que permite a LLM conectarse a sistemas externos, digamos a través de API con entradas estructuradas.

Abre numerosas posibilidades:

• Asistentes autónomos de IA: Los bots pueden interactuar con sistemas internos para tareas como pedidos y devoluciones de clientes, más allá de proporcionar respuestas a consultas.
• Asistentes personales de investigación.: digamos que si está planificando su viaje, los asistentes pueden buscar en la web, rastrear contenido, comparar opciones y resumir resultados en Excel.
• Comandos de voz de IoT: Los modelos pueden controlar dispositivos o sugerir acciones basadas en intenciones detectadas, como ajustar la temperatura del aire acondicionado.

Préstamo de Documentación de llamadas a funciones de GeminiLa llamada a función tiene la siguiente estructura, que funciona igual en OpenAI

1. El usuario emite un mensaje a la aplicación
2. La aplicación pasa el mensaje proporcionado por el usuario y las declaraciones de función, que son una descripción de las herramientas que el modelo podría utilizar.
3. Basado en la Declaración de Función, el modelo sugiere la herramienta a utilizar y los parámetros de solicitud relevantes. Observe que el modelo genera solo la herramienta y los parámetros sugeridos, SIN llamar realmente a las funciones.
4. & 5. Según la respuesta, la aplicación invoca la API correspondiente.

6. y 7. La respuesta de la API se introduce nuevamente en el modelo para generar una respuesta legible por humanos.

8. La aplicación devuelve la respuesta final al usuario y luego repite desde 1.

Esto puede parecer convulso, pero el concepto se ilustrará en detalle con ejemplos.

Antes de profundizar en el código, unas palabras sobre la arquitectura de la aplicación de demostración.

Solución

Aquí construimos un asistente para los turistas que visitan un hotel. El asistente tiene acceso a las siguientes herramientas, lo que le permite acceder a aplicaciones externas.

• get_items, purchase_item: Conéctese al catálogo de productos almacenado en la base de datos a través de API, para recuperar la lista de artículos y realizar una compra, respectivamente.
• rag_pipeline_func: Conéctese al almacén de documentos con recuperación de generación aumentada (RAG) para obtener información de textos no estructurados, por ejemplo, folletos de hoteles.

Pila de tecnología

¡Ahora comencemos!

Preparación

Dirigirse a GitHub para clonar mi código. Los contenidos a continuación se pueden encontrar en el function_calling_demo Computadora portátil.

Por favor cree y active también un entorno virtual, luego pip install -r requirements.txt para instalar los paquetes requeridos

Inicialización

Primero nos conectamos a OpenRouter. Alternativamente usando el original OpenAIChatGenerator sin sobrescribir el api_base_urlTambién funcionaría, siempre que tenga una clave API de OpenAI

import os
from dotenv import load_dotenv
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk
# Set your API key as environment variable before executing this
load_dotenv()
OPENROUTER_API_KEY = os.environ.get('OPENROUTER_API_KEY')
chat_generator = OpenAIChatGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
api_base_url="https://openrouter.ai/api/v1",
model="openai/gpt-4-turbo-preview",
streaming_callback=print_streaming_chunk)

Luego probamos ¿puede el chat_generator ser invocado exitosamente

chat_generator.run(messages=[ChatMessage.from_user("Return this text: 'test'")])

---------- The response should look like this ----------
{'replies': [ChatMessage(content="'test'", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'stop', 'usage': {}})]}

Paso 1: establecer un almacén de datos

Aquí establecemos conexión entre nuestra aplicación y las dos fuentes de datos: almacén de documentos para textos no estructurados, y base de datos de la aplicación vía API

Indexar documentos con una canalización

Proporcionamos textos de muestra en documents para que el modelo realice Generación Aumentada de Recuperación (RAG). Los textos se convierten en incrustaciones y se almacenan en un almacén de documentos en memoria.

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
# Sample documents
documents = [
Document(content="Coffee shop opens at 9am and closes at 5pm."),
Document(content="Gym room opens at 6am and closes at 10pm.")
]
# Create the document store
document_store = InMemoryDocumentStore()
# Create a pipeline to turn the texts into embeddings and store them in the document store
indexing_pipeline = Pipeline()
indexing_pipeline.add_component(
"doc_embedder", SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
)
indexing_pipeline.add_component("doc_writer", DocumentWriter(document_store=document_store))
indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")
indexing_pipeline.run({"doc_embedder": {"documents": documents}})

Debería generar esto, correspondiente a la documents Creamos como muestra.

{'doc_writer': {'documents_written': 2}}

Poner en marcha el servidor API

Se crea un servidor API creado con Flask en db_api.py para conectarse a SQLite. Por favor, gírelo ejecutando python db_api.py en tu terminal

Observe también que se han agregado algunos datos iniciales en db_api.py

Paso 2: definir las funciones

Aquí preparamos las funciones reales para que el modelo las invoque. DESPUÉS Llamada de función (paso 4-5 como se describe en La estructura de la llamada a funciones)

función TRAPO

Es decir, el rag_pipeline_func. Esto es para que el modelo proporcione una respuesta buscando entre los textos almacenados en el Almacén de documentos. Primero definimos la recuperación de RAG como una tubería de Haystack.

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
template = """
Answer the questions based on the given context.
Context:
{% for document in documents %}
{{ document.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
rag_pipe = Pipeline()
rag_pipe.add_component("embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
rag_pipe.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store))
rag_pipe.add_component("prompt_builder", PromptBuilder(template=template))
# Note to llm: We are using OpenAIGenerator, not the OpenAIChatGenerator, because the latter only accepts List[str] as input and cannot accept prompt_builder's str output
rag_pipe.add_component("llm", OpenAIGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
api_base_url="https://openrouter.ai/api/v1",
model="openai/gpt-4-turbo-preview"))
rag_pipe.connect("embedder.embedding", "retriever.query_embedding")
rag_pipe.connect("retriever", "prompt_builder.documents")
rag_pipe.connect("prompt_builder", "llm")

Pruebe si la función funciona

query = “When does the coffee shop open?”
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})

Esto debería producir el siguiente resultado. Observe la replies que el modelo proporcionó proviene de los documentos de muestra que proporcionamos antes

{'llm': {'replies': ['The coffee shop opens at 9am.'],
'meta': [{'model': 'openai/gpt-4-turbo-preview',
'index': 0,
'finish_reason': 'stop',
'usage': {'completion_tokens': 9,
'prompt_tokens': 60,
'total_tokens': 69,
'total_cost': 0.00087}}]}}

Entonces podemos girar el rag_pipe en una función que proporciona la replies solo sin agregar los demás detalles

def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
return {"reply": result["llm"]["replies"][0]}

llamadas API

Definimos el get_items y purchase_itemFunciones para interactuar con la base de datos.

# Flask's default local URL, change it if necessary
db_base_url = 'http://127.0.0.1:5000'
# Use requests to get the data from the database
import requests
import json
# get_categories is supplied as part of the prompt, it is not used as a tool
def get_categories():
response = requests.get(f'{db_base_url}/category')
data = response.json()
return data
def get_items(ids=None,categories=None):
params = {
'id': ids,
'category': categories,
}
response = requests.get(f'{db_base_url}/item', params=params)
data = response.json()
return data
def purchase_item(id,quantity):
headers = {
'Content-type':'application/json', 
'Accept':'application/json'
}
data = {
'id': id,
'quantity': quantity,
}
response = requests.post(f'{db_base_url}/item/purchase', json=data, headers=headers)
return response.json()

Definir la lista de herramientas

Ahora que hemos definido las funciones, debemos permitir que el modelo reconozca esas funciones e instruirles cómo se usan, proporcionándoles descripciones.

Como estamos usando OpenAI aquí, el tools tiene el formato siguiente, siguiendo las formato requerido por Open AI

tools = [
{
"type": "function",
"function": {
"name": "get_items",
"description": "Get a list of items from the database",
"parameters": {
"type": "object",
"properties": {
"ids": {
"type": "string",
"description": "Comma separated list of item ids to fetch",
},
"categories": {
"type": "string",
"description": "Comma separated list of item categories to fetch",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "purchase_item",
"description": "Purchase a particular item",
"parameters": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "The given product ID, product name is not accepted here. Please obtain the product ID from the database first.",
},
"quantity": {
"type": "integer",
"description": "Number of items to purchase",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "rag_pipeline_func",
"description": "Get information from hotel brochure",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to use in the search. Infer this from the user's message. It should be a question or a statement",
}
},
"required": ["query"],
},
},
}
]

Paso 3: ponerlo todo junto

¡Ahora tenemos las entradas necesarias para probar la llamada a funciones! Aquí hacemos algunas cosas:

1. Proporcionar el mensaje inicial al modelo, para darle algo de contexto.
2. Proporcionar un mensaje de muestra generado por el usuario
3. Lo más importante es que pasamos la lista de herramientas al generador de chat en tools

# 1. Initial prompt
context = f"""You are an assistant to tourists visiting a hotel.
You have access to a database of items (which includes {get_categories()}) that tourists can buy, you also have access to the hotel's brochure.
If the tourist's question cannot be answered from the database, you can refer to the brochure.
If the tourist's question cannot be answered from the brochure, you can ask the tourist to ask the hotel staff.
"""
messages = [
ChatMessage.from_system(context),
# 2. Sample message from user
ChatMessage.from_user("Can I buy a coffee?"),
]
# 3. Passing the tools list and invoke the chat generator
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
response

---------- Response ----------
{'replies': [ChatMessage(content='[{"index": 0, "id": "call_AkTWoiJzx5uJSgKW0WAI1yBB", "function": {"arguments": "{\\"categories\\":\\"Food and beverages\\"}", "name": "get_items"}, "type": "function"}]', role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'tool_calls', 'usage': {}})]}

Ahora inspeccionemos la respuesta. Observe cómo la llamada a función devuelve tanto la función elegida por el modelo como los argumentos para invocar la función elegida.

function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)

---------- Response ----------
Function Name: get_items
Function Arguments: {‘categories’: ‘Food and beverages’}

Cuando se le presente otra pregunta, el modelo utilizará otra herramienta que sea más relevante.

# Another question
messages.append(ChatMessage.from_user("Where's the coffee shop?"))
# Invoke the chat generator, and passing the tools list
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)

---------- Response ----------
Function Name: rag_pipeline_func
Function Arguments: {'query': "Where's the coffee shop?"}

Nuevamente, observe que aquí no se invoca ninguna función real; ¡esto es lo que haremos a continuación!

Llamando a la función

Luego podemos introducir los argumentos en la función elegida.

## Find the correspoding function and call it with the given arguments
available_functions = {"get_items": get_items, "purchase_item": purchase_item,"rag_pipeline_func": rag_pipeline_func}
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
print("Function Response:", function_response)

---------- Response ----------
Function Response: {'reply': 'The provided context does not specify a physical location for the coffee shop, only its operating hours. Therefore, I cannot determine where the coffee shop is located based on the given information.'}

La respuesta de rag_pipeline_func Luego se puede pasar como contexto al chat agregándolo debajo del messagespara que el modelo proporcione la respuesta final

messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
response = chat_generator.run(messages=messages)
response_msg = response["replies"][0]
print(response_msg.content)

---------- Response ----------
For the location of the coffee shop within the hotel, I recommend asking the hotel staff directly. They will be able to guide you to it accurately.

¡Ya hemos completado el ciclo de chat!

Paso 4: conviértete en un chat interactivo

El código anterior muestra cómo se pueden realizar llamadas a funciones, pero queremos ir un paso más allá y convertirlo en un chat interactivo.

Aquí muestro dos métodos para hacerlo, desde el más primitivo input() que imprime el diálogo en el propio cuaderno, hasta renderizarlo a través de iluminado para proporcionarle una interfaz de usuario similar a ChatGPT

input() bucle

El código se copia de Tutorial de pajarlo que nos permite probar rápidamente el modelo. Nota: Esta aplicación se creó para demostrar la idea de llamada a funciones y NO pretende ser perfectamente sólida, por ejemplo, admitir el orden de varios elementos al mismo tiempo, sin alucinaciones, etc.

import json
from haystack.dataclasses import ChatMessage, ChatRole
response = None
messages = [
ChatMessage.from_system(context)
]
while True:
# if OpenAI response is a tool call
if response and response["replies"][0].meta["finish_reason"] == "tool_calls":
function_calls = json.loads(response["replies"][0].content)
for function_call in function_calls:
## Parse function calling information
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
## Find the correspoding function and call it with the given arguments
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
## Append function response to the messages list using `ChatMessage.from_function`
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
# Regular Conversation
else:
# Append assistant messages to the messages list
if not messages[-1].is_from(ChatRole.SYSTEM):
messages.append(response["replies"][0])
user_input = input("ENTER YOUR MESSAGE 👇 INFO: Type 'exit' or 'quit' to stop\n")
if user_input.lower() == "exit" or user_input.lower() == "quit":
break
else:
messages.append(ChatMessage.from_user(user_input))
response = chat_generator.run(messages=messages, generation_kwargs={"tools": tools})

Si bien funciona, es posible que queramos tener algo que se vea mejor.

Interfaz optimizada

Streamlit convierte scripts de datos en aplicaciones web que se pueden compartir, lo que proporciona una interfaz de usuario ordenada para nuestra aplicación. El código que se muestra arriba está adaptado a una aplicación Streamlit bajo el streamlit carpeta de mi repositorio

Puedes ejecutarlo mediante:

1. Si aún no lo ha hecho, active el servidor API con python db_api.py
2. Establezca OPENROUTER_API_KEY como variable de entorno, por ejemplo export OPENROUTER_API_KEY = ‘@REPLACE WITH YOUR API KEY’ asumiendo que estás en Linux/ejecutando con git bash
3. Navega hasta el streamlit carpeta en la terminal con cd streamlit
4. Ejecute Streamlit con streamlit run app.py. Se debería crear automáticamente una nueva pestaña en su navegador al ejecutar la aplicación.

¡Eso es básicamente todo! Espero que disfrutes este artículo.

*A menos que se indique lo contrario, todas las imágenes son del autor.