Function Calling ist keine Neuheit. Im Juli 2023 führte OpenAI Function Calling für ihre GPT-Modelle ein, eine Funktion, die nun von Konkurrenten übernommen wird. Die Gemini-API von Google unterstützt sie kürzlich und Anthropic integriert sie in Claude. Function Calling wird für große Sprachmodelle (LLMs) unerlässlich und erweitert deren Fähigkeiten. Umso nützlicher, diese Technik zu erlernen!
Mit diesem Ziel möchte ich ein umfassendes Tutorial erstellen, das über die grundlegende Einführung hinausgeht (es gibt bereits viele Tutorials dafür). Der Fokus liegt auf der praktischen Umsetzung, dem Aufbau eines vollständig autonomen KI-Agenten und der Integration mit Streamlit für eine ChatGPT-ähnliche Oberfläche. Obwohl OpenAI zur Demonstration verwendet wird, kann dieses Tutorial leicht auf andere LLMs angepasst werden, die Function Calling unterstützen, wie Gemini.
Wofür ist Function Calling?
Function Calling ermöglicht Entwicklern, Funktionen (auch Tools oder Aktionen für das Modell genannt, wie Berechnungen durchführen oder Bestellungen aufgeben) zu beschreiben und hat das Modell intelligent entscheiden lassen, ein JSON-Objekt mit Argumenten auszugeben, um diese Funktionen aufzurufen. Einfacher ausgedrückt, erlaubt es:
- Autonome Entscheidungsfindung: Modelle können intelligent Tools zur Beantwortung von Fragen auswählen.
- Zuverlässiges Parsing: Antworten werden im JSON-Format ausgegeben, statt der typischen dialogartigen Antwort. Das mag auf den ersten Blick nicht viel erscheinen, aber genau das erlaubt LLMs die Verbindung zu externen Systemen, beispielsweise über APIs mit strukturierten Eingaben.
Es eröffnet zahlreiche Möglichkeiten:
- Autonome KI-Assistenten: Bots können mit internen Systemen für Aufgaben wie Kundenbestellungen und -retouren interagieren, über die Bereitstellung von Antworten auf Anfragen hinaus.
- Persönliche Recherche-Assistenten: Sagen wir, Sie planen Ihre Reise, dann können Assistenten das Web durchsuchen, Inhalte crawlen, Optionen vergleichen und die Ergebnisse in Excel zusammenfassen.
- IoT-Sprachbefehle: Modelle können Geräte steuern oder Aktionen basierend auf erkannten Absichten vorschlagen, wie die Anpassung der Klimaanlage.
Die Struktur von Function Calling
In Anlehnung an die Gemini-Dokumentation zu Function Calling hat Function Calling die folgende Struktur, die auch bei OpenAI gleich ist:
Bild von der Gemini-Dokumentation zu Function Calling
- Der Nutzer gibt eine Eingabe an die Anwendung
- Die Anwendung übergibt die Nutzereingabe und die Function Declaration(s), eine Beschreibung der Tool(s), die das Modell nutzen könnte
- Basierend auf der Function Declaration schlägt das Modell das zu nutzende Tool und die relevanten Anfrageparameter vor. Beachten Sie, dass das Modell nur das vorgeschlagene Tool und die Parameter ausgibt, OHNE die Funktionen tatsächlich aufzurufen
- & 5. Basierend auf der Antwort ruft die Anwendung die relevante API auf
- & 7. Die Antwort der API wird erneut in das Modell eingespeist, um eine für Menschen lesbare Antwort auszugeben
- Die Anwendung gibt die endgültige Antwort an den Nutzer zurück, dann wird ab 1 wiederholt.
Dies mag kompliziert erscheinen, aber das Konzept wird im Detail mit Beispielen veranschaulicht.
Architektur
Bevor wir in den Code eintauchen, ein paar Worte zur Architektur der Demo-Anwendung.
Lösung
Hier bauen wir einen Assistenten für Touristen auf, die ein Hotel besuchen. Der Assistent hat Zugriff auf die folgenden Tools, die es dem Assistenten ermöglichen, auf externe Anwendungen zuzugreifen.
get_items,purchase_item: Verbindung zum Produktkatalog, der in einer Datenbank über eine API gespeichert ist, um eine Artikelliste abzurufen und einen Kauf zu tätigenrag_pipeline_func: Verbindung zu einer Dokumentensammlung mit Retrieval Augmented Generation (RAG), um Informationen aus unstrukturierten Texten wie Hotelbroschüren zu erhalten
Tech-Stack
- Embedding-Modell: all-MiniLM-L6-v2
- Vektor-Datenbank: Haystacks InMemoryDocumentStore
- LLM: GPT-4 Turbo über OpenRouter zugänglich. Mit OpenRouter können Sie auf verschiedene LLM-APIs aus Hongkong ohne VPN zugreifen. Der Ablauf kann mit geringfügigen Codeänderungen an andere LLMs angepasst werden, vorausgesetzt, sie unterstützen Function Calling, wie Gemini
- LLM-Framework: Haystack wegen der einfachen Handhabung, großartigen Dokumentation und Transparenz bei der Konstruktion von Pipelines. Dieses Tutorial ist eigentlich eine Erweiterung ihres fantastischen Tutorials zum gleichen Thema
Lassen Sie uns beginnen!
Beispiel-Anwendung
Vorbereitung
Wechseln Sie zu Github, um meinen Code zu klonen. Die nachfolgenden Inhalte finden Sie im Notebook function_calling_demo.
Bitte erstellen und aktivieren Sie auch eine virtuelle Umgebung, dann pip install -r requirements.txt, um die erforderlichen Pakete zu installieren.
Initialisierung
Wir verbinden uns zuerst mit OpenRouter. Alternativ würde auch die Verwendung des ursprünglichen OpenAIChatGenerator ohne Überschreiben der api_base_url funktionieren, vorausgesetzt, Sie haben einen OpenAI-API-Schlüssel.
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
# Setzen Sie Ihren API-Schlüssel als Umgebungsvariable, bevor Sie fortfahren
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)
Dann testen wir, ob der chat_generator erfolgreich aufgerufen werden kann:
chat_generator.run(messages=[ChatMessage.from_user("Return this text: 'test'")])
---------- Die Antwort sollte so aussehen ----------
{'replies': [ChatMessage(content="'test'", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'stop', 'usage': {}})]}
Schritt 1: Datenspeicher einrichten
Hier stellen wir die Verbindung zwischen unserer Anwendung und den beiden Datenquellen her: Dokumentenspeicher für unstrukturierte Texte und Anwendungsdatenbank über API.
Indizieren von Dokumenten mit einer Pipeline
Wir stellen Beispieltexte in documents bereit, damit das Modell Retrieval Augmented Generation (RAG) durchführen kann. Die Texte werden in Embeddings umgewandelt und in einem In-Memory-Dokumentenspeicher gespeichert.
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
# Beispieldokumente
documents = [
Document(content="Coffee shop opens at 9am and closes at 5pm."),
Document(content="Gym room opens at 6am and closes at 10pm.")
]
# Erstellen des Dokumentenspeichers
document_store = InMemoryDocumentStore()
# Erstellen einer Pipeline, um die Texte in Embeddings umzuwandeln und im Dokumentenspeicher zu speichern
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}})
Es sollte folgende Ausgabe erscheinen, die den von uns erstellten documents entspricht:
{'doc_writer': {'documents_written': 2}}
API-Server starten
Ein mit Flask erstellter API-Server wird unter db_api.py ausgeführt, um eine Verbindung zu SQLite herzustellen. Bitte starten Sie ihn, indem Sie python db_api.py in Ihrem Terminal ausführen.
Dies würde im Terminal angezeigt, wenn erfolgreich ausgeführt
Beachten Sie auch, dass einige Ausgangsdaten in db_api.py hinzugefügt wurden.
Beispieldaten in der Datenbank
Schritt 2: Funktionen definieren
Hier bereiten wir die tatsächlichen Funktionen vor, die das Modell NACH Function Calling aufrufen kann (Schritt 4-5 wie in Die Struktur von Function Calling beschrieben).
RAG-Funktion
Nämlich die rag_pipeline_func. Damit kann das Modell eine Antwort durch Suche in den im Dokumentenspeicher gespeicherten Texten bereitstellen. Zunächst definieren wir die RAG-Abfrage als Haystack-Pipeline:
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))
# Hinweis zum LLM: Wir verwenden OpenAIGenerator, nicht OpenAIChatGenerator, da Letzterer nur List[str] als Eingabe akzeptiert und den String-Output des prompt_builders nicht verarbeiten kann.
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")
Testen, ob die Funktion funktioniert:
query = "When does the coffee shop open?"
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
Dies sollte folgende Ausgabe liefern. Beachten Sie die replies, die das Modell aus den zuvor bereitgestellten Beispieldokumenten gibt:
{'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}}]}}
Wir können dann die rag_pipe in eine Funktion umwandeln, die nur die replies ohne weitere Details ausgibt:
def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
return {"reply": result["llm"]["replies"][0]}
API-Aufrufe
Wir definieren die Funktionen get_items und purchase_item für die Interaktion mit der Datenbank:
# Flask' Standard-Lokale URL, ändern Sie sie bei Bedarf
db_base_url = 'http://127.0.0.1:5000'
# Verwenden von requests, um die Daten aus der Datenbank abzurufen
import requests
import json
# get_categories wird als Teil des Prompts bereitgestellt, es wird nicht als Tool verwendet
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()
Tool-Liste definieren
Nachdem wir die Funktionen definiert haben, müssen wir dem Modell mitteilen, dass es diese Funktionen erkennen soll, und ihm erklären, wie sie verwendet werden, indem wir Beschreibungen für sie bereitstellen.
Da wir hier OpenAI verwenden, ist tools nach dem von OpenAI geforderten Format formatiert:
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"],
},
},
}
]
Schritt 3: Alles zusammenfügen
Wir haben nun alle erforderlichen Eingaben, um Function Calling zu testen! Hier machen wir ein paar Dinge:
- Wir geben dem Modell den Anfangsprompt, um ihm einen Kontext zu geben
- Wir stellen eine Beispielnachricht eines Nutzers bereit
- Am wichtigsten ist, dass wir die Tool-Liste im
tools-Parameter an den Chat-Generator übergeben
# 1. Anfänglicher 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. Beispielnachricht vom Nutzer
ChatMessage.from_user("Can I buy a coffee?"),
]
# 3. Übergabe der Tool-Liste und Aufruf des Chat-Generators
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
response
---------- Antwort ----------
{'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': {}})]]}
Lassen Sie uns die Antwort inspizieren. Beachten Sie, wie Function Calling sowohl die vom Modell gewählte Funktion als auch die Argumente für den Aufruf der gewählten Funktion zurückgibt.
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)
---------- Antwort ----------
Function Name: get_items
Function Arguments: {'categories': 'Food and beverages'}
Bei einer weiteren Frage wird das Modell ein anderes, relevanters Tool verwenden:
# Weitere Frage
messages.append(ChatMessage.from_user("Where's the coffee shop?"))
# Aufruf des Chat-Generators und Übergabe der Tool-Liste
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)
---------- Antwort ----------
Function Name: rag_pipeline_func
Function Arguments: {'query': "Where's the coffee shop?"}
Beachten Sie erneut, dass hier keine tatsächliche Funktion aufgerufen wird. Das machen wir als Nächstes!
Aufruf der Funktion
Wir können dann die Argumente in die gewählte Funktion einbringen:
## Finden der entsprechenden Funktion und Aufruf mit den gegebenen Argumenten
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)
---------- Antwort ----------
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.'}
Die Antwort von rag_pipeline_func kann dann als Kontext für den Chat an messages angehängt werden, damit das Modell die endgültige Antwort geben kann:
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)
---------- Antwort ----------
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.
Wir haben nun den Chat-Zyklus abgeschlossen!
Schritt 4: In einen interaktiven Chat umwandeln
Der obige Code zeigt, wie Function Calling durchgeführt werden kann, aber wir möchten noch einen Schritt weitergehen und es in einen interaktiven Chat umwandeln.
Hier zeige ich zwei Möglichkeiten, dies zu tun, von der primitiveren input()-Methode, die den Dialog direkt in das Notebook ausgibt, bis hin zur Renderung durch Streamlit, um eine ChatGPT-ähnliche Oberfläche bereitzustellen.
input()-Schleife
Der Code stammt aus dem Haystack-Tutorial und ermöglicht es uns, das Modell schnell zu testen. Hinweis: Diese Anwendung dient der Demonstration der Idee von Function Calling und ist NICHT dafür gedacht, perfekt robust zu sein, z.B. die Bestellung mehrerer Artikel gleichzeitig zu unterstützen, keine Halluzinationen usw.
import json
from haystack.dataclasses import ChatMessage, ChatRole
response = None
messages = [
ChatMessage.from_system(context)
]
while True:
# Falls die OpenAI-Antwort ein Tool-Aufruf ist
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:
## Parsen der Funktion-Aufruf-Informationen
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
## Finden der entsprechenden Funktion und Aufruf mit den gegebenen Argumenten
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
## Anhängen der Funktionsantwort an die messages-Liste mit `ChatMessage.from_function`
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
# Regulärer Dialog
else:
# Assistenten-Nachrichten an die messages-Liste anhängen
if not messages[-1].is_from(ChatRole.SYSTEM):
messages.append(response["replies"][0])
user_input = input("GEBEN SIE IHRE NACHRICHT EIN 👇 INFO: Geben Sie 'exit' oder 'quit' ein, um zu beenden\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})
Ausführung interaktiver Chats in der IDE
Während es funktioniert, wollen wir vielleicht etwas Schöneres haben.
Streamlit-Oberfläche
Streamlit verwandelt Datenskripte in teilbare Web-Apps, die unserer Anwendung eine ordentliche Oberfläche verleihen. Der oben gezeigte Code wurde in eine Streamlit-Anwendung im Ordner streamlit meines Repos angepasst.
Sie können sie wie folgt ausführen:
- Falls noch nicht geschehen, starten Sie den API-Server mit
python db_api.py - Setzen Sie die OPENROUTER_API_KEY als Umgebungsvariable, z.B.
export OPENROUTER_API_KEY='@ERSETZEN SIE DURCH IHREN API-SCHLÜSSEL'unter Linux/Git Bash - Navigieren Sie im Terminal mit
cd streamlitin denstreamlit-Ordner - Starten Sie Streamlit mit
streamlit run app.py. Ein neuer Tab sollte automatisch in Ihrem Browser geöffnet werden und die Anwendung ausführen
Das war’s im Grunde! Ich hoffe, Ihnen hat dieser Artikel gefallen.
