How do you allow users to search Elasticsearch data using natural language?
Build an agent that interprets user intent and generates precise Elasticsearch queries. This guide outlines a smart Elasticsearch agent using LangGraph for orchestration and Google Gemini for natural language understanding and query generation. The agent translates natural language into DSL queries, enabling broader access to data and analytics.
The system uses two core technologies. LangGraph provides the orchestration framework. It defines the multi-step workflow for flexibility and scalability. Google Gemini, a large language model (LLM), handles natural language understanding and generates Elasticsearch DSL. This combination creates an efficient, user-friendly search interface.
A smart Elasticsearch agent is a LangGraph workflow that uses Google Gemini to turn a natural language query into Elasticsearch DSL, runs the DSL on the Elasticsearch cluster and summarizes the results.
.
Purpose of an Elasticsearch Agent
Elasticsearch provides high performance, but its Query DSL is complex. An agent allows users to input queries in natural language. It returns accurate results without requiring users to learn the syntax. This abstraction replaces manual DSL construction with natural conversation. It lowers the barrier to entry for non-technical users.
LangGraph: Core Orchestration Layer for the Elasticsearch Agent
LangGraph uses a graph-based architecture. Each node represents a specific task, such as a database query or LLM call. Edges define the flow of data and context between nodes. This structure supports complex workflow construction. It enables parallel execution. It supports conditional branching. It allows rejoining of paths based on predefined logic. The entire workflow remains visible within a single graph.
LangChain integrates large language models (LLMs) into workflows. It excels at sequential tasks, or prompt chaining, and connects to external data sources. LangGraph builds on LangChain. It supports complex, agent-based workflows with high task dependency. LangGraph manages state and control for dynamic, multi-agent interactions. It does not replace LangChain. It extends LangChain. LangGraph requires LangChain to operate.
LangGraph uses a stateful, graph-based architecture. It moves agent development beyond sequential processing. The system handles adaptive, resilient, and collaborative decisions. It defines complex paths and manages internal state. It supports cyclical behavior, such as reflection. The Elasticsearch Agent adjusts strategies based on intermediate results or failures. If DSL generation fails, the system re-evaluates, retries, or branches. This design handles unexpected inputs and system states. LangGraph supports multi-agent systems. It assigns specialized roles to agents. Examples include query normalization, DSL generation, and summarization. Shared state allows agents to pass context. This modularity enables efficient collaboration. This architecture builds sophisticated AI copilots. It replaces single monolithic LLM calls with distributed, intelligent systems.
Gemini Powers Natural Language Understanding and Elasticsearch DSL Generation
Google Gemini interprets natural language queries. It translates them into structured, executable formats. Specifically, it generates Elasticsearch DSL. This mirrors Gemini in BigQuery. That tool generates SQL from natural language prompts. It also refines and summarizes queries. This parallels the need for an Elasticsearch Agent to generate accurate DSL.
The Gemini API defaults to unstructured text. You can configure it for structured output, such as JSON. This is critical because Elasticsearch DSL is JSON. Two methods achieve this:
- Configure a schema directly on the model. This method offers higher quality and determinism.
- Provide a schema within a text prompt. This method is less deterministic.
Defining a responseSchema ensures output adheres to a specific structure. This reduces post-processing. It enables direct data extraction. This setup reliably parses generated DSL and summaries. Supported MIME types include application/json and text/x.enum. These types control the output format.
Gemini produces constrained, structured output. Precise prompt engineering transforms the LLM from a general text generator into a reliable component. Defining a response schema enforces JSON output with specific keys. This makes Gemini a predictable element in the agent workflow. Reliable automation requires consistent valid JSON. The pipeline fails if the DSL generation step produces invalid output. Structured output allows the Elasticsearch client to parse data. This reduces the need for fragile post-processing. Schemas and precise prompts reduce invalid DSL or irrelevant structures. This increases trust in the agent output. The design suits production environments that require consistency and accuracy.
Prompt engineering guides LLM responses for specific tasks. It shapes output through structured input instructions. For the Elasticsearch Agent, prompts act as a blueprint. They define how Gemini normalizes queries, generates DSL, and summarizes results. Advanced techniques improve DSL accuracy. One method adds the language grammar to the prompt. It uses Extended Backus-Naur Form (EBNF). This clarifies the expected syntax for the model.
Agent Workflow Overview
The agent operates in defined steps. LangGraph defines the workflow. This structure supports future modification and extension.
- User Input: The user provides a natural language query. Example: "Show me recent error logs related to authentication."
- Query Normalization. System corrects spelling and removes redundant terms to prepare input for translation.
- DSL Generation. System sends normalized query to Gemini. Gemini converts input to valid Elasticsearch DSL.
- Elasticsearch Search. System executes generated DSL query on the Elasticsearch cluster.
- Response Summarization. System converts raw results into a concise, readable format.
- Final Output. System presents summarized results to the user.
How agents drive enterprise value
Implementation Steps
Build the agent using these steps:
Prerequisites
- Python: Install version 3.7 or later
- Libraries: Install dependencies using pip.
| pip install langchain langgraph google-generativeai python-dotenv requests |
|---|
- Google Gemini API Key: Obtain an API key from the Google Cloud AI Platform. Enable the Gemini API. Use the free tier with defined usage limits.
- Elasticsearch Cluster: Provide access to an Elasticsearch cluster.
2. Project Setup:
- Create a Project Directory: Create a new directory for the agent project (e.g., es_agent).
Create a `.env` file in your project directory. This file stores your sensitive credentials: the API key and Elasticsearch URL. Add the following lines, replacing placeholders with your actual values:
| GEMINI_API_KEY="YOUR_GEMINI_API_KEY"ES_URL="YOUR_ELASTICSEARCH_URL" # e.g., " https://localhost:9200/_search " |
|---|
- Create three text files in your project directory. These files hold LLM prompts for the graph states: - `query_normalizer_prompt.txt` - `es_dsl_formation_prompt.txt` - `generic_es_response_summariser.txt`
3. Core Code (`es_agent.py`)
Create `es_agent.py` in your project directory. This file contains the agent's main code.
| import os import json import requests import logging from dotenv import load_dotenv import google.generativeai as genai from langgraph.graph import StateGraph, END from langchain_core.runnables import RunnableLambda # ─── Logging Configuration ───────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format= "%(asctime)s %(levelname)s %(name)s ─ %(message)s" , datefmt= "%Y-%m-%d %H:%M:%S" )logger = logging.getLogger(__name__)load_dotenv() # ─── Load environment variables ───────────────────────────────────────────────── GEMINI_API_KEY = os.getenv( "GEMINI_API_KEY" )ES_ENDPOINT = os.getenv( "ES_URL" ) # ─── Load prompt files ─────────────────────────────────────────────────────────── def load_prompt (filename): base_dir = os.path.dirname(os.path.abspath(__file__)) full_path = os.path.join(base_dir, filename) try : with open(full_path, "r" ) as f: text = f.read () logger.info ( f"Loaded prompt from '{full_path}' ({len(text)} bytes)." ) return text except Exception as e: logger.error( f"Error loading prompt file '{full_path}': {e}" ) return "" NORMALIZER_PROMPT = load_prompt( "query_normalizer_prompt.txt" )DSL_PROMPT = load_prompt( "es_dsl_formation_prompt.txt" )SUMMARY_PROMPT = load_prompt( "generic_es_response_summariser.txt" ) # ─── Configure Gemini ──────────────────────────────────────────────────────────── if not GEMINI_API_KEY: logger.warning( "GEMINI_API_KEY is not set in environment; model calls will fail." )genai.configure(api_key=GEMINI_API_KEY)model = genai.GenerativeModel( "gemini-2.0-flash" ) # ─── Shared utility ───────────────────────────────────────────────────────────── def strip_fences (text: str) -> str: """ Remove markdown json fences (```json ... ```). """ original = text text = text.strip() if text.startswith( "```json" ): text = text[ 7 :] if text.endswith( "```" ): text = text[: -3 ] stripped = text.strip() if original != stripped: logger.debug( "Stripped markdown fences from Gemini output." ) return stripped # ─── Step 1: Normalize Query ───────────────────────────────────────────────────── def normalize_query (state: dict) -> dict: nl_query = state.get( "query" , "" ) logger.info ( f"[NormalizeQuery] Received query: '{nl_query}'" ) try : prompt = NORMALIZER_PROMPT.format(user_query=nl_query) logger.debug( f"[NormalizeQuery] Prompt sent to Gemini (length {len(prompt)})." ) response = model.generate_content(prompt) normalized = strip_fences(response.text) logger.info ( f"[NormalizeQuery] Normalized query: '{normalized}'" ) return {**state, "normalized_query" : normalized} except Exception as e: logger.error( f"[NormalizeQuery] Error during normalization: {e}" ) return {**state, "error" : f"Normalization failed: {e}" } # ─── Step 2: NL to DSL ──────────────────────────────────────────────────────────── def generate_es_dsl (state: dict) -> dict: normalized = state.get( "normalized_query" , "" ) logger.info ( f"[GenerateDSL] Converting normalized query to DSL: '{normalized}'" ) try : prompt = DSL_PROMPT.format(nl_query=normalized) logger.debug( f"[GenerateDSL] Prompt sent to Gemini (length {len(prompt)})." ) response = model.generate_content(prompt) dsl_string = strip_fences(response.text) logger.debug( f"[GenerateDSL] Gemini returned DSL string: {dsl_string[: 100 ]}{'...' if len(dsl_string) > 100 else ''}" ) dsl = json.loads(dsl_string) logger.info ( f"[GenerateDSL] Parsed DSL JSON successfully." ) logger.info (dsl_string) return {**state, "es_dsl" : dsl} except json.JSONDecodeError as e: logger.error( f"[GenerateDSL] JSON parsing error: {e}. Raw DSL: {dsl_string}" ) return {**state, "error" : f"Failed to parse DSL JSON: {e}" } except Exception as e: logger.error( f"[GenerateDSL] Unexpected error: {e}" ) return {**state, "error" : f"DSL generation failed: {e}" } # ─── Step 3: Search Elasticsearch ──────────────────────────────────────────────── def search_elasticsearch (state: dict) -> dict: es_dsl = state.get( "es_dsl" ) if not es_dsl: logger.warning( "[QueryES] No DSL found in state; skipping Elasticsearch call." ) return {**state, "error" : "No DSL to execute." } logger.info ( f"[QueryES] Sending DSL to Elasticsearch endpoint: {ES_ENDPOINT}" ) truncated_payload = json.dumps(es_dsl)[: 100 ] + ( "..." if len(json.dumps(es_dsl)) > 100 else "" ) logger.debug( f"[QueryES] DSL payload (truncated): {truncated_payload}" ) try : res = requests.post ( ES_ENDPOINT, json=es_dsl, headers={ "Content-Type" : "application/json" }, verify= False ) res.raise_for_status() es_response = res.json() hits = es_response.get( "hits" , {}).get( "total" , {}).get( "value" , "unknown" ) logger.info ( f"[QueryES] Elasticsearch returned {hits} hits." ) return {**state, "es_raw_response" : es_response} except requests.exceptions.RequestException as e: logger.error( f"[QueryES] HTTP error: {e}" ) return {**state, "error" : f"Elasticsearch HTTP error: {e}" } except json.JSONDecodeError as e: logger.error( f"[QueryES] JSON decode error on ES response: {e}" ) return {**state, "error" : f"ES response JSON decode failed: {e}" } except Exception as e: logger.error( f"[QueryES] Unexpected error: {e}" ) return {**state, "error" : f"Elasticsearch search failed: {e}" } # ─── Step 4: Summarize ──────────────────────────────────────────────────────────── def summarize_response (state: dict) -> dict: if "es_raw_response" not in state: logger.warning( "[Summarize] No ES response found; skipping summarization." ) return state raw = state[ "es_raw_response" ] logger.info ( "[Summarize] Summarizing Elasticsearch response with Gemini." ) try : prompt = SUMMARY_PROMPT.format( es_response=json.dumps(raw), nl_query=state.get( "query" , "" ) ) logger.debug( f"[Summarize] Prompt sent to Gemini (length {len(prompt)})." ) response = model.generate_content(prompt) summary_text = strip_fences(response.text) logger.debug( f"[Summarize] Raw summary text (truncated): {summary_text[: 100 ]}{'...' if len(summary_text) > 100 else ''}" ) try : summary_json = json.loads(summary_text) logger.info ( "[Summarize] Parsed summary JSON successfully." ) return {**state, "summary" : summary_json} except json.JSONDecodeError: logger.warning( "[Summarize] Summary is not valid JSON; returning raw string." ) return {**state, "summary" : summary_text} except Exception as e: logger.error( f"[Summarize] Error during summarization: {e}" ) return {**state, "error" : f"Summarization failed: {e}" } # ─── Step 5: Final Output ──────────────────────────────────────────────────────── def output_result (state: dict) -> dict: logger.info ( "[Output] Generating final output." ) return json.dumps(state[ "summary" ], indent= 2 ) # ─── Build the graph (use plain dict as the state type) ────────────────────────── graph = StateGraph(dict)graph.add_node( "NormalizeQuery" , RunnableLambda(normalize_query))graph.add_node( "GenerateDSL" , RunnableLambda(generate_es_dsl))graph.add_node( "QueryES" , RunnableLambda(search_elasticsearch))graph.add_node( "Summarize" , RunnableLambda(summarize_response))graph.add_node( "Output" , RunnableLambda(output_result)) # Define edges graph.set_entry_point( "NormalizeQuery" )graph.add_edge( "NormalizeQuery" , "GenerateDSL" )graph.add_edge( "GenerateDSL" , "QueryES" )graph.add_edge( "QueryES" , "Summarize" )graph.add_edge( "Summarize" , "Output" )graph.add_edge( "Output" , END) # ─── Compile the graph ─────────────────────────────────────────────────────────── graph_executor = graph.compile() # ─── Expose a helper function to run the agent ─────────────────────────────────── def es_search_agent (nl_query: str): """ Invoke the LangGraph agent with a natural-language query string. """ logger.info ( f"[run_query] Invoking agent with query: '{nl_query}'" ) result_state = graph_executor.invoke({ "query" : nl_query}) return json.loads(result_state) |
|---|
4. Define Prompts
Define prompts to control Gemini behavior. Effective prompts drive accuracy. Use this structure for your prompt files:
| query_normalizer_prompt.txt |
|---|
| You are a query normalizer. Your task is to take a user's search query and normalize it to improve search accuracy. This includes correcting spelling mistakes, removing irrelevant words (like "please" or "thank you"), and standardizing terminology.User Query: {user_query}Normalized Query: |
|---|
| es_dsl_formation_prompt.txt |
|---|
| You are an expert in Elasticsearch DSL (Domain Specific Language). Your task is to translate a natural language search query into a valid Elasticsearch DSL query in JSON format. The DSL query should be as specific as possible to accurately reflect the user's intent. You must enclose the valid JSON object with markdown fences.Natural Language Query: {nl_query}Elasticsearch DSL Query: |
|---|
| generic_es_response_summariser.txt |
|---|
| You are a summarization expert. Your task is to summarize the results of an Elasticsearch query in a concise and informative way for a user. You must enclose the valid JSON object with markdown fences. Only answer in JSON format with the key `summary`.Natural Language Query: {nl_query}Elasticsearch Response: {es_response}Summary: |
|---|
Prompting Guidelines:
- Be specific. Clear instructions improve output quality.
- Provide examples. Include input-output pairs to guide few-shot learning.
- Iterate. Test different prompts. Select the highest-performing version.
- Map data structure: Align prompts with your specific Elasticsearch schema.
- Constrain output: Request JSON and limit allowed keys.
5. Run the agent
Test the agent. Add the following code to the end of es_agent.py:
| if name == "__main__" : query = "Show me recent error logs" result = es_search_agent(query) print(json.dumps(result, indent= 2 )) |
|---|
The agent performs these steps:
- Processes the query
- Generates a DSL query
- Executes the query against Elasticsearch
- Summarizes results
- Prints the final output
6. Secure the Agent
- SSL/TLS: Do not disable SSL certificate verification (verify=False) in production. Configure SSL/TLS for the Elasticsearch cluster. Ensure the agent verifies the certificate.
- Authentication/Authorization: Implement authentication and authorization to control access to the Elasticsearch cluster.
- Input Sanitization: Sanitize user input to prevent injection attacks.
7. Next Steps and Improvements
- Complex Prompts: Test advanced prompt structures to cover diverse query types.
- Context Tracking: Add state management to support multi-turn conversations.
- Failure Recovery: Add robust error handling to manage system failures.
- User Interface: Build a web or command-line interface for agent interaction.
- Performance Tuning: Capture user feedback to refine agent accuracy over time.
- Streaming: Return responses to the user immediately upon arrival.
Key use cases for a smart Elasticsearch Agent include:
- Observability and Security: In IT operations and cybersecurity, the Agent identifies root causes. It decodes error messages, stack traces, and logs. It identifies performance bottlenecks. For security, it collects, stores, and analyzes data. It monitors endpoints and supports threat hunting.
- Search Applications: The Agent supports advanced search features. It enables semantic search to interpret intent. It combines full-text and vector search in hybrid models. It serves as a retrieval engine for Retrieval Augmented Generation (RAG) systems.
- Business Intelligence and Analytics: The Agent simplifies analytics and reporting. It converts natural language into Elasticsearch SQL. It integrates this SQL with Business Intelligence (BI) tools. It aggregates and visualizes large datasets.
- Customer Service and Support. In support scenarios, the agent locates relevant information in large knowledge bases. It processes natural language queries regarding past tickets or product documentation. This approach improves response time and accuracy.
Advanced Elasticsearch Agent Considerations
Secure the Elasticsearch Agent
Security is essential for production applications, particularly those handling sensitive data. Disabling SSL certificate verification (verify=False) creates a critical security risk. Do not use this setting in production. This configuration exposes data to eavesdropping and man-in-the-middle attacks.
Implement the following security measures for the Elasticsearch Agent:
Configure SSL/TLS for node-to-node transport and client HTTP connections. Set `xpack.security.transport.ssl.enabled` to true. Set `xpack.security.http.ssl.enabled` to true.
Network Security
Restrict node access by IP address. Use `xpack.security.transport.filter.allow` and `xpack.security.http.filter.allow`. Configure firewalls to open only ports 9200 (HTTP) and 9300 (transport). Restrict access to management interfaces. Isolate Elasticsearch from public internet access unless required.
Data Protection
Restrict access to sensitive document fields, such as `ssn` and `credit_card`.
Implement document-level security. Filter results based on user context. Users access only authorized data.
Enable audit logging. Set `xpack.security.audit.enabled: true`. Track security events, including failed authentication attempts, privilege escalation attempts, and unusual access patterns.
Apply these security measures systematically. This ensures the agent functions as a trusted enterprise component. Strong security supports compliance with data protection regulations. It also builds user confidence. Data security is mandatory for production systems handling sensitive information.
Summary
Developing a smart Elasticsearch agent improves data usability. Combine LangGraph, Google Gemini, and precise prompts to create an effective assistant. This design helps users search and analyze data efficiently. Prioritize security. Iterate on prompts to optimize performance. Review the linked resource for details on custom AI agents. Build an Elastic agent for your enterprise.
Related reading: LangChain MCP integration for multi-tool AI agents and LangGraph for multi-agent systems in LangChain.
Contact
FAQs
1. What is a smart Elasticsearch agent? Why use one? A smart Elasticsearch agent converts natural language into Elasticsearch DSL. It removes the need to learn syntax. Users can run complex queries easily. This boosts team productivity.
2. How do LangGraph and Google Gemini work together? LangGraph controls the workflow. It manages query normalization, DSL generation, and summarization. Google Gemini interprets user input. It generates accurate DSL in JSON. The combination creates a scalable search assistant.
3. Can I search Elasticsearch using natural language? Yes. The agent accepts plain text, such as “Show me recent authentication errors.” It normalizes the query, builds the DSL, and runs the search. It returns clear, readable results. This allows non-technical users to access Elasticsearch.
4. Which security measures are essential? Security is critical. Enable SSL/TLS. Implement authentication and authorization. Filter sensitive fields. Control document-level access. Enable audit logging. These steps ensure safe enterprise deployment.
5. How does SquareShift help build a smart Elasticsearch agent? SquareShift uses LangGraph and Google Gemini to design enterprise-ready agents. We manage workflow orchestration, DSL generation, and summarization. Teams can search data in natural language. They do not need to learn complex queries.
6. Can SquareShift customize agents for specific industries? Yes. SquareShift adapts prompts, workflows, and DSL logic for specific use cases. This applies to IT operations, cybersecurity, customer support, and analytics. We align results with your business data.
7. What are the benefits of partnering with SquareShift? Enterprises gain faster insights. Productivity improves. Reliance on technical teams decreases. SquareShift delivers secure, scalable, and reliable agents. You unlock the full value of your data.
How does the smart Elasticsearch agent work step by step? The user enters a natural language query. The agent normalizes it, sends it to Gemini to generate Elasticsearch DSL, runs the DSL on the Elasticsearch cluster, summarizes the response and presents the final output. LangGraph defines this workflow, which supports later changes and extensions.
What do you need to build an Elasticsearch agent with LangGraph and Gemini? You need Python 3.7 or later, the langchain, langgraph, google-generativeai, python-dotenv and requests libraries, a Google Gemini API key with the Gemini API enabled, and access to an Elasticsearch cluster. You also create three prompt files for query normalization, DSL formation and response summarization.
Why does the agent use structured JSON output from Gemini? Elasticsearch DSL is JSON, so the pipeline needs valid JSON. Defining a response schema on the model gives higher quality and determinism than putting a schema in the prompt. It reduces post-processing and lets the agent parse generated DSL and summaries reliably.