Using CrewAI with Elasticsearch

pip install elasticsearch==8.17 'crewai[tools]'

import json
import os

from getpass import getpass
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

from crewai import Agent, Crew, Task
from crewai.tools import tool
from crewai_tools import SerperDevTool, WebsiteSearchTool

os.environ["ELASTIC_ENDPOINT"] = "your_es_url"
os.environ["ELASTIC_API_KEY"] = "es_api_key"

_client = Elasticsearch(
    os.environ["ELASTIC_ENDPOINT"],
    api_key=os.environ["ELASTIC_API_KEY"],
)

_client.options(
    request_timeout=60, max_retries=3, retry_on_timeout=True
).inference.put(
    task_type="sparse_embedding",
    inference_id="clothes-inference",
    body={
      "service": "elasticsearch",
        "service_settings": {
          "adaptive_allocations": { 
            "enabled": True
          },
          "num_threads": 1,
          "model_id": ".elser_model_2" 
        }
      }
)

_client.indices.create(
        index="summer-clothes",
        body={
             "mappings": {
                "properties": {
                  "title": {
                    "type": "text",
                    "copy_to": "semantic_field"
                  },
                  "description": {
                    "type": "text",
                    "copy_to": "semantic_field"
                  },
                  "price": {
                    "type": "float"
                  },
                  "semantic_field": {
                    "type": "semantic_text",
                    "inference_id": "clothes-inference"
                  }
                }
             }
        }
    )

documents = [
    {
        "title": "Twist-Detail Crop Top",
        "description": "Fitted crop top in woven, patterned fabric with linen content. Wide shoulder straps, sweetheart neckline, and gathered side seams for a gently draped effect. Twisted detail at center bust, cut-out section at front, and wide smocking at back. Lined",
        "price": 34.99
    },
    {
        "title": "Rib-knit Tank Top",
        "description": "Short, fitted top in a soft rib knit. Extra-narrow shoulder straps and a square neckline.",
        "price": 7.49
    },
    {
        "title": "Linen-blend Shorts",
        "description": "Shorts in an airy, woven linen blend. High, ruffle-trimmed waist, narrow drawstring and covered elastic at waistband, and discreet side pockets.",
        "price": 13.99
    },
    {
        "title": "Twill Cargo Shorts",
        "description": "Fitted shorts in cotton twill with a V-shaped yoke at front and back. High waist, zip fly with button, and patch front pockets.",
        "price": 20.99
    },
    {
        "title": "Slim Fit Ribbed Tank Top",
        "description": "Slim-fit tank top in medium weight, ribbed cotton-blend jersey with a fitted silhouette. Straight-cut hem.",
        "price": 8.49
    },
    {
        "title": "Relaxed Fit Linen Resort Shirt",
        "description": "Relaxed-fit shirt in airy linen. Resort collar, buttons without placket, yoke at back, and short sleeves. Straight-cut hem. Fabric made from linen is breathable, looks great when ironed or wrinkled, and softens over time.",
        "price": 17.99
    },
    {
        "title": "Swim Shorts",
        "description": "Swim shorts in woven fabric. Drawstring and covered elastic at waistband, side pockets, and a back pocket with hook-loop fastener. Small slit at sides. Mesh liner shorts.",
        "price": 14.99
    },
    {
        "title": "Baggy Fit Cargo Shorts",
        "description": "Baggy-fit cargo shorts in cotton canvas with a generous but not oversized silhouette. Zip fly with button, diagonal side pockets, back pockets with flap and snap fasteners, and bellows leg pockets with snap fasteners.",
        "price": 20.99
    },
    {
        "title": "Muslin Shorts",
        "description": "Shorts in airy cotton muslin. High, ruffle-trimmed waist, covered elastic at waistband, and an extra-narrow drawstring with a bead at ends. Discreet side pockets.",
        "price": 15.99
    },
    {
        "title": "Oversized Lyocell-blend Dress",
        "description": "Short, oversized dress in a woven lyocell blend. Gathered, low-cut V-neck with extra-narrow ties at front, 3/4-length, raglan-cut balloon sleeves with narrow elastic at cuffs, and seams at waist and hips with delicate piping. Unlined.",
        "price": 38.99
    }
]

def build_data():
    for doc in documents:
        yield {
            "_index": "summer-clothes",
            "_source": doc
        }

try:
    success, errors = bulk(_client, build_data())
    print(f"{success} documents indexed successfully")
    if errors:
        print("Errors during indexing:", errors)
                
except Exception as e:
    print(f"Error: {str(e)}")

@tool("es tool")
def elasticsearch_tool(question: str) -> str:
    """
    Search in Elasticsearch using hybrid search capabilities.

    Args:
        question (str): The search query to be semantically matched

    Returns:
        str: Concatenated hits from Elasticsearch as string JSON
    """

    response = _client.search(
        index="summer-clothes",
        body={
            "size": 10,
            "_source": {"includes": ["description", "title", "price"]},
            "retriever": {
                "rrf": {
                    "retrievers": [
                        {"standard": {"query": {"match": {"title": question}}}},
                        {
                            "standard": {
                                "query": {
                                    "semantic": {
                                        "field": "semantic_field",
                                        "query": question,
                                    }
                                }
                            }
                        },
                    ]
                }
            },
        },
    )

    hits = response["hits"]["hits"]

    if not hits:
        return ""

    result = json.dumps([hit["_source"] for hit in hits], indent=2)

    return result

os.environ["SERPER_API_KEY"] = "your-key"
os.environ["OPENAI_API_KEY"] = "your-key"

search_tool = SerperDevTool()
web_rag_tool = WebsiteSearchTool()

es_retriever_agent = Agent(
    role="Retriever",
    goal="Retrieve Elasticsearch documents",
    backstory="You are an expert researcher",
    tools=[elasticsearch_tool],
    verbose=True,
)


internet_researcher_agent = Agent(
    role="Research analyst",
    goal="Provide up-to-date market analysis of the industry",
    backstory="You are an expert analyst",
    tools=[search_tool, web_rag_tool],
    verbose=True,
)


writer_agent = Agent(
    role="Content Writer",
    goal="Craft engaging blog posts about the information gathered",
    backstory="A skilled writer with a passion for writing about fashion",
    tools=[],
    verbose=True,
)

es_retriever_task = Task(
    description="Retrieve documents from the Elasticsearch index.",
    expected_output="A list of documents retrieved from the Elasticsearch index based on the query.",
    agent=es_retriever_agent,
)


internet_research_task = Task(
    description="Conduct research on the latest fashion trends for summer 2025. Identify five key trends that are shaping the industry, including popular colors, fabrics, and styles. Use reliable sources such as fashion magazines, retail market reports, and industry analyses. For each trend, provide a concise summary explaining its significance and why it is gaining popularity. Clearly cite your sources using a format like [Source Name]. Your response should be structured and fact-based",
    expected_output="A structured summary of the top five fashion trends for summer 2025, including citations for each trend in the format [Source Name]",
    agent=internet_researcher_agent,
)

write_task = Task(
    description="Compare the fashion trends from the Research Agent with product listings from Elasticsearch. Write a short report highlighting how the store's products align with trends. Use '[Elasticsearch]' for database references and '[Source Name]' for external sources",
    expected_output="A short, structured report combining trend insights with product recommendations, with clearly marked references",
    agent=writer_agent,
    output_file="blog-posts/new_post.md",
)

# Use in a crew
crew = Crew(
    agents=[es_retriever_agent, internet_researcher_agent, writer_agent],
    tasks=[
        es_retriever_task,
        internet_research_task,
        write_task,
    ],
)

# Execute tasks
crew.kickoff()