Die Verwendung der Elasticsearch Inference API zusammen mit Hugging Face-Modellen
Erfahren Sie, wie Sie Elasticsearch mithilfe von Inferenz-Endpoints mit Hugging Face Modellen verbinden und ein mehrsprachiges Blog-Empfehlungssystem mit semantischer Suche und Chat-Abschlüssen erstellen.
In den letzten Aktualisierungen hat Elasticsearch eine native Integration eingeführt, um sich mit Modellen zu verbinden, die auf dem Hugging Face Inference Service gehostet werden. In diesem Beitrag erfahren Sie, wie Sie diese Integration konfigurieren und über einfache API-Aufrufe mithilfe eines großen Sprachmodells (LLM) Inferenzen durchführen können. Wir verwenden SmolLM3-3B, ein leichtes Allzweckmodell mit einem ausgewogenen Gleichgewicht zwischen Ressourcenverbrauch und Antwortqualität.
Voraussetzungen
Elasticsearch 9.3 oder Elastic Cloud Serverless: Sie können ein Cloud-Deployment erstellen, indem Sie diese Anweisungen befolgen, oder stattdessen den
start-localQuickstart verwenden.Python 3.12: Laden Sie Python hier herunter.
Hugging Face Zugriffstoken.
Chat-Abschlüsse unter Verwendung eines Inferenz-Endpoints von Hugging Face
Zuerst erstellen wir ein praktisches Beispiel, das Elasticsearch mit einem Hugging Face Inferenz-Endpoint verbindet, um KI-gestützte Empfehlungen aus einer Reihe von Blogbeiträgen zu generieren. Für die Wissensdatenbank der App verwenden wir einen Datensatz mit Blogartikeln des Unternehmens, der wertvolle, aber oft schwer zugängliche Informationen enthält.
Bei diesem Endpoint ruft die semantische Suche die relevantesten Artikel für eine gegebene Abfrage ab, und ein Hugging Face LLM generiert kurze, kontextuelle Empfehlungen basierend auf diesen Ergebnissen.
Verschaffen wir uns einen groben Überblick über den Informationsfluss, den wir entwickeln werden:
In diesem Artikel testen wir die Fähigkeit von SmolLM3-3B und kombinieren seine kompakte Größe mit einer starken mehrsprachigen Argumentations- und Tool-Aufruffunktion. Basierend auf einer Suchabfrage senden wir alle passenden Inhalte (auf Englisch und Spanisch) an das LLM, um eine Liste empfohlener Artikel mit einer individuell erstellten Beschreibung basierend auf der Suchanfrage und den Ergebnissen zu erstellen.
So könnte die Benutzeroberfläche einer Artikelseite mit einem System zur Generierung von KI-Empfehlungen aussehen.
Sie können die vollständige Implementierung dieser Anwendung im verlinkten Notizbuch finden.
Konfiguration der Elasticsearch Inferenz-Endpoints
Um den Elasticsearch Hugging Face Inferenz-Endpoint zu verwenden, benötigen wir zwei wichtige Elemente: einen Hugging Face API-Schlüssel und eine ausgeführte Hugging Face Endpoint-URL. Dies sollte so aussehen:
PUT _inference/chat_completions/hugging-face-smollm3-3b
{
"service": "hugging_face",
"service_settings": {
"api_key": "hugging-face-access-token",
"url": "url-endpoint"
}
}Der Hugging Face Inferenz-Endpoint in Elasticsearch unterstützt verschiedene Aufgabentypen: text_embedding, completion, chat_completion und rerank. In diesem Blogbeitrag verwenden wir chat_completion, weil das Modell Gesprächsempfehlungen basierend auf den Suchergebnissen und einem Systemprompt generieren kann. Dieser Endpoint ermöglicht es uns, Chatabschlüsse direkt von Elasticsearch aus auf einfache Weise mit der Elasticsearch-API durchzuführen:
POST _inference/chat_completion/hugging-face-smollm3-3b/_stream
{
"messages": [
{ "role": "user", "content": "<user prompt>" }
]
}Dies dient als Kern der Anwendung, der den Prompt und die Suchergebnisse empfängt, die durch das Modell laufen werden. Nachdem die Theorie geklärt ist, können wir mit der Implementierung der Anwendung beginnen.
Einrichten des Inferenz-Endpoints auf Hugging Face
Um das Hugging Face Modell bereitzustellen, werden wir Hugging Face One-Click-Deployments verwenden, einen einfachen und schnellen Service für die Bereitstellung von Modell-Endpoints. Beachten Sie bitte, dass es sich um einen kostenpflichtigen Service handelt, durch dessen Nutzung zusätzliche Kosten entstehen können. In diesem Schritt wird die Modellinstanz erstellt, die zur Generierung der Empfehlungen für die Artikel verwendet wird.
Sie können ein Modell aus dem Ein-Klick-Katalog aussuchen:
Wählen Sie das SmolLM3-3B Modell:
Hier können Sie die URL des Hugging Face Endpoints abrufen:
Wie in der Dokumentation zu den Hugging Face Inferenz-Endpunkten von Elasticsearch erwähnt, erfordert die Textgenerierung ein Modell, das mit der OpenAI-API kompatibel ist. Aus diesem Grund müssen wir den /v1/chat/completions-Subpfad an die Hugging Face Endpoint-URL anhängen. Das Ergebnis sieht wie folgt aus:
https://j2g31h0futopfkli.us-east-1.aws.endpoints.huggingface.cloud/v1/chat/completionsMit dieser Voraussetzung können wir mit dem Codieren in einem Python-Notebook beginnen.
API-Schlüssel für Hugging Face generieren
Erstellen Sie ein Hugging Face Konto und erhalten Sie ein API-Token, indem Sie diesen Anweisungen folgen. Man kann zwischen drei Token-Typen wählen: detailliert (empfohlen für die Produktion, da es nur Zugriff auf bestimmte Ressourcen bietet); Lesezugriff (schreibgeschützt); oder Schreibzugriff (für Lese- und Schreibzugriff). Für dieses Tutorial ist ein Lesezugriffstoken ausreichend, da wir nur den Inferenz-Endpoint aufrufen müssen. Speichern Sie diesen Code für den nächsten Schritt.
Einrichten des Elasticsearch Inferenz-Endpoints
Zunächst legen wir einen Elasticsearch-Python-Client fest:
os.environ["ELASTICSEARCH_API_KEY"] = "your-elasticsearch-api-key"
os.environ["ELASTICSEARCH_URL"] = "https://xxxx.us-central1.gcp.cloud.es.io:443"
es_client = Elasticsearch(
os.environ["ELASTICSEARCH_URL"], api_key=os.environ["ELASTICSEARCH_API_KEY"]
)Danach erstellen wir einen Elasticsearch Inferenz-Endpoint, der das Hugging Face Modell verwendet. Dieser Endpoint ermöglicht es uns, Reaktionen zu generieren, die auf den Blogbeiträgen und dem an das Modell übergebenen Prompt basieren.
INFERENCE_ENDPOINT_ID = "smollm3-3b-pnz"
os.environ["HUGGING_FACE_INFERENCE_ENDPOINT_URL"] = (
"https://j2g31h0futopfkli.us-east-1.aws.endpoints.huggingface.cloud/v1/chat/completions"
)
os.environ["HUGGING_FACE_API_KEY"] = "hf_xxxxx"
resp = es_client.inference.put(
task_type="chat_completion",
inference_id=INFERENCE_ENDPOINT_ID,
body={
"service": "hugging_face",
"service_settings": {
"api_key": os.environ["HUGGING_FACE_API_KEY"],
"url": os.environ["HUGGING_FACE_INFERENCE_ENDPOINT_URL"],
},
},
)Datensatz
Der Datensatz enthält die Blogbeiträge, die abgefragt werden, und repräsentiert einen mehrsprachigen Inhaltssatz, der im gesamten Workflow verwendet wird:
// Articles dataset document example:
{
"id": "6",
"title": "Complete guide to the new API: Endpoints and examples",
"author": "Tomas Hernandez",
"date": "2025-11-06",
"category": "tutorial",
"content": "This guide describes in detail all endpoints of the new API v2. It includes code examples in Python, JavaScript, and cURL for each endpoint. We cover authentication, resource creation, queries, updates, and deletion. We also explain error handling, rate limiting, and best practices. Complete documentation is available on our developer portal."
}Elasticsearch-Mappings
Mit dem definierten Datensatz müssen wir ein Datenschema erstellen, das der Struktur des Blogbeitrags entspricht. Die folgenden Index-Mappings werden verwendet, um die Daten in Elasticsearch zu speichern:
INDEX_NAME = "blog-posts"
mapping = {
"mappings": {
"properties": {
"id": {"type": "keyword"},
"title": {
"type": "object",
"properties": {
"original": {
"type": "text",
"copy_to": "semantic_field",
"fields": {"keyword": {"type": "keyword"}},
},
"translated_title": {
"type": "text",
"fields": {"keyword": {"type": "keyword"}},
},
},
},
"author": {"type": "keyword", "copy_to": "semantic_field"},
"category": {"type": "keyword", "copy_to": "semantic_field"},
"content": {"type": "text", "copy_to": "semantic_field"},
"date": {"type": "date"},
"semantic_field": {"type": "semantic_text"},
}
}
}
es_client.indices.create(index=INDEX_NAME, body=mapping)Hier können wir klar erkennen, wie die Daten strukturiert sind. Wir werden die semantische Suche verwenden, um Ergebnisse basierend auf natürlicher Sprache abzurufen, zusammen mit der copy_to-Eigenschaft, um den Feldinhalt in das semantic_text-Feld zu kopieren. Zusätzlich enthält das title-Feld zwei Unterfelder: Das original-Unterfeld speichert den Titel entweder auf Englisch oder Spanisch, abhängig von der Originalsprache des Artikels, und das translated_title-Unterfeld ist nur für spanische Artikel vorhanden und enthält die englische Übersetzung des Originaltitels.
Ingestieren von Daten
Der folgende Code-Schnipsel überträgt den Datensatz des Blogbeitrags in Elasticsearch mithilfe der Bulk-API:
def build_data(json_file, index_name):
with open(json_file, "r") as f:
data = json.load(f)
for doc in data:
action = {"_index": index_name, "_source": doc}
yield action
try:
success, failed = helpers.bulk(
es_client,
build_data("dataset.json", INDEX_NAME),
)
print(f"{success} documents indexed successfully")
if failed:
print(f"Errors: {failed}")
except Exception as e:
print(f"Error: {str(e)}")Nun, da wir die Artikel in Elasticsearch aufgenommen haben, müssen wir eine Funktion erstellen, die nach dem Feld semantic_text sucht:
def perform_semantic_search(query_text, index_name=INDEX_NAME, size=5):
try:
query = {
"query": {
"match": {
"semantic_field": {
"query": query_text,
}
}
},
"size": size,
}
response = es_client.search(index=index_name, body=query)
hits = response["hits"]["hits"]
return hits
except Exception as e:
print(f"Semantic search error: {str(e)}")
return []Wir benötigen außerdem eine Funktion, die den Endpoint aufruft. In diesem Fall rufen wir den Endpoint mit dem chat_completion Aufgabentyp auf, um Streaming-Antworten zu erhalten:
def stream_chat_completion(messages: list, inference_id: str = INFERENCE_ENDPOINT_ID):
url = f"{ELASTICSEARCH_URL}/_inference/chat_completion/{inference_id}/_stream"
payload = {"messages": messages}
headers = {
"Authorization": f"ApiKey {ELASTICSEARCH_API_KEY}",
"Content-Type": "application/json",
}
try:
response = requests.post(url, json=payload, headers=headers, stream=True)
response.raise_for_status()
for line in response.iter_lines(decode_unicode=True):
if line:
line = line.strip()
if line.startswith("event:"):
continue
if line.startswith("data: "):
data_content = line[6:]
if not data_content.strip() or data_content.strip() == "[DONE]":
continue
try:
chunk_data = json.loads(data_content)
if "choices" in chunk_data and len(chunk_data["choices"]) > 0:
choice = chunk_data["choices"][0]
if "delta" in choice and "content" in choice["delta"]:
content = choice["delta"]["content"]
if content:
yield content
except json.JSONDecodeError as json_err:
print(f"\nJSON decode error: {json_err}")
print(f"Problematic data: {data_content}")
continue
except requests.exceptions.RequestException as e:
yield f"Error: {str(e)}"Nun können wir eine Funktion schreiben, die die semantische Suchfunktion, den chat_completions Inferenz-Endpoint und den Empfehlungs-Endoint aufruft, um die Daten zu generieren, die in den Karten angezeigt werden:
def recommend_articles(search_query, index_name=INDEX_NAME, max_articles=5):
print(f"\n{'='*80}")
print(f"🔍 Search Query: {search_query}")
print(f"{'='*80}\n")
articles = perform_semantic_search(search_query, index_name, size=max_articles)
if not articles:
print("❌ No relevant articles found.")
return None, None
print(f"✅ Found {len(articles)} relevant articles\n")
# Build context with found articles
context = "Available blog articles:\n\n"
for i, article in enumerate(articles, 1):
source = article.get("_source", article)
context += f"Article {i}:\n"
context += f"- Title: {source.get('title', 'N/A')}\n"
context += f"- Author: {source.get('author', 'N/A')}\n"
context += f"- Category: {source.get('category', 'N/A')}\n"
context += f"- Date: {source.get('date', 'N/A')}\n"
context += f"- Content: {source.get('content', 'N/A')}\n\n"
system_prompt = """You are an expert content curator that recommends blog articles.
Write recommendations in a conversational style starting with phrases like:
- "If you're interested in [topic], this article..."
- "This post complements your search with..."
- "For those looking into [topic], this article provides..."
FORMAT REQUIREMENTS:
- Return ONLY a JSON array
- Each element must have EXACTLY these three fields: "article_number", "title", "recommendation"
- If the original title is in spanish, use the "translated_title" subfield in the "title" field
Keep each recommendation concise (2-3 sentences max) and focused on VALUE to the reader.
EXAMPLE OF CORRECT FORMAT:
[
{"article_number": 1, "title": "Article title in english", "recommendation": "If you are interested in [topic], this article provides..."},
{"article_number": 2, "title": "Article title in english", "recommendation": " for those looking into [topic], this article provides..."}
]
Return ONLY the JSON array following this exact structure."""
user_prompt = f"""Search query: "{search_query}"
Generate recommendations for the following articles: {context}
"""
messages = [
{"role": "system", "content": "/no_think"},
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
# LLM generation
print(f"{'='*80}")
print("🤖 Generating personalized recommendations...\n")
full_response = ""
for chunk in stream_chat_completion(messages):
print(chunk, end="", flush=True)
full_response += chunk
return context, articles, full_responseAbschließend müssen wir die Informationen extrahieren und formatieren, um sie zu drucken:
def display_recommendation_cards(articles, recommendations_text):
print("\n" + "=" * 100)
print("📇 RECOMMENDED ARTICLES".center(100))
print("=" * 100 + "\n")
# Parse JSON recommendations - clean tags and extract JSON
recommendations_list = []
try:
# Clean up <think> tags
cleaned_text = re.sub(
r"<think>.*?</think>", "", recommendations_text, flags=re.DOTALL
)
# Remove markdown code blocks ( ... ``` or ``` ... ```)
cleaned_text = re.sub(r"```(?:json)?", "", cleaned_text)
cleaned_text = cleaned_text.strip()
parsed = json.loads(cleaned_text)
# Extract recommendations from list format
for item in parsed:
article_number = item.get("article_number")
title = item.get("title", "")
rec_text = item.get("recommendation", "")
if article_number and rec_text:
recommendations_list.append(
{
"article_number": article_number,
"title": title,
"recommendation": rec_text,
}
)
except json.JSONDecodeError as e:
print(f"⚠️ Could not parse recommendations as JSON: {e}")
return
for i, article in enumerate(articles, 1):
source = article.get("_source", article)
# Card border
print("┌" + "─" * 98 + "┐")
# Find recommendation and title for this article number
recommendation = None
title = None
for rec in recommendations_list:
if rec.get("article_number") == i:
recommendation = rec.get("recommendation")
title = rec.get("title")
break
# Print title
title_lines = textwrap.wrap(f"📌 {title}", width=94)
for line in title_lines:
print(f"│ {line}".ljust(99) + "│")
# Card border
print("├" + "─" * 98 + "┤")
# Print recommendation
if recommendation:
recommendation_lines = textwrap.wrap(recommendation, width=94)
for line in recommendation_lines:
print(f"│ {line}".ljust(99) + "│")
# Card bottom
print("└" + "─" * 98 + "┘")Wir führen einen Test durch, indem wir eine Frage zu den Sicherheitsblogbeiträgen stellen:
search_query = "Security and vulnerabilities"
context, articles, recommendations = recommend_articles(search_query)
print("\nElasticsearch context:\n", context)
# Display visual cards
display_recommendation_cards(articles, recommendations)Hier sehen wir die vom Workflow in der Konsole generierten Karten:
Die vollständigen Ergebnisse, einschließlich aller Treffer und der LLM-Reaktion, können Sie in dieser Datei sehen.
Wir bitten um Artikel zum Thema: „Sicherheit und Schwachstellen“. Diese Frage wird als Suchabfrage gegen die in Elasticsearch gespeicherten Dokumente verwendet. Die abgerufenen Ergebnisse werden dann an das Modell weitergegeben, das basierend auf ihrem Inhalt Empfehlungen generiert. Wie wir sehen können, hat das Modell gute Arbeit geleistet und einen ansprechenden kurzen Text erstellt, der den Leser zum Anklicken motivieren kann.
Fazit
Dieses Beispiel zeigt, wie Elasticsearch und Hugging Face kombiniert werden können, um ein schnelles und effizientes zentrales System für KI-Anwendungen zu schaffen. Dieser Ansatz reduziert den manuellen Aufwand und bietet dank des umfangreichen Modellkatalogs von Hugging Face mehr Flexibilität. Insbesondere die Verwendung von SmolLM3-3B zeigt, wie kompakte, mehrsprachige Modelle in Kombination mit semantischer Suche weiterhin sinnvolles Schlussfolgern und Content-Generierung liefern können. Zusammen bieten diese Tools eine skalierbare und effektive Grundlage für die Entwicklung intelligenter Inhaltsanalysen und mehrsprachiger Anwendungen.
