How to Build Semantic Search with Embeddings

# chunker.py
from dataclasses import dataclass
import tiktoken

@dataclass
class Chunk:
    content: str
    chunk_index: int
    token_count: int

def chunk_text(
    text: str,
    max_tokens: int = 512,
    overlap_tokens: int = 64,
    model: str = "text-embedding-3-small",
) -> list[Chunk]:
    enc = tiktoken.encoding_for_model(model)
    tokens = enc.encode(text)

    chunks: list[Chunk] = []
    start = 0
    idx = 0

    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        content = enc.decode(chunk_tokens)
        chunks.append(Chunk(
            content=content,
            chunk_index=idx,
            token_count=len(chunk_tokens),
        ))
        idx += 1
        start = end - overlap_tokens  # sliding window with overlap

    return chunks

# indexer.py
import asyncio
import hashlib
from openai import AsyncOpenAI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import text
from chunker import chunk_text

client = AsyncOpenAI()
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/semantic_db")

async def embed_batch(texts: list[str]) -> list[list[float]]:
    """Embed up to 2048 texts in one API call."""
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
        encoding_format="float",
    )
    return [item.embedding for item in response.data]

async def index_document(source_id: str, content: str) -> int:
    chunks = chunk_text(content)
    texts = [c.content for c in chunks]

    # Batch embed (max 2048 per call)
    embeddings = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        embeddings.extend(await embed_batch(batch))

    async with AsyncSession(engine) as session:
        for chunk, embedding in zip(chunks, embeddings):
            await session.execute(
                text("""
                    INSERT INTO documents (source_id, chunk_index, content, embedding)
                    VALUES (:source_id, :chunk_index, :content, :embedding)
                    ON CONFLICT (source_id, chunk_index)
                    DO UPDATE SET content = EXCLUDED.content,
                                  embedding = EXCLUDED.embedding
                """),
                {
                    "source_id": source_id,
                    "chunk_index": chunk.chunk_index,
                    "content": chunk.content,
                    "embedding": embedding,
                }
            )
        await session.commit()

    return len(chunks)

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text

app = FastAPI()

class SearchRequest(BaseModel):
    query: str = Field(..., min_length=2, max_length=500)
    top_k: int = Field(default=10, ge=1, le=50)
    min_score: float = Field(default=0.7, ge=0.0, le=1.0)

class SearchResult(BaseModel):
    source_id: str
    chunk_index: int
    content: str
    score: float
    metadata: dict

@app.post("/search", response_model=list[SearchResult])
async def search(req: SearchRequest):
    # Embed the query
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=req.query,
    )
    query_embedding = response.data[0].embedding

    async with AsyncSession(engine) as session:
        rows = await session.execute(
            text("""
                SELECT
                    source_id,
                    chunk_index,
                    content,
                    metadata,
                    1 - (embedding <=> :embedding) AS score
                FROM documents
                WHERE 1 - (embedding <=> :embedding) >= :min_score
                ORDER BY embedding <=> :embedding
                LIMIT :top_k
            """),
            {
                "embedding": query_embedding,
                "min_score": req.min_score,
                "top_k": req.top_k,
            }
        )
        results = rows.fetchall()

    if not results:
        return []

    return [
        SearchResult(
            source_id=row.source_id,
            chunk_index=row.chunk_index,
            content=row.content,
            score=round(row.score, 4),
            metadata=row.metadata or {},
        )
        for row in results
    ]

response = await client.embeddings.create(
    model="text-embedding-3-large",
    input=texts,
    dimensions=512,  # truncate from 3072 → 512
)

# reranker.py
from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, candidates: list[str], top_k: int = 10) -> list[tuple[int, float]]:
    """Returns (original_index, score) pairs sorted by relevance."""
    pairs = [(query, c) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
    return ranked[:top_k]

@app.post("/search", response_model=list[SearchResult])
async def search(req: SearchRequest):
    # ... embed query, fetch top-50 from pgvector ...
    candidates = [row.content for row in raw_results]
    ranked_indices = rerank(req.query, candidates, top_k=req.top_k)

    return [
        SearchResult(
            source_id=raw_results[i].source_id,
            chunk_index=raw_results[i].chunk_index,
            content=raw_results[i].content,
            score=round(float(score), 4),
            metadata=raw_results[i].metadata or {},
        )
        for i, score in ranked_indices
    ]