Traditional keyword search fails when users phrase queries differently from how content is written. AI-powered semantic search solves this by understanding meaning, not just matching words. In this tutorial, you will build a fully functional search engine that understands natural language queries using OpenAI embeddings and a vector database.

By the end, you will have a working system that can search through any text corpus with human-like understanding.

What You Will Build

The architecture is straightforward:

  1. Documents get converted into vector embeddings via OpenAI
  2. Embeddings are stored in a vector database (ChromaDB)
  3. User queries get embedded the same way
  4. The system finds documents with the most similar vectors
  5. Results are ranked by semantic relevance

This is the foundation of RAG (Retrieval-Augmented Generation) systems used in production by companies like Notion, Stripe, and Shopify.

Prerequisites

  • Python 3.9 or higher
  • An OpenAI API key
  • Basic familiarity with Python and REST APIs

Step 1: Set Up Your Project

Create a new directory and install the required dependencies:

mkdir ai-search-engine && cd ai-search-engine
python -m venv venv
source venv/bin/activate
pip install openai chromadb python-dotenv flask

Create a .env file to store your API key:

OPENAI_API_KEY=sk-your-key-here

Step 2: Create the Embedding Pipeline

The core of any semantic search system is the embedding function. This converts text into high-dimensional vectors that capture meaning.

# embeddings.py
import os
from openai import OpenAI
from dotenv import load_dotenv
 
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
 
def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """Convert text into a vector embedding."""
    text = text.replace("\n", " ").strip()
    response = client.embeddings.create(input=[text], model=model)
    return response.data[0].embedding
 
def get_embeddings_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """Batch embed multiple texts for efficiency."""
    cleaned = [t.replace("\n", " ").strip() for t in texts]
    response = client.embeddings.create(input=cleaned, model=model)
    return [item.embedding for item in response.data]

The text-embedding-3-small model offers a good balance between quality and cost at $0.02 per million tokens. For higher accuracy, use text-embedding-3-large.

Step 3: Set Up the Vector Database

ChromaDB is a lightweight vector database that runs locally without any infrastructure setup.

# database.py
import chromadb
from embeddings import get_embedding, get_embeddings_batch
 
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}
)
 
def index_documents(documents: list[dict]):
    """Index a batch of documents into the vector store."""
    texts = [doc["content"] for doc in documents]
    ids = [doc["id"] for doc in documents]
    metadatas = [{"title": doc.get("title", ""), "source": doc.get("source", "")} for doc in documents]
 
    embeddings = get_embeddings_batch(texts)
 
    collection.upsert(
        ids=ids,
        embeddings=embeddings,
        documents=texts,
        metadatas=metadatas
    )
    print(f"Indexed {len(documents)} documents.")
 
def search(query: str, n_results: int = 5) -> list[dict]:
    """Search for documents semantically similar to the query."""
    query_embedding = get_embedding(query)
 
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
        include=["documents", "metadatas", "distances"]
    )
 
    return [
        {
            "content": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
            "score": 1 - results["distances"][0][i]  # Convert distance to similarity
        }
        for i in range(len(results["documents"][0]))
    ]

Step 4: Build the Ingestion Script

You need a way to feed documents into your search engine. Here is a script that processes text files from a directory:

# ingest.py
import os
import hashlib
from database import index_documents
 
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into overlapping chunks for better retrieval."""
    words = text.split()
    chunks = []
    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        if chunk.strip():
            chunks.append(chunk)
    return chunks
 
def ingest_directory(directory: str):
    """Process all text files in a directory."""
    documents = []
    for filename in os.listdir(directory):
        if not filename.endswith((".txt", ".md")):
            continue
        filepath = os.path.join(directory, filename)
        with open(filepath, "r") as f:
            content = f.read()
 
        chunks = chunk_text(content)
        for i, chunk in enumerate(chunks):
            doc_id = hashlib.md5(f"{filename}_{i}".encode()).hexdigest()
            documents.append({
                "id": doc_id,
                "content": chunk,
                "title": filename,
                "source": filepath
            })
 
    index_documents(documents)
    print(f"Ingested {len(documents)} chunks from {directory}")
 
if __name__ == "__main__":
    ingest_directory("./data")

Step 5: Create the Search API

Wrap everything in a simple Flask API so other applications can query your search engine:

# app.py
from flask import Flask, request, jsonify
from database import search
 
app = Flask(__name__)
 
@app.route("/search", methods=["POST"])
def search_endpoint():
    data = request.get_json()
    query = data.get("query", "")
    limit = data.get("limit", 5)
 
    if not query:
        return jsonify({"error": "Query is required"}), 400
 
    results = search(query, n_results=limit)
    return jsonify({"results": results, "query": query})
 
@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok"})
 
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

Step 6: Test Your Search Engine

First, create a data/ directory with some sample text files, then run the ingestion:

mkdir data
# Add some .txt or .md files to ./data/
python ingest.py

Start the server and send a test query:

python app.py &
curl -X POST http://localhost:8000/search \
  -H "Content-Type: application/json" \
  -d '{"query": "how does authentication work", "limit": 3}'

You should see semantically relevant results even if the exact words "authentication" do not appear in your documents.

Practical Tips

Chunking strategy matters. Smaller chunks (200-500 words) give more precise results but lose context. Larger chunks (500-1000 words) preserve context but may dilute relevance. Experiment with your specific data.

Use metadata filtering. ChromaDB supports filtering by metadata fields. Add category, date, or author metadata to let users narrow their search scope.

Cache embeddings aggressively. Embedding API calls cost money. Store embeddings in the database and only re-embed when content changes.

Consider hybrid search. Combine vector search with keyword search (BM25) for the best results. Libraries like rank-bm25 make this straightforward to add.

Monitor costs. The text-embedding-3-small model costs $0.02 per million tokens. A corpus of 10,000 documents (500 words each) costs roughly $0.05 to embed entirely.

Next Steps

Once your basic search engine works, consider these enhancements:

  • Add a reranking step using a cross-encoder model for more accurate top results
  • Implement query expansion to handle ambiguous searches
  • Add a chat interface that uses search results as context (full RAG)
  • Deploy to production using a managed vector database like Pinecone or Weaviate