Usando a API de Inferência Elasticsearch junto com modelos de Hugging Face
Aprenda a conectar o Elasticsearch a modelos do Hugging Face usando endpoints de inferência e a construir um sistema multilíngue de recomendação de blogs com busca semântica e conclusões de chat.
Em atualizações recentes, o Elasticsearch introduziu uma integração nativa para conectar a modelos hospedados no Hugging Face Inference Service. Neste post, vamos explorar como configurar essa integração e realizar inferência por meio de chamadas simples de API usando um grande modelo de linguagem (LLM). Vamos usar SmolLM3-3B, um modelo leve de uso geral com bom equilíbrio entre uso de recursos e qualidade da resposta.
Pré-requisitos
Elasticsearch 9.3 ou Elastic Cloud Serverless: você pode criar uma implantação na nuvem seguindo essas instruções, ou você pode usar o
start-localquickstart.Python 3.12: baixe o Python aqui.
Hugging Face Token de acesso.
Chat completions usando um endpoint de inferência do Hugging Face
Primeiro, vamos construir um exemplo prático que conecta o Elasticsearch a um endpoint de inferência Hugging Face para gerar recomendações baseadas em IA a partir de uma coleção de artigos de blog. Para a base de conhecimento do app, usaremos um conjunto de dados de artigos de blog da empresa, que contém informações valiosas, mas frequentemente difíceis de navegar.
Com este endpoint, a busca semântica recupera os artigos mais relevantes para uma consulta específica, e um Hugging Face LLM gera recomendações curtas e contextuais com base nesses resultados.
Vamos dar uma olhada em uma visão geral do fluxo de informações que vamos criar:
Neste artigo, testaremos a capacidade do SmolLM3-3B de combinar seu tamanho compacto com fortes capacidades de raciocínio multíngue e chamada de ferramentas. Com base em uma consulta de busca, enviaremos todo o conteúdo correspondente (em inglês e espanhol) para o LLM para gerar uma lista de artigos recomendados com uma descrição personalizada com base na consulta de busca e nos resultados.
Veja como poderia ser a UI de um site de artigos com um sistema de geração de recomendações por IA.
Você pode encontrar a implementação completa desta aplicação no notebook vinculado.
Configuração de endpoints de inferência do Elasticsearch
Para usar o endpoint de inferência do Hugging Face no Elasticsearch, precisamos de dois elementos importantes: uma chave de API do Hugging Face e uma URL de endpoint do Hugging Face em execução. Ela deverá ficar assim:
PUT _inference/chat_completions/hugging-face-smollm3-3b
{
"service": "hugging_face",
"service_settings": {
"api_key": "hugging-face-access-token",
"url": "url-endpoint"
}
}O endpoint de inferência Hugging Face no Elasticsearch permite diferentes tipos de tarefas: text_embedding, completion, chat_completion, e rerank. Neste post do blog, usamos chat_completion porque precisamos que o modelo gere recomendações conversacionais baseadas nos resultados de busca e em um prompt do sistema. Esse endpoint nos permite realizar preenchimentos de chat diretamente do Elasticsearch de forma simples usando a API do Elasticsearch:
POST _inference/chat_completion/hugging-face-smollm3-3b/_stream
{
"messages": [
{ "role": "user", "content": "<user prompt>" }
]
}Isso servirá como o núcleo da aplicação, recebendo o prompt e os resultados de busca que passarão pelo modelo. Com a teoria explicada, vamos começar a implementar a aplicação.
Configurando o endpoint de inferência no Hugging Face
Para implantar o modelo Hugging Face, vamos usar implantações Hugging Face One-Click, um serviço fácil e rápido para implantar endpoints de modelos. Lembre-se de que este é um serviço pago, e seu uso pode incorrer em custos adicionais. Esta etapa criará a instância do modelo que será usada para gerar as recomendações dos artigos.
Você pode escolher um modelo do catálogo de um clique.
Vamos escolher o modelo SmolLM3-3B :
A partir daqui, pegue o URL do endpoint do Hugging Face:
Como mencionado na documentação de endpoints de inferência Hugging Face do Elasticsearch, a geração de texto requer um modelo compatível com a API OpenAI. Por esse motivo, precisamos anexar o subcaminho /v1/chat/completions à URL do endpoint Hugging Face. O resultado final ficará assim:
https://j2g31h0futopfkli.us-east-1.aws.endpoints.huggingface.cloud/v1/chat/completionsCom isso pronto, podemos começar a programar em um notebook Python.
Gerando a Chave API do Hugging Face
Crie uma conta Hugging Face e obtenha um token de API seguindo estas instruções. Você pode escolher entre três tipos de token: detalhado (recomendado para produção, pois fornece acesso apenas a recursos específicos); de leitura (para acesso somente leitura); ou de gravação (para acesso de leitura e gravação). Para este tutorial, um token de leitura é suficiente, já que só precisamos chamar o endpoint de inferência. Guarde esta chave para o próximo passo.
Configurando o endpoint de inferência do Elasticsearch
Primeiro, vamos declarar um cliente Elasticsearch Python:
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"]
)Em seguida, vamos criar um endpoint de inferência no Elasticsearch que use o modelo Hugging Face. Esse endpoint nos permitirá gerar respostas com base nos posts do blog e no prompt passado para o modelo.
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"],
},
},
)Conjunto de dados
O conjunto de dados contém os posts do blog que serão consultados, representando um conjunto de conteúdo multilíngue usado em todo o fluxo de trabalho:
// 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."
}Mapeamento do Elasticsearch
Com o conjunto de dados definido, precisamos criar um esquema de dados que se ajuste adequadamente à estrutura do post do blog. Os seguintes mapeamentos de índice serão usados para armazenar os dados no Elasticsearch:
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)Aqui, podemos ver com mais clareza como os dados são estruturados. Usaremos busca semântica para recuperar resultados baseados em linguagem natural, junto com a propriedade copy_to para copiar o conteúdo do campo para o campo semantic_text. Além disso, o campo title contém dois subcampos: o subcampo original armazena o título em inglês ou espanhol, dependendo do idioma original do artigo; e o subcampo translated_title está presente apenas para artigos em espanhol e contém a tradução para o inglês do título original.
Ingestão de dados
O seguinte trecho de código ingere o conjunto de dados de postagens do blog no Elasticsearch usando a 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)}")Agora que os artigos já estão no Elasticsearch, precisamos criar uma função capaz de buscar no campo semantic_text:
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 []Precisamos também de uma função que chame o endpoint de inferência. Neste caso, chamaremos o endpoint usando chat_completion tipo de tarefa para obter respostas de streaming:
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)}"Agora podemos escrever uma função que chama a função de busca semântica, junto com o endpoint de inferência chat_completions e o endpoint de recomendações, para gerar os dados que serão alocados nos cartões:
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_responseFinalmente, precisamos extrair as informações e formatá-las para serem impressas:
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 + "┘")Vamos testar isso fazendo uma pergunta sobre as postagens do blog de segurança:
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)Aqui podemos ver os cartões no console gerados pelo fluxo de trabalho:
Você pode ver os resultados completos, incluindo todos os acertos e a resposta do LLM, neste arquivo.
Estamos pedindo artigos relacionados a: "Segurança e vulnerabilidades." Esta pergunta é usada como consulta de busca nos documentos armazenados no Elasticsearch. Os resultados recuperados são então passados para o modelo, que gera recomendações com base em seu conteúdo. Como podemos ver, o modelo fez um ótimo trabalho criando textos curtos envolventes que podem motivar o leitor a clicar.
Conclusão
Este exemplo mostra como Elasticsearch e Hugging Face podem ser combinados para criar um sistema centralizado rápido e eficiente para aplicações de IA. Essa abordagem reduz o esforço manual e oferece flexibilidade, graças ao extenso catálogo de modelos da Hugging Face. O uso do SmolLM3-3B, em particular, demonstra como modelos compactos e multilíngues ainda podem fornecer raciocínio significativo e geração de conteúdo quando combinados com busca semântica. Juntas, essas ferramentas oferecem uma base escalável e eficaz para construir análises inteligentes de conteúdo e aplicações multilíngues.
