计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
elasticsearch-sdk【效率与协作】Integrates Elasticsearch using elasticsearch-py 8.x with patterns for indexing, search queries, aggregations, vector search, bulk operations, and index lifecycle management.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: elasticsearch-sdk description: Integrates Elasticsearch using elasticsearch-py 8.x with patterns for indexing, search queries, aggregations, vector search, bulk operations, and index lifecycle management. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: elasticsearch, elasticsearch-py, opensearch, full-text search, elastic aggregations, how do i search data from python, elk stack, vector search archetypes:
Integrates Elasticsearch using elasticsearch-py 8.x (also compatible with OpenSearch) with patterns for index management, search queries (term, match, bool), aggregations, bulk indexing, vector/kNN search, index lifecycle management (ILM), and async operations.
Elasticsearch() client with basic_auth or api_key for authenticationhelpers.bulk() for high-throughput indexing — never index documents one-by-onebool query with must/filter/should clauses for complex searchaggs for bucket and metric aggregations, not client-side groupingknn query for vector similarity search (Elasticsearch 8.x+)index.create() with explicit mappings — never rely on dynamic mapping for productionpoint_in_time (PIT) + search_after for deep pagination instead of from/sizeUse this skill when:
from elasticsearch import Elasticsearch, helpers
from elasticsearch.exceptions import (
ConnectionError as ESConnectionError,
NotFoundError,
AuthorizationException,
)
# Elastic Cloud or self-managed cluster
es = Elasticsearch(
"https://localhost:9200",
basic_auth=("elastic", os.environ["ES_PASSWORD"]),
ca_certs="/path/to/ca.crt",
request_timeout=30,
max_retries=3,
retry_on_timeout=True,
)
# Verify connectivity
if not es.ping():
raise RuntimeError("Cannot connect to Elasticsearch")
Checkpoint: Always verify with es.ping() or es.info() at startup. Catch AuthorizationException immediately on auth failures — don't proceed with invalid credentials.
def create_articles_index(es: Elasticsearch, index_name: str) -> dict:
"""Create an index with explicit mappings and settings."""
mappings = {
"properties": {
"title": {
"type": "text",
"fields": {"keyword": {"type": "keyword"}},
"analyzer": "english",
},
"content": {"type": "text", "analyzer": "english"},
"author": {"type": "keyword"},
"published_at": {"type": "date"},
"tags": {"type": "keyword"},
"view_count": {"type": "integer"},
"embedding": {
"type": "dense_vector",
"dims": 768,
"index": True,
"similarity": "cosine",
},
}
}
settings = {
"number_of_shards": 3,
"number_of_replicas": 1,
"refresh_interval": "30s",
}
if not es.indices.exists(index=index_name):
return es.indices.create(index=index_name, mappings=mappings, settings=settings)
return {"acknowledged": True, "existing": True}
Checkpoint: Always define explicit mappings before indexing production data. Use keyword type for exact-match fields and text with analyzers for full-text search. Add dense_vector fields for vector search.
def bulk_index_articles(es: Elasticsearch, index_name: str, articles: list[dict]) -> dict:
"""Bulk index articles with error handling."""
def generate_actions():
for article in articles:
yield {
"_index": index_name,
"_id": article.get("id"),
"_source": {
"title": article["title"],
"content": article["content"],
"author": article["author"],
"published_at": article["published_at"],
"tags": article.get("tags", []),
"view_count": article.get("view_count", 0),
"embedding": article.get("embedding"),
},
}
success, errors = helpers.bulk(
es,
generate_actions(),
chunk_size=500,
request_timeout=60,
raise_on_error=False,
)
return {"indexed": success, "errors": len(errors)}
Checkpoint: Use raise_on_error=False and inspect errors for failures. Use chunk_size to balance throughput and memory. Always check errors count after bulk operations.
def search_articles(
es: Elasticsearch,
index_name: str,
query_text: str,
author: str | None = None,
tags: list[str] | None = None,
min_date: str | None = None,
from_: int = 0,
size: int = 20,
) -> dict:
"""Full-text search with optional filters."""
must_clauses = [{"match": {"content": query_text}}]
filter_clauses = []
if author:
filter_clauses.append({"term": {"author": author}})
if tags:
filter_clauses.append({"terms": {"tags": tags}})
if min_date:
filter_clauses.append({"range": {"published_at": {"gte": min_date}}})
body = {
"query": {
"bool": {
"must": must_clauses,
"filter": filter_clauses,
}
},
"sort": [{"_score": "desc"}, {"published_at": "desc"}],
"from_": from_,
"size": size,
"aggs": {
"by_author": {"terms": {"field": "author", "size": 10}},
"date_histogram": {"date_histogram": {"field": "published_at", "calendar_interval": "month"}},
},
}
return es.search(index=index_name, body=body)
Checkpoint: Use filter context for structured data (dates, keywords) — filter clauses are cached and don't affect scoring. Use must for full-text relevance scoring.
def vector_search(
es: Elasticsearch,
index_name: str,
query_vector: list[float],
k: int = 10,
num_candidates: int = 100,
) -> list[dict]:
"""Perform kNN vector search for semantic similarity."""
body = {
"knn": {
"field": "embedding",
"query_vector": query_vector,
"k": k,
"num_candidates": num_candidates,
},
"_source": {"excludes": ["embedding"]},
}
response = es.search(index=index_name, body=body)
return [hit["_source"] for hit in response["hits"]["hits"]]
def scroll_all_results(es: Elasticsearch, index_name: str, query: dict, page_size: int = 1000):
"""Paginate through all matching results using PIT + search_after."""
pit = es.open_point_in_time(index=index_name, keep_alive="5m")
sort = [{"published_at": "desc"}, {"_shard_doc": "desc"}]
last_sort = None
try:
while True:
body = {
"size": page_size,
"query": query,
"sort": sort,
"pit": {"id": pit["id"], "keep_alive": "5m"},
}
if last_sort:
body["search_after"] = last_sort
response = es.search(body=body)
hits = response["hits"]["hits"]
if not hits:
break
for hit in hits:
yield hit["_source"]
last_sort = hits[-1]["sort"]
finally:
es.close_point_in_time(id=pit["id"])
# ❌ BAD — Indexing one document at a time (slow)
def index_articles_bad(es, index, articles):
for article in articles:
es.index(index=index, document=article)
# ✅ GOOD — Bulk indexing (10-50x faster)
def index_articles_good(es, index, articles):
success, errors = helpers.bulk(es, (
{"_index": index, "_source": a} for a in articles
), raise_on_error=False)
if errors:
log.error(f"Bulk indexing failed for {len(errors)} documents")
# ❌ BAD — Inefficient wildcard query
def search_bad(es, index, term):
return es.search(index=index, query={"wildcard": {"title": f"*{term}*"}})
# ✅ GOOD — Analyzed match query with proper field
def search_good(es, index, term):
return es.search(index=index, query={"match": {"title": {"query": term, "fuzziness": "AUTO"}}})
helpers.bulk() for all batch indexing operationsfilter for structured data and must for full-text relevancenumber_of_shards based on your data volume and node count (20-40GB per shard)ilm (Index Lifecycle Management) for time-series indicesrefresh_interval to -1 during bulk indexing, reset afterscroll for deep pagination in user-facing queries — use search_after or PITwildcard and regexp queries on text fields — they skip the analysis indexmatch_all queries without size limits — can crash the clusterterms queries with thousands of values — use terms_set or filters insteadWhen writing Elasticsearch integration code, structure your output as:
| Skill | Purpose |
|---|---|
coding-mongodb-driver | Document database patterns — alternative to Elasticsearch for non-search workloads |
coding-rag-architecture | RAG pipeline patterns using Elasticsearch as vector store |
coding-postgresql-sdk | PostgreSQL full-text search as an alternative |