使用 Elasticsearch 推理 API 以及 Hugging Face 模型
了解如何使用推理终端将 Elasticsearch 连接到 Hugging Face 模型，并利用语义搜索和聊天补全功能构建多语言博客推荐系统。
在最近的更新中，Elasticsearch 引入了原生集成，用于连接到托管在 Hugging Face Inference Service 上的模型。在本文中，我们将探讨如何配置此集成，并使用大型语言模型 (LLM) 通过简单的 API 调用执行推理。我们将使用 SmolLM3-3B，这是一款轻量级通用模型，在资源使用和答案质量之间取得了良好的平衡。
准备工作
Elasticsearch 9.3 或 Elastic Cloud Serverless：您可以按照这些说明创建云部署，或者改用
start-local快速入门。Python 3.12：在此处下载 Python。
Hugging Face 访问令牌。
使用 Hugging Face 推理终端完成聊天
首先，我们将构建一个实用示例，将 Elasticsearch 连接到 Hugging Face 推理终端，以从博客文章集合中生成 AI 驱动的推荐。对于应用知识库，我们将使用公司博客文章数据集，其中包含有价值但通常难以查找的信息。
通过这个终端，语义搜索可以检索与给定查询最相关的文章，而 Hugging Face LLM 则会根据这些结果生成简短的上下文推荐。
让我们来看看我们将要构建的信息流的高级概述：
在本文中，我们将测试 SmolLM3-3B 是否能将其紧凑的大小与强大的多语言推理和工具调用能力相结合。根据搜索查询，我们将把所有匹配的内容（英语和西班牙语）发送到 LLM，以生成一份推荐文章列表，并根据搜索查询和结果提供自定义描述。
以下是具备 AI 推荐生成系统的文章网站用户界面可能的外观。
您可以在已链接的笔记本中找到此应用程序的完整实现。
配置 Elasticsearch 推理终端
要使用 Elasticsearch Hugging Face 推理终端，我们需要两个重要元素：Hugging Face API 密钥和正在运行的 Hugging Face 终端 URL。它应该如下所示：
PUT _inference/chat_completions/hugging-face-smollm3-3b
{
"service": "hugging_face",
"service_settings": {
"api_key": "hugging-face-access-token",
"url": "url-endpoint"
}
}Elasticsearch 中的 Hugging Face 推理终端支持不同的任务类型：text_embedding、completion、chat_completion 和 rerank。在这篇博客文章中，我们使用 chat_completion 是因为我们需要模型根据搜索结果和系统提示生成对话式推荐。此终端允许我们使用 Elasticsearch API 以简单的方式直接从 Elasticsearch 执行聊天完成：
POST _inference/chat_completion/hugging-face-smollm3-3b/_stream
{
"messages": [
{ "role": "user", "content": "<user prompt>" }
]
}这将作为应用程序的核心，接收通过模型传递的提示和搜索结果。有了理论基础，我们就开始实施应用程序。
在 Hugging Face 上设置推理终端
要部署 Hugging Face 模型，我们将使用 Hugging Face 一键式部署，这是一种用于部署模型终端的简单快速的服务。请记住，这是一项付费服务，使用它可能会产生额外费用。此步骤将创建用于生成文章推荐的模型实例。
您可以从一键目录中选择一个模型：
让我们选择 SmolLM3-3B 模型：
从此处获取 Hugging Face 终端 URL：
正如在 Elasticsearch Hugging Face 推理终端文档中提到的，文本生成需要一个与 OpenAI API 兼容的模型。因此，我们需要将 /v1/chat/completions 子路径附加到 Hugging Face 终端 URL。最终结果将如下所示：
https://j2g31h0futopfkli.us-east-1.aws.endpoints.huggingface.cloud/v1/chat/completions有了这个，我们就可以在 Python 笔记本中开始编码了。
生成 Hugging Face API 密钥
创建 Hugging Face 账户，并按照以下说明获取 API 令牌。您可以选择三种令牌类型：细粒度（推荐用于生产，因为它仅提供对特定资源的访问）、读取（适用于只读访问）或写入（适用于读取和写入访问）。在本教程中，读取令牌就足够了，因为我们只需要调用推理终端。请保存此密钥以备下一步使用。
设置 Elasticsearch 推理终端
首先，让我们声明一个 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"]
)接下来，我们创建一个使用 Hugging Face 模型的 Elasticsearch 推理终端。此终端将允许我们基于博客文章和传递给模型的提示来生成响应。
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"],
},
},
)数据集
该数据集包含将要查询的博客文章，代表整个工作流中使用的多语言内容集：
// 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 映射
定义数据集后，我们需要创建一个适合博客文章结构的数据模式。以下索引映射将用于在 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)在这里，我们可以更清楚地看到数据的结构。我们将使用语义搜索来检索基于自然语言的结果，同时使用 copy_to 属性将字段内容复制到 semantic_text 字段中。此外，title 字段包含两个子字段：original 子字段根据文章的原始语言存储英语或西班牙语标题；而 translated_title 子字段仅存在于西班牙语文章中，并包含原始标题的英语翻译。
采集数据
以下代码片段使用批量 API 将博客文章数据集摄取到 Elasticsearch 中：
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)}")现在，我们已将文章摄取到 Elasticsearch 中，我们需要创建一个能够针对 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 []我们还需要一个调用推理终端的函数。在这种情况下，我们将使用 chat_completion 任务类型调用终端，以获取流式响应：
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)}"现在，我们可以编写一个函数，调用语义搜索函数以及 chat_completions 推理终端和建议终端，以生成将分配到卡片中的数据：
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_response最后，我们需要提取信息并将其格式化以便打印：
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 + "┘")让我们通过询问一个有关安全博客文章的问题来测试一下：
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)如下所示，我们可以看到工作流在控制台中生成的卡片：
您可以在此文件中查看全部结果，包括所有点击和 LLM 响应。
我们正在征集与“安全与漏洞”相关的文章。此问题将用作针对 Elasticsearch 中存储的文档的搜索查询。然后将检索到的结果传递给模型，该模型根据这些结果的内容生成推荐。我们可以看到，该模型出色地生成了引人入胜的短文本，能够激发读者点击的欲望。
结论
本示例展示了如何将 Elasticsearch 和 Hugging Face 结合起来，为 AI 应用程序创建一个快速高效的集中式系统。由于 Hugging Face 拥有丰富的模型目录，这种方法不仅减少了人工操作，还具有灵活性。通过使用 SmolLM3-3B，我们特别看到了紧凑的多语言模型在与语义搜索搭配使用时，仍能提供有意义的推理和内容生成。这些工具共同为构建智能内容分析和多语言应用程序提供了可扩展且高效的基础。
