Trasforma il tuo chatbot in un agente in grado di interagire con API esterne
La chiamata di funzione non è qualcosa di nuovo. Nel luglio 2023, OpenAI ha introdotto la Function Calling per i propri modelli GPT, una funzionalità ora adottata dalla concorrenza. L'API Gemini di Google lo ha recentemente supportato e Anthropic lo sta integrando in Claude. La chiamata di funzione sta diventando essenziale per i modelli linguistici di grandi dimensioni (LLM), migliorandone le capacità. Tanto più utile imparare questa tecnica!
Con questo in mente, mi propongo di scrivere un tutorial completo che copra la chiamata alle funzioni oltre alle introduzioni di base (ci sono già molti tutorial per questo). L'attenzione sarà focalizzata sull'implementazione pratica, costruendo un agente AI completamente autonomo e integrandolo con Streamlit per un'interfaccia simile a ChatGPT. Sebbene OpenAI venga utilizzato a scopo dimostrativo, questo tutorial può essere facilmente adattato per altri LLM che supportano la chiamata di funzione, come Gemini.
La chiamata di funzione consente agli sviluppatori di descrivere le funzioni (ovvero gli strumenti, puoi considerarli come azioni da intraprendere per il modello, come eseguire calcoli o effettuare un ordine) e fare in modo che il modello scelga in modo intelligente di restituire un oggetto JSON contenente argomenti per chiamare tali funzioni . In termini più semplici, consente:
- Processo decisionale autonomo: i modelli possono scegliere in modo intelligente gli strumenti per rispondere alle domande.
- Analisi affidabile: le risposte sono in formato JSON, invece della più tipica risposta simile a un dialogo. Potrebbe non sembrare molto a prima vista, ma questo è ciò che consente a LLM di connettersi a sistemi esterni, ad esempio tramite API con input strutturati.
Apre numerose possibilità:
- Assistenti IA autonomi: i bot possono interagire con i sistemi interni per attività quali ordini e resi dei clienti, oltre a fornire risposte alle domande
- Assistenti personali di ricerca: ad esempio, se stai pianificando il tuo viaggio, gli assistenti possono eseguire ricerche sul Web, eseguire la scansione dei contenuti, confrontare le opzioni e riepilogare i risultati in Excel.
- Comandi vocali dell'IoT: i modelli possono controllare dispositivi o suggerire azioni in base alle intenzioni rilevate, come la regolazione della temperatura AC.
Prendendo in prestito da Documentazione sulla chiamata delle funzioni di GeminiLa chiamata di funzione ha la struttura seguente, che funziona allo stesso modo in OpenAI
- I problemi dell'utente richiedono l'applicazione
- L'applicazione passa il prompt fornito dall'utente e la/e dichiarazione/i di funzione, che è una descrizione degli strumenti che il modello potrebbe utilizzare
- Sulla base della Dichiarazione di Funzione, il modello suggerisce lo strumento da utilizzare e i relativi parametri di richiesta. Si noti che il modello restituisce solo lo strumento e i parametri suggeriti, SENZA chiamare effettivamente le funzioni
- & 5. In base alla risposta, l'applicazione richiama l'API pertinente
6. e 7. La risposta dell'API viene nuovamente inserita nel modello per produrre una risposta leggibile dall'uomo
8. L'applicazione restituisce la risposta finale all'utente, quindi ripete da 1.
Questo potrebbe sembrare contorto, ma il concetto verrà illustrato in dettaglio con esempi
Prima di addentrarci nel codice, qualche parola sull'architettura dell'applicazione demo
Soluzione
Qui costruiamo un assistente per i turisti che visitano un hotel. L'assistente ha accesso ai seguenti strumenti, che gli consentono di accedere ad applicazioni esterne.
get_items,purchase_item: connettersi al catalogo prodotti archiviato nel database tramite API, rispettivamente per recuperare l'elenco degli articoli ed effettuare un acquistorag_pipeline_func: Connessione all'archivio documenti con Retrieval Augmented Generation (RAG) per ottenere informazioni da testi non strutturati, ad esempio brochure di hotel
Pila tecnologica
Adesso cominciamo!
Preparazione
Vai a Github per clonare il mio codice. I contenuti riportati di seguito sono reperibili nel function_calling_demo Taccuino.
Ti invitiamo quindi a creare e attivare anche un ambiente virtuale pip install -r requirements.txt per installare i pacchetti richiesti
Inizializzazione
Per prima cosa ci colleghiamo a OpenRouter. In alternativa utilizzando l'originale OpenAIChatGenerator senza sovrascrivere il file api_base_urlfunzionerebbe anche, a condizione che tu abbia una chiave API 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)
Quindi testiamo il file chat_generator essere invocato con successo
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': {}}))}
Passaggio 1: stabilire l'archivio dati
Qui stabiliamo la connessione tra la nostra applicazione e le due origini dati: Archivio documenti per testi non strutturati, e banca dati dell'applicazione tramite API
Indicizzare i documenti con una pipeline
Forniamo testi di esempio in documents affinché il modello esegua il Retrival Augmented Generation (RAG). I testi vengono trasformati in incorporamenti e archiviati in un archivio documenti in 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}})
Dovrebbe emettere questo, corrispondente a documents abbiamo creato come campione
{'doc_writer': {'documents_written': 2}}
Avvia il server API
Viene creato un server API realizzato con Flask db_api.py per connettersi a SQLite. Per favore, aumentalo correndo python db_api.py nel tuo terminale
Si noti inoltre che sono stati aggiunti alcuni dati iniziali db_api.py
Passaggio 2: definire le funzioni
Qui prepariamo le funzioni effettive per il modello da invocare DOPO Chiamata di funzione (passaggi 4–5 come descritto in La struttura della chiamata di funzione)
Funzione RAG
Vale a dire il rag_pipeline_func. Questo serve al modello per fornire una risposta effettuando una ricerca tra i testi archiviati nell'archivio documenti. Per prima cosa definiamo il recupero RAG come una pipeline 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 OpenAIGeneratortemplate = """
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")
Prova se la funzione funziona
query = “When does the coffee shop open?”
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
Ciò dovrebbe produrre il seguente output. Notare il replies che il modello fornito proviene dai documenti di esempio forniti in precedenza
{'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}})}}
Possiamo quindi girare il rag_pipe in una funzione, che fornisce il replies solo senza aggiungere gli altri dettagli
def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})return {"reply": result("llm")("replies")(0)}
Chiamate API
Definiamo il get_items E purchase_itemfunzioni per interagire con il database
# 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()
Definire l'elenco degli strumenti
Ora che abbiamo definito le funzioni, dobbiamo lasciare che il modello riconosca tali funzioni e istruirli su come vengono utilizzate, fornendo loro descrizioni.
Dato che qui stiamo usando OpenAI, il file tools è formattato come di seguito seguendo il file formato richiesto da 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"),
},
},
}
)
Passaggio 3: mettere tutto insieme
Ora abbiamo gli input necessari per testare la chiamata di funzione! Qui facciamo alcune cose:
- Fornire il prompt iniziale al modello, per dargli un po' di contesto
- Fornisci un messaggio di esempio generato dall'utente
- Ancora più importante, passiamo l'elenco degli strumenti al generatore di chat
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': {}}))}
Ora esaminiamo la risposta. Notare come la chiamata di funzione restituisce sia la funzione scelta dal modello, sia gli argomenti per invocare la funzione scelta.
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’}
Quando viene presentata un'altra domanda, il modello utilizzerà un altro strumento più pertinente
# 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?"}
Ancora una volta, nota che qui non viene richiamata alcuna funzione effettiva, questo è ciò che faremo dopo!
Chiamando la funzione
Possiamo quindi inserire gli argomenti nella funzione scelta
## 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 risposta di rag_pipeline_func può quindi essere passato come contesto alla chat aggiungendolo sotto il file messagesaffinché il modello fornisca la risposta finale
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.
Ora abbiamo completato il ciclo di chat!
Passaggio 4: trasforma in una chat interattiva
Il codice sopra mostra come è possibile eseguire la chiamata di funzione, ma vogliamo fare un ulteriore passo avanti trasformandola in una chat interattiva
Qui mostro due metodi per farlo, dal più primitivo input() che stampa il dialogo nel taccuino stesso, per visualizzarlo Illuminato per fornirgli un'interfaccia utente simile a ChatGPT
input() ciclo continuo
Il codice viene copiato da Il tutorial di Pagliaioche ci permette di testare rapidamente il modello. Nota: questa applicazione è stata creata per dimostrare l'idea della chiamata di funzione e NON è pensata per essere perfettamente robusta, ad esempio supportando l'ordine di più elementi contemporaneamente, senza allucinazioni, ecc.
import json
from haystack.dataclasses import ChatMessage, ChatRoleresponse = 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})
Mentre funziona, potremmo voler avere qualcosa che sembri più carino.
Interfaccia semplificata
Streamlit trasforma gli script di dati in app Web condivisibili, che forniscono un'interfaccia utente ordinata per la nostra applicazione. Il codice mostrato sopra è adattato in un'applicazione Streamlit sotto il streamlit cartella del mio repository
Puoi eseguirlo tramite:
- Se non lo hai già fatto, avvia il server API con
python db_api.py - Imposta OPENROUTER_API_KEY come variabile di ambiente, ad es
export OPENROUTER_API_KEY = ‘@REPLACE WITH YOUR API KEY’supponendo che tu sia su Linux/eseguendo con git bash - Passare a
streamlitcartella nel terminale concd streamlit - Esegui Streamlit con
streamlit run app.py. Una nuova scheda dovrebbe essere creata automaticamente nel browser che esegue l'applicazione
Fondamentalmente è tutto! Spero che questo articolo ti piaccia.
*Se non diversamente specificato, tutte le immagini sono dell'autore
Fonte: towardsdatascience.com
