refactor elasticsearch indexers

Author: ks6088ts

docs/index.md

@@ -12,11 +12,10 @@ uv run python scripts/qdrant_operator.py delete-collection --collection-name qa_
 uv run python scripts/qdrant_operator.py add-documents --collection-name qa_kabuto --verbose
 uv run python scripts/qdrant_operator.py search-documents --collection-name qa_kabuto --question "「鬼灯」を実行すると、KABUTOが急に停止します。原因と対策を教えてください。" --verbose
 
-# Add documents to Elasticsearch
-uv run python -m template_langgraph.tasks.add_documents_to_elasticsearch
-
-# Search Elasticsearch
-uv run python -m template_langgraph.tasks.search_documents_on_elasticsearch
+# Elasticsearch
+uv run python scripts/elasticsearch_operator.py --help
+uv run python scripts/elasticsearch_operator.py add-documents --index-name docs_kabuto --verbose
+uv run python scripts/elasticsearch_operator.py search-documents --index-name docs_kabuto --query "禅モード" --verbose
 
 # Run Kabuto Helpdesk Agent
 uv run python -m template_langgraph.tasks.run_kabuto_helpdesk_agent "KABUTOの起動時に、画面全体が紫色に点滅し、システムがフリーズします。"

scripts/elasticsearch_operator.py

@@ -0,0 +1,109 @@
+import logging
+
+import typer
+from dotenv import load_dotenv
+
+from template_langgraph.loggers import get_logger
+from template_langgraph.tools.elasticsearch_tool import ElasticsearchClientWrapper
+from template_langgraph.tools.pdf_loaders import PdfLoaderWrapper
+
+# Initialize the Typer application
+app = typer.Typer(
+    add_completion=False,
+    help="Elasticsearch operator CLI",
+)
+
+# Set up logging
+logger = get_logger(__name__)
+
+
+@app.command()
+def search_documents(
+    index_name: str = typer.Option(
+        "docs_kabuto",
+        "--index-name",
+        "-i",
+        help="Name of the Elasticsearch index to search documents in",
+    ),
+    query: str = typer.Option(
+        "禅モード",
+        "--query",
+        "-q",
+        help="Query to search in the Elasticsearch index",
+    ),
+    verbose: bool = typer.Option(
+        False,
+        "--verbose",
+        "-v",
+        help="Enable verbose output",
+    ),
+):
+    # Set up logging
+    if verbose:
+        logger.setLevel(logging.DEBUG)
+
+    es = ElasticsearchClientWrapper()
+
+    results = es.search(
+        index_name=index_name,
+        query=query,
+    )
+    logger.info(f"Found {len(results)} results for the question: {query}")
+    for i, result in enumerate(results, start=1):
+        logger.info(f"Result {i}:")
+        logger.info(f"File Name: {result.metadata['source']}")
+        logger.info(f"Content: {result.page_content}")
+        logger.info("-" * 40)
+
+
+@app.command()
+def add_documents(
+    index_name: str = typer.Option(
+        "docs_kabuto",
+        "--index-name",
+        "-i",
+        help="Name of the Elasticsearch index to add documents to",
+    ),
+    verbose: bool = typer.Option(
+        False,
+        "--verbose",
+        "-v",
+        help="Enable verbose output",
+    ),
+):
+    # Set up logging
+    if verbose:
+        logger.setLevel(logging.DEBUG)
+
+    # Create Elasticsearch index
+    es = ElasticsearchClientWrapper()
+    logger.info(f"Creating Elasticsearch index: {index_name}")
+    result = es.create_index(
+        index_name=index_name,
+    )
+    if result:
+        logger.info(f"Created Elasticsearch index: {index_name}")
+    else:
+        logger.warning(f"Index {index_name} already exists.")
+
+    # Load documents from PDF files
+    documents = PdfLoaderWrapper().load_pdf_docs()
+    logger.info(f"Loaded {len(documents)} documents from PDF.")
+
+    # Add documents to Elasticsearch index
+    result = es.add_documents(
+        index_name=index_name,
+        documents=documents,
+    )
+    if result:
+        logger.info(f"Added {len(documents)} documents to Elasticsearch index: {index_name}")
+    else:
+        logger.error(f"Failed to add documents to Elasticsearch index: {index_name}")
+
+
+if __name__ == "__main__":
+    load_dotenv(
+        override=True,
+        verbose=True,
+    )
+    app()

scripts/qdrant_operator.py

@@ -12,7 +12,7 @@
 # Initialize the Typer application
 app = typer.Typer(
     add_completion=False,
-    help="template-langgraph CLI",
+    help="Qdrant operator CLI",
 )
 
 # Set up logging